feat: Support Stream<T> endpoints containing records (serverpod#3266)

Author: tp

packages/serverpod_serialization/lib/src/websocket_messages.dart

@@ -447,16 +447,18 @@ class MethodStreamMessage extends WebSocketMessage
   /// The object that was sent.
   final dynamic object;
 
+  final SerializationManager _serializationManager;
+
   /// Creates a new [MethodStreamMessage].
   /// The [object] must be an object processed by the
   /// [SerializationManager.wrapWithClassName] method.
-  MethodStreamMessage(Map data, SerializationManager serializationManager)
+  MethodStreamMessage(Map data, this._serializationManager)
       : endpoint = data[WebSocketMessageDataKey.endpoint],
         method = data[WebSocketMessageDataKey.method],
         connectionId = UuidValueJsonExtension.fromJson(
             data[WebSocketMessageDataKey.connectionId]),
         parameter = data[WebSocketMessageDataKey.parameter],
-        object = serializationManager
+        object = _serializationManager
             .deserializeByClassName(data[WebSocketMessageDataKey.object]);
 
   /// Builds a [MethodStreamMessage] message.
@@ -479,16 +481,14 @@ class MethodStreamMessage extends WebSocketMessage
   }
 
   @override
-  String toString() => WebSocketMessage._buildMessage(
-        _messageType,
-        {
-          WebSocketMessageDataKey.endpoint: endpoint,
-          WebSocketMessageDataKey.method: method,
-          WebSocketMessageDataKey.connectionId: connectionId,
-          if (parameter != null) WebSocketMessageDataKey.parameter: parameter,
-          WebSocketMessageDataKey.object: object,
-        },
-      ).toString();
+  String toString() => buildMessage(
+        endpoint: endpoint,
+        method: method,
+        connectionId: connectionId,
+        parameter: parameter,
+        object: object,
+        serializationManager: _serializationManager,
+      );
 }
 
 /// A message sent when a bad request is received.

tests/serverpod_test_client/lib/src/protocol/client.dart

@@ -2212,6 +2212,15 @@ class EndpointRecordParameters extends _i1.EndpointRef {
         {'record': _i19.mapRecordToJson(record)},
       );
 
+  _i2.Stream<(int?,)?> streamNullableRecordOfNullableInt(
+          _i2.Stream<(int?,)?> values) =>
+      caller.callStreamingServerEndpoint<_i2.Stream<(int?,)?>, (int?,)?>(
+        'recordParameters',
+        'streamNullableRecordOfNullableInt',
+        {},
+        {'values': values},
+      );
+
   _i2.Future<(int, String)> returnIntStringRecord((int, String) record) =>
       caller.callServerEndpoint<(int, String)>(
         'recordParameters',
@@ -2430,6 +2439,32 @@ class EndpointRecordParameters extends _i1.EndpointRef {
             'returnListOfNestedPositionalAndNamedRecord',
             {'recordList': _i19.mapRecordContainingContainerToJson(recordList)},
           );
+
+  _i2.Stream<
+      List<
+          (
+            (int, String), {
+            (_i10.SimpleData, double) namedSubRecord
+          })?>?> streamNullableListOfNullableNestedPositionalAndNamedRecord(
+    List<((int, String), {(_i10.SimpleData, double) namedSubRecord})?>?
+        initialValue,
+    _i2.Stream<
+            List<((int, String), {(_i10.SimpleData, double) namedSubRecord})?>?>
+        values,
+  ) =>
+      caller.callStreamingServerEndpoint<
+          _i2.Stream<
+              List<
+                  (
+                    (int, String), {
+                    (_i10.SimpleData, double) namedSubRecord
+                  })?>?>,
+          List<((int, String), {(_i10.SimpleData, double) namedSubRecord})?>?>(
+        'recordParameters',
+        'streamNullableListOfNullableNestedPositionalAndNamedRecord',
+        {'initialValue': initialValue},
+        {'values': values},
+      );
 }
 
 /// {@category Endpoint}
@@ -3003,6 +3038,92 @@ class EndpointTestTools extends _i1.EndpointRef {
         {'records': _i19.mapRecordContainingContainerToJson(records)},
       );
 
+  _i2.Stream<
+          (String, (Map<String, int>, {bool flag, _i10.SimpleData simpleData}))>
+      recordEchoStream(
+    (
+      String,
+      (Map<String, int>, {bool flag, _i10.SimpleData simpleData})
+    ) initialValue,
+    _i2.Stream<
+            (
+              String,
+              (Map<String, int>, {bool flag, _i10.SimpleData simpleData})
+            )>
+        stream,
+  ) =>
+          caller.callStreamingServerEndpoint<
+              _i2.Stream<
+                  (
+                    String,
+                    (Map<String, int>, {bool flag, _i10.SimpleData simpleData})
+                  )>,
+              (
+                String,
+                (Map<String, int>, {bool flag, _i10.SimpleData simpleData})
+              )>(
+            'testTools',
+            'recordEchoStream',
+            {'initialValue': initialValue},
+            {'stream': stream},
+          );
+
+  _i2.Stream<List<(String, int)>> listOfRecordEchoStream(
+    List<(String, int)> initialValue,
+    _i2.Stream<List<(String, int)>> stream,
+  ) =>
+      caller.callStreamingServerEndpoint<_i2.Stream<List<(String, int)>>,
+          List<(String, int)>>(
+        'testTools',
+        'listOfRecordEchoStream',
+        {'initialValue': initialValue},
+        {'stream': stream},
+      );
+
+  _i2.Stream<
+      (
+        String,
+        (Map<String, int>, {bool flag, _i10.SimpleData simpleData})
+      )?> nullableRecordEchoStream(
+    (
+      String,
+      (Map<String, int>, {bool flag, _i10.SimpleData simpleData})
+    )? initialValue,
+    _i2.Stream<
+            (
+              String,
+              (Map<String, int>, {bool flag, _i10.SimpleData simpleData})
+            )?>
+        stream,
+  ) =>
+      caller.callStreamingServerEndpoint<
+          _i2.Stream<
+              (
+                String,
+                (Map<String, int>, {bool flag, _i10.SimpleData simpleData})
+              )?>,
+          (
+            String,
+            (Map<String, int>, {bool flag, _i10.SimpleData simpleData})
+          )?>(
+        'testTools',
+        'nullableRecordEchoStream',
+        {'initialValue': initialValue},
+        {'stream': stream},
+      );
+
+  _i2.Stream<List<(String, int)>?> nullableListOfRecordEchoStream(
+    List<(String, int)>? initialValue,
+    _i2.Stream<List<(String, int)>?> stream,
+  ) =>
+      caller.callStreamingServerEndpoint<_i2.Stream<List<(String, int)>?>,
+          List<(String, int)>?>(
+        'testTools',
+        'nullableListOfRecordEchoStream',
+        {'initialValue': initialValue},
+        {'stream': stream},
+      );
+
   _i2.Future<void> logMessageWithSession() => caller.callServerEndpoint<void>(
         'testTools',
         'logMessageWithSession',