Let Minecraft handle packet decoding during playback

We've done this ourselves only because it was easier at the beginning.
However by now (and especially with 1.20.5), Minecraft's packet decoding
has become substantially more sophisticated, to the point that it's
probably a bad idea to try to do it ourselves.

This commit gives back that responsibility to Minecraft by adding the
standard decoder (and encoder because it's required for proper
connection state management on modern versions) to our channel pipeline,
changes the full and the quick replay sender to send raw `ByteBuf`s into
the front of the channel, and moves the posititon of the full replay
sender in the channel pipeline such that it can see and modify packets
after Minecraft has decoded them.

src/main/java/com/replaymod/recording/ReplayModRecording.java

@@ -9,6 +9,7 @@ import com.replaymod.recording.handler.ConnectionEventHandler;
 import com.replaymod.recording.handler.GuiHandler;
 import com.replaymod.recording.mixin.NetworkManagerAccessor;
 import com.replaymod.recording.packet.PacketListener;
+import com.replaymod.replay.ReplayHandler;
 import io.netty.channel.Channel;
 import io.netty.util.AttributeKey;
 import net.minecraft.network.ClientConnection;
@@ -93,7 +94,7 @@ public class ReplayModRecording implements Module {
 
     public void initiateRecording(ClientConnection networkManager) {
         Channel channel = ((NetworkManagerAccessor) networkManager).getChannel();
-        if (channel.pipeline().get("ReplayModReplay_replaySender") != null) return;
+        if (channel.pipeline().get(ReplayHandler.PACKET_HANDLER_NAME) != null) return;
         //#if MC>=11400
         if (channel.hasAttr(ATTR_CHECKED)) return;
         channel.attr(ATTR_CHECKED).set(null);

src/main/java/com/replaymod/replay/FullReplaySender.java

@@ -15,8 +15,8 @@ import com.replaymod.replaystudio.protocol.PacketTypeRegistry;
 import com.replaymod.replaystudio.replay.ReplayFile;
 import de.johni0702.minecraft.gui.utils.EventRegistrations;
 import de.johni0702.minecraft.gui.versions.callbacks.PreTickCallback;
-import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
 import io.netty.channel.ChannelDuplexHandler;
 import io.netty.channel.ChannelHandler.Sharable;
 import io.netty.channel.ChannelHandlerContext;
@@ -28,9 +28,7 @@ import net.minecraft.client.gui.screen.NoticeScreen;
 import net.minecraft.client.world.ClientWorld;
 import net.minecraft.entity.Entity;
 import net.minecraft.entity.player.PlayerEntity;
-import net.minecraft.network.NetworkState;
 import net.minecraft.network.Packet;
-import net.minecraft.network.PacketByteBuf;
 import net.minecraft.network.packet.s2c.play.GameMessageS2CPacket;
 import net.minecraft.network.packet.s2c.play.CustomPayloadS2CPacket;
 import net.minecraft.network.packet.s2c.play.DisconnectS2CPacket;
@@ -140,7 +138,6 @@ import net.minecraft.network.packet.s2c.play.UnloadChunkS2CPacket;
 import net.minecraft.network.packet.s2c.play.ResourcePackSendS2CPacket;
 import net.minecraft.network.packet.s2c.play.SetCameraEntityS2CPacket;
 import net.minecraft.network.packet.s2c.play.TitleS2CPacket;
-import net.minecraft.network.NetworkSide;
 //#else
 //$$ import org.apache.commons.io.Charsets;
 //#endif
@@ -230,9 +227,9 @@ public class FullReplaySender extends ChannelDuplexHandler implements ReplaySend
     protected ReplayFile replayFile;
 
     /**
-     * The channel handler context used to send packets to minecraft.
+     * The channel used to send packets to minecraft.
      */
-    protected ChannelHandlerContext ctx;
+    protected Channel channel;
 
     /**
      * The replay input stream from which new packets are read.
@@ -303,20 +300,17 @@ public class FullReplaySender extends ChannelDuplexHandler implements ReplaySend
     /**
      * Create a new replay sender.
      * @param file The replay file
-     * @param asyncMode {@code true} for async mode, {@code false} otherwise
-     * @see #asyncMode
      */
-    public FullReplaySender(ReplayHandler replayHandler, ReplayFile file, boolean asyncMode) throws IOException {
+    public FullReplaySender(ReplayHandler replayHandler, ReplayFile file) throws IOException {
         this.replayHandler = replayHandler;
         this.replayFile = file;
-        this.asyncMode = asyncMode;
         this.replayLength = file.getMetaData().getDuration();
 
         events.register();
+    }
 
-        if (asyncMode) {
+    public void setChannel(Channel channel) {
-            new Thread(asyncSender, "replaymod-async-sender").start();
+        this.channel = channel;
-        }
     }
 
     /**
@@ -383,8 +377,8 @@ public class FullReplaySender extends ChannelDuplexHandler implements ReplaySend
         syncSender.shutdown();
         events.unregister();
         try {
-            channelInactive(ctx);
+            channel.pipeline().fireChannelInactive();
-            ctx.channel().pipeline().close();
+            channel.pipeline().close();
             FileUtils.deleteDirectory(tempResourcePackFolder);
         } catch(Exception e) {
             e.printStackTrace();
@@ -425,14 +419,9 @@ public class FullReplaySender extends ChannelDuplexHandler implements ReplaySend
             return;
         }
 
-        // When a packet is sent directly, perform no filtering
+        if (msg instanceof Packet) {
-        if(msg instanceof Packet) {
-            super.channelRead(ctx, msg);
-        }
-
-        if (msg instanceof byte[]) {
             try {
-                Packet p = deserializePacket((byte[]) msg);
+                Packet p = (Packet) msg;
 
                 if (p != null) {
                     p = processPacket(p);
@@ -478,29 +467,6 @@ public class FullReplaySender extends ChannelDuplexHandler implements ReplaySend
 
     }
 
-    private Packet deserializePacket(byte[] bytes) throws IOException, IllegalAccessException, InstantiationException {
-        ByteBuf bb = Unpooled.wrappedBuffer(bytes);
-        PacketByteBuf pb = new PacketByteBuf(bb);
-
-        int i = pb.readVarInt();
-
-        NetworkState state = asMc(registry.getState());
-        //#if MC>=12002
-        //$$ Packet p = state.getHandler(NetworkSide.CLIENTBOUND).createPacket(i, pb);
-        //#elseif MC>=11700
-        //$$ Packet p = state.getPacketHandler(NetworkSide.CLIENTBOUND, i, pb);
-        //#else
-        //#if MC>=10800
-        Packet p = state.getPacketHandler(NetworkSide.CLIENTBOUND, i);
-        //#else
-        //$$ Packet p = Packet.generatePacket(state.func_150755_b(), i);
-        //#endif
-        p.read(pb);
-        //#endif
-
-        return p;
-    }
-
     // If we do not give minecraft time to tick, there will be dead entity artifacts left in the world
     // Therefore we have to remove all loaded, dead entities manually if we are in sync mode.
     // We do this after every SpawnX packet and after the destroy entities packet.
@@ -986,13 +952,6 @@ public class FullReplaySender extends ChannelDuplexHandler implements ReplaySend
     //$$ }
     //#endif
 
-    @Override
-    @SuppressWarnings("unchecked")
-    public void channelActive(ChannelHandlerContext ctx) throws Exception {
-        this.ctx = ctx;
-        super.channelActive(ctx);
-    }
-
     @Override
     public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
         // The embedded channel's event loop will consider every thread to be in it and as such provides no
@@ -1073,9 +1032,6 @@ public class FullReplaySender extends ChannelDuplexHandler implements ReplaySend
     private Runnable asyncSender = new Runnable() {
         public void run() {
             try {
-                while (ctx == null && !terminate) {
-                    Thread.sleep(10);
-                }
                 REPLAY_LOOP:
                 while (!terminate) {
                     synchronized (FullReplaySender.this) {
@@ -1125,7 +1081,7 @@ public class FullReplaySender extends ChannelDuplexHandler implements ReplaySend
                                 }
 
                                 // Process packet
-                                channelRead(ctx, nextPacket.bytes);
+                                channel.pipeline().fireChannelRead(Unpooled.wrappedBuffer(nextPacket.bytes));
                                 nextPacket = null;
 
                                 lastTimeStamp = nextTimeStamp;
@@ -1136,7 +1092,7 @@ public class FullReplaySender extends ChannelDuplexHandler implements ReplaySend
                                 // Might be safe to do the same on older versions too, but I'd rather not poke the
                                 // monster that is Forge networking.
                                 //#if MC>=12002
-                                //$$ while (!ctx.channel().config().isAutoRead()) {
+                                //$$ while (!channel.config().isAutoRead()) {
                                 //$$     Thread.sleep(0, 100_000);
                                 //$$ }
                                 //#endif
@@ -1298,10 +1254,6 @@ public class FullReplaySender extends ChannelDuplexHandler implements ReplaySend
 
     private void doSendPacketsTill(int timestamp) {
         try {
-            while (ctx == null && !terminate) { // Make sure channel is ready
-                Thread.sleep(10);
-            }
-
             synchronized (this) {
                 if (timestamp == lastTimeStamp) { // Do nothing if we're already there
                     return;
@@ -1343,7 +1295,7 @@ public class FullReplaySender extends ChannelDuplexHandler implements ReplaySend
                         }
 
                         // Process packet
-                        channelRead(ctx, pd.bytes);
+                        channel.pipeline().fireChannelRead(Unpooled.wrappedBuffer(pd.bytes));
 
                         // MC as of 1.20.2 relies on autoRead, so it can update the connection state on the main
                         // thread before the next packet is read. As such, we need to stall if that was just
@@ -1351,7 +1303,7 @@ public class FullReplaySender extends ChannelDuplexHandler implements ReplaySend
                         // Might be safe to do the same on older versions too, but I'd rather not poke the
                         // monster that is Forge networking.
                         //#if MC>=12002
-                        //$$ while (!ctx.channel().config().isAutoRead()) {
+                        //$$ while (!channel.config().isAutoRead()) {
                         //$$     Thread.sleep(0, 100_000);
                         //$$ }
                         //#endif

src/main/java/com/replaymod/replay/QuickReplaySender.java

@@ -14,13 +14,10 @@ import de.johni0702.minecraft.gui.utils.EventRegistrations;
 import de.johni0702.minecraft.gui.versions.callbacks.PreTickCallback;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerAdapter;
-import io.netty.channel.ChannelHandlerContext;
 import net.minecraft.client.MinecraftClient;
-import net.minecraft.network.NetworkState;
-import net.minecraft.network.NetworkSide;
-import net.minecraft.network.Packet;
 import net.minecraft.network.PacketByteBuf;
 
 import javax.annotation.Nullable;
@@ -31,7 +28,6 @@ import java.util.function.Consumer;
 import com.replaymod.core.utils.WrappedTimer;
 //#endif
 
-import static com.replaymod.core.versions.MCVer.asMc;
 import static com.replaymod.core.versions.MCVer.getMinecraft;
 import static com.replaymod.core.versions.MCVer.getPacketTypeRegistry;
 import static com.replaymod.replay.ReplayModReplay.LOGGER;
@@ -49,7 +45,7 @@ public class QuickReplaySender extends ChannelHandlerAdapter implements ReplaySe
     private final ReplayModReplay mod;
     private final RandomAccessReplay replay;
     private final EventHandler eventHandler = new EventHandler();
-    private ChannelHandlerContext ctx;
+    private Channel channel;
 
     private int currentTimeStamp;
     private double replaySpeed = 1;
@@ -70,6 +66,7 @@ public class QuickReplaySender extends ChannelHandlerAdapter implements ReplaySe
 
             @Override
             protected void dispatch(com.replaymod.replaystudio.protocol.Packet packet) {
+                // Convert ReplayStudio-Netty buffer into MC-Netty buffer
                 com.github.steveice10.netty.buffer.ByteBuf byteBuf = packet.getBuf();
                 int size = byteBuf.readableBytes();
                 if (buf.length < size) {
@@ -78,40 +75,14 @@ public class QuickReplaySender extends ChannelHandlerAdapter implements ReplaySe
                 byteBuf.getBytes(byteBuf.readerIndex(), buf, 0, size);
                 ByteBuf wrappedBuf = Unpooled.wrappedBuffer(buf);
                 wrappedBuf.writerIndex(size);
-                PacketByteBuf packetByteBuf = new PacketByteBuf(wrappedBuf);
+                packet.release();
 
-                NetworkState state = asMc(packet.getRegistry().getState());
+                // Combine id + payload
-                //#if MC>=10809
+                ByteBuf bufWithId = channel.alloc().heapBuffer(2 + wrappedBuf.readableBytes());
-                Packet<?> mcPacket;
+                new PacketByteBuf(bufWithId).writeVarInt(packet.getId());
-                //#else
+                bufWithId.writeBytes(wrappedBuf);
-                //$$ Packet mcPacket;
-                //#endif
-                //#if MC>=12002
-                //$$ mcPacket = state.getHandler(NetworkSide.CLIENTBOUND).createPacket(packet.getId(), packetByteBuf);
-                //#elseif MC>=11700
-                //$$ mcPacket = state.getPacketHandler(NetworkSide.CLIENTBOUND, packet.getId(), packetByteBuf);
-                //#elseif MC>=11500
-                mcPacket = state.getPacketHandler(NetworkSide.CLIENTBOUND, packet.getId());
-                //#else
-                //$$ try {
-                //$$     mcPacket = state.getPacketHandler(NetworkSide.CLIENTBOUND, packet.getId());
-                //$$ } catch (IllegalAccessException | InstantiationException e) {
-                //$$     e.printStackTrace();
-                //$$     return;
-                //$$ }
-                //#endif
-                if (mcPacket != null) {
-                    //#if MC<11700
-                    try {
-                        mcPacket.read(packetByteBuf);
-                    } catch (IOException e) {
-                        e.printStackTrace();
-                        return;
-                    }
-                    //#endif
 
-                    ctx.fireChannelRead(mcPacket);
+                channel.pipeline().fireChannelRead(bufWithId);
-                }
             }
         };
     }
@@ -124,9 +95,8 @@ public class QuickReplaySender extends ChannelHandlerAdapter implements ReplaySe
         eventHandler.unregister();
     }
 
-    @Override
+    public void setChannel(Channel channel) {
-    public void handlerAdded(ChannelHandlerContext ctx) {
+        this.channel = channel;
-        this.ctx = ctx;
     }
 
     public ListenableFuture<Void> getInitializationPromise() {

src/main/java/com/replaymod/replay/ReplayHandler.java

@@ -33,7 +33,9 @@ import net.minecraft.client.MinecraftClient;
 import net.minecraft.client.gui.screen.DownloadingTerrainScreen;
 import net.minecraft.client.network.ClientLoginNetworkHandler;
 import net.minecraft.client.util.Window;
+import net.minecraft.network.DecoderHandler;
 import net.minecraft.network.NetworkState;
+import net.minecraft.network.PacketEncoder;
 import net.minecraft.util.crash.CrashReport;
 import net.minecraft.entity.Entity;
 import net.minecraft.entity.player.PlayerEntity;
@@ -46,13 +48,6 @@ import java.util.*;
 //$$ import net.minecraft.client.resource.server.ServerResourcePackManager;
 //#endif
 
-//#if MC>=12002
-//$$ import io.netty.channel.ChannelDuplexHandler;
-//$$ import io.netty.channel.ChannelPromise;
-//$$ import net.minecraft.network.handler.NetworkStateTransitionHandler;
-//$$ import net.minecraft.network.packet.Packet;
-//#endif
-
 //#if MC>=12000
 //$$ import com.mojang.blaze3d.systems.VertexSorter;
 //$$ import net.minecraft.client.gui.DrawContext;
@@ -129,6 +124,8 @@ import static org.lwjgl.opengl.GL11.GL_DEPTH_BUFFER_BIT;
 
 public class ReplayHandler {
 
+    public static final String PACKET_HANDLER_NAME = "ReplayModReplay_packetHandler";
+
     private static MinecraftClient mc = getMinecraft();
 
     /**
@@ -180,7 +177,7 @@ public class ReplayHandler {
 
         markers = replayFile.getMarkers().or(Collections.emptySet());
 
-        fullReplaySender = new FullReplaySender(this, replayFile, false);
+        fullReplaySender = new FullReplaySender(this, replayFile);
         //#if MC>=10800
         quickReplaySender = new QuickReplaySender(ReplayModReplay.instance, replayFile);
         //#endif
@@ -319,17 +316,25 @@ public class ReplayHandler {
         //$$ ChannelOutboundHandlerAdapter dummyHandler = new ChannelOutboundHandlerAdapter();
         //$$ channel = new EmbeddedChannel(dummyHandler);
         //$$ channel.pipeline().remove(dummyHandler);
+        //$$ channel.pipeline().removeLast();
+        //#endif
+
+        quickReplaySender.setChannel(channel);
+        fullReplaySender.setChannel(channel);
+
+        //#if MC>=12002
+        //$$ channel.pipeline().addLast("decoder", new DecoderHandler(ClientConnection.CLIENTBOUND_PROTOCOL_KEY));
+        //$$ channel.pipeline().addLast("encoder", new PacketEncoder(ClientConnection.SERVERBOUND_PROTOCOL_KEY));
+        //#else
+        channel.pipeline().addLast("decoder", new DecoderHandler(NetworkSide.CLIENTBOUND));
+        channel.pipeline().addLast("encoder", new PacketEncoder(NetworkSide.SERVERBOUND));
         //#endif
-        //#if MC>=10800
-        channel.pipeline().addLast("ReplayModReplay_quickReplaySender", quickReplaySender);
-        //#endif
-        channel.pipeline().addLast("ReplayModReplay_replaySender", fullReplaySender);
         //#if MC>=12002
-        //$$ channel.pipeline().addLast("ReplayModReplay_transition", new DummyNetworkStateTransitionHandler());
         //$$ channel.pipeline().addLast("bundler", new PacketBundler(ClientConnection.CLIENTBOUND_PROTOCOL_KEY));
         //#elseif MC>=11904
         //$$ channel.pipeline().addLast("bundler", new PacketBundler(NetworkSide.CLIENTBOUND));
         //#endif
+        channel.pipeline().addLast(PACKET_HANDLER_NAME, quickMode ? quickReplaySender : fullReplaySender);
         channel.pipeline().addLast("packet_handler", networkManager);
         channel.pipeline().fireChannelActive();
 
@@ -459,6 +464,8 @@ public class ReplayHandler {
             targetCameraPosition = null;
         }
 
+        channel.pipeline().replace(PACKET_HANDLER_NAME, PACKET_HANDLER_NAME, quickMode ? quickReplaySender : fullReplaySender);
+
         if (quickMode) {
             quickReplaySender.register();
             quickReplaySender.restart();
@@ -810,24 +817,4 @@ public class ReplayHandler {
         //$$ }
         //#endif
     }
-
-    //#if MC>=12002
-    //$$ private static class DummyNetworkStateTransitionHandler extends ChannelDuplexHandler {
-    //$$     @Override
-    //$$     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
-    //$$         if (msg instanceof Packet<?> packet) {
-    //$$             NetworkStateTransitionHandler.handle(ctx.channel().attr(ClientConnection.CLIENTBOUND_PROTOCOL_KEY), packet);
-    //$$         }
-    //$$         super.channelRead(ctx, msg);
-    //$$     }
-    //$$
-    //$$     @Override
-    //$$     public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
-    //$$         if (msg instanceof Packet<?> packet) {
-    //$$             NetworkStateTransitionHandler.handle(ctx.channel().attr(ClientConnection.SERVERBOUND_PROTOCOL_KEY), packet);
-    //$$         }
-    //$$         super.write(ctx, msg, promise);
-    //$$     }
-    //$$ }
-    //#endif
 }

versions/1.9.4/mapping.txt

@@ -25,6 +25,8 @@ net.minecraft.network.play.server.SPacketSpawnPlayer getDataManagerEntries() fun
 net.minecraft.client.multiplayer.ServerList saveSingleServer() func_147414_b()
 net.minecraft.util.SoundEvent com.replaymod.core.versions.MCVer.SoundEvent
 
+net.minecraft.network.NettyPacketDecoder net.minecraft.util.MessageDeserializer
+net.minecraft.network.NettyPacketEncoder net.minecraft.util.MessageSerializer
 net.minecraft.network.play.server.SPacketJoinGame net.minecraft.network.play.server.S01PacketJoinGame
 net.minecraft.network.play.server.SPacketChat net.minecraft.network.play.server.S02PacketChat
 net.minecraft.network.play.server.SPacketParticles net.minecraft.network.play.server.S2APacketParticles