You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@servicecomb.apache.org by ya...@apache.org on 2019/11/16 06:52:13 UTC
[servicecomb-java-chassis] branch master updated (fb517e9 ->
b581ced)
This is an automated email from the ASF dual-hosted git repository.
yaohaishi pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/servicecomb-java-chassis.git.
from fb517e9 ServerListFilterExt add switch
new 77cf5df [SCB-1592]upgrade vert.x and netty.(1)update pom and fix compile errors
new bfa60c1 [SCB-1592]upgrade vert.x and netty.(2)fixed interface change
new d1e493f [SCB-1592]upgrade vert.x and netty.(3)fixed testing failures and upgrade tcnetty
new b581ced [SCB-1592]upgrade vert.x and netty.(4)update license and fix conflicts
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../demo/edge/service/EdgeDispatcher.java | 3 +-
.../service/encrypt/EncryptEdgeDispatcher.java | 3 +-
.../config/client/ConfigCenterClient.java | 44 +-
.../edge/core/DefaultEdgeDispatcher.java | 3 +-
.../edge/core/URLMappedEdgeDispatcher.java | 7 +-
.../servicecomb/edge/core/TestEdgeInvocation.java | 7 +-
.../main/java/io/vertx/core/impl/SyncContext.java | 28 +-
.../main/java/io/vertx/core/impl/SyncVertx.java | 53 -
.../main/java/io/vertx/core/impl/VertxImpl.java | 1137 --------------------
.../main/java/io/vertx/core/impl/VertxImpl.java | 1118 -------------------
.../vertx/ext/web/impl/HttpServerRequestUtils.java | 26 -
.../VertxServerRequestToHttpServletRequest.java | 4 +-
.../vertx/stream/InputStreamToReadStream.java | 3 +-
.../vertx/stream/OutputStreamToWriteStream.java | 25 +-
.../ext/web/impl/TestHttpServerRequestUtils.java | 8 -
.../foundation/vertx/TestVertxUtils.java | 6 +-
.../vertx/client/TestClientPoolManager.java | 20 +-
.../vertx/client/tcp/TestTcpClientConnection.java | 1 +
.../foundation/vertx/http/TestReadStreamPart.java | 10 +-
...TestVertxServerRequestToHttpServletRequest.java | 11 +-
...stVertxServerResponseToHttpServletResponse.java | 17 +-
.../foundation/vertx/stream/TestPumpFromPart.java | 4 +-
.../apache/servicecomb/it/edge/EdgeDispatcher.java | 3 +-
.../it/edge/encrypt/EncryptEdgeDispatcher.java | 3 +-
java-chassis-dependencies/default/pom.xml | 6 +-
java-chassis-distribution/src/release/LICENSE | 41 +-
.../metrics/core/TestVertxMetersInitializer.java | 2 +
scripts/travis.sh | 2 +-
.../client/http/WebsocketUtils.java | 46 +-
.../transport/highway/HighwayServerVerticle.java | 4 +-
.../core/http/impl/Http1xConnectionBaseEx.java | 2 +-
.../rest/client/http/TestRestClientInvocation.java | 3 +-
.../transport/rest/vertx/RestServerVerticle.java | 4 +
.../transport/rest/vertx/VertxRestDispatcher.java | 3 +-
.../vertx/accesslog/element/impl/CookieItem.java | 10 +-
.../rest/vertx/TestRestServerVerticle.java | 2 +
.../rest/vertx/TestVertxRestDispatcher.java | 55 +-
.../accesslog/element/impl/CookieItemTest.java | 22 +-
.../vertx/accesslog/impl/AccessLogHandlerTest.java | 10 +-
39 files changed, 247 insertions(+), 2509 deletions(-)
delete mode 100644 foundations/foundation-test-scaffolding/src/main/java/io/vertx/core/impl/SyncVertx.java
delete mode 100644 foundations/foundation-test-scaffolding/src/main/java/io/vertx/core/impl/VertxImpl.java
delete mode 100644 foundations/foundation-vertx/src/main/java/io/vertx/core/impl/VertxImpl.java
delete mode 100644 foundations/foundation-vertx/src/main/java/io/vertx/ext/web/impl/HttpServerRequestUtils.java
[servicecomb-java-chassis] 03/04: [SCB-1592]upgrade vert.x and
netty.(3)fixed testing failures and upgrade tcnetty
Posted by ya...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yaohaishi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/servicecomb-java-chassis.git
commit d1e493f57f222004dad15f58d17fbdc4e45c2cea
Author: liubao <bi...@qq.com>
AuthorDate: Fri Nov 15 15:36:09 2019 +0800
[SCB-1592]upgrade vert.x and netty.(3)fixed testing failures and upgrade tcnetty
---
.../org/apache/servicecomb/edge/core/TestEdgeInvocation.java | 10 ++++++++++
java-chassis-dependencies/default/pom.xml | 2 +-
scripts/travis.sh | 2 +-
.../transport/rest/vertx/TestVertxRestDispatcher.java | 10 +++++++++-
4 files changed, 21 insertions(+), 3 deletions(-)
diff --git a/edge/edge-core/src/test/java/org/apache/servicecomb/edge/core/TestEdgeInvocation.java b/edge/edge-core/src/test/java/org/apache/servicecomb/edge/core/TestEdgeInvocation.java
index 8995cbd..177c44c 100644
--- a/edge/edge-core/src/test/java/org/apache/servicecomb/edge/core/TestEdgeInvocation.java
+++ b/edge/edge-core/src/test/java/org/apache/servicecomb/edge/core/TestEdgeInvocation.java
@@ -50,6 +50,7 @@ import org.junit.Test;
import org.junit.rules.ExpectedException;
import io.vertx.core.Context;
+import io.vertx.core.Vertx;
import io.vertx.core.impl.VertxImpl;
import io.vertx.ext.web.RoutingContext;
import mockit.Deencapsulation;
@@ -61,6 +62,8 @@ import mockit.Mocked;
public class TestEdgeInvocation {
String microserviceName = "ms";
+ @Mocked Vertx vertx;
+
@Mocked
RoutingContext routingContext;
@@ -87,6 +90,13 @@ public class TestEdgeInvocation {
@Before
public void setup() {
+ new Expectations() {
+ {
+ Vertx.currentContext();
+ result = context;
+ }
+ };
+
referenceConfig.setMicroserviceVersionRule(microserviceVersionRule);
referenceConfig.setTransport("rest");
diff --git a/java-chassis-dependencies/default/pom.xml b/java-chassis-dependencies/default/pom.xml
index 2d78a4f..ac2f22c 100644
--- a/java-chassis-dependencies/default/pom.xml
+++ b/java-chassis-dependencies/default/pom.xml
@@ -102,7 +102,7 @@
<spring-cloud.version>Edgware.SR3</spring-cloud.version> <!-- Spring Zuul handling of gzip response has bug in version Dalston.SR1 -->
<swagger.version>1.5.22</swagger.version>
<swagger2markup.version>1.3.3</swagger2markup.version>
- <tcnetty.version>2.0.22.Final</tcnetty.version>
+ <tcnetty.version>2.0.26.Final</tcnetty.version>
<tec-zkclient.version>0.10</tec-zkclient.version>
<tomakehurst.version>2.6.0</tomakehurst.version>
<vertx.version>3.6.3</vertx.version>
diff --git a/scripts/travis.sh b/scripts/travis.sh
index 3d14395..7ce9978 100755
--- a/scripts/travis.sh
+++ b/scripts/travis.sh
@@ -31,7 +31,7 @@ if [ "$1" == "install" ]; then
if [ "$TAGGEDCOMMIT" == "true" ]; then
echo "Skipping the installation as it is tagged commit"
else
- mvn apache-rat:check -Dorg.slf4j.simpleLogger.log.org.apache.maven.cli.transfer.Slf4jMavenTransferListener=warn -B -Pit,samples,distribution
+ mvn apache-rat:check -Dorg.slf4j.simpleLogger.log.org.apache.maven.cli.transfer.Slf4jMavenTransferListener=warn -B -Pit,distribution
if [ $? != 0 ]; then
echo "${red}Rat check failed.${reset}"
exit 1
diff --git a/transports/transport-rest/transport-rest-vertx/src/test/java/org/apache/servicecomb/transport/rest/vertx/TestVertxRestDispatcher.java b/transports/transport-rest/transport-rest-vertx/src/test/java/org/apache/servicecomb/transport/rest/vertx/TestVertxRestDispatcher.java
index 2d8185c..b6e1861 100644
--- a/transports/transport-rest/transport-rest-vertx/src/test/java/org/apache/servicecomb/transport/rest/vertx/TestVertxRestDispatcher.java
+++ b/transports/transport-rest/transport-rest-vertx/src/test/java/org/apache/servicecomb/transport/rest/vertx/TestVertxRestDispatcher.java
@@ -47,6 +47,7 @@ import io.vertx.core.AsyncResult;
import io.vertx.core.Context;
import io.vertx.core.Handler;
import io.vertx.core.MultiMap;
+import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.http.Cookie;
import io.vertx.core.http.HttpMethod;
@@ -278,7 +279,7 @@ public class TestVertxRestDispatcher {
}
@Test
- public void onRequest(@Mocked Context context, @Mocked HttpServerRequest request,
+ public void onRequest(@Mocked Vertx vertx, @Mocked Context context, @Mocked HttpServerRequest request,
@Mocked SocketAddress socketAdrress) {
Map<String, Object> map = new HashMap<>();
RoutingContext routingContext = new MockUp<RoutingContext>() {
@@ -294,6 +295,13 @@ public class TestVertxRestDispatcher {
}
}.getMockInstance();
+ new Expectations() {
+ {
+ Vertx.currentContext();
+ result = context;
+ }
+ };
+
Deencapsulation.invoke(dispatcher, "onRequest", routingContext);
Assert.assertEquals(VertxRestInvocation.class, map.get(RestConst.REST_PRODUCER_INVOCATION).getClass());
[servicecomb-java-chassis] 04/04: [SCB-1592]upgrade vert.x and
netty.(4)update license and fix conflicts
Posted by ya...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yaohaishi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/servicecomb-java-chassis.git
commit b581ced963cbc094eaea1bb5d5ec002e183b45f7
Author: liubao <bi...@qq.com>
AuthorDate: Fri Nov 15 16:06:08 2019 +0800
[SCB-1592]upgrade vert.x and netty.(4)update license and fix conflicts
---
java-chassis-dependencies/default/pom.xml | 4 +--
java-chassis-distribution/src/release/LICENSE | 41 +++++++++++++--------------
2 files changed, 22 insertions(+), 23 deletions(-)
diff --git a/java-chassis-dependencies/default/pom.xml b/java-chassis-dependencies/default/pom.xml
index ac2f22c..5536307 100644
--- a/java-chassis-dependencies/default/pom.xml
+++ b/java-chassis-dependencies/default/pom.xml
@@ -82,7 +82,7 @@
<mock-server.version>3.10.4</mock-server.version>
<narayana.version>5.3.2.Final</narayana.version>
<netflix-commons.version>0.3.0</netflix-commons.version>
- <netty.version>4.1.31.Final</netty.version>
+ <netty.version>4.1.43.Final</netty.version>
<okhttp3.version>3.14.2</okhttp3.version>
<powermock.version>1.6.2</powermock.version>
<prometheus.version>0.6.0</prometheus.version>
@@ -105,7 +105,7 @@
<tcnetty.version>2.0.26.Final</tcnetty.version>
<tec-zkclient.version>0.10</tec-zkclient.version>
<tomakehurst.version>2.6.0</tomakehurst.version>
- <vertx.version>3.6.3</vertx.version>
+ <vertx.version>3.8.3</vertx.version>
<xstream.version>1.4.11.1</xstream.version>
<zipkin.version>2.9.3</zipkin.version>
<zipkin-reporter.version>2.7.13</zipkin-reporter.version>
diff --git a/java-chassis-distribution/src/release/LICENSE b/java-chassis-distribution/src/release/LICENSE
index 9a0d073..a4f65dd 100755
--- a/java-chassis-distribution/src/release/LICENSE
+++ b/java-chassis-distribution/src/release/LICENSE
@@ -242,9 +242,7 @@ You can find a copy of the License at licenses/LICENSE-epl-v20
* JRuby Core (org.jruby:jruby-core:9.2.6.0 - https://github.com/jruby/jruby/jruby-core)
* JRuby Lib Setup (org.jruby:jruby-stdlib:9.2.6.0 - https://github.com/jruby/jruby/jruby-stdlib)
* JRuby Main Maven Artifact (org.jruby:jruby:9.2.6.0 - https://github.com/jruby/jruby/jruby-artifacts/jruby)
-* Vert.x Core (io.vertx:vertx-core:3.6.3 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-core)
-* vertx-web (io.vertx:vertx-web:3.6.3 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-web-parent/vertx-web)
-
+* Vert.x Core (io.vertx:vertx-core:3.8.3 - https://github.com/eclipse-vertx/vert.x)
================================================================
This product bundles libraries which are licensed under the
@@ -517,21 +515,21 @@ For details, see their respective project links.
* nailgun-server (com.martiansoftware:nailgun-server:0.9.1 - http://martiansoftware.com/nailgun)
* netflix-commons-util (com.netflix.netflix-commons:netflix-commons-util:0.1.1 - https://github.com/Netflix/netflix-commons)
* netflix-statistics (com.netflix.netflix-commons:netflix-statistics:0.1.1 - https://github.com/Netflix/netflix-commons)
-* Netty/Buffer (io.netty:netty-buffer:4.1.28.Final - http://netty.io/netty-buffer/)
-* Netty/Codec/DNS (io.netty:netty-codec-dns:4.1.28.Final - http://netty.io/netty-codec-dns/)
-* Netty/Codec/HTTP (io.netty:netty-codec-http:4.1.28.Final - http://netty.io/netty-codec-http/)
-* Netty/Codec/HTTP2 (io.netty:netty-codec-http2:4.1.28.Final - http://netty.io/netty-codec-http2/)
-* Netty/Codec/Socks (io.netty:netty-codec-socks:4.1.28.Final - http://netty.io/netty-codec-socks/)
-* Netty/Codec (io.netty:netty-codec:4.1.28.Final - http://netty.io/netty-codec/)
-* Netty/Common (io.netty:netty-common:4.1.28.Final - http://netty.io/netty-common/)
-* Netty/Handler/Proxy (io.netty:netty-handler-proxy:4.1.28.Final - http://netty.io/netty-handler-proxy/)
-* Netty/Handler (io.netty:netty-handler:4.1.28.Final - http://netty.io/netty-handler/)
-* Netty/Resolver/DNS (io.netty:netty-resolver-dns:4.1.28.Final - http://netty.io/netty-resolver-dns/)
-* Netty/Resolver (io.netty:netty-resolver:4.1.28.Final - http://netty.io/netty-resolver/)
-* Netty/TomcatNative [BoringSSL - Static] (io.netty:netty-tcnative-boringssl-static:2.0.14.Final - https://github.com/netty/netty-tcnative/netty-tcnative-boringssl-static/)
-* Netty/Transport/Native/Epoll (io.netty:netty-transport-native-epoll:4.1.32.Final - http://netty.io/netty-transport-native-epoll/)
-* Netty/Transport/Native/Unix/Common (io.netty:netty-transport-native-unix-common:4.1.32.Final - http://netty.io/netty-transport-native-unix-common/)
-* Netty/Transport (io.netty:netty-transport:4.1.28.Final - http://netty.io/netty-transport/)
+* Netty/Buffer (io.netty:netty-buffer:4.1.43.Final - http://netty.io/netty-buffer/)
+* Netty/Codec/DNS (io.netty:netty-codec-dns:4.1.43.Final - http://netty.io/netty-codec-dns/)
+* Netty/Codec/HTTP (io.netty:netty-codec-http:4.1.43.Final - http://netty.io/netty-codec-http/)
+* Netty/Codec/HTTP2 (io.netty:netty-codec-http2:4.1.43.Final - http://netty.io/netty-codec-http2/)
+* Netty/Codec/Socks (io.netty:netty-codec-socks:4.1.43.Final - http://netty.io/netty-codec-socks/)
+* Netty/Codec (io.netty:netty-codec:4.1.43.Final - http://netty.io/netty-codec/)
+* Netty/Common (io.netty:netty-common:4.1.43.Final - http://netty.io/netty-common/)
+* Netty/Handler/Proxy (io.netty:netty-handler-proxy:4.1.43.Final - http://netty.io/netty-handler-proxy/)
+* Netty/Handler (io.netty:netty-handler:4.1.43.Final - http://netty.io/netty-handler/)
+* Netty/Resolver/DNS (io.netty:netty-resolver-dns:4.1.43.Final - http://netty.io/netty-resolver-dns/)
+* Netty/Resolver (io.netty:netty-resolver:4.1.43.Final - http://netty.io/netty-resolver/)
+* Netty/TomcatNative [BoringSSL - Static] (io.netty:netty-tcnative-boringssl-static:2.0.26.Final - https://github.com/netty/netty-tcnative/netty-tcnative-boringssl-static/)
+* Netty/Transport/Native/Epoll (io.netty:netty-transport-native-epoll:4.1.36.Final - http://netty.io/netty-transport-native-epoll/)
+* Netty/Transport/Native/Unix/Common (io.netty:netty-transport-native-unix-common:4.1.36.Final - http://netty.io/netty-transport-native-unix-common/)
+* Netty/Transport (io.netty:netty-transport:4.1.43.Final - http://netty.io/netty-transport/)
* nio-multipart-parser (org.synchronoss.cloud:nio-multipart-parser:1.1.0 - https://github.com/synchronoss/nio-multipart/nio-multipart-parser)
* Objenesis (org.objenesis:objenesis:2.5.1 - http://objenesis.org)
* OkHttp (com.squareup.okhttp3:okhttp:3.14.2 - https://github.com/square/okhttp/okhttp)
@@ -636,6 +634,7 @@ For details, see their respective project links.
* msg-simple (com.github.fge:msg-simple:1.1 - https://github.com/fge/msg-simple)
* uri-template (com.github.fge:uri-template:0.9 - https://github.com/fge/uri-template)
* Javassist (org.javassist:javassist:3.24.0-GA - http://www.javassist.org/)
-* Vert.x Bridge Common (io.vertx:vertx-bridge-common:3.6.3 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-bridge-common)
-* vertx-auth-common (io.vertx:vertx-auth-common:3.6.3 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-auth/vertx-auth-common)
-* vertx-web-common (io.vertx:vertx-web-common:3.6.3 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-web-parent/vertx-web-common)
+* Vert.x Bridge Common (io.vertx:vertx-bridge-common:3.8.3 - https://github.com/vert-x3/vertx-bridge-common)
+* vertx-auth-common (io.vertx:vertx-auth-common:3.8.3 - https://github.com/vert-x3/vertx-auth)
+* vertx-web-common (io.vertx:vertx-web-common:3.8.3 - https://github.com/vert-x3/vertx-web)
+* vertx-web (io.vertx:vertx-web:3.8.3 - https://github.com/vert-x3/vertx-web)
\ No newline at end of file
[servicecomb-java-chassis] 02/04: [SCB-1592]upgrade vert.x and
netty.(2)fixed interface change
Posted by ya...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yaohaishi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/servicecomb-java-chassis.git
commit bfa60c1fc2b1899b38bdae31e91ead7fdeaa841f
Author: liubao <bi...@qq.com>
AuthorDate: Fri Nov 15 11:23:56 2019 +0800
[SCB-1592]upgrade vert.x and netty.(2)fixed interface change
---
.../demo/edge/service/EdgeDispatcher.java | 3 +-
.../service/encrypt/EncryptEdgeDispatcher.java | 3 +-
.../config/client/ConfigCenterClient.java | 44 +++++++++++----------
.../edge/core/DefaultEdgeDispatcher.java | 3 +-
.../edge/core/URLMappedEdgeDispatcher.java | 7 ++--
.../apache/servicecomb/it/edge/EdgeDispatcher.java | 3 +-
.../it/edge/encrypt/EncryptEdgeDispatcher.java | 3 +-
.../metrics/core/TestVertxMetersInitializer.java | 2 +
.../client/http/WebsocketUtils.java | 46 +++++++++++-----------
.../transport/highway/HighwayServerVerticle.java | 4 +-
.../core/http/impl/Http1xConnectionBaseEx.java | 2 +-
.../rest/client/http/TestRestClientInvocation.java | 3 +-
.../transport/rest/vertx/RestServerVerticle.java | 4 ++
.../transport/rest/vertx/VertxRestDispatcher.java | 3 +-
.../vertx/accesslog/element/impl/CookieItem.java | 10 ++---
.../rest/vertx/TestRestServerVerticle.java | 2 +
.../accesslog/element/impl/CookieItemTest.java | 22 +++++------
.../vertx/accesslog/impl/AccessLogHandlerTest.java | 10 ++---
18 files changed, 91 insertions(+), 83 deletions(-)
diff --git a/demo/demo-edge/edge-service/src/main/java/org/apache/servicecomb/demo/edge/service/EdgeDispatcher.java b/demo/demo-edge/edge-service/src/main/java/org/apache/servicecomb/demo/edge/service/EdgeDispatcher.java
index d5643d3..0eef0ba 100644
--- a/demo/demo-edge/edge-service/src/main/java/org/apache/servicecomb/demo/edge/service/EdgeDispatcher.java
+++ b/demo/demo-edge/edge-service/src/main/java/org/apache/servicecomb/demo/edge/service/EdgeDispatcher.java
@@ -25,7 +25,6 @@ import org.apache.servicecomb.edge.core.EdgeInvocation;
import io.vertx.ext.web.Router;
import io.vertx.ext.web.RoutingContext;
-import io.vertx.ext.web.handler.CookieHandler;
public class EdgeDispatcher extends AbstractEdgeDispatcher {
private CompatiblePathVersionMapper versionMapper = new CompatiblePathVersionMapper();
@@ -38,7 +37,7 @@ public class EdgeDispatcher extends AbstractEdgeDispatcher {
@Override
public void init(Router router) {
String regex = "/api/([^\\\\/]+)/([^\\\\/]+)/(.*)";
- router.routeWithRegex(regex).handler(CookieHandler.create());
+ // cookies handler are enabled by default start from 3.8.3
router.routeWithRegex(regex).handler(createBodyHandler());
router.routeWithRegex(regex).failureHandler(this::onFailure).handler(this::onRequest);
}
diff --git a/demo/demo-edge/edge-service/src/main/java/org/apache/servicecomb/demo/edge/service/encrypt/EncryptEdgeDispatcher.java b/demo/demo-edge/edge-service/src/main/java/org/apache/servicecomb/demo/edge/service/encrypt/EncryptEdgeDispatcher.java
index 5c8a9a5..c4900a6 100644
--- a/demo/demo-edge/edge-service/src/main/java/org/apache/servicecomb/demo/edge/service/encrypt/EncryptEdgeDispatcher.java
+++ b/demo/demo-edge/edge-service/src/main/java/org/apache/servicecomb/demo/edge/service/encrypt/EncryptEdgeDispatcher.java
@@ -30,7 +30,6 @@ import org.slf4j.LoggerFactory;
import io.vertx.core.http.HttpServerRequest;
import io.vertx.ext.web.Router;
import io.vertx.ext.web.RoutingContext;
-import io.vertx.ext.web.handler.CookieHandler;
public class EncryptEdgeDispatcher extends AbstractEdgeDispatcher {
private static final Logger LOGGER = LoggerFactory.getLogger(EncryptEdgeDispatcher.class);
@@ -50,7 +49,7 @@ public class EncryptEdgeDispatcher extends AbstractEdgeDispatcher {
public void init(Router router) {
{
String regex = "/" + prefix + "/([^\\\\/]+)/([^\\\\/]+)/(.*)";
- router.routeWithRegex(regex).handler(CookieHandler.create());
+ // cookies handler are enabled by default start from 3.8.3
router.routeWithRegex(regex).handler(createBodyHandler());
router.routeWithRegex(regex).failureHandler(this::onFailure).handler(this::onRequest);
}
diff --git a/dynamic-config/config-cc/src/main/java/org/apache/servicecomb/config/client/ConfigCenterClient.java b/dynamic-config/config-cc/src/main/java/org/apache/servicecomb/config/client/ConfigCenterClient.java
index ac0f88a..8ad91bd 100644
--- a/dynamic-config/config-cc/src/main/java/org/apache/servicecomb/config/client/ConfigCenterClient.java
+++ b/dynamic-config/config-cc/src/main/java/org/apache/servicecomb/config/client/ConfigCenterClient.java
@@ -70,6 +70,7 @@ import io.vertx.core.http.CaseInsensitiveHeaders;
import io.vertx.core.http.HttpClientOptions;
import io.vertx.core.http.HttpClientRequest;
import io.vertx.core.http.WebSocket;
+import io.vertx.core.http.WebSocketConnectOptions;
import io.vertx.core.http.impl.FrameType;
import io.vertx.core.http.impl.ws.WebSocketFrameImpl;
import io.vertx.core.net.ProxyOptions;
@@ -285,28 +286,35 @@ public class ConfigCenterClient {
Map<String, String> authHeaders = new HashMap<>();
authHeaderProviders.forEach(provider -> authHeaders.putAll(provider.getSignAuthHeaders(
createSignRequest(null, configCenter + url, headers, null))));
-
- client.websocket(refreshPort,
- ipPort.getHostOrIp(),
- url,
- new CaseInsensitiveHeaders().addAll(headers)
- .addAll(authHeaders),
- ws -> {
- ws.exceptionHandler(e -> {
+ WebSocketConnectOptions options = new WebSocketConnectOptions();
+ options.setHost(ipPort.getHostOrIp()).setPort(refreshPort).setURI(url)
+ .setHeaders(new CaseInsensitiveHeaders().addAll(headers)
+ .addAll(authHeaders));
+ client.webSocket(options, asyncResult -> {
+ if (asyncResult.failed()) {
+ LOGGER.error(
+ "watcher connect to config center {} refresh port {} failed. Error message is [{}]",
+ configCenter,
+ refreshPort,
+ asyncResult.cause().getMessage());
+ waiter.countDown();
+ } else {
+ {
+ asyncResult.result().exceptionHandler(e -> {
LOGGER.error("watch config read fail", e);
stopHeartBeatThread();
isWatching = false;
});
- ws.closeHandler(v -> {
+ asyncResult.result().closeHandler(v -> {
LOGGER.warn("watching config connection is closed accidentally");
stopHeartBeatThread();
isWatching = false;
});
- ws.pongHandler(pong -> {
+ asyncResult.result().pongHandler(pong -> {
// ignore, just prevent NPE.
});
- ws.frameHandler(frame -> {
+ asyncResult.result().frameHandler(frame -> {
Buffer action = frame.binaryData();
LOGGER.info("watching config recieved {}", action);
Map<String, Object> mAction = action.toJsonObject().getMap();
@@ -319,18 +327,12 @@ public class ConfigCenterClient {
parseConfigUtils.refreshConfigItemsIncremental(mAction);
}
});
- startHeartBeatThread(ws);
+ startHeartBeatThread(asyncResult.result());
isWatching = true;
waiter.countDown();
- },
- e -> {
- LOGGER.error(
- "watcher connect to config center {} refresh port {} failed. Error message is [{}]",
- configCenter,
- refreshPort,
- e.getMessage());
- waiter.countDown();
- });
+ }
+ }
+ });
});
waiter.await();
}
diff --git a/edge/edge-core/src/main/java/org/apache/servicecomb/edge/core/DefaultEdgeDispatcher.java b/edge/edge-core/src/main/java/org/apache/servicecomb/edge/core/DefaultEdgeDispatcher.java
index b6eadfe..0d195c6 100644
--- a/edge/edge-core/src/main/java/org/apache/servicecomb/edge/core/DefaultEdgeDispatcher.java
+++ b/edge/edge-core/src/main/java/org/apache/servicecomb/edge/core/DefaultEdgeDispatcher.java
@@ -23,7 +23,6 @@ import com.netflix.config.DynamicPropertyFactory;
import io.vertx.ext.web.Router;
import io.vertx.ext.web.RoutingContext;
-import io.vertx.ext.web.handler.CookieHandler;
/**
* Provide an easy mapping dispatcher that starts with a common prefix pattern.
@@ -66,7 +65,7 @@ public class DefaultEdgeDispatcher extends AbstractEdgeDispatcher {
} else {
regex = "/" + prefix + "/([^\\\\/]+)/(.*)";
}
- router.routeWithRegex(regex).handler(CookieHandler.create());
+ // cookies handler are enabled by default start from 3.8.3
router.routeWithRegex(regex).handler(createBodyHandler());
router.routeWithRegex(regex).failureHandler(this::onFailure).handler(this::onRequest);
}
diff --git a/edge/edge-core/src/main/java/org/apache/servicecomb/edge/core/URLMappedEdgeDispatcher.java b/edge/edge-core/src/main/java/org/apache/servicecomb/edge/core/URLMappedEdgeDispatcher.java
index fe30068..c734b40 100644
--- a/edge/edge-core/src/main/java/org/apache/servicecomb/edge/core/URLMappedEdgeDispatcher.java
+++ b/edge/edge-core/src/main/java/org/apache/servicecomb/edge/core/URLMappedEdgeDispatcher.java
@@ -31,7 +31,6 @@ import com.netflix.config.DynamicPropertyFactory;
import io.vertx.ext.web.Router;
import io.vertx.ext.web.RoutingContext;
-import io.vertx.ext.web.handler.CookieHandler;
/**
* Provide a URL mapping based dispatcher. Users configure witch URL patterns dispatch to a target service.
@@ -68,7 +67,7 @@ public class URLMappedEdgeDispatcher extends AbstractEdgeDispatcher {
private Map<String, ConfigurationItem> configurations = new HashMap<>();
public URLMappedEdgeDispatcher() {
- if(this.enabled()) {
+ if (this.enabled()) {
loadConfigurations();
}
}
@@ -85,7 +84,7 @@ public class URLMappedEdgeDispatcher extends AbstractEdgeDispatcher {
@Override
public void init(Router router) {
- router.routeWithRegex(PATTERN_ANY).handler(CookieHandler.create());
+ // cookies handler are enabled by default start from 3.8.3
router.routeWithRegex(PATTERN_ANY).handler(createBodyHandler());
router.routeWithRegex(PATTERN_ANY).failureHandler(this::onFailure).handler(this::onRequest);
}
@@ -138,7 +137,7 @@ public class URLMappedEdgeDispatcher extends AbstractEdgeDispatcher {
configurations.entrySet().forEach(stringConfigurationItemEntry -> {
ConfigurationItem item = stringConfigurationItemEntry.getValue();
LOG.info("config item: key=" + stringConfigurationItemEntry.getKey() + ";pattern=" + item.stringPattern
- + ";service=" + item.microserviceName + ";versionRule=" + item.versionRule);
+ + ";service=" + item.microserviceName + ";versionRule=" + item.versionRule);
});
}
diff --git a/integration-tests/it-edge/src/main/java/org/apache/servicecomb/it/edge/EdgeDispatcher.java b/integration-tests/it-edge/src/main/java/org/apache/servicecomb/it/edge/EdgeDispatcher.java
index 7770c7c..4afda81 100644
--- a/integration-tests/it-edge/src/main/java/org/apache/servicecomb/it/edge/EdgeDispatcher.java
+++ b/integration-tests/it-edge/src/main/java/org/apache/servicecomb/it/edge/EdgeDispatcher.java
@@ -25,7 +25,6 @@ import org.apache.servicecomb.edge.core.EdgeInvocation;
import io.vertx.ext.web.Router;
import io.vertx.ext.web.RoutingContext;
-import io.vertx.ext.web.handler.CookieHandler;
public class EdgeDispatcher extends AbstractEdgeDispatcher {
private CompatiblePathVersionMapper versionMapper = new CompatiblePathVersionMapper();
@@ -38,7 +37,7 @@ public class EdgeDispatcher extends AbstractEdgeDispatcher {
@Override
public void init(Router router) {
String regex = "/api/([^\\\\/]+)/([^\\\\/]+)/(.*)";
- router.routeWithRegex(regex).handler(CookieHandler.create());
+ // cookies handler are enabled by default start from 3.8.3
router.routeWithRegex(regex).handler(createBodyHandler());
router.routeWithRegex(regex).failureHandler(this::onFailure).handler(this::onRequest);
}
diff --git a/integration-tests/it-edge/src/main/java/org/apache/servicecomb/it/edge/encrypt/EncryptEdgeDispatcher.java b/integration-tests/it-edge/src/main/java/org/apache/servicecomb/it/edge/encrypt/EncryptEdgeDispatcher.java
index 6c2bca7..7b48f04 100644
--- a/integration-tests/it-edge/src/main/java/org/apache/servicecomb/it/edge/encrypt/EncryptEdgeDispatcher.java
+++ b/integration-tests/it-edge/src/main/java/org/apache/servicecomb/it/edge/encrypt/EncryptEdgeDispatcher.java
@@ -30,7 +30,6 @@ import org.slf4j.LoggerFactory;
import io.vertx.core.http.HttpServerRequest;
import io.vertx.ext.web.Router;
import io.vertx.ext.web.RoutingContext;
-import io.vertx.ext.web.handler.CookieHandler;
public class EncryptEdgeDispatcher extends AbstractEdgeDispatcher {
private static final Logger LOGGER = LoggerFactory.getLogger(EncryptEdgeDispatcher.class);
@@ -50,7 +49,7 @@ public class EncryptEdgeDispatcher extends AbstractEdgeDispatcher {
public void init(Router router) {
{
String regex = "/" + prefix + "/([^\\\\/]+)/([^\\\\/]+)/(.*)";
- router.routeWithRegex(regex).handler(CookieHandler.create());
+ // cookies handler are enabled by default start from 3.8.3
router.routeWithRegex(regex).handler(createBodyHandler());
router.routeWithRegex(regex).failureHandler(this::onFailure).handler(this::onRequest);
}
diff --git a/metrics/metrics-core/src/test/java/org/apache/servicecomb/metrics/core/TestVertxMetersInitializer.java b/metrics/metrics-core/src/test/java/org/apache/servicecomb/metrics/core/TestVertxMetersInitializer.java
index 536cc47..17850ff 100644
--- a/metrics/metrics-core/src/test/java/org/apache/servicecomb/metrics/core/TestVertxMetersInitializer.java
+++ b/metrics/metrics-core/src/test/java/org/apache/servicecomb/metrics/core/TestVertxMetersInitializer.java
@@ -67,6 +67,8 @@ public class TestVertxMetersInitializer {
public static class TestServerVerticle extends AbstractVerticle {
@Override
+ @SuppressWarnings("deprecation")
+ // TODO: vert.x 3.8.3 does not update startListen to promise, so we keep use deprecated API now. update in newer version.
public void start(Future<Void> startFuture) {
Router mainRouter = Router.router(vertx);
mainRouter.route("/").handler(context -> {
diff --git a/service-registry/src/main/java/org/apache/servicecomb/serviceregistry/client/http/WebsocketUtils.java b/service-registry/src/main/java/org/apache/servicecomb/serviceregistry/client/http/WebsocketUtils.java
index 39db844..1536058 100644
--- a/service-registry/src/main/java/org/apache/servicecomb/serviceregistry/client/http/WebsocketUtils.java
+++ b/service-registry/src/main/java/org/apache/servicecomb/serviceregistry/client/http/WebsocketUtils.java
@@ -27,6 +27,7 @@ import org.slf4j.LoggerFactory;
import io.vertx.core.Handler;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.http.HttpMethod;
+import io.vertx.core.http.WebSocketConnectOptions;
/**
* Created by on 2017/4/28.
@@ -42,28 +43,29 @@ public final class WebsocketUtils {
Handler<Throwable> onConnectFailed) {
HttpClientWithContext vertxHttpClient = WebsocketClientPool.INSTANCE.getClient();
vertxHttpClient.runOnContext(client -> {
- client.websocket(ipPort.getPort(),
- ipPort.getHostOrIp(),
- url,
- RestUtils.getDefaultHeaders().addAll(RestUtils.getSignAuthHeaders(
- RestUtils.createSignRequest(HttpMethod.GET.name(), ipPort, new RequestParam(), url, new HashMap<>()))),
- ws -> {
- onOpen.handle(null);
-
- ws.exceptionHandler(v -> {
- onException.handle(v);
- try {
- ws.close();
- } catch (Exception err) {
- LOGGER.error("ws close error.", err);
- }
- });
- ws.closeHandler(v -> {
- onClose.handle(v);
- });
- ws.handler(onMessage);
- },
- onConnectFailed);
+ WebSocketConnectOptions options = new WebSocketConnectOptions();
+ options.setHost(ipPort.getHostOrIp()).setPort(ipPort.getPort()).setURI(url)
+ .setHeaders(RestUtils.getDefaultHeaders().addAll(RestUtils.getSignAuthHeaders(
+ RestUtils.createSignRequest(HttpMethod.GET.name(), ipPort, new RequestParam(), url, new HashMap<>()))));
+ client.webSocket(options, asyncResult -> {
+ if (asyncResult.failed()) {
+ onConnectFailed.handle(asyncResult.cause());
+ } else {
+ onOpen.handle(null);
+ asyncResult.result().exceptionHandler(v -> {
+ onException.handle(v);
+ try {
+ asyncResult.result().close();
+ } catch (Exception err) {
+ LOGGER.error("ws close error.", err);
+ }
+ });
+ asyncResult.result().closeHandler(v -> {
+ onClose.handle(v);
+ });
+ asyncResult.result().handler(onMessage);
+ }
+ });
});
}
}
diff --git a/transports/transport-highway/src/main/java/org/apache/servicecomb/transport/highway/HighwayServerVerticle.java b/transports/transport-highway/src/main/java/org/apache/servicecomb/transport/highway/HighwayServerVerticle.java
index 3d6efc4..8104ecc 100644
--- a/transports/transport-highway/src/main/java/org/apache/servicecomb/transport/highway/HighwayServerVerticle.java
+++ b/transports/transport-highway/src/main/java/org/apache/servicecomb/transport/highway/HighwayServerVerticle.java
@@ -59,8 +59,10 @@ public class HighwayServerVerticle extends AbstractVerticle {
}
}
+ @SuppressWarnings("deprecation")
+ // TODO: vert.x 3.8.3 does not update startListen to promise, so we keep use deprecated API now. update in newer version.
protected void startListen(Future<Void> startFuture) {
- // 如果本地未配置地址，则表示不必监听，只需要作为客户端使用即可
+ // if listen address is not provided, do not fail and maybe a consumer service.
if (endpointObject == null) {
LOGGER.warn("highway listen address is not configured, will not listen.");
startFuture.complete();
diff --git a/transports/transport-rest/transport-rest-client/src/test/java/io/vertx/core/http/impl/Http1xConnectionBaseEx.java b/transports/transport-rest/transport-rest-client/src/test/java/io/vertx/core/http/impl/Http1xConnectionBaseEx.java
index 524647f..09dab11 100644
--- a/transports/transport-rest/transport-rest-client/src/test/java/io/vertx/core/http/impl/Http1xConnectionBaseEx.java
+++ b/transports/transport-rest/transport-rest-client/src/test/java/io/vertx/core/http/impl/Http1xConnectionBaseEx.java
@@ -22,7 +22,7 @@ import io.vertx.core.impl.ContextInternal;
import io.vertx.core.impl.VertxInternal;
import io.vertx.core.spi.metrics.NetworkMetrics;
-public abstract class Http1xConnectionBaseEx extends Http1xConnectionBase{
+public abstract class Http1xConnectionBaseEx<S extends WebSocketImplBase<S>> extends Http1xConnectionBase<S>{
public Http1xConnectionBaseEx(VertxInternal vertx, ChannelHandlerContext chctx,
ContextInternal context) {
super(vertx, chctx, context);
diff --git a/transports/transport-rest/transport-rest-client/src/test/java/org/apache/servicecomb/transport/rest/client/http/TestRestClientInvocation.java b/transports/transport-rest/transport-rest-client/src/test/java/org/apache/servicecomb/transport/rest/client/http/TestRestClientInvocation.java
index e00c3e3..c690091 100644
--- a/transports/transport-rest/transport-rest-client/src/test/java/org/apache/servicecomb/transport/rest/client/http/TestRestClientInvocation.java
+++ b/transports/transport-rest/transport-rest-client/src/test/java/org/apache/servicecomb/transport/rest/client/http/TestRestClientInvocation.java
@@ -74,6 +74,7 @@ import io.vertx.core.http.HttpClientRequest;
import io.vertx.core.http.HttpClientResponse;
import io.vertx.core.http.RequestOptions;
import io.vertx.core.http.impl.Http1xConnectionBaseEx;
+import io.vertx.core.http.impl.WebSocketImpl;
import io.vertx.core.http.impl.headers.VertxHttpHeaders;
import io.vertx.core.net.NetSocket;
import io.vertx.core.net.SocketAddress;
@@ -128,7 +129,7 @@ public class TestRestClientInvocation {
static long nanoTime = 123;
@Mocked
- Http1xConnectionBaseEx connectionBase;
+ Http1xConnectionBaseEx<WebSocketImpl> connectionBase;
OperationConfig operationConfig = new OperationConfig();
diff --git a/transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/RestServerVerticle.java b/transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/RestServerVerticle.java
index a07c5a6..1047464 100644
--- a/transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/RestServerVerticle.java
+++ b/transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/RestServerVerticle.java
@@ -76,6 +76,8 @@ public class RestServerVerticle extends AbstractVerticle {
this.endpointObject = (URIEndpointObject) endpoint.getAddress();
}
+ @SuppressWarnings("deprecation")
+ // TODO: vert.x 3.8.3 does not update startListen to promise, so we keep use deprecated API now. update in newer version.
@Override
public void start(Future<Void> startFuture) throws Exception {
try {
@@ -212,6 +214,8 @@ public class RestServerVerticle extends AbstractVerticle {
}
}
+ @SuppressWarnings("deprecation")
+ // TODO: vert.x 3.8.3 does not update startListen to promise, so we keep use deprecated API now. update in newer version.
private void startListen(HttpServer server, Future<Void> startFuture) {
server.listen(endpointObject.getPort(), endpointObject.getHostOrIp(), ar -> {
if (ar.succeeded()) {
diff --git a/transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/VertxRestDispatcher.java b/transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/VertxRestDispatcher.java
index b90bc78..763a360 100644
--- a/transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/VertxRestDispatcher.java
+++ b/transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/VertxRestDispatcher.java
@@ -42,7 +42,6 @@ import io.netty.handler.codec.http.multipart.HttpPostRequestDecoder.ErrorDataDec
import io.vertx.core.json.JsonObject;
import io.vertx.ext.web.Router;
import io.vertx.ext.web.RoutingContext;
-import io.vertx.ext.web.handler.CookieHandler;
public class VertxRestDispatcher extends AbstractVertxHttpDispatcher {
private static final Logger LOGGER = LoggerFactory.getLogger(VertxRestDispatcher.class);
@@ -63,7 +62,7 @@ public class VertxRestDispatcher extends AbstractVertxHttpDispatcher {
@Override
public void init(Router router) {
- router.route().handler(CookieHandler.create());
+ // cookies handler are enabled by default start from 3.8.3
router.route().handler(createBodyHandler());
router.route().failureHandler(this::failureHandler).handler(this::onRequest);
}
diff --git a/transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/accesslog/element/impl/CookieItem.java b/transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/accesslog/element/impl/CookieItem.java
index aa20a02..331ef60 100644
--- a/transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/accesslog/element/impl/CookieItem.java
+++ b/transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/accesslog/element/impl/CookieItem.java
@@ -17,12 +17,12 @@
package org.apache.servicecomb.transport.rest.vertx.accesslog.element.impl;
-import java.util.Set;
+import java.util.Map;
import org.apache.servicecomb.transport.rest.vertx.accesslog.AccessLogParam;
import org.apache.servicecomb.transport.rest.vertx.accesslog.element.AccessLogItem;
-import io.vertx.ext.web.Cookie;
+import io.vertx.core.http.Cookie;
import io.vertx.ext.web.RoutingContext;
public class CookieItem implements AccessLogItem<RoutingContext> {
@@ -37,13 +37,13 @@ public class CookieItem implements AccessLogItem<RoutingContext> {
@Override
public String getFormattedItem(AccessLogParam<RoutingContext> accessLogParam) {
- Set<Cookie> cookieSet = accessLogParam.getContextData().cookies();
- if (null == cookieSet) {
+ Map<String, Cookie> cookieMap = accessLogParam.getContextData().cookieMap();
+ if (null == cookieMap) {
return RESULT_NOT_FOUND;
}
String result = null;
- for (Cookie cookie : cookieSet) {
+ for (Cookie cookie : cookieMap.values()) {
if (varName.equals(cookie.getName())) {
result = cookie.getValue();
}
diff --git a/transports/transport-rest/transport-rest-vertx/src/test/java/org/apache/servicecomb/transport/rest/vertx/TestRestServerVerticle.java b/transports/transport-rest/transport-rest-vertx/src/test/java/org/apache/servicecomb/transport/rest/vertx/TestRestServerVerticle.java
index c9846cf..5808aa5 100644
--- a/transports/transport-rest/transport-rest-vertx/src/test/java/org/apache/servicecomb/transport/rest/vertx/TestRestServerVerticle.java
+++ b/transports/transport-rest/transport-rest-vertx/src/test/java/org/apache/servicecomb/transport/rest/vertx/TestRestServerVerticle.java
@@ -61,6 +61,8 @@ public class TestRestServerVerticle {
Future<Void> startFuture = null;
@Before
+ @SuppressWarnings("deprecation")
+ // TODO: vert.x 3.8.3 does not update startListen to promise, so we keep use deprecated API now. update in newer version.
public void setUp() {
instance = new RestServerVerticle();
startFuture = Future.future();
diff --git a/transports/transport-rest/transport-rest-vertx/src/test/java/org/apache/servicecomb/transport/rest/vertx/accesslog/element/impl/CookieItemTest.java b/transports/transport-rest/transport-rest-vertx/src/test/java/org/apache/servicecomb/transport/rest/vertx/accesslog/element/impl/CookieItemTest.java
index d8977aa..68db0db 100644
--- a/transports/transport-rest/transport-rest-vertx/src/test/java/org/apache/servicecomb/transport/rest/vertx/accesslog/element/impl/CookieItemTest.java
+++ b/transports/transport-rest/transport-rest-vertx/src/test/java/org/apache/servicecomb/transport/rest/vertx/accesslog/element/impl/CookieItemTest.java
@@ -17,14 +17,14 @@
package org.apache.servicecomb.transport.rest.vertx.accesslog.element.impl;
-import java.util.HashSet;
+import java.util.HashMap;
import org.apache.servicecomb.transport.rest.vertx.accesslog.AccessLogParam;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
-import io.vertx.ext.web.Cookie;
+import io.vertx.core.http.Cookie;
import io.vertx.ext.web.RoutingContext;
import io.vertx.ext.web.impl.CookieImpl;
@@ -38,13 +38,13 @@ public class CookieItemTest {
public void getFormattedElement() {
AccessLogParam<RoutingContext> param = new AccessLogParam<>();
RoutingContext mockContext = Mockito.mock(RoutingContext.class);
- HashSet<Cookie> cookieSet = new HashSet<>();
+ HashMap<String, Cookie> cookieSet = new HashMap<>();
String cookieValue = "cookieValue";
CookieImpl cookie = new CookieImpl(COOKIE_NAME, cookieValue);
- cookieSet.add(cookie);
+ cookieSet.put(cookie.getName(), cookie);
Mockito.when(mockContext.cookieCount()).thenReturn(1);
- Mockito.when(mockContext.cookies()).thenReturn(cookieSet);
+ Mockito.when(mockContext.cookieMap()).thenReturn(cookieSet);
param.setContextData(mockContext);
String result = ELEMENT.getFormattedItem(param);
@@ -56,10 +56,10 @@ public class CookieItemTest {
public void getFormattedElementOnCookieCountIsZero() {
AccessLogParam<RoutingContext> param = new AccessLogParam<>();
RoutingContext mockContext = Mockito.mock(RoutingContext.class);
- HashSet<Cookie> cookieSet = new HashSet<>();
+ HashMap<String, Cookie> cookieSet = new HashMap<>();
Mockito.when(mockContext.cookieCount()).thenReturn(0);
- Mockito.when(mockContext.cookies()).thenReturn(cookieSet);
+ Mockito.when(mockContext.cookieMap()).thenReturn(cookieSet);
param.setContextData(mockContext);
String result = ELEMENT.getFormattedItem(param);
@@ -73,7 +73,7 @@ public class CookieItemTest {
RoutingContext mockContext = Mockito.mock(RoutingContext.class);
Mockito.when(mockContext.cookieCount()).thenReturn(1);
- Mockito.when(mockContext.cookies()).thenReturn(null);
+ Mockito.when(mockContext.cookieMap()).thenReturn(null);
param.setContextData(mockContext);
String result = ELEMENT.getFormattedItem(param);
@@ -85,13 +85,13 @@ public class CookieItemTest {
public void getFormattedElementOnNotFound() {
AccessLogParam<RoutingContext> param = new AccessLogParam<>();
RoutingContext mockContext = Mockito.mock(RoutingContext.class);
- HashSet<Cookie> cookieSet = new HashSet<>();
+ HashMap<String, Cookie> cookieSet = new HashMap<>();
String cookieValue = "cookieValue";
CookieImpl cookie = new CookieImpl("anotherCookieName", cookieValue);
- cookieSet.add(cookie);
+ cookieSet.put(cookie.getName(), cookie);
Mockito.when(mockContext.cookieCount()).thenReturn(1);
- Mockito.when(mockContext.cookies()).thenReturn(cookieSet);
+ Mockito.when(mockContext.cookieMap()).thenReturn(cookieSet);
param.setContextData(mockContext);
String result = ELEMENT.getFormattedItem(param);
diff --git a/transports/transport-rest/transport-rest-vertx/src/test/java/org/apache/servicecomb/transport/rest/vertx/accesslog/impl/AccessLogHandlerTest.java b/transports/transport-rest/transport-rest-vertx/src/test/java/org/apache/servicecomb/transport/rest/vertx/accesslog/impl/AccessLogHandlerTest.java
index 0ef62e5..d9b788c 100644
--- a/transports/transport-rest/transport-rest-vertx/src/test/java/org/apache/servicecomb/transport/rest/vertx/accesslog/impl/AccessLogHandlerTest.java
+++ b/transports/transport-rest/transport-rest-vertx/src/test/java/org/apache/servicecomb/transport/rest/vertx/accesslog/impl/AccessLogHandlerTest.java
@@ -17,7 +17,7 @@
package org.apache.servicecomb.transport.rest.vertx.accesslog.impl;
-import java.util.HashSet;
+import java.util.HashMap;
import java.util.concurrent.TimeUnit;
import javax.xml.ws.Holder;
@@ -31,10 +31,10 @@ import org.junit.Test;
import org.mockito.Mockito;
import io.vertx.core.Handler;
+import io.vertx.core.http.Cookie;
import io.vertx.core.http.HttpServerRequest;
import io.vertx.core.http.HttpServerResponse;
import io.vertx.core.net.SocketAddress;
-import io.vertx.ext.web.Cookie;
import io.vertx.ext.web.RoutingContext;
import mockit.Mock;
import mockit.MockUp;
@@ -59,7 +59,7 @@ public class AccessLogHandlerTest {
@Test
public void testHandle() {
RoutingContext routingContext = Mockito.mock(RoutingContext.class);
- HashSet<Cookie> cookies = new HashSet<>();
+ HashMap<String, Cookie> cookies = new HashMap<>();
Cookie cookie = Mockito.mock(Cookie.class);
HttpServerResponse httpServerResponse = new MockUp<HttpServerResponse>() {
@Mock
@@ -93,10 +93,10 @@ public class AccessLogHandlerTest {
return 123L;
}
};
- cookies.add(cookie);
+ cookies.put(cookie.getName(), cookie);
Mockito.when(cookie.getName()).thenReturn("cookie-name");
Mockito.when(cookie.getValue()).thenReturn("cookie-value");
- Mockito.when(routingContext.cookies()).thenReturn(cookies);
+ Mockito.when(routingContext.cookieMap()).thenReturn(cookies);
Mockito.when(routingContext.response()).thenReturn(httpServerResponse);
Mockito.when(routingContext.request()).thenReturn(httpServerRequest);
Mockito.when(httpServerRequest.remoteAddress()).thenReturn(remoteSocketAddress);
[servicecomb-java-chassis] 01/04: [SCB-1592]upgrade vert.x and
netty.(1)update pom and fix compile errors
Posted by ya...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yaohaishi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/servicecomb-java-chassis.git
commit 77cf5df1f22b0fe8de69ba24dbc5381171fa6b0c
Author: liubao <bi...@qq.com>
AuthorDate: Fri Nov 15 09:24:30 2019 +0800
[SCB-1592]upgrade vert.x and netty.(1)update pom and fix compile errors
---
.../servicecomb/edge/core/TestEdgeInvocation.java | 7 -
.../main/java/io/vertx/core/impl/SyncContext.java | 28 +-
.../main/java/io/vertx/core/impl/SyncVertx.java | 53 -
.../main/java/io/vertx/core/impl/VertxImpl.java | 1137 --------------------
.../main/java/io/vertx/core/impl/VertxImpl.java | 1118 -------------------
.../vertx/ext/web/impl/HttpServerRequestUtils.java | 26 -
.../VertxServerRequestToHttpServletRequest.java | 4 +-
.../vertx/stream/InputStreamToReadStream.java | 3 +-
.../vertx/stream/OutputStreamToWriteStream.java | 25 +-
.../ext/web/impl/TestHttpServerRequestUtils.java | 8 -
.../foundation/vertx/TestVertxUtils.java | 6 +-
.../vertx/client/TestClientPoolManager.java | 20 +-
.../vertx/client/tcp/TestTcpClientConnection.java | 1 +
.../foundation/vertx/http/TestReadStreamPart.java | 10 +-
...TestVertxServerRequestToHttpServletRequest.java | 11 +-
...stVertxServerResponseToHttpServletResponse.java | 17 +-
.../foundation/vertx/stream/TestPumpFromPart.java | 4 +-
.../rest/vertx/TestVertxRestDispatcher.java | 53 +-
18 files changed, 122 insertions(+), 2409 deletions(-)
diff --git a/edge/edge-core/src/test/java/org/apache/servicecomb/edge/core/TestEdgeInvocation.java b/edge/edge-core/src/test/java/org/apache/servicecomb/edge/core/TestEdgeInvocation.java
index a566085..8995cbd 100644
--- a/edge/edge-core/src/test/java/org/apache/servicecomb/edge/core/TestEdgeInvocation.java
+++ b/edge/edge-core/src/test/java/org/apache/servicecomb/edge/core/TestEdgeInvocation.java
@@ -87,13 +87,6 @@ public class TestEdgeInvocation {
@Before
public void setup() {
- new Expectations(VertxImpl.class) {
- {
- VertxImpl.context();
- result = context;
- }
- };
-
referenceConfig.setMicroserviceVersionRule(microserviceVersionRule);
referenceConfig.setTransport("rest");
diff --git a/foundations/foundation-test-scaffolding/src/main/java/io/vertx/core/impl/SyncContext.java b/foundations/foundation-test-scaffolding/src/main/java/io/vertx/core/impl/SyncContext.java
index 4fd3369..0640b48 100644
--- a/foundations/foundation-test-scaffolding/src/main/java/io/vertx/core/impl/SyncContext.java
+++ b/foundations/foundation-test-scaffolding/src/main/java/io/vertx/core/impl/SyncContext.java
@@ -19,20 +19,28 @@ package io.vertx.core.impl;
import java.util.concurrent.Executor;
import io.vertx.core.AsyncResult;
-import io.vertx.core.Future;
import io.vertx.core.Handler;
+import io.vertx.core.Promise;
+import io.vertx.core.Vertx;
import io.vertx.core.spi.metrics.PoolMetrics;
public class SyncContext extends EventLoopContext {
+ protected VertxInternal owner;
+
public SyncContext() {
this(null);
}
public SyncContext(VertxInternal vertx) {
super(vertx, null, null, null, null, null, null);
- if (SyncVertx.class.isInstance(vertx)) {
- ((SyncVertx) vertx).setContext(this);
- }
+ }
+
+ public VertxInternal owner() {
+ return owner;
+ }
+
+ public void setOwner(VertxInternal owner) {
+ this.owner = owner;
}
@Override
@@ -40,9 +48,9 @@ public class SyncContext extends EventLoopContext {
task.handle(null);
}
- public static <T> void syncExecuteBlocking(Handler<Future<T>> blockingCodeHandler,
+ public static <T> void syncExecuteBlocking(Handler<Promise<T>> blockingCodeHandler,
Handler<AsyncResult<T>> asyncResultHandler) {
- Future<T> res = Future.future();
+ Promise<T> res = Promise.promise();
try {
blockingCodeHandler.handle(res);
@@ -51,11 +59,11 @@ public class SyncContext extends EventLoopContext {
return;
}
- res.setHandler(asyncResultHandler);
+ res.future().setHandler(asyncResultHandler);
}
@Override
- public <T> void executeBlockingInternal(Handler<Future<T>> action, Handler<AsyncResult<T>> resultHandler) {
+ public <T> void executeBlockingInternal(Handler<Promise<T>> action, Handler<AsyncResult<T>> resultHandler) {
syncExecuteBlocking((future) -> {
try {
action.handle(future);
@@ -66,13 +74,13 @@ public class SyncContext extends EventLoopContext {
}
@Override
- public <T> void executeBlocking(Handler<Future<T>> blockingCodeHandler, boolean ordered,
+ public <T> void executeBlocking(Handler<Promise<T>> blockingCodeHandler, boolean ordered,
Handler<AsyncResult<T>> asyncResultHandler) {
syncExecuteBlocking(blockingCodeHandler, asyncResultHandler);
}
@Override
- <T> void executeBlocking(Handler<Future<T>> blockingCodeHandler,
+ <T> void executeBlocking(Handler<Promise<T>> blockingCodeHandler,
Handler<AsyncResult<T>> resultHandler,
Executor exec, TaskQueue queue, @SuppressWarnings("rawtypes") PoolMetrics metrics) {
syncExecuteBlocking(blockingCodeHandler, resultHandler);
diff --git a/foundations/foundation-test-scaffolding/src/main/java/io/vertx/core/impl/SyncVertx.java b/foundations/foundation-test-scaffolding/src/main/java/io/vertx/core/impl/SyncVertx.java
deleted file mode 100644
index c200ed7..0000000
--- a/foundations/foundation-test-scaffolding/src/main/java/io/vertx/core/impl/SyncVertx.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * 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 io.vertx.core.impl;
-
-import io.vertx.core.AsyncResult;
-import io.vertx.core.Handler;
-import io.vertx.core.Vertx;
-import io.vertx.core.VertxOptions;
-import io.vertx.core.net.impl.transport.Transport;
-
-/**
- * after test finished, need to invoke vertx.close
- */
-public class SyncVertx extends VertxImpl {
- private ContextImpl context = new SyncContext(this);
-
- public SyncVertx() {
- this(new VertxOptions(), null);
- }
-
- protected SyncVertx(VertxOptions options, Handler<AsyncResult<Vertx>> resultHandler) {
- super(options, Transport.transport(options.getPreferNativeTransport()));
- init();
- }
-
- @Override
- public ContextImpl getContext() {
- return context;
- }
-
- public void setContext(ContextImpl context) {
- this.context = context;
- }
-
- @Override
- public ContextImpl getOrCreateContext() {
- return context;
- }
-}
diff --git a/foundations/foundation-test-scaffolding/src/main/java/io/vertx/core/impl/VertxImpl.java b/foundations/foundation-test-scaffolding/src/main/java/io/vertx/core/impl/VertxImpl.java
deleted file mode 100644
index 92dd1d0..0000000
--- a/foundations/foundation-test-scaffolding/src/main/java/io/vertx/core/impl/VertxImpl.java
+++ /dev/null
@@ -1,1137 +0,0 @@
-/*
- * Copyright (c) 2011-2017 Contributors to the Eclipse Foundation
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License 2.0 which is available at
- * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
- * which is available at https://www.apache.org/licenses/LICENSE-2.0.
- *
- * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
- */
-
-package io.vertx.core.impl;
-
-import java.io.File;
-import java.io.IOException;
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.function.Supplier;
-
-import io.netty.channel.EventLoop;
-import io.netty.channel.EventLoopGroup;
-import io.netty.resolver.AddressResolverGroup;
-import io.netty.util.ResourceLeakDetector;
-import io.netty.util.concurrent.GenericFutureListener;
-import io.vertx.core.AsyncResult;
-import io.vertx.core.Closeable;
-import io.vertx.core.Context;
-import io.vertx.core.DeploymentOptions;
-import io.vertx.core.Future;
-import io.vertx.core.Handler;
-import io.vertx.core.ServiceHelper;
-import io.vertx.core.TimeoutStream;
-import io.vertx.core.Verticle;
-import io.vertx.core.Vertx;
-import io.vertx.core.VertxOptions;
-import io.vertx.core.datagram.DatagramSocket;
-import io.vertx.core.datagram.DatagramSocketOptions;
-import io.vertx.core.datagram.impl.DatagramSocketImpl;
-import io.vertx.core.dns.AddressResolverOptions;
-import io.vertx.core.dns.DnsClient;
-import io.vertx.core.dns.DnsClientOptions;
-import io.vertx.core.dns.impl.DnsClientImpl;
-import io.vertx.core.eventbus.EventBus;
-import io.vertx.core.eventbus.impl.EventBusImpl;
-import io.vertx.core.eventbus.impl.clustered.ClusteredEventBus;
-import io.vertx.core.file.FileSystem;
-import io.vertx.core.file.impl.FileResolver;
-import io.vertx.core.file.impl.FileSystemImpl;
-import io.vertx.core.file.impl.WindowsFileSystem;
-import io.vertx.core.http.HttpClient;
-import io.vertx.core.http.HttpClientOptions;
-import io.vertx.core.http.HttpServer;
-import io.vertx.core.http.HttpServerOptions;
-import io.vertx.core.http.impl.HttpClientImpl;
-import io.vertx.core.http.impl.HttpServerImpl;
-import io.vertx.core.impl.resolver.DnsResolverProvider;
-import io.vertx.core.json.JsonObject;
-import io.vertx.core.logging.Logger;
-import io.vertx.core.logging.LoggerFactory;
-import io.vertx.core.net.NetClient;
-import io.vertx.core.net.NetClientOptions;
-import io.vertx.core.net.NetServer;
-import io.vertx.core.net.NetServerOptions;
-import io.vertx.core.net.impl.NetClientImpl;
-import io.vertx.core.net.impl.NetServerImpl;
-import io.vertx.core.net.impl.ServerID;
-import io.vertx.core.net.impl.transport.Transport;
-import io.vertx.core.shareddata.SharedData;
-import io.vertx.core.shareddata.impl.SharedDataImpl;
-import io.vertx.core.spi.VerticleFactory;
-import io.vertx.core.spi.VertxMetricsFactory;
-import io.vertx.core.spi.cluster.ClusterManager;
-import io.vertx.core.spi.metrics.Metrics;
-import io.vertx.core.spi.metrics.MetricsProvider;
-import io.vertx.core.spi.metrics.PoolMetrics;
-import io.vertx.core.spi.metrics.VertxMetrics;
-
-/**
- * @author <a href="http://tfox.org">Tim Fox</a>
- */
-public class VertxImpl implements VertxInternal, MetricsProvider {
-
- private static final Logger log = LoggerFactory.getLogger(VertxImpl.class);
-
- private static final String CLUSTER_MAP_NAME = "__vertx.haInfo";
- private static final String NETTY_IO_RATIO_PROPERTY_NAME = "vertx.nettyIORatio";
- private static final int NETTY_IO_RATIO = Integer.getInteger(NETTY_IO_RATIO_PROPERTY_NAME, 50);
-
- static {
- // Netty resource leak detection has a performance overhead and we do not need it in Vert.x
- ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.DISABLED);
- // Use the JDK deflater/inflater by default
- System.setProperty("io.netty.noJdkZlibDecoder", "false");
- }
-
- static VertxImpl vertx(VertxOptions options) {
- VertxImpl vertx = new VertxImpl(options, Transport.transport(options.getPreferNativeTransport()));
- vertx.init();
- return vertx;
- }
-
- static VertxImpl vertx(VertxOptions options, Transport transport) {
- VertxImpl vertx = new VertxImpl(options, transport);
- vertx.init();
- return vertx;
- }
-
- static void clusteredVertx(VertxOptions options, Handler<AsyncResult<Vertx>> resultHandler) {
- VertxImpl vertx = new VertxImpl(options, Transport.transport(options.getPreferNativeTransport()));
- vertx.joinCluster(options, resultHandler);
- }
-
- private final FileSystem fileSystem = getFileSystem();
- private final SharedData sharedData;
- private final VertxMetrics metrics;
- private final ConcurrentMap<Long, InternalTimerHandler> timeouts = new ConcurrentHashMap<>();
- private final AtomicLong timeoutCounter = new AtomicLong(0);
- private final ClusterManager clusterManager;
- private final DeploymentManager deploymentManager;
- private final FileResolver fileResolver;
- private final Map<ServerID, HttpServerImpl> sharedHttpServers = new HashMap<>();
- private final Map<ServerID, NetServerImpl> sharedNetServers = new HashMap<>();
- final WorkerPool workerPool;
- final WorkerPool internalBlockingPool;
- private final ThreadFactory eventLoopThreadFactory;
- private final EventLoopGroup eventLoopGroup;
- private final EventLoopGroup acceptorEventLoopGroup;
- private final BlockedThreadChecker checker;
- private final AddressResolver addressResolver;
- private final AddressResolverOptions addressResolverOptions;
- private final EventBus eventBus;
- private volatile HAManager haManager;
- private boolean closed;
- private volatile Handler<Throwable> exceptionHandler;
- private final Map<String, SharedWorkerPool> namedWorkerPools;
- private final int defaultWorkerPoolSize;
- private final long defaultWorkerMaxExecTime;
- private final CloseHooks closeHooks;
- private final Transport transport;
-
- @SuppressWarnings("rawtypes")
- public VertxImpl(VertxOptions options, Transport transport) {
- // Sanity check
- if (Vertx.currentContext() != null) {
- log.warn("You're already on a Vert.x context, are you sure you want to create a new Vertx instance?");
- }
- closeHooks = new CloseHooks(log);
- checker = new BlockedThreadChecker(options.getBlockedThreadCheckInterval(), options.getBlockedThreadCheckIntervalUnit(), options.getWarningExceptionTime(), options.getWarningExceptionTimeUnit());
- eventLoopThreadFactory = new VertxThreadFactory("vert.x-eventloop-thread-", checker, false, options.getMaxEventLoopExecuteTime(), options.getMaxEventLoopExecuteTimeUnit());
- eventLoopGroup = transport.eventLoopGroup(options.getEventLoopPoolSize(), eventLoopThreadFactory, NETTY_IO_RATIO);
- ThreadFactory acceptorEventLoopThreadFactory = new VertxThreadFactory("vert.x-acceptor-thread-", checker, false, options.getMaxEventLoopExecuteTime(), options.getMaxEventLoopExecuteTimeUnit());
- // The acceptor event loop thread needs to be from a different pool otherwise can get lags in accepted connections
- // under a lot of load
- acceptorEventLoopGroup = transport.eventLoopGroup(1, acceptorEventLoopThreadFactory, 100);
-
- metrics = initialiseMetrics(options);
-
- ExecutorService workerExec = Executors.newFixedThreadPool(options.getWorkerPoolSize(),
- new VertxThreadFactory("vert.x-worker-thread-", checker, true, options.getMaxWorkerExecuteTime(), options.getMaxWorkerExecuteTimeUnit()));
- PoolMetrics workerPoolMetrics = metrics != null ? metrics.createPoolMetrics("worker", "vert.x-worker-thread", options.getWorkerPoolSize()) : null;
- ExecutorService internalBlockingExec = Executors.newFixedThreadPool(options.getInternalBlockingPoolSize(),
- new VertxThreadFactory("vert.x-internal-blocking-", checker, true, options.getMaxWorkerExecuteTime(), options.getMaxWorkerExecuteTimeUnit()));
- PoolMetrics internalBlockingPoolMetrics = metrics != null ? metrics.createPoolMetrics("worker", "vert.x-internal-blocking", options.getInternalBlockingPoolSize()) : null;
- internalBlockingPool = new WorkerPool(internalBlockingExec, internalBlockingPoolMetrics);
- namedWorkerPools = new HashMap<>();
- workerPool = new WorkerPool(workerExec, workerPoolMetrics);
- defaultWorkerPoolSize = options.getWorkerPoolSize();
- defaultWorkerMaxExecTime = options.getMaxWorkerExecuteTime();
-
- this.transport = transport;
- this.fileResolver = new FileResolver(options.getFileSystemOptions());
- this.addressResolverOptions = options.getAddressResolverOptions();
- this.addressResolver = new AddressResolver(this, options.getAddressResolverOptions());
- this.deploymentManager = new DeploymentManager(this);
- if (options.isClustered()) {
- this.clusterManager = getClusterManager(options);
- this.eventBus = new ClusteredEventBus(this, options, clusterManager);
- } else {
- this.clusterManager = null;
- this.eventBus = new EventBusImpl(this);
- }
- this.sharedData = new SharedDataImpl(this, clusterManager);
- }
-
- public void init() {
- eventBus.start(ar -> {});
- if (metrics != null) {
- metrics.vertxCreated(this);
- }
- }
-
- private void joinCluster(VertxOptions options, Handler<AsyncResult<Vertx>> resultHandler) {
- clusterManager.setVertx(this);
- clusterManager.join(ar -> {
- if (ar.succeeded()) {
- createHaManager(options, resultHandler);
- } else {
- log.error("Failed to join cluster", ar.cause());
- resultHandler.handle(Future.failedFuture(ar.cause()));
- }
- });
- }
-
- private void createHaManager(VertxOptions options, Handler<AsyncResult<Vertx>> resultHandler) {
- this.<Map<String, String>>executeBlocking(fut -> {
- fut.complete(clusterManager.getSyncMap(CLUSTER_MAP_NAME));
- }, false, ar -> {
- if (ar.succeeded()) {
- Map<String, String> clusterMap = ar.result();
- haManager = new HAManager(this, deploymentManager, clusterManager, clusterMap, options.getQuorumSize(), options.getHAGroup(), options.isHAEnabled());
- startEventBus(resultHandler);
- } else {
- log.error("Failed to start HAManager", ar.cause());
- resultHandler.handle(Future.failedFuture(ar.cause()));
- }
- });
- }
-
- private void startEventBus(Handler<AsyncResult<Vertx>> resultHandler) {
- eventBus.start(ar -> {
- if (ar.succeeded()) {
- initializeHaManager(resultHandler);
- } else {
- log.error("Failed to start event bus", ar.cause());
- resultHandler.handle(Future.failedFuture(ar.cause()));
- }
- });
- }
-
- private void initializeHaManager(Handler<AsyncResult<Vertx>> resultHandler) {
- this.executeBlocking(fut -> {
- // Init the manager (i.e register listener and check the quorum)
- // after the event bus has been fully started and updated its state
- // it will have also set the clustered changed view handler on the ha manager
- haManager.init();
- fut.complete();
- }, false, ar -> {
- if (ar.succeeded()) {
- if (metrics != null) {
- metrics.vertxCreated(this);
- }
- resultHandler.handle(Future.succeededFuture(this));
- } else {
- log.error("Failed to initialize HAManager", ar.cause());
- resultHandler.handle(Future.failedFuture(ar.cause()));
- }
- });
- }
-
- /**
- * @return The FileSystem implementation for the OS
- */
- protected FileSystem getFileSystem() {
- return Utils.isWindows() ? new WindowsFileSystem(this) : new FileSystemImpl(this);
- }
-
- @Override
- public DatagramSocket createDatagramSocket(DatagramSocketOptions options) {
- return DatagramSocketImpl.create(this, options);
- }
-
- @Override
- public DatagramSocket createDatagramSocket() {
- return createDatagramSocket(new DatagramSocketOptions());
- }
-
- public NetServer createNetServer(NetServerOptions options) {
- return new NetServerImpl(this, options);
- }
-
- @Override
- public NetServer createNetServer() {
- return createNetServer(new NetServerOptions());
- }
-
- public NetClient createNetClient(NetClientOptions options) {
- return new NetClientImpl(this, options);
- }
-
- @Override
- public NetClient createNetClient() {
- return createNetClient(new NetClientOptions());
- }
-
- @Override
- public Transport transport() {
- return transport;
- }
-
- @Override
- public boolean isNativeTransportEnabled() {
- return transport != Transport.JDK;
- }
-
- public FileSystem fileSystem() {
- return fileSystem;
- }
-
- public SharedData sharedData() {
- return sharedData;
- }
-
- public HttpServer createHttpServer(HttpServerOptions serverOptions) {
- return new HttpServerImpl(this, serverOptions);
- }
-
- @Override
- public HttpServer createHttpServer() {
- return createHttpServer(new HttpServerOptions());
- }
-
- public HttpClient createHttpClient(HttpClientOptions options) {
- return new HttpClientImpl(this, options);
- }
-
- @Override
- public HttpClient createHttpClient() {
- return createHttpClient(new HttpClientOptions());
- }
-
- public EventBus eventBus() {
- return eventBus;
- }
-
- public long setPeriodic(long delay, Handler<Long> handler) {
- return scheduleTimeout(getOrCreateContext(), handler, delay, true);
- }
-
- @Override
- public TimeoutStream periodicStream(long delay) {
- return new TimeoutStreamImpl(delay, true);
- }
-
- public long setTimer(long delay, Handler<Long> handler) {
- return scheduleTimeout(getOrCreateContext(), handler, delay, false);
- }
-
- @Override
- public TimeoutStream timerStream(long delay) {
- return new TimeoutStreamImpl(delay, false);
- }
-
- public void runOnContext(Handler<Void> task) {
- ContextImpl context = getOrCreateContext();
- context.runOnContext(task);
- }
-
- // The background pool is used for making blocking calls to legacy synchronous APIs
- public ExecutorService getWorkerPool() {
- return workerPool.executor();
- }
-
- public EventLoopGroup getEventLoopGroup() {
- return eventLoopGroup;
- }
-
- public EventLoopGroup getAcceptorEventLoopGroup() {
- return acceptorEventLoopGroup;
- }
-
- public ContextImpl getOrCreateContext() {
- ContextImpl ctx = getContext();
- if (ctx == null) {
- // We are running embedded - Create a context
- ctx = createEventLoopContext(null, null, new JsonObject(), Thread.currentThread().getContextClassLoader());
- }
- return ctx;
- }
-
- public Map<ServerID, HttpServerImpl> sharedHttpServers() {
- return sharedHttpServers;
- }
-
- public Map<ServerID, NetServerImpl> sharedNetServers() {
- return sharedNetServers;
- }
-
- @Override
- public boolean isMetricsEnabled() {
- return metrics != null;
- }
-
- @Override
- public Metrics getMetrics() {
- return metrics;
- }
-
- public boolean cancelTimer(long id) {
- InternalTimerHandler handler = timeouts.remove(id);
- if (handler != null) {
- handler.context.removeCloseHook(handler);
- return handler.cancel();
- } else {
- return false;
- }
- }
-
- @Override
- public EventLoopContext createEventLoopContext(String deploymentID, WorkerPool workerPool, JsonObject config, ClassLoader tccl) {
- return new EventLoopContext(this, internalBlockingPool, workerPool != null ? workerPool : this.workerPool, deploymentID, config, tccl);
- }
-
- @Override
- public ContextImpl createWorkerContext(boolean multiThreaded, String deploymentID, WorkerPool workerPool, JsonObject config,
- ClassLoader tccl) {
- if (workerPool == null) {
- workerPool = this.workerPool;
- }
- if (multiThreaded) {
- return new MultiThreadedWorkerContext(this, internalBlockingPool, workerPool, deploymentID, config, tccl);
- } else {
- return new WorkerContext(this, internalBlockingPool, workerPool, deploymentID, config, tccl);
- }
- }
-
- @Override
- public DnsClient createDnsClient(int port, String host) {
- return createDnsClient(new DnsClientOptions().setHost(host).setPort(port));
- }
-
- @Override
- public DnsClient createDnsClient() {
- return createDnsClient(new DnsClientOptions());
- }
-
- @Override
- public DnsClient createDnsClient(DnsClientOptions options) {
- String host = options.getHost();
- int port = options.getPort();
- if (host == null || port < 0) {
- DnsResolverProvider provider = new DnsResolverProvider(this, addressResolverOptions);
- InetSocketAddress address = provider.nameServerAddresses().get(0);
- // provide the host and port
- options = new DnsClientOptions(options)
- .setHost(address.getAddress().getHostAddress())
- .setPort(address.getPort());
- }
- return new DnsClientImpl(this, options);
- }
-
- private VertxMetrics initialiseMetrics(VertxOptions options) {
- if (options.getMetricsOptions() != null && options.getMetricsOptions().isEnabled()) {
- VertxMetricsFactory factory = options.getMetricsOptions().getFactory();
- if (factory == null) {
- factory = ServiceHelper.loadFactoryOrNull(VertxMetricsFactory.class);
- if (factory == null) {
- log.warn("Metrics has been set to enabled but no VertxMetricsFactory found on classpath");
- }
- }
- if (factory != null) {
- VertxMetrics metrics = factory.metrics(options);
- Objects.requireNonNull(metrics, "The metric instance created from " + factory + " cannot be null");
- return metrics;
- }
- }
- return null;
- }
-
- private ClusterManager getClusterManager(VertxOptions options) {
- ClusterManager mgr = options.getClusterManager();
- if (mgr == null) {
- String clusterManagerClassName = System.getProperty("vertx.cluster.managerClass");
- if (clusterManagerClassName != null) {
- // We allow specify a sys prop for the cluster manager factory which overrides ServiceLoader
- try {
- Class<?> clazz = Class.forName(clusterManagerClassName);
- mgr = (ClusterManager) clazz.newInstance();
- } catch (Exception e) {
- throw new IllegalStateException("Failed to instantiate " + clusterManagerClassName, e);
- }
- } else {
- mgr = ServiceHelper.loadFactoryOrNull(ClusterManager.class);
- if (mgr == null) {
- throw new IllegalStateException("No ClusterManagerFactory instances found on classpath");
- }
- }
- }
- return mgr;
- }
-
- private long scheduleTimeout(ContextImpl context, Handler<Long> handler, long delay, boolean periodic) {
- if (delay < 1) {
- throw new IllegalArgumentException("Cannot schedule a timer with delay < 1 ms");
- }
- long timerId = timeoutCounter.getAndIncrement();
- InternalTimerHandler task = new InternalTimerHandler(timerId, handler, periodic, delay, context);
- timeouts.put(timerId, task);
- context.addCloseHook(task);
- return timerId;
- }
-
- public static Context context() {
- Thread current = Thread.currentThread();
- if (current instanceof VertxThread) {
- return ((VertxThread) current).getContext();
- }
- return null;
- }
-
- public ContextImpl getContext() {
- ContextImpl context = (ContextImpl) context();
- if (context != null && context.owner == this) {
- return context;
- }
- return null;
- }
-
- public ClusterManager getClusterManager() {
- return clusterManager;
- }
-
- @Override
- public void close() {
- close(null);
- }
-
- private void closeClusterManager(Handler<AsyncResult<Void>> completionHandler) {
- if (clusterManager != null) {
- clusterManager.leave(ar -> {
- if (ar.failed()) {
- log.error("Failed to leave cluster", ar.cause());
- }
- if (completionHandler != null) {
- runOnContext(v -> completionHandler.handle(Future.succeededFuture()));
- }
- });
- } else if (completionHandler != null) {
- runOnContext(v -> completionHandler.handle(Future.succeededFuture()));
- }
- }
-
- @Override
- public synchronized void close(Handler<AsyncResult<Void>> completionHandler) {
- if (closed || eventBus == null) {
- // Just call the handler directly since pools shutdown
- if (completionHandler != null) {
- completionHandler.handle(Future.succeededFuture());
- }
- return;
- }
- closed = true;
-
- closeHooks.run(ar -> {
- deploymentManager.undeployAll(ar1 -> {
- HAManager haManager = haManager();
- Future<Void> haFuture = Future.future();
- if (haManager != null) {
- this.executeBlocking(fut -> {
- haManager.stop();
- fut.complete();
- }, false, haFuture);
- } else {
- haFuture.complete();
- }
- haFuture.setHandler(ar2 -> {
- addressResolver.close(ar3 -> {
- eventBus.close(ar4 -> {
- closeClusterManager(ar5 -> {
- // Copy set to prevent ConcurrentModificationException
- Set<HttpServerImpl> httpServers = new HashSet<>(sharedHttpServers.values());
- Set<NetServerImpl> netServers = new HashSet<>(sharedNetServers.values());
- sharedHttpServers.clear();
- sharedNetServers.clear();
-
- int serverCount = httpServers.size() + netServers.size();
-
- AtomicInteger serverCloseCount = new AtomicInteger();
-
- Handler<AsyncResult<Void>> serverCloseHandler = res -> {
- if (res.failed()) {
- log.error("Failure in shutting down server", res.cause());
- }
- if (serverCloseCount.incrementAndGet() == serverCount) {
- deleteCacheDirAndShutdown(completionHandler);
- }
- };
-
- for (HttpServerImpl server : httpServers) {
- server.closeAll(serverCloseHandler);
- }
- for (NetServerImpl server : netServers) {
- server.closeAll(serverCloseHandler);
- }
- if (serverCount == 0) {
- deleteCacheDirAndShutdown(completionHandler);
- }
- });
- });
- });
- });
- });
- });
- }
-
- @Override
- public void deployVerticle(Verticle verticle) {
- deployVerticle(verticle, new DeploymentOptions(), null);
- }
-
- @Override
- public void deployVerticle(Verticle verticle, Handler<AsyncResult<String>> completionHandler) {
- deployVerticle(verticle, new DeploymentOptions(), completionHandler);
- }
-
- @Override
- public void deployVerticle(String name, Handler<AsyncResult<String>> completionHandler) {
- deployVerticle(name, new DeploymentOptions(), completionHandler);
- }
-
- @Override
- public void deployVerticle(Verticle verticle, DeploymentOptions options) {
- deployVerticle(verticle, options, null);
- }
-
- @Override
- public void deployVerticle(Class<? extends Verticle> verticleClass, DeploymentOptions options) {
- deployVerticle(() -> {
- try {
- return verticleClass.newInstance();
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }, options);
- }
-
- @Override
- public void deployVerticle(Supplier<Verticle> verticleSupplier, DeploymentOptions options) {
- deployVerticle(verticleSupplier, options, null);
- }
-
- @Override
- public void deployVerticle(Verticle verticle, DeploymentOptions options, Handler<AsyncResult<String>> completionHandler) {
- if (options.getInstances() != 1) {
- throw new IllegalArgumentException("Can't specify > 1 instances for already created verticle");
- }
- deployVerticle(() -> verticle, options, completionHandler);
- }
-
- @Override
- public void deployVerticle(Class<? extends Verticle> verticleClass, DeploymentOptions options, Handler<AsyncResult<String>> completionHandler) {
- deployVerticle(() -> {
- try {
- return verticleClass.newInstance();
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }, options, completionHandler);
- }
-
- @Override
- public void deployVerticle(Supplier<Verticle> verticleSupplier, DeploymentOptions options, Handler<AsyncResult<String>> completionHandler) {
- boolean closed;
- synchronized (this) {
- closed = this.closed;
- }
- if (closed) {
- if (completionHandler != null) {
- completionHandler.handle(Future.failedFuture("Vert.x closed"));
- }
- } else {
- deploymentManager.deployVerticle(verticleSupplier, options, completionHandler);
- }
- }
-
- @Override
- public void deployVerticle(String name) {
- deployVerticle(name, new DeploymentOptions(), null);
- }
-
- @Override
- public void deployVerticle(String name, DeploymentOptions options) {
- deployVerticle(name, options, null);
- }
-
- @Override
- public void deployVerticle(String name, DeploymentOptions options, Handler<AsyncResult<String>> completionHandler) {
- if (options.isHa() && haManager() != null && haManager().isEnabled()) {
- haManager().deployVerticle(name, options, completionHandler);
- } else {
- deploymentManager.deployVerticle(name, options, completionHandler);
- }
- }
-
- @Override
- public String getNodeID() {
- return clusterManager.getNodeID();
- }
-
- @Override
- public void undeploy(String deploymentID) {
- undeploy(deploymentID, res -> {
- });
- }
-
- @Override
- public void undeploy(String deploymentID, Handler<AsyncResult<Void>> completionHandler) {
- HAManager haManager = haManager();
- Future<Void> haFuture = Future.future();
- if (haManager != null && haManager.isEnabled()) {
- this.executeBlocking(fut -> {
- haManager.removeFromHA(deploymentID);
- fut.complete();
- }, false, haFuture);
- } else {
- haFuture.complete();
- }
- haFuture.compose(v -> {
- Future<Void> deploymentFuture = Future.future();
- deploymentManager.undeployVerticle(deploymentID, deploymentFuture);
- return deploymentFuture;
- }).setHandler(completionHandler);
- }
-
- @Override
- public Set<String> deploymentIDs() {
- return deploymentManager.deployments();
- }
-
- @Override
- public void registerVerticleFactory(VerticleFactory factory) {
- deploymentManager.registerVerticleFactory(factory);
- }
-
- @Override
- public void unregisterVerticleFactory(VerticleFactory factory) {
- deploymentManager.unregisterVerticleFactory(factory);
- }
-
- @Override
- public Set<VerticleFactory> verticleFactories() {
- return deploymentManager.verticleFactories();
- }
-
- @Override
- public <T> void executeBlockingInternal(Handler<Future<T>> blockingCodeHandler, Handler<AsyncResult<T>> resultHandler) {
- ContextImpl context = getOrCreateContext();
-
- context.executeBlockingInternal(blockingCodeHandler, resultHandler);
- }
-
- @Override
- public <T> void executeBlocking(Handler<Future<T>> blockingCodeHandler, boolean ordered,
- Handler<AsyncResult<T>> asyncResultHandler) {
- ContextImpl context = getOrCreateContext();
- context.executeBlocking(blockingCodeHandler, ordered, asyncResultHandler);
- }
-
- @Override
- public <T> void executeBlocking(Handler<Future<T>> blockingCodeHandler,
- Handler<AsyncResult<T>> asyncResultHandler) {
- executeBlocking(blockingCodeHandler, true, asyncResultHandler);
- }
-
- @Override
- public boolean isClustered() {
- return clusterManager != null;
- }
-
- @Override
- public EventLoopGroup nettyEventLoopGroup() {
- return eventLoopGroup;
- }
-
- // For testing
- public void simulateKill() {
- if (haManager() != null) {
- haManager().simulateKill();
- }
- }
-
- @Override
- public Deployment getDeployment(String deploymentID) {
- return deploymentManager.getDeployment(deploymentID);
- }
-
- @Override
- public synchronized void failoverCompleteHandler(FailoverCompleteHandler failoverCompleteHandler) {
- if (haManager() != null) {
- haManager().setFailoverCompleteHandler(failoverCompleteHandler);
- }
- }
-
- @Override
- public boolean isKilled() {
- return haManager().isKilled();
- }
-
- @Override
- public void failDuringFailover(boolean fail) {
- if (haManager() != null) {
- haManager().failDuringFailover(fail);
- }
- }
-
- @Override
- public VertxMetrics metricsSPI() {
- return metrics;
- }
-
- @Override
- public File resolveFile(String fileName) {
- return fileResolver.resolveFile(fileName);
- }
-
- @Override
- public void resolveAddress(String hostname, Handler<AsyncResult<InetAddress>> resultHandler) {
- addressResolver.resolveHostname(hostname, resultHandler);
- }
-
- @Override
- public AddressResolver addressResolver() {
- return addressResolver;
- }
-
- @Override
- public AddressResolverGroup<InetSocketAddress> nettyAddressResolverGroup() {
- return addressResolver.nettyAddressResolverGroup();
- }
-
- @SuppressWarnings({"unchecked", "rawtypes"})
- private void deleteCacheDirAndShutdown(Handler<AsyncResult<Void>> completionHandler) {
- executeBlockingInternal(fut -> {
- try {
- fileResolver.close();
- fut.complete();
- } catch (IOException e) {
- fut.tryFail(e);
- }
- }, ar -> {
-
- workerPool.close();
- internalBlockingPool.close();
- new ArrayList<>(namedWorkerPools.values()).forEach(WorkerPool::close);
-
- acceptorEventLoopGroup.shutdownGracefully(0, 10, TimeUnit.SECONDS).addListener(new GenericFutureListener() {
- @Override
- public void operationComplete(io.netty.util.concurrent.Future future) throws Exception {
- if (!future.isSuccess()) {
- log.warn("Failure in shutting down acceptor event loop group", future.cause());
- }
- eventLoopGroup.shutdownGracefully(0, 10, TimeUnit.SECONDS).addListener(new GenericFutureListener() {
- @Override
- public void operationComplete(io.netty.util.concurrent.Future future) throws Exception {
- if (!future.isSuccess()) {
- log.warn("Failure in shutting down event loop group", future.cause());
- }
- if (metrics != null) {
- metrics.close();
- }
-
- checker.close();
-
- if (completionHandler != null) {
- eventLoopThreadFactory.newThread(() -> {
- completionHandler.handle(Future.succeededFuture());
- }).start();
- }
- }
- });
- }
- });
- });
- }
-
- public HAManager haManager() {
- return haManager;
- }
-
- private class InternalTimerHandler implements Handler<Void>, Closeable {
- final Handler<Long> handler;
- final boolean periodic;
- final long timerID;
- final ContextImpl context;
- final java.util.concurrent.Future<?> future;
- final AtomicBoolean cancelled;
-
- boolean cancel() {
- if (cancelled.compareAndSet(false, true)) {
- if (metrics != null) {
- metrics.timerEnded(timerID, true);
- }
- future.cancel(false);
- return true;
- } else {
- return false;
- }
- }
-
- InternalTimerHandler(long timerID, Handler<Long> runnable, boolean periodic, long delay, ContextImpl context) {
- this.context = context;
- this.timerID = timerID;
- this.handler = runnable;
- this.periodic = periodic;
- this.cancelled = new AtomicBoolean();
- EventLoop el = context.nettyEventLoop();
- Runnable toRun = () -> context.runOnContext(this);
- if (periodic) {
- future = el.scheduleAtFixedRate(toRun, delay, delay, TimeUnit.MILLISECONDS);
- } else {
- future = el.schedule(toRun, delay, TimeUnit.MILLISECONDS);
- }
- if (metrics != null) {
- metrics.timerCreated(timerID);
- }
- }
-
- public void handle(Void v) {
- if (!cancelled.get()) {
- try {
- handler.handle(timerID);
- } finally {
- if (!periodic) {
- // Clean up after it's fired
- cleanupNonPeriodic();
- }
- }
- }
- }
-
- private void cleanupNonPeriodic() {
- VertxImpl.this.timeouts.remove(timerID);
- if (metrics != null) {
- metrics.timerEnded(timerID, false);
- }
- ContextImpl context = getContext();
- if (context != null) {
- context.removeCloseHook(this);
- }
- }
-
- // Called via Context close hook when Verticle is undeployed
- public void close(Handler<AsyncResult<Void>> completionHandler) {
- VertxImpl.this.timeouts.remove(timerID);
- cancel();
- completionHandler.handle(Future.succeededFuture());
- }
-
- }
-
- /*
- *
- * This class is optimised for performance when used on the same event loop that is was passed to the handler with.
- * However it can be used safely from other threads.
- *
- * The internal state is protected using the synchronized keyword. If always used on the same event loop, then
- * we benefit from biased locking which makes the overhead of synchronized near zero.
- *
- */
- private class TimeoutStreamImpl implements TimeoutStream, Handler<Long> {
-
- private final long delay;
- private final boolean periodic;
-
- private Long id;
- private Handler<Long> handler;
- private Handler<Void> endHandler;
- private long demand;
-
- public TimeoutStreamImpl(long delay, boolean periodic) {
- this.delay = delay;
- this.periodic = periodic;
- this.demand = Long.MAX_VALUE;
- }
-
- @Override
- public synchronized void handle(Long event) {
- try {
- if (demand > 0) {
- demand--;
- handler.handle(event);
- }
- } finally {
- if (!periodic && endHandler != null) {
- endHandler.handle(null);
- }
- }
- }
-
- @Override
- public synchronized TimeoutStream fetch(long amount) {
- demand += amount;
- if (demand < 0) {
- demand = Long.MAX_VALUE;
- }
- return this;
- }
-
- @Override
- public TimeoutStream exceptionHandler(Handler<Throwable> handler) {
- return this;
- }
-
- @Override
- public void cancel() {
- if (id != null) {
- VertxImpl.this.cancelTimer(id);
- }
- }
-
- @Override
- public synchronized TimeoutStream handler(Handler<Long> handler) {
- if (handler != null) {
- if (id != null) {
- throw new IllegalStateException();
- }
- this.handler = handler;
- id = scheduleTimeout(getOrCreateContext(), this, delay, periodic);
- } else {
- cancel();
- }
- return this;
- }
-
- @Override
- public synchronized TimeoutStream pause() {
- demand = 0;
- return this;
- }
-
- @Override
- public synchronized TimeoutStream resume() {
- demand = Long.MAX_VALUE;
- return this;
- }
-
- @Override
- public synchronized TimeoutStream endHandler(Handler<Void> endHandler) {
- this.endHandler = endHandler;
- return this;
- }
- }
-
- class SharedWorkerPool extends WorkerPool {
-
- private final String name;
- private int refCount = 1;
-
- @SuppressWarnings("rawtypes")
- SharedWorkerPool(String name, ExecutorService workerExec, PoolMetrics workerMetrics) {
- super(workerExec, workerMetrics);
- this.name = name;
- }
-
- @Override
- void close() {
- synchronized (VertxImpl.this) {
- if (refCount > 0) {
- refCount = 0;
- super.close();
- }
- }
- }
-
- void release() {
- synchronized (VertxImpl.this) {
- if (--refCount == 0) {
- namedWorkerPools.remove(name);
- super.close();
- }
- }
- }
- }
-
- @Override
- public WorkerExecutorImpl createSharedWorkerExecutor(String name) {
- return createSharedWorkerExecutor(name, defaultWorkerPoolSize);
- }
-
- @Override
- public WorkerExecutorImpl createSharedWorkerExecutor(String name, int poolSize) {
- return createSharedWorkerExecutor(name, poolSize, defaultWorkerMaxExecTime);
- }
-
- @Override
- public synchronized WorkerExecutorImpl createSharedWorkerExecutor(String name, int poolSize, long maxExecuteTime) {
- return createSharedWorkerExecutor(name, poolSize, maxExecuteTime, TimeUnit.NANOSECONDS);
- }
-
- @SuppressWarnings("rawtypes")
- @Override
- public synchronized WorkerExecutorImpl createSharedWorkerExecutor(String name, int poolSize, long maxExecuteTime, TimeUnit maxExecuteTimeUnit) {
- if (poolSize < 1) {
- throw new IllegalArgumentException("poolSize must be > 0");
- }
- if (maxExecuteTime < 1) {
- throw new IllegalArgumentException("maxExecuteTime must be > 0");
- }
- SharedWorkerPool sharedWorkerPool = namedWorkerPools.get(name);
- if (sharedWorkerPool == null) {
- ExecutorService workerExec = Executors.newFixedThreadPool(poolSize, new VertxThreadFactory(name + "-", checker, true, maxExecuteTime, maxExecuteTimeUnit));
- PoolMetrics workerMetrics = metrics != null ? metrics.createPoolMetrics("worker", name, poolSize) : null;
- namedWorkerPools.put(name, sharedWorkerPool = new SharedWorkerPool(name, workerExec, workerMetrics));
- } else {
- sharedWorkerPool.refCount++;
- }
- ContextImpl context = getOrCreateContext();
- WorkerExecutorImpl namedExec = new WorkerExecutorImpl(context, sharedWorkerPool);
- context.addCloseHook(namedExec);
- return namedExec;
- }
-
- @Override
- public Vertx exceptionHandler(Handler<Throwable> handler) {
- exceptionHandler = handler;
- return this;
- }
-
- @Override
- public Handler<Throwable> exceptionHandler() {
- return exceptionHandler;
- }
-
- @Override
- public void addCloseHook(Closeable hook) {
- closeHooks.add(hook);
- }
-
- @Override
- public void removeCloseHook(Closeable hook) {
- closeHooks.remove(hook);
- }
-}
diff --git a/foundations/foundation-vertx/src/main/java/io/vertx/core/impl/VertxImpl.java b/foundations/foundation-vertx/src/main/java/io/vertx/core/impl/VertxImpl.java
deleted file mode 100644
index ef61ee2..0000000
--- a/foundations/foundation-vertx/src/main/java/io/vertx/core/impl/VertxImpl.java
+++ /dev/null
@@ -1,1118 +0,0 @@
-/*
- * Copyright (c) 2011-2017 Contributors to the Eclipse Foundation
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License 2.0 which is available at
- * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
- * which is available at https://www.apache.org/licenses/LICENSE-2.0.
- *
- * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
- */
-
-package io.vertx.core.impl;
-
-import io.netty.channel.EventLoop;
-import io.netty.channel.EventLoopGroup;
-import io.netty.resolver.AddressResolverGroup;
-import io.netty.util.ResourceLeakDetector;
-import io.netty.util.concurrent.GenericFutureListener;
-import io.vertx.core.*;
-import io.vertx.core.Future;
-import io.vertx.core.datagram.DatagramSocket;
-import io.vertx.core.datagram.DatagramSocketOptions;
-import io.vertx.core.datagram.impl.DatagramSocketImpl;
-import io.vertx.core.dns.AddressResolverOptions;
-import io.vertx.core.dns.DnsClient;
-import io.vertx.core.dns.DnsClientOptions;
-import io.vertx.core.dns.impl.DnsClientImpl;
-import io.vertx.core.eventbus.EventBus;
-import io.vertx.core.eventbus.impl.EventBusImpl;
-import io.vertx.core.eventbus.impl.clustered.ClusteredEventBus;
-import io.vertx.core.file.FileSystem;
-import io.vertx.core.file.impl.FileResolver;
-import io.vertx.core.file.impl.FileSystemImpl;
-import io.vertx.core.file.impl.WindowsFileSystem;
-import io.vertx.core.http.HttpClient;
-import io.vertx.core.http.HttpClientOptions;
-import io.vertx.core.http.HttpServer;
-import io.vertx.core.http.HttpServerOptions;
-import io.vertx.core.http.impl.HttpClientImpl;
-import io.vertx.core.http.impl.HttpServerImpl;
-import io.vertx.core.impl.resolver.DnsResolverProvider;
-import io.vertx.core.json.JsonObject;
-import io.vertx.core.logging.Logger;
-import io.vertx.core.logging.LoggerFactory;
-import io.vertx.core.net.NetClient;
-import io.vertx.core.net.NetClientOptions;
-import io.vertx.core.net.NetServer;
-import io.vertx.core.net.NetServerOptions;
-import io.vertx.core.net.impl.NetClientImpl;
-import io.vertx.core.net.impl.NetServerImpl;
-import io.vertx.core.net.impl.ServerID;
-import io.vertx.core.net.impl.transport.Transport;
-import io.vertx.core.shareddata.SharedData;
-import io.vertx.core.shareddata.impl.SharedDataImpl;
-import io.vertx.core.spi.VerticleFactory;
-import io.vertx.core.spi.VertxMetricsFactory;
-import io.vertx.core.spi.cluster.ClusterManager;
-import io.vertx.core.spi.metrics.Metrics;
-import io.vertx.core.spi.metrics.MetricsProvider;
-import io.vertx.core.spi.metrics.PoolMetrics;
-import io.vertx.core.spi.metrics.VertxMetrics;
-
-import java.io.File;
-import java.io.IOException;
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-import java.util.*;
-import java.util.concurrent.*;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.function.Supplier;
-
-/**
- * @author <a href="http://tfox.org">Tim Fox</a>
- */
-public class VertxImpl implements VertxInternal, MetricsProvider {
-
- private static final Logger log = LoggerFactory.getLogger(VertxImpl.class);
-
- private static final String CLUSTER_MAP_NAME = "__vertx.haInfo";
- private static final String NETTY_IO_RATIO_PROPERTY_NAME = "vertx.nettyIORatio";
- private static final int NETTY_IO_RATIO = Integer.getInteger(NETTY_IO_RATIO_PROPERTY_NAME, 50);
-
- static {
- // Netty resource leak detection has a performance overhead and we do not need it in Vert.x
- ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.DISABLED);
- // Use the JDK deflater/inflater by default
- System.setProperty("io.netty.noJdkZlibDecoder", "false");
- }
-
- static VertxImpl vertx(VertxOptions options) {
- VertxImpl vertx = new VertxImpl(options, Transport.transport(options.getPreferNativeTransport()));
- vertx.init();
- return vertx;
- }
-
- static VertxImpl vertx(VertxOptions options, Transport transport) {
- VertxImpl vertx = new VertxImpl(options, transport);
- vertx.init();
- return vertx;
- }
-
- static void clusteredVertx(VertxOptions options, Handler<AsyncResult<Vertx>> resultHandler) {
- VertxImpl vertx = new VertxImpl(options, Transport.transport(options.getPreferNativeTransport()));
- vertx.joinCluster(options, resultHandler);
- }
-
- private final FileSystem fileSystem = getFileSystem();
- private final SharedData sharedData;
- private final VertxMetrics metrics;
- private final ConcurrentMap<Long, InternalTimerHandler> timeouts = new ConcurrentHashMap<>();
- private final AtomicLong timeoutCounter = new AtomicLong(0);
- private final ClusterManager clusterManager;
- private final DeploymentManager deploymentManager;
- private final FileResolver fileResolver;
- private final Map<ServerID, HttpServerImpl> sharedHttpServers = new HashMap<>();
- private final Map<ServerID, NetServerImpl> sharedNetServers = new HashMap<>();
- final WorkerPool workerPool;
- final WorkerPool internalBlockingPool;
- private final ThreadFactory eventLoopThreadFactory;
- private final EventLoopGroup eventLoopGroup;
- private final EventLoopGroup acceptorEventLoopGroup;
- private final BlockedThreadChecker checker;
- private final AddressResolver addressResolver;
- private final AddressResolverOptions addressResolverOptions;
- private final EventBus eventBus;
- private volatile HAManager haManager;
- private boolean closed;
- private volatile Handler<Throwable> exceptionHandler;
- private final Map<String, SharedWorkerPool> namedWorkerPools;
- private final int defaultWorkerPoolSize;
- private final long defaultWorkerMaxExecTime;
- private final CloseHooks closeHooks;
- private final Transport transport;
-
- @SuppressWarnings("rawtypes")
- public VertxImpl(VertxOptions options, Transport transport) {
- // Sanity check
- if (Vertx.currentContext() != null) {
- log.warn("You're already on a Vert.x context, are you sure you want to create a new Vertx instance?");
- }
- closeHooks = new CloseHooks(log);
- checker = new BlockedThreadChecker(options.getBlockedThreadCheckInterval(), options.getBlockedThreadCheckIntervalUnit(), options.getWarningExceptionTime(), options.getWarningExceptionTimeUnit());
- eventLoopThreadFactory = new VertxThreadFactory("vert.x-eventloop-thread-", checker, false, options.getMaxEventLoopExecuteTime(), options.getMaxEventLoopExecuteTimeUnit());
- eventLoopGroup = transport.eventLoopGroup(options.getEventLoopPoolSize(), eventLoopThreadFactory, NETTY_IO_RATIO);
- ThreadFactory acceptorEventLoopThreadFactory = new VertxThreadFactory("vert.x-acceptor-thread-", checker, false, options.getMaxEventLoopExecuteTime(), options.getMaxEventLoopExecuteTimeUnit());
- // The acceptor event loop thread needs to be from a different pool otherwise can get lags in accepted connections
- // under a lot of load
- acceptorEventLoopGroup = transport.eventLoopGroup(1, acceptorEventLoopThreadFactory, 100);
-
- metrics = initialiseMetrics(options);
-
- ExecutorService workerExec = Executors.newFixedThreadPool(options.getWorkerPoolSize(),
- new VertxThreadFactory("vert.x-worker-thread-", checker, true, options.getMaxWorkerExecuteTime(), options.getMaxWorkerExecuteTimeUnit()));
- PoolMetrics workerPoolMetrics = metrics != null ? metrics.createPoolMetrics("worker", "vert.x-worker-thread", options.getWorkerPoolSize()) : null;
- ExecutorService internalBlockingExec = Executors.newFixedThreadPool(options.getInternalBlockingPoolSize(),
- new VertxThreadFactory("vert.x-internal-blocking-", checker, true, options.getMaxWorkerExecuteTime(), options.getMaxWorkerExecuteTimeUnit()));
- PoolMetrics internalBlockingPoolMetrics = metrics != null ? metrics.createPoolMetrics("worker", "vert.x-internal-blocking", options.getInternalBlockingPoolSize()) : null;
- internalBlockingPool = new WorkerPool(internalBlockingExec, internalBlockingPoolMetrics);
- namedWorkerPools = new HashMap<>();
- workerPool = new WorkerPool(workerExec, workerPoolMetrics);
- defaultWorkerPoolSize = options.getWorkerPoolSize();
- defaultWorkerMaxExecTime = options.getMaxWorkerExecuteTime();
-
- this.transport = transport;
- this.fileResolver = new FileResolver(options.getFileSystemOptions());
- this.addressResolverOptions = options.getAddressResolverOptions();
- this.addressResolver = new AddressResolver(this, options.getAddressResolverOptions());
- this.deploymentManager = new DeploymentManager(this);
- if (options.isClustered()) {
- this.clusterManager = getClusterManager(options);
- this.eventBus = new ClusteredEventBus(this, options, clusterManager);
- } else {
- this.clusterManager = null;
- this.eventBus = new EventBusImpl(this);
- }
- this.sharedData = new SharedDataImpl(this, clusterManager);
- }
-
- public void init() {
- eventBus.start(ar -> {});
- if (metrics != null) {
- metrics.vertxCreated(this);
- }
- }
-
- private void joinCluster(VertxOptions options, Handler<AsyncResult<Vertx>> resultHandler) {
- clusterManager.setVertx(this);
- clusterManager.join(ar -> {
- if (ar.succeeded()) {
- createHaManager(options, resultHandler);
- } else {
- log.error("Failed to join cluster", ar.cause());
- resultHandler.handle(Future.failedFuture(ar.cause()));
- }
- });
- }
-
- private void createHaManager(VertxOptions options, Handler<AsyncResult<Vertx>> resultHandler) {
- this.<Map<String, String>>executeBlocking(fut -> {
- fut.complete(clusterManager.getSyncMap(CLUSTER_MAP_NAME));
- }, false, ar -> {
- if (ar.succeeded()) {
- Map<String, String> clusterMap = ar.result();
- haManager = new HAManager(this, deploymentManager, clusterManager, clusterMap, options.getQuorumSize(), options.getHAGroup(), options.isHAEnabled());
- startEventBus(resultHandler);
- } else {
- log.error("Failed to start HAManager", ar.cause());
- resultHandler.handle(Future.failedFuture(ar.cause()));
- }
- });
- }
-
- private void startEventBus(Handler<AsyncResult<Vertx>> resultHandler) {
- eventBus.start(ar -> {
- if (ar.succeeded()) {
- initializeHaManager(resultHandler);
- } else {
- log.error("Failed to start event bus", ar.cause());
- resultHandler.handle(Future.failedFuture(ar.cause()));
- }
- });
- }
-
- private void initializeHaManager(Handler<AsyncResult<Vertx>> resultHandler) {
- this.executeBlocking(fut -> {
- // Init the manager (i.e register listener and check the quorum)
- // after the event bus has been fully started and updated its state
- // it will have also set the clustered changed view handler on the ha manager
- haManager.init();
- fut.complete();
- }, false, ar -> {
- if (ar.succeeded()) {
- if (metrics != null) {
- metrics.vertxCreated(this);
- }
- resultHandler.handle(Future.succeededFuture(this));
- } else {
- log.error("Failed to initialize HAManager", ar.cause());
- resultHandler.handle(Future.failedFuture(ar.cause()));
- }
- });
- }
-
- /**
- * @return The FileSystem implementation for the OS
- */
- protected FileSystem getFileSystem() {
- return Utils.isWindows() ? new WindowsFileSystem(this) : new FileSystemImpl(this);
- }
-
- @Override
- public DatagramSocket createDatagramSocket(DatagramSocketOptions options) {
- return DatagramSocketImpl.create(this, options);
- }
-
- @Override
- public DatagramSocket createDatagramSocket() {
- return createDatagramSocket(new DatagramSocketOptions());
- }
-
- public NetServer createNetServer(NetServerOptions options) {
- return new NetServerImpl(this, options);
- }
-
- @Override
- public NetServer createNetServer() {
- return createNetServer(new NetServerOptions());
- }
-
- public NetClient createNetClient(NetClientOptions options) {
- return new NetClientImpl(this, options);
- }
-
- @Override
- public NetClient createNetClient() {
- return createNetClient(new NetClientOptions());
- }
-
- @Override
- public Transport transport() {
- return transport;
- }
-
- @Override
- public boolean isNativeTransportEnabled() {
- return transport != Transport.JDK;
- }
-
- public FileSystem fileSystem() {
- return fileSystem;
- }
-
- public SharedData sharedData() {
- return sharedData;
- }
-
- public HttpServer createHttpServer(HttpServerOptions serverOptions) {
- return new HttpServerImpl(this, serverOptions);
- }
-
- @Override
- public HttpServer createHttpServer() {
- return createHttpServer(new HttpServerOptions());
- }
-
- public HttpClient createHttpClient(HttpClientOptions options) {
- return new HttpClientImpl(this, options);
- }
-
- @Override
- public HttpClient createHttpClient() {
- return createHttpClient(new HttpClientOptions());
- }
-
- public EventBus eventBus() {
- return eventBus;
- }
-
- public long setPeriodic(long delay, Handler<Long> handler) {
- return scheduleTimeout(getOrCreateContext(), handler, delay, true);
- }
-
- @Override
- public TimeoutStream periodicStream(long delay) {
- return new TimeoutStreamImpl(delay, true);
- }
-
- public long setTimer(long delay, Handler<Long> handler) {
- return scheduleTimeout(getOrCreateContext(), handler, delay, false);
- }
-
- @Override
- public TimeoutStream timerStream(long delay) {
- return new TimeoutStreamImpl(delay, false);
- }
-
- public void runOnContext(Handler<Void> task) {
- ContextImpl context = getOrCreateContext();
- context.runOnContext(task);
- }
-
- // The background pool is used for making blocking calls to legacy synchronous APIs
- public ExecutorService getWorkerPool() {
- return workerPool.executor();
- }
-
- public EventLoopGroup getEventLoopGroup() {
- return eventLoopGroup;
- }
-
- public EventLoopGroup getAcceptorEventLoopGroup() {
- return acceptorEventLoopGroup;
- }
-
- public ContextImpl getOrCreateContext() {
- ContextImpl ctx = getContext();
- if (ctx == null) {
- // We are running embedded - Create a context
- ctx = createEventLoopContext(null, null, new JsonObject(), Thread.currentThread().getContextClassLoader());
- }
- return ctx;
- }
-
- public Map<ServerID, HttpServerImpl> sharedHttpServers() {
- return sharedHttpServers;
- }
-
- public Map<ServerID, NetServerImpl> sharedNetServers() {
- return sharedNetServers;
- }
-
- @Override
- public boolean isMetricsEnabled() {
- return metrics != null;
- }
-
- @Override
- public Metrics getMetrics() {
- return metrics;
- }
-
- public boolean cancelTimer(long id) {
- InternalTimerHandler handler = timeouts.remove(id);
- if (handler != null) {
- handler.context.removeCloseHook(handler);
- return handler.cancel();
- } else {
- return false;
- }
- }
-
- @Override
- public EventLoopContext createEventLoopContext(String deploymentID, WorkerPool workerPool, JsonObject config, ClassLoader tccl) {
- return new EventLoopContext(this, internalBlockingPool, workerPool != null ? workerPool : this.workerPool, deploymentID, config, tccl);
- }
-
- @Override
- public ContextImpl createWorkerContext(boolean multiThreaded, String deploymentID, WorkerPool workerPool, JsonObject config,
- ClassLoader tccl) {
- if (workerPool == null) {
- workerPool = this.workerPool;
- }
- if (multiThreaded) {
- return new MultiThreadedWorkerContext(this, internalBlockingPool, workerPool, deploymentID, config, tccl);
- } else {
- return new WorkerContext(this, internalBlockingPool, workerPool, deploymentID, config, tccl);
- }
- }
-
- @Override
- public DnsClient createDnsClient(int port, String host) {
- return createDnsClient(new DnsClientOptions().setHost(host).setPort(port));
- }
-
- @Override
- public DnsClient createDnsClient() {
- return createDnsClient(new DnsClientOptions());
- }
-
- @Override
- public DnsClient createDnsClient(DnsClientOptions options) {
- String host = options.getHost();
- int port = options.getPort();
- if (host == null || port < 0) {
- DnsResolverProvider provider = new DnsResolverProvider(this, addressResolverOptions);
- InetSocketAddress address = provider.nameServerAddresses().get(0);
- // provide the host and port
- options = new DnsClientOptions(options)
- .setHost(address.getAddress().getHostAddress())
- .setPort(address.getPort());
- }
- return new DnsClientImpl(this, options);
- }
-
- private VertxMetrics initialiseMetrics(VertxOptions options) {
- if (options.getMetricsOptions() != null && options.getMetricsOptions().isEnabled()) {
- VertxMetricsFactory factory = options.getMetricsOptions().getFactory();
- if (factory == null) {
- factory = ServiceHelper.loadFactoryOrNull(VertxMetricsFactory.class);
- if (factory == null) {
- log.warn("Metrics has been set to enabled but no VertxMetricsFactory found on classpath");
- }
- }
- if (factory != null) {
- VertxMetrics metrics = factory.metrics(options);
- Objects.requireNonNull(metrics, "The metric instance created from " + factory + " cannot be null");
- return metrics;
- }
- }
- return null;
- }
-
- private ClusterManager getClusterManager(VertxOptions options) {
- ClusterManager mgr = options.getClusterManager();
- if (mgr == null) {
- String clusterManagerClassName = System.getProperty("vertx.cluster.managerClass");
- if (clusterManagerClassName != null) {
- // We allow specify a sys prop for the cluster manager factory which overrides ServiceLoader
- try {
- Class<?> clazz = Class.forName(clusterManagerClassName);
- mgr = (ClusterManager) clazz.newInstance();
- } catch (Exception e) {
- throw new IllegalStateException("Failed to instantiate " + clusterManagerClassName, e);
- }
- } else {
- mgr = ServiceHelper.loadFactoryOrNull(ClusterManager.class);
- if (mgr == null) {
- throw new IllegalStateException("No ClusterManagerFactory instances found on classpath");
- }
- }
- }
- return mgr;
- }
-
- private long scheduleTimeout(ContextImpl context, Handler<Long> handler, long delay, boolean periodic) {
- if (delay < 1) {
- throw new IllegalArgumentException("Cannot schedule a timer with delay < 1 ms");
- }
- long timerId = timeoutCounter.getAndIncrement();
- InternalTimerHandler task = new InternalTimerHandler(timerId, handler, periodic, delay, context);
- timeouts.put(timerId, task);
- context.addCloseHook(task);
- return timerId;
- }
-
- public static Context context() {
- Thread current = Thread.currentThread();
- if (current instanceof VertxThread) {
- return ((VertxThread) current).getContext();
- }
- return null;
- }
-
- public ContextImpl getContext() {
- ContextImpl context = (ContextImpl) context();
- if (context != null && context.owner == this) {
- return context;
- }
- return null;
- }
-
- public ClusterManager getClusterManager() {
- return clusterManager;
- }
-
- @Override
- public void close() {
- close(null);
- }
-
- private void closeClusterManager(Handler<AsyncResult<Void>> completionHandler) {
- if (clusterManager != null) {
- clusterManager.leave(ar -> {
- if (ar.failed()) {
- log.error("Failed to leave cluster", ar.cause());
- }
- if (completionHandler != null) {
- runOnContext(v -> completionHandler.handle(Future.succeededFuture()));
- }
- });
- } else if (completionHandler != null) {
- runOnContext(v -> completionHandler.handle(Future.succeededFuture()));
- }
- }
-
- @Override
- public synchronized void close(Handler<AsyncResult<Void>> completionHandler) {
- if (closed || eventBus == null) {
- // Just call the handler directly since pools shutdown
- if (completionHandler != null) {
- completionHandler.handle(Future.succeededFuture());
- }
- return;
- }
- closed = true;
-
- closeHooks.run(ar -> {
- deploymentManager.undeployAll(ar1 -> {
- HAManager haManager = haManager();
- Future<Void> haFuture = Future.future();
- if (haManager != null) {
- this.executeBlocking(fut -> {
- haManager.stop();
- fut.complete();
- }, false, haFuture);
- } else {
- haFuture.complete();
- }
- haFuture.setHandler(ar2 -> {
- addressResolver.close(ar3 -> {
- eventBus.close(ar4 -> {
- closeClusterManager(ar5 -> {
- // Copy set to prevent ConcurrentModificationException
- Set<HttpServerImpl> httpServers = new HashSet<>(sharedHttpServers.values());
- Set<NetServerImpl> netServers = new HashSet<>(sharedNetServers.values());
- sharedHttpServers.clear();
- sharedNetServers.clear();
-
- int serverCount = httpServers.size() + netServers.size();
-
- AtomicInteger serverCloseCount = new AtomicInteger();
-
- Handler<AsyncResult<Void>> serverCloseHandler = res -> {
- if (res.failed()) {
- log.error("Failure in shutting down server", res.cause());
- }
- if (serverCloseCount.incrementAndGet() == serverCount) {
- deleteCacheDirAndShutdown(completionHandler);
- }
- };
-
- for (HttpServerImpl server : httpServers) {
- server.closeAll(serverCloseHandler);
- }
- for (NetServerImpl server : netServers) {
- server.closeAll(serverCloseHandler);
- }
- if (serverCount == 0) {
- deleteCacheDirAndShutdown(completionHandler);
- }
- });
- });
- });
- });
- });
- });
- }
-
- @Override
- public void deployVerticle(Verticle verticle) {
- deployVerticle(verticle, new DeploymentOptions(), null);
- }
-
- @Override
- public void deployVerticle(Verticle verticle, Handler<AsyncResult<String>> completionHandler) {
- deployVerticle(verticle, new DeploymentOptions(), completionHandler);
- }
-
- @Override
- public void deployVerticle(String name, Handler<AsyncResult<String>> completionHandler) {
- deployVerticle(name, new DeploymentOptions(), completionHandler);
- }
-
- @Override
- public void deployVerticle(Verticle verticle, DeploymentOptions options) {
- deployVerticle(verticle, options, null);
- }
-
- @Override
- public void deployVerticle(Class<? extends Verticle> verticleClass, DeploymentOptions options) {
- deployVerticle(() -> {
- try {
- return verticleClass.newInstance();
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }, options);
- }
-
- @Override
- public void deployVerticle(Supplier<Verticle> verticleSupplier, DeploymentOptions options) {
- deployVerticle(verticleSupplier, options, null);
- }
-
- @Override
- public void deployVerticle(Verticle verticle, DeploymentOptions options, Handler<AsyncResult<String>> completionHandler) {
- if (options.getInstances() != 1) {
- throw new IllegalArgumentException("Can't specify > 1 instances for already created verticle");
- }
- deployVerticle(() -> verticle, options, completionHandler);
- }
-
- @Override
- public void deployVerticle(Class<? extends Verticle> verticleClass, DeploymentOptions options, Handler<AsyncResult<String>> completionHandler) {
- deployVerticle(() -> {
- try {
- return verticleClass.newInstance();
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }, options, completionHandler);
- }
-
- @Override
- public void deployVerticle(Supplier<Verticle> verticleSupplier, DeploymentOptions options, Handler<AsyncResult<String>> completionHandler) {
- boolean closed;
- synchronized (this) {
- closed = this.closed;
- }
- if (closed) {
- if (completionHandler != null) {
- completionHandler.handle(Future.failedFuture("Vert.x closed"));
- }
- } else {
- deploymentManager.deployVerticle(verticleSupplier, options, completionHandler);
- }
- }
-
- @Override
- public void deployVerticle(String name) {
- deployVerticle(name, new DeploymentOptions(), null);
- }
-
- @Override
- public void deployVerticle(String name, DeploymentOptions options) {
- deployVerticle(name, options, null);
- }
-
- @Override
- public void deployVerticle(String name, DeploymentOptions options, Handler<AsyncResult<String>> completionHandler) {
- if (options.isHa() && haManager() != null && haManager().isEnabled()) {
- haManager().deployVerticle(name, options, completionHandler);
- } else {
- deploymentManager.deployVerticle(name, options, completionHandler);
- }
- }
-
- @Override
- public String getNodeID() {
- return clusterManager.getNodeID();
- }
-
- @Override
- public void undeploy(String deploymentID) {
- undeploy(deploymentID, res -> {
- });
- }
-
- @Override
- public void undeploy(String deploymentID, Handler<AsyncResult<Void>> completionHandler) {
- HAManager haManager = haManager();
- Future<Void> haFuture = Future.future();
- if (haManager != null && haManager.isEnabled()) {
- this.executeBlocking(fut -> {
- haManager.removeFromHA(deploymentID);
- fut.complete();
- }, false, haFuture);
- } else {
- haFuture.complete();
- }
- haFuture.compose(v -> {
- Future<Void> deploymentFuture = Future.future();
- deploymentManager.undeployVerticle(deploymentID, deploymentFuture);
- return deploymentFuture;
- }).setHandler(completionHandler);
- }
-
- @Override
- public Set<String> deploymentIDs() {
- return deploymentManager.deployments();
- }
-
- @Override
- public void registerVerticleFactory(VerticleFactory factory) {
- deploymentManager.registerVerticleFactory(factory);
- }
-
- @Override
- public void unregisterVerticleFactory(VerticleFactory factory) {
- deploymentManager.unregisterVerticleFactory(factory);
- }
-
- @Override
- public Set<VerticleFactory> verticleFactories() {
- return deploymentManager.verticleFactories();
- }
-
- @Override
- public <T> void executeBlockingInternal(Handler<Future<T>> blockingCodeHandler, Handler<AsyncResult<T>> resultHandler) {
- ContextImpl context = getOrCreateContext();
-
- context.executeBlockingInternal(blockingCodeHandler, resultHandler);
- }
-
- @Override
- public <T> void executeBlocking(Handler<Future<T>> blockingCodeHandler, boolean ordered,
- Handler<AsyncResult<T>> asyncResultHandler) {
- ContextImpl context = getOrCreateContext();
- context.executeBlocking(blockingCodeHandler, ordered, asyncResultHandler);
- }
-
- @Override
- public <T> void executeBlocking(Handler<Future<T>> blockingCodeHandler,
- Handler<AsyncResult<T>> asyncResultHandler) {
- executeBlocking(blockingCodeHandler, true, asyncResultHandler);
- }
-
- @Override
- public boolean isClustered() {
- return clusterManager != null;
- }
-
- @Override
- public EventLoopGroup nettyEventLoopGroup() {
- return eventLoopGroup;
- }
-
- // For testing
- public void simulateKill() {
- if (haManager() != null) {
- haManager().simulateKill();
- }
- }
-
- @Override
- public Deployment getDeployment(String deploymentID) {
- return deploymentManager.getDeployment(deploymentID);
- }
-
- @Override
- public synchronized void failoverCompleteHandler(FailoverCompleteHandler failoverCompleteHandler) {
- if (haManager() != null) {
- haManager().setFailoverCompleteHandler(failoverCompleteHandler);
- }
- }
-
- @Override
- public boolean isKilled() {
- return haManager().isKilled();
- }
-
- @Override
- public void failDuringFailover(boolean fail) {
- if (haManager() != null) {
- haManager().failDuringFailover(fail);
- }
- }
-
- @Override
- public VertxMetrics metricsSPI() {
- return metrics;
- }
-
- @Override
- public File resolveFile(String fileName) {
- return fileResolver.resolveFile(fileName);
- }
-
- @Override
- public void resolveAddress(String hostname, Handler<AsyncResult<InetAddress>> resultHandler) {
- addressResolver.resolveHostname(hostname, resultHandler);
- }
-
- @Override
- public AddressResolver addressResolver() {
- return addressResolver;
- }
-
- @Override
- public AddressResolverGroup<InetSocketAddress> nettyAddressResolverGroup() {
- return addressResolver.nettyAddressResolverGroup();
- }
-
- @SuppressWarnings({"unchecked", "rawtypes"})
- private void deleteCacheDirAndShutdown(Handler<AsyncResult<Void>> completionHandler) {
- executeBlockingInternal(fut -> {
- try {
- fileResolver.close();
- fut.complete();
- } catch (IOException e) {
- fut.tryFail(e);
- }
- }, ar -> {
-
- workerPool.close();
- internalBlockingPool.close();
- new ArrayList<>(namedWorkerPools.values()).forEach(WorkerPool::close);
-
- acceptorEventLoopGroup.shutdownGracefully(0, 10, TimeUnit.SECONDS).addListener(new GenericFutureListener() {
- @Override
- public void operationComplete(io.netty.util.concurrent.Future future) throws Exception {
- if (!future.isSuccess()) {
- log.warn("Failure in shutting down acceptor event loop group", future.cause());
- }
- eventLoopGroup.shutdownGracefully(0, 10, TimeUnit.SECONDS).addListener(new GenericFutureListener() {
- @Override
- public void operationComplete(io.netty.util.concurrent.Future future) throws Exception {
- if (!future.isSuccess()) {
- log.warn("Failure in shutting down event loop group", future.cause());
- }
- if (metrics != null) {
- metrics.close();
- }
-
- checker.close();
-
- if (completionHandler != null) {
- eventLoopThreadFactory.newThread(() -> {
- completionHandler.handle(Future.succeededFuture());
- }).start();
- }
- }
- });
- }
- });
- });
- }
-
- public HAManager haManager() {
- return haManager;
- }
-
- private class InternalTimerHandler implements Handler<Void>, Closeable {
- final Handler<Long> handler;
- final boolean periodic;
- final long timerID;
- final ContextImpl context;
- final java.util.concurrent.Future<?> future;
- final AtomicBoolean cancelled;
-
- boolean cancel() {
- if (cancelled.compareAndSet(false, true)) {
- if (metrics != null) {
- metrics.timerEnded(timerID, true);
- }
- future.cancel(false);
- return true;
- } else {
- return false;
- }
- }
-
- InternalTimerHandler(long timerID, Handler<Long> runnable, boolean periodic, long delay, ContextImpl context) {
- this.context = context;
- this.timerID = timerID;
- this.handler = runnable;
- this.periodic = periodic;
- this.cancelled = new AtomicBoolean();
- EventLoop el = context.nettyEventLoop();
- Runnable toRun = () -> context.runOnContext(this);
- if (periodic) {
- future = el.scheduleAtFixedRate(toRun, delay, delay, TimeUnit.MILLISECONDS);
- } else {
- future = el.schedule(toRun, delay, TimeUnit.MILLISECONDS);
- }
- if (metrics != null) {
- metrics.timerCreated(timerID);
- }
- }
-
- public void handle(Void v) {
- if (!cancelled.get()) {
- try {
- handler.handle(timerID);
- } finally {
- if (!periodic) {
- // Clean up after it's fired
- cleanupNonPeriodic();
- }
- }
- }
- }
-
- private void cleanupNonPeriodic() {
- VertxImpl.this.timeouts.remove(timerID);
- if (metrics != null) {
- metrics.timerEnded(timerID, false);
- }
- ContextImpl context = getContext();
- if (context != null) {
- context.removeCloseHook(this);
- }
- }
-
- // Called via Context close hook when Verticle is undeployed
- public void close(Handler<AsyncResult<Void>> completionHandler) {
- VertxImpl.this.timeouts.remove(timerID);
- cancel();
- completionHandler.handle(Future.succeededFuture());
- }
-
- }
-
- /*
- *
- * This class is optimised for performance when used on the same event loop that is was passed to the handler with.
- * However it can be used safely from other threads.
- *
- * The internal state is protected using the synchronized keyword. If always used on the same event loop, then
- * we benefit from biased locking which makes the overhead of synchronized near zero.
- *
- */
- private class TimeoutStreamImpl implements TimeoutStream, Handler<Long> {
-
- private final long delay;
- private final boolean periodic;
-
- private Long id;
- private Handler<Long> handler;
- private Handler<Void> endHandler;
- private long demand;
-
- public TimeoutStreamImpl(long delay, boolean periodic) {
- this.delay = delay;
- this.periodic = periodic;
- this.demand = Long.MAX_VALUE;
- }
-
- @Override
- public synchronized void handle(Long event) {
- try {
- if (demand > 0) {
- demand--;
- handler.handle(event);
- }
- } finally {
- if (!periodic && endHandler != null) {
- endHandler.handle(null);
- }
- }
- }
-
- @Override
- public synchronized TimeoutStream fetch(long amount) {
- demand += amount;
- if (demand < 0) {
- demand = Long.MAX_VALUE;
- }
- return this;
- }
-
- @Override
- public TimeoutStream exceptionHandler(Handler<Throwable> handler) {
- return this;
- }
-
- @Override
- public void cancel() {
- if (id != null) {
- VertxImpl.this.cancelTimer(id);
- }
- }
-
- @Override
- public synchronized TimeoutStream handler(Handler<Long> handler) {
- if (handler != null) {
- if (id != null) {
- throw new IllegalStateException();
- }
- this.handler = handler;
- id = scheduleTimeout(getOrCreateContext(), this, delay, periodic);
- } else {
- cancel();
- }
- return this;
- }
-
- @Override
- public synchronized TimeoutStream pause() {
- demand = 0;
- return this;
- }
-
- @Override
- public synchronized TimeoutStream resume() {
- demand = Long.MAX_VALUE;
- return this;
- }
-
- @Override
- public synchronized TimeoutStream endHandler(Handler<Void> endHandler) {
- this.endHandler = endHandler;
- return this;
- }
- }
-
- class SharedWorkerPool extends WorkerPool {
-
- private final String name;
- private int refCount = 1;
-
- @SuppressWarnings("rawtypes")
- SharedWorkerPool(String name, ExecutorService workerExec, PoolMetrics workerMetrics) {
- super(workerExec, workerMetrics);
- this.name = name;
- }
-
- @Override
- void close() {
- synchronized (VertxImpl.this) {
- if (refCount > 0) {
- refCount = 0;
- super.close();
- }
- }
- }
-
- void release() {
- synchronized (VertxImpl.this) {
- if (--refCount == 0) {
- namedWorkerPools.remove(name);
- super.close();
- }
- }
- }
- }
-
- @Override
- public WorkerExecutorImpl createSharedWorkerExecutor(String name) {
- return createSharedWorkerExecutor(name, defaultWorkerPoolSize);
- }
-
- @Override
- public WorkerExecutorImpl createSharedWorkerExecutor(String name, int poolSize) {
- return createSharedWorkerExecutor(name, poolSize, defaultWorkerMaxExecTime);
- }
-
- @Override
- public synchronized WorkerExecutorImpl createSharedWorkerExecutor(String name, int poolSize, long maxExecuteTime) {
- return createSharedWorkerExecutor(name, poolSize, maxExecuteTime, TimeUnit.NANOSECONDS);
- }
-
- @SuppressWarnings("rawtypes")
- @Override
- public synchronized WorkerExecutorImpl createSharedWorkerExecutor(String name, int poolSize, long maxExecuteTime, TimeUnit maxExecuteTimeUnit) {
- if (poolSize < 1) {
- throw new IllegalArgumentException("poolSize must be > 0");
- }
- if (maxExecuteTime < 1) {
- throw new IllegalArgumentException("maxExecuteTime must be > 0");
- }
- SharedWorkerPool sharedWorkerPool = namedWorkerPools.get(name);
- if (sharedWorkerPool == null) {
- ExecutorService workerExec = Executors.newFixedThreadPool(poolSize, new VertxThreadFactory(name + "-", checker, true, maxExecuteTime, maxExecuteTimeUnit));
- PoolMetrics workerMetrics = metrics != null ? metrics.createPoolMetrics("worker", name, poolSize) : null;
- namedWorkerPools.put(name, sharedWorkerPool = new SharedWorkerPool(name, workerExec, workerMetrics));
- } else {
- sharedWorkerPool.refCount++;
- }
- ContextImpl context = getOrCreateContext();
- WorkerExecutorImpl namedExec = new WorkerExecutorImpl(context, sharedWorkerPool);
- context.addCloseHook(namedExec);
- return namedExec;
- }
-
- @Override
- public Vertx exceptionHandler(Handler<Throwable> handler) {
- exceptionHandler = handler;
- return this;
- }
-
- @Override
- public Handler<Throwable> exceptionHandler() {
- return exceptionHandler;
- }
-
- @Override
- public void addCloseHook(Closeable hook) {
- closeHooks.add(hook);
- }
-
- @Override
- public void removeCloseHook(Closeable hook) {
- closeHooks.remove(hook);
- }
-}
diff --git a/foundations/foundation-vertx/src/main/java/io/vertx/ext/web/impl/HttpServerRequestUtils.java b/foundations/foundation-vertx/src/main/java/io/vertx/ext/web/impl/HttpServerRequestUtils.java
deleted file mode 100644
index 257770e..0000000
--- a/foundations/foundation-vertx/src/main/java/io/vertx/ext/web/impl/HttpServerRequestUtils.java
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * 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 io.vertx.ext.web.impl;
-
-import io.vertx.core.http.HttpServerRequest;
-
-public interface HttpServerRequestUtils {
- static void setPath(HttpServerRequest request, String path) {
- ((HttpServerRequestWrapper) request).setPath(path);
- }
-}
diff --git a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/VertxServerRequestToHttpServletRequest.java b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/VertxServerRequestToHttpServletRequest.java
index a670480..516419d 100644
--- a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/VertxServerRequestToHttpServletRequest.java
+++ b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/VertxServerRequestToHttpServletRequest.java
@@ -95,10 +95,10 @@ public class VertxServerRequestToHttpServletRequest extends AbstractHttpServletR
@Override
public Cookie[] getCookies() {
if (cookies == null) {
- Set<io.vertx.ext.web.Cookie> vertxCookies = context.cookies();
+ Map<String, io.vertx.core.http.Cookie> vertxCookies = context.cookieMap();
Cookie tmpCookies[] = new Cookie[vertxCookies.size()];
int idx = 0;
- for (io.vertx.ext.web.Cookie oneVertxCookie : vertxCookies) {
+ for (io.vertx.core.http.Cookie oneVertxCookie : vertxCookies.values()) {
Cookie cookie = new Cookie(oneVertxCookie.getName(), oneVertxCookie.getValue());
tmpCookies[idx] = cookie;
idx++;
diff --git a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/stream/InputStreamToReadStream.java b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/stream/InputStreamToReadStream.java
index 18ed274..6a0592e 100644
--- a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/stream/InputStreamToReadStream.java
+++ b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/stream/InputStreamToReadStream.java
@@ -28,6 +28,7 @@ import io.vertx.core.AsyncResult;
import io.vertx.core.Context;
import io.vertx.core.Future;
import io.vertx.core.Handler;
+import io.vertx.core.Promise;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.streams.ReadStream;
@@ -120,7 +121,7 @@ public class InputStreamToReadStream implements ReadStream<Buffer> {
}
}
- private synchronized void readInWorker(Future<ReadResult> future) {
+ private synchronized void readInWorker(Promise<ReadResult> future) {
try {
ReadResult readResult = new ReadResult();
readResult.doRead();
diff --git a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/stream/OutputStreamToWriteStream.java b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/stream/OutputStreamToWriteStream.java
index 9e69759..d449ddf 100644
--- a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/stream/OutputStreamToWriteStream.java
+++ b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/stream/OutputStreamToWriteStream.java
@@ -24,9 +24,10 @@ import java.util.concurrent.ConcurrentLinkedQueue;
import org.apache.servicecomb.foundation.common.io.AsyncCloseable;
+import io.vertx.core.AsyncResult;
import io.vertx.core.Context;
-import io.vertx.core.Future;
import io.vertx.core.Handler;
+import io.vertx.core.Promise;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.streams.WriteStream;
@@ -89,19 +90,24 @@ public class OutputStreamToWriteStream implements WriteStream<Buffer>, AsyncClos
@Override
public synchronized WriteStream<Buffer> write(Buffer data) {
+ return write(data, ar -> {
+ if (ar.failed()) {
+ handleException(ar.cause());
+ }
+ });
+ }
+
+ @Override
+ public WriteStream<Buffer> write(Buffer data, Handler<AsyncResult<Void>> handler) {
currentBufferCount++;
buffers.add(data);
context.executeBlocking(this::writeInWorker,
true,
- ar -> {
- if (ar.failed()) {
- handleException(ar.cause());
- }
- });
+ handler);
return this;
}
- protected void writeInWorker(Future<Object> future) {
+ protected void writeInWorker(Promise<Void> future) {
while (true) {
Buffer buffer = buffers.poll();
if (buffer == null) {
@@ -127,6 +133,11 @@ public class OutputStreamToWriteStream implements WriteStream<Buffer>, AsyncClos
@Override
public void end() {
+ end((Handler<AsyncResult<Void>>)null);
+ }
+
+ @Override
+ public void end(Handler<AsyncResult<Void>> handler) {
close();
}
diff --git a/foundations/foundation-vertx/src/test/java/io/vertx/ext/web/impl/TestHttpServerRequestUtils.java b/foundations/foundation-vertx/src/test/java/io/vertx/ext/web/impl/TestHttpServerRequestUtils.java
index 7730bb6..0b4fdbc 100644
--- a/foundations/foundation-vertx/src/test/java/io/vertx/ext/web/impl/TestHttpServerRequestUtils.java
+++ b/foundations/foundation-vertx/src/test/java/io/vertx/ext/web/impl/TestHttpServerRequestUtils.java
@@ -28,14 +28,6 @@ import mockit.Mocked;
public class TestHttpServerRequestUtils {
@Test
- public void setPath(@Mocked HttpServerRequest request) {
- HttpServerRequestWrapper wrapper = new HttpServerRequestWrapper(request);
- HttpServerRequestUtils.setPath(wrapper, "abc");
-
- Assert.assertEquals("abc", wrapper.path());
- }
-
- @Test
public void VertxServerRequestToHttpServletRequest(@Mocked RoutingContext context,
@Mocked HttpServerRequest request) {
HttpServerRequestWrapper wrapper = new HttpServerRequestWrapper(request);
diff --git a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/TestVertxUtils.java b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/TestVertxUtils.java
index 3da853d..8683b1d 100644
--- a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/TestVertxUtils.java
+++ b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/TestVertxUtils.java
@@ -28,7 +28,6 @@ import javax.xml.ws.Holder;
import org.apache.commons.io.FileUtils;
import org.apache.servicecomb.foundation.test.scaffolding.config.ArchaiusUtils;
import org.apache.servicecomb.foundation.vertx.stream.BufferInputStream;
-import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.Test;
@@ -60,7 +59,8 @@ public class TestVertxUtils {
public void testCreateVertxWithFileCPResolving() {
// Prepare
ArchaiusUtils.resetConfig();
- String cacheDirBase = System.getProperty(FileResolver.CACHE_DIR_BASE_PROP_NAME, ".vertx");
+ String cacheDirBase = System.getProperty(FileResolver.CACHE_DIR_BASE_PROP_NAME,
+ System.getProperty("java.io.tmpdir", ".") + File.separator + "vertx-cache");
File file = new File(cacheDirBase);
// create .vertx folder
@@ -92,7 +92,7 @@ public class TestVertxUtils {
@Test
public void testVertxUtilsInitWithOptions() {
VertxOptions oOptions = new VertxOptions();
- oOptions.setClustered(false);
+ oOptions.getEventBusOptions().setClustered(false);
Vertx vertx = VertxUtils.init(oOptions);
Assert.assertNotEquals(null, vertx);
diff --git a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/client/TestClientPoolManager.java b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/client/TestClientPoolManager.java
index 6709ca5..dc86d9b 100644
--- a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/client/TestClientPoolManager.java
+++ b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/client/TestClientPoolManager.java
@@ -145,11 +145,11 @@ public class TestClientPoolManager {
HttpClientWithContext notMatchPool = new HttpClientWithContext(null, null);
pools.add(notMatchPool);
- new Expectations(VertxImpl.class) {
+ new Expectations() {
{
factory.createClientPool(context);
result = new HttpClientWithContext(null, null);
- VertxImpl.context();
+ Vertx.currentContext();
result = context;
}
};
@@ -167,9 +167,9 @@ public class TestClientPoolManager {
pools.add(pool1);
pools.add(pool2);
- new Expectations(VertxImpl.class) {
+ new Expectations() {
{
- VertxImpl.context();
+ Vertx.currentContext();
result = null;
}
};
@@ -188,9 +188,9 @@ public class TestClientPoolManager {
HttpClientWithContext pool = new HttpClientWithContext(null, null);
pools.add(pool);
- new Expectations(VertxImpl.class) {
+ new Expectations() {
{
- VertxImpl.context();
+ Vertx.currentContext();
result = null;
}
};
@@ -203,9 +203,9 @@ public class TestClientPoolManager {
HttpClientWithContext pool = new HttpClientWithContext(null, null);
pools.add(pool);
- new Expectations(VertxImpl.class) {
+ new Expectations() {
{
- VertxImpl.context();
+ Vertx.currentContext();
result = otherContext;
otherContext.owner();
result = otherVertx;
@@ -220,9 +220,9 @@ public class TestClientPoolManager {
HttpClientWithContext pool = new HttpClientWithContext(null, null);
pools.add(pool);
- new Expectations(VertxImpl.class) {
+ new Expectations() {
{
- VertxImpl.context();
+ Vertx.currentContext();
result = workerContext;
workerContext.owner();
result = vertx;
diff --git a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/client/tcp/TestTcpClientConnection.java b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/client/tcp/TestTcpClientConnection.java
index 860e141..c2756e9 100644
--- a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/client/tcp/TestTcpClientConnection.java
+++ b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/client/tcp/TestTcpClientConnection.java
@@ -172,6 +172,7 @@ public class TestTcpClientConnection {
{
tcpClientPackage.getMsgId();
result = msgId;
+ tcpClientConnection.write((ByteBuf) any);
}
};
new MockUp<Context>(context) {
diff --git a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/http/TestReadStreamPart.java b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/http/TestReadStreamPart.java
index 5080839..50c151d 100644
--- a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/http/TestReadStreamPart.java
+++ b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/http/TestReadStreamPart.java
@@ -38,12 +38,13 @@ import org.junit.Test;
import org.junit.rules.ExpectedException;
import io.vertx.core.Handler;
+import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.file.FileSystemException;
import io.vertx.core.file.OpenOptions;
import io.vertx.core.http.HttpClientResponse;
-import io.vertx.core.impl.ContextInternal;
-import io.vertx.core.impl.SyncVertx;
+import io.vertx.core.impl.SyncContext;
+import io.vertx.core.impl.VertxInternal;
import io.vertx.core.streams.WriteStream;
import mockit.Expectations;
import mockit.Mock;
@@ -51,9 +52,9 @@ import mockit.MockUp;
import mockit.Mocked;
public class TestReadStreamPart {
- static SyncVertx vertx = new SyncVertx();
+ static Vertx vertx = Vertx.vertx();
- static ContextInternal context = vertx.getContext();
+ static SyncContext context = new SyncContext();
static String src = "src";
@@ -68,6 +69,7 @@ public class TestReadStreamPart {
@Before
public void setup() throws IOException {
+ context.setOwner((VertxInternal)vertx);
inputStream.reset();
}
diff --git a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/http/TestVertxServerRequestToHttpServletRequest.java b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/http/TestVertxServerRequestToHttpServletRequest.java
index 0ec60b8..bcec03f 100644
--- a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/http/TestVertxServerRequestToHttpServletRequest.java
+++ b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/http/TestVertxServerRequestToHttpServletRequest.java
@@ -20,9 +20,8 @@ package org.apache.servicecomb.foundation.vertx.http;
import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
-import java.util.LinkedHashSet;
+import java.util.LinkedHashMap;
import java.util.Map;
-import java.util.Set;
import javax.servlet.AsyncContext;
import javax.servlet.ServletInputStream;
@@ -118,12 +117,12 @@ public class TestVertxServerRequestToHttpServletRequest {
@Test
public void testGetCookies() {
- Set<io.vertx.ext.web.Cookie> vertxCookies = new LinkedHashSet<>();
- vertxCookies.add(io.vertx.ext.web.Cookie.cookie("c1", "c1v"));
- vertxCookies.add(io.vertx.ext.web.Cookie.cookie("c2", "c2v"));
+ Map<String, io.vertx.core.http.Cookie> vertxCookies = new LinkedHashMap<>();
+ vertxCookies.put("c1", io.vertx.core.http.Cookie.cookie("c1", "c1v"));
+ vertxCookies.put("c2", io.vertx.core.http.Cookie.cookie("c2", "c2v"));
new Expectations() {
{
- context.cookies();
+ context.cookieMap();
result = vertxCookies;
}
};
diff --git a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/http/TestVertxServerResponseToHttpServletResponse.java b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/http/TestVertxServerResponseToHttpServletResponse.java
index 215ea56..e82c662 100644
--- a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/http/TestVertxServerResponseToHttpServletResponse.java
+++ b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/http/TestVertxServerResponseToHttpServletResponse.java
@@ -42,14 +42,13 @@ import org.junit.rules.ExpectedException;
import io.vertx.core.AsyncResult;
import io.vertx.core.Context;
-import io.vertx.core.Future;
import io.vertx.core.Handler;
import io.vertx.core.MultiMap;
+import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.http.HttpServerResponse;
import io.vertx.core.impl.SyncContext;
-import io.vertx.core.impl.VertxImpl;
import io.vertx.core.streams.WriteStream;
import mockit.Deencapsulation;
import mockit.Expectations;
@@ -139,9 +138,9 @@ public class TestVertxServerResponseToHttpServletResponse {
}
}.getMockInstance();
- new Expectations(VertxImpl.class) {
+ new Expectations() {
{
- VertxImpl.context();
+ Vertx.currentContext();
result = context;
}
};
@@ -154,7 +153,7 @@ public class TestVertxServerResponseToHttpServletResponse {
}
@Mock
- <T> void executeBlocking(Handler<Future<T>> blockingCodeHandler, boolean ordered,
+ <T> void executeBlocking(Handler<Promise<T>> blockingCodeHandler, boolean ordered,
Handler<AsyncResult<T>> resultHandler) {
SyncContext.syncExecuteBlocking(blockingCodeHandler, resultHandler);
}
@@ -170,9 +169,9 @@ public class TestVertxServerResponseToHttpServletResponse {
@Test
public void construct_invalid() throws IOException {
- new Expectations(VertxImpl.class) {
+ new Expectations() {
{
- VertxImpl.context();
+ Vertx.currentContext();
result = null;
}
};
@@ -270,9 +269,9 @@ public class TestVertxServerResponseToHttpServletResponse {
@Test
public void flushBuffer_diffContext() throws IOException {
- new Expectations(VertxImpl.class) {
+ new Expectations() {
{
- VertxImpl.context();
+ Vertx.currentContext();
result = null;
}
};
diff --git a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/stream/TestPumpFromPart.java b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/stream/TestPumpFromPart.java
index 9ba8a28..befcc2e 100644
--- a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/stream/TestPumpFromPart.java
+++ b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/stream/TestPumpFromPart.java
@@ -33,7 +33,7 @@ import org.junit.Assert;
import org.junit.Test;
import io.vertx.core.Context;
-import io.vertx.core.Future;
+import io.vertx.core.Promise;
import io.vertx.core.impl.SyncContext;
import mockit.Expectations;
import mockit.Mock;
@@ -121,7 +121,7 @@ public class TestPumpFromPart {
public void pump_read_error() throws IOException {
new MockUp<InputStreamToReadStream>() {
@Mock
- void readInWorker(Future<ReadResult> future) {
+ void readInWorker(Promise<ReadResult> future) {
future.fail(error);
}
};
diff --git a/transports/transport-rest/transport-rest-vertx/src/test/java/org/apache/servicecomb/transport/rest/vertx/TestVertxRestDispatcher.java b/transports/transport-rest/transport-rest-vertx/src/test/java/org/apache/servicecomb/transport/rest/vertx/TestVertxRestDispatcher.java
index f02ee0e..2d8185c 100644
--- a/transports/transport-rest/transport-rest-vertx/src/test/java/org/apache/servicecomb/transport/rest/vertx/TestVertxRestDispatcher.java
+++ b/transports/transport-rest/transport-rest-vertx/src/test/java/org/apache/servicecomb/transport/rest/vertx/TestVertxRestDispatcher.java
@@ -42,11 +42,13 @@ import org.junit.Before;
import org.junit.Test;
import io.netty.handler.codec.http.multipart.HttpPostRequestDecoder.ErrorDataDecoderException;
+import io.vertx.codegen.annotations.Nullable;
import io.vertx.core.AsyncResult;
import io.vertx.core.Context;
import io.vertx.core.Handler;
import io.vertx.core.MultiMap;
import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.Cookie;
import io.vertx.core.http.HttpMethod;
import io.vertx.core.http.HttpServerRequest;
import io.vertx.core.http.HttpServerResponse;
@@ -292,12 +294,6 @@ public class TestVertxRestDispatcher {
}
}.getMockInstance();
- new Expectations(VertxImpl.class) {
- {
- VertxImpl.context();
- result = context;
- }
- };
Deencapsulation.invoke(dispatcher, "onRequest", routingContext);
Assert.assertEquals(VertxRestInvocation.class, map.get(RestConst.REST_PRODUCER_INVOCATION).getClass());
@@ -381,12 +377,22 @@ class MockHttpServerResponse implements HttpServerResponse {
}
@Override
+ public void end(Handler<AsyncResult<Void>> handler) {
+
+ }
+
+ @Override
public void end(String chunk) {
responseEnded = true;
responseChunk = chunk;
}
@Override
+ public void end(String s, Handler<AsyncResult<Void>> handler) {
+
+ }
+
+ @Override
public HttpServerResponse exceptionHandler(Handler<Throwable> handler) {
return null;
}
@@ -397,6 +403,11 @@ class MockHttpServerResponse implements HttpServerResponse {
}
@Override
+ public HttpServerResponse write(Buffer buffer, Handler<AsyncResult<Void>> handler) {
+ return null;
+ }
+
+ @Override
public HttpServerResponse setWriteQueueMaxSize(int maxSize) {
return null;
}
@@ -492,11 +503,21 @@ class MockHttpServerResponse implements HttpServerResponse {
}
@Override
+ public HttpServerResponse write(String s, String s1, Handler<AsyncResult<Void>> handler) {
+ return null;
+ }
+
+ @Override
public HttpServerResponse write(String chunk) {
return null;
}
@Override
+ public HttpServerResponse write(String s, Handler<AsyncResult<Void>> handler) {
+ return null;
+ }
+
+ @Override
public HttpServerResponse writeContinue() {
return null;
}
@@ -507,11 +528,21 @@ class MockHttpServerResponse implements HttpServerResponse {
}
@Override
+ public void end(String s, String s1, Handler<AsyncResult<Void>> handler) {
+
+ }
+
+ @Override
public void end(Buffer chunk) {
}
@Override
+ public void end(Buffer buffer, Handler<AsyncResult<Void>> handler) {
+
+ }
+
+ @Override
public HttpServerResponse sendFile(String filename, long offset, long length) {
return null;
}
@@ -589,4 +620,14 @@ class MockHttpServerResponse implements HttpServerResponse {
public HttpServerResponse writeCustomFrame(int type, int flags, Buffer payload) {
return null;
}
+
+ @Override
+ public HttpServerResponse addCookie(Cookie cookie) {
+ return null;
+ }
+
+ @Override
+ public @Nullable Cookie removeCookie(String s, boolean b) {
+ return null;
+ }
}