Merge branch 'master' into chore/add-test-for-rpc-modules

Author: htynkn

LICENSE

@@ -263,4 +263,13 @@ This product bundles and repackages the following code in Google Guava 16.0.1, w
 
  * com.google.common.util.concurrent.ExecutionList
  * com.google.common.util.concurrent.ListenableFuture
- * com.google.common.util.concurrent.ListenableFutureTask
+ * com.google.common.util.concurrent.ListenableFutureTask
+
+For the package com.alibaba.dubbo.common.threadlocal:
+
+This product contains a modified portion of 'Netty', an event-driven asynchronous network application framework also
+ under a "Apache License 2.0" license, see https://github.com/netty/netty/blob/4.1/LICENSE.txt:
+
+ * io.netty.util.concurrent.FastThreadLocal
+ * io.netty.util.internal.InternalThreadLocalMap
+

dependencies-bom/pom.xml

@@ -84,7 +84,7 @@
         <curator_version>2.12.0</curator_version>
         <jedis_version>2.9.0</jedis_version>
         <xmemcached_version>1.3.6</xmemcached_version>
-        <cxf_version>3.0.14</cxf_version>
+        <cxf_version>3.1.15</cxf_version>
         <thrift_version>0.8.0</thrift_version>
         <hessian_version>4.0.38</hessian_version>
         <servlet_version>3.1.0</servlet_version>
@@ -107,6 +107,9 @@
         <logback_version>1.2.2</logback_version>
         <commons_lang3_version>3.4</commons_lang3_version>
         <embedded_redis_version>0.6</embedded_redis_version>
+
+        <jaxb_version>2.2.7</jaxb_version>
+        <activation_version>1.2.0</activation_version>
     </properties>
 
     <dependencyManagement>
@@ -321,6 +324,34 @@
                 <artifactId>commons-lang3</artifactId>
                 <version>${commons_lang3_version}</version>
             </dependency>
+
+            <!-- for dubbo-rpc-webservice -->
+            <dependency>
+                <groupId>javax.xml.bind</groupId>
+                <artifactId>jaxb-api</artifactId>
+                <version>${jaxb_version}</version>
+            </dependency>
+            <dependency>
+                <groupId>com.sun.xml.bind</groupId>
+                <artifactId>jaxb-impl</artifactId>
+                <version>${jaxb_version}</version>
+            </dependency>
+            <dependency>
+                <groupId>com.sun.xml.bind</groupId>
+                <artifactId>jaxb-core</artifactId>
+                <version>${jaxb_version}</version>
+            </dependency>
+            <dependency>
+                <groupId>javax.activation</groupId>
+                <artifactId>javax.activation-api</artifactId>
+                <version>${activation_version}</version>
+            </dependency>
+            <dependency>
+                <groupId>com.sun.activation</groupId>
+                <artifactId>javax.activation</artifactId>
+                <version>${activation_version}</version>
+            </dependency>
+
             <!-- Test lib -->
             <dependency>
                 <groupId>org.apache.curator</groupId>

dubbo-bootstrap/src/main/java/org/apache/dubbo/bootstrap/DubboBootstrap.java

@@ -16,66 +16,65 @@
  */
 package org.apache.dubbo.bootstrap;
 
-import com.alibaba.dubbo.common.extension.ExtensionLoader;
-import com.alibaba.dubbo.common.logger.Logger;
-import com.alibaba.dubbo.common.logger.LoggerFactory;
+import com.alibaba.dubbo.config.DubboShutdownHook;
 import com.alibaba.dubbo.config.ServiceConfig;
-import com.alibaba.dubbo.registry.support.AbstractRegistryFactory;
-import com.alibaba.dubbo.rpc.Protocol;
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * A bootstrap class to easily start and stop Dubbo via programmatic API.
  * The bootstrap class will be responsible to cleanup the resources during stop.
  */
 public class DubboBootstrap {
 
-    private static final Logger logger = LoggerFactory.getLogger(DubboBootstrap.class);
-
     /**
      * The list of ServiceConfig
      */
     private List<ServiceConfig> serviceConfigList;
 
     /**
-     * Has it already been destroyed or not?
+     * Whether register the shutdown hook during start?
      */
-    private final AtomicBoolean destroyed;
+    private final boolean registerShutdownHookOnStart;
 
     /**
      * The shutdown hook used when Dubbo is running under embedded environment
      */
-    private Thread shutdownHook;
+    private DubboShutdownHook shutdownHook;
 
     public DubboBootstrap() {
+        this(true, DubboShutdownHook.getDubboShutdownHook());
+    }
+
+    public DubboBootstrap(boolean registerShutdownHookOnStart) {
+        this(registerShutdownHookOnStart, DubboShutdownHook.getDubboShutdownHook());
+    }
+
+    public DubboBootstrap(boolean registerShutdownHookOnStart, DubboShutdownHook shutdownHook) {
         this.serviceConfigList = new ArrayList<ServiceConfig>();
-        this.destroyed = new AtomicBoolean(false);
-        this.shutdownHook = new Thread(new Runnable() {
-            @Override
-            public void run() {
-                if (logger.isInfoEnabled()) {
-                    logger.info("Run shutdown hook now.");
-                }
-                destroy();
-            }
-        }, "DubboShutdownHook");
+        this.shutdownHook = shutdownHook;
+        this.registerShutdownHookOnStart = registerShutdownHookOnStart;
     }
 
     /**
      * Register service config to bootstrap, which will be called during {@link DubboBootstrap#stop()}
      * @param serviceConfig the service
      * @return the bootstrap instance
      */
-    public DubboBootstrap regsiterServiceConfig(ServiceConfig serviceConfig) {
+    public DubboBootstrap registerServiceConfig(ServiceConfig serviceConfig) {
         serviceConfigList.add(serviceConfig);
         return this;
     }
 
     public void start() {
-        registerShutdownHook();
+        if (registerShutdownHookOnStart) {
+            registerShutdownHook();
+        } else {
+            // DubboShutdown hook has been registered in AbstractConfig,
+            // we need to remove it explicitly
+            removeShutdownHook();
+        }
         for (ServiceConfig serviceConfig: serviceConfigList) {
             serviceConfig.export();
         }
@@ -85,8 +84,10 @@ public void stop() {
         for (ServiceConfig serviceConfig: serviceConfigList) {
             serviceConfig.unexport();
         }
-        destroy();
-        removeShutdownHook();
+        shutdownHook.destroyAll();
+        if (registerShutdownHookOnStart) {
+            removeShutdownHook();
+        }
     }
 
     /**
@@ -107,27 +108,4 @@ public void removeShutdownHook() {
             // ignore - VM is already shutting down
         }
     }
-
-    /**
-     * Destroy all the resources, including registries and protocols.
-     */
-    private void destroy() {
-        if (!destroyed.compareAndSet(false, true)) {
-            return;
-        }
-        // destroy all the registries
-        AbstractRegistryFactory.destroyAll();
-        // destroy all the protocols
-        ExtensionLoader<Protocol> loader = ExtensionLoader.getExtensionLoader(Protocol.class);
-        for (String protocolName : loader.getLoadedExtensions()) {
-            try {
-                Protocol protocol = loader.getLoadedExtension(protocolName);
-                if (protocol != null) {
-                    protocol.destroy();
-                }
-            } catch (Throwable t) {
-                logger.warn(t.getMessage(), t);
-            }
-        }
-    }
 }

dubbo-cluster/src/main/java/com/alibaba/dubbo/rpc/cluster/loadbalance/ConsistentHashLoadBalance.java

@@ -101,10 +101,10 @@ private String toKey(Object[] args) {
 
         private Invoker<T> selectForKey(long hash) {
             Map.Entry<Long, Invoker<T>> entry = virtualInvokers.tailMap(hash, true).firstEntry();
-        	if (entry == null) {
-        		entry = virtualInvokers.firstEntry();
-        	}
-        	return entry.getValue();
+            if (entry == null) {
+                entry = virtualInvokers.firstEntry();
+            }
+            return entry.getValue();
         }
 
         private long hash(byte[] digest, int number) {

dubbo-cluster/src/main/java/com/alibaba/dubbo/rpc/cluster/support/FailbackClusterInvoker.java

@@ -18,12 +18,13 @@
 
 import com.alibaba.dubbo.common.logger.Logger;
 import com.alibaba.dubbo.common.logger.LoggerFactory;
-import com.alibaba.dubbo.common.utils.NamedThreadFactory;
+import com.alibaba.dubbo.common.threadlocal.NamedInternalThreadFactory;
 import com.alibaba.dubbo.rpc.Invocation;
 import com.alibaba.dubbo.rpc.Invoker;
 import com.alibaba.dubbo.rpc.Result;
 import com.alibaba.dubbo.rpc.RpcException;
 import com.alibaba.dubbo.rpc.RpcResult;
+import com.alibaba.dubbo.rpc.RpcContext;
 import com.alibaba.dubbo.rpc.cluster.Directory;
 import com.alibaba.dubbo.rpc.cluster.LoadBalance;
 
@@ -50,7 +51,13 @@ public class FailbackClusterInvoker<T> extends AbstractClusterInvoker<T> {
 
     private static final long RETRY_FAILED_PERIOD = 5 * 1000;
 
-    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2, new NamedThreadFactory("failback-cluster-timer", true));
+    /**
+     * Use {@link NamedInternalThreadFactory} to produce {@link com.alibaba.dubbo.common.threadlocal.InternalThread}
+     * which with the use of {@link com.alibaba.dubbo.common.threadlocal.InternalThreadLocal} in {@link RpcContext}.
+     */
+    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2,
+            new NamedInternalThreadFactory("failback-cluster-timer", true));
+
     private final ConcurrentMap<Invocation, AbstractClusterInvoker<?>> failed = new ConcurrentHashMap<Invocation, AbstractClusterInvoker<?>>();
     private volatile ScheduledFuture<?> retryFuture;
 

dubbo-cluster/src/main/java/com/alibaba/dubbo/rpc/cluster/support/ForkingClusterInvoker.java

@@ -17,7 +17,7 @@
 package com.alibaba.dubbo.rpc.cluster.support;
 
 import com.alibaba.dubbo.common.Constants;
-import com.alibaba.dubbo.common.utils.NamedThreadFactory;
+import com.alibaba.dubbo.common.threadlocal.NamedInternalThreadFactory;
 import com.alibaba.dubbo.rpc.Invocation;
 import com.alibaba.dubbo.rpc.Invoker;
 import com.alibaba.dubbo.rpc.Result;
@@ -43,7 +43,12 @@
  */
 public class ForkingClusterInvoker<T> extends AbstractClusterInvoker<T> {
 
-    private final ExecutorService executor = Executors.newCachedThreadPool(new NamedThreadFactory("forking-cluster-timer", true));
+    /**
+     * Use {@link NamedInternalThreadFactory} to produce {@link com.alibaba.dubbo.common.threadlocal.InternalThread}
+     * which with the use of {@link com.alibaba.dubbo.common.threadlocal.InternalThreadLocal} in {@link RpcContext}.
+     */
+    private final ExecutorService executor = Executors.newCachedThreadPool(
+            new NamedInternalThreadFactory("forking-cluster-timer", true));
 
     public ForkingClusterInvoker(Directory<T> directory) {
         super(directory);

dubbo-common/src/main/java/com/alibaba/dubbo/common/Constants.java

@@ -156,7 +156,7 @@ public class Constants {
 
     public static final String LOADBALANCE_KEY = "loadbalance";
 
-    // key for router type, for e.g., "script"/"file",  corresponding to ScriptRouterFactory.NAME, FileRouterFactory.NAME 
+    // key for router type, for e.g., "script"/"file",  corresponding to ScriptRouterFactory.NAME, FileRouterFactory.NAME
     public static final String ROUTER_KEY = "router";
 
     public static final String CLUSTER_KEY = "cluster";
@@ -624,7 +624,7 @@ public class Constants {
     public static final String QOS_PORT = "qos.port";
 
     public static final String ACCEPT_FOREIGN_IP = "qos.accept.foreign.ip";
-    
+
     public static final String HESSIAN2_REQUEST_KEY = "hessian2.request";
 
     public static final boolean DEFAULT_HESSIAN2_REQUEST = false;

dubbo-common/src/main/java/com/alibaba/dubbo/common/Version.java

@@ -23,17 +23,28 @@
 import java.net.URL;
 import java.security.CodeSource;
 import java.util.Enumeration;
+import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Map;
 import java.util.Set;
 
 /**
  * Version
  */
 public final class Version {
-
-    private static final String DEFAULT_DUBBO_VERSION = "2.0.0";
     private static final Logger logger = LoggerFactory.getLogger(Version.class);
-    private static final String VERSION = getVersion(Version.class, DEFAULT_DUBBO_VERSION);
+
+    // Dubbo RPC protocol version
+    public static final String DEFAULT_DUBBO_PROTOCOL_VERSION = "2.0.2";
+    // Dubbo implementation version, usually is jar version.
+    private static final String VERSION = getVersion(Version.class, "");
+
+    /**
+     * For protocol compatibility purpose.
+     * Because {@link #isSupportResponseAttatchment} is checked for every call, int compare expect to has higher performance than string.
+     */
+    private static final int LOWEST_VERSION_FOR_RESPONSE_ATTATCHMENT = 202; // 2.0.2
+    private static final Map<String, Integer> VERSION2INT = new HashMap<String, Integer>();
 
     static {
         // check if there's duplicated jar
@@ -43,10 +54,39 @@ public final class Version {
     private Version() {
     }
 
+    public static String getProtocolVersion() {
+        return DEFAULT_DUBBO_PROTOCOL_VERSION;
+    }
+
     public static String getVersion() {
         return VERSION;
     }
 
+    public static boolean isSupportResponseAttatchment(String version) {
+        if (version == null || version.length() == 0) {
+            return false;
+        }
+        return getIntVersion(version) >= LOWEST_VERSION_FOR_RESPONSE_ATTATCHMENT;
+    }
+
+    public static int getIntVersion(String version) {
+        Integer v = VERSION2INT.get(version);
+        if (v == null) {
+            v = parseInt(version);
+            VERSION2INT.put(version, v);
+        }
+        return v;
+    }
+
+    private static int parseInt(String version) {
+        int v = 0;
+        String[] vArr = version.split("\\.");
+        int len = vArr.length;
+        for (int i = 1; i <= len; i++) {
+            v += Integer.parseInt(vArr[len - i]) * Math.pow(10, i - 1);
+        }
+        return v;
+    }
 
     private static boolean hasResource(String path) {
         try {

dubbo-common/src/main/java/com/alibaba/dubbo/common/threadlocal/InternalThreadLocal.java

@@ -1,18 +1,17 @@
 /*
- * 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
+ * Copyright 2014 The Netty Project
  *
- *     http://www.apache.org/licenses/LICENSE-2.0
+ * The Netty Project 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.
+ * 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 com.alibaba.dubbo.common.threadlocal;