feat: manual reconnect

Useful when e.g. a consumer is cancelled.

src/AmqpConnectionManager.ts

@@ -245,6 +245,31 @@ export default class AmqpConnectionManager extends EventEmitter implements IAmqp
         return !!this._currentConnection;
     }
 
+    /** Force reconnect - noop unless connected */
+    reconnect(): void {
+        if (this._closed) {
+            throw new Error('cannot reconnect after close');
+        }
+
+        // If we have a connection, close it and immediately connect again.
+        // Wait for ordinary reconnect otherwise.
+        if (this._currentConnection) {
+            this._currentConnection.removeAllListeners();
+            this._currentConnection
+                .close()
+                .catch(() => {
+                    // noop
+                })
+                .then(() => {
+                    this._currentConnection = undefined;
+                    this._disconnectSent = true;
+                    this.emit('disconnect', { err: new Error('forced reconnect') });
+                    return this._connect();
+                })
+                .catch(neverThrows);
+        }
+    }
+
     /** The current connection. */
     get connection(): Connection | undefined {
         return this._currentConnection;

test/AmqpConnectionManagerTest.ts

@@ -1,6 +1,7 @@
 import origAmpq from 'amqplib';
 import chai from 'chai';
 import chaiString from 'chai-string';
+import { once } from 'events';
 import * as promiseTools from 'promise-tools';
 import AmqpConnectionManager from '../src/AmqpConnectionManager';
 import { FakeAmqp, FakeConnection } from './fixtures';
@@ -324,6 +325,22 @@ describe('AmqpConnectionManager', function () {
             );
         }));
 
+    it('should be able to manually reconnect', async () => {
+        amqp = new AmqpConnectionManager('amqp://localhost');
+        await once(amqp, 'connect');
+
+        amqp.reconnect();
+        await once(amqp, 'disconnect');
+        await once(amqp, 'connect');
+    });
+
+    it('should throw on manual reconnect after close', async () => {
+        amqp = new AmqpConnectionManager('amqp://localhost');
+        await once(amqp, 'connect');
+        await amqp.close()
+        expect(amqp.reconnect).to.throw()
+    })
+
     it('should create and clean up channel wrappers', async function () {
         amqp = new AmqpConnectionManager('amqp://localhost');
         const channel = amqp.createChannel({ name: 'test-chan' });