Merge 2bf89b0 into 8227f0c

foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/ClientClosedEvent.java

@@ -0,0 +1,37 @@
+/*
+ * 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.servicecomb.foundation.vertx;
+
+public class ClientClosedEvent {
+  private final String address;
+
+  private final int totalConnectedCount;
+
+  public String getAddress() {
+    return address;
+  }
+
+  public int getTotalConnectedCount() {
+    return totalConnectedCount;
+  }
+
+  public ClientClosedEvent(String address, int totalConnectedCount) {
+    this.address = address;
+    this.totalConnectedCount = totalConnectedCount;
+  }
+}

foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/ClientConnectedEvent.java

@@ -0,0 +1,39 @@
+/*
+ * 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.servicecomb.foundation.vertx;
+
+import io.vertx.core.net.NetSocket;
+
+public class ClientConnectedEvent {
+  private final NetSocket netSocket;
+
+  private final int totalConnectedCount;
+
+  public NetSocket getNetSocket() {
+    return netSocket;
+  }
+
+  public int getTotalConnectedCount() {
+    return totalConnectedCount;
+  }
+
+  public ClientConnectedEvent(NetSocket netSocket, int totalConnectedCount) {
+    this.netSocket = netSocket;
+    this.totalConnectedCount = totalConnectedCount;
+  }
+}

foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/server/TcpServer.java

@@ -18,23 +18,35 @@
 package org.apache.servicecomb.foundation.vertx.server;
 
 import java.net.InetSocketAddress;
+import java.util.concurrent.atomic.AtomicInteger;
 
+import org.apache.servicecomb.foundation.common.event.EventManager;
 import org.apache.servicecomb.foundation.common.net.URIEndpointObject;
 import org.apache.servicecomb.foundation.ssl.SSLCustom;
 import org.apache.servicecomb.foundation.ssl.SSLOption;
 import org.apache.servicecomb.foundation.ssl.SSLOptionFactory;
 import org.apache.servicecomb.foundation.vertx.AsyncResultCallback;
+import org.apache.servicecomb.foundation.vertx.ClientConnectedEvent;
 import org.apache.servicecomb.foundation.vertx.VertxTLSBuilder;
 
+import com.netflix.config.DynamicPropertyFactory;
+
 import io.vertx.core.Vertx;
 import io.vertx.core.net.NetServer;
 import io.vertx.core.net.NetServerOptions;
 
 public class TcpServer {
   private URIEndpointObject endpointObject;
 
+  private final AtomicInteger connectedCounter;
+
+  private final int connectionLimit;
+
   public TcpServer(URIEndpointObject endpointObject) {
     this.endpointObject = endpointObject;
+    this.connectedCounter = new AtomicInteger(0);
+    this.connectionLimit = DynamicPropertyFactory.getInstance()
+        .getIntProperty("servicecomb.highway.server.connection-limit", Integer.MAX_VALUE).get();
   }
 
   public void init(Vertx vertx, String sslKey, AsyncResultCallback<InetSocketAddress> callback) {
@@ -57,8 +69,18 @@ public void init(Vertx vertx, String sslKey, AsyncResultCallback<InetSocketAddre
     }
 
     netServer.connectHandler(netSocket -> {
-      TcpServerConnection connection = createTcpServerConnection();
-      connection.init(netSocket);
+      if (connectedCounter.get() < connectionLimit) {
+        int connectedCount = connectedCounter.incrementAndGet();
+        if (connectedCount <= connectionLimit) {
+          TcpServerConnection connection = createTcpServerConnection();
+          connection.init(netSocket, connectedCounter);
+          EventManager.post(new ClientConnectedEvent(netSocket, connectedCount));
+          return;
+        } else {
+          connectedCounter.decrementAndGet();
+        }
+      }
+      netSocket.close();
     });
 
     InetSocketAddress socketAddress = endpointObject.getSocketAddress();

foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/server/TcpServerConnection.java

@@ -16,6 +16,10 @@
  */
 package org.apache.servicecomb.foundation.vertx.server;
 
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.servicecomb.foundation.common.event.EventManager;
+import org.apache.servicecomb.foundation.vertx.ClientClosedEvent;
 import org.apache.servicecomb.foundation.vertx.tcp.TcpConnection;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -28,7 +32,7 @@ public class TcpServerConnection extends TcpConnection {
 
   protected TcpParser splitter;
 
-  public void init(NetSocket netSocket) {
+  public void init(NetSocket netSocket, AtomicInteger connectedCounter) {
     // currently, socket always be NetSocketImpl
     this.initNetSocket((NetSocketImpl) netSocket);
 
@@ -46,6 +50,9 @@ public void init(NetSocket netSocket) {
       LOGGER.error("disconected from {}, in thread {}",
           remoteAddress,
           Thread.currentThread().getName());
+
+      int connectedCount = connectedCounter.decrementAndGet();
+      EventManager.post(new ClientClosedEvent(remoteAddress, connectedCount));
     });
 
     netSocket.handler(splitter);

foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/server/TestTcpServer.java

@@ -18,6 +18,7 @@
 package org.apache.servicecomb.foundation.vertx.server;
 
 import java.net.InetSocketAddress;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.servicecomb.foundation.common.net.URIEndpointObject;
 import org.apache.servicecomb.foundation.vertx.AsyncResultCallback;
@@ -41,8 +42,8 @@ public TcpServerForTest(URIEndpointObject endpointObject) {
     protected TcpServerConnection createTcpServerConnection() {
       return new TcpServerConnection() {
         @Override
-        public void init(NetSocket netSocket) {
-          super.init(netSocket);
+        public void init(NetSocket netSocket, AtomicInteger connectedCounter) {
+          super.init(netSocket, connectedCounter);
         }
       };
     }

foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/server/TestTcpServerConnection.java

@@ -16,6 +16,8 @@
  */
 package org.apache.servicecomb.foundation.vertx.server;
 
+import java.util.concurrent.atomic.AtomicInteger;
+
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -29,7 +31,7 @@ public void test(@Mocked NetSocketImpl netSocket) {
     connection.setProtocol("p");
     connection.setZipName("z");
 
-    connection.init(netSocket);
+    connection.init(netSocket, new AtomicInteger());
 
     Assert.assertEquals(netSocket, connection.getNetSocket());
   }

integration-tests/pom.xml

@@ -41,6 +41,7 @@
     <module>spring-zuul-tracing-tests</module>
     <module>spring-pojo-tests</module>
     <module>dynamic-config-tests</module>
+    <module>spring-pojo-connection-limit-test</module>
   </modules>
 
   <dependencyManagement>

integration-tests/spring-pojo-connection-limit-test/pom.xml

@@ -0,0 +1,68 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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/xsd/maven-4.0.0.xsd">
+  <parent>
+    <artifactId>integration-tests</artifactId>
+    <groupId>org.apache.servicecomb.tests</groupId>
+    <version>1.0.0-SNAPSHOT</version>
+  </parent>
+  <modelVersion>4.0.0</modelVersion>
+
+  <artifactId>spring-pojo-connection-limit-test</artifactId>
+  <name>Java Chassis::Integration Tests::Spring POJO Connection Limit</name>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.servicecomb.tests</groupId>
+      <artifactId>pojo-test</artifactId>
+      <version>1.0.0-SNAPSHOT</version>
+      <type>test-jar</type>
+      <exclusions>
+        <exclusion>
+          <groupId>org.apache.servicecomb.demo</groupId>
+          <artifactId>demo-signature</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-autoconfigure</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.servicecomb</groupId>
+      <artifactId>spring-boot-starter-provider</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-test</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework</groupId>
+      <artifactId>spring-test</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.hibernate</groupId>
+      <artifactId>hibernate-validator</artifactId>
+    </dependency>
+
+  </dependencies>
+
+</project>

integration-tests/spring-pojo-connection-limit-test/src/test/java/org/apache/servicecomb/demo/pojo/test/PojoSpringConnectionLimitIntegrationTest.java

@@ -0,0 +1,47 @@
+/*
+ * 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.servicecomb.demo.pojo.test;
+
+import static org.apache.servicecomb.serviceregistry.client.LocalServiceRegistryClientImpl.LOCAL_REGISTRY_FILE_KEY;
+import static org.junit.Assert.fail;
+
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+@RunWith(SpringRunner.class)
+@SpringBootTest(classes = PojoSpringMain.class)
+public class PojoSpringConnectionLimitIntegrationTest {
+  @BeforeClass
+  public static void setUpClass() {
+    System.setProperty(LOCAL_REGISTRY_FILE_KEY, "notExistJustForceLocal");
+  }
+
+  @Test
+  public void remoteHelloPojo_sayHello() {
+    try {
+      PojoService.hello.SayHello("whatever");
+      fail("connection limit failed");
+    } catch (Exception e) {
+      Assert.assertEquals("java.io.IOException: socket closed", e.getCause().toString());
+    }
+  }
+}

integration-tests/spring-pojo-connection-limit-test/src/test/java/org/apache/servicecomb/demo/pojo/test/PojoSpringMain.java

@@ -0,0 +1,31 @@
+/*
+ * 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.servicecomb.demo.pojo.test;
+
+import org.apache.servicecomb.springboot.starter.provider.EnableServiceComb;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+@EnableServiceComb
+public class PojoSpringMain {
+
+  public static void main(final String[] args) {
+    SpringApplication.run(PojoSpringMain.class, args);
+  }
+}

integration-tests/spring-pojo-connection-limit-test/src/test/resources/log4j.properties

@@ -0,0 +1,20 @@
+#
+# 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.
+#
+log4j.rootLogger=INFO, stdout
+log4j.appender.stdout=org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=%d [%-15.15t] %-5p %-30.30c{1} - %m%n