minecraft-src/net/minecraft/network/Connection.java

package net.minecraft.network;

import com.google.common.base.Suppliers;
import com.google.common.collect.Queues;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.mojang.logging.LogUtils;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelException;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelOutboundHandler;
import io.netty.channel.ChannelOutboundHandlerAdapter;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.ChannelPromise;
import io.netty.channel.DefaultEventLoopGroup;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.local.LocalChannel;
import io.netty.channel.local.LocalServerChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.flow.FlowControlHandler;
import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.handler.timeout.TimeoutException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.channels.ClosedChannelException;
import java.util.Objects;
import java.util.Queue;
import java.util.concurrent.RejectedExecutionException;
import java.util.function.Consumer;
import java.util.function.Supplier;
import javax.crypto.Cipher;
import net.minecraft.SharedConstants;
import net.minecraft.Util;
import net.minecraft.network.chat.Component;
import net.minecraft.network.protocol.BundlerInfo;
import net.minecraft.network.protocol.Packet;
import net.minecraft.network.protocol.PacketFlow;
import net.minecraft.network.protocol.common.ClientboundDisconnectPacket;
import net.minecraft.network.protocol.handshake.ClientIntent;
import net.minecraft.network.protocol.handshake.ClientIntentionPacket;
import net.minecraft.network.protocol.handshake.HandshakeProtocols;
import net.minecraft.network.protocol.handshake.ServerHandshakePacketListener;
import net.minecraft.network.protocol.login.ClientLoginPacketListener;
import net.minecraft.network.protocol.login.ClientboundLoginDisconnectPacket;
import net.minecraft.network.protocol.login.LoginProtocols;
import net.minecraft.network.protocol.status.ClientStatusPacketListener;
import net.minecraft.network.protocol.status.StatusProtocols;
import net.minecraft.server.RunningOnDifferentThreadException;
import net.minecraft.util.Mth;
import net.minecraft.util.debugchart.LocalSampleLogger;
import org.apache.commons.lang3.Validate;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.Marker;
import org.slf4j.MarkerFactory;

public class Connection extends SimpleChannelInboundHandler<Packet<?>> {
	private static final float AVERAGE_PACKETS_SMOOTHING = 0.75F;
	private static final Logger LOGGER = LogUtils.getLogger();
	public static final Marker ROOT_MARKER = MarkerFactory.getMarker("NETWORK");
	public static final Marker PACKET_MARKER = Util.make(MarkerFactory.getMarker("NETWORK_PACKETS"), marker -> marker.add(ROOT_MARKER));
	public static final Marker PACKET_RECEIVED_MARKER = Util.make(MarkerFactory.getMarker("PACKET_RECEIVED"), marker -> marker.add(PACKET_MARKER));
	public static final Marker PACKET_SENT_MARKER = Util.make(MarkerFactory.getMarker("PACKET_SENT"), marker -> marker.add(PACKET_MARKER));
	public static final Supplier<NioEventLoopGroup> NETWORK_WORKER_GROUP = Suppliers.memoize(
		() -> new NioEventLoopGroup(0, new ThreadFactoryBuilder().setNameFormat("Netty Client IO #%d").setDaemon(true).build())
	);
	public static final Supplier<EpollEventLoopGroup> NETWORK_EPOLL_WORKER_GROUP = Suppliers.memoize(
		() -> new EpollEventLoopGroup(0, new ThreadFactoryBuilder().setNameFormat("Netty Epoll Client IO #%d").setDaemon(true).build())
	);
	public static final Supplier<DefaultEventLoopGroup> LOCAL_WORKER_GROUP = Suppliers.memoize(
		() -> new DefaultEventLoopGroup(0, new ThreadFactoryBuilder().setNameFormat("Netty Local Client IO #%d").setDaemon(true).build())
	);
	private static final ProtocolInfo<ServerHandshakePacketListener> INITIAL_PROTOCOL = HandshakeProtocols.SERVERBOUND;
	private final PacketFlow receiving;
	private volatile boolean sendLoginDisconnect = true;
	private final Queue<Consumer<Connection>> pendingActions = Queues.<Consumer<Connection>>newConcurrentLinkedQueue();
	/**
	 * The active channel
	 */
	private Channel channel;
	/**
	 * The address of the remote party
	 */
	private SocketAddress address;
	@Nullable
	private volatile PacketListener disconnectListener;
	/**
	 * The PacketListener instance responsible for processing received packets
	 */
	@Nullable
	private volatile PacketListener packetListener;
	@Nullable
	private DisconnectionDetails disconnectionDetails;
	private boolean encrypted;
	private boolean disconnectionHandled;
	private int receivedPackets;
	private int sentPackets;
	private float averageReceivedPackets;
	private float averageSentPackets;
	private int tickCount;
	private boolean handlingFault;
	@Nullable
	private volatile DisconnectionDetails delayedDisconnect;
	@Nullable
	BandwidthDebugMonitor bandwidthDebugMonitor;

	public Connection(PacketFlow receiving) {
		this.receiving = receiving;
	}

	@Override
	public void channelActive(ChannelHandlerContext channelHandlerContext) throws Exception {
		super.channelActive(channelHandlerContext);
		this.channel = channelHandlerContext.channel();
		this.address = this.channel.remoteAddress();
		if (this.delayedDisconnect != null) {
			this.disconnect(this.delayedDisconnect);
		}
	}

	@Override
	public void channelInactive(ChannelHandlerContext channelHandlerContext) {
		this.disconnect(Component.translatable("disconnect.endOfStream"));
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext channelHandlerContext, Throwable throwable) {
		if (throwable instanceof SkipPacketException) {
			LOGGER.debug("Skipping packet due to errors", throwable.getCause());
		} else {
			boolean bl = !this.handlingFault;
			this.handlingFault = true;
			if (this.channel.isOpen()) {
				if (throwable instanceof TimeoutException) {
					LOGGER.debug("Timeout", throwable);
					this.disconnect(Component.translatable("disconnect.timeout"));
				} else {
					Component component = Component.translatable("disconnect.genericReason", "Internal Exception: " + throwable);
					PacketListener packetListener = this.packetListener;
					DisconnectionDetails disconnectionDetails;
					if (packetListener != null) {
						disconnectionDetails = packetListener.createDisconnectionInfo(component, throwable);
					} else {
						disconnectionDetails = new DisconnectionDetails(component);
					}

					if (bl) {
						LOGGER.debug("Failed to sent packet", throwable);
						if (this.getSending() == PacketFlow.CLIENTBOUND) {
							Packet<?> packet = (Packet<?>)(this.sendLoginDisconnect ? new ClientboundLoginDisconnectPacket(component) : new ClientboundDisconnectPacket(component));
							this.send(packet, PacketSendListener.thenRun(() -> this.disconnect(disconnectionDetails)));
						} else {
							this.disconnect(disconnectionDetails);
						}

						this.setReadOnly();
					} else {
						LOGGER.debug("Double fault", throwable);
						this.disconnect(disconnectionDetails);
					}
				}
			}
		}
	}

	protected void channelRead0(ChannelHandlerContext context, Packet<?> packet) {
		if (this.channel.isOpen()) {
			PacketListener packetListener = this.packetListener;
			if (packetListener == null) {
				throw new IllegalStateException("Received a packet before the packet listener was initialized");
			} else {
				if (packetListener.shouldHandleMessage(packet)) {
					try {
						genericsFtw(packet, packetListener);
					} catch (RunningOnDifferentThreadException var5) {
					} catch (RejectedExecutionException var6) {
						this.disconnect(Component.translatable("multiplayer.disconnect.server_shutdown"));
					} catch (ClassCastException var7) {
						LOGGER.error("Received {} that couldn't be processed", packet.getClass(), var7);
						this.disconnect(Component.translatable("multiplayer.disconnect.invalid_packet"));
					}

					this.receivedPackets++;
				}
			}
		}
	}

	private static <T extends PacketListener> void genericsFtw(Packet<T> packet, PacketListener listener) {
		packet.handle((T)listener);
	}

	private void validateListener(ProtocolInfo<?> protocolInfo, PacketListener packetListener) {
		Validate.notNull(packetListener, "packetListener");
		PacketFlow packetFlow = packetListener.flow();
		if (packetFlow != this.receiving) {
			throw new IllegalStateException("Trying to set listener for wrong side: connection is " + this.receiving + ", but listener is " + packetFlow);
		} else {
			ConnectionProtocol connectionProtocol = packetListener.protocol();
			if (protocolInfo.id() != connectionProtocol) {
				throw new IllegalStateException("Listener protocol (" + connectionProtocol + ") does not match requested one " + protocolInfo);
			}
		}
	}

	private static void syncAfterConfigurationChange(ChannelFuture future) {
		try {
			future.syncUninterruptibly();
		} catch (Exception var2) {
			if (var2 instanceof ClosedChannelException) {
				LOGGER.info("Connection closed during protocol change");
			} else {
				throw var2;
			}
		}
	}

	public <T extends PacketListener> void setupInboundProtocol(ProtocolInfo<T> protocolInfo, T packetInfo) {
		this.validateListener(protocolInfo, packetInfo);
		if (protocolInfo.flow() != this.getReceiving()) {
			throw new IllegalStateException("Invalid inbound protocol: " + protocolInfo.id());
		} else {
			this.packetListener = packetInfo;
			this.disconnectListener = null;
			UnconfiguredPipelineHandler.InboundConfigurationTask inboundConfigurationTask = UnconfiguredPipelineHandler.setupInboundProtocol(protocolInfo);
			BundlerInfo bundlerInfo = protocolInfo.bundlerInfo();
			if (bundlerInfo != null) {
				PacketBundlePacker packetBundlePacker = new PacketBundlePacker(bundlerInfo);
				inboundConfigurationTask = inboundConfigurationTask.andThen(
					channelHandlerContext -> channelHandlerContext.pipeline().addAfter("decoder", "bundler", packetBundlePacker)
				);
			}

			syncAfterConfigurationChange(this.channel.writeAndFlush(inboundConfigurationTask));
		}
	}

	public void setupOutboundProtocol(ProtocolInfo<?> protocolInfo) {
		if (protocolInfo.flow() != this.getSending()) {
			throw new IllegalStateException("Invalid outbound protocol: " + protocolInfo.id());
		} else {
			UnconfiguredPipelineHandler.OutboundConfigurationTask outboundConfigurationTask = UnconfiguredPipelineHandler.setupOutboundProtocol(protocolInfo);
			BundlerInfo bundlerInfo = protocolInfo.bundlerInfo();
			if (bundlerInfo != null) {
				PacketBundleUnpacker packetBundleUnpacker = new PacketBundleUnpacker(bundlerInfo);
				outboundConfigurationTask = outboundConfigurationTask.andThen(
					channelHandlerContext -> channelHandlerContext.pipeline().addAfter("encoder", "unbundler", packetBundleUnpacker)
				);
			}

			boolean bl = protocolInfo.id() == ConnectionProtocol.LOGIN;
			syncAfterConfigurationChange(this.channel.writeAndFlush(outboundConfigurationTask.andThen(channelHandlerContext -> this.sendLoginDisconnect = bl)));
		}
	}

	public void setListenerForServerboundHandshake(PacketListener packetListener) {
		if (this.packetListener != null) {
			throw new IllegalStateException("Listener already set");
		} else if (this.receiving == PacketFlow.SERVERBOUND && packetListener.flow() == PacketFlow.SERVERBOUND && packetListener.protocol() == INITIAL_PROTOCOL.id()) {
			this.packetListener = packetListener;
		} else {
			throw new IllegalStateException("Invalid initial listener");
		}
	}

	public void initiateServerboundStatusConnection(String hostName, int port, ClientStatusPacketListener packetListener) {
		this.initiateServerboundConnection(hostName, port, StatusProtocols.SERVERBOUND, StatusProtocols.CLIENTBOUND, packetListener, ClientIntent.STATUS);
	}

	public void initiateServerboundPlayConnection(String hostName, int port, ClientLoginPacketListener packetListener) {
		this.initiateServerboundConnection(hostName, port, LoginProtocols.SERVERBOUND, LoginProtocols.CLIENTBOUND, packetListener, ClientIntent.LOGIN);
	}

	public <S extends ServerboundPacketListener, C extends ClientboundPacketListener> void initiateServerboundPlayConnection(
		String hostName, int port, ProtocolInfo<S> serverboundProtocol, ProtocolInfo<C> clientbountProtocol, C packetListener, boolean isTransfer
	) {
		this.initiateServerboundConnection(
			hostName, port, serverboundProtocol, clientbountProtocol, packetListener, isTransfer ? ClientIntent.TRANSFER : ClientIntent.LOGIN
		);
	}

	private <S extends ServerboundPacketListener, C extends ClientboundPacketListener> void initiateServerboundConnection(
		String hostName, int port, ProtocolInfo<S> serverboundProtocol, ProtocolInfo<C> clientboundProtocol, C packetListener, ClientIntent intention
	) {
		if (serverboundProtocol.id() != clientboundProtocol.id()) {
			throw new IllegalStateException("Mismatched initial protocols");
		} else {
			this.disconnectListener = packetListener;
			this.runOnceConnected(connection -> {
				this.setupInboundProtocol(clientboundProtocol, packetListener);
				connection.sendPacket(new ClientIntentionPacket(SharedConstants.getCurrentVersion().getProtocolVersion(), hostName, port, intention), null, true);
				this.setupOutboundProtocol(serverboundProtocol);
			});
		}
	}

	public void send(Packet<?> packet) {
		this.send(packet, null);
	}

	public void send(Packet<?> packet, @Nullable PacketSendListener sendListener) {
		this.send(packet, sendListener, true);
	}

	public void send(Packet<?> packet, @Nullable PacketSendListener listener, boolean flush) {
		if (this.isConnected()) {
			this.flushQueue();
			this.sendPacket(packet, listener, flush);
		} else {
			this.pendingActions.add((Consumer)connection -> connection.sendPacket(packet, listener, flush));
		}
	}

	public void runOnceConnected(Consumer<Connection> action) {
		if (this.isConnected()) {
			this.flushQueue();
			action.accept(this);
		} else {
			this.pendingActions.add(action);
		}
	}

	private void sendPacket(Packet<?> packet, @Nullable PacketSendListener sendListener, boolean flush) {
		this.sentPackets++;
		if (this.channel.eventLoop().inEventLoop()) {
			this.doSendPacket(packet, sendListener, flush);
		} else {
			this.channel.eventLoop().execute(() -> this.doSendPacket(packet, sendListener, flush));
		}
	}

	private void doSendPacket(Packet<?> packet, @Nullable PacketSendListener sendListener, boolean flush) {
		ChannelFuture channelFuture = flush ? this.channel.writeAndFlush(packet) : this.channel.write(packet);
		if (sendListener != null) {
			channelFuture.addListener(future -> {
				if (future.isSuccess()) {
					sendListener.onSuccess();
				} else {
					Packet<?> packetx = sendListener.onFailure();
					if (packetx != null) {
						ChannelFuture channelFuturex = this.channel.writeAndFlush(packetx);
						channelFuturex.addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
					}
				}
			});
		}

		channelFuture.addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
	}

	public void flushChannel() {
		if (this.isConnected()) {
			this.flush();
		} else {
			this.pendingActions.add(Connection::flush);
		}
	}

	private void flush() {
		if (this.channel.eventLoop().inEventLoop()) {
			this.channel.flush();
		} else {
			this.channel.eventLoop().execute(() -> this.channel.flush());
		}
	}

	/**
	 * Will iterate through the outboundPacketQueue and dispatch all Packets
	 */
	private void flushQueue() {
		if (this.channel != null && this.channel.isOpen()) {
			synchronized (this.pendingActions) {
				Consumer<Connection> consumer;
				while ((consumer = (Consumer<Connection>)this.pendingActions.poll()) != null) {
					consumer.accept(this);
				}
			}
		}
	}

	/**
	 * Checks timeouts and processes all packets received
	 */
	public void tick() {
		this.flushQueue();
		if (this.packetListener instanceof TickablePacketListener tickablePacketListener) {
			tickablePacketListener.tick();
		}

		if (!this.isConnected() && !this.disconnectionHandled) {
			this.handleDisconnection();
		}

		if (this.channel != null) {
			this.channel.flush();
		}

		if (this.tickCount++ % 20 == 0) {
			this.tickSecond();
		}

		if (this.bandwidthDebugMonitor != null) {
			this.bandwidthDebugMonitor.tick();
		}
	}

	protected void tickSecond() {
		this.averageSentPackets = Mth.lerp(0.75F, (float)this.sentPackets, this.averageSentPackets);
		this.averageReceivedPackets = Mth.lerp(0.75F, (float)this.receivedPackets, this.averageReceivedPackets);
		this.sentPackets = 0;
		this.receivedPackets = 0;
	}

	/**
	 * Returns the socket address of the remote side. Server-only.
	 */
	public SocketAddress getRemoteAddress() {
		return this.address;
	}

	public String getLoggableAddress(boolean logIps) {
		if (this.address == null) {
			return "local";
		} else {
			return logIps ? this.address.toString() : "IP hidden";
		}
	}

	/**
	 * Closes the channel with a given reason. The reason is stored for later and will be used for informational purposes (info log on server,
	 * disconnection screen on the client). This method is also called on the client when the server requests disconnection via
	 * {@code ClientboundDisconnectPacket}.
	 * 
	 * Closing the channel this way does not send any disconnection packets, it simply terminates the underlying netty channel.
	 */
	public void disconnect(Component message) {
		this.disconnect(new DisconnectionDetails(message));
	}

	public void disconnect(DisconnectionDetails disconnectionDetails) {
		if (this.channel == null) {
			this.delayedDisconnect = disconnectionDetails;
		}

		if (this.isConnected()) {
			this.channel.close().awaitUninterruptibly();
			this.disconnectionDetails = disconnectionDetails;
		}
	}

	/**
	 * True if this {@code Connection} uses a memory connection (single player game). False may imply both an active TCP connection or simply no active connection at all
	 */
	public boolean isMemoryConnection() {
		return this.channel instanceof LocalChannel || this.channel instanceof LocalServerChannel;
	}

	/**
	 * The receiving packet direction (i.e. SERVERBOUND on the server and CLIENTBOUND on the client).
	 */
	public PacketFlow getReceiving() {
		return this.receiving;
	}

	/**
	 * The sending packet direction (i.e. SERVERBOUND on the client and CLIENTBOUND on the server)
	 */
	public PacketFlow getSending() {
		return this.receiving.getOpposite();
	}

	public static Connection connectToServer(InetSocketAddress address, boolean useEpollIfAvailable, @Nullable LocalSampleLogger sampleLogger) {
		Connection connection = new Connection(PacketFlow.CLIENTBOUND);
		if (sampleLogger != null) {
			connection.setBandwidthLogger(sampleLogger);
		}

		ChannelFuture channelFuture = connect(address, useEpollIfAvailable, connection);
		channelFuture.syncUninterruptibly();
		return connection;
	}

	public static ChannelFuture connect(InetSocketAddress address, boolean useEpollIfAvailable, Connection connection) {
		Class<? extends SocketChannel> class_;
		EventLoopGroup eventLoopGroup;
		if (Epoll.isAvailable() && useEpollIfAvailable) {
			class_ = EpollSocketChannel.class;
			eventLoopGroup = (EventLoopGroup)NETWORK_EPOLL_WORKER_GROUP.get();
		} else {
			class_ = NioSocketChannel.class;
			eventLoopGroup = (EventLoopGroup)NETWORK_WORKER_GROUP.get();
		}

		return new Bootstrap().group(eventLoopGroup).handler(new ChannelInitializer<Channel>() {
			@Override
			protected void initChannel(Channel channel) {
				try {
					channel.config().setOption(ChannelOption.TCP_NODELAY, true);
				} catch (ChannelException var3) {
				}

				ChannelPipeline channelPipeline = channel.pipeline().addLast("timeout", new ReadTimeoutHandler(30));
				Connection.configureSerialization(channelPipeline, PacketFlow.CLIENTBOUND, false, connection.bandwidthDebugMonitor);
				connection.configurePacketHandler(channelPipeline);
			}
		}).channel(class_).connect(address.getAddress(), address.getPort());
	}

	private static String outboundHandlerName(boolean clientbound) {
		return clientbound ? "encoder" : "outbound_config";
	}

	private static String inboundHandlerName(boolean serverbound) {
		return serverbound ? "decoder" : "inbound_config";
	}

	public void configurePacketHandler(ChannelPipeline pipeline) {
		pipeline.addLast("hackfix", new ChannelOutboundHandlerAdapter() {
			@Override
			public void write(ChannelHandlerContext channelHandlerContext, Object object, ChannelPromise channelPromise) throws Exception {
				super.write(channelHandlerContext, object, channelPromise);
			}
		}).addLast("packet_handler", this);
	}

	public static void configureSerialization(ChannelPipeline pipeline, PacketFlow flow, boolean memoryOnly, @Nullable BandwidthDebugMonitor bandwithDebugMonitor) {
		PacketFlow packetFlow = flow.getOpposite();
		boolean bl = flow == PacketFlow.SERVERBOUND;
		boolean bl2 = packetFlow == PacketFlow.SERVERBOUND;
		pipeline.addLast("splitter", createFrameDecoder(bandwithDebugMonitor, memoryOnly))
			.addLast(new FlowControlHandler())
			.addLast(inboundHandlerName(bl), (ChannelHandler)(bl ? new PacketDecoder<>(INITIAL_PROTOCOL) : new UnconfiguredPipelineHandler.Inbound()))
			.addLast("prepender", createFrameEncoder(memoryOnly))
			.addLast(outboundHandlerName(bl2), (ChannelHandler)(bl2 ? new PacketEncoder<>(INITIAL_PROTOCOL) : new UnconfiguredPipelineHandler.Outbound()));
	}

	private static ChannelOutboundHandler createFrameEncoder(boolean memoryOnly) {
		return (ChannelOutboundHandler)(memoryOnly ? new NoOpFrameEncoder() : new Varint21LengthFieldPrepender());
	}

	private static ChannelInboundHandler createFrameDecoder(@Nullable BandwidthDebugMonitor bandwithDebugMonitor, boolean memoryOnly) {
		if (!memoryOnly) {
			return new Varint21FrameDecoder(bandwithDebugMonitor);
		} else {
			return (ChannelInboundHandler)(bandwithDebugMonitor != null ? new MonitorFrameDecoder(bandwithDebugMonitor) : new NoOpFrameDecoder());
		}
	}

	public static void configureInMemoryPipeline(ChannelPipeline pipeline, PacketFlow flow) {
		configureSerialization(pipeline, flow, true, null);
	}

	/**
	 * Prepares a clientside Connection for a local in-memory connection ("single player").
	 * Establishes a connection to the socket supplied and configures the channel pipeline (only the packet handler is necessary,
	 * since this is for an in-memory connection). Returns the newly created instance.
	 */
	public static Connection connectToLocalServer(SocketAddress address) {
		final Connection connection = new Connection(PacketFlow.CLIENTBOUND);
		new Bootstrap().group((EventLoopGroup)LOCAL_WORKER_GROUP.get()).handler(new ChannelInitializer<Channel>() {
			@Override
			protected void initChannel(Channel channel) {
				ChannelPipeline channelPipeline = channel.pipeline();
				Connection.configureInMemoryPipeline(channelPipeline, PacketFlow.CLIENTBOUND);
				connection.configurePacketHandler(channelPipeline);
			}
		}).channel(LocalChannel.class).connect(address).syncUninterruptibly();
		return connection;
	}

	/**
	 * Enables encryption for this connection using the given decrypting and encrypting ciphers.
	 * This adds new handlers to this connection's pipeline which handle the decrypting and encrypting.
	 * This happens as part of the normal network handshake.
	 * 
	 * @see net.minecraft.network.protocol.login.ClientboundHelloPacket
	 * @see net.minecraft.network.protocol.login.ServerboundKeyPacket
	 */
	public void setEncryptionKey(Cipher decryptingCipher, Cipher encryptingCipher) {
		this.encrypted = true;
		this.channel.pipeline().addBefore("splitter", "decrypt", new CipherDecoder(decryptingCipher));
		this.channel.pipeline().addBefore("prepender", "encrypt", new CipherEncoder(encryptingCipher));
	}

	public boolean isEncrypted() {
		return this.encrypted;
	}

	/**
	 * Returns {@code true} if this {@code Connection} has an active channel, {@code false} otherwise.
	 */
	public boolean isConnected() {
		return this.channel != null && this.channel.isOpen();
	}

	/**
	 * Returns {@code true} while this connection is still connecting, i.e. {@link #channelActive} has not fired yet.
	 */
	public boolean isConnecting() {
		return this.channel == null;
	}

	/**
	 * Gets the current handler for processing packets
	 */
	@Nullable
	public PacketListener getPacketListener() {
		return this.packetListener;
	}

	@Nullable
	public DisconnectionDetails getDisconnectionDetails() {
		return this.disconnectionDetails;
	}

	/**
	 * Switches the channel to manual reading modus
	 */
	public void setReadOnly() {
		if (this.channel != null) {
			this.channel.config().setAutoRead(false);
		}
	}

	/**
	 * Enables or disables compression for this connection. If {@code threshold} is >= 0 then a {@link CompressionDecoder} and {@link CompressionEncoder}
	 * are installed in the pipeline or updated if they already exist. If {@code threshold} is < 0 then any such codec are removed.
	 * 
	 * Compression is enabled as part of the connection handshake when the server sends {@link net.minecraft.network.protocol.login.ClientboundLoginCompressionPacket}.
	 */
	public void setupCompression(int threshold, boolean validateDecompressed) {
		if (threshold >= 0) {
			if (this.channel.pipeline().get("decompress") instanceof CompressionDecoder compressionDecoder) {
				compressionDecoder.setThreshold(threshold, validateDecompressed);
			} else {
				this.channel.pipeline().addAfter("splitter", "decompress", new CompressionDecoder(threshold, validateDecompressed));
			}

			if (this.channel.pipeline().get("compress") instanceof CompressionEncoder compressionEncoder) {
				compressionEncoder.setThreshold(threshold);
			} else {
				this.channel.pipeline().addAfter("prepender", "compress", new CompressionEncoder(threshold));
			}
		} else {
			if (this.channel.pipeline().get("decompress") instanceof CompressionDecoder) {
				this.channel.pipeline().remove("decompress");
			}

			if (this.channel.pipeline().get("compress") instanceof CompressionEncoder) {
				this.channel.pipeline().remove("compress");
			}
		}
	}

	/**
	 * Checks if the channel is no longer active and if so, processes the disconnection
	 * by notifying the current packet listener, which will handle things like removing the player from the world (serverside) or
	 * showing the disconnection screen (clientside).
	 */
	public void handleDisconnection() {
		if (this.channel != null && !this.channel.isOpen()) {
			if (this.disconnectionHandled) {
				LOGGER.warn("handleDisconnection() called twice");
			} else {
				this.disconnectionHandled = true;
				PacketListener packetListener = this.getPacketListener();
				PacketListener packetListener2 = packetListener != null ? packetListener : this.disconnectListener;
				if (packetListener2 != null) {
					DisconnectionDetails disconnectionDetails = (DisconnectionDetails)Objects.requireNonNullElseGet(
						this.getDisconnectionDetails(), () -> new DisconnectionDetails(Component.translatable("multiplayer.disconnect.generic"))
					);
					packetListener2.onDisconnect(disconnectionDetails);
				}
			}
		}
	}

	public float getAverageReceivedPackets() {
		return this.averageReceivedPackets;
	}

	public float getAverageSentPackets() {
		return this.averageSentPackets;
	}

	public void setBandwidthLogger(LocalSampleLogger bandwithLogger) {
		this.bandwidthDebugMonitor = new BandwidthDebugMonitor(bandwithLogger);
	}
}