Skip to content

[Service Bus] Move settlement & renewLock methods from message to receiver #11962

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ramya-rao-a merged 9 commits into from
Oct 22, 2020
Merged

[Service Bus] Move settlement & renewLock methods from message to receiver #11962

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
7797d02
[Service Bus] Move settlement methods to receiver
ramya-rao-a Oct 21, 2020
a4d73e4
Revert change to body in amqpannotatedmessage
ramya-rao-a Oct 21, 2020
eb4a395
Clean up
ramya-rao-a Oct 21, 2020
41eef8f
Fix streaming receiver tests
ramya-rao-a Oct 21, 2020
11c467b
Remove obsolete casting
ramya-rao-a Oct 22, 2020
3ec99ef
jsdoc update for settlement methods as per feedback
ramya-rao-a Oct 22, 2020
c77d5df
Merge remote-tracking branch 'origin' into settlement
ramya-rao-a Oct 22, 2020
cef80ad
Formatting
ramya-rao-a Oct 22, 2020
52b585f
Enable non session backupMessageSettlement tests
ramya-rao-a Oct 22, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 27 additions & 31 deletions sdk/servicebus/service-bus/review/service-bus.api.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -192,9 +192,9 @@ export interface GetMessageIteratorOptions extends OperationOptionsBase {
}

// @public
export interface MessageHandlers<ReceivedMessageT> {
export interface MessageHandlers {
processError(args: ProcessErrorArgs): Promise<void>;
processMessage(message: ReceivedMessageT): Promise<void>;
processMessage(message: ServiceBusReceivedMessage): Promise<void>;
}

export { MessagingError }
Expand Down Expand Up @@ -348,19 +348,19 @@ export class ServiceBusAdministrationClient extends ServiceClient {
export class ServiceBusClient {
constructor(connectionString: string, options?: ServiceBusClientOptions);
constructor(fullyQualifiedNamespace: string, credential: TokenCredential, options?: ServiceBusClientOptions);
acceptNextSession(queueName: string, options?: AcceptSessionOptions<"peekLock">): Promise<ServiceBusSessionReceiver<ServiceBusReceivedMessageWithLock>>;
acceptNextSession(queueName: string, options: AcceptSessionOptions<"receiveAndDelete">): Promise<ServiceBusSessionReceiver<ServiceBusReceivedMessage>>;
acceptNextSession(topicName: string, subscriptionName: string, options?: AcceptSessionOptions<"peekLock">): Promise<ServiceBusSessionReceiver<ServiceBusReceivedMessageWithLock>>;
acceptNextSession(topicName: string, subscriptionName: string, options: AcceptSessionOptions<"receiveAndDelete">): Promise<ServiceBusSessionReceiver<ServiceBusReceivedMessage>>;
acceptSession(queueName: string, sessionId: string, options?: AcceptSessionOptions<"peekLock">): Promise<ServiceBusSessionReceiver<ServiceBusReceivedMessageWithLock>>;
acceptSession(queueName: string, sessionId: string, options: AcceptSessionOptions<"receiveAndDelete">): Promise<ServiceBusSessionReceiver<ServiceBusReceivedMessage>>;
acceptSession(topicName: string, subscriptionName: string, sessionId: string, options?: AcceptSessionOptions<"peekLock">): Promise<ServiceBusSessionReceiver<ServiceBusReceivedMessageWithLock>>;
acceptSession(topicName: string, subscriptionName: string, sessionId: string, options: AcceptSessionOptions<"receiveAndDelete">): Promise<ServiceBusSessionReceiver<ServiceBusReceivedMessage>>;
acceptNextSession(queueName: string, options?: AcceptSessionOptions<"peekLock">): Promise<ServiceBusSessionReceiver>;
acceptNextSession(queueName: string, options: AcceptSessionOptions<"receiveAndDelete">): Promise<ServiceBusSessionReceiver>;
acceptNextSession(topicName: string, subscriptionName: string, options?: AcceptSessionOptions<"peekLock">): Promise<ServiceBusSessionReceiver>;
acceptNextSession(topicName: string, subscriptionName: string, options: AcceptSessionOptions<"receiveAndDelete">): Promise<ServiceBusSessionReceiver>;
acceptSession(queueName: string, sessionId: string, options?: AcceptSessionOptions<"peekLock">): Promise<ServiceBusSessionReceiver>;
acceptSession(queueName: string, sessionId: string, options: AcceptSessionOptions<"receiveAndDelete">): Promise<ServiceBusSessionReceiver>;
acceptSession(topicName: string, subscriptionName: string, sessionId: string, options?: AcceptSessionOptions<"peekLock">): Promise<ServiceBusSessionReceiver>;
acceptSession(topicName: string, subscriptionName: string, sessionId: string, options: AcceptSessionOptions<"receiveAndDelete">): Promise<ServiceBusSessionReceiver>;
close(): Promise<void>;
createReceiver(queueName: string, options?: CreateReceiverOptions<"peekLock">): ServiceBusReceiver<ServiceBusReceivedMessageWithLock>;
createReceiver(queueName: string, options: CreateReceiverOptions<"receiveAndDelete">): ServiceBusReceiver<ServiceBusReceivedMessage>;
createReceiver(topicName: string, subscriptionName: string, options?: CreateReceiverOptions<"peekLock">): ServiceBusReceiver<ServiceBusReceivedMessageWithLock>;
createReceiver(topicName: string, subscriptionName: string, options: CreateReceiverOptions<"receiveAndDelete">): ServiceBusReceiver<ServiceBusReceivedMessage>;
createReceiver(queueName: string, options?: CreateReceiverOptions<"peekLock">): ServiceBusReceiver;
Copy link
Copy Markdown
Contributor

@HarshaNalluru HarshaNalluru Oct 21, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we aren't returning different types, there is no point in using the generics.
It could just be CreateReceiverOptions with an attribute "receiveMode".

Copy link
Copy Markdown
Contributor

@HarshaNalluru HarshaNalluru Oct 21, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

same for acceptSession and any other generics involving receiveMode

Copy link
Copy Markdown
Contributor Author

@ramya-rao-a ramya-rao-a Oct 21, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please see the note in the PR description on this topic

I have retained the overloads for createReceiver and createSessionReceiver methods even though there is no difference between the overloads anymore. This is to facilitate the next step of ensuring that users in ReceiveAndDelete mode do not see the settlement methods or the renewMessageLock method.

Copy link
Copy Markdown
Contributor

@HarshaNalluru HarshaNalluru Oct 21, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice!

createReceiver(queueName: string, options: CreateReceiverOptions<"receiveAndDelete">): ServiceBusReceiver;
createReceiver(topicName: string, subscriptionName: string, options?: CreateReceiverOptions<"peekLock">): ServiceBusReceiver;
createReceiver(topicName: string, subscriptionName: string, options: CreateReceiverOptions<"receiveAndDelete">): ServiceBusReceiver;
createSender(queueOrTopicName: string): ServiceBusSender;
fullyQualifiedNamespace: string;
}
Expand Down Expand Up @@ -431,31 +431,27 @@ export interface ServiceBusReceivedMessage extends ServiceBusMessage {
}

// @public
export interface ServiceBusReceivedMessageWithLock extends ServiceBusReceivedMessage {
abandon(propertiesToModify?: {
export interface ServiceBusReceiver {
abandonMessage(message: ServiceBusReceivedMessage, propertiesToModify?: {
[key: string]: any;
}): Promise<void>;
complete(): Promise<void>;
deadLetter(options?: DeadLetterOptions & {
close(): Promise<void>;
completeMessage(message: ServiceBusReceivedMessage): Promise<void>;
deadLetterMessage(message: ServiceBusReceivedMessage, options?: DeadLetterOptions & {
[key: string]: any;
}): Promise<void>;
defer(propertiesToModify?: {
deferMessage(message: ServiceBusReceivedMessage, propertiesToModify?: {
[key: string]: any;
}): Promise<void>;
renewLock(): Promise<Date>;
}

// @public
export interface ServiceBusReceiver<ReceivedMessageT> {
close(): Promise<void>;
entityPath: string;
getMessageIterator(options?: GetMessageIteratorOptions): AsyncIterableIterator<ReceivedMessageT>;
getMessageIterator(options?: GetMessageIteratorOptions): AsyncIterableIterator<ServiceBusReceivedMessage>;
isClosed: boolean;
peekMessages(maxMessageCount: number, options?: PeekMessagesOptions): Promise<ServiceBusReceivedMessage[]>;
receiveDeferredMessages(sequenceNumbers: Long | Long[], options?: OperationOptionsBase): Promise<ReceivedMessageT[]>;
receiveMessages(maxMessageCount: number, options?: ReceiveMessagesOptions): Promise<ReceivedMessageT[]>;
receiveDeferredMessages(sequenceNumbers: Long | Long[], options?: OperationOptionsBase): Promise<ServiceBusReceivedMessage[]>;
receiveMessages(maxMessageCount: number, options?: ReceiveMessagesOptions): Promise<ServiceBusReceivedMessage[]>;
receiveMode: "peekLock" | "receiveAndDelete";
subscribe(handlers: MessageHandlers<ReceivedMessageT>, options?: SubscribeOptions): {
renewMessageLock(message: ServiceBusReceivedMessage): Promise<Date>;
subscribe(handlers: MessageHandlers, options?: SubscribeOptions): {
close(): Promise<void>;
};
}
Expand All @@ -473,13 +469,13 @@ export interface ServiceBusSender {
}

// @public
export interface ServiceBusSessionReceiver<ReceivedMessageT extends ServiceBusReceivedMessage | ServiceBusReceivedMessageWithLock> extends ServiceBusReceiver<ReceivedMessageT> {
export interface ServiceBusSessionReceiver extends ServiceBusReceiver {
getSessionState(options?: OperationOptionsBase): Promise<any>;
renewSessionLock(options?: OperationOptionsBase): Promise<Date>;
readonly sessionId: string;
readonly sessionLockedUntilUtc: Date;
setSessionState(state: any, options?: OperationOptionsBase): Promise<void>;
subscribe(handlers: MessageHandlers<ReceivedMessageT>, options?: SubscribeOptions): {
subscribe(handlers: MessageHandlers, options?: SubscribeOptions): {
close(): Promise<void>;
};
}
Expand Down
4 changes: 2 additions & 2 deletions sdk/servicebus/service-bus/src/diagnostics/instrumentServiceBusMessage.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,7 @@ export function createProcessingSpan(
// NOTE: the connectionConfig also has an entityPath property but that only
// represents the optional entityPath in their connection string which is NOT
// what we want for tracing.
receiver: Pick<ServiceBusReceiver<any>, "entityPath">,
receiver: Pick<ServiceBusReceiver, "entityPath">,
connectionConfig: Pick<ConnectionContext["config"], "host">,
options?: OperationOptionsBase
): Span {
Expand Down Expand Up @@ -145,7 +145,7 @@ export function createProcessingSpan(
*/
export function createAndEndProcessingSpan(
receivedMessages: ServiceBusReceivedMessage | ServiceBusReceivedMessage[],
receiver: Pick<ServiceBusReceiver<any>, "entityPath">,
receiver: Pick<ServiceBusReceiver, "entityPath">,
connectionConfig: Pick<ConnectionContext["config"], "host">,
options?: OperationOptionsBase
): void {
Expand Down
1 change: 0 additions & 1 deletion sdk/servicebus/service-bus/src/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,6 @@ export {
AmqpMessageProperties,
DeadLetterOptions,
ServiceBusReceivedMessage,
ServiceBusReceivedMessageWithLock,
ServiceBusMessage
} from "./serviceBusMessage";
export { ServiceBusMessageBatch } from "./serviceBusMessageBatch";
Expand Down
8 changes: 4 additions & 4 deletions sdk/servicebus/service-bus/src/models.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@

import { OperationOptionsBase } from "./modelsToBeSharedWithEventHubs";
import Long from "long";
import { ServiceBusReceivedMessage } from "./serviceBusMessage";
import { MessagingError } from "@azure/core-amqp";

/**
Expand Down Expand Up @@ -36,13 +37,13 @@ export interface ProcessErrorArgs {
/**
* The general message handler interface (used for streamMessages).
*/
export interface MessageHandlers<ReceivedMessageT> {
export interface MessageHandlers {
/**
* Handler that processes messages from service bus.
*
* @param message A message received from Service Bus.
*/
processMessage(message: ReceivedMessageT): Promise<void>;
processMessage(message: ServiceBusReceivedMessage): Promise<void>;
/**
* Handler that processes errors that occur during receiving.
* @param args The error and additional context to indicate where
Expand All @@ -55,8 +56,7 @@ export interface MessageHandlers<ReceivedMessageT> {
* @internal
* @ignore
*/
export interface InternalMessageHandlers<ReceivedMessageT>
extends MessageHandlers<ReceivedMessageT> {
export interface InternalMessageHandlers extends MessageHandlers {
/**
* Called when the connection is initialized but before we subscribe to messages or add credits.
*
Expand Down
Loading