Remove timeout function from SnapshotRequest.

Author: tosh-coding

src/AsyncFiberWorks/Channels/IRequesterSnapshotChannel.cs

@@ -15,7 +15,6 @@ public interface IRequesterSnapshotChannel<T>
         /// <param name="fiber">the target executor to receive the message</param>
         /// <param name="control"></param>
         /// <param name="receive"></param>
-        /// <param name="timeoutInMs">For initial snapshot</param>
-        IDisposable PrimedSubscribe(IExecutionContext fiber, Action<SnapshotRequestControlEvent> control, Action<T> receive, int timeoutInMs);
+        IDisposable PrimedSubscribe(IExecutionContext fiber, Action<SnapshotRequestControlEvent> control, Action<T> receive);
     }
 }

src/AsyncFiberWorks/Channels/SnapshotChannel.cs

@@ -19,10 +19,9 @@ public class SnapshotChannel<T> : ISnapshotChannel<T>
         /// <param name="fiber">the target executor to receive the message</param>
         /// <param name="control"></param>
         /// <param name="receive"></param>
-        /// <param name="timeoutInMs">For initial snapshot</param>
-        public IDisposable PrimedSubscribe(IExecutionContext fiber, Action<SnapshotRequestControlEvent> control, Action<T> receive, int timeoutInMs)
+        public IDisposable PrimedSubscribe(IExecutionContext fiber, Action<SnapshotRequestControlEvent> control, Action<T> receive)
         {
-            var requester = new SnapshotRequest<T>(fiber, control, receive, timeoutInMs);
+            var requester = new SnapshotRequest<T>(fiber, control, receive);
             requester.StartSubscribe(_requestChannel, _updatesChannel);
             return requester;
         }

src/AsyncFiberWorks/Channels/SnapshotRequest.cs

@@ -14,7 +14,6 @@ internal class SnapshotRequest<T> : IDisposable
         private readonly IExecutionContext _fiber;
         private readonly Action<SnapshotRequestControlEvent> _control;
         private readonly Action<T> _receive;
-        private readonly int _timeoutInMs;
 
         private IReply<T> _reply;
         private bool _disposed = false;
@@ -26,13 +25,11 @@ internal class SnapshotRequest<T> : IDisposable
         /// <param name="fiber">the target executor to receive the message</param>
         /// <param name="control"></param>
         /// <param name="receive"></param>
-        /// <param name="timeoutInMs">For initial snapshot</param>
-        public SnapshotRequest(IExecutionContext fiber, Action<SnapshotRequestControlEvent> control, Action<T> receive, int timeoutInMs)
+        public SnapshotRequest(IExecutionContext fiber, Action<SnapshotRequestControlEvent> control, Action<T> receive)
         {
             _fiber = fiber;
             _control = control;
             _receive = receive;
-            _timeoutInMs = timeoutInMs;
         }
 
         internal void StartSubscribe(RequestReplyChannel<object, T> requestChannel, MessageHandlerList<T> _updatesChannel)
@@ -44,21 +41,7 @@ internal void StartSubscribe(RequestReplyChannel<object, T> requestChannel, Mess
             }
             _reply = reply;
 
-            var workFiber = new PoolFiber();
-            var timeoutTimer = workFiber.Schedule(() =>
-            {
-                lock (_lock)
-                {
-                    if (_reply == null)
-                    {
-                        return;
-                    }
-                    _reply.Dispose();
-                    _reply = null;
-                }
-                _fiber.Enqueue(() => _control(SnapshotRequestControlEvent.Timeout));
-            }, _timeoutInMs);
-            reply.SetCallbackOnReceive(() => workFiber.Enqueue(() =>
+            reply.SetCallbackOnReceive(() => DefaultThreadPool.Instance.Queue((_) =>
             {
                 T result;
                 bool successToFirstReceive = reply.TryReceive(out result);
@@ -71,7 +54,6 @@ internal void StartSubscribe(RequestReplyChannel<object, T> requestChannel, Mess
                     _reply.Dispose();
                     _reply = null;
                 }
-                timeoutTimer.Dispose();
 
                 if (!successToFirstReceive)
                 {
@@ -137,11 +119,6 @@ public enum SnapshotRequestControlEvent : byte
         /// </summary>
         None = 0,
 
-        /// <summary>
-        /// A timeout occurred.
-        /// </summary>
-        Timeout = 1,
-
         /// <summary>
         /// The state changed during a connection attempt.
         /// </summary>

src/AsyncFiberWorksTests/Examples/BasicExamples.cs

@@ -223,12 +223,10 @@ public void Snapshot()
                 var requesterThread = new ThreadPoolAdaptorFromQueueForThread();
                 var fiberRequest = new PoolFiber(requesterThread, new DefaultExecutor());
                 var receivedValues = new List<int>();
+                var timeoutTimerCancellation = new Unsubscriber();
                 Action<SnapshotRequestControlEvent> actionControl = (controlEvent) =>
                 {
-                    if (controlEvent == SnapshotRequestControlEvent.Timeout)
-                    {
-                        Assert.Fail("SnapshotRequestControlEvent.Timeout");
-                    }
+                    timeoutTimerCancellation.Dispose();
                     if (controlEvent == SnapshotRequestControlEvent.Connecting)
                     {
                         return;
@@ -271,7 +269,13 @@ public void Snapshot()
                     receivedValues.Add(v);
                     Console.WriteLine("Received: " + v);
                 };
-                var handleReceive = channel.PrimedSubscribe(fiberRequest, actionControl, actionReceive, 5000);
+                var handleReceive = channel.PrimedSubscribe(fiberRequest, actionControl, actionReceive);
+                var timeoutTimer = fiberRequest.Schedule(() =>
+                {
+                    handleReceive.Dispose();
+                    Assert.Fail("SnapshotRequestControlEvent.Timeout");
+                }, 5000);
+                timeoutTimerCancellation.AppendDisposable(timeoutTimer);
 
                 requesterThread.Run();
                 handleReceive.Dispose();