Refactoring

Update connection test projects.
Fix ReadByteArray method in MemoryStreamReader

ConnectionTest/Client/Program.cs

@@ -2,21 +2,72 @@
 using System.Net;
 using System.Threading;
 using ZeroLevel;
+using ZeroLevel.Network;
+using ZeroLevel.Services.HashFunctions;
+using ZeroLevel.Services.Serialization;
 
 namespace Client
 {
+    public class Info
+        : IBinarySerializable
+    {
+        public uint Id;
+        public uint Length;
+        public uint Checksum;
+
+        public void Deserialize(IBinaryReader reader)
+        {
+            this.Id = reader.ReadUInt32();
+            this.Length = reader.ReadUInt32();
+            this.Checksum = reader.ReadUInt32();
+        }
+
+        public void Serialize(IBinaryWriter writer)
+        {
+            writer.WriteUInt32(this.Id);
+            writer.WriteUInt32(this.Length);
+            writer.WriteUInt32(this.Checksum);
+        }
+    }
+
+    public class Fragment
+        : IBinarySerializable
+    {
+        public uint Id;
+        public uint Offset;
+        public uint Checksum;
+        public byte[] Payload;
+
+        public void Deserialize(IBinaryReader reader)
+        {
+            this.Id = reader.ReadUInt32();
+            this.Offset = reader.ReadUInt32();
+            this.Checksum = reader.ReadUInt32();
+            this.Payload = reader.ReadBytes();
+        }
+
+        public void Serialize(IBinaryWriter writer)
+        {
+            writer.WriteUInt32(this.Id);
+            writer.WriteUInt32(this.Offset);
+            writer.WriteUInt32(this.Checksum);
+            writer.WriteBytes(this.Payload);
+        }
+    }
+
     class Program
     {
+        private readonly static XXHashUnsafe _hash = new XXHashUnsafe(667);
+
         static void Main(string[] args)
         {
             Log.AddConsoleLogger();
             var ex = Bootstrap.CreateExchange();
-
             var address = ReadIP();
             var port = ReadPort();
             var client = ex.GetConnection(new IPEndPoint(address, port));
-            Console.WriteLine("Esc - exit\r\nEnter - recreate connection\r\nSpace - send request");
+
-            long index = 0;
+            uint index = 0;
             while (true)
             {
                 if (Console.KeyAvailable)
@@ -26,60 +77,124 @@ namespace Client
                         case ConsoleKey.Escape:
                             client?.Dispose();
                             return;
-                        case ConsoleKey.Enter:
+                    }
-                            address = ReadIP();
+                }
-                            port = ReadPort();
+                if (index % 2 == 0)
-                            try
                 {
-                                if (client != null)
+                    SendDataEqParts(client, index, 1024 * 1024 + index * 3 + 1);
+                }
+                else
                 {
-                                    client.Dispose();
+                    SendDataDiffParts(client, index, 1024 * 1024 + index * 3 + 1);
                 }
-
+                index++;
-                                client = ex.GetConnection(new IPEndPoint(address, port));
             }
-                            catch (Exception exc)
-                            {
-                                Log.Error(exc, "Fault recreate connection");
         }
-                            break;
+
-                        case ConsoleKey.Spacebar:
+        static void SendDataDiffParts(IClient client, uint id, uint length)
-                            Log.Info("Send request");
+        {
-                            if (client == null)
+            var payload = GetByteArray(length);
+            var full_checksum = _hash.Hash(payload);
+            var info = new Info { Checksum = full_checksum, Id = id, Length = length };
+            if (client.Request<Info, bool>("start", info, res =>
+            {
+                Log.Info($"Success start sending packet '{id}'");
+            }))
+            {
+                uint size = 1;
+                uint offset = 0;
+                while (offset < payload.Length)
                 {
-                                client = ex.GetConnection(new IPEndPoint(address, port));
+                    var fragment = GetFragment(id, payload, offset, size);
-                                if (client == null)
+                    if (!client.Request<Fragment, bool>("part", fragment, res =>
+                    {
+                        if (!res)
+                        {
+                            Log.Info($"Fault server incoming packet '{id}' fragment. Offset: '{offset}'. Size: '{size}' bytes.");
+                        }
+                    }))
                     {
-                                    Log.Info("No connection");
+                        Log.Warning($"Can't start send packet '{id}' fragment. Offset: '{offset}'. Size: '{size}' bytes. No connection");
                     }
-                                continue;
+                    offset += size;
+                    size += 1;
                 }
-                            if (false == client.Request<string, string>("time", "Time reqeust", s => { Log.Info($"Got time response '{s}'"); }))
+                client.Send<uint>("complete", id);
+            }
+            else
             {
-                                Log.Warning("Send time request fault");
+                Log.Warning($"Can't start send packet '{id}'. No connection");
             }
-                            break;
         }
+
+        static void SendDataEqParts(IClient client, uint id, uint length)
+        {
+            var payload = GetByteArray(length);
+            var full_checksum = _hash.Hash(payload);
+            var info = new Info { Checksum = full_checksum, Id = id, Length = length };
+            if (client.Request<Info, bool>("start", info, res =>
+            {
+                if (res)
+                {
+                    Log.Info($"Success start sending packet '{id}'");
                 }
-                Thread.Sleep(100);
+                else
-                index++;
-                if (index % 50 == 0)
                 {
-                    if (client == null)
+                    Log.Info($"Fault server start incoming packet '{id}'");
+                }
+            }))
             {
-                        client = ex.GetConnection(new IPEndPoint(address, port));
+                uint size = 4096;
-                        if (client == null)
+                uint offset = 0;
+
+                while (offset < payload.Length)
+                {
+                    var fragment = GetFragment(id, payload, offset, size);
+                    if (!client.Request<Fragment, bool>("part", fragment, res =>
                     {
-                            Log.Info("No connection");
+                        if (!res)
+                        {
+                            Log.Info($"Fault server incoming packet '{id}' fragment. Offset: '{offset}'. Size: '{size}' bytes.");
                         }
-                        continue;
+                    }))
+                    {
+                        Log.Warning($"Can't start send packet '{id}' fragment. Offset: '{offset}'. Size: '{size}' bytes. No connection");
                     }
-                    if (false == client.Request<string, string>("whois", "Whois reqeust", s => { Log.Info($"Got whois response '{s}'"); }))
+                    offset += size;
+                }
+                client.Send<uint>("complete", id);
+            }
+            else
             {
-                        Log.Warning("Send whois request fault");
+                Log.Warning($"Can't start send packet '{id}'. No connection");
             }
         }
+
+        private static Fragment GetFragment(uint id, byte[] data, uint offset, uint size)
+        {
+            int diff = (int)(-(data.Length - (offset + size)));
+            if (diff > 0)
+            {
+                size -= (uint)diff;
+            }
+            var payload = new byte[size];
+            Array.Copy(data, offset, payload, 0, size);
+            var ch = _hash.Hash(payload);
+            return new Fragment
+            {
+                Id = id,
+                Checksum = ch,
+                Offset = offset,
+                Payload = payload
+            };
         }
+
+        private static byte[] GetByteArray(uint size)
+        {
+            Random rnd = new Random();
+            Byte[] b = new Byte[size];
+            rnd.NextBytes(b);
+            return b;
         }
 
         static IPAddress ReadIP()

ConnectionTest/Server/Program.cs

@@ -1,26 +1,152 @@
 using System;
+using System.Collections.Generic;
 using ZeroLevel;
+using ZeroLevel.Services.HashFunctions;
+using ZeroLevel.Services.Serialization;
 
 namespace Server
 {
+    public class Info
+        : IBinarySerializable
+    {
+        public uint Id;
+        public uint Length;
+        public uint Checksum;
+
+        public void Deserialize(IBinaryReader reader)
+        {
+            this.Id = reader.ReadUInt32();
+            this.Length = reader.ReadUInt32();
+            this.Checksum = reader.ReadUInt32();
+        }
+
+        public void Serialize(IBinaryWriter writer)
+        {
+            writer.WriteUInt32(this.Id);
+            writer.WriteUInt32(this.Length);
+            writer.WriteUInt32(this.Checksum);
+        }
+    }
+
+    public class Fragment
+        : IBinarySerializable
+    {
+        public uint Id;
+        public uint Offset;
+        public uint Checksum;
+        public byte[] Payload;
+
+        public void Deserialize(IBinaryReader reader)
+        {
+            this.Id = reader.ReadUInt32();
+            this.Offset = reader.ReadUInt32();
+            this.Checksum = reader.ReadUInt32();
+            this.Payload = reader.ReadBytes();
+        }
+
+        public void Serialize(IBinaryWriter writer)
+        {
+            writer.WriteUInt32(this.Id);
+            writer.WriteUInt32(this.Offset);
+            writer.WriteUInt32(this.Checksum);
+            writer.WriteBytes(this.Payload);
+        }
+    }
+
+    public class Data
+    {
+        public uint Checksum;
+        public uint Length;
+        public uint ActualLength;
+        public byte[] Payload;
+    }
+
     class Program
     {
+        private readonly static Dictionary<uint, Data> _incoming = new Dictionary<uint, Data>();
+        private readonly static XXHashUnsafe _hash = new XXHashUnsafe(667);
+
         static void Main(string[] args)
         {
-            Log.AddConsoleLogger();
+            Log.AddConsoleLogger(ZeroLevel.Logging.LogLevel.Warning);
             var ex = Bootstrap.CreateExchange();
-
             var port = ReadPort();
-
             var server = ex.UseHost(port);
+
             server.RegisterInbox<string, string>("time", (c, s) => { Log.Info($"Request time: [{s}]"); return DateTime.UtcNow.ToString("yyyy-MM-dd HH:mm:ss"); });
             server.RegisterInbox<string, string>("whois", (c, s) => { Log.Info($"Request whois: [{s}]"); return $"[{Environment.MachineName}] {Environment.UserDomainName}\\{Environment.UserName}"; });
 
+            server.RegisterInbox<Info, bool>("start", (c, i) =>
+            {
+                Start(i);
+                return true;
+            });
+
+            server.RegisterInbox<Fragment, bool>("part", (c, p) =>
+            {
+                return WriteFragment(p);
+            });
+
+            server.RegisterInbox<uint>("complete", (c, id) => Complete(id));
+
+            Log.Warning("Started");
+
             server.OnConnect += Server_OnConnect;
             server.OnDisconnect += Server_OnDisconnect;
             Console.ReadKey();
         }
 
+        private static void Start(Info info)
+        {
+            var data = new Data
+            {
+                Checksum = info.Checksum,
+                ActualLength = 0,
+                Length = info.Length,
+                Payload = new byte[info.Length]
+            };
+            _incoming.Add(info.Id, data);
+            Log.Info($"Start incoming data id '{info.Id}'. {info.Length} bytes. Checksum: {info.Checksum}");
+        }
+
+        private static bool WriteFragment(Fragment fragment)
+        {
+            var checksum = _hash.Hash(fragment.Payload);
+            if (checksum != fragment.Checksum)
+            {
+                Log.Warning($"[WriteFragment] Wrong checksum (checksum: {checksum} expected: {fragment.Checksum})! ID: '{fragment.Id}'. Offset: '{fragment.Offset}'. Length: '{fragment.Payload.Length}' bytes.");
+                return false;
+            }
+            if (!_incoming.ContainsKey(fragment.Id))
+            {
+                Log.Warning($"[WriteFragment] Data ID: '{fragment.Id}' not found. Offset: '{fragment.Offset}'. Length: '{fragment.Payload.Length}' bytes.");
+                return false;
+            }
+            Array.Copy(fragment.Payload, 0, _incoming[fragment.Id].Payload, fragment.Offset, fragment.Payload.Length);
+            return true;
+        }
+
+        private static void Complete(uint id)
+        {
+            if (_incoming.ContainsKey(id))
+            {
+                var checksum = _hash.Hash(_incoming[id].Payload);
+                if (checksum != _incoming[id].Checksum)
+                {
+                    Log.Warning($"[Complete] Wrong checksum (checksum: {checksum} expected: {_incoming[id].Checksum})! ID: '{id}'");
+                }
+                else
+                {
+                    Log.Info($"Data '{id}' successfully received");
+                }
+                _incoming.Remove(id);
+            }
+            else
+            {
+                Log.Warning($"[Complete] Data ID '{id}' not found");
+            }
+        }
+
         private static void Server_OnDisconnect(ZeroLevel.Network.ISocketClient obj)
         {
             Log.Info($"Client disconnected: {obj.Endpoint.Address}:{obj.Endpoint.Port}");

ZeroLevel/Services/Serialization/MemoryStreamReader.cs

@@ -749,16 +749,7 @@ namespace ZeroLevel.Services.Serialization
 
         public byte[] ReadByteArray()
         {
-            int count = ReadInt32();
+            return ReadBytes();
-            var array = new byte[count];
-            if (count > 0)
-            {
-                for (int i = 0; i < count; i++)
-                {
-                    array[i] = ReadByte();
-                }
-            }
-            return array;
         }
 
         public byte[][] ReadByteArrayArray()