feat: extraction of generic Packet distribution service

Continuity.AuthServer/AuthSession.cs

@@ -6,6 +6,7 @@ using Continuity.AuthServer.Packets;
 using MassTransit.Mediator;
 using Microsoft.Extensions.Logging;
 using NetCoreServer;
+using Rai.PacketMediator;
 using Wonderking.Packets;
 
 namespace Continuity.AuthServer;
@@ -30,11 +31,11 @@ public class AuthSession : TcpSession
         return base.Send(buffer);
     }
 
-    public Task SendAsync(IPacket packet)
+    public Task SendAsync(IOutgoingPacket packet)
     {
         var type = packet.GetType();
         _logger.LogInformation("Packet of type {Type} is being serialized", type.Name);
-        var packetIdAttribute = type.GetCustomAttribute<PacketIdAttribute>();
+        var packetIdAttribute = type.GetCustomAttribute<WonderkingPacketIdAttribute>();
         if (packetIdAttribute == null)
         {
             return Task.CompletedTask;

Continuity.AuthServer/Consumers/PacketConsumer.cs

@@ -1,23 +1,26 @@
 // Copyright (c) 2023 Timothy Schenk. Subject to the GNU AGPL Version 3 License.
 
 using Continuity.AuthServer.Packets;
-using Continuity.AuthServer.Services;
 using MassTransit;
+using NetCoreServer;
+using Rai.PacketMediator;
+using Wonderking.Packets;
 
 namespace Continuity.AuthServer.Consumers;
 
 public class PacketConsumer : IConsumer<RawPacket>
 {
-    private readonly PacketDistributorService _distributorService;
+    private readonly PacketDistributorService<OperationCode, TcpSession> _distributorService;
 
-    public PacketConsumer(PacketDistributorService distributorService)
+    public PacketConsumer(PacketDistributorService<OperationCode, TcpSession> distributorService)
     {
         _distributorService = distributorService;
     }
 
     public Task Consume(ConsumeContext<RawPacket> context)
     {
-        _distributorService.AddPacket(context.Message);
+        _distributorService.AddPacket(context.Message.MessageBody, context.Message.OperationCode,
+            context.Message.Session);
         return Task.CompletedTask;
     }
 }

Continuity.AuthServer/PacketHandlers/ChannelSelectionHandler.cs

@@ -5,13 +5,14 @@ using Continuity.AuthServer.DB.Documents;
 using DotNext.Collections.Generic;
 using Microsoft.EntityFrameworkCore;
 using NetCoreServer;
+using Rai.PacketMediator;
 using Wonderking.Packets.Incoming;
 using Wonderking.Packets.Outgoing;
 using Wonderking.Packets.Outgoing.Data;
 
 namespace Continuity.AuthServer.PacketHandlers;
 
-public partial class ChannelSelectionHandler : IPacketHandler<ChannelSelectionPacket>
+public partial class ChannelSelectionHandler : IPacketHandler<ChannelSelectionPacket, TcpSession>
 {
     private readonly WonderkingContext _wonderkingContext;
 

Continuity.AuthServer/PacketHandlers/CharacterCreationHandler.cs

@@ -5,6 +5,7 @@ using Continuity.AuthServer.DB.Documents;
 using Continuity.AuthServer.Services;
 using Microsoft.EntityFrameworkCore;
 using NetCoreServer;
+using Rai.PacketMediator;
 using Wonderking.Game.Data.Character;
 using Wonderking.Game.Mapping;
 using Wonderking.Packets.Incoming;
@@ -13,7 +14,7 @@ using Wonderking.Packets.Outgoing.Data;
 
 namespace Continuity.AuthServer.PacketHandlers;
 
-public class CharacterCreationHandler : IPacketHandler<CharacterCreationPacket>
+public class CharacterCreationHandler : IPacketHandler<CharacterCreationPacket, TcpSession>
 {
     private readonly CharacterStatsMappingConfiguration _characterStatsMapping;
     private readonly ItemObjectPoolService _itemObjectPoolService;

Continuity.AuthServer/PacketHandlers/CharacterDeletionHandler.cs

@@ -3,12 +3,13 @@
 using Continuity.AuthServer.DB;
 using Microsoft.EntityFrameworkCore;
 using NetCoreServer;
+using Rai.PacketMediator;
 using Wonderking.Packets.Incoming;
 using Wonderking.Packets.Outgoing;
 
 namespace Continuity.AuthServer.PacketHandlers;
 
-public class CharacterDeletionHandler : IPacketHandler<CharacterDeletePacket>
+public class CharacterDeletionHandler : IPacketHandler<CharacterDeletePacket, TcpSession>
 {
     private readonly WonderkingContext _wonderkingContext;
 

Continuity.AuthServer/PacketHandlers/CharacterNameCheckHandler.cs

@@ -3,12 +3,13 @@
 using Continuity.AuthServer.DB;
 using Microsoft.EntityFrameworkCore;
 using NetCoreServer;
+using Rai.PacketMediator;
 using Wonderking.Packets.Incoming;
 using Wonderking.Packets.Outgoing;
 
 namespace Continuity.AuthServer.PacketHandlers;
 
-public class CharacterNameCheckHandler : IPacketHandler<CharacterNameCheckPacket>
+public class CharacterNameCheckHandler : IPacketHandler<CharacterNameCheckPacket, TcpSession>
 {
     private readonly WonderkingContext _wonderkingContext;
 

Continuity.AuthServer/PacketHandlers/IPacketHandler.cs

@@ -1,37 +0,0 @@
-// Copyright (c) 2023 Timothy Schenk. Subject to the GNU AGPL Version 3 License.
-
-using System.Diagnostics;
-using JetBrains.Annotations;
-using NetCoreServer;
-using Wonderking.Packets;
-
-namespace Continuity.AuthServer.PacketHandlers;
-
-[UsedImplicitly(ImplicitUseTargetFlags.WithInheritors)]
-public interface IPacketHandler<in T> : IPacketHandler where T : IPacket
-{
-    [UsedImplicitly]
-    public Task HandleAsync(T packet, TcpSession session);
-
-    async Task<bool> IPacketHandler.TryHandleAsync(IPacket packet, TcpSession session)
-    {
-        if (packet is not T tPacket)
-        {
-            return false;
-        }
-
-        using (var activity = new ActivitySource(nameof(Server)).StartActivity("HandleAsync"))
-        {
-            activity?.SetTag("Handler", this.ToString());
-            activity?.SetTag("PacketId", packet.ToString());
-            await HandleAsync(tPacket, session);
-        }
-
-        return true;
-    }
-}
-
-public interface IPacketHandler
-{
-    Task<bool> TryHandleAsync(IPacket packet, TcpSession session);
-}

Continuity.AuthServer/PacketHandlers/LoginHandler.cs

@@ -11,13 +11,14 @@ using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.Logging;
 using NetCoreServer;
+using Rai.PacketMediator;
 using Wonderking.Packets.Incoming;
 using Wonderking.Packets.Outgoing;
 using Wonderking.Packets.Outgoing.Data;
 
 namespace Continuity.AuthServer.PacketHandlers;
 
-public class LoginHandler : IPacketHandler<LoginInfoPacket>
+public class LoginHandler : IPacketHandler<LoginInfoPacket, TcpSession>
 {
     private readonly IConfiguration _configuration;
     private readonly ILogger<LoginHandler> _logger;

Continuity.AuthServer/Program.cs

@@ -12,12 +12,15 @@ using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Diagnostics.HealthChecks;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
+using NetCoreServer;
 using Npgsql;
 using OpenTelemetry.Logs;
 using OpenTelemetry.Metrics;
 using OpenTelemetry.Resources;
 using OpenTelemetry.Trace;
+using Rai.PacketMediator;
 using Wonderking.Game.Mapping;
+using Wonderking.Packets;
 
 var builder = Host.CreateApplicationBuilder();
 #if DEBUG
@@ -83,7 +86,8 @@ if (configuration.GetValue<bool>("Tracing:Enabled"))
     {
         tracing.AddZipkinExporter(options =>
             options.Endpoint = new Uri(configuration["Zipkin:Endpoint"] ?? string.Empty));
-        tracing.AddOtlpExporter(options => options.Endpoint = new Uri(configuration["OTLP:Tracing:Endpoint"] ?? string.Empty));
+        tracing.AddOtlpExporter(options =>
+            options.Endpoint = new Uri(configuration["OTLP:Tracing:Endpoint"] ?? string.Empty));
     });
 }
 
@@ -102,12 +106,13 @@ builder.Services.AddSingleton<CharacterStatsMappingConfiguration>(
         File.ReadAllText("config/character-stats.mapping.json")) ?? throw new InvalidOperationException());
 
 builder.Services.AddSingleton<ILoggerFactory>(loggerFactory);
-builder.Services.AddSingleton<PacketDistributorService>();
+builder.Services.AddSingleton<PacketDistributorService<OperationCode, TcpSession>>();
 builder.Services.AddSingleton<ItemObjectPoolService>();
 builder.Services.AddHostedService(provider =>
     provider.GetService<ItemObjectPoolService>() ?? throw new InvalidOperationException());
 builder.Services.AddHostedService(provider =>
-    provider.GetService<PacketDistributorService>() ?? throw new InvalidOperationException());
+    provider.GetService<PacketDistributorService<OperationCode, TcpSession>>() ??
+    throw new InvalidOperationException());
 builder.Services.AddMassTransit(x =>
 {
     x.UsingInMemory((context, configurator) => configurator.ConfigureEndpoints(context));

Continuity.AuthServer/Services/PacketDistributorService.cs

@@ -1,187 +0,0 @@
-// Copyright (c) 2023 Timothy Schenk. Subject to the GNU AGPL Version 3 License.
-
-using System.Collections.Concurrent;
-using System.Collections.Immutable;
-using System.Diagnostics;
-using System.Reflection;
-using Continuity.AuthServer.LoggerMessages;
-using Continuity.AuthServer.PacketHandlers;
-using Continuity.AuthServer.Packets;
-using DotNext.Collections.Generic;
-using DotNext.Linq.Expressions;
-using DotNext.Metaprogramming;
-using MassTransit.Internals;
-using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.Hosting;
-using Microsoft.Extensions.Logging;
-using Microsoft.VisualBasic.CompilerServices;
-using Newtonsoft.Json;
-using Wonderking.Packets;
-
-namespace Continuity.AuthServer.Services;
-
-using static CodeGenerator;
-using static ExpressionBuilder;
-
-public class PacketDistributorService : IHostedService, IDisposable
-{
-    private readonly ConcurrentQueue<RawPacket> _concurrentQueue;
-
-    private readonly ILogger<PacketDistributorService> _logger;
-    private readonly IServiceProvider _serviceProvider;
-
-    private ImmutableDictionary<OperationCode,
-        Func<byte[], IPacket>> _deserializationMap;
-
-    private ConcurrentDictionary<OperationCode, IPacketHandler> _packetHandlersInstantiation;
-    private readonly ActivitySource _activitySource;
-
-    public PacketDistributorService(ILogger<PacketDistributorService> logger, IServiceProvider serviceProvider)
-    {
-        _concurrentQueue = new ConcurrentQueue<RawPacket>();
-        _logger = logger;
-        _serviceProvider = serviceProvider;
-        _activitySource = new ActivitySource(nameof(Server));
-    }
-
-    public Task StartAsync(CancellationToken cancellationToken)
-    {
-        var tempDeserializationMap =
-            new Dictionary<OperationCode, Func<byte[], IPacket>>();
-
-        var wonderkingAssembly = Assembly.GetAssembly(typeof(IPacket));
-        var packetsTypes = GetPacketsWithId(wonderkingAssembly);
-        var packetHandlers = GetAllPacketHandlersWithId(Assembly.GetExecutingAssembly());
-        _packetHandlersInstantiation = new ConcurrentDictionary<OperationCode, IPacketHandler>();
-        packetHandlers.ForEach(x =>
-        {
-            var packetHandler =
-                ActivatorUtilities.GetServiceOrCreateInstance(_serviceProvider,
-                    x.Value);
-            _packetHandlersInstantiation.TryAdd(x.Key, packetHandler as IPacketHandler);
-        });
-        foreach (var packetsType in packetsTypes)
-        {
-            var lambda = Lambda<Func<byte[], IPacket>>(fun =>
-            {
-                var argPacketData = fun[0];
-                var newPacket = packetsType.Value.New();
-
-                var packetVariable = DeclareVariable(packetsType.Value, "packet");
-                Assign(packetVariable, newPacket);
-                Call(packetVariable, nameof(IPacket.Deserialize), argPacketData);
-
-                Return(packetVariable);
-            }).Compile();
-            _logger.PacketCreationFunctionCreated(packetsType.Key);
-            tempDeserializationMap.Add(packetsType.Key, lambda);
-        }
-
-        _deserializationMap = tempDeserializationMap.ToImmutableDictionary();
-        return Task.CompletedTask;
-    }
-
-    public Task StopAsync(CancellationToken cancellationToken)
-    {
-        return Task.CompletedTask;
-    }
-
-    private Dictionary<OperationCode, Type> GetPacketsWithId(Assembly executingAssembly)
-    {
-        // ! : We are filtering if types that don't have an instance of the required Attribute
-        var packetsWithId = executingAssembly.GetTypes().AsParallel()
-            .Where(type => type.HasInterface(typeof(IPacket)) && type is { IsInterface: false, IsAbstract: false })
-            .Where(type => type.Namespace?.Contains("Incoming") ?? false)
-            .Select(type => new { Type = type, Attribute = type.GetCustomAttribute<PacketIdAttribute>() })
-            .Where(item => item.Attribute is not null)
-            .ToDictionary(item => item.Attribute!.Code, item => item.Type);
-        if (packetsWithId is not { Count: 0 })
-        {
-            packetsWithId.AsParallel()
-                .ForAll(packet => _logger.PacketWithIdAdded(packet.Key, packet.Value.FullName));
-            return packetsWithId;
-        }
-
-        _logger.NoPacketsFound();
-        throw new IncompleteInitialization();
-    }
-
-    private Dictionary<OperationCode, Type> GetAllPacketHandlersWithId(Assembly assembly)
-    {
-        // ! : We are filtering if types that don't have an instance of the required Attribute
-        var packetHandlersWithId = assembly.GetTypes().AsParallel()
-            .Where(t =>
-                t is { IsClass: true, IsAbstract: false } && Array.Exists(t
-                    .GetInterfaces(), i =>
-                    i.IsGenericType && i.GetGenericTypeDefinition() == typeof(IPacketHandler<>)))
-            .Select(type => new
-            {
-                Type = type,
-                PacketId = type
-                    .GetInterfaces().First(t1 =>
-                        t1 is { IsGenericType: true } && t1.GetGenericTypeDefinition() == typeof(IPacketHandler<>))
-                    .GetGenericArguments()[0].GetCustomAttribute<PacketIdAttribute>()?.Code
-            })
-            .Where(x => x.PacketId is not null)
-            .ToDictionary(
-                x => x.PacketId!.Value, x => x.Type
-            );
-
-        if (packetHandlersWithId is not { Count: 0 })
-        {
-            packetHandlersWithId.AsParallel().ForAll(packetHandler =>
-                _logger.PacketHandlerWithIdAdded(packetHandler.Key, packetHandler.Value.FullName));
-            return packetHandlersWithId;
-        }
-
-        _logger.NoPacketHandlersFound();
-        throw new IncompleteInitialization();
-    }
-
-    public void AddPacket(RawPacket rawPacket)
-    {
-        _concurrentQueue.Enqueue(rawPacket);
-        DequeueRawPacket();
-        _logger.PacketReceived(rawPacket.OperationCode);
-    }
-
-    private void DequeueRawPacket()
-    {
-        if (_concurrentQueue.TryDequeue(out var item))
-        {
-            ThreadPool.QueueUserWorkItem(InvokePacketHandler, item, preferLocal: false);
-        }
-    }
-
-    private void InvokePacketHandler(RawPacket item)
-    {
-        IPacket packet;
-        _logger.PacketDequeued(item.Session.Id, item.OperationCode);
-        if (!_deserializationMap.TryGetValue(item.OperationCode, out var value))
-        {
-            _logger.PacketTypeNotFound(item.OperationCode);
-            return;
-        }
-
-        using (var packetParsingActivity = _activitySource.StartActivity("PacketParsing"))
-        {
-            packetParsingActivity?.SetTag("PacketId", item.OperationCode);
-            packet = value(item.MessageBody);
-        }
-
-        using (var packetHandlerActivity = _activitySource.StartActivity("PacketHandler"))
-        {
-            packetHandlerActivity?.SetTag("PacketId", item.OperationCode);
-            _ = _packetHandlersInstantiation[item.OperationCode].TryHandleAsync(packet, item.Session);
-        }
-
-        _logger.PacketData(JsonConvert.SerializeObject(packet));
-        _logger.PacketFinished(item.Session.Id, item.OperationCode);
-    }
-
-    public void Dispose()
-    {
-        GC.SuppressFinalize(this);
-        _activitySource.Dispose();
-    }
-}

Continuity.sln

@@ -6,6 +6,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Benchmarks", "Benchmarks\Be
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Wonderking", "Wonderking\Wonderking.csproj", "{6B53A10B-C397-4347-BB00-A12272D0528E}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Rai.PacketMediator", "Rai.PacketMediator\Rai.PacketMediator.csproj", "{D6FA787F-6B95-4679-BC6F-EED10B591E5C}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -24,5 +26,9 @@ Global
 		{6B53A10B-C397-4347-BB00-A12272D0528E}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{6B53A10B-C397-4347-BB00-A12272D0528E}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{6B53A10B-C397-4347-BB00-A12272D0528E}.Release|Any CPU.Build.0 = Release|Any CPU
+		{D6FA787F-6B95-4679-BC6F-EED10B591E5C}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{D6FA787F-6B95-4679-BC6F-EED10B591E5C}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{D6FA787F-6B95-4679-BC6F-EED10B591E5C}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{D6FA787F-6B95-4679-BC6F-EED10B591E5C}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 EndGlobal

Rai.PacketMediator/IBidirectionalPacket.cs

@@ -2,11 +2,7 @@
 
 using JetBrains.Annotations;
 
-namespace Wonderking.Packets;
+namespace Rai.PacketMediator;
 
 [UsedImplicitly(ImplicitUseTargetFlags.WithInheritors)]
-public interface IPacket
+public interface IBidirectionalPacket : IOutgoingPacket, IIncomingPacket;
-{
-    public void Deserialize(byte[] data);
-    public byte[] Serialize();
-}

Rai.PacketMediator/IIncomingPacket.cs

@@ -0,0 +1,8 @@
+// Copyright (c) 2023 Timothy Schenk. Subject to the GNU AGPL Version 3 License.
+
+namespace Rai.PacketMediator;
+
+public interface IIncomingPacket : IPacket
+{
+    public void Deserialize(byte[] data);
+}

Rai.PacketMediator/IOutgoingPacket.cs

@@ -0,0 +1,8 @@
+// Copyright (c) 2023 Timothy Schenk. Subject to the GNU AGPL Version 3 License.
+
+namespace Rai.PacketMediator;
+
+public interface IOutgoingPacket : IPacket
+{
+    public byte[] Serialize();
+}

Rai.PacketMediator/IPacket.cs

@@ -0,0 +1,5 @@
+// Copyright (c) 2023 Timothy Schenk. Subject to the GNU AGPL Version 3 License.
+
+namespace Rai.PacketMediator;
+
+public interface IPacket;

Rai.PacketMediator/IPacketHandler.cs

@@ -0,0 +1,33 @@
+// Copyright (c) 2023 Timothy Schenk. Subject to the GNU AGPL Version 3 License.
+
+using System.Diagnostics;
+using JetBrains.Annotations;
+
+namespace Rai.PacketMediator;
+
+[UsedImplicitly(ImplicitUseTargetFlags.WithInheritors)]
+public interface IPacketHandler<in T, in S> : IPacketHandler<S> where T : IIncomingPacket
+{
+    [UsedImplicitly]
+    public Task HandleAsync(T packet, S session);
+
+    async Task<bool> IPacketHandler<S>.TryHandleAsync(IIncomingPacket packet, S session)
+    {
+        if (packet is not T tPacket)
+        {
+            return false;
+        }
+
+        using var activity = new ActivitySource(nameof(PacketMediator)).StartActivity(nameof(HandleAsync));
+        activity?.AddTag("Handler", this.ToString());
+        activity?.AddTag("Packet", packet.ToString());
+        await HandleAsync(tPacket, session);
+
+        return true;
+    }
+}
+
+public interface IPacketHandler<in S>
+{
+    Task<bool> TryHandleAsync(IIncomingPacket packet, S session);
+}

Rai.PacketMediator/PacketDistributorService.cs

@@ -0,0 +1,171 @@
+// Copyright (c) 2023 Timothy Schenk. Subject to the GNU AGPL Version 3 License.
+
+using System.Collections.Concurrent;
+using System.Collections.Immutable;
+using System.Diagnostics;
+using System.Reflection;
+using DotNext.Collections.Generic;
+using DotNext.Linq.Expressions;
+using DotNext.Metaprogramming;
+using MassTransit.Internals;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.VisualBasic.CompilerServices;
+
+namespace Rai.PacketMediator;
+
+using static CodeGenerator;
+
+public class PacketDistributorService<T, S> : Microsoft.Extensions.Hosting.IHostedService, IDisposable where T : Enum
+{
+    private readonly ConcurrentQueue<ValueTuple<byte[], T, S>> _concurrentQueue;
+
+    private readonly IServiceProvider _serviceProvider;
+    private readonly Assembly[] _sourcesContainingPackets;
+
+    private ImmutableDictionary<T,
+        Func<byte[], IIncomingPacket>> _deserializationMap;
+
+    private ConcurrentDictionary<T, IPacketHandler<S>?> _packetHandlersInstantiation;
+    private readonly ActivitySource _activitySource;
+
+    public PacketDistributorService(IServiceProvider serviceProvider,
+        Assembly[] sourcesContainingPackets)
+    {
+        _concurrentQueue = new ConcurrentQueue<ValueTuple<byte[], T, S>>();
+        _serviceProvider = serviceProvider;
+        _sourcesContainingPackets = sourcesContainingPackets;
+        _activitySource = new ActivitySource(nameof(PacketMediator));
+    }
+
+    private Dictionary<T, Type> RetrievePacketsDictionary()
+    {
+        using var activity = this._activitySource.StartActivity();
+        var packetsWithId = this._sourcesContainingPackets.SelectMany(a => a.GetTypes()
+            .Where(type =>
+                type.HasInterface(typeof(IIncomingPacket)) && type is { IsInterface: false, IsAbstract: false })
+            .Select(type => new { Type = type, Attribute = type.GetCustomAttribute<PacketIdAttribute<T>>() })
+            .Where(item => item.Attribute is not null)
+            .ToDictionary(item => item.Attribute!.Code, item => item.Type)).ToDictionary();
+
+        activity?.AddTag("AmountOfPackets", packetsWithId.Count);
+
+        return packetsWithId;
+    }
+
+    private Dictionary<T, Type> GetAllPacketHandlersWithId()
+    {
+        using var activity = this._activitySource.StartActivity();
+        var packetHandlersWithId = this._sourcesContainingPackets.SelectMany(assembly => assembly.GetTypes()
+            .Where(t =>
+                t is { IsClass: true, IsAbstract: false } && Array.Exists(t
+                    .GetInterfaces(), i =>
+                    i.IsGenericType && i.GetGenericTypeDefinition() == typeof(IPacketHandler<S>)))
+            .Select(type => new
+            {
+                Type = type,
+                PacketId = type
+                    .GetInterfaces().First(t1 =>
+                        t1 is { IsGenericType: true } && t1.GetGenericTypeDefinition() == typeof(IPacketHandler<S>))
+                    .GetGenericArguments().First(t => t.HasAttribute<PacketIdAttribute<T>>())
+                    .GetAttribute<PacketIdAttribute<T>>().First().Code
+            })
+            .ToDictionary(
+                x => x.PacketId, x => x.Type
+            )).ToDictionary();
+        activity?.AddTag("AmountOfPacketHandlers", packetHandlersWithId.Count);
+        return packetHandlersWithId;
+    }
+
+    public void Dispose()
+    {
+        GC.SuppressFinalize(this);
+        _activitySource.Dispose();
+    }
+
+    public Task StartAsync(CancellationToken cancellationToken)
+    {
+        var packetDictionary = RetrievePacketsDictionary();
+
+        if (packetDictionary is { Count: 0 })
+        {
+            throw new IncompleteInitialization();
+        }
+
+        var packetHandlers = GetAllPacketHandlersWithId();
+
+        if (packetHandlers is { Count: 0 })
+        {
+            throw new IncompleteInitialization();
+        }
+
+        var tempDeserializationMap =
+            new Dictionary<T, Func<byte[], IIncomingPacket>>();
+        _packetHandlersInstantiation = new ConcurrentDictionary<T, IPacketHandler<S>?>();
+        packetHandlers.ForEach(x =>
+        {
+            var packetHandler =
+                ActivatorUtilities.GetServiceOrCreateInstance(_serviceProvider,
+                    x.Value);
+            _packetHandlersInstantiation.TryAdd(x.Key, packetHandler as IPacketHandler<S>);
+        });
+        foreach (var packetsType in packetDictionary)
+        {
+            var lambda = Lambda<Func<byte[], IIncomingPacket>>(fun =>
+            {
+                var argPacketData = fun[0];
+                var newPacket = packetsType.Value.New();
+
+                var packetVariable = DeclareVariable(packetsType.Value, "packet");
+                Assign(packetVariable, newPacket);
+                Call(packetVariable, nameof(IIncomingPacket.Deserialize), argPacketData);
+
+                Return(packetVariable);
+            }).Compile();
+            tempDeserializationMap.Add(packetsType.Key, lambda);
+        }
+
+        _deserializationMap = tempDeserializationMap.ToImmutableDictionary();
+        return Task.CompletedTask;
+    }
+
+    public void AddPacket(byte[] packetData, T operationCode, S session)
+    {
+        _concurrentQueue.Enqueue((packetData, operationCode, session));
+        DequeueRawPacket();
+    }
+
+    private void DequeueRawPacket()
+    {
+        if (_concurrentQueue.TryDequeue(out var item))
+        {
+            ThreadPool.QueueUserWorkItem(InvokePacketHandler, item, preferLocal: false);
+        }
+    }
+
+    private void InvokePacketHandler((byte[], T, S) valueTuple)
+    {
+        IIncomingPacket packet;
+        var (packetData, operationCode, session) = valueTuple;
+        if (!_deserializationMap.TryGetValue(operationCode, out var func))
+        {
+            return;
+        }
+
+        using (var packetParsingActivity = _activitySource.StartActivity("PacketParsing"))
+        {
+            packetParsingActivity?.SetTag("PacketId", operationCode);
+            packet = func(packetData);
+        }
+
+        using (var packetHandlerActivity = _activitySource.StartActivity("PacketHandler"))
+        {
+            packetHandlerActivity?.SetTag("PacketId", operationCode);
+            _ = _packetHandlersInstantiation[operationCode]?.TryHandleAsync(packet, session);
+        }
+    }
+
+    public Task StopAsync(CancellationToken cancellationToken)
+    {
+        return Task.CompletedTask;
+    }
+}

Rai.PacketMediator/PacketIdAttribute.cs

@@ -0,0 +1,14 @@
+// Copyright (c) 2023 Timothy Schenk. Subject to the GNU AGPL Version 3 License.
+
+namespace Rai.PacketMediator;
+
+[AttributeUsage(AttributeTargets.Class, Inherited = false)]
+public abstract class PacketIdAttribute<T> : Attribute where T : Enum
+{
+    protected PacketIdAttribute(T code)
+    {
+        Code = code;
+    }
+
+    public T Code { get; }
+}

Rai.PacketMediator/Rai.PacketMediator.csproj

@@ -0,0 +1,20 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <TargetFramework>net8.0</TargetFramework>
+        <ImplicitUsings>enable</ImplicitUsings>
+        <Nullable>enable</Nullable>
+    </PropertyGroup>
+
+    <ItemGroup>
+      <PackageReference Include="DotNext" Version="5.0.1" />
+      <PackageReference Include="DotNext.Metaprogramming" Version="5.0.1" />
+      <PackageReference Include="DotNext.Threading" Version="5.0.1" />
+      <PackageReference Include="DotNext.Unsafe" Version="5.0.1" />
+      <PackageReference Include="JetBrains.Annotations" Version="2023.3.0" />
+      <PackageReference Include="MassTransit" Version="8.1.3" />
+      <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="8.0.0" />
+      <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="8.0.0" />
+    </ItemGroup>
+
+</Project>

Wonderking/Packets/Incoming/ChannelSelectionPacket.cs

@@ -1,9 +1,11 @@
 // Copyright (c) 2023 Timothy Schenk. Subject to the GNU AGPL Version 3 License.
 
+using Rai.PacketMediator;
+
 namespace Wonderking.Packets.Incoming;
 
-[PacketId(OperationCode.ChannelSelection)]
+[WonderkingPacketId(OperationCode.ChannelSelection)]
-public class ChannelSelectionPacket : IPacket
+public class ChannelSelectionPacket : IIncomingPacket
 {
     public required ushort ServerId { get; set; }
     public required ushort ChannelId { get; set; }

Wonderking/Packets/Incoming/CharacterCreationPacket.cs

@@ -1,12 +1,13 @@
 // Copyright (c) 2023 Timothy Schenk. Subject to the GNU AGPL Version 3 License.
 
 using System.Text;
+using Rai.PacketMediator;
 using Wonderking.Game.Data.Character;
 
 namespace Wonderking.Packets.Incoming;
 
-[PacketId(OperationCode.CharacterCreation)]
+[WonderkingPacketId(OperationCode.CharacterCreation)]
-public class CharacterCreationPacket : IPacket
+public class CharacterCreationPacket : IIncomingPacket
 {
     public required byte Slot { get; set; }
     public required byte Unknown { get; set; }

Wonderking/Packets/Incoming/CharacterDeletePacket.cs

@@ -1,11 +1,12 @@
 // Copyright (c) 2023 Timothy Schenk. Subject to the GNU AGPL Version 3 License.
 
 using System.Text;
+using Rai.PacketMediator;
 
 namespace Wonderking.Packets.Incoming;
 
-[PacketId(OperationCode.CharacterDeletion)]
+[WonderkingPacketId(OperationCode.CharacterDeletion)]
-public class CharacterDeletePacket : IPacket
+public class CharacterDeletePacket : IIncomingPacket
 {
     public byte Slot { get; set; }
     public string Name { get; set; }

Wonderking/Packets/Incoming/CharacterNameCheckPacket.cs

@@ -1,11 +1,12 @@
 // Copyright (c) 2023 Timothy Schenk. Subject to the GNU AGPL Version 3 License.
 
 using System.Text;
+using Rai.PacketMediator;
 
 namespace Wonderking.Packets.Incoming;
 
-[PacketId(OperationCode.CharacterNameCheck)]
+[WonderkingPacketId(OperationCode.CharacterNameCheck)]
-public class CharacterNameCheckPacket : IPacket
+public class CharacterNameCheckPacket : IIncomingPacket
 {
     public required string Name { get; set; }
 

Wonderking/Packets/Incoming/LoginInfoPacket.cs

@@ -1,11 +1,12 @@
 // Copyright (c) 2023 Timothy Schenk. Subject to the GNU AGPL Version 3 License.
 
 using System.Text;
+using Rai.PacketMediator;
 
 namespace Wonderking.Packets.Incoming;
 
-[PacketId(OperationCode.LoginInfo)]
+[WonderkingPacketId(OperationCode.LoginInfo)]
-public class LoginInfoPacket : IPacket
+public class LoginInfoPacket : IIncomingPacket
 {
     public required string Username { get; set; }
 

Wonderking/Packets/Outgoing/ChannelSelectionResponsePacket.cs

@@ -2,12 +2,13 @@
 
 using System.Buffers.Binary;
 using System.Text;
+using Rai.PacketMediator;
 using Wonderking.Packets.Outgoing.Data;
 
 namespace Wonderking.Packets.Outgoing;
 
-[PacketId(OperationCode.ChannelSelectionResponse)]
+[WonderkingPacketId(OperationCode.ChannelSelectionResponse)]
-public class ChannelSelectionResponsePacket : IPacket
+public class ChannelSelectionResponsePacket : IOutgoingPacket
 {
     public required byte ChannelIsFullFlag { get; set; }
     public required string Endpoint { get; set; }

Wonderking/Packets/Outgoing/CharacterCreationResponsePacket.cs

@@ -2,12 +2,13 @@
 
 using System.Buffers.Binary;
 using System.Text;
+using Rai.PacketMediator;
 using Wonderking.Packets.Outgoing.Data;
 
 namespace Wonderking.Packets.Outgoing;
 
-[PacketId(OperationCode.CharacterCreationResponse)]
+[WonderkingPacketId(OperationCode.CharacterCreationResponse)]
-public class CharacterCreationResponsePacket : IPacket
+public class CharacterCreationResponsePacket : IOutgoingPacket
 {
     public required CharacterData Character { get; set; }
     public required int Slot { get; set; }

Wonderking/Packets/Outgoing/CharacterDeleteResponsePacket.cs

@@ -1,11 +1,12 @@
 // Copyright (c) 2023 Timothy Schenk. Subject to the GNU AGPL Version 3 License.
 
+using Rai.PacketMediator;
 using NotSupportedException = System.NotSupportedException;
 
 namespace Wonderking.Packets.Outgoing;
 
-[PacketId(OperationCode.CharacterDeletionResponse)]
+[WonderkingPacketId(OperationCode.CharacterDeletionResponse)]
-public class CharacterDeleteResponsePacket : IPacket
+public class CharacterDeleteResponsePacket : IOutgoingPacket
 {
     public required byte HasToBeZero { get; set; }
 

Wonderking/Packets/Outgoing/CharacterNameCheckPacketResponse.cs

@@ -1,9 +1,11 @@
 // Copyright (c) 2023 Timothy Schenk. Subject to the GNU AGPL Version 3 License.
 
+using Rai.PacketMediator;
+
 namespace Wonderking.Packets.Outgoing;
 
-[PacketId(OperationCode.CharacterNameCheckResponse)]
+[WonderkingPacketId(OperationCode.CharacterNameCheckResponse)]
-public class CharacterNameCheckPacketResponse : IPacket
+public class CharacterNameCheckPacketResponse : IOutgoingPacket
 {
     public required bool IsTaken { get; set; }
 

Wonderking/Packets/Outgoing/CharacterSelectionSetGuildNamePacket.cs

@@ -1,11 +1,12 @@
 // Copyright (c) 2023 Timothy Schenk. Subject to the GNU AGPL Version 3 License.
 
 using System.Text;
+using Rai.PacketMediator;
 
 namespace Wonderking.Packets.Outgoing;
 
-[PacketId(OperationCode.CharacterSelectionSetGuildName)]
+[WonderkingPacketId(OperationCode.CharacterSelectionSetGuildName)]
-public class CharacterSelectionSetGuildNamePacket : IPacket
+public class CharacterSelectionSetGuildNamePacket : IOutgoingPacket
 {
     public required string[] GuildNames { get; set; }
 

Wonderking/Packets/Outgoing/LoginResponsePacket.cs

@@ -1,12 +1,13 @@
 // Copyright (c) 2023 Timothy Schenk. Subject to the GNU AGPL Version 3 License.
 
 using System.Buffers.Binary;
+using Rai.PacketMediator;
 using Wonderking.Packets.Outgoing.Data;
 
 namespace Wonderking.Packets.Outgoing;
 
-[PacketId(OperationCode.LoginResponse)]
+[WonderkingPacketId(OperationCode.LoginResponse)]
-public class LoginResponsePacket : IPacket
+public class LoginResponsePacket : IOutgoingPacket
 {
     public required LoginResponseReason ResponseReason { get; set; }
     public required byte UnknownFlag { get; set; } = 1;

Wonderking/Packets/WonderkingPacketIdAttribute.cs

@@ -1,14 +1,13 @@
 // Copyright (c) 2023 Timothy Schenk. Subject to the GNU AGPL Version 3 License.
 
+using Rai.PacketMediator;
+
 namespace Wonderking.Packets;
 
 [AttributeUsage(AttributeTargets.Class, Inherited = false)]
-public class PacketIdAttribute : Attribute
+public class WonderkingPacketIdAttribute : PacketIdAttribute<OperationCode>
 {
-    public PacketIdAttribute(OperationCode code)
+    public WonderkingPacketIdAttribute(OperationCode code) : base(code)
     {
-        Code = code;
     }
-
-    public OperationCode Code { get; }
 }

Wonderking/Wonderking.csproj

@@ -41,4 +41,7 @@
   <ItemGroup>
     <Folder Include="Game\Writer\"/>
   </ItemGroup>
+  <ItemGroup>
+    <ProjectReference Include="..\Rai.PacketMediator\Rai.PacketMediator.csproj" />
+  </ItemGroup>
 </Project>