fix: force messages to be provessed slowly to ensure topology

packages/transport/stream/src/__tests__/stream.test.ts

@@ -185,7 +185,7 @@ describe("streams", function () {
 			let session: TestSessionStream;
 			let streams: ReturnType<typeof createMetrics>[];
 
-			beforeAll(async () => {});
+			beforeAll(async () => { });
 
 			beforeEach(async () => {
 				// 0 and 2 not connected
@@ -248,7 +248,7 @@ describe("streams", function () {
 			let session: TestSessionStream;
 			let streams: ReturnType<typeof createMetrics>[];
 
-			beforeAll(async () => {});
+			beforeAll(async () => { });
 
 			beforeEach(async () => {
 				// 0 and 2 not connected
@@ -565,7 +565,7 @@ describe("streams", function () {
 			let session: TestSessionStream;
 			let streams: ReturnType<typeof createMetrics>[];
 
-			beforeAll(async () => {});
+			beforeAll(async () => { });
 
 			beforeEach(async () => {
 				session = await connected(3, {
@@ -717,7 +717,7 @@ describe("streams", function () {
 			let streams: ReturnType<typeof createMetrics>[];
 			const data = new Uint8Array([1, 2, 3]);
 
-			beforeAll(async () => {});
+			beforeAll(async () => { });
 
 			beforeEach(async () => {
 				session = await connected(3, {
@@ -742,6 +742,14 @@ describe("streams", function () {
 			});
 
 			it("messages are only sent once to each peer", async () => {
+				streams.forEach((stream) => {
+					const processFn = stream.stream.processMessage.bind(stream.stream);
+					stream.stream.processMessage = async (a, b, c) => {
+						await delay(200);
+						return processFn(a, b, c);
+					};
+				});
+
 				let totalWrites = 10;
 				expect(streams[0].ack).toHaveLength(0);
 
@@ -837,7 +845,7 @@ describe("streams", function () {
 			let streams: ReturnType<typeof createMetrics>[];
 			const data = new Uint8Array([1, 2, 3]);
 
-			beforeAll(async () => {});
+			beforeAll(async () => { });
 
 			beforeEach(async () => {
 				session = await disconnected(5, {
@@ -1202,7 +1210,7 @@ describe("streams", function () {
 		let streams: ReturnType<typeof createMetrics>[];
 		let timer: ReturnType<typeof setTimeout>;
 
-		beforeAll(async () => {});
+		beforeAll(async () => { });
 
 		beforeEach(async () => {
 			session = await connected(3, {
@@ -1937,13 +1945,13 @@ describe("join/leave", () => {
 					{
 						mode: seekDelivery[i]
 							? new SeekDelivery({
-									redundancy: 1,
-									to: [slow.publicKey, fast.publicKey]
-							  })
+								redundancy: 1,
+								to: [slow.publicKey, fast.publicKey]
+							})
 							: new SilentDelivery({
-									redundancy: 1,
-									to: [slow.publicKey, fast.publicKey]
-							  }) // undefined ?
+								redundancy: 1,
+								to: [slow.publicKey, fast.publicKey]
+							}) // undefined ?
 					}
 				);