using System.Collections.Concurrent; using System.Reflection; using MassTransit.Internals; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; using Microsoft.VisualBasic.CompilerServices; using Newtonsoft.Json; using Server.PacketHandlers; using Server.Packets; namespace Server.Services; public class PacketDistributorService : IHostedService { private readonly ConcurrentQueue _concurrentQueue; private readonly ILogger _logger; private readonly Dictionary _packetsTypes; private readonly Dictionary _packetHandlers; private readonly IServiceProvider _serviceProvider; public PacketDistributorService(ILogger logger, IServiceProvider serviceProvider) { _concurrentQueue = new ConcurrentQueue(); _logger = logger; _serviceProvider = serviceProvider; _packetHandlers = new Dictionary(); _packetsTypes = new Dictionary(); var packetsWithId = Assembly.GetEntryAssembly()?.GetTypes().AsParallel() .Where(type => type.GetCustomAttribute() != null && type.HasInterface(typeof(IPacket)) && !type.IsInterface) //.Select(Activator.CreateInstance).Cast() .ToDictionary(packet => packet.GetCustomAttribute()!.Code); if (packetsWithId == null || packetsWithId.Count == 0) { _logger.LogCritical("No Packets have been found"); throw new IncompleteInitialization(); } packetsWithId.AsParallel().ForAll(packet => { _logger.LogTrace("Packet with ID: {PacketID} has been added as {PacketName}", packet.Key, packet.Value.FullName); }); _packetsTypes = packetsWithId; var packetHandlersWithId = Assembly.GetEntryAssembly()?.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.IsGenericType && t.GetGenericTypeDefinition() == typeof(IPacketHandler<>)) .GetGenericArguments()[0].GetCustomAttribute().Code); if (packetHandlersWithId == null || packetHandlersWithId.Count == 0) { _logger.LogCritical("No PacketHandlers have been found"); throw new IncompleteInitialization(); } packetHandlersWithId.AsParallel().ForAll(packetHandler => { _logger.LogTrace("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) { _concurrentQueue.Enqueue(rawPacket); Task.Run(() => DequeueAndProcessAsync()); _logger.LogInformation("Packet with ID: {MessageOperationCode} has been received", rawPacket.OperationCode); } private async Task DequeueAndProcessAsync() { if (_concurrentQueue.TryDequeue(out var item)) { Task.Run(() => { _logger.LogTrace("[{TempId}] Packet with ID: {MessageOperationCode} is being dequeued", item.Session.Id, item.OperationCode); var packetType = _packetsTypes[item.OperationCode]; var packet = (IPacket)Activator.CreateInstance(packetType)!; packet.Deserialize(item.MessageBody); var packetHandler = ActivatorUtilities.GetServiceOrCreateInstance(_serviceProvider, _packetHandlers[item.OperationCode]); packetHandler.GetType().GetMethod("HandleAsync") ?.Invoke(packetHandler, new Object[] { packet, item.Session }); _logger.LogDebug("Packet data {PacketData}", JsonConvert.SerializeObject(packet)); _logger.LogTrace("[{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; } }