Connecting to non AMQP server should fail (#78)

carlhoerberg · web-flow · commit 7a59414d5e31 · 2023-08-08T09:45:29.000+02:00
* Set timeout between connect and connected
* keep timeout if heartbeat is enabled
* enable tcp keepalive if heartbeats are disabled
diff --git a/src/amqp-socket-client.ts b/src/amqp-socket-client.ts
@@ -53,9 +53,18 @@ export class AMQPClient extends AMQPBaseClient {
       writable: true,
       enumerable: false // hide it from console.log etc.
     })
+    // enable socket read timeout during connection establishment
+    socket.setTimeout((this.heartbeat || 60) * 1000)
+    // enable TCP keepalive if AMQP heartbeats are disabled
+    if (this.heartbeat === 0) socket.setKeepAlive(true, 60)
     return new Promise((resolve, reject) => {
+      socket.on('timeout', () => reject(new AMQPError("timeout", this)))
       socket.on('error', (err) => reject(new AMQPError(err.message, this)))
-      this.connectPromise = [resolve, reject]
+      const onConnect = (conn : AMQPBaseClient) => {
+        socket.setTimeout(this.heartbeat * 1000) // reset timeout if heartbeats are disabled
+        resolve(conn)
+      }
+      this.connectPromise = [onConnect, reject]
     })
   }
 
diff --git a/test-browser/websocket.ts b/test-browser/websocket.ts
@@ -663,3 +663,8 @@ test("raises when channelMax is reached", async () => {
   const ch1 = await conn.channel(1)
   await expect(ch1.basicQos(10)).resolves.toBeUndefined()
 }, 20_000)
+
+test('should fail to connect to an AMQP port', async () => {
+  const amqp = new AMQPWebSocketClient("ws://127.0.0.1:5672/ws/amqp")
+  await expect(amqp.connect()).rejects.toThrow()
+})
diff --git a/test/test.ts b/test/test.ts
@@ -673,3 +673,8 @@ test('can update-secret', async () => {
   const conn = await amqp.connect()
   await expect(conn.updateSecret("foobar", "no reason")).resolves.toBeUndefined()
 })
+
+test('should fail to connect to HTTP', async () => {
+  const amqp = new AMQPClient("amqp://127.0.0.1:15672?heartbeat=1")
+  await expect(amqp.connect()).rejects.toThrow()
+})
