continuity/Server/Services/PacketDistributorService.cs

namespace Server.Services;

using System.Collections.Concurrent;
using System.Collections.Immutable;
using System.Reflection;
using DotNext.Collections.Generic;
using MassTransit.Internals;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.VisualBasic.CompilerServices;
using Newtonsoft.Json;
using PacketHandlers;
using Packets;
using static DotNext.Linq.Expressions.ExpressionBuilder;
using static DotNext.Metaprogramming.CodeGenerator;

public class PacketDistributorService : IHostedService
{
    private readonly ConcurrentQueue<RawPacket> concurrentQueue;

    private readonly
        ImmutableDictionary<OperationCode,
            Func<byte[], IPacket>> deserializationMap;

    private readonly ILogger<PacketDistributorService> logger;
    private readonly ConcurrentDictionary<OperationCode, object> packetHandlersInstantiation;

    private readonly IServiceProvider serviceProvider;

    public PacketDistributorService(ILogger<PacketDistributorService> logger, IServiceProvider serviceProvider)
    {
        this.concurrentQueue = new ConcurrentQueue<RawPacket>();
        this.logger = logger;
        this.serviceProvider = serviceProvider;
        var tempDeserializationMap =
            new Dictionary<OperationCode, Func<byte[], IPacket>>();

        var executingAssembly = Assembly.GetExecutingAssembly();
        var packetsTypes = this.GetPacketsWithId(executingAssembly);
        var packetHandlers = this.GetAllPacketHandlersWithId(executingAssembly);
        this.packetHandlersInstantiation = new ConcurrentDictionary<OperationCode, object>();
        packetHandlers.ForEach(x =>
        {
            var packetHandler =
                ActivatorUtilities.GetServiceOrCreateInstance(this.serviceProvider,
                    x.Value);
            this.packetHandlersInstantiation.TryAdd(x.Key, packetHandler);
        });
        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.LogInformation("Packet creation function created for {Opcode}", packetsType.Key);
            tempDeserializationMap.Add(packetsType.Key, lambda);
        }

        this.deserializationMap = tempDeserializationMap.ToImmutableDictionary();
    }

    public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask;

    public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;

    private Dictionary<OperationCode, Type> GetPacketsWithId(Assembly executingAssembly)
    {
        var packetsWithId = executingAssembly.GetTypes().AsParallel()
            .Where(type => type.HasInterface(typeof(IPacket)) && !type.IsInterface && !type.IsAbstract)
            .Where(type => type.GetCustomAttribute<PacketIdAttribute>() != null)
            .ToDictionary(type => type.GetCustomAttribute<PacketIdAttribute>().Code);
        if (packetsWithId is not { Count: 0 })
        {
            packetsWithId.AsParallel().ForAll(packet => this.logger.LogTrace("Packet with ID: {PacketID} has been added as {PacketName}", packet.Key,
                    packet.Value.FullName));
            return packetsWithId;
        }

        this.logger.LogCritical("No Packets have been found");
        throw new IncompleteInitialization();
    }

    private Dictionary<OperationCode, Type> GetAllPacketHandlersWithId(Assembly assembly)
    {
        var packetHandlersWithId = assembly.GetTypes().AsParallel().Where(t =>
            t is { IsClass: true, IsAbstract: false } && t
                .GetInterfaces().Any(i =>
                    i.IsGenericType && i.GetGenericTypeDefinition() == typeof(IPacketHandler<>))).ToDictionary(type =>
            type.GetInterfaces().First(t =>
                    t is { IsGenericType: true } && t.GetGenericTypeDefinition() == typeof(IPacketHandler<>))
                .GetGenericArguments().First().GetCustomAttribute<PacketIdAttribute>().Code);

        if (packetHandlersWithId is not { Count: 0 })
        {
            packetHandlersWithId.AsParallel().ForAll(packetHandler => this.logger.LogTrace("PacketHandler with ID: {PacketID} has been added as {PacketName}",
                    packetHandler.Key,
                    packetHandler.Value.FullName));
            return packetHandlersWithId;
        }

        this.logger.LogCritical("No PacketHandlers have been found");
        throw new IncompleteInitialization();
    }

    public void AddPacket(RawPacket rawPacket)
    {
        this.concurrentQueue.Enqueue(rawPacket);
        this.DequeueRawPacket();
        this.logger.LogInformation("Packet with ID: {MessageOperationCode} has been received",
            rawPacket.OperationCode);
    }

    private void DequeueRawPacket()
    {
        if (this.concurrentQueue.TryDequeue(out var item))
        {
            ThreadPool.QueueUserWorkItem(this.InvokePacketHandler, item, true);
        }
    }

    private void InvokePacketHandler(RawPacket item)
    {
        this.logger.LogTrace("[{TempId}] Packet with ID: {MessageOperationCode} is being dequeued",
            item.Session.Id, item.OperationCode);
        if (!this.deserializationMap.ContainsKey(item.OperationCode))
        {
            this.logger.LogInformation("Couldn't find Packet type for Id: {Opcode}", item.OperationCode);
            return;
        }

        var packet = this.deserializationMap[item.OperationCode](item.MessageBody);
        this.logger.LogInformation("Packet data {PacketData}", JsonConvert.SerializeObject(packet));
        this.packetHandlersInstantiation[item.OperationCode].GetType().GetMethod("HandleAsync")
            ?.Invoke(this.packetHandlersInstantiation[item.OperationCode], new object[] { packet, item.Session });

        this.logger.LogTrace("[{TempId}] Packet with ID: {MessageOperationCode} has finished",
            item.Session.Id,
            item.OperationCode);
    }
}