Destroy rakPipeline() properly

Author: Alemiz112

transport-raknet/src/main/java/org/cloudburstmc/netty/channel/raknet/RakServerChannel.java

@@ -30,6 +30,7 @@
 import org.cloudburstmc.netty.handler.codec.raknet.server.RakServerRateLimiter;
 import org.cloudburstmc.netty.handler.codec.raknet.server.RakServerRouteHandler;
 import org.cloudburstmc.netty.handler.codec.raknet.server.RakServerTailHandler;
+import org.cloudburstmc.netty.util.RakUtils;
 
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
@@ -91,13 +92,18 @@ public RakChildChannel getChildChannel(SocketAddress address) {
 
     private void onChildClosed(ChannelFuture channelFuture) {
         RakChildChannel channel = (RakChildChannel) channelFuture.channel();
-        channel.rakPipeline().fireChannelInactive();
-        channel.rakPipeline().fireChannelUnregistered();
         this.childChannelMap.remove(channel.remoteAddress());
 
         if (this.config().getMetrics() != null) {
             this.config().getMetrics().channelClose(channel.remoteAddress());
         }
+
+        channel.rakPipeline().fireChannelInactive();
+        channel.rakPipeline().fireChannelUnregistered();
+        // Need to use reflection to destroy pipeline because
+        // DefaultChannelPipeline.destroy() is only called when channel.isOpen() is false,
+        // but the method is called on parent channel, and there is no other way to destroy pipeline.
+        RakUtils.destroyChannelPipeline(channel.rakPipeline());
     }
 
     @Override

transport-raknet/src/main/java/org/cloudburstmc/netty/handler/codec/raknet/common/RakUnhandledMessagesQueue.java

@@ -48,6 +48,7 @@ public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
     public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
         if (this.future != null) {
             this.future.cancel(false);
+            this.future = null;
         }
 
         EncapsulatedPacket message;

transport-raknet/src/main/java/org/cloudburstmc/netty/util/RakUtils.java

@@ -18,16 +18,19 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
+import io.netty.channel.ChannelPipeline;
 import io.netty.channel.DefaultChannelPipeline;
 
 import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
 import java.net.*;
 import java.util.Queue;
 
 public class RakUtils {
 
     private static final Constructor<DefaultChannelPipeline> DEFAULT_CHANNEL_PIPELINE_CONSTRUCTOR;
+    private static final Method PIPELINE_DESTROY_METHOD;
 
     static {
         try {
@@ -37,6 +40,14 @@ public class RakUtils {
         } catch (NoSuchMethodException e) {
             throw new AssertionError("Unable to find DefaultChannelPipeline(Channel) constructor", e);
         }
+
+        try {
+            Method method = DefaultChannelPipeline.class.getDeclaredMethod("destroy");
+            method.setAccessible(true);
+            PIPELINE_DESTROY_METHOD = method;
+        } catch (NoSuchMethodException e) {
+            throw new AssertionError("Unable to find DefaultChannelPipeline.destroy() method", e);
+        }
     }
 
     public static DefaultChannelPipeline newChannelPipeline(Channel channel) {
@@ -47,6 +58,14 @@ public static DefaultChannelPipeline newChannelPipeline(Channel channel) {
         }
     }
 
+    public static void destroyChannelPipeline(ChannelPipeline pipeline) {
+        try {
+            PIPELINE_DESTROY_METHOD.invoke(pipeline);
+        } catch (IllegalAccessException | InvocationTargetException e) {
+            throw new IllegalStateException("Unable to destroy DefaultChannelPipeline", e);
+        }
+    }
+
     private static final int AF_INET6 = 23;
 
     public static InetSocketAddress readAddress(ByteBuf buffer) {