Merge pull request #745 from riversdark0/feature/StreamingHubOnConnected

Add StreamingHub OnConnected support

src/MagicOnion.Server/Hubs/StreamingHub.cs

@@ -88,6 +88,15 @@ protected virtual ValueTask OnConnecting()
         return CompletedTask;
     }
 
+    /// <summary>
+    /// Called after connect (headers and marker have been sent).
+    /// Allow the server send message to the client or broadcast to group.
+    /// </summary>
+    protected virtual ValueTask OnConnected()
+    {
+        return CompletedTask;
+    }
+
     /// <summary>
     /// Called after disconnect.
     /// </summary>
@@ -158,6 +167,11 @@ async Task HandleMessageAsync()
         // NOTE: To prevent buffering by AWS ALB or reverse-proxy.
         await writer.WriteAsync(MarkerResponseBytes);
 
+        // Call OnConnected after sending the headers and marker.
+        // The server can send messages or broadcast to client after OnConnected.
+        // eg: Send the current game state to the client.
+        await OnConnected();
+
         var handlers = StreamingHubHandlerRepository.GetHandlers(Context.MethodHandler);
 
         // Main loop of StreamingHub.

tests/MagicOnion.Server.Tests/StreamingHubTest.cs

@@ -11,6 +11,7 @@ namespace MagicOnion.Server.Tests;
 
 public interface IMessageReceiver
 {
+    void VoidOnConnected(int x, string y, double z);
     //Task ZeroArgument();
     //Task OneArgument(int x);
     //Task MoreArgument(int x, string y, double z);
@@ -60,6 +61,11 @@ protected override async ValueTask OnConnecting()
         group = await Group.AddAsync("global");
     }
 
+    protected override async ValueTask OnConnected()
+    {
+        BroadcastToSelf(group).VoidOnConnected(123, "foo", 12.3f);
+    }
+
     protected override async ValueTask OnDisconnected()
     {
         if (group != null) await group.RemoveAsync(Context);
@@ -134,6 +140,15 @@ public BasicStreamingHubTest(ITestOutputHelper logger, ServerFixture<TestHub> se
         this.channel = server.DefaultChannel;
     }
 
+    [Fact]
+    public async Task OnConnected()
+    {
+        client = await StreamingHubClient.ConnectAsync<ITestHub, IMessageReceiver>(channel, this);
+        var x = await voidOnConnectedTask.Task;
+        x.Should().Be((123, "foo", 12.3f));
+        await client.DisposeAsync();
+    }
+
     [Fact]
     public async Task ZeroArgument()
     {
@@ -327,6 +342,13 @@ public async Task RetrunOneArgument3()
     //    one3Task.TrySetResult(x);
     //}
 
+    TaskCompletionSource<(int, string, double)> voidOnConnectedTask = new TaskCompletionSource<(int, string, double)>();
+    void IMessageReceiver.VoidOnConnected(int x, string y, double z)
+    {
+        voidOnConnectedTask.TrySetResult((x, y, z));
+    }
+
+
     TaskCompletionSource<(int, string, double)> voidmoreTask = new TaskCompletionSource<(int, string, double)>();
     void IMessageReceiver.VoidMoreArgument(int x, string y, double z)
     {
@@ -399,7 +421,7 @@ public async Task StatusCodeAsync()
     [StreamingHubTestFilter]
     public async Task FilterCheckAsync()
     {
-            
+
     }
 }
 
@@ -415,7 +437,7 @@ public override async ValueTask Invoke(StreamingHubContext context, Func<Streami
         context.Items["HubFilter1_BF"] = "AfterOK";
     }
 }
-    
+
 public class MoreCheckHubTest : IEmptyReceiver, IDisposable, IClassFixture<ServerFixture<MoreCheckHub>>
 {
     ITestOutputHelper logger;