fix(datachannel): sending order is now preserved correctly (#1038)

This upgrades `binarypack` to version 2.

`pack` now returns `ArrayBuffer`s which can be sent over all data
channel types without asynchronous conversion.
This implementation is also much faster than version 1.

Closes #746

e2e/datachannel/serialization.spec.ts

@@ -25,10 +25,11 @@ describe("DataChannel:Default", () => {
  await P.init();
  });
  it("should transfer numbers", serializationTest("./numbers.js"));
- /** ordering bug: chunked string not in order */
- // it('should transfer strings', serializationTest("./strings.js"))
+ it("should transfer strings", serializationTest("./strings.js"));
  it("should transfer objects", serializationTest("./objects.js"));
  it("should transfer arrays", serializationTest("./arrays.js"));
- /** can't send bug */
- // it('should transfer typed arrays / array buffers', serializationTest("./arraybuffers.js"))
+ it(
+ "should transfer typed arrays / array buffers",
+ serializationTest("./arraybuffers.js"),
+ );
 });

lib/dataconnection.ts

@@ -1,4 +1,4 @@
-import { util } from "./util";
+import { concatArrayBuffers, util } from "./util";
 import logger from "./logger";
 import { Negotiator } from "./negotiator";
 import { ConnectionType, SerializationType, ServerMessageType } from "./enums";
@@ -65,7 +65,7 @@ export class DataConnection
  private _buffering = false;
  private _chunkedData: {
  [id: number]: {
- data: Blob[];
+ data: Uint8Array[];
  count: number;
  total: number;
  };
@@ -119,9 +119,7 @@ export class DataConnection
  /** Called by the Negotiator when the DataChannel is ready. */
  override _initializeDataChannel(dc: RTCDataChannel): void {
  this._dc = dc;
- if (!util.supports.binaryBlob || util.supports.reliable) {
- this.dataChannel.binaryType = "arraybuffer";
- }
+ this.dataChannel.binaryType = "arraybuffer";
 
  this.dataChannel.onopen = () => {
  logger.log(`DC#${this.connectionId} dc connection success`);
@@ -193,7 +191,7 @@ export class DataConnection
  __peerData: number;
  n: number;
  total: number;
- data: Blob;
+ data: ArrayBuffer;
  }): void {
  const id = data.__peerData;
  const chunkInfo = this._chunkedData[id] || {
@@ -202,7 +200,7 @@ export class DataConnection
  total: data.total,
  };
 
- chunkInfo.data[data.n] = data.data;
+ chunkInfo.data[data.n] = new Uint8Array(data.data);
  chunkInfo.count++;
  this._chunkedData[id] = chunkInfo;
 
@@ -211,8 +209,8 @@ export class DataConnection
  delete this._chunkedData[id];
 
  // We've received all the chunks--time to construct the complete data.
- const data = new Blob(chunkInfo.data);
- this._handleDataMessage({ data });
+ const data = concatArrayBuffers(chunkInfo.data);
+ this.emit("data", util.unpack(data));
  }
  }
 
@@ -283,6 +281,11 @@ export class DataConnection
  return;
  }
 
+ if (data instanceof Blob) {
+ data.arrayBuffer().then((ab) => this.send(ab));
+ return;
+ }
+
  if (this.serialization === SerializationType.JSON) {
  this._bufferedSend(this.stringify(data));
  } else if (
@@ -291,18 +294,12 @@ export class DataConnection
  ) {
  const blob = util.pack(data);
 
- if (!chunked && blob.size > util.chunkedMTU) {
+ if (!chunked && blob.byteLength > util.chunkedMTU) {
  this._sendChunks(blob);
  return;
  }
 
- if (!util.supports.binaryBlob) {
- // We only do this if we really need to (e.g. blobs are not supported),
- // because this conversion is costly.
- this._encodingQueue.enque(blob);
- } else {
- this._bufferedSend(blob);
- }
+ this._bufferedSend(blob);
  } else {
  this._bufferedSend(data);
  }
@@ -364,7 +361,7 @@ export class DataConnection
  }
  }
 
- private _sendChunks(blob: Blob): void {
+ private _sendChunks(blob: ArrayBuffer): void {
  const blobs = util.chunk(blob);
  logger.log(`DC#${this.connectionId} Try to send ${blobs.length} chunks...`);
 

lib/util.ts

@@ -131,10 +131,10 @@ export class Util {
  private _dataCount: number = 1;
 
  chunk(
- blob: Blob,
- ): { __peerData: number; n: number; total: number; data: Blob }[] {
+ blob: ArrayBuffer,
+ ): { __peerData: number; n: number; total: number; data: ArrayBuffer }[] {
  const chunks = [];
- const size = blob.size;
+ const size = blob.byteLength;
  const total = Math.ceil(size / util.chunkedMTU);
 
  let index = 0;
@@ -208,3 +208,16 @@ export class Util {
  * :::
  */
 export const util = new Util();
+export function concatArrayBuffers(bufs: Uint8Array[]) {
+ let size = 0;
+ for (const buf of bufs) {
+ size += buf.byteLength;
+ }
+ const result = new Uint8Array(size);
+ let offset = 0;
+ for (const buf of bufs) {
+ result.set(buf, offset);
+ offset += buf.byteLength;
+ }
+ return result;
+}

package.json

@@ -195,7 +195,7 @@
  "dependencies": {
  "@swc/helpers": "^0.5.0",
  "eventemitter3": "^4.0.7",
- "peerjs-js-binarypack": "1.0.2",
+ "peerjs-js-binarypack": "^2.0.0",
  "webrtc-adapter": "^8.0.0"
  }
 }