temp: remove all async dealermethods

planetarium · Dec 20, 2024 · 422e951 · 422e951
1 parent 6fe52ff
commit 422e951
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 5 deletions.
diff --git a/src/Libplanet.Net/Transports/NetMQChannel.cs b/src/Libplanet.Net/Transports/NetMQChannel.cs
@@ -65,13 +65,15 @@ public void Open()
 
         public async IAsyncEnumerable<NetMQMessage> SendMessageAsync(
             NetMQMessage message,
+            TimeSpan? timeout,
             int expectedResponses,
             [EnumeratorCancellation] CancellationToken cancellationToken)
         {
             var channel = Channel.CreateUnbounded<NetMQMessage>();
             await _requests.Writer.WriteAsync(
                 new MessageRequest(
                     message,
+                    timeout,
                     expectedResponses,
                     channel,
                     cancellationToken),
@@ -88,14 +90,19 @@ private async Task ProcessRuntime(CancellationToken ct)
         {
             using var dealer = new DealerSocket();
             dealer.Options.DisableTimeWait = true;
-            dealer.Connect(await _peer.ResolveNetMQAddressAsync());
+            var address = await _peer.ResolveNetMQAddressAsync();
+            _logger.Debug("[NetMQChannel] Connecting {Address}", address);
+            dealer.Connect(address);
             while (!ct.IsCancellationRequested)
             {
                 MessageRequest req = await _requests.Reader.ReadAsync(ct);
                 _lastUpdated = DateTimeOffset.UtcNow;
                 CancellationTokenSource linked =
                     CancellationTokenSource.CreateLinkedTokenSource(ct, req.CancellationToken);
-                _logger.Debug("[NetMQChannel] Trying to send message {Message}", req.Message);
+                _logger.Debug(
+                    "[NetMQChannel] Trying to send message {Message} (count: {ExpectedResponses})",
+                    req.Message,
+                    req.ExpectedResponses);
                 if (!dealer.TrySendMultipartMessage(req.Message))
                 {
                     _logger.Debug(
@@ -109,9 +116,16 @@ private async Task ProcessRuntime(CancellationToken ct)
 
                 foreach (var i in Enumerable.Range(0, req.ExpectedResponses))
                 {
-                    NetMQMessage raw = await dealer.ReceiveMultipartMessageAsync(
-                        cancellationToken: linked.Token
-                    );
+                    _logger.Debug(
+                        "[NetMQChannel] Waiting for replies... (#{Index})", i);
+                    var raw = new NetMQMessage();
+                    if (!dealer.TryReceiveMultipartMessage(TimeSpan.FromSeconds(5), ref raw))
+                    {
+                        break;
+                    }
+
+                    _logger.Debug(
+                        "[NetMQChannel] Successfully received replies #{Index}", i);
                     _lastUpdated = DateTimeOffset.UtcNow;
 
                     await req.Channel.Writer.WriteAsync(raw, linked.Token);
@@ -125,18 +139,22 @@ private readonly struct MessageRequest
         {
             public MessageRequest(
                 NetMQMessage message,
+                TimeSpan? timeout,
                 in int expectedResponses,
                 Channel<NetMQMessage> channel,
                 CancellationToken cancellationToken)
             {
                 Message = message;
+                Timeout = timeout;
                 ExpectedResponses = expectedResponses;
                 Channel = channel;
                 CancellationToken = cancellationToken;
             }
 
             public NetMQMessage Message { get; }
 
+            public TimeSpan? Timeout { get; }
+
             public int ExpectedResponses { get; }
 
             public Channel<NetMQMessage> Channel { get; }

diff --git a/src/Libplanet.Net/Transports/NetMQTransport.cs b/src/Libplanet.Net/Transports/NetMQTransport.cs
@@ -343,6 +343,7 @@ CancellationToken cancellationToken
 
                 await foreach (var raw in channel.SendMessageAsync(
                                    rawMessage,
+                                   timeout,
                                    expectedResponses,
                                    linkedCt))
                 {