Connection closing optimization (#697)

Author: GhostBoyBoy

amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/proxy/ProxyConnection.java

@@ -17,7 +17,6 @@
 import static java.nio.charset.StandardCharsets.US_ASCII;
 import static java.nio.charset.StandardCharsets.UTF_8;
 
-import com.google.common.collect.Sets;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
@@ -28,7 +27,6 @@
 import io.streamnative.pulsar.handlers.amqp.AmqpProtocolHandler;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicInteger;
 import lombok.Getter;
@@ -232,17 +230,6 @@ public void receiveConnectionOpen(AMQShortString virtualHost, AMQShortString cap
             }
         }
         vhost = virtualHostStr;
-
-        proxyService.getVhostConnectionMap().compute(vhost, (v, set) -> {
-            if (set == null) {
-                Set<ProxyConnection> proxyConnectionSet =  Sets.newConcurrentHashSet();
-                proxyConnectionSet.add(this);
-                return proxyConnectionSet;
-            } else {
-                set.add(this);
-                return set;
-            }
-        });
         handleConnect(new AtomicInteger(5));
     }
 

amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/proxy/ProxyHandler.java

@@ -82,7 +82,7 @@ protected void initChannel(SocketChannel ch) throws Exception {
         channelFuture.addListener(future -> {
             if (!future.isSuccess()) {
                 // Close the connection if the connection attempt has failed.
-                clientChannel.close();
+                proxyConnection.close();
             }
         });
         state = State.Init;
@@ -94,8 +94,8 @@ private class ProxyBackendHandler extends ChannelInboundHandlerAdapter implement
             ClientMethodProcessor<ClientChannelMethodProcessor>, FutureListener<Void> {
 
         private ChannelHandlerContext cnx;
-        private AMQMethodBody connectResponseBody;
-        private AmqpClientDecoder clientDecoder;
+        private final AMQMethodBody connectResponseBody;
+        private final AmqpClientDecoder clientDecoder;
 
         ProxyBackendHandler(AMQMethodBody responseBody) {
             this.connectResponseBody = responseBody;
@@ -107,11 +107,11 @@ public void operationComplete(Future future) throws Exception {
             // This is invoked when the write operation on the paired connection
             // is completed
             if (future.isSuccess()) {
-                brokerChannel.read();
+                cnx.channel().read();
             } else {
                 log.warn("[{}] [{}] Failed to write on proxy connection. Closing both connections.", clientChannel,
-                        brokerChannel, future.cause());
-                clientChannel.close();
+                        cnx.channel(), future.cause());
+                proxyConnection.close();
             }
 
         }
@@ -123,8 +123,8 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception {
             super.channelActive(ctx);
             for (Object msg : connectMsgList) {
                 ((ByteBuf) msg).retain();
-                brokerChannel.writeAndFlush(msg).addListener(future -> {
-                    brokerChannel.read();
+                ctx.channel().writeAndFlush(msg).addListener(future -> {
+                    ctx.channel().read();
                 });
             }
         }
@@ -166,6 +166,7 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E
             cause.printStackTrace();
             log.error("[" + vhost + "] ProxyBackendHandler [exceptionCaught] - msg: " + cause.getMessage(), cause);
             state = State.Failed;
+            proxyConnection.close();
         }
 
         @Override

amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/proxy/ProxyService.java

@@ -16,15 +16,12 @@
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 
-import com.google.common.collect.Maps;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.Channel;
 import io.netty.channel.EventLoopGroup;
 import io.netty.util.concurrent.DefaultThreadFactory;
 import java.io.Closeable;
 import java.io.IOException;
-import java.util.Map;
-import java.util.Set;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.broker.PulsarService;
@@ -37,23 +34,18 @@
 public class ProxyService implements Closeable {
 
     @Getter
-    private ProxyConfiguration proxyConfig;
+    private final ProxyConfiguration proxyConfig;
     @Getter
-    private PulsarService pulsarService;
+    private final PulsarService pulsarService;
     @Getter
     private LookupHandler lookupHandler;
 
     private Channel listenChannel;
-    private EventLoopGroup acceptorGroup;
+    private final EventLoopGroup acceptorGroup;
     @Getter
-    private EventLoopGroup workerGroup;
+    private final EventLoopGroup workerGroup;
 
-    private DefaultThreadFactory acceptorThreadFactory = new DefaultThreadFactory("amqp-redirect-acceptor");
-    private DefaultThreadFactory workerThreadFactory = new DefaultThreadFactory("amqp-redirect-io");
-    private static final int numThreads = Runtime.getRuntime().availableProcessors();
-
-    @Getter
-    private static final Map<String, Set<ProxyConnection>> vhostConnectionMap = Maps.newConcurrentMap();
+    private static final int NUM_THREADS = Runtime.getRuntime().availableProcessors();
 
     private String tenant;
 
@@ -63,8 +55,10 @@ public ProxyService(ProxyConfiguration proxyConfig, PulsarService pulsarService)
         this.proxyConfig = proxyConfig;
         this.pulsarService = pulsarService;
         this.tenant = this.proxyConfig.getAmqpTenant();
-        acceptorGroup = EventLoopUtil.newEventLoopGroup(1, false, acceptorThreadFactory);
-        workerGroup = EventLoopUtil.newEventLoopGroup(numThreads, false, workerThreadFactory);
+        this.acceptorGroup = EventLoopUtil.newEventLoopGroup(1, false,
+                new DefaultThreadFactory("amqp-redirect-acceptor"));
+        this.workerGroup = EventLoopUtil.newEventLoopGroup(NUM_THREADS, false,
+                new DefaultThreadFactory("amqp-redirect-io"));
     }
 
     private void configValid(ProxyConfiguration proxyConfig) {
@@ -89,16 +83,6 @@ public void start() throws Exception {
         this.lookupHandler = new PulsarServiceLookupHandler(proxyConfig, pulsarService);
     }
 
-    private void releaseConnection(String namespaceName) {
-        log.info("release connection");
-        if (vhostConnectionMap.containsKey(namespaceName)) {
-            Set<ProxyConnection> proxyConnectionSet = vhostConnectionMap.get(namespaceName);
-            for (ProxyConnection proxyConnection : proxyConnectionSet) {
-                proxyConnection.close();
-            }
-        }
-    }
-
     @Override
     public void close() throws IOException {
         if (lookupHandler != null) {