Add cancellation support to all async operations

lib/awaitableSender.ts

@@ -12,6 +12,7 @@ import { Session } from "./session";
 import {
  OperationTimeoutError, InsufficientCreditError, SendOperationFailedError
 } from "./errorDefinitions";
+import { AbortSignalLike, createAbortError } from "./util/utils";
 
 /**
  * Describes the interface for the send operation Promise which contains a reference to resolve,
@@ -170,9 +171,12 @@ export class AwaitableSender extends BaseSender {
  * @param {number} [format] The message format. Specify this if a message with custom format needs
  * to be sent. `0` implies the standard AMQP 1.0 defined format. If no value is provided, then the
  * given message is assumed to be of type Message interface and encoded appropriately.
+ * @param {AbortSignalLike} abortSignal A signal to cancel the send operation. This does not
+ * guarantee that the message will not be sent. It only stops listening for an acknowledgement from
+ * the remote endpoint.
  * @returns {Promise<Delivery>} Promise<Delivery> The delivery information about the sent message.
  */
- send(msg: Message | Buffer, tag?: Buffer | string, format?: number): Promise<Delivery> {
+ send(msg: Message | Buffer, tag?: Buffer | string, format?: number, abortSignal?: AbortSignalLike): Promise<Delivery> {
  return new Promise<Delivery>((resolve, reject) => {
  log.sender("[%s] Sender '%s' on amqp session '%s', credit: %d available: %d",
  this.connection.id, this.name, this.session.id, this.credit,
@@ -187,12 +191,44 @@ export class AwaitableSender extends BaseSender {
  return reject(new OperationTimeoutError(message));
  }, this.sendTimeoutInSeconds * 1000);
 
+ const onAbort = () => {
+ const promise = this.deliveryDispositionMap.get(delivery.id) as PromiseLike;
+ clearTimeout(promise.timer);
+ const deleteResult = this.deliveryDispositionMap.delete(delivery.id);
+ log.sender(
+ "[%s] Event: 'abort', Successfully deleted the delivery with id %d from the " +
+ " map of sender '%s' on amqp session '%s' and cleared the timer: %s.",
+ this.connection.id, delivery.id, this.name, this.session.id, deleteResult
+ );
+ const err = createAbortError("Send request has been cancelled.");
+ log.error("[%s] %s", this.connection.id, err.message);
+ promise.reject(err);
+ };
+
+ const removeAbortListener = () => {
+ abortSignal?.removeEventListener("abort", onAbort);
+ };
+
  const delivery = (this._link as RheaSender).send(msg, tag, format);
  this.deliveryDispositionMap.set(delivery.id, {
- resolve: resolve,
- reject: reject,
+ resolve: () => {
+ resolve();
+ removeAbortListener();
+ },
+ reject: (reason?: any) => {
+ reject(reason);
+ removeAbortListener();
+ },
  timer: timer
  });
+
+ if (abortSignal) {
+ if (abortSignal.aborted) {
+ onAbort();
+ } else {
+ abortSignal.addEventListener("abort", onAbort);
+ }
+ }
  } else {
  // Please send the message after some time.
  const msg =

lib/connection.ts

@@ -9,7 +9,7 @@ import { Sender, SenderOptions } from "./sender";
 import { Receiver, ReceiverOptions } from "./receiver";
 import { Container } from "./container";
 import { defaultOperationTimeoutInSeconds } from "./util/constants";
-import { Func, EmitParameters, emitEvent } from "./util/utils";
+import { Func, EmitParameters, emitEvent, AbortSignalLike, createAbortError } from "./util/utils";
 import {
  ConnectionEvents, SessionEvents, SenderEvents, ReceiverEvents, create_connection, websocket_connect,
  ConnectionOptions as RheaConnectionOptions, Connection as RheaConnection, AmqpError, Dictionary,
@@ -48,6 +48,36 @@ export interface ReceiverOptionsWithSession extends ReceiverOptions {
  session?: Session;
 }
 
+/**
+ * Set of options to use when running Connection.open()
+ */
+export interface ConnectionOpenOptions {
+ /**
+ * A signal used to cancel the Connection.open() operation.
+ */
+ abortSignal?: AbortSignalLike;
+}
+
+/**
+ * Set of options to use when running Connection.close()
+ */
+export interface ConnectionCloseOptions {
+ /**
+ * A signal used to cancel the Connection.close() operation.
+ */
+ abortSignal?: AbortSignalLike;
+}
+
+/**
+ * Set of options to use when running Connection.createSession()
+ */
+export interface SessionCreateOptions {
+ /**
+ * A signal used to cancel the Connection.createSession() operation.
+ */
+ abortSignal?: AbortSignalLike;
+}
+
 /**
  * Describes the options that can be provided while creating an AMQP connection.
  * @interface ConnectionOptions
@@ -264,17 +294,20 @@ export class Connection extends Entity {
 
  /**
  * Creates a new amqp connection.
+ * @param options A set of options including a signal used to cancel the operation.
  * @return {Promise<Connection>} Promise<Connection>
  * - **Resolves** the promise with the Connection object when rhea emits the "connection_open" event.
  * - **Rejects** the promise with an AmqpError when rhea emits the "connection_close" event
- * while trying to establish an amqp connection.
+ * while trying to establish an amqp connection or with an AbortError if the operation was cancelled.
  */
- open(): Promise<Connection> {
+ open(options?: ConnectionOpenOptions): Promise<Connection> {
  return new Promise((resolve, reject) => {
  if (!this.isOpen()) {
 
  let onOpen: Func<RheaEventContext, void>;
  let onClose: Func<RheaEventContext, void>;
+ let onAbort: Func<void, void>;
+ const abortSignal = options?.abortSignal;
  let waitTimer: any;
 
  const removeListeners: Function = () => {
@@ -283,6 +316,7 @@ export class Connection extends Entity {
  this._connection.removeListener(ConnectionEvents.connectionOpen, onOpen);
  this._connection.removeListener(ConnectionEvents.connectionClose, onClose);
  this._connection.removeListener(ConnectionEvents.disconnected, onClose);
+ abortSignal?.removeEventListener("abort", onAbort);
  };
 
  onOpen = (context: RheaEventContext) => {
@@ -299,6 +333,14 @@ export class Connection extends Entity {
  return reject(err);
  };
 
+ onAbort = () => {
+ removeListeners();
+ this._connection.close();
+ const err = createAbortError("Connection open request has been cancelled.");
+ log.error("[%s] [%s]", this.id, err.message);
+ return reject(err);
+ };
+
  const actionAfterTimeout = () => {
  removeListeners();
  const msg: string = `Unable to open the amqp connection "${this.id}" due to operation timeout.`;
@@ -314,6 +356,14 @@ export class Connection extends Entity {
  log.connection("[%s] Trying to create a new amqp connection.", this.id);
  this._connection.connect();
  this.actionInitiated++;
+
+ if (abortSignal) {
+ if (abortSignal.aborted) {
+ onAbort();
+ } else {
+ abortSignal.addEventListener("abort", onAbort);
+ }
+ }
  } else {
  return resolve(this);
  }
@@ -323,25 +373,33 @@ export class Connection extends Entity {
 
  /**
  * Closes the amqp connection.
+ * @param options A set of options including a signal used to cancel the operation.
+ * When the abort signal in the options is fired, the local endpoint is closed.
+ * This does not guarantee that the remote has closed as well. It only stops listening for
+ * an acknowledgement that the remote endpoint is closed as well.
  * @return {Promise<void>} Promise<void>
  * - **Resolves** the promise when rhea emits the "connection_close" event.
  * - **Rejects** the promise with an AmqpError when rhea emits the "connection_error" event while
- * trying to close an amqp connection.
+ * trying to close an amqp connection or with an AbortError if the operation was cancelled.
  */
- close(): Promise<void> {
+ close(options?: ConnectionCloseOptions): Promise<void> {
  return new Promise<void>((resolve, reject) => {
  log.error("[%s] The connection is open ? -> %s", this.id, this.isOpen());
  if (this.isOpen()) {
  let onClose: Func<RheaEventContext, void>;
  let onError: Func<RheaEventContext, void>;
  let onDisconnected: Func<RheaEventContext, void>;
+ let onAbort: Func<void, void>;
+ const abortSignal = options?.abortSignal;
  let waitTimer: any;
+
  const removeListeners = () => {
  clearTimeout(waitTimer);
  this.actionInitiated--;
  this._connection.removeListener(ConnectionEvents.connectionError, onError);
  this._connection.removeListener(ConnectionEvents.connectionClose, onClose);
  this._connection.removeListener(ConnectionEvents.disconnected, onDisconnected);
+ abortSignal?.removeEventListener("abort", onAbort);
  };
 
  onClose = (context: RheaEventContext) => {
@@ -366,6 +424,13 @@ export class Connection extends Entity {
  log.error("[%s] Connection got disconnected while closing itself: %O.", this.id, error);
  };
 
+ onAbort = () => {
+ removeListeners();
+ const err = createAbortError("Connection close request has been cancelled.");
+ log.error("[%s] [%s]", this.id, err.message);
+ return reject(err);
+ };
+
  const actionAfterTimeout = () => {
  removeListeners();
  const msg: string = `Unable to close the amqp connection "${this.id}" due to operation timeout.`;
@@ -380,6 +445,14 @@ export class Connection extends Entity {
  waitTimer = setTimeout(actionAfterTimeout, this.options!.operationTimeoutInSeconds! * 1000);
  this._connection.close();
  this.actionInitiated++;
+
+ if (abortSignal) {
+ if (abortSignal.aborted) {
+ onAbort();
+ } else {
+ abortSignal.addEventListener("abort", onAbort);
+ }
+ }
  } else {
  return resolve();
  }
@@ -452,19 +525,22 @@ export class Connection extends Entity {
 
  /**
  * Creates an amqp session on the provided amqp connection.
+ * @param options A set of options including a signal used to cancel the operation.
  * @return {Promise<Session>} Promise<Session>
  * - **Resolves** the promise with the Session object when rhea emits the "session_open" event.
  * - **Rejects** the promise with an AmqpError when rhea emits the "session_close" event while
- * trying to create an amqp session.
+ * trying to create an amqp session or with an AbortError if the operation was cancelled.
  */
- createSession(): Promise<Session> {
+ createSession(options?: SessionCreateOptions): Promise<Session> {
  return new Promise((resolve, reject) => {
  const rheaSession = this._connection.create_session();
  const session = new Session(this, rheaSession);
  session.actionInitiated++;
  let onOpen: Func<RheaEventContext, void>;
  let onClose: Func<RheaEventContext, void>;
  let onDisconnected: Func<RheaEventContext, void>;
+ let onAbort: Func<void, void>;
+ const abortSignal = options?.abortSignal;
  let waitTimer: any;
 
  const removeListeners = () => {
@@ -473,6 +549,7 @@ export class Connection extends Entity {
  rheaSession.removeListener(SessionEvents.sessionOpen, onOpen);
  rheaSession.removeListener(SessionEvents.sessionClose, onClose);
  rheaSession.connection.removeListener(ConnectionEvents.disconnected, onDisconnected);
+ abortSignal?.removeEventListener("abort", onAbort);
  };
 
  onOpen = (context: RheaEventContext) => {
@@ -498,6 +575,14 @@ export class Connection extends Entity {
  return reject(error);
  };
 
+ onAbort = () => {
+ removeListeners();
+ rheaSession.close();
+ const err = createAbortError("Create session request has been cancelled.");
+ log.error("[%s] [%s]", this.id, err.message);
+ return reject(err);
+ };
+
  const actionAfterTimeout = () => {
  removeListeners();
  const msg: string = `Unable to create the amqp session due to operation timeout.`;
@@ -512,19 +597,30 @@ export class Connection extends Entity {
  log.session("[%s] Calling amqp session.begin().", this.id);
  waitTimer = setTimeout(actionAfterTimeout, this.options!.operationTimeoutInSeconds! * 1000);
  rheaSession.begin();
+
+ if (abortSignal) {
+ if (abortSignal.aborted) {
+ onAbort();
+ } else {
+ abortSignal.addEventListener("abort", onAbort);
+ }
+ }
  });
  }
 
  /**
  * Creates an amqp sender link. It either uses the provided session or creates a new one.
+ * - **Resolves** the promise with the Sender object when rhea emits the "sender_open" event.
+ * - **Rejects** the promise with an AmqpError when rhea emits the "sender_close" event while
+ * trying to create an amqp session or with an AbortError if the operation was cancelled.
  * @param {SenderOptionsWithSession} options Optional parameters to create a sender link.
  * @return {Promise<Sender>} Promise<Sender>.
  */
- async createSender(options?: SenderOptionsWithSession): Promise<Sender> {
+ async createSender(options?: SenderOptionsWithSession & { abortSignal?: AbortSignalLike; }): Promise<Sender> {
  if (options && options.session && options.session.createSender) {
  return options.session.createSender(options);
  }
- const session = await this.createSession();
+ const session = await this.createSession({ abortSignal: options?.abortSignal });
  return session.createSender(options);
  }
 
@@ -540,24 +636,27 @@ export class Connection extends Entity {
  *
  * @return Promise<AwaitableSender>.
  */
- async createAwaitableSender(options?: AwaitableSenderOptionsWithSession): Promise<AwaitableSender> {
+ async createAwaitableSender(options?: AwaitableSenderOptionsWithSession & { abortSignal?: AbortSignalLike; }): Promise<AwaitableSender> {
  if (options && options.session && options.session.createAwaitableSender) {
  return options.session.createAwaitableSender(options);
  }
- const session = await this.createSession();
+ const session = await this.createSession({ abortSignal: options?.abortSignal });
  return session.createAwaitableSender(options);
  }
 
  /**
  * Creates an amqp receiver link. It either uses the provided session or creates a new one.
+ * - **Resolves** the promise with the Sender object when rhea emits the "receiver_open" event.
+ * - **Rejects** the promise with an AmqpError when rhea emits the "receiver_close" event while
+ * trying to create an amqp session or with an AbortError if the operation was cancelled.
  * @param {ReceiverOptionsWithSession} options Optional parameters to create a receiver link.
  * @return {Promise<Receiver>} Promise<Receiver>.
  */
- async createReceiver(options?: ReceiverOptionsWithSession): Promise<Receiver> {
+ async createReceiver(options?: ReceiverOptionsWithSession & { abortSignal?: AbortSignalLike; }): Promise<Receiver> {
  if (options && options.session && options.session.createReceiver) {
  return options.session.createReceiver(options);
  }
- const session = await this.createSession();
+ const session = await this.createSession({ abortSignal: options?.abortSignal });
  return session.createReceiver(options);
  }
 
@@ -572,17 +671,17 @@ export class Connection extends Entity {
  * @return {Promise<ReqResLink>} Promise<ReqResLink>
  */
  async createRequestResponseLink(senderOptions: SenderOptions, receiverOptions: ReceiverOptions,
- providedSession?: Session): Promise<ReqResLink> {
+ providedSession?: Session, abortSignal?: AbortSignal): Promise<ReqResLink> {
  if (!senderOptions) {
  throw new Error(`Please provide sender options.`);
  }
  if (!receiverOptions) {
  throw new Error(`Please provide receiver options.`);
  }
- const session = providedSession || await this.createSession();
+ const session = providedSession || await this.createSession({ abortSignal });
  const [sender, receiver] = await Promise.all([
- session.createSender(senderOptions),
- session.createReceiver(receiverOptions)
+ session.createSender({ ...senderOptions, abortSignal }),
+ session.createReceiver({ ...receiverOptions, abortSignal })
  ]);
  log.connection("[%s] Successfully created the sender '%s' and receiver '%s' on the same " +
  "amqp session '%s'.", this.id, sender.name, receiver.name, session.id);