Move 'BindingProxies' from 'NetMQTransport' into 'TurnClient'

Libplanet.Stun/Stun/TurnClient.cs

@@ -7,6 +7,7 @@
 using System.Net.Sockets;
 using System.Threading;
 using System.Threading.Tasks;
+using Libplanet.Net;
 using Libplanet.Stun.Messages;
 using Nito.AsyncEx;
 using Serilog;
@@ -218,6 +219,30 @@ public void Dispose()
             }
         }
 
+        public async Task BindingProxies(
+            CancellationToken cancellationToken,
+            int? listenPort)
+        {
+            while (!cancellationToken.IsCancellationRequested)
+            {
+#pragma warning disable IDE0067 // We'll dispose of `stream` in proxy task.
+                NetworkStream stream = await AcceptRelayedStreamAsync();
+#pragma warning restore IDE0067
+
+                // TODO We should expose the interface so that library users
+                // can limit / manage the task.
+                Func<Task> f = async () =>
+                {
+                    using var proxy = new NetworkStreamProxy(stream);
+                    await proxy.StartAsync(IPAddress.Loopback, listenPort.Value);
+                };
+
+#pragma warning disable CS4014
+                Task.Run(f).ContinueWith(_ => stream.Dispose());
+#pragma warning restore CS4014
+            }
+        }
+
         private async Task SendMessageAsync(
             NetworkStream stream,
             StunMessage message,

Libplanet/Net/NetMQTransport.cs

@@ -227,16 +227,11 @@ public async Task StartAsync(CancellationToken cancellationToken)
             _logger.Information($"Listen on {_listenPort}");
 
             _cancellationToken = cancellationToken;
-            var tasks = new List<Task>();
 
             if (!(_turnClient is null))
             {
                 _publicIPAddress = (await _turnClient.GetMappedAddressAsync()).Address;
-
-                if (await _turnClient.IsBehindNAT())
-                {
-                    _behindNAT = true;
-                }
+                _behindNAT = await _turnClient.IsBehindNAT();
             }
 
             if (_behindNAT)
@@ -246,12 +241,10 @@ public async Task StartAsync(CancellationToken cancellationToken)
                 );
                 _endPoint = new DnsEndPoint(turnEp.Address.ToString(), turnEp.Port);
 
-                // FIXME should be parameterized
-                tasks.Add(BindingProxies(_cancellationToken));
-                tasks.Add(BindingProxies(_cancellationToken));
-                tasks.Add(BindingProxies(_cancellationToken));
+                var tasks = BindingMultipleProxies(_cancellationToken, _listenPort, 3);
                 tasks.Add(RefreshAllocate(_cancellationToken));
                 tasks.Add(RefreshPermissions(_cancellationToken));
+                await await Task.WhenAny(tasks);
             }
             else if (_host is null)
             {
@@ -850,34 +843,6 @@ await dealer.SendMultipartMessageAsync(
                 DateTimeOffset.UtcNow - startedTime);
         }
 
-        // FIXME: Separate turn related features outside of Transport if possible.
-        private async Task BindingProxies(CancellationToken cancellationToken)
-        {
-            while (!cancellationToken.IsCancellationRequested)
-            {
-                try
-                {
-#pragma warning disable IDE0067  // We'll dispose of `stream` in proxy task.
-                    NetworkStream stream = await _turnClient.AcceptRelayedStreamAsync();
-#pragma warning restore IDE0067
-
-                    // TODO We should expose the interface so that library users
-                    // can limit / manage the task.
-#pragma warning disable CS4014
-                    Task.Run(async () =>
-                    {
-                        using var proxy = new NetworkStreamProxy(stream);
-                        await proxy.StartAsync(IPAddress.Loopback, _listenPort.Value);
-                    }).ContinueWith(_ => stream.Dispose());
-#pragma warning restore CS4014
-                }
-                catch (Exception e)
-                {
-                    _logger.Error(e, "An unexpected exception occurred. try again...");
-                }
-            }
-        }
-
         private void CheckStarted()
         {
             if (!Running)
@@ -1011,6 +976,16 @@ private Task RunPoller(NetMQPoller poller) =>
                 TaskScheduler.Default
             );
 
+        private List<Task> BindingMultipleProxies(
+            CancellationToken cancellationToken,
+            int? listenPort,
+            int count)
+        {
+            return Enumerable.Range(1, count)
+                .Select(x => _turnClient.BindingProxies(_cancellationToken, listenPort.Value))
+                .ToList();
+        }
+
         private readonly struct MessageRequest
         {
             private readonly int _retried;