From c9e3d408f3d0d5bb0f12010ae0e2553a0abd8918 Mon Sep 17 00:00:00 2001 From: "a.bozhenov" Date: Tue, 25 Jun 2019 19:11:20 +0300 Subject: [PATCH] dirty_network_refactoring no works! --- ZeroLevel/Services/HashFunctions/Murmur3.cs | 164 ++++++++++++ ZeroLevel/Services/Network/Models/Frame.cs | 24 -- ZeroLevel/Services/Network/ZBaseNetwork.cs | 5 - .../Services/_Network/Contracts/IRouter.cs | 35 +++ .../_Network/Contracts/ISocketClient.cs | 12 + ZeroLevel/Services/_Network/FrameParser.cs | 240 ++++++++++++++++++ ZeroLevel/Services/_Network/FrameType.cs | 10 + ZeroLevel/Services/_Network/IncomingRouter.cs | 183 +++++++++++++ .../Services/_Network/NetworkDelegates.cs | 7 + ZeroLevel/Services/_Network/NetworkNode.cs | 121 +++++++++ .../Services/_Network/NetworkPacketFactory.cs | 90 +++++++ 11 files changed, 862 insertions(+), 29 deletions(-) create mode 100644 ZeroLevel/Services/HashFunctions/Murmur3.cs create mode 100644 ZeroLevel/Services/_Network/Contracts/IRouter.cs create mode 100644 ZeroLevel/Services/_Network/Contracts/ISocketClient.cs create mode 100644 ZeroLevel/Services/_Network/FrameParser.cs create mode 100644 ZeroLevel/Services/_Network/FrameType.cs create mode 100644 ZeroLevel/Services/_Network/IncomingRouter.cs create mode 100644 ZeroLevel/Services/_Network/NetworkDelegates.cs create mode 100644 ZeroLevel/Services/_Network/NetworkNode.cs create mode 100644 ZeroLevel/Services/_Network/NetworkPacketFactory.cs diff --git a/ZeroLevel/Services/HashFunctions/Murmur3.cs b/ZeroLevel/Services/HashFunctions/Murmur3.cs new file mode 100644 index 0000000..45dc758 --- /dev/null +++ b/ZeroLevel/Services/HashFunctions/Murmur3.cs @@ -0,0 +1,164 @@ +using System; +using System.Runtime.CompilerServices; + +namespace ZeroLevel.Services.HashFunctions +{ + public static class Murmur3 + { + // 128 bit output, 64 bit platform version + + public static ulong READ_SIZE = 16; + private static ulong C1 = 0x87c37b91114253d5L; + private static ulong C2 = 0x4cf5ad432745937fL; + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + private static ulong MixKey1(ulong k1) + { + k1 *= C1; + k1 = k1.RotateLeft(31); + k1 *= C2; + return k1; + } + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + private static ulong MixKey2(ulong k2) + { + k2 *= C2; + k2 = k2.RotateLeft(33); + k2 *= C1; + return k2; + } + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + private static ulong MixFinal(ulong k) + { + // avalanche bits + + k ^= k >> 33; + k *= 0xff51afd7ed558ccdL; + k ^= k >> 33; + k *= 0xc4ceb9fe1a85ec53L; + k ^= k >> 33; + return k; + } + + public static byte[] ComputeHash(byte[] bb, ulong seed = 0) + { + var h1 = seed; + ulong h2 = 0; + var length = 0UL; + int pos = 0; + ulong remaining = (ulong)bb.Length; + // read 128 bits, 16 bytes, 2 longs in eacy cycle + while (remaining >= READ_SIZE) + { + ulong k1 = bb.GetUInt64(pos); + pos += 8; + ulong k2 = bb.GetUInt64(pos); + pos += 8; + length += READ_SIZE; + remaining -= READ_SIZE; + + h1 ^= MixKey1(k1); + h1 = h1.RotateLeft(27); + h1 += h2; + h1 = h1 * 5 + 0x52dce729; + h2 ^= MixKey2(k2); + h2 = h2.RotateLeft(31); + h2 += h1; + h2 = h2 * 5 + 0x38495ab5; + } + + // if the input MOD 16 != 0 + if (remaining > 0) + { + ulong k1 = 0; + ulong k2 = 0; + length += remaining; + // little endian (x86) processing + switch (remaining) + { + case 15: + k2 ^= (ulong)bb[pos + 14] << 48; // fall through + goto case 14; + case 14: + k2 ^= (ulong)bb[pos + 13] << 40; // fall through + goto case 13; + case 13: + k2 ^= (ulong)bb[pos + 12] << 32; // fall through + goto case 12; + case 12: + k2 ^= (ulong)bb[pos + 11] << 24; // fall through + goto case 11; + case 11: + k2 ^= (ulong)bb[pos + 10] << 16; // fall through + goto case 10; + case 10: + k2 ^= (ulong)bb[pos + 9] << 8; // fall through + goto case 9; + case 9: + k2 ^= (ulong)bb[pos + 8]; // fall through + goto case 8; + case 8: + k1 ^= bb.GetUInt64(pos); + break; + case 7: + k1 ^= (ulong)bb[pos + 6] << 48; // fall through + goto case 6; + case 6: + k1 ^= (ulong)bb[pos + 5] << 40; // fall through + goto case 5; + case 5: + k1 ^= (ulong)bb[pos + 4] << 32; // fall through + goto case 4; + case 4: + k1 ^= (ulong)bb[pos + 3] << 24; // fall through + goto case 3; + case 3: + k1 ^= (ulong)bb[pos + 2] << 16; // fall through + goto case 2; + case 2: + k1 ^= (ulong)bb[pos + 1] << 8; // fall through + goto case 1; + case 1: + k1 ^= (ulong)bb[pos]; // fall through + break; + default: + throw new Exception("Something went wrong with remaining bytes calculation."); + } + h1 ^= MixKey1(k1); + h2 ^= MixKey2(k2); + } + + + h1 ^= length; + h2 ^= length; + + h1 += h2; + h2 += h1; + + h1 = Murmur3.MixFinal(h1); + h2 = Murmur3.MixFinal(h2); + + h1 += h2; + h2 += h1; + + var hash = new byte[Murmur3.READ_SIZE]; + Array.Copy(BitConverter.GetBytes(h1), 0, hash, 0, 8); + Array.Copy(BitConverter.GetBytes(h2), 0, hash, 8, 8); + return hash; + } + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public static ulong RotateLeft(this ulong original, int bits) => + (original << bits) | (original >> (64 - bits)); + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public static ulong RotateRight(this ulong original, int bits) => + (original >> bits) | (original << (64 - bits)); + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public static ulong GetUInt64(this byte[] bb, int pos) => + (ulong)(bb[pos++] | bb[pos++] << 8 | bb[pos++] << 16 | bb[pos++] << 24); + } +} diff --git a/ZeroLevel/Services/Network/Models/Frame.cs b/ZeroLevel/Services/Network/Models/Frame.cs index e516bdb..7510301 100644 --- a/ZeroLevel/Services/Network/Models/Frame.cs +++ b/ZeroLevel/Services/Network/Models/Frame.cs @@ -2,7 +2,6 @@ using System.Collections.Generic; using System.Runtime.Serialization; using System.Text; -using System.Threading; using ZeroLevel.Services.Pools; using ZeroLevel.Services.Serialization; @@ -22,7 +21,6 @@ namespace ZeroLevel.Network var frame = _pool.Allocate(); frame.Inbox = null; frame.Payload = null; - frame.IsRequest = false; return frame; } @@ -31,25 +29,14 @@ namespace ZeroLevel.Network _pool.Free(this); } - private static long _id_counter = 0; - - internal static long GetMessageId() => Interlocked.Increment(ref _id_counter); - - [DataMember] - public long FrameId { get; set; } - [DataMember] public string Inbox { get; set; } [DataMember] public byte[] Payload { get; set; } - [DataMember] - public bool IsRequest { get; set; } - public Frame() { - FrameId = GetMessageId(); } public Frame(Frame other) @@ -63,18 +50,14 @@ namespace ZeroLevel.Network public void Deserialize(IBinaryReader reader) { - this.FrameId = reader.ReadLong(); this.Inbox = reader.ReadString(); this.Payload = reader.ReadBytes(); - this.IsRequest = reader.ReadBoolean(); } public void Serialize(IBinaryWriter writer) { - writer.WriteLong(this.FrameId); writer.WriteString(this.Inbox); writer.WriteBytes(this.Payload); - writer.WriteBoolean(this.IsRequest); } public T Read() @@ -109,11 +92,6 @@ namespace ZeroLevel.Network this.Payload = Encoding.UTF32.GetBytes(data); } - public override int GetHashCode() - { - return this.FrameId.GetHashCode(); - } - public override bool Equals(object obj) { return this.Equals(obj as Frame); @@ -126,8 +104,6 @@ namespace ZeroLevel.Network return true; if (this.GetType() != other.GetType()) return false; - if (this.IsRequest != other.IsRequest) return false; - if (this.FrameId != other.FrameId) return false; if (string.Compare(this.Inbox, other.Inbox, true) != 0) return false; if (ArrayExtensions.UnsafeEquals(this.Payload, other.Payload) == false) return false; return true; diff --git a/ZeroLevel/Services/Network/ZBaseNetwork.cs b/ZeroLevel/Services/Network/ZBaseNetwork.cs index 65bf4bc..acd7055 100644 --- a/ZeroLevel/Services/Network/ZBaseNetwork.cs +++ b/ZeroLevel/Services/Network/ZBaseNetwork.cs @@ -44,11 +44,6 @@ namespace ZeroLevel.Network private const int DEFAULT_MAX_FRAME_PAYLOAD_SIZE = 1024 * 1024 * 32; public readonly static int MAX_FRAME_PAYLOAD_SIZE; - /// - /// Starting byte of the data packet header - /// - public const byte PACKET_HEADER_START_BYTE = 181; - /// /// The size of the message queue to send /// diff --git a/ZeroLevel/Services/_Network/Contracts/IRouter.cs b/ZeroLevel/Services/_Network/Contracts/IRouter.cs new file mode 100644 index 0000000..d0d22f5 --- /dev/null +++ b/ZeroLevel/Services/_Network/Contracts/IRouter.cs @@ -0,0 +1,35 @@ +namespace ZeroLevel.Services._Network +{ + public interface IRouter + { + #region Messages + void RegisterInbox(string inbox, MessageHandler handler); + void RegisterInbox(string inbox, MessageHandler handler); + + // Default inboxe + void RegisterInbox(MessageHandler handler); + void RegisterInbox(MessageHandler handler); + #endregion + + #region Requests + void RegisterInbox(string inbox, RequestHandler handler); + void RegisterInbox(string inbox, RequestHandler handler); + + // Default inboxe + void RegisterInbox(RequestHandler handler); + void RegisterInbox(RequestHandler handler); + #endregion + } + + public interface IClient + { + void Send(string inbox); + void Send(string inbox, byte[] data); + void Send(string inbox, T message); + + byte[] Request(string inbox); + byte[] Request(string inbox, byte[] data); + Tresponse Request(string inbox); + Tresponse Request(string inbox, Trequest request); + } +} diff --git a/ZeroLevel/Services/_Network/Contracts/ISocketClient.cs b/ZeroLevel/Services/_Network/Contracts/ISocketClient.cs new file mode 100644 index 0000000..9a12485 --- /dev/null +++ b/ZeroLevel/Services/_Network/Contracts/ISocketClient.cs @@ -0,0 +1,12 @@ +using System; + +namespace ZeroLevel.Services._Network +{ + public interface ISocketClient + { + event Action OnIncomingData; + void UseKeepAlive(TimeSpan period); + void Send(byte[] data); + byte[] Request(byte[] data); + } +} diff --git a/ZeroLevel/Services/_Network/FrameParser.cs b/ZeroLevel/Services/_Network/FrameParser.cs new file mode 100644 index 0000000..8aa34c4 --- /dev/null +++ b/ZeroLevel/Services/_Network/FrameParser.cs @@ -0,0 +1,240 @@ +using System; +using System.Threading.Tasks; +using ZeroLevel.Network; +using ZeroLevel.Services._Network; + +namespace ZeroLevel._Network +{ + public sealed class FrameParser + { + #region private models + private enum ParserState + { + WaitNew, + WaitSize, + WaitIdentity, + Proceeding + } + + private class _Accum + { + public int Identity; + public int Size; + public byte[] Payload; + public FrameType Type; + + public bool SizeFilled; + public bool IdentityFilled; + public bool PayloadFilled; + public bool Corrupted; + + + public void Reset(byte magic) + { + Identity = 0; + Size = 0; + offset = 0; + Payload = null; + SizeFilled = false; + IdentityFilled = false; + PayloadFilled = false; + Corrupted = false; + + switch (magic) + { + case NetworkPacketFactory.MAGIC: Type = FrameType.Message; break; + case NetworkPacketFactory.MAGIC_REQUEST: Type = FrameType.Request; break; + case NetworkPacketFactory.MAGIC_RESPONSE: Type = FrameType.Response; break; + case NetworkPacketFactory.MAGIC_KEEP_ALIVE: Type = FrameType.KeepAlive; break; + } + } + + private byte[] _size_buf = new byte[4]; + private int offset; + + public int WriteSize(byte[] buf, int start, int length) + { + for (; offset < 4 && start < length; offset++, start++) + { + _size_buf[offset] = buf[start]; + } + if (offset == 4) + { + Size = BitConverter.ToInt32(_size_buf, 0); + SizeFilled = true; + offset = 0; + if (Size == 0) + { + // At least 1 byte with checksum must be + Corrupted = true; + } + } + return start; + } + + public int WriteIdentity(byte[] buf, int start, int length) + { + for (; offset < 4 && start < length; offset++, start++) + { + _size_buf[offset] = buf[start]; + } + if (offset == 4) + { + Identity = BitConverter.ToInt32(_size_buf, 0); + IdentityFilled = true; + offset = 0; + } + return start; + } + + public int WritePayload(byte[] buf, int start, int length) + { + if (Payload == null) + { + Payload = new byte[Size]; + var mask = ((byte)(NetworkPacketFactory.MAGIC ^ _size_buf[0] ^ _size_buf[1] ^ _size_buf[2] ^ _size_buf[3])); + if (buf[start] != mask) + { + Corrupted = true; + return start; + } + start = start + 1; + } + int i = start; + for (; offset < Size && i < length; offset++, i++) + { + Payload[offset] = buf[i]; + } + if (offset == Size) + { + var mask = ((byte)(NetworkPacketFactory.MAGIC ^ _size_buf[0] ^ _size_buf[1] ^ _size_buf[2] ^ _size_buf[3])); + DeHashData(Payload, mask); + PayloadFilled = true; + } + return i; + } + + private static byte DeHashData(byte[] data, byte initialmask) + { + if (data.Length == 0) return 0; + byte checksum = initialmask; + for (var i = data.Length - 1; i > 0; i--) + { + data[i] ^= data[i - 1]; + checksum ^= data[i]; + } + data[0] ^= initialmask; + checksum ^= data[0]; + return checksum; + } + } + + #endregion private models + public event Action OnIncoming; + + private readonly _Accum _accum = new _Accum(); + private ParserState _state = ParserState.WaitNew; + private readonly object _push_lock = new object(); + + /// + /// Parse with state machine + /// + public void Push(byte[] part, int length) + { + lock (_push_lock) + { + __Push(part, 0, length); + } + } + + private void FireOnFrame(FrameType type, int identity, byte[] payload) + { + try + { + Task.Run(() => OnIncoming?.Invoke(type, identity, payload)); + } + catch (Exception ex) + { + Log.SystemError(ex, "[FrameParser.FireOnFrame] Fault handle incoming data"); + } + } + + private void __Push(byte[] part, int position, int length) + { + if (part == null || length == 0 || position >= length) return; + while (position < length) + { + switch (_state) + { + case ParserState.WaitNew: + { + for (; position < length; position++) + { + // Search for the beginning of the package header + if ((part[position] & NetworkPacketFactory.MAGIC) == NetworkPacketFactory.MAGIC) + { + _accum.Reset(part[position]); + _state = ParserState.WaitSize; + position++; + break; + } + } + } + break; + + case ParserState.WaitSize: + { + position = _accum.WriteSize(part, position, length); + if (_accum.SizeFilled) + { + if (_accum.Corrupted || _accum.Size < 1 || _accum.Size > ZBaseNetwork.MAX_FRAME_PAYLOAD_SIZE) + { + _state = ParserState.WaitNew; + } + else + { + switch (_accum.Type) + { + case FrameType.KeepAlive: + case FrameType.Message: + _state = ParserState.Proceeding; + break; + case FrameType.Request: + case FrameType.Response: + _state = ParserState.WaitIdentity; + break; + } + } + } + } + break; + + case ParserState.WaitIdentity: + { + position = _accum.WriteIdentity(part, position, length); + if (_accum.IdentityFilled) + { + _state = ParserState.Proceeding; + } + } + break; + + case ParserState.Proceeding: + { + position = _accum.WritePayload(part, position, length); + if (_accum.Corrupted) + { + _state = ParserState.WaitNew; + } + else if (_accum.PayloadFilled) + { + FireOnFrame(_accum.Type, _accum.Identity, _accum.Payload); + _state = ParserState.WaitNew; + } + } + break; + } + } + } + } +} \ No newline at end of file diff --git a/ZeroLevel/Services/_Network/FrameType.cs b/ZeroLevel/Services/_Network/FrameType.cs new file mode 100644 index 0000000..6a92336 --- /dev/null +++ b/ZeroLevel/Services/_Network/FrameType.cs @@ -0,0 +1,10 @@ +namespace ZeroLevel.Services._Network +{ + public enum FrameType + { + Message = 0, + Request = 1, + Response = 2, + KeepAlive = 3 + } +} diff --git a/ZeroLevel/Services/_Network/IncomingRouter.cs b/ZeroLevel/Services/_Network/IncomingRouter.cs new file mode 100644 index 0000000..c06bd98 --- /dev/null +++ b/ZeroLevel/Services/_Network/IncomingRouter.cs @@ -0,0 +1,183 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Linq.Expressions; +using System.Reflection; +using ZeroLevel.Services.Invokation; + +namespace ZeroLevel.Services._Network +{ + public class Router + : IRouter + { + #region Routing + + private sealed class MRInvoker + { + /// + /// Creates a compiled expression for a quick method call, returns the identifier of the expression and a delegate for the call. + /// + private static Invoker CreateCompiledExpression(MethodInfo method) + { + var targetArg = Expression.Parameter(typeof(object)); // Target + var argsArg = Expression.Parameter(typeof(object[])); // Method's args + var parameters = method.GetParameters(); + Expression body = Expression.Call( + method.IsStatic + ? null + : Expression.Convert(targetArg, method.DeclaringType), // Method's type + method, + parameters.Select((p, i) => + Expression.Convert(Expression.ArrayIndex(argsArg, Expression.Constant(i)), p.ParameterType))); + if (body.Type == typeof(void)) + body = Expression.Block(body, Expression.Constant(null)); + else if (body.Type.IsValueType) + body = Expression.Convert(body, typeof(object)); + return Expression.Lambda(body, targetArg, argsArg).Compile(); + } + + private static Invoker CreateCompiledExpression(Delegate handler) + { + return CreateCompiledExpression(handler.GetMethodInfo()); + } + + private object _instance; + private Invoker _invoker; + private Type _typeReq; + private Type _typeResp; + private bool _noArguments = false; + + public static MRInvoker Create(Action handler) + { + return new MRInvoker + { + _noArguments = true, + _typeReq = null, + _typeResp = null, + _instance = handler.Target, + _invoker = CreateCompiledExpression(handler) + }; + } + + public static MRInvoker Create(Action handler) + { + return new MRInvoker + { + _typeReq = typeof(T), + _typeResp = null, + _instance = handler.Target, + _invoker = CreateCompiledExpression(handler) + }; + } + + public static MRInvoker Create(Func handler) + { + return new MRInvoker + { + _typeReq = typeof(Treq), + _typeResp = typeof(Tresp), + _instance = handler.Target, + _invoker = CreateCompiledExpression(handler) + }; + } + + public static MRInvoker Create(Func handler) + { + return new MRInvoker + { + _typeReq = null, + _typeResp = typeof(Tresp), + _instance = handler.Target, + _invoker = CreateCompiledExpression(handler) + }; + } + + public object Invoke(Frame frame, IZBackward client) + { + if (_typeResp == null) + { + var incoming = MessageSerializer.DeserializeCompatible(_typeReq, frame.Payload); + if (_noArguments) + { + this._invoker.Invoke(this._instance, new object[] { frame.FrameId, client }); + } + else + { + this._invoker.Invoke(this._instance, new object[] { incoming, frame.FrameId, client }); + } + } + else if (_typeReq == null) + { + return this._invoker.Invoke(this._instance, new object[] { frame.FrameId, client }); + } + else + { + var incoming = MessageSerializer.DeserializeCompatible(_typeReq, frame.Payload); + return this._invoker.Invoke(this._instance, new object[] { incoming, frame.FrameId, client }); + } + return null; + } + } + + private readonly Dictionary> _handlers = + new Dictionary>(); + + private readonly Dictionary _requestors = + new Dictionary(); + + #endregion Routing + + public void Incoming(FrameType type, byte[] data) + { + switch (type) + { + case FrameType.Message: + break; + case FrameType.Request: + break; + case FrameType.Response: + break; + } + } + + public void RegisterInbox(string inbox, MessageHandler handler) + { + throw new System.NotImplementedException(); + } + + public void RegisterInbox(string inbox, MessageHandler handler) + { + throw new System.NotImplementedException(); + } + + public void RegisterInbox(MessageHandler handler) + { + throw new System.NotImplementedException(); + } + + public void RegisterInbox(MessageHandler handler) + { + throw new System.NotImplementedException(); + } + + public void RegisterInbox(string inbox, RequestHandler handler) + { + throw new System.NotImplementedException(); + } + + public void RegisterInbox(string inbox, RequestHandler handler) + { + throw new System.NotImplementedException(); + } + + public void RegisterInbox(RequestHandler handler) + { + throw new System.NotImplementedException(); + } + + public void RegisterInbox(RequestHandler handler) + { + throw new System.NotImplementedException(); + } + } +} diff --git a/ZeroLevel/Services/_Network/NetworkDelegates.cs b/ZeroLevel/Services/_Network/NetworkDelegates.cs new file mode 100644 index 0000000..0a1db6b --- /dev/null +++ b/ZeroLevel/Services/_Network/NetworkDelegates.cs @@ -0,0 +1,7 @@ +namespace ZeroLevel.Services._Network +{ + public delegate void MessageHandler(ISocketClient client); + public delegate void MessageHandler(ISocketClient client, T message); + public delegate Tresponse RequestHandler(ISocketClient client); + public delegate Tresponse RequestHandler(ISocketClient client, Trequest request); +} diff --git a/ZeroLevel/Services/_Network/NetworkNode.cs b/ZeroLevel/Services/_Network/NetworkNode.cs new file mode 100644 index 0000000..b09ccc0 --- /dev/null +++ b/ZeroLevel/Services/_Network/NetworkNode.cs @@ -0,0 +1,121 @@ +using System; +using ZeroLevel._Network; + +namespace ZeroLevel.Services._Network +{ + public class NetworkNode + : IClient, IRouter + { + private FrameParser _parser = new FrameParser(); + private readonly ISocketClient _client; + private readonly IRouter _router; + private DateTime _lastConnectionTime; + + public NetworkNode(ISocketClient client, IRouter router) + { + _lastConnectionTime = DateTime.UtcNow; + _client = client; + _router = router; + _parser.OnIncoming += _parser_OnIncoming; + _client.OnIncomingData += _readerWriter_OnIncomingData; + } + + private void _readerWriter_OnIncomingData(byte[] data, int length) + { + _parser.Push(data, length); + } + + private void _parser_OnIncoming(FrameType type, int identity, byte[] data) + { + switch (type) + { + case FrameType.KeepAlive: + _lastConnectionTime = DateTime.UtcNow; + break; + case FrameType.Message: + break; + case FrameType.Request: + break; + case FrameType.Response: + break; + } + } + + public void Send(string inbox) + { + throw new System.NotImplementedException(); + } + + public void Send(string inbox, byte[] data) + { + throw new System.NotImplementedException(); + } + + public void Send(string inbox, T message) + { + throw new System.NotImplementedException(); + } + + public byte[] Request(string inbox) + { + throw new System.NotImplementedException(); + } + + public byte[] Request(string inbox, byte[] data) + { + throw new System.NotImplementedException(); + } + + public Tresponse Request(string inbox) + { + throw new System.NotImplementedException(); + } + + public Tresponse Request(string inbox, Trequest request) + { + throw new System.NotImplementedException(); + } + + #region IRouter + public void RegisterInbox(string inbox, MessageHandler handler) + { + _router.RegisterInbox(inbox, handler); + } + + public void RegisterInbox(string inbox, MessageHandler handler) + { + _router.RegisterInbox(inbox, handler); + } + + public void RegisterInbox(MessageHandler handler) + { + _router.RegisterInbox(handler); + } + + public void RegisterInbox(MessageHandler handler) + { + _router.RegisterInbox(handler); + } + + public void RegisterInbox(string inbox, RequestHandler handler) + { + _router.RegisterInbox(inbox, handler); + } + + public void RegisterInbox(string inbox, RequestHandler handler) + { + _router.RegisterInbox(inbox, handler); + } + + public void RegisterInbox(RequestHandler handler) + { + _router.RegisterInbox(handler); + } + + public void RegisterInbox(RequestHandler handler) + { + _router.RegisterInbox(handler); + } + #endregion + } +} diff --git a/ZeroLevel/Services/_Network/NetworkPacketFactory.cs b/ZeroLevel/Services/_Network/NetworkPacketFactory.cs new file mode 100644 index 0000000..b3a0471 --- /dev/null +++ b/ZeroLevel/Services/_Network/NetworkPacketFactory.cs @@ -0,0 +1,90 @@ +using System; +using System.Runtime.CompilerServices; +using System.Threading; + +namespace ZeroLevel.Services._Network +{ + public static class NetworkPacketFactory + { + public const byte MAGIC = 153; + public const byte MAGIC_REQUEST = 155; + public const byte MAGIC_RESPONSE = 185; + public const byte MAGIC_KEEP_ALIVE = 187; + + private static int _current_request_id = 0; + + private static byte[] _keep_alive = new byte[] { 187, 0, 0, 0, 4, 128, 64, 32, 42 }; + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public static byte[] KeepAliveMessage() => _keep_alive; + + public static byte[] Message(byte[] data) + { + var packet = new byte[data.Length + 6]; + packet[0] = MAGIC; + Array.Copy(BitConverter.GetBytes(data.Length), 0, packet, 1, 4); + packet[5] = (byte)(packet[0] ^ packet[1] ^ packet[2] ^ packet[3] ^ packet[4]); + HashData(data, packet[5]); + Array.Copy(data, 0, packet, 6, data.Length); + return packet; + } + + public static byte[] Reqeust(byte[] data, out int requestId) + { + var packet = new byte[data.Length + 6 + 4]; + packet[0] = (MAGIC | MAGIC_REQUEST); + Array.Copy(BitConverter.GetBytes(data.Length), 0, packet, 1, 4); + packet[5] = (byte)(packet[0] ^ packet[1] ^ packet[2] ^ packet[3] ^ packet[4]); + + requestId = Interlocked.Increment(ref _current_request_id); + var id = BitConverter.GetBytes(requestId); + packet[6] = id[0]; + packet[7] = id[1]; + packet[8] = id[2]; + packet[9] = id[3]; + + HashData(data, packet[5]); + Array.Copy(data, 0, packet, 10, data.Length); + return packet; + } + + public static byte[] Response(byte[] data, int requestId) + { + var packet = new byte[data.Length + 6 + 4]; + packet[0] = (MAGIC | MAGIC_RESPONSE); + Array.Copy(BitConverter.GetBytes(data.Length), 0, packet, 1, 4); + packet[5] = (byte)(packet[0] ^ packet[1] ^ packet[2] ^ packet[3] ^ packet[4]); + + var id = BitConverter.GetBytes(requestId); + packet[6] = id[0]; + packet[7] = id[1]; + packet[8] = id[2]; + packet[9] = id[3]; + + HashData(data, packet[5]); + Array.Copy(data, 0, packet, 10, data.Length); + return packet; + } + + private static void HashData(byte[] data, byte initialmask) + { + if (data == null || data.Length == 0) return; + int i = 1; + data[0] ^= initialmask; + for (; i < (data.Length - 8); i += 8) + { + data[i + 0] ^= data[i - 1]; + data[i + 1] ^= data[i + 0]; + data[i + 2] ^= data[i + 1]; + data[i + 3] ^= data[i + 2]; + data[i + 4] ^= data[i + 3]; + data[i + 5] ^= data[i + 4]; + data[i + 6] ^= data[i + 5]; + data[i + 7] ^= data[i + 6]; + } + for (; i < data.Length; i++) + { + data[i] ^= data[i - 1]; + } + } + } +}