Cancel idle abortion timer

Receiving max idle time exceeded left abort_idle timer running when
connection was already closed causing possibly second disconnected event.

The CLOSE frame error was not cleared, but it was resent on reconnect after
the client had send the timeout error.

Author: kohtala

lib/connection.js

@@ -239,7 +239,7 @@ var Connection = function (options, container) {
     this.conn_established_counter = 0;
     this.heartbeat_out = undefined;
     this.heartbeat_in = undefined;
-    this.abort_idle = false;
+    this.abort_idle = undefined;
     this.socket_ready = false;
     this.scheduled_reconnect = undefined;
     this.default_sender = undefined;
@@ -274,7 +274,8 @@ Connection.prototype._disconnect = function() {
 
 Connection.prototype._reconnect = function() {
     if (this.abort_idle) {
-        this.abort_idle = false;
+        clearTimeout(this.abort_idle);
+        this.abort_idle = undefined;
         this.local.close.error = undefined;
         this.state = new EndpointState();
         this.state.open();
@@ -302,7 +303,10 @@ Connection.prototype._reset_remote_state = function() {
 
 Connection.prototype.connect = function () {
     this.is_server = false;
-    this.abort_idle = false;
+    if (this.abort_idle) {
+        clearTimeout(this.abort_idle);
+        this.abort_idle = undefined;
+    }
     this._reset_remote_state();
     this._connect(this.options.connection_details(this.conn_established_counter));
     this.open();
@@ -341,6 +345,7 @@ Connection.prototype.accept = function (socket) {
 
 Connection.prototype.abort_socket = function (socket) {
     if (socket === this.socket) {
+        this.abort_idle = undefined;
         log.io('[%s] aborting socket', this.options.id);
         this.socket.end();
         if (this.socket.removeAllListeners) {
@@ -575,11 +580,10 @@ Connection.prototype.input = function (buff) {
 
 Connection.prototype.idle = function () {
     if (!this.is_closed()) {
-        this.abort_idle = true;
         this.closed_with_non_fatal_error = true;
         this.local.close.error = {condition:'amqp:resource-limit-exceeded', description:'max idle time exceeded'};
         this.close();
-        setTimeout(this.abort_socket.bind(this, this.socket), 1000);
+        this.abort_idle = setTimeout(this.abort_socket.bind(this, this.socket), 1000);
     }
 };
 
@@ -602,6 +606,10 @@ Connection.prototype._disconnected = function (error) {
         clearTimeout(this.heartbeat_in);
         this.heartbeat_in = undefined;
     }
+    if (this.abort_idle) {
+        clearTimeout(this.abort_idle);
+        this.abort_idle = undefined;
+    }
     var was_closed_with_non_fatal_error = this.closed_with_non_fatal_error;
     if (this.closed_with_non_fatal_error) {
         this.closed_with_non_fatal_error = false;
@@ -756,6 +764,7 @@ Connection.prototype._write_open = function () {
 
 Connection.prototype._write_close = function () {
     this._write_frame(0, this.local.close);
+    this.local.close.error = undefined;
 };
 
 Connection.prototype.on_begin = function (frame) {

package.json

@@ -29,8 +29,7 @@
         "ts-node": "^10.4.0",
         "typescript": "^4.5.5",
         "uglify-js": "",
-        "ws": "^6.0.0",
-        "wtfnode": "^0.8.4"
+        "ws": "^6.0.0"
     },
     "scripts": {
         "lint": "eslint lib/*.js",

test/connections.ts

@@ -163,6 +163,7 @@ describe('connection error handling', function () {
         var error_handler_called: boolean;
         var close_handler_called: boolean;
         container.on('connection_open', function (context: rhea.EventContext) {
+            assert.strictEqual(context.connection.error, undefined);
             context.connection.close({ condition: 'amqp:connection:forced', description: 'testing error on close' });
         });
         container.on('connection_close', function (context: rhea.EventContext) {

test/idle.ts

@@ -0,0 +1,176 @@
+/*
+ * Copyright 2015 Red Hat Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import * as assert from "assert";
+import { AddressInfo, Server } from "net";
+import * as rhea from "../";
+
+describe('idle', function () {
+    var server: rhea.Container;
+    var listener: Server;
+
+    beforeEach(function() {
+    })
+
+    afterEach(function () {
+        listener.close();
+    });
+
+    describe('server has idle_time_out', function () {
+        beforeEach(function (done: Function) {
+            server = rhea.create_container({ non_fatal_errors: [] });
+            listener = server.listen({ port: 0, idle_time_out: 300 });
+            listener.on('listening', function () {
+                done();
+            });
+        });
+
+        // There are setups where server detects idle due to suspend while
+        // networking and all is well.
+        it('client does not send within idle_time_out', function (done: Function) {
+            var state = 0;
+            server.on('connection_open', function (context: rhea.EventContext) {
+                assert.strictEqual(++state, 1);
+            });
+            server.on('connection_error', function (context: rhea.EventContext) {
+                assert.fail('server must not receive error');
+            });
+            server.on('connection_close', function (context: rhea.EventContext) {
+                assert.strictEqual(++state, 4);
+            });
+            server.on('disconnected', function (context: rhea.EventContext) {
+                assert.strictEqual(context.reconnecting, undefined);
+                assert.strictEqual(++state, 5);
+                setTimeout(() => { done() }, 100);
+            });
+
+            var client: rhea.Container = rhea.create_container();
+            var conn = client.connect({
+                port: (listener.address() as AddressInfo).port
+            });
+            conn.on('connection_open', function (context: rhea.EventContext) {
+                const connection = context.connection;
+                assert.strictEqual(++state, 2);
+                assert.strictEqual(connection.remote.open.idle_time_out, 300);
+                // Disable idle timer
+                connection.remote.open.idle_time_out = 0;
+                if (connection.heartbeat_out) clearTimeout(connection.heartbeat_out);
+            });
+            conn.on('connection_error', function (context: rhea.EventContext) {
+                assert.strictEqual(++state, 3);
+                var error = context.connection.error;
+                assert.strictEqual((error as any).condition, 'amqp:resource-limit-exceeded');
+                assert.strictEqual((error as any).description, 'max idle time exceeded');
+            });
+            conn.on('disconnected', function (context: rhea.EventContext) {
+                assert.fail('disconnected shouldnt have been called');
+            });
+        });
+    });
+
+    describe('server has no idle_time_out', function () {
+        beforeEach(function (done: Function) {
+            server = rhea.create_container({ non_fatal_errors: [] });
+            listener = server.listen({ port: 0 });
+            listener.on('listening', function () {
+                done();
+            });
+        });
+
+        it('server does not send within idle_time_out', function (done: Function) {
+            var state = 0;
+            var server_opens = 0;
+            var server_closes = 0;
+            server.on('connection_open', function (context: rhea.EventContext) {
+                ++server_opens;
+                const connection = context.connection;
+                assert.strictEqual(connection.remote.open.idle_time_out, 500);
+                if (server_opens == 1) {
+                    // Disable idle timer
+                    assert.strictEqual(++state, 1);
+                    connection.remote.open.idle_time_out = 0;
+                    if (connection.heartbeat_out) clearTimeout(connection.heartbeat_out);
+                } else {
+                    // Reconnect after failure. Just close.
+                    assert.strictEqual(++state, 7);
+                    connection.close({ condition: 'time to end this test', description: 'well done' })
+                }
+            });
+            server.on('connection_error', function (context: rhea.EventContext) {
+                var error = context.connection.error;
+                assert.strictEqual((error as any).condition, 'amqp:resource-limit-exceeded');
+                assert.strictEqual((error as any).description, 'max idle time exceeded');
+                assert.strictEqual(++state, 3);
+            });
+            server.on('connection_close', function (context: rhea.EventContext) {
+                ++server_closes;
+                if (server_closes == 1) {
+                    assert.strictEqual(++state, 4);
+                } else {
+                    done();
+                }
+            });
+            server.on('disconnected', function (context: rhea.EventContext) {
+                assert.fail('disconnected shouldnt have been called');
+            });
+
+            var client: rhea.Container = rhea.create_container();
+            var conn = client.connect({
+                port: (listener.address() as AddressInfo).port,
+                idle_time_out: 500
+            });
+            var client_opens = 0;
+            var client_closes = 0;
+            var client_disconnects = 0;
+            conn.on('connection_open', function (context: rhea.EventContext) {
+                client_opens++;
+                assert.strictEqual(client_opens, server_opens);
+                if (client_opens === 2) {
+                    assert.strictEqual(++state, 8);
+                    conn.close();
+                } else {
+                    assert.strictEqual(++state, 2);
+                }
+            });
+            conn.on('connection_error', function (context: rhea.EventContext) {
+                var error = context.connection.error;
+                assert.strictEqual((error as any).condition, 'time to end this test');
+                assert.strictEqual(++state, 9);
+            });
+            conn.on('connection_close', function (context: rhea.EventContext) {
+                client_closes++;
+                if (client_closes == 1) {
+                    assert.strictEqual(client_closes, 1);
+                    assert.strictEqual(server_opens, 1);
+                    assert.strictEqual(client_opens, 1);
+                    assert.strictEqual(++state, 5);
+                } else {
+                    assert.strictEqual(client_closes, 2);
+                    assert.strictEqual(server_opens, 2);
+                    assert.strictEqual(client_opens, 2);
+                    assert.strictEqual(++state, 10);
+                }
+            });
+            conn.on('disconnected', function (context: rhea.EventContext) {
+                client_disconnects++;
+                assert.strictEqual(client_disconnects, 1);
+                assert.strictEqual(server_opens, 1);
+                assert.strictEqual(context.reconnecting, true);
+                assert.strictEqual(++state, 6);
+            });
+        });
+    });
+});