ChannelMessageHandlers: Make RegisterHandler() not remove the existing handler if another one with same id is given

[ibc]

• Do not remove existing handler if an attempt to register a new one with same id happens. Tis prevents the following issue discovered while adding a new unit test:

FAIL  node/tests/test-PipeTransport.js
 ✓ router.pipeToRouter() succeeds with audio (24 ms)
 ✓ router.pipeToRouter() succeeds with video (21 ms)
 ### NEW TEST:
 ✓ router.pipeToRouter() fails if both Routers belong to the same Worker (32 ms)
 ✓ router.createPipeTransport() with enableRtx succeeds (6 ms)
 ✓ router.createPipeTransport() with invalid srtpParameters must fail (3 ms)
 ✓ router.createPipeTransport() with enableSrtp succeeds (9 ms)
 ✓ router.createPipeTransport() with fixed port succeeds (8 ms)
 ✓ transport.consume() for a pipe Producer succeeds (6 ms)
 ### UNRELATED FAILING TEST:
 ✕ producer.pause() and producer.resume() are transmitted to pipe Consumer (2 ms)
 ✓ producer.close() is transmitted to pipe Consumer (2 ms)
 ✓ router.pipeToRouter() succeeds with data (7 ms)
 ✓ transport.dataConsume() for a pipe DataProducer succeeds (2 ms)
 ✓ dataProducer.close() is transmitted to pipe DataConsumer (2 ms)
 ✓ router.pipeToRouter() called twice generates a single PipeTransport pair (14 ms)
 ✓ router.pipeToRouter() called in two Routers passing one to each other as argument generates a single a single PipeTransport pair (10 ms)

 ● producer.pause() and producer.resume() are transmitted to pipe Consumer

   Channel request handler with ID c9cba815-1128-4500-aa4f-7edbc135b889 not found [method:producer.resume]

It doesn't make sense that such a test fails with that error (producer not found in the worker). To simplify the error, I've added a temporal code in the new test that clearly shows the issue:

test('router.pipeToRouter() fails if both Routers belong to the same Worker', async () =>
{
	const router1bis = await worker1.createRouter({ mediaCodecs });

	await expect(router1.pipeToRouter(
		{
			producerId : videoProducer.id,
			router     : router1bis
		}))
		.rejects
		.toThrow(Error);

	// TODO: Temporal code to show an unexpected error.
	{
		// This is ok.
		expect(videoProducer.closed).toBe(false);

		// However these will fail since the handler Id does not exist in the worker.
		// This is impossible and must be a bug.
		await videoProducer.resume();
		await videoProducer.pause();
	}

	router1bis.close();
}, 2000);

[ibc]

To be clear, it looks like this code is closing/deleting videoProducer in the worker (it shouldn't):

router1.pipeToRouter(
	{
		producerId : videoProducer.id,
		router     : router1bis
	});

Ok, the code throws because both routers are in the same worker, but that should NOT close the given producer.

@nazar-pc I cannot find anything that could cause this error. I've verified tat Router::pipeToRouter() method in Router.ts does not close the given producer at all.

[ibc]

I know where the bug is.

[ibc]

Adding test in Rust.

[ibc]

@nazar-pc I definitely need your help in the Rust test:

failures:

---- pipe_transport::pipe_to_router_fails_if_both_routers_belong_to_the_same_worker stdout ----
Err(ProduceFailed(Request(Response { reason: "Channel request handler with ID d41aa2bc-588c-467f-941f-4b7bfe1e8f63 already exists [method:transport.produce]" })))

thread 'pipe_transport::pipe_to_router_fails_if_both_routers_belong_to_the_same_worker' panicked at 'assertion failed: matches!(result,\n    Err(PipeProducerToRouterError ::\n    ProduceFailed(ProduceError :: AlreadyExists(producer_id))))', rust/tests/integration/pipe_transport.rs:584:9
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace


failures:
    pipe_transport::pipe_to_router_fails_if_both_routers_belong_to_the_same_worker

[nazar-pc]

The error comes from

mediasoup/rust/src/router/transport.rs

Lines 403 to 407 in 6ac1cd2

	if let Some(id) = &producer_options.id {
	if self.router().has_producer(id) {
	return Err(ProduceError::AlreadyExists(*id));
	}
	}

As far as Rust side is concerned, router certainly contains such producer already and it doesn't make any sense to produce it again.

I don't think I fully understand what you're trying to do in this PR.

[ibc]

The error comes from

No, it doesn't. The error comes from router1.pipe_producer_to_router() as expected given that I'm calling it by passing as argument a router1bis Router which was created in the same worker. Remember the limitation due to the global id handler?

[ibc]

As far as Rust side is concerned, router certainly contains such producer already and it doesn't make any sense to produce it again.

No, this is not the error.

[nazar-pc]

I see. You'll need to inspect error message then and convert it to ProduceError::AlreadyExists (and I think similarly for data producer as well). Otherwise it returns generic error straight from the worker as a reason string.

[ibc]

• 1 works fine. No error here.
• 2 fails (as expected) since there is already a handler with same id the the worker.
• 3 fails because definitely I don't know what to put here. This is an error generated by the Worker rather than by Rust. However when doing assert!(matches!(result, xxxxx) is seems that xxxxx MUST be an enum of PipeProducerToRouterError, but none of those are being generated because the error is NOT generated by Rust but later by the Worker.

[ibc]

I see. You'll need to inspect error message then and convert it to ProduceError::AlreadyExists (and I think similarly for data producer as well). Otherwise it returns generic error straight from the worker as a reason string.

Please help here, I've spent lot long time already trying to figure out the syntax here without success. I don't know how to inspect the error message.

[ibc]

I've already searched for all error matching in Rust tests. None is similar to what it's needed here. And there is nothing about data producers that can be taken as guide.

[nazar-pc]

As you figured out, the root error is obviously PipeProducerToRouterError, which is an enum. Specifically it failed because with PipeProducerToRouterError::ProduceFailed, which in turn contains enum that explains why produce failed, in this case variant is ProduceError::Request, meaning request failed, which failed because worker sent corresponding response, hence RequestError::Response with an error message Channel request handler with ID d91991e8-fcbe-4069-8660-668cb94dc6dd already exists [method:transport.produce]. If you want to catch that, you can check if error matches that and then inspect internal reason message.

So it is just a few enums nested in one another with narrowing reason of an error.

I have pushed corresponding change, let me know if that is sufficient or I should still intercept it and replace with ProduceError::AlreadyExists instead.

[ibc]

Thanks, IMHO looks good, we don't need to be super error specific in corner cases like this.

Said that, I don't think this is ok (pipe_producer_to_router() method):

In case the pipe_producer creation fails, the method should close the pipe_consumer (as we do in TS).

[nazar-pc]

In case the pipe_producer creation fails, the method should close the pipe_consumer (as we do in TS).

We have RAII in Rust, since that consumer is not stored anywhere (due to error exiting early) it'll be destroyed on drop automatically with destructor (Drop trait implementation in Rust). So it works the same way as TypeScript, just doesn't require explicit code for that.

[ibc]

lovely!

impl Drop for Inner {
    fn drop(&mut self) {
        debug!("drop()");

        self.close(true);
    }
}