[service-bus] Check abortSignal in SessionReceiver.subscribe()

sdk/servicebus/service-bus/review/service-bus.api.md

@@ -189,7 +189,7 @@ export interface MessageHandlerOptions extends MessageHandlerOptionsBase {
 }
 
 // @public
-export interface MessageHandlerOptionsBase {
+export interface MessageHandlerOptionsBase extends OperationOptionsBase {
     autoComplete?: boolean;
     maxConcurrentCalls?: number;
 }
@@ -459,7 +459,7 @@ export interface ServiceBusSessionReceiver<ReceivedMessageT extends ServiceBusRe
 }
 
 // @public
-export interface SessionSubscribeOptions extends OperationOptionsBase, MessageHandlerOptionsBase {
+export interface SessionSubscribeOptions extends MessageHandlerOptionsBase {
 }
 
 // @public
@@ -482,7 +482,7 @@ export interface SqlRuleFilter {
 export type SubQueue = "deadLetter" | "transferDeadLetter";
 
 // @public
-export interface SubscribeOptions extends OperationOptionsBase, MessageHandlerOptions {
+export interface SubscribeOptions extends MessageHandlerOptions {
 }
 
 // @public

sdk/servicebus/service-bus/src/models.ts

@@ -120,18 +120,18 @@ export interface GetMessageIteratorOptions extends OperationOptionsBase {}
 /**
  * Options used when subscribing to a Service Bus queue or subscription.
  */
-export interface SubscribeOptions extends OperationOptionsBase, MessageHandlerOptions {}
+export interface SubscribeOptions extends MessageHandlerOptions {}
 
 /**
  * Options used when subscribing to a Service Bus queue or subscription.
  */
-export interface SessionSubscribeOptions extends OperationOptionsBase, MessageHandlerOptionsBase {}
+export interface SessionSubscribeOptions extends MessageHandlerOptionsBase {}
 
 /**
  * Describes the options passed to `registerMessageHandler` method when receiving messages from a
  * Queue/Subscription.
  */
-export interface MessageHandlerOptionsBase {
+export interface MessageHandlerOptionsBase extends OperationOptionsBase {
   /**
    * @property Indicates whether the `complete()` method on the message should automatically be
    * called by the sdk after the user provided onMessage handler has been executed.

sdk/servicebus/service-bus/src/session/messageSession.ts

@@ -18,10 +18,14 @@ import { OnAmqpEventAsPromise, OnError, OnMessage } from "../core/messageReceive
 import { logger } from "../log";
 import { DispositionType, InternalReceiveMode, ServiceBusMessageImpl } from "../serviceBusMessage";
 import { logError, throwErrorIfConnectionClosed } from "../util/errors";
-import { calculateRenewAfterDuration, convertTicksToDate } from "../util/utils";
+import {
+  calculateRenewAfterDuration,
+  convertTicksToDate,
+  StandardAbortMessage
+} from "../util/utils";
 import { BatchingReceiverLite, MinimalReceiver } from "../core/batchingReceiver";
 import { onMessageSettled, DeferredPromiseAndTimer } from "../core/shared";
-import { AbortSignalLike } from "@azure/abort-controller";
+import { AbortError, AbortSignalLike } from "@azure/abort-controller";
 import { ReceiverHelper } from "../core/receiverHelper";
 import { CreateSessionReceiverOptions, MessageHandlerOptionsBase } from "../models";
 
@@ -618,6 +622,11 @@ export class MessageSession extends LinkEntity<Receiver> {
    */
   subscribe(onMessage: OnMessage, onError: OnError, options?: MessageHandlerOptionsBase): void {
     if (!options) options = {};
+
+    if (options.abortSignal?.aborted) {
+      throw new AbortError(StandardAbortMessage);
+    }
+
     this._isReceivingMessagesForSubscriber = true;
     if (typeof options.maxConcurrentCalls === "number" && options.maxConcurrentCalls > 0) {
       this.maxConcurrentCalls = options.maxConcurrentCalls;

sdk/servicebus/service-bus/test/internal/abortSignal.spec.ts

@@ -18,6 +18,9 @@ import {
 import { createConnectionContextForTests } from "./unittestUtils";
 import { StandardAbortMessage } from "../../src/util/utils";
 import { isLinkLocked } from "../utils/misc";
+import { ServiceBusSessionReceiverImpl } from "../../src/receivers/sessionReceiver";
+import { Constants } from "@azure/core-amqp";
+import { ServiceBusReceiverImpl } from "../../src/receivers/receiver";
 
 describe("AbortSignal", () => {
   const testMessageThatDoesntMatter = {
@@ -314,4 +317,89 @@ describe("AbortSignal", () => {
       assert.isFalse(isLinkLocked(messageReceiver));
     });
   });
+
+  describe("subscribe", () => {
+    /**
+     * SessionReceiver is a bit of an odd duck because it doesn't do initialization
+     * in its subscribe() call (like non-session receiver) so our normal abortSignal
+     * code isn't running there. So we have to check this separately from Receiver.
+     */
+    it("SessionReceiver.subscribe", async () => {
+      const session = await ServiceBusSessionReceiverImpl.createInitializedSessionReceiver(
+        createConnectionContextForTests({
+          onCreateReceiverCalled: (receiver) => {
+            (receiver as any).source = {
+              filter: {
+                [Constants.sessionFilterName]: "hello"
+              }
+            };
+
+            (receiver as any).properties = {
+              ["com.microsoft:locked-until-utc"]: Date.now()
+            };
+          }
+        }),
+        "entityPath",
+        "peekLock",
+        {
+          sessionId: "hello"
+        }
+      );
+
+      try {
+        const abortSignal = createAbortSignalForTest(true);
+        const receivedErrors: Error[] = [];
+
+        session.subscribe(
+          {
+            processMessage: async (_msg) => {},
+            processError: async (err) => {
+              receivedErrors.push(err);
+            }
+          },
+          {
+            abortSignal
+          }
+        );
+
+        assert.equal(receivedErrors[0].message, "The operation was aborted.");
+        assert.equal(receivedErrors[0].name, "AbortError");
+      } finally {
+        await session.close();
+      }
+    });
+
+    it("Receiver.subscribe", async () => {
+      const receiver = new ServiceBusReceiverImpl(
+        createConnectionContextForTests(),
+        "entityPath",
+        "peekLock"
+      );
+
+      try {
+        const abortSignal = createAbortSignalForTest(true);
+        const receivedErrors: Error[] = [];
+
+        await new Promise<void>((resolve) => {
+          receiver.subscribe(
+            {
+              processMessage: async (_msg: any) => {},
+              processError: async (err: Error) => {
+                resolve();
+                receivedErrors.push(err);
+              }
+            },
+            {
+              abortSignal
+            }
+          );
+        });
+
+        assert.equal(receivedErrors[0].message, "The operation was aborted.");
+        assert.equal(receivedErrors[0].name, "AbortError");
+      } finally {
+        await receiver.close();
+      }
+    });
+  });
 });