Changes on transport layer

- Removed dependency over ReactiveSockets
- Added TcpChannel as an adapter of TcpClient to replace ReactiveSocket
- Applied Disposable pattern over all the Channel implementations

src/Client/Client.cs

@@ -23,14 +23,14 @@ public class Client : IClient
 		readonly IRepository<PacketIdentifier> packetIdentifierRepository;
 		readonly ProtocolConfiguration configuration;
 
-        public Client(IBufferedChannel<byte> socket, 
+        public Client(IChannel<byte[]> binaryChannel, 
 			IPacketChannelFactory channelFactory, 
 			IPacketChannelAdapter channelAdapter,
 			IProtocolFlowProvider flowProvider,
 			IRepositoryProvider repositoryProvider,
 			ProtocolConfiguration configuration)
         {
-			var channel = channelFactory.CreateChannel (socket);
+			var channel = channelFactory.Create (binaryChannel);
 
 			this.protocolChannel = channelAdapter.Adapt (channel);
 			this.flowProvider = flowProvider;
@@ -196,7 +196,7 @@ private void CloseChannel ()
 		{
 			this.IsConnected = false; 
 			this.Id = null;
-			this.protocolChannel.Close ();
+			this.protocolChannel.Dispose ();
 		}
 	}
 }

src/Client/Client.csproj

@@ -19,8 +19,8 @@
     <Compile Include="..\Server\ProtocolChannel.cs">
       <Link>ProtocolChannel.cs</Link>
     </Compile>
-    <Compile Include="..\Server\ReactiveSocketChannel.cs">
-      <Link>ReactiveSocketChannel.cs</Link>
+    <Compile Include="..\Server\TcpChannel.cs">
+      <Link>TcpChannel.cs</Link>
     </Compile>
     <Compile Include="ApplicationMessage.cs" />
     <Compile Include="Client.cs" />

src/Client/Properties/Resources.resx

@@ -132,4 +132,7 @@
   <data name="Client_ConnectionTimeout" xml:space="preserve">
     <value>A timeout occured while waiting for the client connection confirmation</value>
   </data>
+  <data name="TcpChannel_ClientMustBeConnected" xml:space="preserve">
+    <value>Client must be connected</value>
+  </data>
 </root>

src/Portable/IChannel.cs

@@ -3,7 +3,7 @@
 
 namespace Hermes
 {
-	public interface IChannel<T>
+	public interface IChannel<T> : IDisposable
     {
 		bool IsConnected { get; }
 
@@ -12,7 +12,5 @@ public interface IChannel<T>
 		IObservable<T> Sender { get; }
 
         Task SendAsync(T message);
-
-		void Close();
     }
 }

src/Portable/IPacketBuffer.cs

@@ -0,0 +1,7 @@
+namespace Hermes
+{
+	public interface IPacketBuffer
+	{
+		bool TryGetPacket (byte[] sequence, out byte[] packet);
+	}
+}

src/Portable/IPacketChannelFactory.cs

@@ -4,6 +4,6 @@ namespace Hermes
 {
 	public interface IPacketChannelFactory
 	{
-		IChannel<IPacket> CreateChannel (IBufferedChannel<byte> socket);
+		IChannel<IPacket> Create (IChannel<byte[]> binaryChannel);
 	}
 }

src/Portable/PacketBuffer.cs

@@ -0,0 +1,99 @@
+using System.Collections.Generic;
+using System.Linq;
+
+namespace Hermes
+{
+	public class PacketBuffer : IPacketBuffer
+	{
+		bool readStarted;
+		bool remainingLengthRead;
+		int remainingLength = 0;
+		bool packetReady;
+
+		readonly IList<byte> mainBuffer;
+		readonly IList<byte> pendingBuffer;
+
+		public PacketBuffer ()
+		{
+			this.mainBuffer = new List<byte> ();
+			this.pendingBuffer = new List<byte> ();
+		}
+
+		public bool TryGetPacket (byte[] sequence, out byte[] packet)
+		{
+			this.Buffer (sequence);
+
+			if (this.packetReady) {
+				packet = this.mainBuffer.ToArray ();
+				this.Reset ();
+			} else {
+				packet = default (byte[]);
+			}
+
+			return packet != default(byte[]);
+		}
+
+		private void Buffer(byte[] sequence)
+		{
+			foreach(var @byte in sequence) {
+				this.Buffer (@byte);
+			}
+		}
+
+		private void Buffer (byte @byte)
+		{
+			if (this.packetReady) {
+				this.pendingBuffer.Add (@byte);
+				return;
+			}
+
+			this.mainBuffer.Add(@byte);
+
+			if (!this.readStarted)
+			{
+				this.readStarted = true;
+				return;
+			}
+
+			if (!this.remainingLengthRead)
+			{
+				if ((@byte & 128) == 0) {
+					var bytesLenght = default (int);
+
+					this.remainingLength = Protocol.Encoding.DecodeRemainingLength(mainBuffer.ToArray(), out bytesLenght);
+					this.remainingLengthRead = true;
+
+					if (remainingLength == 0)
+						this.packetReady = true;
+				}
+
+				return;
+			}
+
+			if (remainingLength == 1)
+			{
+				this.packetReady = true;
+			}
+			else
+			{
+				remainingLength--;
+			}
+		}
+
+		private void Reset()
+		{
+			this.mainBuffer.Clear();
+			this.readStarted = false;
+			this.remainingLengthRead = false;
+			this.remainingLength = 0;
+			this.packetReady = false;
+
+			if (this.pendingBuffer.Any ()) {
+				var pendingSequence = this.pendingBuffer.ToArray ();
+
+				this.pendingBuffer.Clear ();
+				this.Buffer (pendingSequence);
+			}
+		}
+	}
+}

src/Portable/PacketChannel.cs

@@ -7,6 +7,8 @@ namespace Hermes
 {
 	public class PacketChannel : IChannel<IPacket>
 	{
+		bool disposed;
+
 		readonly IChannel<byte[]> innerChannel;
 		readonly IPacketManager manager;
         readonly Subject<IPacket> receiver;
@@ -39,6 +41,9 @@ public PacketChannel (IChannel<byte[]> innerChannel, IPacketManager manager)
 
 		public async Task SendAsync (IPacket packet)
 		{
+			if (this.disposed)
+				throw new ObjectDisposedException (this.GetType ().FullName);
+
 			try {
 				var bytes = await this.manager.GetBytesAsync (packet);
 
@@ -50,12 +55,23 @@ public async Task SendAsync (IPacket packet)
 			}
 		}
 
-		public void Close ()
+		public void Dispose ()
 		{
-			this.subscription.Dispose ();
-			this.innerChannel.Close ();
-			this.receiver.OnCompleted ();
-			this.sender.OnCompleted ();
+			this.Dispose (true);
+			GC.SuppressFinalize (this);
+		}
+
+		protected virtual void Dispose (bool disposing)
+		{
+			if (this.disposed) return;
+
+			if (disposing) {
+				this.subscription.Dispose ();
+				this.innerChannel.Dispose ();
+				this.receiver.OnCompleted ();
+				this.sender.OnCompleted ();
+				this.disposed = true;
+			}
 		}
 	}
 }

src/Portable/PacketChannelFactory.cs

@@ -13,10 +13,8 @@ public PacketChannelFactory (ITopicEvaluator topicEvaluator)
 			this.topicEvaluator = topicEvaluator;
 		}
 
-		public IChannel<IPacket> CreateChannel (IBufferedChannel<byte> socket)
+		public IChannel<IPacket> Create (IChannel<byte[]> binaryChannel)
 		{
-			var binaryChannel = new BinaryChannel (socket);
-
 			var formatters = this.GetFormatters();
 			var manager = new PacketManager (formatters);