Add connection num feature (#1038)

Author: nanfeng1999

sofa-boot-project/sofa-boot-core/rpc-sofa-boot/src/main/java/com/alipay/sofa/rpc/boot/runtime/adapter/helper/ConsumerConfigHelper.java

@@ -77,6 +77,7 @@ public ConsumerConfig getConsumerConfig(Contract contract, RpcBinding binding) {
         Object callbackHandler = param.getCallbackHandler();
         String genericInterface = param.getGenericInterface();
         String loadBalancer = param.getLoadBalancer();
+        Integer connectionNum = param.getConnectionNum();
         Boolean lazy = param.getLazy();
         Boolean check = param.getCheck();
         String mockMode = param.getMockMode();
@@ -119,6 +120,9 @@ public ConsumerConfig getConsumerConfig(Contract contract, RpcBinding binding) {
         if (StringUtils.hasText(loadBalancer)) {
             consumerConfig.setLoadBalancer(loadBalancer);
         }
+        if (connectionNum != null) {
+            consumerConfig.setConnectionNum(connectionNum);
+        }
         if (lazy != null) {
             consumerConfig.setLazy(lazy);
         }

sofa-boot-project/sofa-boot-core/rpc-sofa-boot/src/main/java/com/alipay/sofa/rpc/boot/runtime/binding/RpcBindingXmlConstants.java

@@ -35,6 +35,7 @@ public class RpcBindingXmlConstants {
     public static final String TAG_TIMEOUT           = "timeout";
     public static final String TAG_ADDRESS_WAIT_TIME = "address-wait-time";
     public static final String TAG_CONNECT_TIMEOUT   = "connect.timeout";
+    public static final String TAG_CONNECT_NUM       = "connect.num";
     public static final String TAG_RETRIES           = "retries";
     public static final String TAG_TYPE              = "type";
     public static final String TAG_CALLBACK_CLASS    = "callback-class";

sofa-boot-project/sofa-boot-core/rpc-sofa-boot/src/main/java/com/alipay/sofa/rpc/boot/runtime/converter/RpcBindingConverter.java

@@ -207,6 +207,8 @@ protected void parseGlobalAttrs(Element element, RpcBindingParam param,
             .getAttribute(RpcBindingXmlConstants.TAG_ADDRESS_WAIT_TIME));
         Integer connectTimeout = SofaBootRpcParserUtil.parseInteger(element
             .getAttribute(RpcBindingXmlConstants.TAG_CONNECT_TIMEOUT));
+        Integer connectionNum = SofaBootRpcParserUtil.parseInteger(element
+            .getAttribute(RpcBindingXmlConstants.TAG_CONNECT_NUM));
         Integer retries = SofaBootRpcParserUtil.parseInteger(element
             .getAttribute(RpcBindingXmlConstants.TAG_RETRIES));
         String type = element.getAttribute(RpcBindingXmlConstants.TAG_TYPE);
@@ -273,6 +275,10 @@ protected void parseGlobalAttrs(Element element, RpcBindingParam param,
         if (StringUtils.hasText(loadBalancer)) {
             param.setLoadBalancer(loadBalancer);
         }
+
+        if (connectionNum != null) {
+            param.setConnectionNum(connectionNum);
+        }
         if (lazy != null) {
             param.setLazy(lazy);
         }
@@ -549,6 +555,7 @@ protected void convertReferenceAnnotation(RpcBindingParam bindingParam,
         if (StringUtils.hasText(callbackRef)) {
             bindingParam.setCallbackHandler(applicationContext.getBean(callbackRef));
         }
+        bindingParam.setConnectionNum(sofaReferenceBindingAnnotation.connectionNum());
         bindingParam.setLazy(sofaReferenceBindingAnnotation.lazy());
 
         String registryAlias = sofaReferenceBindingAnnotation.registry();

sofa-boot-project/sofa-boot-core/rpc-sofa-boot/src/main/java/com/alipay/sofa/rpc/boot/runtime/param/RpcBindingParam.java

@@ -43,6 +43,8 @@ public abstract class RpcBindingParam implements BindingParam {
 
     protected Integer                    connectTimeout;
 
+    protected Integer                    connectionNum;
+
     protected Integer                    retries;
 
     protected String                     type;
@@ -144,6 +146,24 @@ public void setConnectTimeout(Integer connectTimeout) {
         this.connectTimeout = connectTimeout;
     }
 
+    /**
+     * Getter method for property <code>connectionNum</code>.
+     *
+     * @return property value of connectionNum
+     */
+    public Integer getConnectionNum() {
+        return connectionNum;
+    }
+
+    /**
+     * Setter method for property <code>connectionNum</code>.
+     *
+     * @param connectionNum value to be assigned to property connectionNum
+     */
+    public void setConnectionNum(Integer connectionNum) {
+        this.connectionNum = connectionNum;
+    }
+
     /**
      * Getter method for property <code>retries</code>.
      *
@@ -488,6 +508,10 @@ public boolean equals(Object o) {
             : that.connectTimeout != null) {
             return false;
         }
+        if (connectionNum != null ? !connectionNum.equals(that.connectionNum)
+            : that.connectionNum != null) {
+            return false;
+        }
         if (retries != null ? !retries.equals(that.retries) : that.retries != null) {
             return false;
         }
@@ -557,6 +581,7 @@ public int hashCode() {
         int result = timeout != null ? timeout.hashCode() : 0;
         result = 31 * result + (addressWaitTime != null ? addressWaitTime.hashCode() : 0);
         result = 31 * result + (connectTimeout != null ? connectTimeout.hashCode() : 0);
+        result = 31 * result + (connectionNum != null ? connectionNum.hashCode() : 0);
         result = 31 * result + (retries != null ? retries.hashCode() : 0);
         result = 31 * result + (type != null ? type.hashCode() : 0);
         result = 31 * result + (callbackClass != null ? callbackClass.hashCode() : 0);

sofa-boot-project/sofa-boot-core/rpc-sofa-boot/src/test/java/com/alipay/sofa/rpc/boot/test/SofaBootRpcAllTest.java

@@ -22,6 +22,7 @@
 import com.alipay.sofa.rpc.boot.container.ConsumerConfigContainer;
 import com.alipay.sofa.rpc.boot.runtime.param.RestBindingParam;
 import com.alipay.sofa.rpc.boot.test.bean.annotation.AnnotationService;
+import com.alipay.sofa.rpc.boot.test.bean.connectionnum.ConnectionNumService;
 import com.alipay.sofa.rpc.boot.test.bean.direct.DirectService;
 import com.alipay.sofa.rpc.boot.test.bean.dubbo.DubboService;
 import com.alipay.sofa.rpc.boot.test.bean.filter.FilterService;
@@ -124,6 +125,9 @@ public class SofaBootRpcAllTest {
     @Autowired
     private LazyService                lazyServiceDubbo;
 
+    @Autowired
+    private ConnectionNumService       connectionNumService;
+
     @Autowired
     @Qualifier("sofaGreeterTripleRef")
     private SofaGreeterTriple.IGreeter sofaGreeterTripleRef;
@@ -143,6 +147,9 @@ public class SofaBootRpcAllTest {
     @SofaReference(binding = @SofaReferenceBinding(bindingType = "bolt", timeout = 1000), jvmFirst = false, uniqueId = "timeout")
     private AnnotationService          annotationConsumerTimeoutService;
 
+    @SofaReference(binding = @SofaReferenceBinding(bindingType = "rest", connectionNum = 100), jvmFirst = false, uniqueId = "connectionNum")
+    private AnnotationService          annotationConsumerConnectionNumService;
+
     @SofaClientFactory
     private ClientFactory              clientFactory;
 
@@ -291,6 +298,31 @@ public void testLoadBalancerAnnotation() throws NoSuchFieldException, IllegalAcc
         Assert.assertTrue("Found roundrobin reference", found);
     }
 
+    @Test
+    public void testConnectionNum() throws NoSuchFieldException, IllegalAccessException {
+        Field consumerConfigMapField = ConsumerConfigContainer.class
+            .getDeclaredField("consumerConfigMap");
+        consumerConfigMapField.setAccessible(true);
+        ConcurrentMap<Binding, ConsumerConfig> consumerConfigMap = (ConcurrentMap<Binding, ConsumerConfig>) consumerConfigMapField
+            .get(consumerConfigContainer);
+
+        boolean found1 = false;
+        boolean found2 = false;
+        for (ConsumerConfig consumerConfig : consumerConfigMap.values()) {
+            if ("connectionNum".equals(consumerConfig.getUniqueId())
+                && AnnotationService.class.getName().equals(consumerConfig.getInterfaceId())) {
+                found1 = true;
+                Assert.assertEquals(100, consumerConfig.getConnectionNum());
+            } else if (connectionNumService.getClass().getName()
+                .startsWith(consumerConfig.getInterfaceId())) {
+                found2 = true;
+                Assert.assertEquals(300, consumerConfig.getConnectionNum());
+            }
+        }
+        Assert.assertTrue("Found annotation reference", found1);
+        Assert.assertTrue("Found xml reference", found2);
+    }
+
     @Test
     public void testRestSwagger() throws IOException {
         HttpClient httpClient = HttpClientBuilder.create().build();

sofa-boot-project/sofa-boot-core/rpc-sofa-boot/src/test/java/com/alipay/sofa/rpc/boot/test/bean/connectionnum/ConnectionNumService.java

@@ -0,0 +1,29 @@
+/*
+ * 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 com.alipay.sofa.rpc.boot.test.bean.connectionnum;
+
+import javax.ws.rs.*;
+
+@Path("/webapi")
+@Consumes("application/json;charset=UTF-8")
+@Produces("application/json;charset=UTF-8")
+public interface ConnectionNumService {
+    @GET
+    @Path("/connectionNumService/{num}")
+    String sayConnectionNum(@PathParam("num") String string);
+
+}

sofa-boot-project/sofa-boot-core/rpc-sofa-boot/src/test/java/com/alipay/sofa/rpc/boot/test/bean/connectionnum/ConnectionNumServiceImpl.java

@@ -0,0 +1,24 @@
+/*
+ * 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 com.alipay.sofa.rpc.boot.test.bean.connectionnum;
+
+public class ConnectionNumServiceImpl implements ConnectionNumService {
+    @Override
+    public String sayConnectionNum(String string) {
+        return string;
+    }
+}

sofa-boot-project/sofa-boot-core/rpc-sofa-boot/src/test/resources/spring/test_all.xml

@@ -162,6 +162,18 @@
         </sofa:binding.dubbo>
     </sofa:reference>
 
+    <!-- connectionNum -->
+    <bean id="connectionNumServiceImpl" class="com.alipay.sofa.rpc.boot.test.bean.connectionnum.ConnectionNumServiceImpl"/>
+    <sofa:service ref="connectionNumServiceImpl" interface="com.alipay.sofa.rpc.boot.test.bean.connectionnum.ConnectionNumService">
+        <sofa:binding.rest/>
+    </sofa:service>
+
+    <sofa:reference jvm-first="false" id="connectionNumService" interface="com.alipay.sofa.rpc.boot.test.bean.connectionnum.ConnectionNumService">
+        <sofa:binding.rest>
+            <sofa:global-attrs connect.num="300"/>
+        </sofa:binding.rest>
+    </sofa:reference>
+
 
     <!-- invoke sync-->
     <bean id="sofaGreeterTripleImpl" class="com.alipay.sofa.rpc.boot.test.bean.triple.TripleGreeterImpl"/>

sofa-boot-project/sofa-boot-core/runtime-sofa-boot/src/main/java/com/alipay/sofa/runtime/api/annotation/SofaReferenceBinding.java

@@ -100,6 +100,13 @@
      */
     String registry() default "";
 
+    /**
+     * the number of long connections per ref
+     *
+     * @return
+     */
+    int connectionNum() default 1;
+
     /**
      * delay init connection
      *