release inactive long-polling channel from ClientsBox to avoid memory leak

Author: dawnbreaks

.gitignore

@@ -5,3 +5,5 @@
 /target
 
 /gnupg
+.idea
+*.iml

src/main/java/com/corundumstudio/socketio/handler/ClientHead.java

@@ -15,19 +15,6 @@
  */
 package com.corundumstudio.socketio.handler;
 
-import java.net.SocketAddress;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Queue;
-import java.util.Set;
-import java.util.UUID;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import com.corundumstudio.socketio.Configuration;
 import com.corundumstudio.socketio.DisconnectableHub;
 import com.corundumstudio.socketio.HandshakeData;
@@ -43,13 +30,20 @@
 import com.corundumstudio.socketio.store.Store;
 import com.corundumstudio.socketio.store.StoreFactory;
 import com.corundumstudio.socketio.transport.NamespaceClient;
-
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.handler.codec.http.HttpHeaderNames;
 import io.netty.util.AttributeKey;
 import io.netty.util.internal.PlatformDependent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.SocketAddress;
+import java.util.*;
+import java.util.Map.Entry;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 public class ClientHead {
 
@@ -105,6 +99,14 @@ public void bindChannel(Channel channel, Transport transport) {
         sendPackets(transport, channel);
     }
 
+    public void releasePollingChannel(Channel channel) {
+        TransportState state = channels.get(Transport.POLLING);
+        if(channel.equals(state.getChannel())) {
+            clientsBox.remove(channel);
+            state.update(null);
+        }
+    }
+
     public String getOrigin() {
         return handshakeData.getHttpHeaders().get(HttpHeaderNames.ORIGIN);
     }

src/main/java/com/corundumstudio/socketio/transport/PollingTransport.java

@@ -15,15 +15,6 @@
  */
 package com.corundumstudio.socketio.transport;
 
-import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
-
-import java.io.IOException;
-import java.util.List;
-import java.util.UUID;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import com.corundumstudio.socketio.Transport;
 import com.corundumstudio.socketio.handler.AuthorizeHandler;
 import com.corundumstudio.socketio.handler.ClientHead;
@@ -33,19 +24,21 @@
 import com.corundumstudio.socketio.messages.XHROptionsMessage;
 import com.corundumstudio.socketio.messages.XHRPostMessage;
 import com.corundumstudio.socketio.protocol.PacketDecoder;
-
 import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandler.Sharable;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
-import io.netty.handler.codec.http.DefaultHttpResponse;
-import io.netty.handler.codec.http.FullHttpRequest;
-import io.netty.handler.codec.http.HttpHeaderNames;
-import io.netty.handler.codec.http.HttpMethod;
-import io.netty.handler.codec.http.HttpResponse;
-import io.netty.handler.codec.http.HttpResponseStatus;
-import io.netty.handler.codec.http.QueryStringDecoder;
+import io.netty.handler.codec.http.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.UUID;
+
+import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
 
 @Sharable
 public class PollingTransport extends ChannelInboundHandlerAdapter {
@@ -186,4 +179,15 @@ private void sendError(ChannelHandlerContext ctx) {
         ctx.channel().writeAndFlush(res).addListener(ChannelFutureListener.CLOSE);
     }
 
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        final Channel channel = ctx.channel();
+        ClientHead client = clientsBox.get(channel);
+        if (client != null && client.isTransportChannel(ctx.channel(), Transport.POLLING)) {
+            log.debug("channel inactive {}", client.getSessionId());
+            client.releasePollingChannel(channel);
+        }
+        super.channelInactive(ctx);
+    }
+
 }