You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@servicecomb.apache.org by GitBox <gi...@apache.org> on 2018/04/22 08:32:26 UTC
[GitHub] liubao68 closed pull request #667: [SCB-486] Edge route download
liubao68 closed pull request #667: [SCB-486] Edge route download
URL: https://github.com/apache/incubator-servicecomb-java-chassis/pull/667
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/RestConst.java b/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/RestConst.java
index 7e45aa8cd..1261ca9f6 100644
--- a/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/RestConst.java
+++ b/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/RestConst.java
@@ -54,4 +54,6 @@ private RestConst() {
public static final String REST_REQUEST = "servicecomb-rest-request";
public static final String CONSUMER_HEADER = "servicecomb-rest-consumer-header";
+
+ public static final String READ_STREAM_PART = "servicecomb-readStreamPart";
}
diff --git a/demo/demo-edge/business-2.0.0/src/main/java/org/apache/servicecomb/demo/edge/business/Impl.java b/demo/demo-edge/business-2.0.0/src/main/java/org/apache/servicecomb/demo/edge/business/Impl.java
index 3db2a7cb6..e24f36ebd 100644
--- a/demo/demo-edge/business-2.0.0/src/main/java/org/apache/servicecomb/demo/edge/business/Impl.java
+++ b/demo/demo-edge/business-2.0.0/src/main/java/org/apache/servicecomb/demo/edge/business/Impl.java
@@ -17,18 +17,39 @@
package org.apache.servicecomb.demo.edge.business;
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.RandomAccessFile;
+import java.nio.charset.StandardCharsets;
+
+import org.apache.commons.io.FileUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.servicecomb.demo.edge.model.AppClientDataRsp;
import org.apache.servicecomb.demo.edge.model.ChannelRequestBase;
import org.apache.servicecomb.demo.edge.model.ResultWithInstance;
import org.apache.servicecomb.provider.rest.common.RestSchema;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
+import io.swagger.annotations.ApiResponse;
+import io.swagger.annotations.ApiResponses;
+
@RestSchema(schemaId = "news-v2")
@RequestMapping(path = "/business/v2")
public class Impl {
+ File tempDir = new File("target/downloadTemp");
+
+ public Impl() throws IOException {
+ FileUtils.forceMkdir(tempDir);
+ }
+
@RequestMapping(path = "/channel/news/subscribe", method = RequestMethod.POST)
public AppClientDataRsp subscribeNewsColumn(@RequestBody ChannelRequestBase request) {
AppClientDataRsp response = new AppClientDataRsp();
@@ -46,4 +67,30 @@ public ResultWithInstance add(int x, int y) {
public ResultWithInstance dec(int x, int y) {
return ResultWithInstance.create(x - y);
}
+
+ @GetMapping(path = "/download")
+ @ApiResponses({
+ @ApiResponse(code = 200, response = File.class, message = ""),
+ })
+ public ResponseEntity<InputStream> download() throws IOException {
+ return ResponseEntity
+ .ok()
+ .header(HttpHeaders.CONTENT_TYPE, MediaType.TEXT_PLAIN_VALUE)
+ .header(HttpHeaders.CONTENT_DISPOSITION, "attachment;filename=download.txt")
+ .body(new ByteArrayInputStream("download".getBytes(StandardCharsets.UTF_8)));
+ }
+
+ protected File createBigFile() throws IOException {
+ File file = new File(tempDir, "bigFile.txt");
+ file.delete();
+ RandomAccessFile randomAccessFile = new RandomAccessFile(file, "rw");
+ randomAccessFile.setLength(10 * 1024 * 1024);
+ randomAccessFile.close();
+ return file;
+ }
+
+ @GetMapping(path = "/bigFile")
+ public File bigFile() throws IOException {
+ return createBigFile();
+ }
}
diff --git a/demo/demo-edge/consumer/src/main/java/org/apache/servicecomb/demo/edge/consumer/Consumer.java b/demo/demo-edge/consumer/src/main/java/org/apache/servicecomb/demo/edge/consumer/Consumer.java
index bd2abad71..284a8832e 100644
--- a/demo/demo-edge/consumer/src/main/java/org/apache/servicecomb/demo/edge/consumer/Consumer.java
+++ b/demo/demo-edge/consumer/src/main/java/org/apache/servicecomb/demo/edge/consumer/Consumer.java
@@ -17,11 +17,13 @@
package org.apache.servicecomb.demo.edge.consumer;
+import java.io.InputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.servicecomb.core.Endpoint;
import org.apache.servicecomb.core.endpoint.EndpointsCache;
@@ -34,6 +36,7 @@
import org.apache.servicecomb.serviceregistry.api.registry.Microservice;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.util.Assert;
@@ -75,6 +78,9 @@ public Consumer() {
public void run() {
prepareEdge();
+ testDownload();
+ testDownloadBigFile();
+
invoke("/v1/add", 2, 1, addV1Result);
invoke("/v1/add", 3, 1, addV1Result);
invoke("/v1/add", 4, 1, addV1Result);
@@ -100,6 +106,36 @@ public void run() {
checkResult("v2/dec", decV2Result, "2.0.0");
}
+ protected void testDownloadBigFile() {
+ String url = edgePrefix + "/v2/bigFile";
+ AtomicInteger size = new AtomicInteger();
+
+ template.execute(url, HttpMethod.GET, req -> {
+ }, resp -> {
+ byte[] buf = new byte[1 * 1024 * 1024];
+ try (InputStream is = resp.getBody()) {
+ for (;;) {
+ int len = is.read(buf);
+ if (len == -1) {
+ break;
+ }
+
+ size.addAndGet(len);
+ }
+ }
+ return null;
+ });
+ Assert.isTrue(size.get() == 10 * 1024 * 1024);
+ System.out.println("test download bigFile finished");
+ }
+
+ protected void testDownload() {
+ String url = edgePrefix + "/v2/download";
+ String content = template.getForObject(url, String.class);
+ Assert.isTrue("download".equals(content));
+ System.out.println("test download finished");
+ }
+
private void checkResult(String name, List<ResultWithInstance> results, String... expectedVersions) {
Set<String> versions = new HashSet<>();
Set<String> remained = new HashSet<>(Arrays.asList(expectedVersions));
diff --git a/demo/perf/src/main/java/org/apache/servicecomb/demo/perf/RedisClientUtils.java b/demo/perf/src/main/java/org/apache/servicecomb/demo/perf/RedisClientUtils.java
index b8e5add89..0ce693a8a 100644
--- a/demo/perf/src/main/java/org/apache/servicecomb/demo/perf/RedisClientUtils.java
+++ b/demo/perf/src/main/java/org/apache/servicecomb/demo/perf/RedisClientUtils.java
@@ -40,7 +40,7 @@ public static void init(Vertx vertx) throws InterruptedException {
.setHost(PerfConfiguration.redisHost)
.setPort(PerfConfiguration.redisPort)
.setAuth(PerfConfiguration.redisPassword);
- ClientPoolFactory<RedisClient> factory = () -> {
+ ClientPoolFactory<RedisClient> factory = (ctx) -> {
return RedisClient.create(vertx, redisOptions);
};
clientMgr = new ClientPoolManager<>(vertx, factory);
diff --git a/edge/edge-core/src/main/java/org/apache/servicecomb/edge/core/EdgeBootListener.java b/edge/edge-core/src/main/java/org/apache/servicecomb/edge/core/EdgeBootListener.java
index 1762980b8..55ee39261 100644
--- a/edge/edge-core/src/main/java/org/apache/servicecomb/edge/core/EdgeBootListener.java
+++ b/edge/edge-core/src/main/java/org/apache/servicecomb/edge/core/EdgeBootListener.java
@@ -20,6 +20,8 @@
import org.apache.commons.configuration.Configuration;
import org.apache.servicecomb.core.BootListener;
import org.apache.servicecomb.core.executor.ExecutorManager;
+import org.apache.servicecomb.transport.rest.client.TransportClientConfig;
+import org.apache.servicecomb.transport.rest.vertx.TransportConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
@@ -36,6 +38,9 @@ public void onBootEvent(BootEvent event) {
return;
}
+ TransportClientConfig.setRestTransportClientCls(EdgeRestTransportClient.class);
+ TransportConfig.setRestServerVerticle(EdgeRestServerVerticle.class);
+
String defaultExecutor = DynamicPropertyFactory.getInstance()
.getStringProperty(ExecutorManager.KEY_EXECUTORS_DEFAULT, null)
.get();
diff --git a/edge/edge-core/src/main/java/org/apache/servicecomb/edge/core/EdgeInvocation.java b/edge/edge-core/src/main/java/org/apache/servicecomb/edge/core/EdgeInvocation.java
index e8988876d..4991443a6 100644
--- a/edge/edge-core/src/main/java/org/apache/servicecomb/edge/core/EdgeInvocation.java
+++ b/edge/edge-core/src/main/java/org/apache/servicecomb/edge/core/EdgeInvocation.java
@@ -36,9 +36,12 @@
import org.apache.servicecomb.serviceregistry.consumer.MicroserviceVersionRule;
import org.apache.servicecomb.serviceregistry.definition.DefinitionConst;
+import io.vertx.core.Vertx;
import io.vertx.ext.web.RoutingContext;
public class EdgeInvocation extends AbstractRestInvocation {
+ public static final String EDGE_INVOCATION_CONTEXT = "edgeInvocationContext";
+
protected String microserviceName;
protected MicroserviceVersionRule microserviceVersionRule;
@@ -121,6 +124,8 @@ protected void createInvocation() {
this.invocation = InvocationFactory.forConsumer(referenceConfig,
restOperationMeta.getOperationMeta(),
null);
+ this.invocation.setSync(false);
+ this.invocation.getHandlerContext().put(EDGE_INVOCATION_CONTEXT, Vertx.currentContext());
this.invocation.setResponseExecutor(new ReactiveResponseExecutor());
}
}
diff --git a/edge/edge-core/src/main/java/org/apache/servicecomb/edge/core/EdgeRestServerVerticle.java b/edge/edge-core/src/main/java/org/apache/servicecomb/edge/core/EdgeRestServerVerticle.java
new file mode 100644
index 000000000..a7d7529aa
--- /dev/null
+++ b/edge/edge-core/src/main/java/org/apache/servicecomb/edge/core/EdgeRestServerVerticle.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.servicecomb.edge.core;
+
+import org.apache.servicecomb.transport.rest.client.RestTransportClient;
+import org.apache.servicecomb.transport.rest.vertx.RestServerVerticle;
+
+public class EdgeRestServerVerticle extends RestServerVerticle {
+ @Override
+ public void start() throws Exception {
+ super.start();
+
+ RestTransportClient restClient = (RestTransportClient) config().getValue(RestTransportClient.class.getName());
+ restClient.getClientMgr().findClientPool(false, context);
+ }
+}
diff --git a/edge/edge-core/src/main/java/org/apache/servicecomb/edge/core/EdgeRestTransportClient.java b/edge/edge-core/src/main/java/org/apache/servicecomb/edge/core/EdgeRestTransportClient.java
new file mode 100644
index 000000000..bfac38676
--- /dev/null
+++ b/edge/edge-core/src/main/java/org/apache/servicecomb/edge/core/EdgeRestTransportClient.java
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.servicecomb.edge.core;
+
+import org.apache.servicecomb.core.Invocation;
+import org.apache.servicecomb.foundation.vertx.client.http.HttpClientWithContext;
+import org.apache.servicecomb.transport.rest.client.RestTransportClient;
+
+import io.vertx.core.Context;
+
+public class EdgeRestTransportClient extends RestTransportClient {
+ @Override
+ protected HttpClientWithContext findHttpClientPool(Invocation invocation) {
+ Context invocationContext = (Context) invocation.getHandlerContext().get(EdgeInvocation.EDGE_INVOCATION_CONTEXT);
+ return clientMgr.findClientPool(invocation.isSync(), invocationContext);
+ }
+}
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 5cbadbf8b..2f821a840 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
@@ -227,5 +227,7 @@ public void createInvocation(@Mocked MicroserviceVersionMeta microserviceVersion
edgeInvocation.createInvocation();
Invocation invocation = Deencapsulation.getField(edgeInvocation, "invocation");
Assert.assertThat(invocation.getResponseExecutor(), Matchers.instanceOf(ReactiveResponseExecutor.class));
+ Assert.assertFalse(invocation.isSync());
+ Assert.assertSame(context, invocation.getHandlerContext().get(EdgeInvocation.EDGE_INVOCATION_CONTEXT));
}
}
diff --git a/foundations/foundation-test-scaffolding/src/main/java/org/apache/servicecomb/foundation/test/scaffolding/log/LogCollector.java b/foundations/foundation-test-scaffolding/src/main/java/org/apache/servicecomb/foundation/test/scaffolding/log/LogCollector.java
index 33e580809..53359e35c 100644
--- a/foundations/foundation-test-scaffolding/src/main/java/org/apache/servicecomb/foundation/test/scaffolding/log/LogCollector.java
+++ b/foundations/foundation-test-scaffolding/src/main/java/org/apache/servicecomb/foundation/test/scaffolding/log/LogCollector.java
@@ -21,6 +21,7 @@
import org.apache.log4j.Appender;
import org.apache.log4j.AppenderSkeleton;
+import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.log4j.spi.LoggingEvent;
@@ -48,6 +49,11 @@ public LogCollector() {
Logger.getRootLogger().addAppender(appender);
}
+ public LogCollector setLogLevel(String logName, Level level) {
+ Logger.getLogger(logName).setLevel(level);
+ return this;
+ }
+
public List<LoggingEvent> getEvents() {
return events;
}
diff --git a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/VertxUtils.java b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/VertxUtils.java
index 8a31563de..4aba612f2 100644
--- a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/VertxUtils.java
+++ b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/VertxUtils.java
@@ -37,6 +37,7 @@
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Context;
import io.vertx.core.DeploymentOptions;
+import io.vertx.core.Verticle;
import io.vertx.core.Vertx;
import io.vertx.core.VertxOptions;
import io.vertx.core.buffer.Buffer;
@@ -88,7 +89,7 @@ private VertxUtils() {
}
// deploy Verticle and wait for its success. do not call this method in event-loop thread
- public static <VERTICLE extends AbstractVerticle> boolean blockDeploy(Vertx vertx,
+ public static <VERTICLE extends Verticle> boolean blockDeploy(Vertx vertx,
Class<VERTICLE> cls,
DeploymentOptions options) throws InterruptedException {
Holder<Boolean> result = new Holder<>();
diff --git a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/client/ClientPoolFactory.java b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/client/ClientPoolFactory.java
index 0445247ca..a8f38bd15 100644
--- a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/client/ClientPoolFactory.java
+++ b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/client/ClientPoolFactory.java
@@ -17,6 +17,8 @@
package org.apache.servicecomb.foundation.vertx.client;
+import io.vertx.core.Context;
+
public interface ClientPoolFactory<CLIENT_POOL> {
- CLIENT_POOL createClientPool();
+ CLIENT_POOL createClientPool(Context context);
}
diff --git a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/client/ClientPoolManager.java b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/client/ClientPoolManager.java
index a79b86d7d..b867593eb 100644
--- a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/client/ClientPoolManager.java
+++ b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/client/ClientPoolManager.java
@@ -56,28 +56,36 @@ public ClientPoolManager(Vertx vertx, ClientPoolFactory<CLIENT_POOL> factory) {
this.factory = factory;
}
- public CLIENT_POOL createClientPool() {
- CLIENT_POOL pool = factory.createClientPool();
- addPool(pool);
+ public CLIENT_POOL createClientPool(Context context) {
+ CLIENT_POOL pool = factory.createClientPool(context);
+ addPool(context, pool);
return pool;
}
- protected void addPool(CLIENT_POOL pool) {
- Vertx.currentContext().put(id, pool);
+ protected void addPool(Context context, CLIENT_POOL pool) {
+ context.put(id, pool);
pools.add(pool);
}
public CLIENT_POOL findClientPool(boolean sync) {
+ return findClientPool(sync, null);
+ }
+
+ public CLIENT_POOL findClientPool(boolean sync, Context targetContext) {
if (sync) {
return findThreadBindClientPool();
}
// reactive mode
- return findByContext();
+ return findByContext(targetContext);
}
protected CLIENT_POOL findByContext() {
- Context currentContext = Vertx.currentContext();
+ return findByContext(null);
+ }
+
+ protected CLIENT_POOL findByContext(Context targetContext) {
+ Context currentContext = targetContext != null ? targetContext : Vertx.currentContext();
if (currentContext != null
&& currentContext.owner() == vertx
&& currentContext.isEventLoopContext()) {
@@ -89,7 +97,7 @@ protected CLIENT_POOL findByContext() {
// this will make "client.thread-count" bigger than which in microservice.yaml
// maybe it's better to remove "client.thread-count", just use "rest/highway.thread-count"
- return createClientPool();
+ return createClientPool(currentContext);
}
// not in correct context:
diff --git a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/client/ClientVerticle.java b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/client/ClientVerticle.java
index 19e972bea..69044aec9 100644
--- a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/client/ClientVerticle.java
+++ b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/client/ClientVerticle.java
@@ -24,6 +24,7 @@
public class ClientVerticle<CLIENT_POOL> extends AbstractVerticle {
private static final Logger LOGGER = LoggerFactory.getLogger(ClientVerticle.class);
+
public static final String CLIENT_MGR = "clientMgr";
@SuppressWarnings("unchecked")
@@ -31,7 +32,7 @@
public void start() throws Exception {
try {
ClientPoolManager<CLIENT_POOL> clientMgr = (ClientPoolManager<CLIENT_POOL>) config().getValue(CLIENT_MGR);
- clientMgr.createClientPool();
+ clientMgr.createClientPool(context);
} catch (Throwable e) {
// vert.x got some states that not print error and execute call back in VertexUtils.blockDeploy, we add a log our self.
LOGGER.error("", e);
diff --git a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/client/http/HttpClientPoolFactory.java b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/client/http/HttpClientPoolFactory.java
index 945e8726f..a6b1f60b3 100644
--- a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/client/http/HttpClientPoolFactory.java
+++ b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/client/http/HttpClientPoolFactory.java
@@ -20,7 +20,6 @@
import org.apache.servicecomb.foundation.vertx.client.ClientPoolFactory;
import io.vertx.core.Context;
-import io.vertx.core.Vertx;
import io.vertx.core.http.HttpClient;
import io.vertx.core.http.HttpClientOptions;
@@ -33,8 +32,7 @@ public HttpClientPoolFactory(HttpClientOptions httpClientOptions) {
}
@Override
- public HttpClientWithContext createClientPool() {
- Context context = Vertx.currentContext();
+ public HttpClientWithContext createClientPool(Context context) {
HttpClient httpClient = context.owner().createHttpClient(httpClientOptions);
return new HttpClientWithContext(httpClient, context);
diff --git a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/client/tcp/AbstractTcpClientPoolFactory.java b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/client/tcp/AbstractTcpClientPoolFactory.java
index e87c283e0..3d1b298cd 100644
--- a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/client/tcp/AbstractTcpClientPoolFactory.java
+++ b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/client/tcp/AbstractTcpClientPoolFactory.java
@@ -32,8 +32,7 @@ public AbstractTcpClientPoolFactory(TcpClientConfig normalClientConfig, TcpClien
}
@Override
- public CLIENT_POOL createClientPool() {
- Context context = Vertx.currentContext();
+ public CLIENT_POOL createClientPool(Context context) {
Vertx vertx = context.owner();
NetClientWrapper netClientWrapper = new NetClientWrapper(vertx, normalClientConfig, sslClientConfig);
diff --git a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/ReadStreamPart.java b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/ReadStreamPart.java
new file mode 100644
index 000000000..401460992
--- /dev/null
+++ b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/ReadStreamPart.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.servicecomb.foundation.vertx.http;
+
+import java.util.concurrent.CompletableFuture;
+
+import javax.servlet.http.Part;
+
+import org.apache.servicecomb.foundation.common.part.AbstractPart;
+
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.streams.Pump;
+import io.vertx.core.streams.ReadStream;
+import io.vertx.core.streams.WriteStream;
+
+/**
+ * this is not a really part type, all method extend from AbstractPart is undefined except:<br>
+ * 1.getContentType<br>
+ * 2.getSubmittedFileName<br>
+ * extend from AbstractPart just because want to make it be Part type,
+ * so that can be sent by
+ * {@link org.apache.servicecomb.foundation.vertx.http.VertxServerResponseToHttpServletResponse#sendPart(Part) VertxServerResponseToHttpServletResponse.sendPart}
+ */
+public class ReadStreamPart extends AbstractPart {
+ private ReadStream<Buffer> readStream;
+
+ public ReadStreamPart(ReadStream<Buffer> readStream) {
+ this.readStream = readStream;
+
+ readStream.pause();
+ }
+
+ public CompletableFuture<Void> saveToWriteStream(WriteStream<Buffer> writeStream) {
+ CompletableFuture<Void> future = new CompletableFuture<>();
+
+ writeStream.exceptionHandler(future::completeExceptionally);
+ readStream.exceptionHandler(future::completeExceptionally);
+ readStream.endHandler(future::complete);
+
+ // if readStream(HttpClientResponse) and writeStream(HttpServerResponse)
+ // belongs to difference eventloop
+ // maybe will cause deadlock
+ // if happened, vertx will print deadlock stacks
+ Pump.pump(readStream, writeStream).start();
+ readStream.resume();
+
+ return future;
+ }
+}
diff --git a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/VertxServerResponseToHttpServletResponse.java b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/VertxServerResponseToHttpServletResponse.java
index d467058f9..8f55886a6 100644
--- a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/VertxServerResponseToHttpServletResponse.java
+++ b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/VertxServerResponseToHttpServletResponse.java
@@ -132,23 +132,26 @@ public void internalFlushBuffer() {
@Override
public CompletableFuture<Void> sendPart(Part part) {
- CompletableFuture<Void> future = new CompletableFuture<Void>();
-
prepareSendPartHeader(part);
+ if (ReadStreamPart.class.isInstance(part)) {
+ return ((ReadStreamPart) part).saveToWriteStream(this.serverResponse);
+ }
+
+ CompletableFuture<Void> future = new CompletableFuture<Void>();
try {
InputStream is = part.getInputStream();
context.runOnContext(v -> {
- InputStreamToReadStream aa = new InputStreamToReadStream(context.owner(), is);
- aa.exceptionHandler(t -> {
+ InputStreamToReadStream inputStreamToReadStream = new InputStreamToReadStream(context.owner(), is);
+ inputStreamToReadStream.exceptionHandler(t -> {
clearPartResource(part, is);
future.completeExceptionally(t);
});
- aa.endHandler(V -> {
+ inputStreamToReadStream.endHandler(V -> {
clearPartResource(part, is);
future.complete(null);
});
- Pump.pump(aa, serverResponse).start();
+ Pump.pump(inputStreamToReadStream, serverResponse).start();
});
} catch (IOException e) {
future.completeExceptionally(e);
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 884aecc59..4548e5be3 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
@@ -86,14 +86,12 @@ boolean isEventLoopContext() {
public void createClientPool(@Mocked HttpClientWithContext pool) {
new Expectations(VertxImpl.class) {
{
- factory.createClientPool();
+ factory.createClientPool(context);
result = pool;
- VertxImpl.context();
- result = context;
}
};
- Assert.assertSame(pool, poolMgr.createClientPool());
+ Assert.assertSame(pool, poolMgr.createClientPool(context));
Assert.assertSame(pool, context.get(id));
Assert.assertThat(pools, Matchers.contains(pool));
}
@@ -104,7 +102,7 @@ public void findClientPool_sync(@Mocked HttpClientWithContext pool1, @Mocked Htt
{
poolMgr.findThreadBindClientPool();
result = pool1;
- poolMgr.findByContext();
+ poolMgr.findByContext(null);
result = pool2;
}
};
@@ -148,7 +146,7 @@ public void findByContext_reactive() {
new Expectations(VertxImpl.class) {
{
- factory.createClientPool();
+ factory.createClientPool(context);
result = new HttpClientWithContext(null, null);
VertxImpl.context();
result = context;
diff --git a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/client/TestClientVerticle.java b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/client/TestClientVerticle.java
index 0c018fef6..6f7637239 100644
--- a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/client/TestClientVerticle.java
+++ b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/client/TestClientVerticle.java
@@ -38,7 +38,7 @@ public void start(@Mocked Context context) throws Exception {
AtomicInteger count = new AtomicInteger();
ClientPoolManager<HttpClientWithContext> clientMgr = new MockUp<ClientPoolManager<HttpClientWithContext>>() {
@Mock
- HttpClientWithContext createClientPool() {
+ HttpClientWithContext createClientPool(Context context) {
count.incrementAndGet();
return null;
}
diff --git a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/client/http/TestHttpClientPoolFactory.java b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/client/http/TestHttpClientPoolFactory.java
index 7f0f7ebee..08b9a32b1 100644
--- a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/client/http/TestHttpClientPoolFactory.java
+++ b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/client/http/TestHttpClientPoolFactory.java
@@ -36,15 +36,13 @@
public void createClientPool(@Mocked Vertx vertx, @Mocked Context context, @Mocked HttpClient httpClient) {
new Expectations(VertxImpl.class) {
{
- VertxImpl.context();
- result = context;
context.owner();
result = vertx;
vertx.createHttpClient(httpClientOptions);
result = httpClient;
}
};
- HttpClientWithContext pool = factory.createClientPool();
+ HttpClientWithContext pool = factory.createClientPool(context);
Assert.assertSame(context, pool.context());
Assert.assertSame(httpClient, pool.getHttpClient());
diff --git a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/client/tcp/TestAbstractTcpClientPoolFactory.java b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/client/tcp/TestAbstractTcpClientPoolFactory.java
index 5c868ff9b..a95aa3f9d 100644
--- a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/client/tcp/TestAbstractTcpClientPoolFactory.java
+++ b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/client/tcp/TestAbstractTcpClientPoolFactory.java
@@ -36,13 +36,11 @@
public void createClientPool(@Mocked Vertx vertx, @Mocked Context context) {
new Expectations(VertxImpl.class) {
{
- VertxImpl.context();
- result = context;
context.owner();
result = vertx;
}
};
- TcpClientConnectionPool pool = factory.createClientPool();
+ TcpClientConnectionPool pool = factory.createClientPool(context);
Assert.assertSame(normalClientConfig, pool.netClientWrapper.getClientConfig(false));
Assert.assertSame(sslClientConfig, pool.netClientWrapper.getClientConfig(true));
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 9a7877ac0..d50ecb84c 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
@@ -100,6 +100,12 @@ public void send_inWorkingStatus(@Mocked AbstractTcpClientPackage tcpClientPacka
result = byteBuf;
}
};
+
+ new MockUp<Context>(context) {
+ @Mock
+ void runOnContext(Handler<Void> action) {
+ }
+ };
tcpClientConnection.send(tcpClientPackage, ar -> {
});
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
new file mode 100644
index 000000000..cffd96498
--- /dev/null
+++ b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/http/TestReadStreamPart.java
@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.servicecomb.foundation.vertx.http;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.concurrent.ExecutionException;
+
+import org.apache.servicecomb.foundation.vertx.stream.InputStreamToReadStream;
+import org.hamcrest.Matchers;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+import io.vertx.core.AsyncResult;
+import io.vertx.core.Future;
+import io.vertx.core.Handler;
+import io.vertx.core.Vertx;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.streams.WriteStream;
+import mockit.Mock;
+import mockit.MockUp;
+import mockit.Mocked;
+
+public class TestReadStreamPart {
+ @Mocked
+ Vertx vertx;
+
+ String src = "src";
+
+ InputStreamToReadStream readStream;
+
+ ReadStreamPart part;
+
+ InputStream inputStream = new ByteArrayInputStream(src.getBytes());
+
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
+
+
+ @Before
+ public void setup() {
+ readStream = new InputStreamToReadStream(vertx, inputStream);
+ part = new ReadStreamPart(readStream);
+
+ new MockUp<Vertx>(vertx) {
+ @Mock
+ <T> void executeBlocking(Handler<Future<T>> blockingCodeHandler, boolean ordered,
+ Handler<AsyncResult<T>> resultHandler) {
+ Future<T> future = Future.future();
+ blockingCodeHandler.handle(future);
+ future.setHandler(resultHandler);
+ }
+ };
+ }
+
+ @Test
+ public void saveToWriteStream() throws InterruptedException, ExecutionException {
+ Buffer buf = Buffer.buffer();
+ WriteStream<Buffer> writeStream = new MockUp<WriteStream<Buffer>>() {
+ @Mock
+ WriteStream<Buffer> write(Buffer data) {
+ buf.appendBuffer(data);
+ return null;
+ }
+ }.getMockInstance();
+
+ part.saveToWriteStream(writeStream).get();
+
+ Assert.assertEquals(src, buf.toString());
+ }
+
+ @Test
+ public void saveToWriteStream_writeException() throws InterruptedException, ExecutionException {
+ Error error = new Error();
+ WriteStream<Buffer> writeStream = new MockUp<WriteStream<Buffer>>() {
+ Handler<Throwable> exceptionHandler;
+
+ @Mock
+ WriteStream<Buffer> exceptionHandler(Handler<Throwable> handler) {
+ this.exceptionHandler = handler;
+ return null;
+ }
+
+ @Mock
+ WriteStream<Buffer> write(Buffer data) {
+ exceptionHandler.handle(error);
+ return null;
+ }
+ }.getMockInstance();
+
+ expectedException.expect(ExecutionException.class);
+ expectedException.expectCause(Matchers.sameInstance(error));
+
+ part.saveToWriteStream(writeStream).get();
+ }
+
+ @Test
+ public void saveToWrite_readException(@Mocked WriteStream<Buffer> writeStream)
+ throws InterruptedException, ExecutionException {
+ Error error = new Error();
+ new MockUp<InputStream>(inputStream) {
+ @Mock
+ int read(byte b[]) throws IOException {
+ throw error;
+ }
+ };
+
+ expectedException.expect(ExecutionException.class);
+ expectedException.expectCause(Matchers.sameInstance(error));
+
+ part.saveToWriteStream(writeStream).get();
+ }
+}
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 fd6202cee..f66dd1b7e 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
@@ -47,6 +47,7 @@
import io.vertx.core.buffer.Buffer;
import io.vertx.core.http.HttpServerResponse;
import io.vertx.core.impl.VertxImpl;
+import io.vertx.core.streams.WriteStream;
import mockit.Deencapsulation;
import mockit.Expectations;
import mockit.Mock;
@@ -366,6 +367,21 @@ public void sendPart_inputStreamBreak(@Mocked Part part, @Mocked InputStream inp
future.get();
}
+ @SuppressWarnings("unchecked")
+ @Test
+ public void sendPart_ReadStreamPart(@Mocked ReadStreamPart part)
+ throws IOException, InterruptedException, ExecutionException {
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ new Expectations() {
+ {
+ part.saveToWriteStream((WriteStream<Buffer>) any);
+ result = future;
+ }
+ };
+
+ Assert.assertSame(future, response.sendPart(part));
+ }
+
@Test
public void sendPart_succ(@Mocked Part part, @Mocked InputStream inputStream)
throws IOException, InterruptedException, ExecutionException {
diff --git a/transports/transport-rest/transport-rest-client/pom.xml b/transports/transport-rest/transport-rest-client/pom.xml
index d7e2996fa..d7f03638e 100644
--- a/transports/transport-rest/transport-rest-client/pom.xml
+++ b/transports/transport-rest/transport-rest-client/pom.xml
@@ -16,36 +16,40 @@
~ limitations under the License.
-->
-<project
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
- xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
- <modelVersion>4.0.0</modelVersion>
- <parent>
- <groupId>org.apache.servicecomb</groupId>
- <artifactId>transport-rest</artifactId>
- <version>1.0.0-m2-SNAPSHOT</version>
- </parent>
- <artifactId>transport-rest-client</artifactId>
- <name>Java Chassis::Transports::Rest::Client</name>
+<project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
+ xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.servicecomb</groupId>
+ <artifactId>transport-rest</artifactId>
+ <version>1.0.0-m2-SNAPSHOT</version>
+ </parent>
+ <artifactId>transport-rest-client</artifactId>
+ <name>Java Chassis::Transports::Rest::Client</name>
- <dependencies>
- <dependency>
- <groupId>org.apache.servicecomb</groupId>
- <artifactId>foundation-vertx</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.servicecomb</groupId>
- <artifactId>common-rest</artifactId>
- </dependency>
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-log4j12</artifactId>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>log4j</groupId>
- <artifactId>log4j</artifactId>
- <scope>test</scope>
- </dependency>
- </dependencies>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.servicecomb</groupId>
+ <artifactId>foundation-vertx</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.servicecomb</groupId>
+ <artifactId>common-rest</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.servicecomb</groupId>
+ <artifactId>foundation-test-scaffolding</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
</project>
diff --git a/transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/RestTransportClient.java b/transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/RestTransportClient.java
index 1fed06523..8c4381417 100644
--- a/transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/RestTransportClient.java
+++ b/transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/RestTransportClient.java
@@ -17,7 +17,11 @@
package org.apache.servicecomb.transport.rest.client;
+import java.util.List;
+
+import org.apache.servicecomb.common.rest.filter.HttpClientFilter;
import org.apache.servicecomb.core.Invocation;
+import org.apache.servicecomb.foundation.common.utils.SPIServiceUtils;
import org.apache.servicecomb.foundation.vertx.VertxTLSBuilder;
import org.apache.servicecomb.foundation.vertx.VertxUtils;
import org.apache.servicecomb.foundation.vertx.client.ClientPoolManager;
@@ -25,7 +29,7 @@
import org.apache.servicecomb.foundation.vertx.client.http.HttpClientPoolFactory;
import org.apache.servicecomb.foundation.vertx.client.http.HttpClientWithContext;
import org.apache.servicecomb.swagger.invocation.AsyncResponse;
-import org.apache.servicecomb.transport.rest.client.http.VertxHttpMethod;
+import org.apache.servicecomb.transport.rest.client.http.RestClientInvocation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -33,15 +37,22 @@
import io.vertx.core.Vertx;
import io.vertx.core.http.HttpClientOptions;
-public final class RestTransportClient {
+public class RestTransportClient {
private static final Logger LOGGER = LoggerFactory.getLogger(RestTransportClient.class);
private static final String SSL_KEY = "rest.consumer";
- private ClientPoolManager<HttpClientWithContext> clientMgr;
+ protected ClientPoolManager<HttpClientWithContext> clientMgr;
+
+ private List<HttpClientFilter> httpClientFilters;
+ public ClientPoolManager<HttpClientWithContext> getClientMgr() {
+ return clientMgr;
+ }
public void init(Vertx vertx) throws Exception {
+ httpClientFilters = SPIServiceUtils.getSortedService(HttpClientFilter.class);
+
HttpClientOptions httpClientOptions = createHttpClientOptions();
clientMgr = new ClientPoolManager<>(vertx, new HttpClientPoolFactory(httpClientOptions));
@@ -61,13 +72,18 @@ private static HttpClientOptions createHttpClientOptions() {
return httpClientOptions;
}
- public void send(Invocation invocation, AsyncResponse asyncResp) throws Exception {
- HttpClientWithContext httpClientWithContext = clientMgr.findClientPool(invocation.isSync());
+ public void send(Invocation invocation, AsyncResponse asyncResp) {
+ HttpClientWithContext httpClientWithContext = findHttpClientPool(invocation);
+ RestClientInvocation restClientInvocation = new RestClientInvocation(httpClientWithContext, httpClientFilters);
try {
- VertxHttpMethod.INSTANCE.doMethod(httpClientWithContext, invocation, asyncResp);
- } catch (Exception e) {
+ restClientInvocation.invoke(invocation, asyncResp);
+ } catch (Throwable e) {
asyncResp.fail(invocation.getInvocationType(), e);
LOGGER.error("vertx rest transport send error.", e);
}
}
+
+ protected HttpClientWithContext findHttpClientPool(Invocation invocation) {
+ return clientMgr.findClientPool(invocation.isSync());
+ }
}
diff --git a/transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/RestTransportClientManager.java b/transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/RestTransportClientManager.java
index f969ba533..3dfd785aa 100644
--- a/transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/RestTransportClientManager.java
+++ b/transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/RestTransportClientManager.java
@@ -27,10 +27,11 @@
// same instance in AbstractTranport. need refactor in future.
private final Vertx transportVertx = VertxUtils.getOrCreateVertxByName("transport", null);
- private RestTransportClient restClient = new RestTransportClient();
+ private RestTransportClient restClient;
private RestTransportClientManager() {
try {
+ restClient = TransportClientConfig.getRestTransportClientCls().newInstance();
restClient.init(transportVertx);
} catch (Exception e) {
throw new IllegalStateException("Failed to init RestTransportClient.", e);
diff --git a/transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/TransportClientConfig.java b/transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/TransportClientConfig.java
index 980326710..15d122682 100644
--- a/transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/TransportClientConfig.java
+++ b/transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/TransportClientConfig.java
@@ -20,9 +20,19 @@
import com.netflix.config.DynamicPropertyFactory;
public final class TransportClientConfig {
+ private static Class<? extends RestTransportClient> restTransportClientCls = RestTransportClient.class;
+
private TransportClientConfig() {
}
+ public static Class<? extends RestTransportClient> getRestTransportClientCls() {
+ return restTransportClientCls;
+ }
+
+ public static void setRestTransportClientCls(Class<? extends RestTransportClient> restTransportClientCls) {
+ TransportClientConfig.restTransportClientCls = restTransportClientCls;
+ }
+
public static int getThreadCount() {
return DynamicPropertyFactory.getInstance().getIntProperty("cse.rest.client.thread-count", 1).get();
}
@@ -40,8 +50,10 @@ public static int getConnectionIdleTimeoutInSeconds() {
public static boolean getConnectionKeepAlive() {
return DynamicPropertyFactory.getInstance().getBooleanProperty("cse.rest.client.connection.keepAlive", true).get();
}
-
+
public static boolean getConnectionCompression() {
- return DynamicPropertyFactory.getInstance().getBooleanProperty("cse.rest.client.connection.compression", false).get();
+ return DynamicPropertyFactory.getInstance()
+ .getBooleanProperty("cse.rest.client.connection.compression", false)
+ .get();
}
}
diff --git a/transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/http/DefaultHttpClientFilter.java b/transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/http/DefaultHttpClientFilter.java
index 8d8555d1d..097c39cad 100644
--- a/transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/http/DefaultHttpClientFilter.java
+++ b/transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/http/DefaultHttpClientFilter.java
@@ -61,8 +61,12 @@ protected ProduceProcessor findProduceProcessor(RestOperationMeta restOperation,
return restOperation.findProduceProcessor(contentTypeForFind);
}
- @Override
- public Response afterReceiveResponse(Invocation invocation, HttpServletResponseEx responseEx) {
+ protected Object extractResult(Invocation invocation, HttpServletResponseEx responseEx) {
+ Object result = invocation.getHandlerContext().get(RestConst.READ_STREAM_PART);
+ if (result != null) {
+ return result;
+ }
+
OperationMeta operationMeta = invocation.getOperationMeta();
ResponseMeta responseMeta = operationMeta.findResponseMeta(responseEx.getStatus());
RestOperationMeta swaggerRestOperation = operationMeta.getExtData(RestConst.SWAGGER_REST_OPERATION);
@@ -75,16 +79,19 @@ public Response afterReceiveResponse(Invocation invocation, HttpServletResponseE
responseEx.getStatus(),
responseEx.getStatusType().getReasonPhrase(),
responseEx.getHeader(HttpHeaders.CONTENT_TYPE));
- Exception exception = ExceptionFactory.createConsumerException(new CommonExceptionData(msg));
- return Response.consumerFailResp(exception);
+ return ExceptionFactory.createConsumerException(new CommonExceptionData(msg));
}
- Object result = null;
try {
- result = produceProcessor.decodeResponse(responseEx.getBodyBuffer(), responseMeta.getJavaType());
+ return produceProcessor.decodeResponse(responseEx.getBodyBuffer(), responseMeta.getJavaType());
} catch (Exception e) {
- return Response.consumerFailResp(e);
+ return ExceptionFactory.createConsumerException(e);
}
+ }
+
+ @Override
+ public Response afterReceiveResponse(Invocation invocation, HttpServletResponseEx responseEx) {
+ Object result = extractResult(invocation, responseEx);
Response response =
Response.create(responseEx.getStatusType(), result);
diff --git a/transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/http/VertxHttpMethod.java b/transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/http/RestClientInvocation.java
similarity index 59%
rename from transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/http/VertxHttpMethod.java
rename to transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/http/RestClientInvocation.java
index d4f2aa109..9a586a134 100644
--- a/transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/http/VertxHttpMethod.java
+++ b/transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/http/RestClientInvocation.java
@@ -19,6 +19,8 @@
import java.util.List;
+import javax.servlet.http.Part;
+
import org.apache.servicecomb.common.rest.RestConst;
import org.apache.servicecomb.common.rest.codec.param.RestClientRequestImpl;
import org.apache.servicecomb.common.rest.definition.RestOperationMeta;
@@ -26,13 +28,14 @@
import org.apache.servicecomb.core.Invocation;
import org.apache.servicecomb.core.definition.OperationMeta;
import org.apache.servicecomb.core.transport.AbstractTransport;
+import org.apache.servicecomb.foundation.common.http.HttpStatus;
import org.apache.servicecomb.foundation.common.net.IpPort;
import org.apache.servicecomb.foundation.common.net.URIEndpointObject;
import org.apache.servicecomb.foundation.common.utils.JsonUtils;
-import org.apache.servicecomb.foundation.common.utils.SPIServiceUtils;
import org.apache.servicecomb.foundation.vertx.client.http.HttpClientWithContext;
import org.apache.servicecomb.foundation.vertx.http.HttpServletRequestEx;
import org.apache.servicecomb.foundation.vertx.http.HttpServletResponseEx;
+import org.apache.servicecomb.foundation.vertx.http.ReadStreamPart;
import org.apache.servicecomb.foundation.vertx.http.VertxClientRequestToHttpServletRequest;
import org.apache.servicecomb.foundation.vertx.http.VertxClientResponseToHttpServletResponse;
import org.apache.servicecomb.serviceregistry.api.Const;
@@ -43,33 +46,42 @@
import org.springframework.util.StringUtils;
import io.vertx.core.buffer.Buffer;
-import io.vertx.core.http.HttpClient;
import io.vertx.core.http.HttpClientRequest;
import io.vertx.core.http.HttpClientResponse;
import io.vertx.core.http.HttpMethod;
import io.vertx.core.http.RequestOptions;
-public class VertxHttpMethod {
- private static final Logger LOGGER = LoggerFactory.getLogger(VertxHttpMethod.class);
+public class RestClientInvocation {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RestClientInvocation.class);
+
+ private HttpClientWithContext httpClientWithContext;
+
+ private Invocation invocation;
+
+ private AsyncResponse asyncResp;
+
+ private List<HttpClientFilter> httpClientFilters;
+
+ private HttpClientRequest clientRequest;
+
+ private HttpClientResponse clientResponse;
- public static final VertxHttpMethod INSTANCE = new VertxHttpMethod();
+ public RestClientInvocation(HttpClientWithContext httpClientWithContext, List<HttpClientFilter> httpClientFilters) {
+ this.httpClientWithContext = httpClientWithContext;
+ this.httpClientFilters = httpClientFilters;
+ }
- static List<HttpClientFilter> httpClientFilters = SPIServiceUtils.getSortedService(HttpClientFilter.class);
+ public void invoke(Invocation invocation, AsyncResponse asyncResp) throws Exception {
+ this.invocation = invocation;
+ this.asyncResp = asyncResp;
- public void doMethod(HttpClientWithContext httpClientWithContext, Invocation invocation,
- AsyncResponse asyncResp) throws Exception {
OperationMeta operationMeta = invocation.getOperationMeta();
RestOperationMeta swaggerRestOperation = operationMeta.getExtData(RestConst.SWAGGER_REST_OPERATION);
- String path = this.createRequestPath(invocation, swaggerRestOperation);
+ String path = this.createRequestPath(swaggerRestOperation);
IpPort ipPort = (IpPort) invocation.getEndpoint().getAddress();
- HttpClientRequest clientRequest =
- this.createRequest(httpClientWithContext.getHttpClient(),
- invocation,
- ipPort,
- path,
- asyncResp);
+ createRequest(ipPort, path);
clientRequest.putHeader(org.apache.servicecomb.core.Const.TARGET_MICROSERVICE, invocation.getMicroserviceName());
RestClientRequestImpl restClientRequest =
new RestClientRequestImpl(clientRequest, httpClientWithContext.context().owner(), asyncResp);
@@ -82,13 +94,28 @@ public void doMethod(HttpClientWithContext httpClientWithContext, Invocation inv
}
clientRequest.exceptionHandler(e -> {
- LOGGER.error(e.toString());
+ LOGGER.error("Failed to send request to {}.", ipPort.getSocketAddress(), e);
asyncResp.fail(invocation.getInvocationType(), e);
});
-
+ clientRequest.connectionHandler(connection -> {
+ LOGGER.info("http connection connected, local:{}, remote:{}.",
+ connection.localAddress(),
+ connection.remoteAddress());
+ connection.closeHandler(v -> {
+ LOGGER.info("http connection closed, local:{}, remote:{}.",
+ connection.localAddress(),
+ connection.remoteAddress());
+ });
+ connection.exceptionHandler(e -> {
+ LOGGER.info("http connection exception, local:{}, remote:{}.",
+ connection.localAddress(),
+ connection.remoteAddress(),
+ e);
+ });
+ });
// 从业务线程转移到网络线程中去发送
httpClientWithContext.runOnContext(httpClient -> {
- this.setCseContext(invocation, clientRequest);
+ this.setCseContext();
clientRequest.setTimeout(AbstractTransport.getRequestTimeoutProperty().get());
try {
restClientRequest.end();
@@ -99,15 +126,14 @@ public void doMethod(HttpClientWithContext httpClientWithContext, Invocation inv
});
}
- private HttpMethod getMethod(Invocation invocation) {
+ private HttpMethod getMethod() {
OperationMeta operationMeta = invocation.getOperationMeta();
RestOperationMeta swaggerRestOperation = operationMeta.getExtData(RestConst.SWAGGER_REST_OPERATION);
String method = swaggerRestOperation.getHttpMethod();
return HttpMethod.valueOf(method);
}
- HttpClientRequest createRequest(HttpClient client, Invocation invocation, IpPort ipPort, String path,
- AsyncResponse asyncResp) {
+ void createRequest(IpPort ipPort, String path) {
URIEndpointObject endpoint = (URIEndpointObject) invocation.getEndpoint().getAddress();
RequestOptions requestOptions = new RequestOptions();
requestOptions.setHost(ipPort.getHostOrIp())
@@ -115,51 +141,64 @@ HttpClientRequest createRequest(HttpClient client, Invocation invocation, IpPort
.setSsl(endpoint.isSslEnabled())
.setURI(path);
- HttpMethod method = getMethod(invocation);
+ HttpMethod method = getMethod();
LOGGER.debug("Sending request by rest, method={}, qualifiedName={}, path={}, endpoint={}.",
method,
invocation.getMicroserviceQualifiedName(),
path,
invocation.getEndpoint().getEndpoint());
- HttpClientRequest request = client.request(method, requestOptions, response -> {
- handleResponse(invocation, response, asyncResp);
- });
- return request;
+ clientRequest = httpClientWithContext.getHttpClient().request(method, requestOptions, this::handleResponse);
}
- void handleResponse(Invocation invocation, HttpClientResponse clientResponse,
- AsyncResponse asyncResp) {
+ protected void handleResponse(HttpClientResponse httpClientResponse) {
+ this.clientResponse = httpClientResponse;
+
+ if (HttpStatus.isSuccess(clientResponse.statusCode())
+ && Part.class.equals(invocation.getOperationMeta().getMethod().getReturnType())) {
+ ReadStreamPart part = new ReadStreamPart(httpClientResponse);
+ invocation.getHandlerContext().put(RestConst.READ_STREAM_PART, part);
+ processResponseBody(null);
+ return;
+ }
+
+ httpClientResponse.exceptionHandler(e -> {
+ LOGGER.error("Failed to receive response from {}.", httpClientResponse.netSocket().remoteAddress(), e);
+ asyncResp.fail(invocation.getInvocationType(), e);
+ });
+
clientResponse.bodyHandler(responseBuf -> {
- // 此时是在网络线程中，不应该就地处理，通过dispatcher转移线程
- invocation.getResponseExecutor().execute(() -> {
- try {
- HttpServletResponseEx responseEx =
- new VertxClientResponseToHttpServletResponse(clientResponse, responseBuf);
- for (HttpClientFilter filter : httpClientFilters) {
- Response response = filter.afterReceiveResponse(invocation, responseEx);
- if (response != null) {
- asyncResp.complete(response);
- return;
- }
+ processResponseBody(responseBuf);
+ });
+ }
+
+ protected void processResponseBody(Buffer responseBuf) {
+ invocation.getResponseExecutor().execute(() -> {
+ try {
+ HttpServletResponseEx responseEx =
+ new VertxClientResponseToHttpServletResponse(clientResponse, responseBuf);
+ for (HttpClientFilter filter : httpClientFilters) {
+ Response response = filter.afterReceiveResponse(invocation, responseEx);
+ if (response != null) {
+ asyncResp.complete(response);
+ return;
}
- } catch (Throwable e) {
- asyncResp.fail(invocation.getInvocationType(), e);
}
- });
+ } catch (Throwable e) {
+ asyncResp.fail(invocation.getInvocationType(), e);
+ }
});
}
- protected void setCseContext(Invocation invocation, HttpClientRequest request) {
+ protected void setCseContext() {
try {
String cseContext = JsonUtils.writeValueAsString(invocation.getContext());
- request.putHeader(org.apache.servicecomb.core.Const.CSE_CONTEXT, cseContext);
+ clientRequest.putHeader(org.apache.servicecomb.core.Const.CSE_CONTEXT, cseContext);
} catch (Exception e) {
- LOGGER.debug(e.toString());
+ LOGGER.debug("Failed to encode and set cseContext.", e);
}
}
- protected String createRequestPath(Invocation invocation,
- RestOperationMeta swaggerRestOperation) throws Exception {
+ protected String createRequestPath(RestOperationMeta swaggerRestOperation) throws Exception {
URIEndpointObject address = (URIEndpointObject) invocation.getEndpoint().getAddress();
String urlPrefix = address.getFirst(Const.URL_PREFIX);
diff --git a/transports/transport-rest/transport-rest-client/src/test/java/org/apache/servicecomb/transport/rest/client/http/TestDefaultHttpClientFilter.java b/transports/transport-rest/transport-rest-client/src/test/java/org/apache/servicecomb/transport/rest/client/http/TestDefaultHttpClientFilter.java
index 1d79e60c3..8e6168a24 100644
--- a/transports/transport-rest/transport-rest-client/src/test/java/org/apache/servicecomb/transport/rest/client/http/TestDefaultHttpClientFilter.java
+++ b/transports/transport-rest/transport-rest-client/src/test/java/org/apache/servicecomb/transport/rest/client/http/TestDefaultHttpClientFilter.java
@@ -18,6 +18,8 @@
package org.apache.servicecomb.transport.rest.client.http;
import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
import javax.ws.rs.core.HttpHeaders;
import javax.ws.rs.core.Response.Status;
@@ -28,6 +30,7 @@
import org.apache.servicecomb.core.Invocation;
import org.apache.servicecomb.core.definition.OperationMeta;
import org.apache.servicecomb.foundation.vertx.http.HttpServletResponseEx;
+import org.apache.servicecomb.foundation.vertx.http.ReadStreamPart;
import org.apache.servicecomb.swagger.invocation.Response;
import org.apache.servicecomb.swagger.invocation.exception.CommonExceptionData;
import org.apache.servicecomb.swagger.invocation.exception.InvocationException;
@@ -92,6 +95,20 @@ public void testFindProduceProcessorJsonWithCharset(@Mocked RestOperationMeta re
Assert.assertSame(produceProcessor, filter.findProduceProcessor(restOperation, responseEx));
}
+ @Test
+ public void extractResult_readStreamPart(@Mocked Invocation invocation, @Mocked ReadStreamPart part) {
+ Map<String, Object> handlerContext = new HashMap<>();
+ handlerContext.put(RestConst.READ_STREAM_PART, part);
+ new Expectations() {
+ {
+ invocation.getHandlerContext();
+ result = handlerContext;
+ }
+ };
+
+ Assert.assertSame(part, filter.extractResult(invocation, null));
+ }
+
@Test
public void testAfterReceiveResponseNullProduceProcessor(@Mocked Invocation invocation,
@Mocked HttpServletResponseEx responseEx,
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
new file mode 100644
index 000000000..76a6eabf6
--- /dev/null
+++ b/transports/transport-rest/transport-rest-client/src/test/java/org/apache/servicecomb/transport/rest/client/http/TestRestClientInvocation.java
@@ -0,0 +1,351 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.servicecomb.transport.rest.client.http;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import javax.servlet.http.Part;
+
+import org.apache.log4j.Level;
+import org.apache.servicecomb.common.rest.RestConst;
+import org.apache.servicecomb.common.rest.definition.RestOperationMeta;
+import org.apache.servicecomb.common.rest.definition.path.URLPathBuilder;
+import org.apache.servicecomb.common.rest.filter.HttpClientFilter;
+import org.apache.servicecomb.core.Endpoint;
+import org.apache.servicecomb.core.Invocation;
+import org.apache.servicecomb.core.definition.OperationMeta;
+import org.apache.servicecomb.core.executor.ReactiveExecutor;
+import org.apache.servicecomb.foundation.common.net.URIEndpointObject;
+import org.apache.servicecomb.foundation.common.utils.ReflectUtils;
+import org.apache.servicecomb.foundation.test.scaffolding.log.LogCollector;
+import org.apache.servicecomb.foundation.vertx.client.http.HttpClientWithContext;
+import org.apache.servicecomb.foundation.vertx.http.ReadStreamPart;
+import org.apache.servicecomb.serviceregistry.api.Const;
+import org.apache.servicecomb.swagger.invocation.AsyncResponse;
+import org.apache.servicecomb.swagger.invocation.Response;
+import org.apache.servicecomb.swagger.invocation.exception.InvocationException;
+import org.hamcrest.Matchers;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import io.vertx.core.Context;
+import io.vertx.core.Handler;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.HttpClient;
+import io.vertx.core.http.HttpClientRequest;
+import io.vertx.core.http.HttpClientResponse;
+import io.vertx.core.http.HttpMethod;
+import io.vertx.core.http.RequestOptions;
+import io.vertx.core.net.NetSocket;
+import io.vertx.core.net.SocketAddress;
+import mockit.Deencapsulation;
+import mockit.Mock;
+import mockit.MockUp;
+import mockit.Mocked;
+
+public class TestRestClientInvocation {
+ Handler<Throwable> exceptionHandler;
+
+ Handler<Buffer> bodyHandler;
+
+ Map<String, String> headers = new HashMap<>();
+
+ HttpClientRequest request = mock(HttpClientRequest.class);
+
+ HttpClient httpClient = mock(HttpClient.class);
+
+ Context context = mock(Context.class);
+
+ HttpClientWithContext httpClientWithContext = new HttpClientWithContext(httpClient, context);
+
+ Invocation invocation = mock(Invocation.class);
+
+ Response response;
+
+ AsyncResponse asyncResp = resp -> {
+ response = resp;
+ };
+
+ OperationMeta operationMeta = mock(OperationMeta.class);
+
+ Endpoint endpoint = mock(Endpoint.class);
+
+ RestOperationMeta swaggerRestOperation = mock(RestOperationMeta.class);
+
+ URLPathBuilder urlPathBuilder = mock(URLPathBuilder.class);
+
+ URIEndpointObject address = mock(URIEndpointObject.class);
+
+ List<HttpClientFilter> httpClientFilters = new ArrayList<>();
+
+ RestClientInvocation restClientInvocation = new RestClientInvocation(httpClientWithContext, httpClientFilters);
+
+ Map<String, Object> handlerContext = new HashMap<>();
+
+ @SuppressWarnings("unchecked")
+ @Before
+ public void setup() {
+ Deencapsulation.setField(restClientInvocation, "clientRequest", request);
+ Deencapsulation.setField(restClientInvocation, "invocation", invocation);
+ Deencapsulation.setField(restClientInvocation, "asyncResp", asyncResp);
+
+ when(invocation.getOperationMeta()).thenReturn(operationMeta);
+ when(swaggerRestOperation.getPathBuilder()).thenReturn(urlPathBuilder);
+ when(swaggerRestOperation.getHttpMethod()).thenReturn("GET");
+ when(operationMeta.getExtData(RestConst.SWAGGER_REST_OPERATION)).thenReturn(swaggerRestOperation);
+ when(invocation.getEndpoint()).thenReturn(endpoint);
+ when(endpoint.getAddress()).thenReturn(address);
+ when(invocation.getHandlerContext()).then(answer -> handlerContext);
+ when(httpClient.request((HttpMethod) Mockito.any(), (RequestOptions) Mockito.any(), Mockito.any()))
+ .thenReturn(request);
+ doAnswer(a -> {
+ exceptionHandler = (Handler<Throwable>) a.getArguments()[0];
+ return request;
+ }).when(request).exceptionHandler(any());
+ doAnswer(a -> {
+ headers.put(a.getArgumentAt(0, String.class), a.getArgumentAt(1, String.class));
+ return request;
+ }).when(request).putHeader((String) any(), (String) any());
+ doAnswer(a -> {
+ ((Handler<Void>) a.getArguments()[0]).handle(null);
+ return null;
+ }).when(context).runOnContext(any());
+ }
+
+ @Test
+ public void invoke(@Mocked Response resp) throws Exception {
+ doAnswer(a -> {
+ asyncResp.complete(resp);
+ return null;
+ }).when(request).end();
+ restClientInvocation.invoke(invocation, asyncResp);
+
+ Assert.assertSame(resp, response);
+ }
+
+ @Test
+ public void invoke_endThrow() throws Exception {
+ Mockito.doThrow(Error.class).when(request).end();
+ restClientInvocation.invoke(invocation, asyncResp);
+
+ Assert.assertThat(((InvocationException) response.getResult()).getCause(), Matchers.instanceOf(Error.class));
+ }
+
+ @Test
+ public void invoke_requestThrow() throws Exception {
+ Throwable t = new Error();
+ doAnswer(a -> {
+ exceptionHandler.handle(t);
+ return null;
+ }).when(request).end();
+ restClientInvocation.invoke(invocation, asyncResp);
+ restClientInvocation.invoke(invocation, asyncResp);
+
+ Assert.assertThat(((InvocationException) response.getResult()).getCause(), Matchers.sameInstance(t));
+ }
+
+ @Test
+ public void testSetCseContext() {
+ Map<String, String> contextMap = Collections.singletonMap("k", "v");
+ when(invocation.getContext()).thenReturn(contextMap);
+
+ restClientInvocation.setCseContext();
+
+ Assert.assertEquals("{x-cse-context={\"k\":\"v\"}}", headers.toString());
+ }
+
+ @Test
+ public void testSetCseContext_failed() {
+ LogCollector logCollector = new LogCollector();
+ logCollector.setLogLevel(RestClientInvocation.class.getName(), Level.DEBUG);
+ Deencapsulation.setField(restClientInvocation, "invocation", null);
+
+ restClientInvocation.setCseContext();
+
+ Assert.assertEquals("Failed to encode and set cseContext.", logCollector.getEvents().get(0).getMessage());
+ logCollector.teardown();
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void handleResponse() {
+ HttpClientResponse httpClientResponse = mock(HttpClientResponse.class);
+ doAnswer(a -> {
+ bodyHandler = (Handler<Buffer>) a.getArguments()[0];
+ return httpClientResponse;
+ }).when(httpClientResponse).bodyHandler(any());
+
+ Buffer buf = Buffer.buffer();
+ new MockUp<RestClientInvocation>(restClientInvocation) {
+ @Mock
+ void processResponseBody(Buffer responseBuf) {
+ asyncResp.success(buf);
+ }
+ };
+
+ restClientInvocation.handleResponse(httpClientResponse);
+ bodyHandler.handle(buf);
+
+ Assert.assertSame(buf, response.getResult());
+ }
+
+ public Part returnPart() {
+ return null;
+ }
+
+ @Test
+ public void handleResponse_readStreamPart() {
+ HttpClientResponse httpClientResponse = mock(HttpClientResponse.class);
+ when(httpClientResponse.statusCode()).thenReturn(200);
+ Method method = ReflectUtils.findMethod(this.getClass(), "returnPart");
+ when(operationMeta.getMethod()).thenReturn(method);
+ new MockUp<RestClientInvocation>(restClientInvocation) {
+ @Mock
+ void processResponseBody(Buffer responseBuf) {
+ }
+ };
+
+ restClientInvocation.handleResponse(httpClientResponse);
+
+ Assert.assertThat(handlerContext.get(RestConst.READ_STREAM_PART), Matchers.instanceOf(ReadStreamPart.class));
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void handleResponse_responseException() {
+ HttpClientResponse httpClientResponse = mock(HttpClientResponse.class);
+
+ NetSocket netSocket = mock(NetSocket.class);
+ when(httpClientResponse.netSocket()).thenReturn(netSocket);
+ when(netSocket.remoteAddress()).thenReturn(mock(SocketAddress.class));
+
+ doAnswer(a -> {
+ exceptionHandler = (Handler<Throwable>) a.getArguments()[0];
+ return httpClientResponse;
+ }).when(httpClientResponse).exceptionHandler(any());
+
+ restClientInvocation.handleResponse(httpClientResponse);
+ Error error = new Error();
+ exceptionHandler.handle(error);
+
+ Assert.assertThat(((InvocationException) response.getResult()).getCause(), Matchers.sameInstance(error));
+ }
+
+ @Test
+ public void processResponseBody() {
+ Response resp = Response.ok(null);
+
+ HttpClientResponse httpClientResponse = mock(HttpClientResponse.class);
+ Deencapsulation.setField(restClientInvocation, "clientResponse", httpClientResponse);
+
+ {
+ HttpClientFilter filter = mock(HttpClientFilter.class);
+ when(filter.afterReceiveResponse(any(), any())).thenReturn(null);
+ httpClientFilters.add(filter);
+ }
+ {
+ HttpClientFilter filter = mock(HttpClientFilter.class);
+ when(filter.afterReceiveResponse(any(), any())).thenReturn(resp);
+ httpClientFilters.add(filter);
+ }
+
+ when(invocation.getResponseExecutor()).thenReturn(new ReactiveExecutor());
+
+ restClientInvocation.processResponseBody(null);
+
+ Assert.assertSame(resp, response);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void processResponseBody_throw() {
+ HttpClientResponse httpClientResponse = mock(HttpClientResponse.class);
+ Deencapsulation.setField(restClientInvocation, "clientResponse", httpClientResponse);
+
+ {
+ HttpClientFilter filter = mock(HttpClientFilter.class);
+ when(filter.afterReceiveResponse(any(), any())).thenThrow(Error.class);
+ httpClientFilters.add(filter);
+ }
+
+ when(invocation.getResponseExecutor()).thenReturn(new ReactiveExecutor());
+
+ restClientInvocation.processResponseBody(null);
+
+ Assert.assertThat(((InvocationException) response.getResult()).getCause(), Matchers.instanceOf(Error.class));
+ }
+
+ @Test
+ public void createRequestPath_NoUrlPrefixNoPath() throws Exception {
+ when(address.getFirst(Const.URL_PREFIX)).thenReturn(null);
+
+ when(urlPathBuilder.createRequestPath(any())).thenReturn("/path");
+
+ String path = restClientInvocation.createRequestPath(swaggerRestOperation);
+ Assert.assertEquals("/path", path);
+ }
+
+ @Test
+ public void createRequestPath_noUrlPrefixHavePath() throws Exception {
+ handlerContext.put(RestConst.REST_CLIENT_REQUEST_PATH, "/client/path");
+ when(address.getFirst(Const.URL_PREFIX)).thenReturn(null);
+
+ String path = restClientInvocation.createRequestPath(swaggerRestOperation);
+ Assert.assertEquals("/client/path", path);
+ }
+
+ @Test
+ public void createRequestPath_haveUrlPrefixNoPath() throws Exception {
+ when(address.getFirst(Const.URL_PREFIX)).thenReturn("/prefix");
+
+ when(urlPathBuilder.createRequestPath(any())).thenReturn("/path");
+
+ String path = restClientInvocation.createRequestPath(swaggerRestOperation);
+ Assert.assertEquals("/prefix/path", path);
+ }
+
+ @Test
+ public void createRequestPath_haveUrlPrefixHavePath() throws Exception {
+ when(address.getFirst(Const.URL_PREFIX)).thenReturn("/prefix");
+ handlerContext.put(RestConst.REST_CLIENT_REQUEST_PATH, "/client/path");
+
+ String path = restClientInvocation.createRequestPath(swaggerRestOperation);
+ Assert.assertEquals("/prefix/client/path", path);
+ }
+
+ @Test
+ public void createRequestPath_haveUrlPrefixHavePathAndStartWith() throws Exception {
+ when(address.getFirst(Const.URL_PREFIX)).thenReturn("/prefix");
+ handlerContext.put(RestConst.REST_CLIENT_REQUEST_PATH, "/prefix/client/path");
+
+ String path = restClientInvocation.createRequestPath(swaggerRestOperation);
+ Assert.assertEquals("/prefix/client/path", path);
+ }
+}
diff --git a/transports/transport-rest/transport-rest-client/src/test/java/org/apache/servicecomb/transport/rest/client/http/TestVertxHttpMethod.java b/transports/transport-rest/transport-rest-client/src/test/java/org/apache/servicecomb/transport/rest/client/http/TestVertxHttpMethod.java
deleted file mode 100644
index 1c9dcfe9c..000000000
--- a/transports/transport-rest/transport-rest-client/src/test/java/org/apache/servicecomb/transport/rest/client/http/TestVertxHttpMethod.java
+++ /dev/null
@@ -1,302 +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 org.apache.servicecomb.transport.rest.client.http;
-
-import static org.junit.Assert.fail;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.servicecomb.common.rest.RestConst;
-import org.apache.servicecomb.common.rest.codec.produce.ProduceProcessor;
-import org.apache.servicecomb.common.rest.definition.RestOperationMeta;
-import org.apache.servicecomb.common.rest.definition.path.URLPathBuilder;
-import org.apache.servicecomb.core.Endpoint;
-import org.apache.servicecomb.core.Invocation;
-import org.apache.servicecomb.core.definition.OperationMeta;
-import org.apache.servicecomb.foundation.common.net.IpPort;
-import org.apache.servicecomb.foundation.common.net.URIEndpointObject;
-import org.apache.servicecomb.foundation.vertx.client.http.HttpClientWithContext;
-import org.apache.servicecomb.serviceregistry.api.Const;
-import org.apache.servicecomb.swagger.invocation.AsyncResponse;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-import org.mockito.Mockito;
-
-import io.vertx.core.Context;
-import io.vertx.core.Handler;
-import io.vertx.core.http.HttpClient;
-import io.vertx.core.http.HttpClientRequest;
-import io.vertx.core.http.HttpClientResponse;
-import io.vertx.core.http.HttpMethod;
-import mockit.Expectations;
-import mockit.Injectable;
-import mockit.Mock;
-import mockit.MockUp;
-import mockit.Mocked;
-
-public class TestVertxHttpMethod extends VertxHttpMethod {
-
- HttpClientRequest request;
-
- @Before
- public void setup() {
- request = mock(HttpClientRequest.class);
- }
-
- @Test
- public void testDoMethodNullPointerException(@Mocked HttpClient httpClient) throws Exception {
- Context context = new MockUp<Context>() {
- @Mock
- public void runOnContext(Handler<Void> action) {
- action.handle(null);
- }
- }.getMockInstance();
- HttpClientWithContext httpClientWithContext = new HttpClientWithContext(httpClient, context);
-
- Invocation invocation = mock(Invocation.class);
- AsyncResponse asyncResp = mock(AsyncResponse.class);
-
- try {
- this.doMethod(httpClientWithContext, invocation, asyncResp);
- fail("Expect to throw NullPointerException, but got none");
- } catch (NullPointerException e) {
- }
- }
-
- @Test
- public void testDoMethod(@Mocked HttpClient httpClient, @Injectable URIEndpointObject address) throws Exception {
- Context context = new MockUp<Context>() {
- @Mock
- public void runOnContext(Handler<Void> action) {
- action.handle(null);
- }
- }.getMockInstance();
- HttpClientWithContext httpClientWithContext = new HttpClientWithContext(httpClient, context);
-
- Invocation invocation = mock(Invocation.class);
- AsyncResponse asyncResp = mock(AsyncResponse.class);
- OperationMeta operationMeta = mock(OperationMeta.class);
- RestOperationMeta swaggerRestOperation = mock(RestOperationMeta.class);
-
- Endpoint endpoint = mock(Endpoint.class);
- when(invocation.getOperationMeta()).thenReturn(operationMeta);
- URLPathBuilder urlPathBuilder = mock(URLPathBuilder.class);
- when(swaggerRestOperation.getPathBuilder()).thenReturn(urlPathBuilder);
- operationMeta.getExtData(RestConst.SWAGGER_REST_OPERATION);
- when(operationMeta.getExtData(RestConst.SWAGGER_REST_OPERATION)).thenReturn(swaggerRestOperation);
- when(invocation.getEndpoint()).thenReturn(endpoint);
- when(endpoint.getAddress()).thenReturn(address);
-
- when(request.exceptionHandler(Mockito.any())).then(answer -> null);
- Map<String, Object> map = new HashMap<>();
- when(invocation.getHandlerContext()).then(answer -> map);;
-
- this.doMethod(httpClientWithContext, invocation, asyncResp);
- Assert.assertTrue(true);
- }
-
- @Test
- public void testCreateRequest() {
- HttpClient client = mock(HttpClient.class);
- Invocation invocation = mock(Invocation.class);
- OperationMeta operationMeta = mock(OperationMeta.class);
- Endpoint endpoint = mock(Endpoint.class);
- URIEndpointObject address = mock(URIEndpointObject.class);
- when(invocation.getEndpoint()).thenReturn(endpoint);
- when(endpoint.getAddress()).thenReturn(address);
- when(address.isSslEnabled()).thenReturn(false);
- when(invocation.getOperationMeta()).thenReturn(operationMeta);
- RestOperationMeta swaggerRestOperation = mock(RestOperationMeta.class);
- when(operationMeta.getExtData(RestConst.SWAGGER_REST_OPERATION)).thenReturn(swaggerRestOperation);
- IpPort ipPort = mock(IpPort.class);
- when(ipPort.getPort()).thenReturn(10);
- when(ipPort.getHostOrIp()).thenReturn("ever");
- AsyncResponse asyncResp = mock(AsyncResponse.class);
- List<HttpMethod> methods = new ArrayList<>(
- Arrays.asList(HttpMethod.GET, HttpMethod.PUT, HttpMethod.POST, HttpMethod.DELETE, HttpMethod.PATCH));
- for (HttpMethod method : methods) {
- when(swaggerRestOperation.getHttpMethod()).thenReturn(method.toString());
- HttpClientRequest obj =
- VertxHttpMethod.INSTANCE.createRequest(client, invocation, ipPort, "good", asyncResp);
- Assert.assertNull(obj);
- }
- }
-
- @Test
- public void testSetCseContext() {
- boolean status = false;
- try {
- Invocation invocation = mock(Invocation.class);
- HttpClientResponse httpResponse = mock(HttpClientResponse.class);
- OperationMeta operationMeta = mock(OperationMeta.class);
- RestOperationMeta swaggerRestOperation = mock(RestOperationMeta.class);
- HttpClientRequest request = mock(HttpClientRequest.class);
-
- Endpoint endpoint = mock(Endpoint.class);
- when(invocation.getOperationMeta()).thenReturn(operationMeta);
- URLPathBuilder urlPathBuilder = mock(URLPathBuilder.class);
- when(swaggerRestOperation.getPathBuilder()).thenReturn(urlPathBuilder);
- operationMeta.getExtData(RestConst.SWAGGER_REST_OPERATION);
- when(operationMeta.getExtData(RestConst.SWAGGER_REST_OPERATION)).thenReturn(swaggerRestOperation);
- when(invocation.getEndpoint()).thenReturn(endpoint);
- String contentType = httpResponse.getHeader("Content-Type");
- ProduceProcessor produceProcessor = mock(ProduceProcessor.class);
- when(swaggerRestOperation.findProduceProcessor(contentType)).thenReturn(produceProcessor);
- this.setCseContext(invocation, request);
- } catch (Exception ex) {
- status = true;
- }
- Assert.assertFalse(status);
- }
-
- @Test
- public void testHandleResponse() {
- boolean status = false;
- try {
- Invocation invocation = mock(Invocation.class);
- AsyncResponse asyncResp = mock(AsyncResponse.class);
- HttpClientResponse httpResponse = mock(HttpClientResponse.class);
- OperationMeta operationMeta = mock(OperationMeta.class);
- RestOperationMeta swaggerRestOperation = mock(RestOperationMeta.class);
-
- Endpoint endpoint = mock(Endpoint.class);
- when(invocation.getOperationMeta()).thenReturn(operationMeta);
- URLPathBuilder urlPathBuilder = mock(URLPathBuilder.class);
- when(swaggerRestOperation.getPathBuilder()).thenReturn(urlPathBuilder);
- operationMeta.getExtData(RestConst.SWAGGER_REST_OPERATION);
- when(operationMeta.getExtData(RestConst.SWAGGER_REST_OPERATION)).thenReturn(swaggerRestOperation);
- when(invocation.getEndpoint()).thenReturn(endpoint);
-
- String contentType = httpResponse.getHeader("Content-Type");
- ProduceProcessor produceProcessor = mock(ProduceProcessor.class);
- when(swaggerRestOperation.findProduceProcessor(contentType)).thenReturn(produceProcessor);
- this.handleResponse(invocation, httpResponse, asyncResp);
- } catch (Exception ex) {
- status = true;
- }
- Assert.assertFalse(status);
- }
-
- @Override
- protected HttpClientRequest createRequest(HttpClient client, Invocation invocation, IpPort ipPort, String path,
- AsyncResponse asyncResp) {
- return request;
- }
-
- @Test
- public void testCreateRequestPathNoUrlPrefixNoPath(@Injectable Invocation invocation,
- @Injectable RestOperationMeta swaggerRestOperation, @Injectable Endpoint endpoint,
- @Injectable URIEndpointObject address, @Injectable URLPathBuilder builder) throws Exception {
- new Expectations() {
- {
- endpoint.getAddress();
- result = address;
- builder.createRequestPath((Object[]) any);
- result = "/path";
- }
- };
- String path = this.createRequestPath(invocation, swaggerRestOperation);
- Assert.assertEquals("/path", path);
- }
-
- @Test
- public void testCreateRequestPathNoUrlPrefixHavePath(@Injectable Invocation invocation,
- @Injectable RestOperationMeta swaggerRestOperation, @Injectable Endpoint endpoint,
- @Injectable URIEndpointObject address, @Injectable URLPathBuilder builder) throws Exception {
- Map<String, Object> contextMap = new HashMap<>();
- contextMap.put(RestConst.REST_CLIENT_REQUEST_PATH, "/client/path");
-
- new Expectations() {
- {
- endpoint.getAddress();
- result = address;
- invocation.getHandlerContext();
- result = contextMap;
- }
- };
- String path = this.createRequestPath(invocation, swaggerRestOperation);
- Assert.assertEquals("/client/path", path);
- }
-
- @Test
- public void testCreateRequestPathHaveUrlPrefixNoPath(@Injectable Invocation invocation,
- @Injectable RestOperationMeta swaggerRestOperation, @Injectable Endpoint endpoint,
- @Injectable URIEndpointObject address, @Injectable URLPathBuilder builder) throws Exception {
- new Expectations() {
- {
- endpoint.getAddress();
- result = address;
- address.getFirst(Const.URL_PREFIX);
- result = "/root";
- builder.createRequestPath((Object[]) any);
- result = "/path";
- }
- };
- String path = this.createRequestPath(invocation, swaggerRestOperation);
- Assert.assertEquals("/root/path", path);
- }
-
- @Test
- public void testCreateRequestPathHaveUrlPrefixHavePath(@Injectable Invocation invocation,
- @Injectable RestOperationMeta swaggerRestOperation, @Injectable Endpoint endpoint,
- @Injectable URIEndpointObject address, @Injectable URLPathBuilder builder) throws Exception {
- Map<String, Object> contextMap = new HashMap<>();
- contextMap.put(RestConst.REST_CLIENT_REQUEST_PATH, "/client/path");
-
- new Expectations() {
- {
- endpoint.getAddress();
- result = address;
- address.getFirst(Const.URL_PREFIX);
- result = "/root";
- invocation.getHandlerContext();
- result = contextMap;
- }
- };
- String path = this.createRequestPath(invocation, swaggerRestOperation);
- Assert.assertEquals("/root/client/path", path);
- }
-
- @Test
- public void testCreateRequestPathHaveUrlPrefixHavePathAndStartWith(@Injectable Invocation invocation,
- @Injectable RestOperationMeta swaggerRestOperation, @Injectable Endpoint endpoint,
- @Injectable URIEndpointObject address, @Injectable URLPathBuilder builder) throws Exception {
- Map<String, Object> contextMap = new HashMap<>();
- contextMap.put(RestConst.REST_CLIENT_REQUEST_PATH, "/client/path");
-
- new Expectations() {
- {
- endpoint.getAddress();
- result = address;
- address.getFirst(Const.URL_PREFIX);
- result = "/client";
- invocation.getHandlerContext();
- result = contextMap;
- }
- };
- String path = this.createRequestPath(invocation, swaggerRestOperation);
- Assert.assertEquals("/client/path", path);
- }
-}
diff --git a/transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/TransportConfig.java b/transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/TransportConfig.java
index fee033c21..01f13c8d5 100644
--- a/transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/TransportConfig.java
+++ b/transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/TransportConfig.java
@@ -21,6 +21,8 @@
import com.netflix.config.DynamicPropertyFactory;
import com.netflix.config.DynamicStringProperty;
+import io.vertx.core.Verticle;
+
public final class TransportConfig {
public static final int DEFAULT_SERVER_THREAD_COUNT = 1;
@@ -32,9 +34,19 @@
// 32K
public static final int DEFAULT_SERVER_MAX_HEADER_SIZE = 32 * 1024;
+ private static Class<? extends Verticle> restServerVerticle = RestServerVerticle.class;
+
private TransportConfig() {
}
+ public static Class<? extends Verticle> getRestServerVerticle() {
+ return restServerVerticle;
+ }
+
+ public static void setRestServerVerticle(Class<? extends Verticle> restServerVerticle) {
+ TransportConfig.restServerVerticle = restServerVerticle;
+ }
+
public static String getAddress() {
DynamicStringProperty address =
DynamicPropertyFactory.getInstance().getStringProperty("cse.rest.address", null);
diff --git a/transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/VertxRestTransport.java b/transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/VertxRestTransport.java
index 66f4638db..008792c64 100644
--- a/transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/VertxRestTransport.java
+++ b/transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/VertxRestTransport.java
@@ -74,8 +74,9 @@ public boolean init() throws Exception {
DeploymentOptions options = new DeploymentOptions().setInstances(TransportConfig.getThreadCount());
SimpleJsonObject json = new SimpleJsonObject();
json.put(ENDPOINT_KEY, getEndpoint());
+ json.put(RestTransportClient.class.getName(), restClient);
options.setConfig(json);
- return VertxUtils.blockDeploy(transportVertx, RestServerVerticle.class, options);
+ return VertxUtils.blockDeploy(transportVertx, TransportConfig.getRestServerVerticle(), options);
}
@Override
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services