flutter-webrtc
diff --git a/‎example/lib/src/data_channel_sample.dart
Lines changed: 10 additions & 0 deletions b/‎example/lib/src/data_channel_sample.dart
Lines changed: 10 additions & 0 deletions
diff --git a/‎lib/rtc_data_channel.dart
Lines changed: 24 additions & 4 deletions b/‎lib/rtc_data_channel.dart
Lines changed: 24 additions & 4 deletions
@@ -61,6 +61,16 @@ class _DataChannelSampleState extends State<DataChannelSample> {
         // do something with message.binary
       }
     };
+    // or alternatively:
+    dataChannel.messageStream.listen((message) {
+      if (message.type == MessageType.text) {
+        print(message.text);
+      }
+      else {
+        // do something with message.binary
+      }
+    });
+    
     dataChannel.send(RTCDataChannelMessage("Hello!"));
     dataChannel.send(RTCDataChannelMessage.fromBinary(
       Uint8List(5)
 
@@ -5,6 +5,7 @@ import 'package:flutter/services.dart';
 import 'utils.dart';
 import 'dart:io' show Platform;
 
+
 enum MessageType {
   text, binary
 }
@@ -83,6 +84,7 @@ class RTCDataChannelMessage {
 
   /// Binary contents of this message as a base64 encoded [String].
   /// Use only on binary messages.
+  /// See: [isBinary].
   String get binaryAsBase64 => base64.encode(_data);
 }
 
@@ -120,7 +122,20 @@ class RTCDataChannel {
   /// binary data as a [Uint8List] or text data as a [String].
   RTCDataChannelOnMessageCallback onMessage;
 
-  RTCDataChannel(this._peerConnectionId, this._label, this._dataChannelId){
+  final _stateChangeController = StreamController<RTCDataChannelState>.broadcast(sync: true);
+  final _messageController = StreamController<RTCDataChannelMessage>.broadcast(sync: true);
+
+  /// Stream of state change events. Emits the new state on change.
+  /// Closes when the [RTCDataChannel] is closed.
+  Stream<RTCDataChannelState> stateChangeStream;
+
+  /// Stream of incoming messages. Emits the message.
+  /// Closes when the [RTCDataChannel] is closed.
+  Stream<RTCDataChannelMessage> messageStream;
+
+  RTCDataChannel(this._peerConnectionId, this._label, this._dataChannelId) {
+    stateChangeStream = _stateChangeController.stream;
+    messageStream = _messageController.stream;
     _eventSubscription = _eventChannelFor(_dataChannelId)
       .receiveBroadcastStream()
       .listen(eventListener, onError: errorListener);
@@ -133,8 +148,10 @@ class RTCDataChannel {
       case 'dataChannelStateChanged':
         //int dataChannelId = map['id'];
         _state = rtcDataChannelStateForString(map['state']);
-        if (this.onDataChannelState != null)
+        if (this.onDataChannelState != null) {
           this.onDataChannelState(_state);
+        }
+        _stateChangeController.add(_state);
         break;
       case 'dataChannelReceiveMessage':
         //int dataChannelId = map['id'];
@@ -153,9 +170,10 @@ class RTCDataChannel {
         else {
           message = RTCDataChannelMessage(data);
         }
-
-        if (this.onMessage != null)
+        if (this.onMessage != null) {
           this.onMessage(message);
+        }
+        _messageController.add(message);
         break;
     }
   }
@@ -200,6 +218,8 @@ class RTCDataChannel {
   }
 
   Future<void> close() async {
+    await _stateChangeController.close();
+    await _messageController.close();
     await _eventSubscription?.cancel();
     await _channel.invokeMethod('dataChannelClose',
         <String, dynamic>{'peerConnectionId': _peerConnectionId, 'dataChannelId': _dataChannelId});