[change] refactor and move netty implementations to tars-netty.

Author: kyy1996

Contributing.md

@@ -13,3 +13,4 @@ If you contributed but cannot find your ID here, please submit PR and add your G
 - woodwind
 - XenoAmess
 - yukkiball
+- Kongyuanyuan

core/pom.xml

@@ -21,12 +21,6 @@
     </properties>
 
     <dependencies>
-        <dependency>
-            <groupId>io.netty</groupId>
-            <artifactId>netty-tcnative</artifactId>
-            <version>2.0.20.Final</version> <!-- see table for correct version -->
-            <scope>runtime</scope>
-        </dependency>
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>
@@ -36,7 +30,6 @@
         <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>
-            <version>1.7.30</version>
         </dependency>
         <dependency>
             <groupId>com.google.guava</groupId>
@@ -52,7 +45,6 @@
         <dependency>
             <groupId>io.netty</groupId>
             <artifactId>netty-all</artifactId>
-            <version>4.1.56.Final</version>
         </dependency>
         <dependency>
             <groupId>javax.servlet</groupId>
@@ -96,11 +88,5 @@
             <version>4.3.2</version>
             <scope>compile</scope>
         </dependency>
-        <dependency>
-            <groupId>org.apache.httpcomponents</groupId>
-            <artifactId>httpcore</artifactId>
-            <version>4.3.2</version>
-            <scope>compile</scope>
-        </dependency>
     </dependencies>
 </project>

core/src/main/java/com/qq/tars/client/rpc/NettyTransporter.java

@@ -25,20 +25,10 @@
 import com.qq.tars.rpc.common.Url;
 import com.qq.tars.rpc.common.util.concurrent.ConcurrentHashSet;
 import com.qq.tars.rpc.exc.ClientException;
-import com.qq.tars.rpc.protocol.tars.TarsServantResponse;
 import com.qq.tars.support.log.LoggerFactory;
-import io.netty.channel.Channel;
 import org.slf4j.Logger;
 
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
@@ -89,23 +79,21 @@ public void refresh() {
         ScheduledExecutorManager.getInstance().schedule(() -> destroy(brokenInvokers), Math.max(servantProxyConfig.getAsyncTimeout(), servantProxyConfig.getSyncTimeout()), TimeUnit.MILLISECONDS);
     }
 
-    protected RPCClient[] getClients(Url url) throws IOException {
+    protected RPCClient[] getClients(Url url) {
         int connections = url.getParameter(Constants.TARS_CLIENT_CONNECTIONS, Constants.DEFAULT_CONNECTION);
-        RPCClient[] clients = new NettyServantClient[connections];
+        RPCClient[] clients = new RPCClient[connections];
         for (int i = 0; i < clients.length; i++) {
             clients[i] = initClient(url);
         }
         return clients;
     }
 
     protected RPCClient initClient(Url url) {
-        RPCClient client = null;
         try {
-            client = NettyTransporter.connect(url, servantProxyConfig, new InnerDefaultHandler());
+            return TransporterAbstractFactory.getInstance().getTransporterFactory().connect(url, servantProxyConfig);
         } catch (Throwable e) {
             throw new ClientException(servantProxyConfig.getSimpleObjectName(), "Fail to create client|" + url.toIdentityString() + "|" + e.getLocalizedMessage(), e);
         }
-        return client;
     }
 
     protected ConcurrentHashSet<Invoker<T>> initInvoker() {
@@ -158,43 +146,4 @@ private void destroy(Collection<Invoker<T>> invokers) {
             }
         }
     }
-
-    private static class InnerDefaultHandler implements ChannelHandler {
-
-        @Override
-        public void connected(Channel channel) {
-
-        }
-
-        @Override
-        public void disconnected(Channel channel) {
-
-        }
-
-        @Override
-        public void send(Channel channel, Object message) {
-
-        }
-
-        @Override
-        public void received(Channel channel, Object message) {
-            TarsServantResponse response = (TarsServantResponse) message;
-            if (logger.isDebugEnabled()) {
-                System.out.println();
-                logger.debug("[tars]netty receive message id is " + response.getRequestId());
-            }
-            TicketFeature.getFeature(response.getRequestId()).complete(response);
-        }
-
-        @Override
-        public void caught(Channel channel, Throwable exception) {
-
-        }
-
-        @Override
-        public void destroy() {
-
-        }
-    }
-
 }

core/src/main/java/com/qq/tars/client/rpc/ServantProtocolInvoker.java

@@ -0,0 +1,41 @@
+package com.qq.tars.client.rpc;
+
+import com.google.common.collect.ImmutableList;
+import com.qq.tars.support.log.LoggerFactory;
+import org.slf4j.Logger;
+
+import java.util.List;
+import java.util.ServiceLoader;
+
+/**
+ * Abstract factory for transporter.
+ * Used to load different transporter implementation(such as netty, reactor, restTemplate, etc) through SPI:{@code com.qq.tars.client.rpc.TransporterFactory}
+ *
+ * @author kongyuanyuan
+ */
+public class TransporterAbstractFactory {
+    private static final Logger log = LoggerFactory.getTransporterLogger();
+
+    private static final TransporterAbstractFactory INSTANCE = new TransporterAbstractFactory();
+    private final TransporterFactory defaultTransporterFactory;
+
+    private TransporterAbstractFactory() {
+        ServiceLoader<TransporterFactory> transporterFactories = ServiceLoader.load(TransporterFactory.class, this.getClass().getClassLoader());
+        List<TransporterFactory> transporterFactoryList = ImmutableList.copyOf(transporterFactories);
+        if (transporterFactoryList.isEmpty()) {
+            throw new IllegalStateException("No TransporterFactory implementation found on the classpath through SPI. Try to add tars-netty in pom.xml.");
+        }
+        defaultTransporterFactory = transporterFactoryList.get(0);
+        if (transporterFactoryList.size() > 1) {
+            log.warn("More than one transporter factory found. {} will be used.", defaultTransporterFactory.getClass().getCanonicalName());
+        }
+    }
+
+    public static TransporterAbstractFactory getInstance() {
+        return INSTANCE;
+    }
+
+    public TransporterFactory getTransporterFactory() {
+        return defaultTransporterFactory;
+    }
+}

core/src/main/java/com/qq/tars/client/rpc/TarsDecoder.java

@@ -0,0 +1,43 @@
+package com.qq.tars.client.rpc;
+
+import com.qq.tars.client.ServantProxyConfig;
+import com.qq.tars.rpc.common.Url;
+import com.qq.tars.server.config.ServantAdapterConfig;
+import com.qq.tars.server.core.Processor;
+
+/**
+ * Abstract factory for creating servers and clients in transportation。
+ *
+ * @author kongyuanyuan
+ */
+public interface TransporterFactory {
+
+    /**
+     * Connect with a server node using the given servant config.
+     *
+     * @param url                server node url
+     * @param servantProxyConfig servant config
+     * @return rpc client instance
+     */
+    RPCClient connect(Url url, ServantProxyConfig servantProxyConfig);
+
+    /**
+     * Connect with a server node using the given servant config and channel handler.
+     *
+     * @param url                server node url
+     * @param servantProxyConfig servant config
+     * @param channelHandler     custom channel handler. Can be used in unit tests.
+     * @return rpc client instance
+     */
+    RPCClient connect(Url url, ServantProxyConfig servantProxyConfig, ChannelHandler channelHandler);
+
+    /**
+     * Get the server instance for transportation which can be used to bind ip/port with {@link TransporterServer#bind()} method.
+     *
+     * @param servantAdapterConfig the servant adapter config which is provided by the platform.
+     * @param processor            the request processor.
+     * @return transporter server instance
+     * @see com.qq.tars.server.core.TarsServantProcessor
+     */
+    TransporterServer getTransporterServer(ServantAdapterConfig servantAdapterConfig, Processor processor);
+}

core/src/main/java/com/qq/tars/client/rpc/TransporterAbstractFactory.java

@@ -0,0 +1,13 @@
+package com.qq.tars.client.rpc;
+
+/**
+ * 传输层服务器。需要实现{@code bind}方法来绑定服务地址与端口号。
+ *
+ * @author kongyuanyuan
+ */
+public interface TransporterServer {
+    /**
+     * 执行绑定动作。
+     */
+    void bind();
+}

core/src/main/java/com/qq/tars/client/rpc/TransporterFactory.java

@@ -17,7 +17,7 @@
 package com.qq.tars.client.support;
 
 import com.qq.tars.client.CommunicatorConfig;
-import com.qq.tars.client.rpc.NettyServantClient;
+import com.qq.tars.client.rpc.RPCClient;
 import com.qq.tars.common.util.concurrent.TaskQueue;
 import com.qq.tars.common.util.concurrent.TaskThreadFactory;
 import com.qq.tars.common.util.concurrent.TaskThreadPoolExecutor;
@@ -35,7 +35,7 @@ public class ClientPoolManager {
     public static ThreadPoolExecutor getClientThreadPoolExecutor(CommunicatorConfig communicatorConfig) {
         ThreadPoolExecutor clientPoolExecutor = clientThreadPoolMap.get(communicatorConfig);
         if (clientPoolExecutor == null) {
-            synchronized (NettyServantClient.class) {
+            synchronized (RPCClient.class) {
                 clientPoolExecutor = clientThreadPoolMap.get(communicatorConfig);
                 if (clientPoolExecutor == null) {
                     clientThreadPoolMap.put(communicatorConfig, createThreadPool(communicatorConfig));

core/src/main/java/com/qq/tars/client/rpc/TransporterServer.java

@@ -17,7 +17,7 @@
 package com.qq.tars.client.support;
 
 import com.qq.tars.client.CommunicatorConfig;
-import com.qq.tars.client.rpc.NettyServantClient;
+import com.qq.tars.client.rpc.RPCClient;
 import com.qq.tars.common.util.concurrent.TaskQueue;
 import com.qq.tars.common.util.concurrent.TaskThreadFactory;
 import com.qq.tars.common.util.concurrent.TaskThreadPoolExecutor;
@@ -34,7 +34,7 @@ public static ThreadPoolExecutor getClientThreadPoolExecutor(CommunicatorConfig
         String contextIdentity = resolveCurrentContextIdentity();
         ThreadPoolExecutor clientPoolExecutor = clientThreadPoolMap.get(contextIdentity);
         if (clientPoolExecutor == null) {
-            synchronized (NettyServantClient.class) {
+            synchronized (RPCClient.class) {
                 clientPoolExecutor = clientThreadPoolMap.get(contextIdentity);
                 if (clientPoolExecutor == null) {
                     clientThreadPoolMap.put(contextIdentity, createThreadPool(communicatorConfig));