Merge pull request #149 from uProxy/trevj-delay-datachannel-closures

have DataChannel block close() call until all messages have been sent

Author: trevj

Gruntfile.coffee

@@ -53,7 +53,8 @@ taskManager.add 'browserify_specs', [
   'browserify:buildToolsTaskmanagerSpec'
   'browserify:loggingSpec'
   'browserify:loggingProviderSpec'
-  'browserify:webrtcSpec'
+  'browserify:peerconnectionSpec'
+  'browserify:datachannelSpec'
   'browserify:queueSpec'
 ]
 
@@ -215,7 +216,8 @@ module.exports = (grunt) ->
       handlerSpec: Rule.browserifySpec 'handler/queue'
       loggingProviderSpec: Rule.browserifySpec 'loggingprovider/loggingprovider'
       loggingSpec: Rule.browserifySpec 'logging/logging'
-      webrtcSpec: Rule.browserifySpec 'webrtc/peerconnection'
+      peerconnectionSpec: Rule.browserifySpec 'webrtc/peerconnection'
+      datachannelSpec: Rule.browserifySpec 'webrtc/datachannel'
       queueSpec: Rule.browserifySpec 'queue/queue'
       # Browserify sample apps main freedom module and core environments
       copypasteFreedomChatFreedomModule: Rule.browserify 'samples/copypaste-freedom-chat/freedom-module'

package.json

@@ -1,7 +1,7 @@
 {
   "name": "uproxy-lib",
   "description": "Shared libraries for uProxy projects.",
-  "version": "21.0.0",
+  "version": "22.0.0",
   "repository": {
     "type": "git",
     "url": "https://github.com/uProxy/uproxy-lib"

src/webrtc/datachannel.spec.ts

@@ -0,0 +1,63 @@
+/// <reference path='../../../third_party/typings/es6-promise/es6-promise.d.ts' />
+/// <reference path='../../../third_party/typings/jasmine/jasmine.d.ts' />
+/// <reference path="../../../third_party/freedom-typings/rtcpeerconnection.d.ts" />
+
+import MockFreedomRtcDataChannel = require('../freedom/mocks/mock-rtcdatachannel');
+
+import freedomMocker = require('../freedom/mocks/mock-freedom-in-module-env');
+freedom = freedomMocker.makeMockFreedomInModuleEnv({
+  'rtcdatachannel': () => { return new MockFreedomRtcDataChannel(); }
+});
+
+import datachannel = require('./datachannel');
+
+describe('DataChannel', function() {
+  var mockRtcDataChannel :MockFreedomRtcDataChannel;
+
+  beforeEach(function() {
+    mockRtcDataChannel = new MockFreedomRtcDataChannel();
+  });
+
+  // Ensure that close() waits for all outgoing unsent messages to be sent:
+  //   https://github.com/uProxy/uproxy/issues/1218
+  it('close waits for all data to be sent', (done) => {
+    // The core.rtcdatachannel is initially open.
+    spyOn(mockRtcDataChannel, 'getReadyState').and.returnValue(
+        Promise.resolve('open'));
+
+    // Have core.rtcdatachannel emit an onclose event when its
+    // close() method is invoked.
+    var closeSpy = spyOn(mockRtcDataChannel, 'close').and.callFake(() => {
+      mockRtcDataChannel.handleEvent('onclose');
+    });
+
+    // Pretend the core.rtcdatachannel is buffering a lot of data.
+    // This will cause DataChannel to buffer sends.
+    var getBufferedAmountSpy = spyOn(mockRtcDataChannel, 'getBufferedAmount');
+    getBufferedAmountSpy.and.callFake(() => {
+      return Promise.resolve(datachannel.PC_QUEUE_LIMIT);
+    });
+
+    var sendBufferSpy = spyOn(mockRtcDataChannel, 'sendBuffer');
+
+    datachannel.createFromRtcDataChannel(mockRtcDataChannel).then(
+        (channel:datachannel.DataChannel) => {
+      channel.send({
+        buffer: new Uint8Array([0, 1, 2]).buffer
+      });
+
+      channel.close();
+
+      // At this point, our message should not yet have been sent.
+      // Now that we've requested the channel be closed, "unblock" the
+      // core.rtcdatachannel to drain the message queue.
+      expect(sendBufferSpy).not.toHaveBeenCalled();
+      getBufferedAmountSpy.and.returnValue(Promise.resolve(0));
+
+      channel.onceClosed.then(() => {
+        expect(sendBufferSpy).toHaveBeenCalled();
+        done();
+      });
+    });
+  });
+});

src/webrtc/datachannel.ts

@@ -25,7 +25,7 @@ var CHUNK_SIZE = 1024 * 15;
 // the 16MB for Chrome 37+ and "100 messages" of previous versions mentioned):
 //   https://code.google.com/p/webrtc/issues/detail?id=2866
 // CONSIDER: make it 0. There is no size in the spec.
-var PC_QUEUE_LIMIT = 1024 * 250;
+export var PC_QUEUE_LIMIT = 1024 * 250;
 // Javascript has trouble representing integers larger than 2^53. So we simply
 // don't support trying to send array's bigger than that.
 var MAX_MESSAGE_SIZE = Math.pow(2, 53);
@@ -70,9 +70,9 @@ export interface DataChannel {
   // cleared.  There can only be one listener at a time.  Pass null to unset.
   setOverflowListener(listener:(overflow:boolean) => void) : void;
 
-  // Closes this data channel.
+  // Closes this data channel once all outgoing messages have been sent.
   // A channel cannot be re-opened once this has been called.
-  close() : void;
+  close() : Promise<void>;
 
   toString() : string;
 }
@@ -109,6 +109,13 @@ export class DataChannelClass implements DataChannel {
   public onceOpened      :Promise<void>;
   public onceClosed      :Promise<void>;
 
+  // True iff close() has been called.
+  private draining_ = false;
+  private fulfillDrained_ :() => void;
+  private onceDrained_ = new Promise((F, R) => {
+    this.fulfillDrained_ = F;
+  });
+
   private opennedSuccessfully_ :boolean;
   private rejectOpened_  :(e:Error) => void;
 
@@ -119,10 +126,8 @@ export class DataChannelClass implements DataChannel {
 
   // |rtcDataChannel_| is the freedom rtc data channel.
   // |label_| is the rtcDataChannel_.getLabel() result
-  // |id| is the Freedom GUID for the underlying browser object.
   constructor(private rtcDataChannel_:freedom_RTCDataChannel.RTCDataChannel,
-              private label_:string,
-              id:string) {
+              private label_:string) {
     this.dataFromPeerQueue = new handler.Queue<Data,void>();
     this.toPeerDataQueue_ = new handler.Queue<Data,void>();
     this.toPeerDataBytes_ = 0;
@@ -153,7 +158,7 @@ export class DataChannelClass implements DataChannel {
     });
     this.onceOpened.then(() => {
       this.opennedSuccessfully_ = true;
-      this.toPeerDataQueue_.setNextHandler(this.handleSendDataToPeer_);
+      this.conjestionControlSendHandler();
     });
     this.onceClosed.then(() => {
         if(!this.opennedSuccessfully_) {
@@ -164,6 +169,10 @@ export class DataChannelClass implements DataChannel {
         }
         this.opennedSuccessfully_ = false;
       });
+    this.onceDrained_.then(() => {
+      log.debug('all messages sent, closing');
+      this.rtcDataChannel_.close();
+    });
   }
 
 
@@ -303,6 +312,9 @@ export class DataChannelClass implements DataChannel {
       } else {
         if (this.toPeerDataQueue_.getLength() === 0) {
           this.setOverflow_(false);
+          if (this.draining_) {
+            this.fulfillDrained_();
+          }
         }
         // This processes one block from the queue, which (in Chrome) is
         // expected to be no larger than 4 KB.  We will then go through the
@@ -316,8 +328,17 @@ export class DataChannelClass implements DataChannel {
     });
   }
 
-  public close = () : void => {
-    this.rtcDataChannel_.close();
+  public close = () : Promise<void> => {
+    log.debug('close requested (%1 messages to send)',
+        this.toPeerDataQueue_.getLength());
+
+    this.draining_ = true;
+
+    if (this.toPeerDataQueue_.getLength() === 0) {
+      this.fulfillDrained_();
+    }
+
+    return this.onceClosed;
   }
 
   public getBrowserBufferedAmount = () : Promise<number> => {
@@ -343,15 +364,19 @@ export class DataChannelClass implements DataChannel {
   }
 }  // class DataChannelClass
 
-
-// Static, async constructor that returns a DataChannel ready for use.
-// |id| is the GUID generated by Freedom to identify the underlying
-//     browser object.
+// Static constructor which constructs a core.rtcdatachannel instance
+// given a core.rtcdatachannel GUID.
 export function createFromFreedomId(id:string) : Promise<DataChannel> {
-  var rtcDataChannel = freedom['core.rtcdatachannel'](id);
+  return createFromRtcDataChannel(freedom['core.rtcdatachannel'](id));
+}
+
+// Static constructor which constructs a core.rtcdatachannel instance
+// given a core.rtcdatachannel instance.
+export function createFromRtcDataChannel(
+    rtcDataChannel:freedom_RTCDataChannel.RTCDataChannel) : Promise<DataChannel> {
   return rtcDataChannel.setBinaryType('arraybuffer').then(() => {
     return rtcDataChannel.getLabel().then((label:string) => {
-      return new DataChannelClass(rtcDataChannel, label, id);
+      return new DataChannelClass(rtcDataChannel, label);
     });
   });
 }

src/webrtc/peerconnection.spec.ts

@@ -17,7 +17,7 @@ freedom = freedomMocker.makeMockFreedomInModuleEnv({
 import PeerConnection = require('./peerconnection');
 import PeerConnectionClass = PeerConnection.PeerConnectionClass;
 
-describe('WebRtc / PeerConnection', function() {
+describe('PeerConnection', function() {
   var mockRtcPeerConnection :MockFreedomRtcPeerConnection;
 
   beforeEach(function() {