continuity/Server/PacketDistributorService.cs

using System.Collections.Concurrent;
using System.Reflection;
using System.Text;
using System.Text.Json;
using System.Text.Json.Nodes;
using MassTransit.Internals;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.VisualBasic.CompilerServices;
using Newtonsoft.Json;

namespace Server;

public class PacketDistributorService : IHostedService
{
    private readonly ConcurrentQueue<RawPacket> _concurrentQueue;
    private readonly ILogger<PacketDistributorService> _logger;
    private readonly Dictionary<OperationCode, Type> _packetsTypes;
    private readonly Dictionary<OperationCode, Type> _packetHandlers;
    private readonly IServiceProvider _serviceProvider;

    public PacketDistributorService(ILogger<PacketDistributorService> logger, IServiceProvider serviceProvider)
    {
        _concurrentQueue = new ConcurrentQueue<RawPacket>();
        _logger = logger;
        _serviceProvider = serviceProvider;
        _packetHandlers = new Dictionary<OperationCode, Type>();
        _packetsTypes = new Dictionary<OperationCode, Type>();

        var packetsWithId = Assembly.GetEntryAssembly()?.GetTypes().AsParallel()
            .Where(type => type.GetCustomAttribute<PacketId>() != null && type.HasInterface(typeof(IPacket)) &&
                           !type.IsInterface)
            //.Select(Activator.CreateInstance).Cast<IPacket>()
            .ToDictionary(packet => packet.GetCustomAttribute<PacketId>()!.Code);

        if (packetsWithId == null || packetsWithId.Count == 0)
        {
            _logger.LogCritical("No Packets have been found");
            throw new IncompleteInitialization();
        }

        packetsWithId.AsParallel().ForAll(packet =>
        {
            _logger.LogInformation("Packet with ID: {PacketID} has been added as {PacketName}", packet.Key,
                packet.Value.FullName);
        });
        _packetsTypes = packetsWithId;


        var packetHandlersWithId = Assembly.GetEntryAssembly()?.GetTypes().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.IsGenericType && t.GetGenericTypeDefinition() == typeof(IPacketHandler<>))
                .GetGenericArguments()[0].GetCustomAttribute<PacketId>().Code);


        if (packetHandlersWithId == null || packetHandlersWithId.Count == 0)
        {
            _logger.LogCritical("No PacketHandlers have been found");
            throw new IncompleteInitialization();
        }

        packetHandlersWithId.AsParallel().ForAll(packetHandler =>
        {
            _logger.LogInformation("Packet with ID: {PacketID} has been added as {PacketName}", packetHandler.Key,
                packetHandler.Value.FullName);
        });

        _packetHandlers = packetHandlersWithId;
    }

    public Task StartAsync(CancellationToken cancellationToken)
    {
        return Task.CompletedTask;
    }

    public void AddPacket(RawPacket rawPacket)
    {
        _logger.LogInformation("Packet with ID: {MessageOperationCode} has been received",
            rawPacket.OperationCode);
        _concurrentQueue.Enqueue(rawPacket);
        Parallel.Invoke(DequeueAndProcessAsync);
    }

    private async void DequeueAndProcessAsync()
    {
        if (_concurrentQueue.TryDequeue(out var item))
        {
            Parallel.Invoke(() =>
            {
                _logger.LogInformation("[{TempId}] Packet with ID: {MessageOperationCode} is being dequeued",
                    item.Session.Id, item.OperationCode);
                var packetType = _packetsTypes[item.OperationCode];
                var packet = Activator.CreateInstance(packetType);
                packetType.GetFields().AsParallel().ForAll(field =>
                {
                    var typeOfField = field.FieldType;
                    var fieldOffsetAttribute = field.GetCustomAttribute<FieldOffsetAttribute>();
                    object value;
                    _logger.LogInformation("Type of field of {fieldName}: {Name}", field.Name, typeOfField.Name);
                    switch (typeOfField.Name)
                    {
                        case nameof(String):
                        {
                            value = Encoding.ASCII.GetString(item.MessageBody, fieldOffsetAttribute.Offset,
                                fieldOffsetAttribute.Size);
                            break;
                        }
                        default:
                        {
                            value = 0;
                            break;
                        }
                    }

                    field.SetValue(packet, value);
                });
                var packetHandler =
                    ActivatorUtilities.GetServiceOrCreateInstance(_serviceProvider,
                        _packetHandlers[item.OperationCode]);
                packetHandler.GetType().GetMethod("Handle")?.Invoke(packetHandler, new[] { packet });
                _logger.LogInformation("Packet data {PacketData}", JsonConvert.SerializeObject(packet));
                _logger.LogInformation("[{TempId}] Packet with ID: {MessageOperationCode} has finished",
                    item.Session.Id,
                    item.OperationCode);
            });
        }
        else
        {
            await Task.Delay(100); // Delay to prevent busy-waiting, can be adjusted based on needs
        }
    }

    public Task StopAsync(CancellationToken cancellationToken)
    {
        return Task.CompletedTask;
    }
}