add NewPongMessage method and integrate retry policy in EngineIO4Adapter

Author: doghappy

src/SocketIOClient.Serializer/BaseJsonSerializer.cs

@@ -70,6 +70,14 @@ public ProtocolMessage NewPingMessage()
         };
     }
 
+    public ProtocolMessage NewPongMessage()
+    {
+        return new ProtocolMessage
+        {
+            Text = "3",
+        };
+    }
+
     protected abstract IMessage NewMessage(MessageType type, string text);
 
     private static List<ProtocolMessage> GetSerializeResult(string text, IEnumerable<byte[]> bytes)

src/SocketIOClient.Serializer/ISerializer.cs

@@ -10,5 +10,6 @@ public interface ISerializer
         List<ProtocolMessage> Serialize(object[] data);
         IMessage Deserialize(string text);
         ProtocolMessage NewPingMessage();
+        ProtocolMessage NewPongMessage();
     }
 }

src/SocketIOClient/V2/DefaultSessionFactory.cs

@@ -26,7 +26,7 @@ public ISession New(EngineIO eio, SessionOptions options)
         var randomDelayRetryPolicy = new RandomDelayRetryPolicy(random);
         IEngineIOAdapter engineIOAdapter = eio == EngineIO.V3
             ? new EngineIO3Adapter(stopwatch, serializer, httpAdapter, options.Timeout, randomDelayRetryPolicy)
-            : new EngineIO4Adapter(stopwatch, httpAdapter, options.Timeout);
+            : new EngineIO4Adapter(stopwatch, serializer, httpAdapter, options.Timeout, randomDelayRetryPolicy);
         return new HttpSession(
             options,
             engineIOAdapter,

src/SocketIOClient/V2/Session/EngineIOHttpAdapter/EngineIO4Adapter.cs

@@ -5,6 +5,7 @@
 using System.Threading.Tasks;
 using SocketIOClient.Core;
 using SocketIOClient.Core.Messages;
+using SocketIOClient.Serializer;
 using SocketIOClient.V2.Infrastructure;
 using SocketIOClient.V2.Observers;
 using SocketIOClient.V2.Protocol;
@@ -16,19 +17,25 @@ public class EngineIO4Adapter : IEngineIOAdapter
 {
     public EngineIO4Adapter(
         IStopwatch stopwatch,
+        ISerializer serializer,
         IProtocolAdapter protocolAdapter,
-        TimeSpan timeout)
+        TimeSpan timeout,
+        IRetriable retryPolicy)
     {
         _stopwatch = stopwatch;
+        _serializer = serializer;
         _protocolAdapter = protocolAdapter;
         _timeout = timeout;
+        _retryPolicy = retryPolicy;
     }
 
     private const string Delimiter = "\u001E";
     private readonly IStopwatch _stopwatch;
+    private readonly ISerializer _serializer;
     private readonly IProtocolAdapter _protocolAdapter;
     private readonly List<IMyObserver<IMessage>> _observers = [];
     private readonly TimeSpan _timeout;
+    private readonly IRetriable _retryPolicy;
 
     public IHttpRequest ToHttpRequest(ICollection<byte[]> bytes)
     {
@@ -96,12 +103,13 @@ public async Task ProcessMessageAsync(IMessage message)
 
     private async Task HandlePingMessageAsync()
     {
+        var pongProtocolMessage = _serializer.NewPongMessage();
         _stopwatch.Restart();
-        using var cts = new CancellationTokenSource(_timeout);
-        await _protocolAdapter.SendAsync(new ProtocolMessage
+        await _retryPolicy.RetryAsync(3, async () =>
         {
-            Text = "3",
-        }, cts.Token);
+            using var cts = new CancellationTokenSource(_timeout);
+            await _protocolAdapter.SendAsync(pongProtocolMessage, cts.Token);
+        });
         _stopwatch.Stop();
         var pong = new PongMessage
         {

tests/SocketIOClient.Serializer.NewtonsoftJson.Tests/NewtonJsonSerializerTests.cs

@@ -505,4 +505,16 @@ public void NewPingMessage_WhenCalled_ReturnPingMessage()
                 Text = "2",
             });
     }
+
+    [Fact]
+    public void NewPongMessage_WhenCalled_ReturnPingMessage()
+    {
+        var pong = _serializer.NewPongMessage();
+        pong.Should()
+            .BeEquivalentTo(new ProtocolMessage
+            {
+                Type = ProtocolMessageType.Text,
+                Text = "3",
+            });
+    }
 }

tests/SocketIOClient.UnitTests/V2/Serializer/SystemTextJson/SystemJsonSerializerTests.cs

@@ -516,4 +516,16 @@ public void NewPingMessage_WhenCalled_ReturnPingMessage()
                 Text = "2",
             });
     }
+
+    [Fact]
+    public void NewPongMessage_WhenCalled_ReturnPingMessage()
+    {
+        var pong = _serializer.NewPongMessage();
+        pong.Should()
+            .BeEquivalentTo(new ProtocolMessage
+            {
+                Type = ProtocolMessageType.Text,
+                Text = "3",
+            });
+    }
 }

tests/SocketIOClient.UnitTests/V2/Session/EngineIOHttpAdapter/EngineIO4AdapterTests.cs

@@ -2,8 +2,11 @@
 using JetBrains.Annotations;
 using NSubstitute;
 using SocketIOClient.Core;
+using SocketIOClient.Core.Messages;
+using SocketIOClient.Serializer;
 using SocketIOClient.V2;
 using SocketIOClient.V2.Infrastructure;
+using SocketIOClient.V2.Observers;
 using SocketIOClient.V2.Protocol;
 using SocketIOClient.V2.Protocol.Http;
 using SocketIOClient.V2.Session.EngineIOHttpAdapter;
@@ -15,13 +18,22 @@ public class EngineIO4AdapterTests
     public EngineIO4AdapterTests()
     {
         _stopwatch = Substitute.For<IStopwatch>();
+        _serializer = Substitute.For<ISerializer>();
         _protocolAdapter = Substitute.For<IProtocolAdapter>();
-        _adapter = new EngineIO4Adapter(_stopwatch, _protocolAdapter, TimeSpan.FromSeconds(1));
+        _retryPolicy = Substitute.For<IRetriable>();
+        _adapter = new EngineIO4Adapter(
+            _stopwatch,
+            _serializer,
+            _protocolAdapter,
+            TimeSpan.FromSeconds(1),
+            _retryPolicy);
     }
 
     private readonly IStopwatch _stopwatch;
+    private readonly ISerializer _serializer;
     private readonly IProtocolAdapter _protocolAdapter;
     private readonly EngineIO4Adapter _adapter;
+    private readonly IRetriable _retryPolicy;
 
     [Fact]
     public void ToHttpRequest_GivenAnEmptyArray_ThrowException()
@@ -162,4 +174,18 @@ public void ToHttpRequest_GivenValidContent_ReturnSameBodyText(string content, s
         var req = _adapter.ToHttpRequest(content);
         req.BodyText.Should().Be(expected);
     }
+
+    [Fact]
+    public async Task ProcessMessageAsync_PingMessage_NotifyObserverWithDuration()
+    {
+        var observer = Substitute.For<IMyObserver<IMessage>>();
+        _adapter.Subscribe(observer);
+        _stopwatch.Elapsed.Returns(TimeSpan.FromSeconds(1));
+
+        await _adapter.ProcessMessageAsync(new PingMessage());
+
+        await observer
+            .Received(1)
+            .OnNextAsync(Arg.Is<IMessage>(m => ((PongMessage)m).Duration == TimeSpan.FromSeconds(1)));
+    }
 }