fix peerjs-js-binarypack library

fix large file sending

changelog.md

@@ -6,12 +6,15 @@ All notable changes to this project will be documented in this file.
 
 ## 1.1.0 (vNEXT)
 
-- remove deprecated `RtpDataChannels` and `DtlsSrtpKeyAgreement` options
-- remove grunt from deps, upgrade deps versions
+- removed: deprecated `RtpDataChannels` and `DtlsSrtpKeyAgreement` options
+- removed: grunt from deps, upgrade deps versions
+- removed: Reliable dep because modern browsers supports `RTCDataChannel.ordered` property
 
-- add TURN server to default config
+- added: TURN server to default config
 
 - fixed: emit error message, then destory/disconnect when error occured
+- fixed: use `peerjs-js-binarypack` instead of `js-binarypack`
+- fixed: sending large files via DataConnection #121
 
 <a name="1.0.4"></a>
 

index.d.ts

@@ -199,7 +199,6 @@ declare namespace Peer {
     data: boolean;
     binaryBlob: boolean;
     reliable: boolean;
-    sctp: boolean;
   }
 
   interface util {

lib/dataconnection.ts

@@ -1,6 +1,5 @@
-import { Reliable } from "reliable";
 import { util } from "./util";
-import logger, { LogLevel } from "./logger";
+import logger from "./logger";
 import { Negotiator } from "./negotiator";
 import {
   ConnectionType,
@@ -11,12 +10,14 @@ import {
 import { Peer } from "./peer";
 import { BaseConnection } from "./baseconnection";
 import { ServerMessage } from "./servermessage";
+import { EncodingQueue } from './encodingQueue';
 
 /**
  * Wraps a DataChannel between two Peers.
  */
 export class DataConnection extends BaseConnection {
   private static readonly ID_PREFIX = "dc_";
+  private static readonly MAX_BUFFERED_AMOUNT = 8 * 1024 * 1024;
 
   private _negotiator: Negotiator;
   readonly label: string;
@@ -30,11 +31,17 @@ export class DataConnection extends BaseConnection {
   private _buffer: any[] = [];
   private _bufferSize = 0;
   private _buffering = false;
-  private _chunkedData = {};
+  private _chunkedData: {
+    [id: number]: {
+      data: Blob[],
+      count: number,
+      total: number
+    }
+  } = {};
 
   private _peerBrowser: any;
   private _dc: RTCDataChannel;
-  private _reliable: Reliable;
+  private _encodingQueue = new EncodingQueue();
 
   get dataChannel(): RTCDataChannel {
     return this._dc;
@@ -50,12 +57,21 @@ export class DataConnection extends BaseConnection {
 
     this.label = this.options.label || this.connectionId;
     this.serialization = this.options.serialization || SerializationType.Binary;
-    this.reliable = this.options.reliable;
+    this.reliable = !!this.options.reliable;
 
     if (this.options._payload) {
       this._peerBrowser = this.options._payload.browser;
     }
 
+    this._encodingQueue.on('done', (ab: ArrayBuffer) => {
+      this._bufferedSend(ab);
+    });
+
+    this._encodingQueue.on('error', () => {
+      logger.error(`DC#${this.connectionId}: Error occured in encoding from blob to arraybuffer, close DC`);
+      this.close();
+    });
+
     this._negotiator = new Negotiator(this);
 
     this._negotiator.startConnection(
@@ -77,58 +93,48 @@ export class DataConnection extends BaseConnection {
     }
 
     this.dataChannel.onopen = () => {
-      logger.log("Data channel connection success");
+      logger.log(`DC#${this.connectionId} dc connection success`);
       this._open = true;
       this.emit(ConnectionEventType.Open);
     };
 
-    // Use the Reliable shim for non Firefox browsers
-    if (!util.supports.reliable && this.reliable) {
-      const isLoggingEnable = logger.logLevel > LogLevel.Disabled;
-      this._reliable = new Reliable(this.dataChannel, isLoggingEnable);
-    }
+    this.dataChannel.onmessage = (e) => {
+      logger.log(`DC#${this.connectionId} dc onmessage:`, e.data);
+      this._handleDataMessage(e);
+    };
 
-    if (this._reliable) {
-      this._reliable.onmessage = (msg) => {
-        this.emit(ConnectionEventType.Data, msg);
-      };
-    } else {
-      this.dataChannel.onmessage = (e) => {
-        this._handleDataMessage(e);
-      };
-    }
     this.dataChannel.onclose = () => {
-      logger.log("DataChannel closed for:", this.peer);
+      logger.log(`DC#${this.connectionId} dc closed for:`, this.peer);
       this.close();
     };
   }
 
   // Handles a DataChannel message.
-  private _handleDataMessage({ data }): void {
+  private _handleDataMessage({ data }: { data: Blob | ArrayBuffer | string }): void {
     const datatype = data.constructor;
 
     const isBinarySerialization = this.serialization === SerializationType.Binary ||
       this.serialization === SerializationType.BinaryUTF8;
 
-    let deserializedData = data;
+    let deserializedData: any = data;
 
     if (isBinarySerialization) {
       if (datatype === Blob) {
         // Datatype should never be blob
-        util.blobToArrayBuffer(data, (ab) => {
+        util.blobToArrayBuffer(data as Blob, (ab) => {
           const unpackedData = util.unpack(ab);
           this.emit(ConnectionEventType.Data, unpackedData);
         });
         return;
       } else if (datatype === ArrayBuffer) {
-        deserializedData = util.unpack(data);
+        deserializedData = util.unpack(data as ArrayBuffer);
       } else if (datatype === String) {
         // String fallback for binary data for browsers that don't support binary yet
-        const ab = util.binaryStringToArrayBuffer(data);
+        const ab = util.binaryStringToArrayBuffer(data as string);
         deserializedData = util.unpack(ab);
       }
     } else if (this.serialization === SerializationType.JSON) {
-      deserializedData = JSON.parse(data);
+      deserializedData = JSON.parse(data as string);
     }
 
     // Check if we've chunked--if so, piece things back together.
@@ -141,7 +147,7 @@ export class DataConnection extends BaseConnection {
     super.emit(ConnectionEventType.Data, deserializedData);
   }
 
-  private _handleChunk(data: any): void {
+  private _handleChunk(data: { __peerData: number, n: number, total: number, data: Blob }): void {
     const id = data.__peerData;
     const chunkInfo = this._chunkedData[id] || {
       data: [],
@@ -184,6 +190,19 @@ export class DataConnection extends BaseConnection {
       this.provider = null;
     }
 
+    if (this.dataChannel) {
+      this.dataChannel.onopen = null;
+      this.dataChannel.onmessage = null;
+      this.dataChannel.onclose = null;
+      this._dc = null;
+    }
+
+    if (this._encodingQueue) {
+      this._encodingQueue.destroy();
+      this._encodingQueue.removeAllListeners();
+      this._encodingQueue = null;
+    }
+
     if (!this.open) {
       return;
     }
@@ -205,13 +224,6 @@ export class DataConnection extends BaseConnection {
       return;
     }
 
-    if (this._reliable) {
-      // Note: reliable shim sending will make it so that you cannot customize
-      // serialization.
-      this._reliable.send(data);
-      return;
-    }
-
     if (this.serialization === SerializationType.JSON) {
       this._bufferedSend(JSON.stringify(data));
     } else if (
@@ -220,28 +232,15 @@ export class DataConnection extends BaseConnection {
     ) {
       const blob = util.pack(data);
 
-      // For Chrome-Firefox interoperability, we need to make Firefox "chunk"
-      // the data it sends out.
-      const needsChunking =
-        util.chunkedBrowsers[this._peerBrowser] ||
-        util.chunkedBrowsers[util.browser];
-
-      if (needsChunking && !chunked && blob.size > util.chunkedMTU) {
+      if (!chunked && blob.size > util.chunkedMTU) {
         this._sendChunks(blob);
         return;
       }
 
-      // DataChannel currently only supports strings.
-      if (!util.supports.sctp) {
-        util.blobToBinaryString(blob, (str) => {
-          this._bufferedSend(str);
-        });
-      } else if (!util.supports.binaryBlob) {
+      if (!util.supports.binaryBlob) {
         // We only do this if we really need to (e.g. blobs are not supported),
         // because this conversion is costly.
-        util.blobToArrayBuffer(blob, (ab) => {
-          this._bufferedSend(ab);
-        });
+        this._encodingQueue.enque(blob);
       } else {
         this._bufferedSend(blob);
       }
@@ -263,16 +262,23 @@ export class DataConnection extends BaseConnection {
       return false;
     }
 
+    if (this.dataChannel.bufferedAmount > DataConnection.MAX_BUFFERED_AMOUNT) {
+      this._buffering = true;
+      setTimeout(() => {
+        this._buffering = false;
+        this._tryBuffer();
+      }, 50);
+
+      return false;
+    }
+
     try {
       this.dataChannel.send(msg);
     } catch (e) {
+      logger.error(`DC#:${this.connectionId} Error when sending:`, e);
       this._buffering = true;
 
-      setTimeout(() => {
-        // Try again.
-        this._buffering = false;
-        this._tryBuffer();
-      }, 100);
+      this.close();
 
       return false;
     }
@@ -299,8 +305,9 @@ export class DataConnection extends BaseConnection {
     }
   }
 
-  private _sendChunks(blob): void {
+  private _sendChunks(blob: Blob): void {
     const blobs = util.chunk(blob);
+    logger.log(`DC#${this.connectionId} Try to send ${blobs.length} chunks...`);
 
     for (let blob of blobs) {
       this.send(blob, true);

lib/encodingQueue.ts

@@ -0,0 +1,64 @@
+import { EventEmitter } from "eventemitter3";
+import logger from "./logger";
+
+export class EncodingQueue extends EventEmitter {
+  readonly fileReader: FileReader = new FileReader();
+
+  private _queue: Blob[] = [];
+  private _processing: boolean = false;
+
+  constructor() {
+    super();
+
+    this.fileReader.onload = (evt) => {
+      this._processing = false;
+
+      if (evt.target) {
+        this.emit('done', evt.target.result as ArrayBuffer);
+      }
+
+      this.doNextTask();
+    };
+
+    this.fileReader.onerror = (evt) => {
+      logger.error(`EncodingQueue error:`, evt);
+      this._processing = false;
+      this.destroy();
+      this.emit('error', evt);
+    }
+  }
+
+  get queue(): Blob[] {
+    return this._queue;
+  }
+
+  get size(): number {
+    return this.queue.length;
+  }
+
+  get processing(): boolean {
+    return this._processing;
+  }
+
+  enque(blob: Blob): void {
+    this.queue.push(blob);
+
+    if (this.processing) return;
+
+    this.doNextTask();
+  }
+
+  destroy(): void {
+    this.fileReader.abort();
+    this._queue = [];
+  }
+
+  private doNextTask(): void {
+    if (this.size === 0) return;
+    if (this.processing) return;
+
+    this._processing = true;
+
+    this.fileReader.readAsArrayBuffer(this.queue.shift());
+  }
+}

lib/negotiator.ts

@@ -1,4 +1,3 @@
-import * as Reliable from "reliable";
 import { util } from "./util";
 import logger from "./logger";
 import { MediaConnection } from "./mediaconnection";
@@ -28,11 +27,7 @@ export class Negotiator {
       if (this.connection.type === ConnectionType.Data) {
         const dataConnection = <DataConnection>this.connection;
 
-        let config = {};
-
-        if (!util.supports.sctp) {
-          config = { reliable: options.reliable };
-        }
+        const config: RTCDataChannelInit = { ordered: !!options.reliable };
 
         const dataChannel = peerConnection.createDataChannel(
           dataConnection.label,
@@ -203,13 +198,6 @@ export class Negotiator {
 
       logger.log("Created offer.");
 
-      if (!util.supports.sctp && this.connection.type === ConnectionType.Data) {
-        const dataConnection = <DataConnection>this.connection;
-        if (dataConnection.reliable) {
-          offer.sdp = Reliable.higherBandwidthSDP(offer.sdp);
-        }
-      }
-
       if (this.connection.options.sdpTransform && typeof this.connection.options.sdpTransform === 'function') {
         offer.sdp = this.connection.options.sdpTransform(offer.sdp) || offer.sdp;
       }
@@ -267,13 +255,6 @@ export class Negotiator {
       const answer = await peerConnection.createAnswer();
       logger.log("Created answer.");
 
-      if (!util.supports.sctp && this.connection.type === ConnectionType.Data) {
-        const dataConnection = <DataConnection>this.connection;
-        if (dataConnection.reliable) {
-          answer.sdp = Reliable.higherBandwidthSDP(answer.sdp);
-        }
-      }
-
       if (this.connection.options.sdpTransform && typeof this.connection.options.sdpTransform === 'function') {
         answer.sdp = this.connection.options.sdpTransform(answer.sdp) || answer.sdp;
       }

lib/util.ts

@@ -1,4 +1,4 @@
-import * as BinaryPack from "js-binarypack";
+import * as BinaryPack from "peerjs-js-binarypack";
 import { Supports } from './supports';
 import { UtilSupportsObj } from '..';
 
@@ -35,7 +35,6 @@ export const util = new class {
       data: false,
       binaryBlob: false,
       reliable: false,
-      sctp: false,
     };
 
     if (!supported.webRTC) return supported;
@@ -50,10 +49,9 @@ export const util = new class {
       let dc: RTCDataChannel;
 
       try {
-        dc = pc.createDataChannel("_PEERJSTEST");
+        dc = pc.createDataChannel("_PEERJSTEST", { ordered: true });
         supported.data = true;
         supported.reliable = !!dc.ordered;
-        supported.sctp = !!pc.sctp;
 
         // Binary test
         try {
@@ -88,26 +86,25 @@ export const util = new class {
 
   // Binary stuff
 
-  private _dataCount = 1;
+  private _dataCount: number = 1;
 
-  // chunks a blob.
-  chunk(bl: Blob): any[] {
+  chunk(blob: Blob): { __peerData: number, n: number, total: number, data: Blob }[] {
     const chunks = [];
-    const size = bl.size;
+    const size = blob.size;
     const total = Math.ceil(size / util.chunkedMTU);
 
-    let index;
-    let start = (index = 0);
+    let index = 0;
+    let start = 0;
 
     while (start < size) {
       const end = Math.min(size, start + util.chunkedMTU);
-      const b = bl.slice(start, end);
+      const b = blob.slice(start, end);
 
       const chunk = {
         __peerData: this._dataCount,
         n: index,
         data: b,
-        total: total
+        total,
       };
 
       chunks.push(chunk);
@@ -121,32 +118,22 @@ export const util = new class {
     return chunks;
   }
 
-  blobToArrayBuffer(blob: Blob, cb: (arg: string | ArrayBuffer | null) => void): void {
+  blobToArrayBuffer(blob: Blob, cb: (arg: ArrayBuffer | null) => void): FileReader {
     const fr = new FileReader();
 
     fr.onload = function (evt) {
       if (evt.target) {
-        cb(evt.target.result);
+        cb(evt.target.result as ArrayBuffer);
       }
     };
 
     fr.readAsArrayBuffer(blob);
-  }
-
-  blobToBinaryString(blob: Blob, cb: (arg: string | ArrayBuffer | null) => void): void {
-    const fr = new FileReader();
-
-    fr.onload = function (evt) {
-      if (evt.target) {
-        cb(evt.target.result);
-      }
-    };
 
-    fr.readAsBinaryString(blob);
+    return fr;
   }
 
   binaryStringToArrayBuffer(binary: string): ArrayBuffer | SharedArrayBuffer {
-    let byteArray = new Uint8Array(binary.length);
+    const byteArray = new Uint8Array(binary.length);
 
     for (let i = 0; i < binary.length; i++) {
       byteArray[i] = binary.charCodeAt(i) & 0xff;

package-lock.json

@@ -5300,11 +5300,6 @@
         }
       }
     },
-    "js-binarypack": {
-      "version": "0.0.9",
-      "resolved": "https://registry.npmjs.org/js-binarypack/-/js-binarypack-0.0.9.tgz",
-      "integrity": "sha1-RUJD094hKWHMFRSi8Rnewvr2QDU="
-    },
     "js-levenshtein": {
       "version": "1.1.6",
       "resolved": "https://registry.npmjs.org/js-levenshtein/-/js-levenshtein-1.1.6.tgz",
@@ -6597,9 +6592,7 @@
       }
     },
     "peerjs-js-binarypack": {
-      "version": "1.0.0",
-      "resolved": "https://registry.npmjs.org/peerjs-js-binarypack/-/peerjs-js-binarypack-1.0.0.tgz",
-      "integrity": "sha512-XRk2K2zF75ers+J5GrXeB3J2VrICVxRnkH9RjJepXgd9AK2o0lnMt3Stjyp7QlYPZ6xWmeJxR9m2YjBMmiThDw=="
+      "version": "1.0.1"
     },
     "performance-now": {
       "version": "2.1.0",
@@ -7657,13 +7650,6 @@
         }
       }
     },
-    "reliable": {
-      "version": "git+https://github.com/michelle/reliable.git#70604f577ae55a2eb015c17d73cc8d9dce5f9ec4",
-      "from": "git+https://github.com/michelle/reliable.git",
-      "requires": {
-        "js-binarypack": "0.0.9"
-      }
-    },
     "remove-trailing-separator": {
       "version": "1.1.0",
       "resolved": "https://registry.npmjs.org/remove-trailing-separator/-/remove-trailing-separator-1.1.0.tgz",

package.json

@@ -38,8 +38,7 @@
     "eventemitter3": "^3.1.2",
     "opencollective": "^1.0.3",
     "opencollective-postinstall": "^2.0.0",
-    "peerjs-js-binarypack": "1.0.0",
-    "reliable": "git+https://github.com/michelle/reliable.git",
+    "peerjs-js-binarypack": "1.0.1",
     "webrtc-adapter": "^7.3.0"
   },
   "collective": {