add func handler

Author: doghappy

src/SocketIOClient/V2/Observers/IMyObserver.cs

@@ -1,6 +1,8 @@
+using System.Threading.Tasks;
+
 namespace SocketIOClient.V2.Observers;
 
 public interface IMyObserver<in T>
 {
-    void OnNext(T protocolMessage);
+    Task OnNextAsync(T protocolMessage);
 }

src/SocketIOClient/V2/Protocol/Http/HttpAdapter.cs

@@ -59,7 +59,7 @@ private async Task HandleResponseAsync(IHttpResponse response)
         var incomingMessage = await GetMessageAsync(response);
         foreach (var observer in _observers)
         {
-            observer.OnNext(incomingMessage);
+            await observer.OnNextAsync(incomingMessage);
         }
     }
 

src/SocketIOClient/V2/Session/HttpSession.cs

@@ -36,15 +36,16 @@ public HttpSession(
 
     public int PendingDeliveryCount => _messageQueue.Count;
 
-    public void OnNext(ProtocolMessage protocolMessage)
+    public Task OnNextAsync(ProtocolMessage protocolMessage)
     {
         if (protocolMessage.Type == ProtocolMessageType.Bytes)
         {
             OnNextBytesMessage(protocolMessage.Bytes);
-            return;
+            return Task.CompletedTask;
         }
         var messages = _engineIOAdapter.GetMessages(protocolMessage.Text);
         HandleMessages(messages);
+        return Task.CompletedTask;
     }
 
     private void HandleMessages(IEnumerable<ProtocolMessage> messages)
@@ -92,7 +93,7 @@ private void NotifyObservers(IMessage message)
     {
         foreach (var observer in _observers)
         {
-            observer.OnNext(message);
+            observer.OnNextAsync(message);
         }
     }
 

src/SocketIOClient/V2/SocketIO.cs

@@ -24,7 +24,7 @@ public class SocketIO : ISocketIO
 
 
     private readonly Dictionary<int, Action<IAckMessage>> _ackHandlers = new();
-    private readonly Dictionary<int, Func<SocketIOResponse, Task>> _funcHandlers = new();
+    private readonly Dictionary<int, Func<IAckMessage, Task>> _funcHandlers = new();
     private readonly SocketIOOptions _options;
 
 
@@ -42,15 +42,6 @@ public SocketIO(Uri uri, SocketIOOptions options)
     {
     }
 
-    private IEngineIOAdapter NewEnginIOAdapter()
-    {
-        if (_options.EIO == EngineIO.V3)
-        {
-            return new EngineIO3Adapter();
-        }
-        return new EngineIO4Adapter();
-    }
-
     public Task ConnectAsync()
     {
         _session = SessionFactory.New(_options.EIO);
@@ -81,12 +72,27 @@ public async Task EmitAsync(string eventName, Action<IAckMessage> ack)
         _ackHandlers.Add(PacketId, ack);
     }
 
-    public void OnNext(IMessage message)
+    public async Task EmitAsync(string eventName, Func<IAckMessage, Task> ack)
+    {
+        ThrowIfNotConnected();
+        PacketId++;
+        await _session.SendAsync([eventName], CancellationToken.None);
+        _funcHandlers.Add(PacketId, ack);
+    }
+
+    public async Task OnNextAsync(IMessage message)
     {
         if (message.Type == MessageType.Ack)
         {
             var ackMessage = (IAckMessage)message;
-            _ackHandlers[ackMessage.Id](ackMessage);
+            if (_ackHandlers.TryGetValue(ackMessage.Id, out var ack))
+            {
+                ack(ackMessage);
+            }
+            else if (_funcHandlers.TryGetValue(ackMessage.Id, out var func))
+            {
+                await func(ackMessage);
+            }
         }
     }
 }

tests/SocketIOClient.UnitTests/V2/Protocol/Http/HttpAdapterTests.cs

@@ -27,7 +27,7 @@ public async Task SendProtocolMessageAsync_WhenCalled_OnNextShouldBeTriggered()
 
         await _httpAdapter.SendAsync(new ProtocolMessage(), CancellationToken.None);
 
-        observer.Received().OnNext(Arg.Any<ProtocolMessage>());
+        observer.Received().OnNextAsync(Arg.Any<ProtocolMessage>());
     }
 
     [Fact]
@@ -38,6 +38,6 @@ public async Task SendHttpRequestAsync_WhenCalled_OnNextShouldBeTriggered()
 
         await _httpAdapter.SendAsync(new HttpRequest(), CancellationToken.None);
 
-        observer.Received().OnNext(Arg.Any<ProtocolMessage>());
+        observer.Received().OnNextAsync(Arg.Any<ProtocolMessage>());
     }
 }

tests/SocketIOClient.UnitTests/V2/Session/HttpSessionTests.cs

@@ -129,7 +129,7 @@ public async Task Integration_HttpAdapterPushedMessages_MessagesWillBeForwardedT
         session.Subscribe(observer);
         var captured = new List<IMessage>();
         observer
-            .When(x => x.OnNext(Arg.Any<IMessage>()))
+            .When(x => x.OnNextAsync(Arg.Any<IMessage>()))
             .Do(call => captured.Add(call.Arg<IMessage>()));
 
         await httpAdapter.SendAsync(new ProtocolMessage(), CancellationToken.None);
@@ -164,9 +164,9 @@ public void OnNext_BinaryMessageIsNotReady_NoMessageWillBePushed()
         };
         _engineIOAdapter.GetMessages(Arg.Any<string>()).Returns([protocolMessage]);
 
-        _session.OnNext(protocolMessage);
+        _session.OnNextAsync(protocolMessage);
 
-        observer.Received(0).OnNext(Arg.Any<IMessage>());
+        observer.Received(0).OnNextAsync(Arg.Any<IMessage>());
         _session.PendingDeliveryCount.Should().Be(1);
     }
 
@@ -194,12 +194,12 @@ public void OnNext_BinaryMessageReady_MessageWillBePushed()
                     Type = ProtocolMessageType.Bytes,
                 },
             ]);
-        _session.OnNext(new ProtocolMessage
+        _session.OnNextAsync(new ProtocolMessage
         {
             Type = ProtocolMessageType.Text,
         });
 
-        observer.Received(1).OnNext(Arg.Any<IBinaryMessage>());
+        observer.Received(1).OnNextAsync(Arg.Any<IBinaryMessage>());
         _session.PendingDeliveryCount.Should().Be(0);
     }
 
@@ -223,12 +223,12 @@ public void OnNext_BinaryAckMessageIsNotReady_NoMessageWillBePushed()
                     Type = ProtocolMessageType.Text,
                 },
             ]);
-        _session.OnNext(new ProtocolMessage
+        _session.OnNextAsync(new ProtocolMessage
         {
             Type = ProtocolMessageType.Text,
         });
 
-        observer.Received(0).OnNext(Arg.Any<IMessage>());
+        observer.Received(0).OnNextAsync(Arg.Any<IMessage>());
         _session.PendingDeliveryCount.Should().Be(1);
     }
 
@@ -253,16 +253,16 @@ public void OnNext_BinaryAckMessageReady_MessageWillBePushed()
                     Type = ProtocolMessageType.Text,
                 },
             ]);
-        _session.OnNext(new ProtocolMessage
+        _session.OnNextAsync(new ProtocolMessage
         {
             Type = ProtocolMessageType.Text,
         });
-        _session.OnNext(new ProtocolMessage
+        _session.OnNextAsync(new ProtocolMessage
         {
             Type = ProtocolMessageType.Bytes,
         });
 
-        observer.Received(1).OnNext(Arg.Any<IBinaryMessage>());
+        observer.Received(1).OnNextAsync(Arg.Any<IBinaryMessage>());
         _session.PendingDeliveryCount.Should().Be(0);
     }
 }

tests/SocketIOClient.UnitTests/V2/SocketIOTests.cs

@@ -35,7 +35,7 @@ await _io.Invoking(x => x.EmitAsync("event", _ => { }))
     }
 
     [Fact]
-    public async Task EmitAsync_AckEvent_PacketIdIncrementBy1()
+    public async Task EmitAsync_AckEventAction_PacketIdIncrementBy1()
     {
         await _io.ConnectAsync();
         await _io.EmitAsync("event", _ => { });
@@ -44,7 +44,7 @@ public async Task EmitAsync_AckEvent_PacketIdIncrementBy1()
     }
 
     [Fact]
-    public async Task EmitAsync_AckEventAndGotResponse_HandlerIsCalled()
+    public async Task EmitAsync_AckEventActionAndGotResponse_HandlerIsCalled()
     {
         var ackCalled = false;
 
@@ -54,7 +54,36 @@ public async Task EmitAsync_AckEventAndGotResponse_HandlerIsCalled()
         {
             Id = _io.PacketId,
         };
-        _io.OnNext(ackMessage);
+        _io.OnNextAsync(ackMessage);
+
+        ackCalled.Should().BeTrue();
+    }
+
+    [Fact]
+    public async Task EmitAsync_AckEventFunc_PacketIdIncrementBy1()
+    {
+        await _io.ConnectAsync();
+        await _io.EmitAsync("event", _ => Task.CompletedTask);
+
+        _io.PacketId.Should().Be(1);
+    }
+
+    [Fact]
+    public async Task EmitAsync_AckEventFuncAndGotResponse_HandlerIsCalled()
+    {
+        var ackCalled = false;
+
+        await _io.ConnectAsync();
+        await _io.EmitAsync("event", _ =>
+        {
+            ackCalled = true;
+            return Task.CompletedTask;
+        });
+        var ackMessage = new SystemJsonAckMessage
+        {
+            Id = _io.PacketId,
+        };
+        _io.OnNextAsync(ackMessage);
 
         ackCalled.Should().BeTrue();
     }