merge thrift

dubbo-all/pom.xml

@@ -184,6 +184,13 @@
             <scope>compile</scope>
             <optional>true</optional>
         </dependency>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-rpc-native-thrift</artifactId>
+            <version>${project.version}</version>
+            <scope>compile</scope>
+            <optional>true</optional>
+        </dependency>
         <dependency>
             <groupId>org.apache.dubbo</groupId>
             <artifactId>dubbo-rpc-memcached</artifactId>
@@ -591,6 +598,7 @@
                                     <include>org.apache.dubbo:dubbo-metadata-report-consul</include>
                                     <include>org.apache.dubbo:dubbo-metadata-report-etcd</include>
                                     <include>org.apache.dubbo:dubbo-serialization-native-hession</include>
+                                    <include>org.apache.dubbo:dubbo-rpc-native-thrift</include>
                                 </includes>
                             </artifactSet>
                             <transformers>

dubbo-bom/pom.xml

@@ -202,6 +202,11 @@
                 <artifactId>dubbo-rpc-thrift</artifactId>
                 <version>${project.version}</version>
             </dependency>
+            <dependency>
+                <groupId>org.apache.dubbo</groupId>
+                <artifactId>dubbo-rpc-native-thrift</artifactId>
+                <version>${project.version}</version>
+            </dependency>
             <dependency>
                 <groupId>org.apache.dubbo</groupId>
                 <artifactId>dubbo-rpc-memcached</artifactId>

dubbo-dependencies-bom/pom.xml

@@ -106,7 +106,7 @@
         <consul_process_version>2.0.0</consul_process_version>
         <xmemcached_version>1.3.6</xmemcached_version>
         <cxf_version>3.1.15</cxf_version>
-        <thrift_version>0.8.0</thrift_version>
+        <thrift_version>0.12.0</thrift_version>
         <hessian_version>4.0.38</hessian_version>
         <servlet_version>3.1.0</servlet_version>
         <jetty_version>9.4.11.v20180605</jetty_version>

dubbo-distribution/pom.xml

@@ -140,6 +140,11 @@
             <artifactId>dubbo-rpc-thrift</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-rpc-native-thrift</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.dubbo</groupId>
             <artifactId>dubbo-rpc-memcached</artifactId>

dubbo-rpc/dubbo-rpc-native-thrift/pom.xml

@@ -0,0 +1,49 @@
+<!--
+  Licensed to the Apache Software Foundation (ASF) under one or more
+  contributor license agreements.  See the NOTICE file distributed with
+  this work for additional information regarding copyright ownership.
+  The ASF licenses this file to You under the Apache License, Version 2.0
+  (the "License"); you may not use this file except in compliance with
+  the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.dubbo</groupId>
+        <artifactId>dubbo-rpc</artifactId>
+        <version>${revision}</version>
+    </parent>
+    <artifactId>dubbo-rpc-native-thrift</artifactId>
+    <packaging>jar</packaging>
+    <name>${project.artifactId}</name>
+    <description>The thrift rpc module of dubbo project</description>
+    <properties>
+        <skip_maven_deploy>false</skip_maven_deploy>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-rpc-api</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.thrift</groupId>
+            <artifactId>libthrift</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-serialization-jdk</artifactId>
+            <version>${revision}</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

dubbo-rpc/dubbo-rpc-native-thrift/src/main/java/org/apache/dubbo/rpc/protocol/nativethrift/ThriftProtocol.java

@@ -0,0 +1,179 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.dubbo.rpc.protocol.nativethrift;
+
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.protocol.AbstractProxyProtocol;
+import org.apache.thrift.TException;
+import org.apache.thrift.TMultiplexedProcessor;
+import org.apache.thrift.TProcessor;
+import org.apache.thrift.protocol.TCompactProtocol;
+import org.apache.thrift.protocol.TMultiplexedProtocol;
+import org.apache.thrift.protocol.TProtocol;
+import org.apache.thrift.server.TServer;
+import org.apache.thrift.server.TThreadedSelectorServer;
+import org.apache.thrift.transport.TFramedTransport;
+import org.apache.thrift.transport.TNonblockingServerSocket;
+import org.apache.thrift.transport.TSocket;
+import org.apache.thrift.transport.TTransport;
+
+import java.lang.reflect.Constructor;
+import java.net.InetSocketAddress;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ *  native thrift protocol
+ */
+public class ThriftProtocol extends AbstractProxyProtocol {
+
+    public static final int DEFAULT_PORT = 40880;
+
+    public static final String NAME = "native-thrift";
+    public static final String THRIFT_IFACE = "$Iface";
+    public static final String THRIFT_PROCESSOR = "$Processor";
+    public static final String THRIFT_CLIENT = "$Client";
+
+    private static final Map<String, TServer> serverMap = new HashMap<>();
+    private TMultiplexedProcessor processor = new TMultiplexedProcessor();
+
+    @Override
+    public int getDefaultPort() {
+        return DEFAULT_PORT;
+    }
+
+    public ThriftProtocol() {
+        super(TException.class, RpcException.class);
+    }
+
+    @Override
+    protected <T> Runnable doExport(T impl, Class<T> type, URL url) throws RpcException {
+        return exportThreadedSelectorServer(impl, type, url);
+    }
+
+    @Override
+    protected <T> T doRefer(Class<T> type, URL url) throws RpcException {
+        return doReferFrameAndCompact(type, url);
+    }
+
+    public ThriftProtocol(Class<?>... exceptions) {
+        super(exceptions);
+    }
+
+    private <T> Runnable exportThreadedSelectorServer(T impl, Class<T> type, URL url) throws RpcException {
+
+        TThreadedSelectorServer.Args tArgs = null;
+        String typeName = type.getName();
+
+        TServer tserver = null;
+        if (typeName.endsWith(THRIFT_IFACE)) {
+            String processorClsName = typeName.substring(0, typeName.indexOf(THRIFT_IFACE)) + THRIFT_PROCESSOR;
+            try {
+                Class<?> clazz = Class.forName(processorClsName);
+                Constructor constructor = clazz.getConstructor(type);
+                try {
+                    TProcessor tprocessor = (TProcessor) constructor.newInstance(impl);
+                    processor.registerProcessor(typeName,tprocessor);
+
+                    tserver = serverMap.get(url.getAddress());
+                    if(tserver == null) {
+
+                        /**Solve the problem of only 50 of the default number of concurrent connections*/
+                        TNonblockingServerSocket.NonblockingAbstractServerSocketArgs args = new TNonblockingServerSocket.NonblockingAbstractServerSocketArgs();
+                        /**1000 connections*/
+                        args.backlog(1000);
+                        args.bindAddr(new InetSocketAddress(url.getHost(), url.getPort()));
+                        /**timeout: 10s */
+                        args.clientTimeout(10000);
+
+                        TNonblockingServerSocket transport = new TNonblockingServerSocket(args);
+
+                        tArgs = new TThreadedSelectorServer.Args(transport);
+                        tArgs.workerThreads(200);
+                        tArgs.selectorThreads(4);
+                        tArgs.acceptQueueSizePerThread(256);
+                        tArgs.processor(processor);
+                        tArgs.transportFactory(new TFramedTransport.Factory());
+                        tArgs.protocolFactory(new TCompactProtocol.Factory());
+                    }else{
+                        return null; // if server is starting, return and do nothing here
+                    }
+                } catch (Exception e) {
+                    logger.error(e.getMessage(), e);
+                    throw new RpcException("Fail to create nativethrift server(" + url + ") : " + e.getMessage(), e);
+                }
+            } catch (Exception e) {
+                logger.error(e.getMessage(), e);
+                throw new RpcException("Fail to create nativethrift server(" + url + ") : " + e.getMessage(), e);
+            }
+        }
+
+        if (tserver == null && tArgs == null) {
+            logger.error("Fail to create nativethrift server(" + url + ") due to null args");
+            throw new RpcException("Fail to create nativethrift server(" + url + ") due to null args");
+        }
+        final TServer thriftServer =  new TThreadedSelectorServer(tArgs);
+        serverMap.put(url.getAddress(),thriftServer);
+
+        new Thread(() -> {
+            logger.info("Start Thrift ThreadedSelectorServer");
+            thriftServer.serve();
+            logger.info("Thrift ThreadedSelectorServer started.");
+        }).start();
+
+        return () -> {
+            try {
+                logger.info("Close Thrift NonblockingServer");
+                thriftServer.stop();
+            } catch (Throwable e) {
+                logger.warn(e.getMessage(), e);
+            }
+        };
+    }
+
+    private <T> T doReferFrameAndCompact(Class<T> type, URL url) throws RpcException {
+
+        try {
+            T thriftClient = null;
+            String typeName = type.getName();
+            if (typeName.endsWith(THRIFT_IFACE)) {
+                String clientClsName = typeName.substring(0, typeName.indexOf(THRIFT_IFACE)) + THRIFT_CLIENT;
+                Class<?> clazz = Class.forName(clientClsName);
+                Constructor constructor = clazz.getConstructor(TProtocol.class);
+                try {
+                    TSocket tSocket = new TSocket(url.getHost(), url.getPort());
+                    TTransport transport = new TFramedTransport(tSocket);
+                    TProtocol tprotocol = new TCompactProtocol(transport);
+                    TMultiplexedProtocol protocol = new TMultiplexedProtocol(tprotocol,typeName);
+                    thriftClient = (T) constructor.newInstance(protocol);
+                    transport.open();
+                    logger.info("nativethrift client opened for service(" + url + ")");
+                } catch (Exception e) {
+                    logger.error(e.getMessage(), e);
+                    throw new RpcException("Fail to create remote client:" + e.getMessage(), e);
+                }
+            }
+            return thriftClient;
+        } catch (Exception e) {
+            logger.error(e.getMessage(), e);
+            throw new RpcException("Fail to create remote client for service(" + url + "): " + e.getMessage(), e);
+        }
+    }
+
+}

dubbo-rpc/dubbo-rpc-native-thrift/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Protocol

@@ -0,0 +1 @@
+native-thrift=org.apache.dubbo.rpc.protocol.nativethrift.ThriftProtocol

dubbo-rpc/dubbo-rpc-native-thrift/src/test/idls/DemoService.thrift

@@ -0,0 +1,17 @@
+namespace java org.apache.dubbo.rpc.protocol.nativethrift
+namespace go demo
+/*Demo service define file,can be generated to inteface files*/
+/*Here test the 7 kind of data type*/
+service DemoService {
+    string sayHello(1:required string name);
+
+    bool hasName( 1:required bool hasName);
+
+    string sayHelloTimes(1:required string name, 2:required i32 times);
+
+    void timeOut(1:required i32 millis);
+
+    string customException();
+
+    string context(1:required string name);
+}

dubbo-rpc/dubbo-rpc-native-thrift/src/test/idls/UserService.thrift

@@ -0,0 +1,6 @@
+namespace java org.apache.dubbo.rpc.protocol.nativethrift
+namespace go demo
+
+service UserService {
+    string find(1:required i32 id);
+}