Fix subscriber data channel state events reporting the wrong channel (#1172)

_onDataChannel registered a nested listener: every state change on a
subscriber data channel attached a new listener to the publisher
reliable channel and emitted SubscriberDataChannelStateUpdatedEvent from
that channel's state instead. So subscriber state changes produced no
immediate event, the events that did arrive carried the publisher
channel's state (for the lossy label too), and inner listeners
accumulated uncancelled for the lifetime of the connection.

Now each subscriber channel emits its own state directly from a single
listener, mirroring the publisher-side wiring.

Also initializes the late stateChangeStream field in MockDataChannel,
which previously threw LateInitializationError whenever engine code
touched it in tests. Regression test verified to fail against the old
wiring.

Pre-existing bug surfaced by Devin on the #1170 reformat diff, fixed
separately to keep that PR mechanical.

Author: hiroshihorie

.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"

lib/src/core/engine.dart

@@ -809,23 +809,21 @@ class Engine extends Disposable with EventsEmittable<EngineEvent> {
         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}');

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 = <SubscriberDataChannelStateUpdatedEvent>[];
+      final listener = engine.createListener()..on<SubscriberDataChannelStateUpdatedEvent>(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<void>.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 = <SubscriberDataChannelStateUpdatedEvent>[];
+      final listener = engine.createListener()..on<SubscriberDataChannelStateUpdatedEvent>(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<void>.delayed(Duration.zero);
+
+      expect(events, hasLength(3));
+      expect(events.every((e) => e.state == rtc.RTCDataChannelState.RTCDataChannelOpen), isTrue);
+    });
+  });
+}

test/mock/datachannel_mock.dart

@@ -24,6 +24,9 @@ class MockDataChannel extends RTCDataChannel {
   late StreamController<RTCDataChannelState> stateChangeStreamController;
   MockDataChannel(this._id, this._label) {
     stateChangeStreamController = StreamController<RTCDataChannelState>.broadcast();
+    // The base class declares this as a bare `late` field, reading it before
+    // assignment throws LateInitializationError
+    stateChangeStream = stateChangeStreamController.stream;
   }
 
   @override