feat: Support method streaming. (serverpod#2567)

BenAuerDev · Aug 6, 2024 · c048a22 · c048a22
1 parent b595b1a
commit c048a22
Show file tree

Hide file tree

Showing 32 changed files with 2,735 additions and 63 deletions.
diff --git a/examples/auth_example/auth_example_client/lib/src/protocol/client.dart b/examples/auth_example/auth_example_client/lib/src/protocol/client.dart
@@ -48,6 +48,7 @@ class Client extends _i1.ServerpodClient {
       StackTrace,
     )? onFailedCall,
     Function(_i1.MethodCallContext)? onSucceededCall,
+    bool? disconnectStreamsOnLostInternetConnection,
   }) : super(
           host,
           _i4.Protocol(),
@@ -57,6 +58,8 @@ class Client extends _i1.ServerpodClient {
           connectionTimeout: connectionTimeout,
           onFailedCall: onFailedCall,
           onSucceededCall: onSucceededCall,
+          disconnectStreamsOnLostInternetConnection:
+              disconnectStreamsOnLostInternetConnection,
         ) {
     example = EndpointExample(this);
     modules = _Modules(this);

diff --git a/examples/chat/chat_client/lib/src/protocol/client.dart b/examples/chat/chat_client/lib/src/protocol/client.dart
@@ -54,6 +54,7 @@ class Client extends _i1.ServerpodClient {
       StackTrace,
     )? onFailedCall,
     Function(_i1.MethodCallContext)? onSucceededCall,
+    bool? disconnectStreamsOnLostInternetConnection,
   }) : super(
           host,
           _i6.Protocol(),
@@ -63,6 +64,8 @@ class Client extends _i1.ServerpodClient {
           connectionTimeout: connectionTimeout,
           onFailedCall: onFailedCall,
           onSucceededCall: onSucceededCall,
+          disconnectStreamsOnLostInternetConnection:
+              disconnectStreamsOnLostInternetConnection,
         ) {
     channels = EndpointChannels(this);
     modules = _Modules(this);

diff --git a/...serverpod/lib/src/server/websocket_request_handlers/method_websocket_request_handler.dart b/...serverpod/lib/src/server/websocket_request_handlers/method_websocket_request_handler.dart
@@ -121,6 +121,8 @@ class MethodWebsocketRequestHandler {
       );
       return OpenMethodStreamResponse.buildMessage(
         connectionId: message.connectionId,
+        endpoint: message.endpoint,
+        method: message.method,
         responseType: OpenMethodStreamResponseType.endpointNotFound,
       );
     }
@@ -133,6 +135,8 @@ class MethodWebsocketRequestHandler {
       );
       return OpenMethodStreamResponse.buildMessage(
         connectionId: message.connectionId,
+        endpoint: message.endpoint,
+        method: message.method,
         responseType: OpenMethodStreamResponseType.endpointNotFound,
       );
     }
@@ -142,6 +146,8 @@ class MethodWebsocketRequestHandler {
         'Endpoint method is not a valid stream method: $message',
       );
       return OpenMethodStreamResponse.buildMessage(
+        endpoint: message.endpoint,
+        method: message.method,
         connectionId: message.connectionId,
         responseType: OpenMethodStreamResponseType.endpointNotFound,
       );
@@ -160,6 +166,8 @@ class MethodWebsocketRequestHandler {
         'Failed to parse parameters for open stream request: $message',
       );
       return OpenMethodStreamResponse.buildMessage(
+        endpoint: message.endpoint,
+        method: message.method,
         connectionId: message.connectionId,
         responseType: OpenMethodStreamResponseType.invalidArguments,
       );
@@ -176,6 +184,8 @@ class MethodWebsocketRequestHandler {
         'Failed to parse input streams for open stream request: $message',
       );
       return OpenMethodStreamResponse.buildMessage(
+        endpoint: message.endpoint,
+        method: message.method,
         connectionId: message.connectionId,
         responseType: OpenMethodStreamResponseType.invalidArguments,
       );
@@ -206,11 +216,15 @@ class MethodWebsocketRequestHandler {
       return switch (authFailed.reason) {
         AuthenticationFailureReason.insufficientAccess =>
           OpenMethodStreamResponse.buildMessage(
+            endpoint: message.endpoint,
+            method: message.method,
             connectionId: message.connectionId,
             responseType: OpenMethodStreamResponseType.authorizationDeclined,
           ),
         AuthenticationFailureReason.unauthenticated =>
           OpenMethodStreamResponse.buildMessage(
+            endpoint: message.endpoint,
+            method: message.method,
             connectionId: message.connectionId,
             responseType: OpenMethodStreamResponseType.authenticationFailed,
           ),
@@ -228,6 +242,8 @@ class MethodWebsocketRequestHandler {
     );
 
     return OpenMethodStreamResponse.buildMessage(
+      endpoint: message.endpoint,
+      method: message.method,
       connectionId: message.connectionId,
       responseType: OpenMethodStreamResponseType.success,
     );
@@ -574,6 +590,8 @@ class _MethodStreamManager {
     required OpenMethodStreamCommand message,
     required Server server,
   }) {
+    /// Will never be stopped listened to if the method does not return.
+    /// This is a potential memory leak.
     methodConnector.call(session, args, streamParams).listen(
       (value) {
         _postMessage(