wonderking/continuity
2
0
Fork 0
Code Issues Pull requests 16 Projects 1 Releases Packages 1 Wiki Activity Actions

refactor: general reordering for PacketDistributor

Browse source
This commit is contained in:
Timothy Schenk 2023-08-11 11:19:43 +02:00
parent e6996d32ec
commit f0e0c8edfc
1 changed files with 51 additions and 44 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

95
Server/Services/PacketDistributorService.cs
View file

@ -27,47 +27,52 @@ public class PacketDistributorService : IHostedService
_packetHandlers = new Dictionary<OperationCode, Type>();
_packetsTypes = new Dictionary<OperationCode, Type>();
var packetsWithId = Assembly.GetEntryAssembly()?.GetTypes().AsParallel()
var executingAssembly = Assembly.GetExecutingAssembly();
_packetsTypes = GetPacketsWithId(executingAssembly);
_packetHandlers = GetAllPacketHandlersWithId(executingAssembly);
}
private Dictionary<OperationCode, Type> GetPacketsWithId(Assembly executingAssembly)
{
var packetsWithId = executingAssembly.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)
if (packetsWithId is not { 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);
});
return packetsWithId;
}
packetsWithId.AsParallel().ForAll(packet =>
{
_logger.LogTrace("Packet with ID: {PacketID} has been added as {PacketName}", packet.Key,
packet.Value.FullName);
});
_packetsTypes = packetsWithId;
_logger.LogCritical("No Packets have been found");
throw new IncompleteInitialization();
}
var packetHandlersWithId = Assembly.GetEntryAssembly()?.GetTypes().AsParallel().Where(t =>
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.IsGenericType && t.GetGenericTypeDefinition() == typeof(IPacketHandler<>))
.GetGenericArguments()[0].GetCustomAttribute<PacketId>().Code);
if (packetHandlersWithId == null || packetHandlersWithId.Count == 0)
if (packetHandlersWithId is not { Count: 0 })
{
_logger.LogCritical("No PacketHandlers have been found");
throw new IncompleteInitialization();
packetHandlersWithId.AsParallel().ForAll(packetHandler =>
{
_logger.LogTrace("PacketHandler with ID: {PacketID} has been added as {PacketName}", packetHandler.Key,
packetHandler.Value.FullName);
});
return packetHandlersWithId;
}
packetHandlersWithId.AsParallel().ForAll(packetHandler =>
{
_logger.LogTrace("Packet with ID: {PacketID} has been added as {PacketName}", packetHandler.Key,
packetHandler.Value.FullName);
});
_packetHandlers = packetHandlersWithId;
_logger.LogCritical("No PacketHandlers have been found");
throw new IncompleteInitialization();
}
public Task StartAsync(CancellationToken cancellationToken)
@ -78,32 +83,16 @@ public class PacketDistributorService : IHostedService
public void AddPacket(RawPacket rawPacket)
{
_concurrentQueue.Enqueue(rawPacket);
Task.Run(() => DequeueAndProcessAsync());
Task.Run(() => DequeueRawPacketAsync());
_logger.LogInformation("Packet with ID: {MessageOperationCode} has been received",
rawPacket.OperationCode);
}
private async Task DequeueAndProcessAsync()
private async Task DequeueRawPacketAsync()
{
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);
});
Task.Run(() => { InvokePacketHandler(item); });
}
else
{
@ -111,6 +100,24 @@ public class PacketDistributorService : IHostedService
}
}
private void InvokePacketHandler(RawPacket? item)
{
_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);
}
public Task StopAsync(CancellationToken cancellationToken)
{
return Task.CompletedTask;