More impl with unit tests

csharp/Practice01-End/src/MyClient/Driver/MyDriverClient.cs

@@ -13,6 +13,8 @@ internal interface IMyDriverClient : IDisposable
         void AddQuery(int queryId);
 
         void RemoveQuery(int queryId);
+
+        MyData Receive();
     }
 
     internal interface IMyDriverClientFactory
@@ -54,6 +56,11 @@ public void RemoveQuery(int queryId)
             this._client.RemoveQuery(queryId);
         }
 
+        public MyData Receive()
+        {
+            return this._client.Receive();
+        }
+
         public void Dispose()
         {
             this._client.Dispose();

csharp/Practice01-End/src/MyClient/MyClient.csproj

@@ -51,6 +51,7 @@
     <Compile Include="IMySubscriber.cs" />
     <Compile Include="MyConnection.cs" />
     <Compile Include="MyConnector.cs" />
+    <Compile Include="MyReceiver.cs" />
     <Compile Include="MyRequest.cs" />
     <Compile Include="MyRequestSender.cs" />
     <Compile Include="MySubscription.cs" />

csharp/Practice01-End/src/MyClient/MyReceiver.cs

@@ -0,0 +1,76 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Collections.Concurrent;
+using MyDriver;
+using System.Threading;
+using log4net;
+
+namespace MyClient
+{
+    internal interface IMyReceiver
+    {
+        void Process();
+
+        BlockingCollection<MyData> DataProduced { get; }
+
+        CancellationToken CancellationToken { get; }
+    }
+
+    internal class MyReceiver : IMyReceiver
+    {
+        private static ILog Logger = LogManager.GetLogger(typeof(MyReceiver));
+
+        private IMyConnector _connector;
+        private CancellationTokenSource _cts;
+
+        public MyReceiver(IMyConnector connector)
+        {
+            this._connector = connector;
+            this._cts = new CancellationTokenSource();
+            this.DataProduced = new BlockingCollection<MyData>();
+        }
+
+        public void Process()
+        {
+            var client = this._connector.Client;
+            if (client == null)
+            {
+                Logger.Info("Connector isn't prepared, stop processing.");
+                return;
+            }
+
+            while (true)
+            {
+                MyData data;
+
+                try
+                {
+                    data = client.Receive();
+                }
+                catch (MyDriverException ex)
+                {
+                    Logger.Error("Exception occurred when receiving, close the client and stop processing.", ex);
+
+                    this._connector.CloseClient();
+                    this._cts.Cancel();
+                    return;
+                }
+
+                if (data == null)
+                {
+                    Logger.Info("The client is closed by others, stop processing.");
+                    this._cts.Cancel();
+                    return;
+                }
+
+                this.DataProduced.Add(data);
+            }
+        }
+
+        public BlockingCollection<MyData> DataProduced { get; private set; }
+
+        public CancellationToken CancellationToken { get { return this._cts.Token; } }
+    }
+}

csharp/Practice01-End/src/MyClient/MyRequestSender.cs

@@ -34,7 +34,7 @@ public void Process()
             var client = this._connector.Client;
             if (client == null)
             {
-                Logger.Info("Connector isn't prepared. stop processing.");
+                Logger.Info("Connector isn't prepared, stop processing.");
                 return;
             }
 

csharp/Practice01-End/tests/MyClient.Tests/DummyException.cs

@@ -0,0 +1,9 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+
+namespace MyClient.Tests
+{
+    public class DummyException : Exception { }
+}

csharp/Practice01-End/tests/MyClient.Tests/MyClient.Tests.csproj

@@ -51,7 +51,9 @@
   <ItemGroup>
     <Compile Include="MyConnectionTest.cs" />
     <Compile Include="MyConnectorTest.cs" />
+    <Compile Include="MyReceiverTest.cs" />
     <Compile Include="MyRequestSenderTest.cs" />
+    <Compile Include="DummyException.cs" />
     <Compile Include="Properties\AssemblyInfo.cs" />
     <Compile Include="Threading\DelayThreadUtils.cs" />
   </ItemGroup>

csharp/Practice01-End/tests/MyClient.Tests/MyReceiverTest.cs

@@ -0,0 +1,86 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using MyClient.Driver;
+using Moq;
+using Xunit;
+using MyDriver;
+
+namespace MyClient.Tests
+{
+    public class MyReceiverTest
+    {
+        internal Mock<IMyDriverClient> _clientMock;
+        internal Mock<IMyConnector> _connectorMock;
+        internal MyReceiver _receiver;
+
+        public MyReceiverTest()
+        {
+            this._clientMock = new Mock<IMyDriverClient>(MockBehavior.Strict);
+
+            this._connectorMock = new Mock<IMyConnector>(MockBehavior.Strict);
+            this._connectorMock.Setup(c => c.Client).Returns(this._clientMock.Object);
+
+            this._receiver = new MyReceiver(this._connectorMock.Object);
+        }
+
+        public class Process : MyReceiverTest
+        {
+            [Fact]
+            public void DataReceived_PutIntoCollection()
+            {
+                var dataList = Enumerable.Range(0, 10).Select(i => new MyData(0, i.ToString())).ToList();
+
+                var setup = this._clientMock.SetupSequence(c => c.Receive());
+                foreach (var data in dataList)
+                {
+                    setup = setup.Returns(data);
+                }
+
+                setup.Throws(new DummyException());
+
+                Assert.Throws(typeof(DummyException), () => this._receiver.Process());
+
+                Assert.False(this._receiver.CancellationToken.IsCancellationRequested);
+
+                var producedList = this._receiver.DataProduced.ToList();
+                Assert.Equal(dataList.Count, producedList.Count);
+                for (var i = 0; i < dataList.Count; i++)
+                {
+                    Assert.Same(dataList[i], producedList[i]);
+                }
+            }
+
+            [Fact]
+            public void ErrorOccurred_TokenCanceled_CloseClient()
+            {
+                this._connectorMock.Setup(c => c.CloseClient());
+
+                this._clientMock.SetupSequence(c => c.Receive()).Returns(new MyData(0, "")).Throws(new MyDriverException("Test"));
+
+                this._receiver.Process();
+
+                Assert.True(this._receiver.CancellationToken.IsCancellationRequested);
+
+                Assert.Equal(1, this._receiver.DataProduced.Count);
+                Assert.False(this._receiver.DataProduced.IsAddingCompleted);
+
+                this._connectorMock.Verify(c => c.CloseClient(), Times.Once());
+            }
+
+            [Fact]
+            public void NullReceived_TokenCanceled()
+            {
+                this._clientMock.SetupSequence(c => c.Receive()).Returns(new MyData(0, "")).Returns(null);
+
+                this._receiver.Process();
+
+                Assert.True(this._receiver.CancellationToken.IsCancellationRequested);
+
+                Assert.Equal(1, this._receiver.DataProduced.Count);
+                Assert.False(this._receiver.DataProduced.IsAddingCompleted);
+            }
+        }
+    }
+}