WIP

Author: mayuki

Directory.Packages.props

@@ -9,7 +9,7 @@
     <MessagePackVersion>3.1.4</MessagePackVersion>
     <MicrosoftCodeAnalysisVersion>4.3.1</MicrosoftCodeAnalysisVersion>
     <MicrosoftCodeAnalysisVersionUnity>3.9.0</MicrosoftCodeAnalysisVersionUnity>
-    <MulticasterVersion>2.1.1</MulticasterVersion>
+    <MulticasterVersion>0.1.0-ci-experimental-altchannel-20260209-063841</MulticasterVersion>
   </PropertyGroup>
   <ItemGroup>
     <PackageVersion Include="BenchmarkDotNet" Version="0.13.12" />

NuGet.Config

@@ -2,8 +2,14 @@
 <configuration>
   <packageSourceMapping>
     <clear />
+    <packageSource key="Canary-Build">
+      <package pattern="Multicaster*" />
+    </packageSource>
     <packageSource key="nuget.org">
       <package pattern="*" />
     </packageSource>
   </packageSourceMapping>
+  <packageSources>
+    <add key="Canary-Build" value="https://pkgs.dev.azure.com/cysharp/Public/_packaging/Canary-Build/nuget/v3/index.json" />
+  </packageSources>
 </configuration>

src/MagicOnion.Client.SourceGenerator/CodeGen/StaticStreamingHubClientGenerator.cs

@@ -115,32 +115,32 @@ static void EmitHelperMethods(StreamingHubClientBuildContext ctx)
         if (ctx.EnableStreamingHubDiagnosticHandler)
         {
             ctx.Writer.AppendLineWithFormat($$"""
-                            global::System.Threading.Tasks.Task<TResponse> WriteMessageWithResponseDiagnosticTaskAsync<TRequest, TResponse>(int methodId, TRequest message, global::MagicOnion.TransportReliability reliability = global::MagicOnion.TransportReliability.Reliable, [global::System.Runtime.CompilerServices.CallerMemberName] string callerMemberName = default!)
+                            global::System.Threading.Tasks.Task<TResponse> WriteMessageWithResponseDiagnosticTaskAsync<TRequest, TResponse>(int methodId, TRequest message, [global::System.Runtime.CompilerServices.CallerMemberName] string callerMemberName = default!)
                             {
                                 if (diagnosticHandler is null)
                                 {
-                                    return base.WriteMessageWithResponseTaskAsync<TRequest, TResponse>(methodId, message, reliability);
+                                    return base.WriteMessageWithResponseTaskAsync<TRequest, TResponse>(methodId, message);
                                 }
 
                                 return diagnosticHandler.OnMethodInvoke(this, methodId, callerMemberName, message, isFireAndForget: true, base.WriteMessageWithResponseValueTaskOfTAsync<TRequest, TResponse>).AsTask();
                             }
 
-                            async global::System.Threading.Tasks.ValueTask WriteMessageWithResponseDiagnosticValueTaskAsync<TRequest, TResponse>(int methodId, TRequest message, global::MagicOnion.TransportReliability reliability = global::MagicOnion.TransportReliability.Reliable, [global::System.Runtime.CompilerServices.CallerMemberName] string callerMemberName = default!)
+                            async global::System.Threading.Tasks.ValueTask WriteMessageWithResponseDiagnosticValueTaskAsync<TRequest, TResponse>(int methodId, TRequest message, [global::System.Runtime.CompilerServices.CallerMemberName] string callerMemberName = default!)
                             {
                                 if (diagnosticHandler is null)
                                 {
-                                    await base.WriteMessageWithResponseValueTaskAsync<TRequest, TResponse>(methodId, message, reliability);
+                                    await base.WriteMessageWithResponseValueTaskAsync<TRequest, TResponse>(methodId, message);
                                     return;
                                 }
 
                                 await diagnosticHandler.OnMethodInvoke(this, methodId, callerMemberName, message, isFireAndForget: true, base.WriteMessageWithResponseValueTaskOfTAsync<TRequest, TResponse>);
                             }
 
-                            global::System.Threading.Tasks.ValueTask<TResponse> WriteMessageWithResponseDiagnosticValueTaskOfTAsync<TRequest, TResponse>(int methodId, TRequest message, global::MagicOnion.TransportReliability reliability = global::MagicOnion.TransportReliability.Reliable, [global::System.Runtime.CompilerServices.CallerMemberName] string callerMemberName = default!)
+                            global::System.Threading.Tasks.ValueTask<TResponse> WriteMessageWithResponseDiagnosticValueTaskOfTAsync<TRequest, TResponse>(int methodId, TRequest message, [global::System.Runtime.CompilerServices.CallerMemberName] string callerMemberName = default!)
                             {
                                 if (diagnosticHandler is null)
                                 {
-                                    return base.WriteMessageWithResponseValueTaskOfTAsync<TRequest, TResponse>(methodId, message, reliability);
+                                    return base.WriteMessageWithResponseValueTaskOfTAsync<TRequest, TResponse>(methodId, message);
                                 }
 
                                 return diagnosticHandler.OnMethodInvoke(this, methodId, callerMemberName, message, isFireAndForget: true, base.WriteMessageWithResponseValueTaskOfTAsync<TRequest, TResponse>);

src/MagicOnion.Client/Internal/ClientDataChannel.cs

@@ -162,7 +162,7 @@ async Task RunReceiveLoop()
                             SendAckFromClient();
                             connectedTcs.TrySetResult(true);
                             break;
-                        case 0x03: // Data
+                        case 0x11: // Data (from Server)
                             if (result.Buffer.Length > 1 + 8 + 8 /* Sequence */)
                             {
                                 var sequence = BitConverter.ToUInt64(result.Buffer, 1 + 8);

src/MagicOnion.Client/StreamingHubClientBase.cs

@@ -277,6 +277,7 @@ internal async Task __ConnectAndSubscribeAsync(CancellationToken connectAndSubsc
             }
             catch (Exception ex)
             {
+                // TODO: Error logging / retry?
                 dataChannel.Dispose();
                 dataChannel = null;
             }
@@ -330,13 +331,50 @@ void EnsureConnected()
         }
     }
 
+    readonly Channel<StreamingHubPayload> receivedDataQueue = Channel.CreateUnbounded<StreamingHubPayload>();
+
+    async Task RunDataConsumerLoopAsync(SynchronizationContext? syncContext, CancellationToken cancellationToken)
+    {
+        while (!cancellationToken.IsCancellationRequested)
+        {
+            if (!await receivedDataQueue.Reader.WaitToReadAsync().ConfigureAwait(false))
+            {
+                break;
+            }
+
+            while (receivedDataQueue.Reader.TryRead(out var payload))
+            {
+                ConsumeData(syncContext, payload);
+            }
+        }
+    }
+
+    async Task RunDataChannelReaderLoopAsync(CancellationToken cancellationToken)
+    {
+        if (dataChannel is null) throw new InvalidOperationException();
+        while (!cancellationToken.IsCancellationRequested && await dataChannel.DataReader.WaitToReadAsync().ConfigureAwait(false))
+        {
+            while (dataChannel.DataReader.TryRead(out var payload))
+            {
+                receivedDataQueue.Writer.TryWrite(payload);
+            }
+        }
+    }
+
     async Task StartSubscribe(SynchronizationContext? syncContext, Task<bool> firstMoveNext, CancellationToken subscriptionToken)
     {
         EnsureConnected();
 
         var disconnectionReason = new DisconnectionReason(DisconnectionType.CompletedNormally, null);
         writerTask = RunWriterLoopAsync(subscriptionToken);
 
+        RunDataConsumerLoopAsync(syncContext, subscriptionToken);
+
+        if (dataChannel is not null)
+        {
+            RunDataChannelReaderLoopAsync(subscriptionToken);
+        }
+
         var reader = this.reader;
         try
         {
@@ -345,7 +383,7 @@ async Task StartSubscribe(SynchronizationContext? syncContext, Task<bool> firstM
             {
                 try
                 {
-                    ConsumeData(syncContext, reader.Current);
+                    receivedDataQueue.Writer.TryWrite(reader.Current);
                 }
                 catch (Exception ex)
                 {
@@ -393,6 +431,7 @@ async Task StartSubscribe(SynchronizationContext? syncContext, Task<bool> firstM
         }
         finally
         {
+            receivedDataQueue.Writer.TryComplete();
             disconnected = true;
 
             try

src/MagicOnion.Server/Hubs/Internal/DataChannel/DataChannelServer.cs

@@ -76,7 +76,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
                         channel.SetConnected();
                         break;
                     case 0x10:
-                        // Data
+                        // Data (from Client)
                         data = data.Slice(9);
                         if (data.Length < 8 + 1)
                         {

src/MagicOnion.Server/Hubs/Internal/DataChannel/ServerDataChannel.cs

@@ -102,7 +102,7 @@ public void SendPayload(StreamingHubPayload payload)
         var span = array.AsSpan(0, length);
         try
         {
-            span[0] = 0x10; // Data
+            span[0] = 0x11; // Data (from Server)
             BitConverter.TryWriteBytes(span.Slice(1, 8), SessionId);
             BitConverter.TryWriteBytes(span.Slice(9, 8), outgoingSequence);
             payload.Span.CopyTo(span.Slice(17, payload.Length));

src/MagicOnion.Server/Hubs/Internal/HubReceiverMethodReliabilityMap.cs

@@ -1,5 +1,6 @@
 using System.Collections.Frozen;
 using System.Reflection;
+using MagicOnion.Internal;
 
 namespace MagicOnion.Server.Hubs.Internal;
 
@@ -17,7 +18,7 @@ public static HubReceiverMethodReliabilityMap Create<TReceiver>()
         var reliabilityByMethods = targetType.GetMethods()
             .Concat(targetType.GetInterfaces().SelectMany(x => x.GetMethods()))
             .Select(x => (x.Name, x.ReturnType, Reliability: x.GetCustomAttribute<TransportAttribute>()?.Reliability ?? reliabilityForType))
-            .ToDictionary(k => k.Name, v => (v.Reliability, v.ReturnType, MethodId: v.Name.GetHashCode()));
+            .ToDictionary(k => k.Name, v => (v.Reliability, v.ReturnType, MethodId: FNV1A32.GetHashCode(v.Name))); // TODO: MethodIdAttribute
 
         // Validate
         if (reliabilityByMethods.FirstOrDefault(x => IsAwaitable(x.Value.ReturnType) && x.Value.Reliability != TransportReliability.Reliable) is { Key: not null } method)

src/MagicOnion.Server/Hubs/Internal/MagicOnionRemoteReceiverWriter.cs

@@ -1,24 +1,42 @@
 using Cysharp.Runtime.Multicast.Remoting;
 using MagicOnion.Internal;
+using MagicOnion.Server.Hubs.Internal.DataChannel;
 
 namespace MagicOnion.Server.Hubs.Internal;
 
 internal class MagicOnionRemoteReceiverWriter : IRemoteReceiverWriter
 {
     readonly StreamingServiceContext<StreamingHubPayload, StreamingHubPayload> writer;
     readonly HubReceiverMethodReliabilityMap reliabilityMap;
+    readonly ServerDataChannel? dataChannel;
 
     public IRemoteClientResultPendingTaskRegistry PendingTasks { get; }
 
-    public MagicOnionRemoteReceiverWriter(StreamingServiceContext<StreamingHubPayload, StreamingHubPayload> writer, IRemoteClientResultPendingTaskRegistry pendingTasks, HubReceiverMethodReliabilityMap reliabilityMap)
+    public MagicOnionRemoteReceiverWriter(
+        StreamingServiceContext<StreamingHubPayload, StreamingHubPayload> writer,
+        IRemoteClientResultPendingTaskRegistry pendingTasks,
+        HubReceiverMethodReliabilityMap reliabilityMap,
+        ServerDataChannel? dataChannel)
     {
         this.writer = writer;
         PendingTasks = pendingTasks;
         this.reliabilityMap = reliabilityMap;
+        this.dataChannel = dataChannel;
     }
 
-    public void Write(ReadOnlyMemory<byte> payload)
+    public void Write(InvocationWriteContext context)
     {
-        writer.QueueResponseStreamWrite(StreamingHubPayloadPool.Shared.RentOrCreate(payload.Span));
+        var payload = StreamingHubPayloadPool.Shared.RentOrCreate(context.Payload.Span);
+        if (dataChannel is not null &&
+            reliabilityMap.ReliabilityByMethodId.TryGetValue(context.MethodId, out var reliability) &&
+            reliability != TransportReliability.Reliable)
+        {
+            // Unreliable or Reliable unordered
+            dataChannel.SendPayload(payload);
+        }
+        else
+        {
+            writer.QueueResponseStreamWrite(payload);
+        }
     }
 }

src/MagicOnion.Server/Hubs/StreamingHub.cs

@@ -47,6 +47,8 @@ public abstract class StreamingHubBase<THubInterface, TReceiver> : ServiceBase<T
     // HACK: If the ID of the message is `-1`, the client will ignore the message.
     static ReadOnlySpan<byte> MarkerResponseBytes => [0x93, 0xff, 0x00, 0x0c]; // MsgPack: [-1, 0, nil]
 
+    static readonly HubReceiverMethodReliabilityMap reliabilityMap = HubReceiverMethodReliabilityMap.Create<TReceiver>();
+
     readonly Channel<(StreamingHubPayload Payload, UniqueHashDictionary<StreamingHubHandler> Handlers, int MethodId, int MessageId, ReadOnlyMemory<byte> Body, bool HasResponse)> requests
         = Channel.CreateBounded<(StreamingHubPayload, UniqueHashDictionary<StreamingHubHandler>, int, int, ReadOnlyMemory<byte>, bool)>(new BoundedChannelOptions(capacity: 10)
         {
@@ -106,12 +108,11 @@ async Task<DuplexStreamingResult<StreamingHubPayload, StreamingHubPayload>> IStr
         handlers = streamingHubFeature.Handlers;
 
         var isDataChannelCreated = TryCreateDataChannel();
-        var reliabilityMap = HubReceiverMethodReliabilityMap.Create<TReceiver>();
 
         var remoteProxyFactory = serviceProvider.GetRequiredService<IRemoteProxyFactory>();
         var remoteSerializer = serviceProvider.GetRequiredService<IRemoteSerializer>();
         this.remoteClientResultPendingTasks = serviceProvider.GetRequiredService<IRemoteClientResultPendingTaskRegistry>();
-        this.Client = remoteProxyFactory.CreateDirect<TReceiver>(new MagicOnionRemoteReceiverWriter(StreamingServiceContext, remoteClientResultPendingTasks, reliabilityMap), remoteSerializer);
+        this.Client = remoteProxyFactory.CreateDirect<TReceiver>(new MagicOnionRemoteReceiverWriter(StreamingServiceContext, remoteClientResultPendingTasks, reliabilityMap, isDataChannelCreated ? dataChannel : null), remoteSerializer);
 
         this.Group = new HubGroupRepository<TReceiver>(Client, StreamingServiceContext, streamingHubFeature.GroupProvider);