diff --git a/Components/MineSharp.Protocol/MinecraftClient.cs b/Components/MineSharp.Protocol/MinecraftClient.cs index 4a069c2..498d267 100644 --- a/Components/MineSharp.Protocol/MinecraftClient.cs +++ b/Components/MineSharp.Protocol/MinecraftClient.cs @@ -532,18 +532,36 @@ internal void SetCompression(int threshold) stream!.SetCompression(threshold); } - internal async Task HandleBundleDelimiter() + internal void HandleBundleDelimiter() { var bundledPackets = Interlocked.Exchange(ref this.bundledPackets, null); if (bundledPackets != null) { - Logger.Debug("Processing bundled packets"); - var tasks = bundledPackets.Select( - p => HandleIncomingPacket(p.Type, p.Buffer)) - .ToArray(); + _ = Task.Run(() => ProcessBundledPackets(bundledPackets), CancellationToken); + } + else + { + if (Interlocked.CompareExchange(ref this.bundledPackets, new(), null) != null) + { + Logger.Warn("Bundling could not be enabled because it was already enabled. This is a race condition."); + } + } + } + + private async Task ProcessBundledPackets(ConcurrentQueue<(PacketType, PacketBuffer)> packets) + { + Logger.Debug($"Processing {packets.Count} bundled packets"); + try + { + // wiki.vg: the client is guaranteed to process every packet in the bundle on the same tick + // we don't guarantee that. + // TODO: process bundled packets within a single tick + var tasks = packets.Select( + p => HandleIncomingPacket(p.Item1, p.Item2)) + .ToArray(); // no clearing required the queue will no longer be used and will get GCed - //bundledPackets.Clear(); + // bundledPackets.Clear(); await Task.WhenAll(tasks); @@ -552,16 +570,9 @@ internal async Task HandleBundleDelimiter() Logger.Error(faultedTask.Exception, "Error handling bundled packet."); } } - else + catch (Exception e) { - if (Interlocked.CompareExchange(ref this.bundledPackets, new(), null) == null) - { - Logger.Debug("Bundling packets!"); - } - else - { - Logger.Warn("Bundling could not be enabled because it was already enabled. This is a race condition."); - } + Logger.Error(e, "Error handling bundled packets."); } } @@ -602,6 +613,14 @@ private async Task ReceivePackets() Logger.Trace("Received packet {PacketType}. GameState = {GameState}, PacketId = {PacketId}", packetType, gameState, packetId); + // handle BundleDelimiter packet here, because there is a race condition where some + // packets may be read before HandleBundleDelimiter is invoked through a handler + if (packetType == PacketType.CB_Play_BundleDelimiter) + { + HandleBundleDelimiter(); + continue; + } + if (gameState != GameState.Play) { await HandleIncomingPacket(packetType, buffer); diff --git a/Components/MineSharp.Protocol/Packets/Handlers/PlayPacketHandler.cs b/Components/MineSharp.Protocol/Packets/Handlers/PlayPacketHandler.cs index 765f3fa..7765e37 100644 --- a/Components/MineSharp.Protocol/Packets/Handlers/PlayPacketHandler.cs +++ b/Components/MineSharp.Protocol/Packets/Handlers/PlayPacketHandler.cs @@ -31,7 +31,6 @@ public override Task HandleIncoming(IPacket packet) return packet switch { KeepAlivePacket keepAlive => HandleKeepAlive(keepAlive), - BundleDelimiterPacket bundleDelimiter => HandleBundleDelimiter(bundleDelimiter), PingPacket ping => HandlePing(ping), DisconnectPacket disconnect => HandleDisconnect(disconnect), LoginPacket login => HandleLogin(login), @@ -41,7 +40,7 @@ public override Task HandleIncoming(IPacket packet) public override bool HandlesIncoming(PacketType type) { - return type is PacketType.CB_Play_KeepAlive or PacketType.CB_Play_BundleDelimiter or PacketType.CB_Play_Ping + return type is PacketType.CB_Play_KeepAlive or PacketType.CB_Play_Ping or PacketType.CB_Play_KickDisconnect or PacketType.CB_Play_Login; } @@ -50,11 +49,6 @@ private Task HandleKeepAlive(KeepAlivePacket packet) return client.SendPacket(new Serverbound.Play.KeepAlivePacket(packet.KeepAliveId)); } - private Task HandleBundleDelimiter(BundleDelimiterPacket bundleDelimiter) - { - return client.HandleBundleDelimiter(); - } - private Task HandlePing(PingPacket ping) { return client.SendPacket(new PongPacket(ping.Id));