Move all PacketLib local channel classes to Geyser

bootstrap/bungeecord/src/main/java/org/geysermc/platform/bungeecord/GeyserBungeeInjector.java

@@ -25,7 +25,6 @@
 
 package org.geysermc.platform.bungeecord;
 
-import com.github.steveice10.packetlib.io.local.LocalServerChannelWrapper;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
@@ -42,7 +41,9 @@ import net.md_5.bungee.event.EventHandler;
 import net.md_5.bungee.netty.PipelineUtils;
 import org.geysermc.connector.GeyserConnector;
 import org.geysermc.connector.bootstrap.GeyserBootstrap;
-import org.geysermc.connector.common.GeyserInjector;
+import org.geysermc.connector.common.connection.GeyserInjector;
+import org.geysermc.connector.common.connection.LocalServerChannelWrapper;
+import org.geysermc.connector.common.connection.LocalSession;
 
 import java.lang.reflect.Field;
 import java.lang.reflect.Method;
@@ -122,7 +123,7 @@ public class GeyserBungeeInjector extends GeyserInjector implements Listener {
 
         ChannelFuture channelFuture = (new ServerBootstrap()
                 .channel(LocalServerChannelWrapper.class)
-                .childHandler(new ChannelInitializer<Channel>() {
+                .childHandler(new ChannelInitializer<>() {
                     @Override
                     protected void initChannel(Channel ch) throws Exception {
                         if (proxy.getConfig().getServers() == null) {
@@ -156,6 +157,14 @@ public class GeyserBungeeInjector extends GeyserInjector implements Listener {
             this.proxy.getPluginManager().registerListener(this.plugin, this);
             this.eventRegistered = true;
         }
+
+        // Only affects Waterfall, but there is no sure way to differentiate between a proxy with this patch and a proxy without this patch
+        // Patch causing the issue: https://github.com/PaperMC/Waterfall/blob/7e6af4cef64d5d377a6ffd00a534379e6efa94cf/BungeeCord-Patches/0045-Don-t-use-a-bytebuf-for-packet-decoding.patch
+        // If native compression is enabled, then this line is tripped up if a heap buffer is sent over in such a situation
+        // as a new direct buffer is not created with that patch (HeapByteBufs throw an UnsupportedOperationException here):
+        // https://github.com/SpigotMC/BungeeCord/blob/a283aaf724d4c9a815540cd32f3aafaa72df9e05/native/src/main/java/net/md_5/bungee/jni/zlib/NativeZlib.java#L43
+        // This issue could be mitigated down the line by preventing Bungee from setting compression
+        LocalSession.createDirectByteBufAllocator();
     }
 
     @Override

bootstrap/spigot/src/main/java/org/geysermc/platform/spigot/GeyserSpigotInjector.java

@@ -25,7 +25,6 @@
 
 package org.geysermc.platform.spigot;
 
-import com.github.steveice10.packetlib.io.local.LocalServerChannelWrapper;
 import com.viaversion.viaversion.bukkit.handlers.BukkitChannelInitializer;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.*;
@@ -33,7 +32,8 @@ import io.netty.channel.local.LocalAddress;
 import io.netty.util.concurrent.DefaultThreadFactory;
 import org.bukkit.Bukkit;
 import org.geysermc.connector.bootstrap.GeyserBootstrap;
-import org.geysermc.connector.common.GeyserInjector;
+import org.geysermc.connector.common.connection.GeyserInjector;
+import org.geysermc.connector.common.connection.LocalServerChannelWrapper;
 
 import java.lang.reflect.Field;
 import java.lang.reflect.Method;

bootstrap/velocity/src/main/java/org/geysermc/platform/velocity/GeyserVelocityInjector.java

@@ -25,13 +25,13 @@
 
 package org.geysermc.platform.velocity;
 
-import com.github.steveice10.packetlib.io.local.LocalServerChannelWrapper;
 import com.velocitypowered.api.proxy.ProxyServer;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.*;
 import io.netty.channel.local.LocalAddress;
 import org.geysermc.connector.bootstrap.GeyserBootstrap;
-import org.geysermc.connector.common.GeyserInjector;
+import org.geysermc.connector.common.connection.GeyserInjector;
+import org.geysermc.connector.common.connection.LocalServerChannelWrapper;
 
 import java.lang.reflect.Field;
 import java.util.function.Supplier;

connector/pom.xml

@@ -163,15 +163,20 @@
             </exclusions>
         </dependency>
         <dependency>
-            <groupId>com.github.RednedEpic</groupId>
+            <groupId>com.github.steveice10</groupId>
-            <artifactId>PacketLib</artifactId>
+            <artifactId>packetlib</artifactId>
-            <version>9d4b476</version>
+            <version>2.1-SNAPSHOT</version>
             <scope>compile</scope>
             <exclusions>
                 <exclusion>
                     <groupId>io.netty</groupId>
                     <artifactId>netty-all</artifactId>
                 </exclusion>
+                <exclusion>
+                    <!-- This is still experimental - additionally, it could only really benefit standalone -->
+                    <groupId>io.netty.incubator</groupId>
+                    <artifactId>netty-incubator-transport-native-io_uring</artifactId>
+                </exclusion>
             </exclusions>
         </dependency>
         <dependency>

connector/src/main/java/org/geysermc/connector/GeyserConnector.java

@@ -29,6 +29,7 @@ import com.fasterxml.jackson.core.JsonParser;
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.github.steveice10.mc.protocol.MinecraftConstants;
+import com.github.steveice10.packetlib.tcp.TcpSession;
 import com.nukkitx.network.raknet.RakNetConstants;
 import com.nukkitx.network.util.EventLoops;
 import com.nukkitx.protocol.bedrock.BedrockServer;
@@ -46,12 +47,12 @@ import org.geysermc.connector.configuration.GeyserConfiguration;
 import org.geysermc.connector.metrics.Metrics;
 import org.geysermc.connector.network.ConnectorServerEventHandler;
 import org.geysermc.connector.network.session.GeyserSession;
-import org.geysermc.connector.network.translators.chat.MessageTranslator;
-import org.geysermc.connector.registry.BlockRegistries;
-import org.geysermc.connector.registry.Registries;
 import org.geysermc.connector.network.translators.PacketTranslatorRegistry;
+import org.geysermc.connector.network.translators.chat.MessageTranslator;
 import org.geysermc.connector.network.translators.item.ItemTranslator;
 import org.geysermc.connector.network.translators.world.WorldManager;
+import org.geysermc.connector.registry.BlockRegistries;
+import org.geysermc.connector.registry.Registries;
 import org.geysermc.connector.scoreboard.ScoreboardUpdater;
 import org.geysermc.connector.skin.FloodgateSkinUploader;
 import org.geysermc.connector.utils.*;
@@ -70,8 +71,12 @@ import java.net.InetSocketAddress;
 import java.net.UnknownHostException;
 import java.security.Key;
 import java.text.DecimalFormat;
-import java.util.*;
+import java.util.HashMap;
-import java.util.concurrent.*;
+import java.util.Map;
+import java.util.Properties;
+import java.util.UUID;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -191,6 +196,9 @@ public class GeyserConnector {
             }
         }
 
+        // Ensure that PacketLib does not create an event loop for handling packets; we'll do that ourselves
+        TcpSession.USE_EVENT_LOOP_FOR_PACKETS = false;
+
         TimeSyncer timeSyncer = null;
         if (config.getRemote().getAuthType() == AuthType.FLOODGATE) {
             timeSyncer = new TimeSyncer(Constants.NTP_SERVER);

connector/src/main/java/org/geysermc/connector/common/connection/ChannelWrapper.java

@@ -0,0 +1,265 @@
+/*
+ * Copyright (c) 2019-2021 GeyserMC. http://geysermc.org
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ *
+ * @author GeyserMC
+ * @link https://github.com/GeyserMC/Geyser
+ */
+
+package org.geysermc.connector.common.connection;
+
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.channel.*;
+import io.netty.util.Attribute;
+import io.netty.util.AttributeKey;
+
+import java.net.SocketAddress;
+
+public class ChannelWrapper implements Channel {
+    protected final Channel source;
+    private volatile SocketAddress remoteAddress;
+
+    public ChannelWrapper(Channel channel) {
+        this.source = channel;
+    }
+
+    @Override
+    public SocketAddress localAddress() {
+        return source.localAddress();
+    }
+
+    @Override
+    public SocketAddress remoteAddress() {
+        if (remoteAddress == null) {
+            return source.remoteAddress();
+        }
+        return remoteAddress;
+    }
+
+    public void remoteAddress(SocketAddress socketAddress) {
+        remoteAddress = socketAddress;
+    }
+
+    @Override
+    public ChannelId id() {
+        return source.id();
+    }
+
+    @Override
+    public EventLoop eventLoop() {
+        return source.eventLoop();
+    }
+
+    @Override
+    public Channel parent() {
+        return source.parent();
+    }
+
+    @Override
+    public ChannelConfig config() {
+        return source.config();
+    }
+
+    @Override
+    public boolean isOpen() {
+        return source.isOpen();
+    }
+
+    @Override
+    public boolean isRegistered() {
+        return source.isRegistered();
+    }
+
+    @Override
+    public boolean isActive() {
+        return source.isActive();
+    }
+
+    @Override
+    public ChannelMetadata metadata() {
+        return source.metadata();
+    }
+
+    @Override
+    public ChannelFuture closeFuture() {
+        return source.closeFuture();
+    }
+
+    @Override
+    public boolean isWritable() {
+        return source.isWritable();
+    }
+
+    @Override
+    public long bytesBeforeUnwritable() {
+        return source.bytesBeforeUnwritable();
+    }
+
+    @Override
+    public long bytesBeforeWritable() {
+        return source.bytesBeforeWritable();
+    }
+
+    @Override
+    public Unsafe unsafe() {
+        return source.unsafe();
+    }
+
+    @Override
+    public ChannelPipeline pipeline() {
+        return source.pipeline();
+    }
+
+    @Override
+    public ByteBufAllocator alloc() {
+        return source.alloc();
+    }
+
+    @Override
+    public ChannelFuture bind(SocketAddress socketAddress) {
+        return source.bind(socketAddress);
+    }
+
+    @Override
+    public ChannelFuture connect(SocketAddress socketAddress) {
+        return source.connect(socketAddress);
+    }
+
+    @Override
+    public ChannelFuture connect(SocketAddress socketAddress, SocketAddress socketAddress1) {
+        return source.connect(socketAddress, socketAddress1);
+    }
+
+    @Override
+    public ChannelFuture disconnect() {
+        return source.disconnect();
+    }
+
+    @Override
+    public ChannelFuture close() {
+        return source.disconnect();
+    }
+
+    @Override
+    public ChannelFuture deregister() {
+        return source.deregister();
+    }
+
+    @Override
+    public ChannelFuture bind(SocketAddress socketAddress, ChannelPromise channelPromise) {
+        return source.bind(socketAddress, channelPromise);
+    }
+
+    @Override
+    public ChannelFuture connect(SocketAddress socketAddress, ChannelPromise channelPromise) {
+        return source.connect(socketAddress, channelPromise);
+    }
+
+    @Override
+    public ChannelFuture connect(SocketAddress socketAddress, SocketAddress socketAddress1, ChannelPromise channelPromise) {
+        return source.connect(socketAddress, socketAddress1, channelPromise);
+    }
+
+    @Override
+    public ChannelFuture disconnect(ChannelPromise channelPromise) {
+        return source.disconnect(channelPromise);
+    }
+
+    @Override
+    public ChannelFuture close(ChannelPromise channelPromise) {
+        return source.close(channelPromise);
+    }
+
+    @Override
+    public ChannelFuture deregister(ChannelPromise channelPromise) {
+        return source.deregister(channelPromise);
+    }
+
+    @Override
+    public Channel read() {
+        source.read();
+        return this;
+    }
+
+    @Override
+    public ChannelFuture write(Object o) {
+        return source.write(o);
+    }
+
+    @Override
+    public ChannelFuture write(Object o, ChannelPromise channelPromise) {
+        return source.write(o, channelPromise);
+    }
+
+    @Override
+    public Channel flush() {
+        return source.flush();
+    }
+
+    @Override
+    public ChannelFuture writeAndFlush(Object o, ChannelPromise channelPromise) {
+        return source.writeAndFlush(o, channelPromise);
+    }
+
+    @Override
+    public ChannelFuture writeAndFlush(Object o) {
+        return source.writeAndFlush(o);
+    }
+
+    @Override
+    public ChannelPromise newPromise() {
+        return source.newPromise();
+    }
+
+    @Override
+    public ChannelProgressivePromise newProgressivePromise() {
+        return source.newProgressivePromise();
+    }
+
+    @Override
+    public ChannelFuture newSucceededFuture() {
+        return source.newSucceededFuture();
+    }
+
+    @Override
+    public ChannelFuture newFailedFuture(Throwable throwable) {
+        return source.newFailedFuture(throwable);
+    }
+
+    @Override
+    public ChannelPromise voidPromise() {
+        return source.voidPromise();
+    }
+
+    @Override
+    public <T> Attribute<T> attr(AttributeKey<T> attributeKey) {
+        return source.attr(attributeKey);
+    }
+
+    @Override
+    public <T> boolean hasAttr(AttributeKey<T> attributeKey) {
+        return source.hasAttr(attributeKey);
+    }
+
+    @Override
+    public int compareTo(Channel o) {
+        return source.compareTo(o);
+    }
+}

connector/src/main/java/org/geysermc/connector/common/connection/DefaultChannelPipelinePublic.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright (c) 2019-2021 GeyserMC. http://geysermc.org
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ *
+ * @author GeyserMC
+ * @link https://github.com/GeyserMC/Geyser
+ */
+
+package org.geysermc.connector.common.connection;
+
+import io.netty.channel.Channel;
+import io.netty.channel.DefaultChannelPipeline;
+
+/**
+ * Exists solely to make DefaultChannelPipeline's protected constructor public
+ */
+public class DefaultChannelPipelinePublic extends DefaultChannelPipeline {
+    public DefaultChannelPipelinePublic(Channel channel) {
+        super(channel);
+    }
+}

connector/src/main/java/org/geysermc/connector/common/connection/GeyserInjector.java

@@ -23,7 +23,7 @@
  * @link https://github.com/GeyserMC/Geyser
  */
 
-package org.geysermc.connector.common;
+package org.geysermc.connector.common.connection;
 
 import io.netty.channel.ChannelFuture;
 import lombok.Getter;
@@ -47,7 +47,6 @@ public abstract class GeyserInjector {
     protected SocketAddress serverSocketAddress;
 
     /**
-     *
      * @param bootstrap the bootstrap of the Geyser instance.
      */
     public void initializeLocalChannel(GeyserBootstrap bootstrap) {

connector/src/main/java/org/geysermc/connector/common/connection/LocalChannelWithRemoteAddress.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright (c) 2019-2021 GeyserMC. http://geysermc.org
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ *
+ * @author GeyserMC
+ * @link https://github.com/GeyserMC/Geyser
+ */
+
+package org.geysermc.connector.common.connection;
+
+import io.netty.channel.local.LocalChannel;
+
+import java.net.InetSocketAddress;
+
+/**
+ * Client -> server storing the spoofed remote address.
+ */
+public class LocalChannelWithRemoteAddress extends LocalChannel {
+    private InetSocketAddress spoofedAddress;
+
+    public InetSocketAddress spoofedRemoteAddress() {
+        return spoofedAddress;
+    }
+
+    public void spoofedRemoteAddress(InetSocketAddress socketAddress) {
+        this.spoofedAddress = socketAddress;
+    }
+}

connector/src/main/java/org/geysermc/connector/common/connection/LocalChannelWrapper.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright (c) 2019-2021 GeyserMC. http://geysermc.org
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ *
+ * @author GeyserMC
+ * @link https://github.com/GeyserMC/Geyser
+ */
+
+package org.geysermc.connector.common.connection;
+
+import io.netty.channel.DefaultChannelPipeline;
+import io.netty.channel.local.LocalChannel;
+import io.netty.channel.local.LocalServerChannel;
+
+import java.net.InetSocketAddress;
+
+public class LocalChannelWrapper extends LocalChannel {
+    private final ChannelWrapper wrapper;
+    /**
+     * {@link #newChannelPipeline()} is called during super, so this exists until the wrapper can be initialized.
+     */
+    private volatile ChannelWrapper tempWrapper;
+
+    public LocalChannelWrapper() {
+        wrapper = new ChannelWrapper(this);
+    }
+
+    public LocalChannelWrapper(LocalServerChannel parent, LocalChannel peer) {
+        super(parent, peer);
+        if (tempWrapper == null) {
+            this.wrapper = new ChannelWrapper(this);
+        } else {
+            this.wrapper = tempWrapper;
+        }
+        wrapper.remoteAddress(new InetSocketAddress(0));
+    }
+
+    public ChannelWrapper wrapper() {
+        return wrapper;
+    }
+
+    @Override
+    protected DefaultChannelPipeline newChannelPipeline() {
+        if (wrapper != null) {
+            return new DefaultChannelPipelinePublic(wrapper);
+        } else if (tempWrapper != null) {
+            return new DefaultChannelPipelinePublic(tempWrapper);
+        } else {
+            tempWrapper = new ChannelWrapper(this);
+            return new DefaultChannelPipelinePublic(tempWrapper);
+        }
+    }
+}

connector/src/main/java/org/geysermc/connector/common/connection/LocalServerChannelWrapper.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright (c) 2019-2021 GeyserMC. http://geysermc.org
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ *
+ * @author GeyserMC
+ * @link https://github.com/GeyserMC/Geyser
+ */
+
+package org.geysermc.connector.common.connection;
+
+import io.netty.channel.local.LocalChannel;
+import io.netty.channel.local.LocalServerChannel;
+
+/**
+ * If the incoming channel if an instance of LocalChannelWithRemoteAddress, this server creates a LocalChannelWrapper
+ * for the other end and attaches the spoofed remote address
+ */
+public class LocalServerChannelWrapper extends LocalServerChannel {
+    @Override
+    protected LocalChannel newLocalChannel(LocalChannel peer) {
+        // LocalChannel here should be an instance of LocalChannelWithRemoteAddress, which we can use to set the "remote address" on the other end
+        if (peer instanceof LocalChannelWithRemoteAddress) {
+            LocalChannelWrapper channel = new LocalChannelWrapper(this, peer);
+            channel.wrapper().remoteAddress(((LocalChannelWithRemoteAddress) peer).spoofedRemoteAddress());
+            return channel;
+        }
+        return super.newLocalChannel(peer);
+    }
+}

connector/src/main/java/org/geysermc/connector/common/connection/LocalSession.java

@@ -0,0 +1,142 @@
+/*
+ * Copyright (c) 2019-2021 GeyserMC. http://geysermc.org
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ *
+ * @author GeyserMC
+ * @link https://github.com/GeyserMC/Geyser
+ */
+
+package org.geysermc.connector.common.connection;
+
+import com.github.steveice10.packetlib.BuiltinFlags;
+import com.github.steveice10.packetlib.packet.PacketProtocol;
+import com.github.steveice10.packetlib.tcp.*;
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.channel.*;
+import io.netty.channel.unix.PreferredDirectByteBufAllocator;
+import io.netty.handler.codec.haproxy.*;
+
+import java.net.Inet4Address;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+
+/**
+ * Manages a Minecraft Java session over our LocalChannel implementations.
+ */
+public final class LocalSession extends TcpSession {
+    private static DefaultEventLoopGroup DEFAULT_EVENT_LOOP_GROUP;
+    private static PreferredDirectByteBufAllocator PREFERRED_DIRECT_BYTE_BUF_ALLOCATOR = null;
+
+    private final SocketAddress targetAddress;
+    private final String clientIp;
+
+    public LocalSession(String host, int port, SocketAddress targetAddress, String clientIp, PacketProtocol protocol) {
+        super(host, port, protocol);
+        this.targetAddress = targetAddress;
+        this.clientIp = clientIp;
+    }
+
+    @Override
+    public void connect() {
+        if (this.disconnected) {
+            throw new IllegalStateException("Session has already been disconnected.");
+        }
+
+        if (DEFAULT_EVENT_LOOP_GROUP == null) {
+            DEFAULT_EVENT_LOOP_GROUP = new DefaultEventLoopGroup();
+        }
+
+        try {
+            final Bootstrap bootstrap = new Bootstrap();
+            bootstrap.channel(LocalChannelWithRemoteAddress.class);
+            bootstrap.handler(new ChannelInitializer<LocalChannelWithRemoteAddress>() {
+                @Override
+                public void initChannel(LocalChannelWithRemoteAddress channel) {
+                    channel.spoofedRemoteAddress(new InetSocketAddress(clientIp, 0));
+                    getPacketProtocol().newClientSession(LocalSession.this);
+
+                    refreshReadTimeoutHandler(channel);
+                    refreshWriteTimeoutHandler(channel);
+
+                    ChannelPipeline pipeline = channel.pipeline();
+                    pipeline.addLast("encryption", new TcpPacketEncryptor(LocalSession.this));
+                    pipeline.addLast("sizer", new TcpPacketSizer(LocalSession.this));
+                    pipeline.addLast("codec", new TcpPacketCodec(LocalSession.this));
+                    pipeline.addLast("manager", LocalSession.this);
+
+                    addHAProxySupport(pipeline);
+                }
+            }).group(DEFAULT_EVENT_LOOP_GROUP).option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getConnectTimeout() * 1000);
+
+            if (PREFERRED_DIRECT_BYTE_BUF_ALLOCATOR != null) {
+                bootstrap.option(ChannelOption.ALLOCATOR, PREFERRED_DIRECT_BYTE_BUF_ALLOCATOR);
+            }
+
+            bootstrap.remoteAddress(targetAddress);
+
+            bootstrap.connect().addListener((future) -> {
+                if (!future.isSuccess()) {
+                    exceptionCaught(null, future.cause());
+                }
+            });
+        } catch(Throwable t) {
+            exceptionCaught(null, t);
+        }
+    }
+
+    // TODO duplicate code
+    private void addHAProxySupport(ChannelPipeline pipeline) {
+        InetSocketAddress clientAddress = getFlag(BuiltinFlags.CLIENT_PROXIED_ADDRESS);
+        if (getFlag(BuiltinFlags.ENABLE_CLIENT_PROXY_PROTOCOL, false) && clientAddress != null) {
+            pipeline.addFirst("proxy-protocol-packet-sender", new ChannelInboundHandlerAdapter() {
+                @Override
+                public void channelActive(ChannelHandlerContext ctx) throws Exception {
+                    HAProxyProxiedProtocol proxiedProtocol = clientAddress.getAddress() instanceof Inet4Address ? HAProxyProxiedProtocol.TCP4 : HAProxyProxiedProtocol.TCP6;
+                    InetSocketAddress remoteAddress;
+                    if (ctx.channel().remoteAddress() instanceof InetSocketAddress) {
+                        remoteAddress = (InetSocketAddress) ctx.channel().remoteAddress();
+                    } else {
+                        remoteAddress = new InetSocketAddress(host, port);
+                    }
+                    ctx.channel().writeAndFlush(new HAProxyMessage(
+                            HAProxyProtocolVersion.V2, HAProxyCommand.PROXY, proxiedProtocol,
+                            clientAddress.getAddress().getHostAddress(), remoteAddress.getAddress().getHostAddress(),
+                            clientAddress.getPort(), remoteAddress.getPort()
+                    ));
+                    ctx.pipeline().remove(this);
+                    ctx.pipeline().remove("proxy-protocol-encoder");
+                    super.channelActive(ctx);
+                }
+            });
+            pipeline.addFirst("proxy-protocol-encoder", HAProxyMessageEncoder.INSTANCE);
+        }
+    }
+
+    /**
+     * Should only be called when direct ByteBufs should be preferred. At this moment, this should only be called on BungeeCord.
+     */
+    public static void createDirectByteBufAllocator() {
+        if (PREFERRED_DIRECT_BYTE_BUF_ALLOCATOR == null) {
+            PREFERRED_DIRECT_BYTE_BUF_ALLOCATOR = new PreferredDirectByteBufAllocator();
+            PREFERRED_DIRECT_BYTE_BUF_ALLOCATOR.updateAllocator(ByteBufAllocator.DEFAULT);
+        }
+    }
+}

connector/src/main/java/org/geysermc/connector/network/session/GeyserSession.java

@@ -50,6 +50,7 @@ import com.github.steveice10.packetlib.BuiltinFlags;
 import com.github.steveice10.packetlib.event.session.*;
 import com.github.steveice10.packetlib.packet.Packet;
 import com.github.steveice10.packetlib.tcp.TcpClientSession;
+import com.github.steveice10.packetlib.tcp.TcpSession;
 import com.nukkitx.math.GenericMath;
 import com.nukkitx.math.vector.*;
 import com.nukkitx.protocol.bedrock.BedrockPacket;
@@ -76,6 +77,7 @@ import org.geysermc.common.PlatformType;
 import org.geysermc.connector.GeyserConnector;
 import org.geysermc.connector.command.CommandSender;
 import org.geysermc.connector.common.AuthType;
+import org.geysermc.connector.common.connection.LocalSession;
 import org.geysermc.connector.configuration.EmoteOffhandWorkaroundOption;
 import org.geysermc.connector.entity.Entity;
 import org.geysermc.connector.entity.ItemFrameEntity;
@@ -123,7 +125,7 @@ public class GeyserSession implements CommandSender {
      * If this is manually called, ensure that any exceptions are properly handled.
      */
     private final EventLoop eventLoop;
-    private TcpClientSession downstream;
+    private TcpSession downstream;
     @Setter
     private AuthData authData;
     @Setter
@@ -715,8 +717,14 @@ public class GeyserSession implements CommandSender {
         // Start ticking
         tickThread = eventLoop.scheduleAtFixedRate(this::tick, 50, 50, TimeUnit.MILLISECONDS);
 
-        downstream = new TcpClientSession(this.remoteAddress, this.remotePort, protocol);
+        if (connector.getBootstrap().getSocketAddress() != null) {
+            // We're going to connect through the JVM and not through TCP
+            downstream = new LocalSession(this.remoteAddress, this.remotePort,
+                    connector.getBootstrap().getSocketAddress(), upstream.getAddress().getAddress().getHostAddress(), this.protocol);
+        } else {
+            downstream = new TcpClientSession(this.remoteAddress, this.remotePort, this.protocol);
             disableSrvResolving();
+        }
         if (connector.getConfig().getRemote().isUseProxyProtocol()) {
             downstream.setFlag(BuiltinFlags.ENABLE_CLIENT_PROXY_PROTOCOL, true);
             downstream.setFlag(BuiltinFlags.CLIENT_PROXIED_ADDRESS, upstream.getAddress());
@@ -793,7 +801,7 @@ public class GeyserSession implements CommandSender {
                 loggingIn = false;
                 loggedIn = true;
 
-                if (downstream.isInternallyConnecting()) {
+                if (downstream instanceof LocalSession) {
                     // Connected directly to the server
                     connector.getLogger().info(LanguageUtils.getLocaleStringLog("geyser.network.remote.connect_internal",
                             authData.getName(), protocol.getProfile().getName()));
@@ -860,7 +868,7 @@ public class GeyserSession implements CommandSender {
                     disconnectMessage = MessageTranslator.convertMessageLenient(event.getReason());
                 }
 
-                if (downstream != null && downstream.isInternallyConnecting()) {
+                if (downstream instanceof LocalSession) {
                     connector.getLogger().info(LanguageUtils.getLocaleStringLog("geyser.network.remote.disconnect_internal", authData.getName(), disconnectMessage));
                 } else {
                     connector.getLogger().info(LanguageUtils.getLocaleStringLog("geyser.network.remote.disconnect", authData.getName(), remoteAddress, disconnectMessage));
@@ -890,27 +898,9 @@ public class GeyserSession implements CommandSender {
         if (!daylightCycle) {
             setDaylightCycle(true);
         }
-        boolean internalConnect = false;
-        if (connector.getBootstrap().getSocketAddress() != null) {
-            try {
-                // Only affects Waterfall, but there is no sure way to differentiate between a proxy with this patch and a proxy without this patch
-                // Patch causing the issue: https://github.com/PaperMC/Waterfall/blob/7e6af4cef64d5d377a6ffd00a534379e6efa94cf/BungeeCord-Patches/0045-Don-t-use-a-bytebuf-for-packet-decoding.patch
-                // If native compression is enabled, then this line is tripped up if a heap buffer is sent over in such a situation
-                // as a new direct buffer is not created with that patch (HeapByteBufs throw an UnsupportedOperationException here):
-                // https://github.com/SpigotMC/BungeeCord/blob/a283aaf724d4c9a815540cd32f3aafaa72df9e05/native/src/main/java/net/md_5/bungee/jni/zlib/NativeZlib.java#L43
-                // This issue could be mitigated down the line by preventing Bungee from setting compression
-                downstream.setFlag(BuiltinFlags.USE_ONLY_DIRECT_BUFFERS, connector.getPlatformType() == PlatformType.BUNGEECORD);
 
-                downstream.connectInternal(connector.getBootstrap().getSocketAddress(), upstream.getAddress().getAddress().getHostAddress());
-                internalConnect = true;
-            } catch (Exception e) {
-                e.printStackTrace();
-            }
-        }
-        if (!internalConnect) {
         downstream.connect();
     }
-    }
 
     public void disconnect(String reason) {
         if (!closed) {