Refactor peer-connection (#370)

* Add main & secondary stream configs

* Refactor dynamic configuration handling in Core.applyDynamicConfig

* Refactored types

* Fix jsdoc in types

* Refactor mergeConfigs function to use Object.create(null)

* Refactor mergeConfigs function to handle restricted properties

* Fix types

* Refactor core to use StreamConfig instead of CoreConfig

* Refactor core configuration to use partial updates in HlsJsP2PEngine and ShakaP2PEngine

* Refactor core configuration to use partial updates in HlsJsP2PEngine and ShakaP2PEngine

* Refactor core configuration to use StreamConfig instead of CoreConfig

* Add tests for core utils

* Update eslint config to ignore test folder

* Refactor core configuration to use DefinedCoreConfig

* Refactor overrideConfig

* Small improvements

* Small improvements

* Add isP2PDisabled in static stream configuration

* Refactor applyDynamicConfig method to handle changes in isP2PDisabled and stream configurations

* Refactor Peer class connection event handling

* Refactor segment loading logic to improve P2P functionality

* Refactor stream configuration handling in Core.applyDynamicConfig

* Simplified dynamic configuration handling

* Improved CoreConfig description in documentation

* Remove redundant comments from codebase

* Refactor segment loading logic to improve P2P functionality

* Refactor naming for enhanced readability

* Refactor segment loading logic to handle P2P functionality

* Refactor error handling in RequestsContainer.destroy method

* Improved naming

* Improvements in Core and RequestsContainer

* Refactor Peer class connection event listeners

* Refactor error handling in catch block

* Refactor Peer class connection event listeners and add missing event handlers

* Refactor peer-connection

* Resolved type error

Author: DimaDemchenko

packages/p2p-media-loader-core/package.json

@@ -58,6 +58,7 @@
     "vitest": "^1.6.0"
   },
   "devDependencies": {
+    "@types/streamx": "^2.9.5",
     "vite-plugin-node-polyfills": "^0.21.0"
   }
 }

packages/p2p-media-loader-core/src/declarations.d.ts

@@ -1,4 +1,6 @@
 declare module "bittorrent-tracker" {
+  import type { Duplex, WritableEvents } from "streamx";
+
   export default class Client {
     constructor(options: {
       infoHash: Uint8Array;
@@ -31,22 +33,15 @@ declare module "bittorrent-tracker" {
 
   export type PeerEvents = {
     connect: () => void;
-    data: (data: Uint8Array) => void;
-    close: () => void;
-    finish: () => void;
-    end: () => void;
-    error: (error: { code: string }) => void;
-  };
+  } & WritableEvents<unknown>;
 
-  export type PeerConnection = {
+  export type PeerConnection = Duplex & {
     id: string;
     idUtf8: string;
     initiator: boolean;
-    _channel: RTCDataChannel;
     on<E extends keyof PeerEvents>(event: E, handler: PeerEvents[E]): void;
+    off<E extends keyof PeerEvents>(event: E, handler: PeerEvents[E]): void;
     send(data: string | ArrayBuffer): void;
-    write(data: string | ArrayBuffer): void;
-    destroy(): void;
   };
 }
 

packages/p2p-media-loader-core/src/p2p/peer-protocol.ts

@@ -63,7 +63,6 @@ export class PeerProtocol {
       throw new Error(`Some segment data is already uploading.`);
     }
     const chunks = getBufferChunks(data, this.peerConfig.webRtcMaxMessageSize);
-    const channel = this.connection._channel;
     const { promise, resolve, reject } = Utils.getControlledPromise<void>();
 
     let isUploadingSegmentData = false;
@@ -82,26 +81,28 @@ export class PeerProtocol {
         return;
       }
 
-      while (channel.bufferedAmount <= channel.bufferedAmountLowThreshold) {
+      while (true) {
         const chunk = chunks.next().value;
 
         if (!chunk) {
           resolve();
           break;
         }
 
-        this.connection.send(chunk);
+        const drained = this.connection.write(chunk);
         this.onChunkUploaded(chunk.byteLength, this.connection.idUtf8);
+        if (!drained) break;
       }
     };
 
     try {
-      channel.addEventListener("bufferedamountlow", sendChunk);
+      this.connection.on("drain", sendChunk);
       isUploadingSegmentData = true;
       sendChunk();
       await promise;
     } finally {
-      channel.removeEventListener("bufferedamountlow", sendChunk);
+      this.connection.off("drain", sendChunk);
+
       if (this.uploadingContext === uploadingContext) {
         this.uploadingContext = undefined;
       }

packages/p2p-media-loader-core/src/p2p/peer.ts

@@ -304,12 +304,16 @@ export class Peer {
     this.destroy();
   };
 
-  private onConnectionError = (error: { code: string }) => {
+  private onConnectionError = (error: Error) => {
     this.logger(`peer connection error ${this.id} %O`, error);
 
-    if (error.code === "ERR_DATA_CHANNEL") {
+    const code = (error as { code?: string }).code;
+
+    if (code === "ERR_DATA_CHANNEL") {
+      this.destroy();
+    } else if (code === "ERR_CONNECTION_FAILURE") {
       this.destroy();
-    } else if (error.code === "ERR_CONNECTION_FAILURE") {
+    } else if (code === "ERR_CONNECTION_FAILURE") {
       this.destroy();
     }
   };