Make subchannels be ordered

CHANGELOG.md

@@ -4,6 +4,7 @@
 ### NEXT
 
 * liburing: avoid extra memcpy on RTP ([PR #1258](https://github.com/versatica/mediasoup/pull/1258)).
+* Make subchannels be ordered ([PR #1261](https://github.com/versatica/mediasoup/pull/1261)).
 
 
 ### 3.13.10

node/src/DataConsumer.ts

@@ -565,6 +565,18 @@ export class DataConsumer<DataConsumerAppData extends AppData = AppData>
 	{
 		logger.debug('setSubchannels()');
 
+		// Optimistic subchannels update. We do this to make the |subchannels|
+		// getter be immediately updated with given values. Later it will be
+		// updated again with the response from the worker.
+		const subchannelsSet = new Set<number>();
+
+		for (const subchannel of subchannels)
+		{
+			subchannelsSet.add(Math.abs(subchannel % 65536));
+		}
+
+		this.#subchannels = Array.from(subchannelsSet).sort();
+
 		/* Build Request. */
 		const requestOffset = new FbsDataConsumer.SetSubchannelsRequestT(
 			subchannels

node/src/tests/test-DataConsumer.ts

@@ -71,7 +71,7 @@ test('transport.consumeData() succeeds', async () =>
 	expect(dataConsumer1.label).toBe('foo');
 	expect(dataConsumer1.protocol).toBe('bar');
 	expect(dataConsumer1.paused).toBe(false);
-	expect(dataConsumer1.subchannels.sort((a, b) => a - b)).toEqual([ 0, 1, 2, 100, 65535 ]);
+	expect(dataConsumer1.subchannels).toEqual([ 0, 1, 2, 100, 65535 ]);
 	expect(dataConsumer1.appData).toEqual({ baz: 'LOL' });
 
 	const dump = await router.dump();
@@ -132,9 +132,17 @@ test('dataConsumer.getStats() succeeds', async () =>
 
 test('dataConsumer.setSubchannels() succeeds', async () =>
 {
-	await dataConsumer1.setSubchannels([ 999, 999, 998, 65536 ]);
+	const expectedSubchannels = [ 0, 998, 999 ];
 
-	expect(dataConsumer1.subchannels.sort((a, b) => a - b)).toEqual([ 0, 998, 999 ]);
+	const promise = dataConsumer1.setSubchannels([ 999, 999, 998, 65536 ]);
+
+	// Before even subchannels are updated in worker side, its value in Node must
+	// already be updated.
+	expect(dataConsumer1.subchannels).toEqual(expectedSubchannels);
+
+	// And also once the promise resolves.
+	await promise;
+	expect(dataConsumer1.subchannels).toEqual(expectedSubchannels);
 }, 2000);
 
 test('transport.consumeData() on a DirectTransport succeeds', async () =>

rust/tests/integration/data_consumer.rs

@@ -153,11 +153,7 @@ fn consume_data_succeeds() {
         }
         assert_eq!(data_consumer.label().as_str(), "foo");
         assert_eq!(data_consumer.protocol().as_str(), "bar");
-
-        let mut sorted_subchannels = data_consumer.subchannels();
-        sorted_subchannels.sort();
-
-        assert_eq!(sorted_subchannels, [0, 1, 2, 100, 65535]);
+        assert_eq!(data_consumer.subchannels(), [0, 1, 2, 100, 65535]);
         assert_eq!(
             data_consumer
                 .app_data()

worker/include/RTC/DataConsumer.hpp

@@ -6,7 +6,7 @@
 #include "Channel/ChannelSocket.hpp"
 #include "RTC/SctpDictionaries.hpp"
 #include "RTC/Shared.hpp"
-#include <absl/container/flat_hash_set.h>
+#include <absl/container/btree_set.h>
 #include <string>
 
 namespace RTC
@@ -126,7 +126,7 @@ namespace RTC
 		RTC::SctpStreamParameters sctpStreamParameters;
 		std::string label;
 		std::string protocol;
-		absl::flat_hash_set<uint16_t> subchannels;
+		absl::btree_set<uint16_t> subchannels;
 		bool transportConnected{ false };
 		bool sctpAssociationConnected{ false };
 		bool paused{ false };