Add pipe boolean option to transport.consume() (versatica#494)

LewisWolfgang · Jan 4, 2021 · 6961afb · 6961afb
1 parent e06c153
commit 6961afb
Show file tree

Hide file tree

Showing 13 changed files with 253 additions and 110 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,8 +1,11 @@
 # Changelog
 
 
-### 3.6.30 (WIP)
+### 3.7.30 (WIP)
 
+* Add `pipe` option to `transport.consume()`(PR #494).
+  - So the receiver will get all streams from the `Producer`.
+  - It works for any kind of transport (but `PipeTransport` which is always like this).
 * Update NPM deps.
 * Add `LICENSE` and `PATENTS` files in `libwebrtc` dependency (issue #495).
 * Added `worker/src/Utils/README_BASE64_UTILS` (issue #497).

diff --git a/lib/Consumer.d.ts b/lib/Consumer.d.ts
@@ -32,6 +32,11 @@ export declare type ConsumerOptions = {
      * If unset, the highest ones are selected.
      */
     preferredLayers?: ConsumerLayers;
+    /**
+     * Whether this Consumer should consume all RTP streams generated by the
+     * Producer.
+     */
+    pipe?: boolean;
     /**
      * Custom application data.
      */

diff --git a/lib/Consumer.d.ts.map b/lib/Consumer.d.ts.map
diff --git a/lib/Transport.d.ts b/lib/Transport.d.ts
@@ -166,7 +166,7 @@ export declare class Transport extends EnhancedEventEmitter {
      *
      * @virtual
      */
-    consume({ producerId, rtpCapabilities, paused, preferredLayers, appData }: ConsumerOptions): Promise<Consumer>;
+    consume({ producerId, rtpCapabilities, paused, preferredLayers, pipe, appData }: ConsumerOptions): Promise<Consumer>;
     /**
      * Create a DataProducer.
      */

diff --git a/lib/Transport.d.ts.map b/lib/Transport.d.ts.map
diff --git a/lib/Transport.js b/lib/Transport.js
@@ -275,7 +275,7 @@ class Transport extends EnhancedEventEmitter_1.EnhancedEventEmitter {
      *
      * @virtual
      */
-    async consume({ producerId, rtpCapabilities, paused = false, preferredLayers, appData = {} }) {
+    async consume({ producerId, rtpCapabilities, paused = false, preferredLayers, pipe = false, appData = {} }) {
         logger.debug('consume()');
         if (!producerId || typeof producerId !== 'string')
             throw new TypeError('missing producerId');
@@ -287,25 +287,31 @@ class Transport extends EnhancedEventEmitter_1.EnhancedEventEmitter {
         if (!producer)
             throw Error(`Producer with id "${producerId}" not found`);
         // This may throw.
-        const rtpParameters = ortc.getConsumerRtpParameters(producer.consumableRtpParameters, rtpCapabilities);
+        const rtpParameters = ortc.getConsumerRtpParameters(producer.consumableRtpParameters, rtpCapabilities, pipe);
         // Set MID.
-        rtpParameters.mid = `${this._nextMidForConsumers++}`;
-        // We use up to 8 bytes for MID (string).
-        if (this._nextMidForConsumers === 100000000) {
-            logger.error(`consume() | reaching max MID value "${this._nextMidForConsumers}"`);
-            this._nextMidForConsumers = 0;
+        if (!pipe) {
+            rtpParameters.mid = `${this._nextMidForConsumers++}`;
+            // We use up to 8 bytes for MID (string).
+            if (this._nextMidForConsumers === 100000000) {
+                logger.error(`consume() | reaching max MID value "${this._nextMidForConsumers}"`);
+                this._nextMidForConsumers = 0;
+            }
         }
         const internal = { ...this._internal, consumerId: uuid_1.v4(), producerId };
         const reqData = {
             kind: producer.kind,
             rtpParameters,
-            type: producer.type,
+            type: pipe ? 'pipe' : producer.type,
             consumableRtpEncodings: producer.consumableRtpParameters.encodings,
             paused,
             preferredLayers
         };
         const status = await this._channel.request('transport.consume', internal, reqData);
-        const data = { kind: producer.kind, rtpParameters, type: producer.type };
+        const data = {
+            kind: producer.kind,
+            rtpParameters,
+            type: pipe ? 'pipe' : producer.type
+        };
         const consumer = new Consumer_1.Consumer({
             internal,
             data,

diff --git a/lib/ortc.d.ts b/lib/ortc.d.ts
@@ -118,7 +118,7 @@ export declare function canConsume(consumableParams: RtpParameters, caps: RtpCap
  * to reduce codecs, codecs' RTCP feedback and header extensions, and also enables
  * or disabled RTX.
  */
-export declare function getConsumerRtpParameters(consumableParams: RtpParameters, caps: RtpCapabilities): RtpParameters;
+export declare function getConsumerRtpParameters(consumableParams: RtpParameters, caps: RtpCapabilities, pipe: boolean): RtpParameters;
 /**
  * Generate RTP parameters for a pipe Consumer.
  *