reset the default session for a connection if that session is closed

Author: grs

lib/connection.js

@@ -82,6 +82,7 @@ function session_per_connection(conn) {
         'get_session' : function () {
             if (!ssn) {
                 ssn = conn.create_session();
+                ssn.observers.on('session_close', function () { ssn = null; });
                 ssn.begin();
             }
             return ssn;

lib/session.js

@@ -410,6 +410,7 @@ var Session = function (connection, local_channel) {
     this.links = {}; // map by name
     this.options = {};
     Object.defineProperty(this, 'error', { get:  function() { return this.remote.end ? this.remote.end.error : undefined; }});
+    this.observers = new EventEmitter();
 };
 Session.prototype = Object.create(EventEmitter.prototype);
 Session.prototype.constructor = Session;
@@ -436,6 +437,7 @@ Session.prototype._reconnect = function() {
 
 Session.prototype.dispatch = function(name) {
     log.events('[%s] Session got event: %s', this.connection.options.id, name);
+    EventEmitter.prototype.emit.apply(this.observers, arguments);
     if (this.listeners(name).length) {
         EventEmitter.prototype.emit.apply(this, arguments);
         return true;

test/links.ts

@@ -693,4 +693,32 @@ describe('miscellaneous', function() {
         });
         client.connect(listener.address() as any).open_receiver();
     });
+
+    it('handles session close', function(done: Function) {
+        var closed_session = false;
+        var outgoing = ['one', 'two'] as any;
+        var incoming = [] as any;
+        server.on('sender_open', function(context: rhea.EventContext) {
+            context.sender!.send({subject:outgoing.shift()} as rhea.Message);
+        });
+        server.once('accepted', function(context: rhea.EventContext) {
+            closed_session = true;
+            context.session!.close();
+        });
+        client.on('session_close', function (context: rhea.EventContext) {
+            context.connection.open_receiver();
+        });
+        client.on('message', function (context: rhea.EventContext) {
+            incoming.push(context.message!.subject);
+            if (incoming.length == 2) {
+                context.connection.close();
+            }
+        });
+        client.on('connection_close', function (context: rhea.EventContext) {
+            assert.deepEqual(incoming, ['one', 'two']);
+            done();
+        });
+        var conn = client.connect({port: (listener.address() as any).port, id: 'client'});
+        conn.open_receiver();
+    });
 });