diff --git a/.changes/subscriber-dc-state-events b/.changes/subscriber-dc-state-events new file mode 100644 index 000000000..896e2155c --- /dev/null +++ b/.changes/subscriber-dc-state-events @@ -0,0 +1 @@ +patch type="fixed" "Subscriber data channel state events now report the subscriber channel state and correct reliability type, and no longer leak listeners" diff --git a/lib/src/core/engine.dart b/lib/src/core/engine.dart index cf2c4c162..91b49f7b1 100644 --- a/lib/src/core/engine.dart +++ b/lib/src/core/engine.dart @@ -809,23 +809,21 @@ class Engine extends Disposable with EventsEmittable { logger.fine('Server opened DC label: ${dc.label}'); _reliableDCSub = dc; _reliableDCSub?.onMessage = _onDCMessage; - _reliableDCSub?.stateChangeStream.listen((state) => - _reliableDCPub?.stateChangeStream.listen((state) => events.emit(SubscriberDataChannelStateUpdatedEvent( - isPrimary: _subscriberPrimary, - state: state, - type: Reliability.reliable, - )))); + _reliableDCSub?.stateChangeStream.listen((state) => events.emit(SubscriberDataChannelStateUpdatedEvent( + isPrimary: _subscriberPrimary, + state: state, + type: Reliability.reliable, + ))); break; case _lossyDCLabel: logger.fine('Server opened DC label: ${dc.label}'); _lossyDCSub = dc; _lossyDCSub?.onMessage = _onDCMessage; - _lossyDCSub?.stateChangeStream.listen((event) => - _reliableDCPub?.stateChangeStream.listen((state) => events.emit(SubscriberDataChannelStateUpdatedEvent( - isPrimary: _subscriberPrimary, - state: state, - type: Reliability.lossy, - )))); + _lossyDCSub?.stateChangeStream.listen((state) => events.emit(SubscriberDataChannelStateUpdatedEvent( + isPrimary: _subscriberPrimary, + state: state, + type: Reliability.lossy, + ))); break; default: logger.warning('Unknown DC label: ${dc.label}'); diff --git a/test/core/data_channel_state_test.dart b/test/core/data_channel_state_test.dart new file mode 100644 index 000000000..0380141fc --- /dev/null +++ b/test/core/data_channel_state_test.dart @@ -0,0 +1,86 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +@Timeout(Duration(seconds: 10)) +library; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:flutter_webrtc/flutter_webrtc.dart' as rtc; + +import 'package:livekit_client/livekit_client.dart'; +import 'package:livekit_client/src/internal/events.dart'; +import '../mock/datachannel_mock.dart'; +import '../mock/e2e_container.dart'; +import '../mock/peerconnection_mock.dart'; + +void main() { + setUp(resetMockDataChannels); + + group('subscriber data channel state events', () { + // Regression: _onDataChannel used to register a nested listener on the + // publisher reliable channel instead of emitting the subscriber channel's + // own state. Subscriber state changes produced no immediate event, later + // events carried the wrong channel's state, and inner listeners + // accumulated on every state change. + test('emit the subscriber channel state with the correct reliability type', () async { + final container = E2EContainer(); + addTearDown(container.dispose); + await container.connectRoom(); + + final engine = container.room.engine; + final events = []; + final listener = engine.createListener()..on(events.add); + addTearDown(listener.dispose); + + // Simulate the server opening subscriber-side channels. + final reliableSub = MockDataChannel(8, '_reliable'); + final lossySub = MockDataChannel(9, '_lossy'); + final onDataChannel = engine.subscriber!.pc.onDataChannel!; + onDataChannel(reliableSub); + onDataChannel(lossySub); + + reliableSub.stateChangeStreamController.add(rtc.RTCDataChannelState.RTCDataChannelClosing); + lossySub.stateChangeStreamController.add(rtc.RTCDataChannelState.RTCDataChannelClosed); + await Future.delayed(Duration.zero); + + expect(events, hasLength(2)); + expect(events[0].type, Reliability.reliable); + expect(events[0].state, rtc.RTCDataChannelState.RTCDataChannelClosing); + expect(events[1].type, Reliability.lossy); + expect(events[1].state, rtc.RTCDataChannelState.RTCDataChannelClosed); + }); + + test('repeated state changes do not multiply events', () async { + final container = E2EContainer(); + addTearDown(container.dispose); + await container.connectRoom(); + + final engine = container.room.engine; + final events = []; + final listener = engine.createListener()..on(events.add); + addTearDown(listener.dispose); + + final reliableSub = MockDataChannel(8, '_reliable'); + engine.subscriber!.pc.onDataChannel!(reliableSub); + + for (var i = 0; i < 3; i++) { + reliableSub.stateChangeStreamController.add(rtc.RTCDataChannelState.RTCDataChannelOpen); + } + await Future.delayed(Duration.zero); + + expect(events, hasLength(3)); + expect(events.every((e) => e.state == rtc.RTCDataChannelState.RTCDataChannelOpen), isTrue); + }); + }); +} diff --git a/test/mock/datachannel_mock.dart b/test/mock/datachannel_mock.dart index c6b80b646..e63121f83 100644 --- a/test/mock/datachannel_mock.dart +++ b/test/mock/datachannel_mock.dart @@ -24,6 +24,9 @@ class MockDataChannel extends RTCDataChannel { late StreamController stateChangeStreamController; MockDataChannel(this._id, this._label) { stateChangeStreamController = StreamController.broadcast(); + // The base class declares this as a bare `late` field, reading it before + // assignment throws LateInitializationError + stateChangeStream = stateChangeStreamController.stream; } @override