Add binary message decoding/encoding support for message and optional payload decoder. (braverhealth#105) (#12)

* add msgpack_dart decoding

* refactor: replace useMessagePack with payloadDecoder in MessageSerializer and PhoenixSocketOptions

* add tests for binry message encoding and decoding

* fix ci

---------

Co-authored-by: Neelansh Sethi <sethineelansh@gmail.com>
Co-authored-by: matehat <mathieu@braver.health>

Author: 3 people

.github/workflows/test.yaml

@@ -18,20 +18,19 @@ jobs:
         ports:
           - 4001:4001
           - 4002:4002
-        
+
       proxy:
         image: ghcr.io/shopify/toxiproxy
         ports:
           - 8474:8474
           - 4004:4004
 
     steps:
-
       - name: Setup Dart
         uses: dart-lang/setup-dart@v1
         with:
           architecture: x64
-          sdk: "3.1.2"
+          sdk: "3.7.2"
 
       - name: Fetch sources
         uses: actions/checkout@v2

lib/src/message.dart

@@ -1,5 +1,5 @@
+import 'dart:typed_data';
 import 'package:logging/logging.dart';
-
 import 'channel.dart';
 import 'events.dart';
 import 'socket.dart';
@@ -53,6 +53,23 @@ class Message {
     this.payload,
   });
 
+  /// Build a [Message] with binary payload.
+  factory Message.binary({
+    String? joinRef,
+    String? ref,
+    String? topic,
+    required PhoenixChannelEvent event,
+    required Uint8List payload,
+  }) {
+    return Message(
+      joinRef: joinRef,
+      ref: ref,
+      topic: topic,
+      event: event,
+      payload: payload,
+    );
+  }
+
   /// Reference of the channel on which the message is received.
   ///
   /// Used by the [PhoenixSocket] to route the message on the proper
@@ -73,8 +90,8 @@ class Message {
 
   /// The payload of this message.
   ///
-  /// This needs to be a JSON-encodable object.
-  final Map<String, dynamic>? payload;
+  /// This can be either a JSON-encodable Map or a Uint8List for binary data.
+  final dynamic payload;
 
   /// Encode a message to a JSON-encodable list of values.
   Object encode() {

lib/src/message_serializer.dart

@@ -1,32 +1,86 @@
 import 'dart:convert';
+import 'dart:typed_data';
 
-import 'message.dart';
+import 'package:phoenix_socket/phoenix_socket.dart';
+import 'package:phoenix_socket/src/utils/serializer.dart';
+import 'package:phoenix_socket/src/utils/map_utils.dart';
 
 typedef DecoderCallback = dynamic Function(String rawData);
 typedef EncoderCallback = String Function(Object? data);
+typedef PayloadDecoderCallback = dynamic Function(Uint8List payload);
 
 /// Default class to serialize [Message] instances to JSON.
 class MessageSerializer {
+  static const int headerLength = 1;
+  static const int metaLength = 4;
+
+  static const Map<String, int> kinds = {
+    'push': 0,
+    'reply': 1,
+    'broadcast': 2,
+  };
+
   final DecoderCallback _decoder;
   final EncoderCallback _encoder;
+  final PayloadDecoderCallback? _payloadDecoder;
 
   /// Default constructor returning the singleton instance of this class.
   const MessageSerializer({
     DecoderCallback decoder = jsonDecode,
     EncoderCallback encoder = jsonEncode,
+    PayloadDecoderCallback? payloadDecoder,
   })  : _decoder = decoder,
-        _encoder = encoder;
+        _encoder = encoder,
+        _payloadDecoder = payloadDecoder;
+
+  /// Encode a [Message] into a raw string or a Uint8List.
+  ///
+  /// If the message has a binary payload, it will be encoded using the
+  /// [BinaryDecoder.binaryEncode] method. Otherwise, the message will be
+  /// encoded using the [encoder] callback.
+  /// Given a [Message], return the raw string that would be sent through
+  /// a websocket.
+  dynamic encode(Message message) {
+    if (message.payload is Uint8List) {
+      return BinaryDecoder.binaryEncode(message);
+    }
+    return _encoder(message.encode());
+  }
 
-  /// Yield a [Message] from some raw string arriving from a websocket.
+  /// Decode a [Message] from a raw string or a Uint8List.
+  ///
+  /// If the message has a binary payload, it will be decoded using the
+  /// [BinaryDecoder.binaryDecode] method. Otherwise, the message will be
+  /// decoded using the [decoder] callback.
   Message decode(dynamic rawData) {
-    if (rawData is String || rawData is List<int>) {
+    if (rawData is String) {
       return Message.fromJson(_decoder(rawData));
+    } else if (rawData is Uint8List) {
+      final rawMap = BinaryDecoder.binaryDecode(rawData);
+      return Message(
+        joinRef: rawMap['join_ref'],
+        ref: rawMap['ref'],
+        topic: rawMap['topic'],
+        event: PhoenixChannelEvent.custom(rawMap['event']),
+        payload: _getPayload(rawMap['payload']),
+      );
     } else {
       throw ArgumentError('Received a non-string or a non-list of integers');
     }
   }
 
-  /// Given a [Message], return the raw string that would be sent through
-  /// a websocket.
-  String encode(Message message) => _encoder(message.encode());
+  dynamic _getPayload(dynamic payLoad) {
+    if (_payloadDecoder != null && payLoad is Uint8List) {
+      final deserializedPayload = _payloadDecoder!(payLoad);
+      if (deserializedPayload is Map) {
+        return MapUtils.deepConvertToStringDynamic(deserializedPayload);
+      } else if (deserializedPayload is Uint8List) {
+        return deserializedPayload;
+      } else {
+        return {'data': deserializedPayload};
+      }
+    } else {
+      return payLoad;
+    }
+  }
 }

lib/src/socket.dart

@@ -1,6 +1,7 @@
 import 'dart:async';
 import 'dart:core';
 import 'dart:math';
+import 'dart:typed_data';
 
 import 'package:logging/logging.dart';
 import 'package:phoenix_socket/src/utils/iterable_extensions.dart';
@@ -86,7 +87,7 @@ class PhoenixSocket {
 
   final BehaviorSubject<PhoenixSocketEvent> _stateStreamController =
       BehaviorSubject();
-  final StreamController<String> _receiveStreamController =
+  final StreamController<dynamic> _receiveStreamController =
       StreamController.broadcast();
   final String _endpoint;
   final StreamController<Message> _topicMessages = StreamController();
@@ -396,7 +397,7 @@ class PhoenixSocket {
   ///
   /// Used to define a custom message type for proper data decoding
   onSocketDataCallback(message) {
-    if (message is String) {
+    if (message is String || message is Uint8List) {
       if (!_receiveStreamController.isClosed) {
         _receiveStreamController.add(message);
       }

lib/src/socket_options.dart

@@ -5,14 +5,17 @@ import 'message_serializer.dart';
 /// Provided durations are all in milliseconds.
 class PhoenixSocketOptions {
   /// Create a PhoenixSocketOptions
-  const PhoenixSocketOptions({
+  PhoenixSocketOptions({
     /// The duration after which a connection attempt
     /// is considered failed
     Duration? timeout,
 
     /// The interval between heartbeat roundtrips
     Duration? heartbeat,
 
+    /// Function to decode binary payloads
+    PayloadDecoderCallback? payloadDecoder,
+
     /// The list of delays between reconnection attempts.
     ///
     /// The last duration will be repeated until it works.
@@ -38,7 +41,8 @@ class PhoenixSocketOptions {
     this.dynamicParams,
     MessageSerializer? serializer,
   })  : _timeout = timeout ?? const Duration(seconds: 10),
-        serializer = serializer ?? const MessageSerializer(),
+        serializer =
+            serializer ?? MessageSerializer(payloadDecoder: payloadDecoder),
         _heartbeat = heartbeat ?? const Duration(seconds: 30),
         assert(!(params != null && dynamicParams != null),
             "Can't set both params and dynamicParams");

lib/src/utils/map_utils.dart

@@ -0,0 +1,27 @@
+/// Utility functions for handling map operations.
+class MapUtils {
+  /// Recursively converts all keys in a map to strings and handles nested maps and lists.
+  ///
+  /// [input] is the map to be converted.
+  /// Returns a new map with all keys converted to strings and nested structures handled.
+  static Map<String, dynamic> deepConvertToStringDynamic(
+      Map<dynamic, dynamic> input) {
+    return input.map((key, value) {
+      if (value is Map) {
+        return MapEntry(key.toString(), deepConvertToStringDynamic(value));
+      } else if (value is List) {
+        return MapEntry(
+            key.toString(),
+            value.map((element) {
+              if (element is Map) {
+                return deepConvertToStringDynamic(element);
+              } else {
+                return element;
+              }
+            }).toList());
+      } else {
+        return MapEntry(key.toString(), value);
+      }
+    });
+  }
+}