emit sendable event when capacity becomes available in outgoing session buffer

Author: grs

lib/session.js

@@ -533,11 +533,11 @@ Session.prototype.find_link = function (filter) {
 };
 
 Session.prototype.each_receiver = function (action, filter) {
-    this.each_link(util.receiver_filter(filter));
+    this.each_link(action, util.receiver_filter(filter));
 };
 
 Session.prototype.each_sender = function (action, filter) {
-    this.each_link(util.sender_filter(filter));
+    this.each_link(action, util.sender_filter(filter));
 };
 
 Session.prototype.each_link = function (action, filter) {
@@ -587,13 +587,25 @@ Session.prototype.is_closed = function () {
     return this.connection.is_closed() || this.is_itself_closed();
 };
 
+function notify_sendable(sender) {
+    sender.dispatch('sendable', sender._context());
+}
+
+function is_sender_sendable(sender) {
+    return sender.is_open() && sender.sendable();
+}
+
 Session.prototype._process = function () {
     do {
         if (this.state.need_open()) {
             this.output(this.local.begin);
         }
 
+        var was_blocked = this.outgoing.deliveries.available() === 0;
         this.outgoing.process();
+        if (was_blocked && this.outgoing.deliveries.available()) {
+            this.each_sender(notify_sendable, is_sender_sendable);
+        }
         this.incoming.process(this);
         for (var k in this.links) {
             this.links[k]._process();

test/links.ts

@@ -757,4 +757,38 @@ describe('miscellaneous', function() {
             done();
         });
     });
+
+    it('sends presettled messages when sendable', function(done: Function) {
+        server.options.credit_window = 0;
+        var outgoing = [] as any;
+        var incoming = [] as any;
+        var sent = 0;
+        var settled = 0;
+        for (var i = 0; i < 10000; i++) {
+            outgoing.push('message-' + i);
+        }
+        server.on('receiver_open', function (context: rhea.EventContext) {
+            context.receiver!.add_credit(10000);
+        });
+        server.on('message', function (context: rhea.EventContext) {
+            incoming.push(context.message!.body);
+            if (incoming.length === outgoing.length) {
+                context.receiver!.close();
+            }
+        });
+        var conn = client.connect(listener.address() as any);
+        var s = conn.open_sender({snd_settle_mode:1});
+        client.on('sendable', function (context: rhea.EventContext) {
+            while (context.sender!.sendable() && sent < outgoing.length) {
+                context.sender!.send({body:outgoing[sent++]});
+            }
+        });
+        client.on('sender_close', function (context: rhea.EventContext) {
+            context.connection.close();
+        });
+        client.on('connection_close', function (context: rhea.EventContext) {
+            assert.deepEqual(incoming, outgoing);
+            done();
+        });
+    });
 });