This is an automated email from the ASF dual-hosted git repository.
liubao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/servicecomb-java-chassis.git
The following commit(s) were added to refs/heads/master by this push:
new 65a114106 [SCB-2885]refactoring buffer reader to support SSE (#4382)
65a114106 is described below
commit 65a1141065ba45ef891fb46a4e41a786219eed9f
Author: liubao68 <[email protected]>
AuthorDate: Tue Jun 25 21:24:26 2024 +0800
[SCB-2885]refactoring buffer reader to support SSE (#4382)
* [SCB-2885]refactoring buffer reader to support SSE
* fix test error
* fix test error
---
.../common/rest/RestProducerInvocationFlow.java | 22 +---
.../rest/filter/inner/RestServerCodecFilter.java | 131 ++++++++++++++++-----
.../filter/inner/RestServerCodecFilterTest.java | 7 +-
.../core/invocation/ProducerInvocationFlow.java | 6 +-
.../invocation/ProducerInvocationFlowTest.java | 2 +-
.../vertx/http/AbstractHttpServletResponse.java | 11 ++
.../vertx/http/HttpServletResponseEx.java | 5 +
.../vertx/http/StandardHttpServletResponseEx.java | 18 ++-
.../VertxServerResponseToHttpServletResponse.java | 28 +++--
.../http/TestStandardHttpServletResponseEx.java | 2 +-
...stVertxServerResponseToHttpServletResponse.java | 6 +-
.../highway/HighwayProducerInvocationFlow.java | 2 +-
12 files changed, 173 insertions(+), 67 deletions(-)
diff --git
a/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/RestProducerInvocationFlow.java
b/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/RestProducerInvocationFlow.java
index dab6d6f6d..727346c82 100644
---
a/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/RestProducerInvocationFlow.java
+++
b/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/RestProducerInvocationFlow.java
@@ -16,8 +16,6 @@
*/
package org.apache.servicecomb.common.rest;
-import static
org.apache.servicecomb.common.rest.filter.inner.RestServerCodecFilter.isDownloadFileResponseType;
-
import org.apache.servicecomb.common.rest.codec.produce.ProduceProcessor;
import
org.apache.servicecomb.common.rest.codec.produce.ProduceProcessorManager;
import org.apache.servicecomb.common.rest.filter.inner.RestServerCodecFilter;
@@ -25,11 +23,9 @@ import org.apache.servicecomb.core.Invocation;
import org.apache.servicecomb.core.exception.Exceptions;
import org.apache.servicecomb.core.invocation.InvocationCreator;
import org.apache.servicecomb.core.invocation.ProducerInvocationFlow;
-import org.apache.servicecomb.foundation.common.utils.PartUtils;
import org.apache.servicecomb.foundation.vertx.http.HttpServletRequestEx;
import org.apache.servicecomb.foundation.vertx.http.HttpServletResponseEx;
import org.apache.servicecomb.swagger.invocation.Response;
-import org.apache.servicecomb.swagger.invocation.exception.InvocationException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -48,32 +44,26 @@ public class RestProducerInvocationFlow extends
ProducerInvocationFlow {
protected Invocation sendCreateInvocationException(Throwable throwable) {
try {
Response response = Exceptions.toProducerResponse(null, throwable);
- RestServerCodecFilter.encodeResponse(null, response, false,
DEFAULT_PRODUCE_PROCESSOR, responseEx);
+ RestServerCodecFilter.encodeResponse(null, response,
DEFAULT_PRODUCE_PROCESSOR, responseEx);
} catch (Throwable e) {
LOGGER.error("Failed to send response when prepare invocation failed,
request uri:{}",
requestEx.getRequestURI(), e);
}
- flushResponse(null);
+ endResponse(null);
return null;
}
@Override
- protected void sendResponse(Invocation invocation, Response response) {
+ protected void endResponse(Invocation invocation, Response response) {
invocation.getInvocationStageTrace().startProviderSendResponse();
- boolean failed = response.getResult() instanceof InvocationException;
- if (!failed && isDownloadFileResponseType(invocation, response)) {
- responseEx.sendPart(PartUtils.getSinglePart(null, response.getResult()))
- .whenComplete((r, e) -> flushResponse(invocation));
- return;
- }
- flushResponse(invocation);
+ endResponse(invocation);
}
- private void flushResponse(Invocation invocation) {
+ private void endResponse(Invocation invocation) {
try {
- responseEx.flushBuffer();
+ responseEx.endResponse();
} catch (Throwable flushException) {
LOGGER.error("Failed to flush rest response, operation:{}, request
uri:{}",
invocation == null ? "NA" :
diff --git
a/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/filter/inner/RestServerCodecFilter.java
b/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/filter/inner/RestServerCodecFilter.java
index 348ee7f19..d47dd4bce 100644
---
a/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/filter/inner/RestServerCodecFilter.java
+++
b/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/filter/inner/RestServerCodecFilter.java
@@ -39,14 +39,16 @@ import org.apache.servicecomb.core.filter.EdgeFilter;
import org.apache.servicecomb.core.filter.Filter;
import org.apache.servicecomb.core.filter.FilterNode;
import org.apache.servicecomb.core.filter.ProviderFilter;
-import org.apache.servicecomb.foundation.common.utils.AsyncUtils;
+import org.apache.servicecomb.foundation.common.utils.PartUtils;
import org.apache.servicecomb.foundation.vertx.http.HttpServletRequestEx;
import org.apache.servicecomb.foundation.vertx.http.HttpServletResponseEx;
import org.apache.servicecomb.foundation.vertx.stream.BufferOutputStream;
import org.apache.servicecomb.swagger.invocation.Response;
import org.apache.servicecomb.swagger.invocation.context.TransportContext;
-import org.apache.servicecomb.swagger.invocation.exception.CommonExceptionData;
import org.apache.servicecomb.swagger.invocation.exception.InvocationException;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -106,54 +108,121 @@ public class RestServerCodecFilter extends
AbstractFilter implements ProviderFil
protected CompletableFuture<Response> encodeResponse(Invocation invocation,
Response response) {
invocation.onEncodeResponseStart(response);
-
HttpTransportContext transportContext = invocation.getTransportContext();
-
+ HttpServletResponseEx responseEx = transportContext.getResponseEx();
// TODO: response support JsonView
ProduceProcessor produceProcessor = ProduceProcessorManager.INSTANCE
.createProduceProcessor(invocation.getOperationMeta(),
response.getStatusCode(),
invocation.getRequestEx().getHeader(HttpHeaders.ACCEPT), null);
- HttpServletResponseEx responseEx = transportContext.getResponseEx();
- boolean download = isDownloadFileResponseType(invocation, response);
- return encodeResponse(invocation, response, download, produceProcessor,
responseEx)
+ return encodeResponse(invocation, response, produceProcessor, responseEx)
.whenComplete((r, e) -> invocation.onEncodeResponseFinish());
}
- public static CompletableFuture<Response> encodeResponse(Invocation
invocation, Response response, boolean download,
+ private static boolean isFailedResponse(Response response) {
+ return response.getResult() instanceof InvocationException;
+ }
+
+ private static CompletableFuture<Response> writePart(
+ HttpServletResponseEx responseEx, Object data, Response response) {
+ CompletableFuture<Response> result = new CompletableFuture<>();
+ responseEx.sendPart(PartUtils.getSinglePart(null, data))
+ .whenComplete((r, e) -> {
+ if (e != null) {
+ result.completeExceptionally(e);
+ return;
+ }
+ result.complete(response);
+ });
+ return result;
+ }
+
+ private static CompletableFuture<Response> writeResponse(
+ HttpServletResponseEx responseEx, ProduceProcessor produceProcessor,
Object data, Response response,
+ boolean commit) {
+ try (BufferOutputStream output = new BufferOutputStream(Buffer.buffer())) {
+ produceProcessor.encodeResponse(output, data);
+
+ CompletableFuture<Response> result = new CompletableFuture<>();
+ responseEx.setBodyBuffer(output.getBuffer()); // For extensions usage
+ if (commit) {
+ responseEx.setContentLength(output.getBuffer().length());
+ }
+ responseEx.sendBuffer(output.getBuffer()).whenComplete((v, e) -> {
+ if (e != null) {
+ result.completeExceptionally(e);
+ return;
+ }
+ result.complete(response);
+ });
+ return result;
+ } catch (Throwable e) {
+ LOGGER.error("internal service error must be fixed.", e);
+ responseEx.setStatus(500);
+ return CompletableFuture.failedFuture(e);
+ }
+ }
+
+ public static CompletableFuture<Response> encodeResponse(Invocation
invocation, Response response,
ProduceProcessor produceProcessor, HttpServletResponseEx responseEx) {
responseEx.setStatus(response.getStatusCode());
copyHeadersToHttpResponse(invocation, response.getHeaders(), responseEx);
- boolean failed = response.getResult() instanceof InvocationException;
+ if (isFailedResponse(response)) {
+ responseEx.setContentType(produceProcessor.getName());
+ return writeResponse(responseEx, produceProcessor,
((InvocationException) response.getResult()).getErrorData(),
+ response, true);
+ }
+
+ if (isDownloadFileResponseType(invocation, response)) {
+ return writePart(responseEx, response.getResult(), response);
+ }
- if (!failed && download) {
- return CompletableFuture.completedFuture(response);
+ if (isServerSendEvent(response)) {
+ responseEx.setContentType(produceProcessor.getName());
+ writeServerSendEvent(response, produceProcessor, responseEx);
}
responseEx.setContentType(produceProcessor.getName());
- try (BufferOutputStream output = new BufferOutputStream(Buffer.buffer())) {
- if (failed) {
- produceProcessor.encodeResponse(output, ((InvocationException)
response.getResult()).getErrorData());
- } else {
- produceProcessor.encodeResponse(output, response.getResult());
+ return writeResponse(responseEx, produceProcessor, response.getResult(),
response, true);
+ }
+
+ private static void writeServerSendEvent(Response response, ProduceProcessor
produceProcessor,
+ HttpServletResponseEx responseEx) {
+ responseEx.setChunked(true);
+ CompletableFuture<Response> result = new CompletableFuture<>();
+ Publisher<?> publisher = response.getResult();
+ publisher.subscribe(new Subscriber<Object>() {
+ Subscription subscription;
+
+ @Override
+ public void onSubscribe(Subscription s) {
+ s.request(1);
+ subscription = s;
}
- responseEx.setBodyBuffer(output.getBuffer());
+ @Override
+ public void onNext(Object o) {
+ writeResponse(responseEx, produceProcessor, o, response,
false).whenComplete((r, e) -> {
+ if (e != null) {
+ subscription.cancel();
+ result.completeExceptionally(e);
+ return;
+ }
+ subscription.request(1);
+ });
+ }
- return CompletableFuture.completedFuture(response);
- } catch (Throwable e) {
- LOGGER.error("internal service error must be fixed.", e);
- try (BufferOutputStream output = new
BufferOutputStream(Buffer.buffer())) {
- responseEx.setStatus(500);
- produceProcessor.encodeResponse(output, new CommonExceptionData("500",
"Internal Server Error"));
- responseEx.setBodyBuffer(output.getBuffer());
- return CompletableFuture.completedFuture(response);
- } catch (Throwable e1) {
- // we have no idea how to do, no response given to client.
- return AsyncUtils.completeExceptionally(e);
+ @Override
+ public void onError(Throwable t) {
+ result.completeExceptionally(t);
}
- }
+
+ @Override
+ public void onComplete() {
+ result.complete(response);
+ }
+ });
}
/**
@@ -167,6 +236,10 @@ public class RestServerCodecFilter extends AbstractFilter
implements ProviderFil
invocation.findResponseType(response.getStatusCode()).getRawClass());
}
+ public static boolean isServerSendEvent(Response response) {
+ return response.getResult() instanceof Publisher<?>;
+ }
+
public static void copyHeadersToHttpResponse(Invocation invocation, MultiMap
headers,
HttpServletResponseEx responseEx) {
if (headers != null) {
diff --git
a/common/common-rest/src/test/java/org/apache/servicecomb/common/rest/filter/inner/RestServerCodecFilterTest.java
b/common/common-rest/src/test/java/org/apache/servicecomb/common/rest/filter/inner/RestServerCodecFilterTest.java
index 559a6af10..0decec28b 100644
---
a/common/common-rest/src/test/java/org/apache/servicecomb/common/rest/filter/inner/RestServerCodecFilterTest.java
+++
b/common/common-rest/src/test/java/org/apache/servicecomb/common/rest/filter/inner/RestServerCodecFilterTest.java
@@ -23,6 +23,7 @@ import static
jakarta.ws.rs.core.Response.Status.INTERNAL_SERVER_ERROR;
import static
org.apache.servicecomb.core.SCBEngine.CFG_KEY_TURN_DOWN_STATUS_WAIT_SEC;
import static
org.apache.servicecomb.core.SCBEngine.DEFAULT_TURN_DOWN_STATUS_WAIT_SEC;
+import java.io.ByteArrayInputStream;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
@@ -41,6 +42,7 @@ import org.apache.servicecomb.core.definition.SchemaMeta;
import org.apache.servicecomb.core.filter.AbstractFilter;
import org.apache.servicecomb.core.filter.FilterNode;
import org.apache.servicecomb.core.invocation.InvocationFactory;
+import org.apache.servicecomb.foundation.common.part.InputStreamPart;
import
org.apache.servicecomb.foundation.test.scaffolding.exception.RuntimeExceptionWithoutStackTrace;
import org.apache.servicecomb.foundation.vertx.http.HttpServletRequestEx;
import org.apache.servicecomb.foundation.vertx.http.HttpServletResponseEx;
@@ -88,6 +90,8 @@ public class RestServerCodecFilterTest {
final MultiMap headers = MultiMap.caseInsensitiveMultiMap();
+ final Part part = new InputStreamPart("ok", new ByteArrayInputStream(new
byte[] {1, 2}));
+
final FilterNode nextNode = new FilterNode(new AbstractFilter() {
@Override
public String getName() {
@@ -96,7 +100,7 @@ public class RestServerCodecFilterTest {
@Override
public CompletableFuture<Response> onFilter(Invocation invocation,
FilterNode nextNode) {
- Response response = Response.ok("ok");
+ Response response = Response.ok(part);
response.setHeaders(headers);
return CompletableFuture.completedFuture(response);
}
@@ -125,6 +129,7 @@ public class RestServerCodecFilterTest {
Mockito.when(schemaMeta.getMicroserviceMeta()).thenReturn(microserviceMeta);
Mockito.when(operationMeta.buildBaseProviderRuntimeType()).thenReturn(invocationRuntimeType);
invocation = Mockito.spy(InvocationFactory.forProvider(endpoint,
operationMeta, null));
+
Mockito.when(responseEx.sendPart(part)).thenReturn(CompletableFuture.completedFuture(null));
}
private void mockDecodeRequestFail() {
diff --git
a/core/src/main/java/org/apache/servicecomb/core/invocation/ProducerInvocationFlow.java
b/core/src/main/java/org/apache/servicecomb/core/invocation/ProducerInvocationFlow.java
index 88371433c..7304fabae 100644
---
a/core/src/main/java/org/apache/servicecomb/core/invocation/ProducerInvocationFlow.java
+++
b/core/src/main/java/org/apache/servicecomb/core/invocation/ProducerInvocationFlow.java
@@ -76,7 +76,7 @@ public abstract class ProducerInvocationFlow {
response = Response.createFail(new
InvocationException(Status.INTERNAL_SERVER_ERROR,
new CommonExceptionData("Internal error, check logs for
details.")));
}
- sendResponse(invocation, response);
+ endResponse(invocation, response);
finishInvocation(invocation, response);
});
return;
@@ -91,7 +91,7 @@ public abstract class ProducerInvocationFlow {
response = Response.createFail(new
InvocationException(Status.INTERNAL_SERVER_ERROR,
new CommonExceptionData("Internal error, check logs for
details.")));
}
- sendResponse(invocation, response);
+ endResponse(invocation, response);
finishInvocation(invocation, response);
});
}
@@ -102,5 +102,5 @@ public abstract class ProducerInvocationFlow {
protected abstract Invocation sendCreateInvocationException(Throwable
throwable);
- protected abstract void sendResponse(Invocation invocation, Response
response);
+ protected abstract void endResponse(Invocation invocation, Response
response);
}
diff --git
a/core/src/test/java/org/apache/servicecomb/core/invocation/ProducerInvocationFlowTest.java
b/core/src/test/java/org/apache/servicecomb/core/invocation/ProducerInvocationFlowTest.java
index 7bba26f4d..e2e3b2c42 100644
---
a/core/src/test/java/org/apache/servicecomb/core/invocation/ProducerInvocationFlowTest.java
+++
b/core/src/test/java/org/apache/servicecomb/core/invocation/ProducerInvocationFlowTest.java
@@ -50,7 +50,7 @@ public class ProducerInvocationFlowTest {
}
@Override
- protected void sendResponse(Invocation invocation, Response response) {
+ protected void endResponse(Invocation invocation, Response response) {
sendInvocation = invocation;
}
}
diff --git
a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/AbstractHttpServletResponse.java
b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/AbstractHttpServletResponse.java
index 2a0559bab..43e438c14 100644
---
a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/AbstractHttpServletResponse.java
+++
b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/AbstractHttpServletResponse.java
@@ -25,6 +25,7 @@ import java.util.Locale;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
+import io.vertx.core.buffer.Buffer;
import jakarta.servlet.ServletOutputStream;
import jakarta.servlet.http.Cookie;
import jakarta.servlet.http.Part;
@@ -88,6 +89,11 @@ public abstract class AbstractHttpServletResponse extends
BodyBufferSupportImpl
throw new Error("not supported method");
}
+ @Override
+ public void endResponse() throws IOException {
+ throw new Error("not supported method");
+ }
+
@Override
public void resetBuffer() {
throw new Error("not supported method");
@@ -222,4 +228,9 @@ public abstract class AbstractHttpServletResponse extends
BodyBufferSupportImpl
public CompletableFuture<Void> sendPart(Part body) {
throw new Error("not supported method");
}
+
+ @Override
+ public CompletableFuture<Void> sendBuffer(Buffer buffer) {
+ throw new Error("not supported method");
+ }
}
diff --git
a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/HttpServletResponseEx.java
b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/HttpServletResponseEx.java
index e6ec15dc0..a565ed779 100644
---
a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/HttpServletResponseEx.java
+++
b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/HttpServletResponseEx.java
@@ -21,6 +21,7 @@ import java.io.IOException;
import java.io.PrintWriter;
import java.util.concurrent.CompletableFuture;
+import io.vertx.core.buffer.Buffer;
import io.vertx.core.http.HttpHeaders;
import jakarta.servlet.ServletOutputStream;
import jakarta.servlet.http.HttpServletResponse;
@@ -36,10 +37,14 @@ public interface HttpServletResponseEx extends
HttpServletResponse, BodyBufferSu
CompletableFuture<Void> sendPart(Part body);
+ CompletableFuture<Void> sendBuffer(Buffer buffer);
+
default void setChunked(boolean chunked) {
setHeader(HttpHeaders.TRANSFER_ENCODING.toString(),
HttpHeaders.CHUNKED.toString());
}
+ void endResponse() throws IOException;
+
@Override
default ServletOutputStream getOutputStream() throws IOException {
throw new IOException("Not allowed");
diff --git
a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/StandardHttpServletResponseEx.java
b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/StandardHttpServletResponseEx.java
index 430f2ace8..38db248ef 100644
---
a/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/StandardHttpServletResponseEx.java
+++
b/foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/StandardHttpServletResponseEx.java
@@ -82,11 +82,7 @@ public class StandardHttpServletResponseEx extends
HttpServletResponseWrapper im
}
@Override
- public void flushBuffer() throws IOException {
- byte[] bytes = getBodyBytes();
- if (bytes != null) {
- getOutputStream().write(bytes, 0, getBodyBytesLength());
- }
+ public void endResponse() throws IOException {
super.flushBuffer();
}
@@ -121,4 +117,16 @@ public class StandardHttpServletResponseEx extends
HttpServletResponseWrapper im
Context context = Vertx.currentContext();
return new PumpFromPart(context, part).toOutputStream(outputStream, false);
}
+
+ @Override
+ public CompletableFuture<Void> sendBuffer(Buffer buffer) {
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ try {
+ getOutputStream().write(buffer.getBytes(), 0, buffer.length());
+ future.complete(null);
+ } catch (IOException e) {
+ future.completeExceptionally(e);
+ }
+ 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 7e28e8bb7..bd4f45b5e 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
@@ -26,6 +26,7 @@ import
org.apache.servicecomb.foundation.vertx.stream.PumpFromPart;
import io.vertx.core.Context;
import io.vertx.core.Vertx;
+import io.vertx.core.buffer.Buffer;
import io.vertx.core.http.HttpServerResponse;
import jakarta.servlet.http.Part;
import jakarta.ws.rs.core.HttpHeaders;
@@ -50,6 +51,11 @@ public class VertxServerResponseToHttpServletResponse
extends AbstractHttpServle
serverResponse.headers().set(HttpHeaders.CONTENT_TYPE, type);
}
+ @Override
+ public void setContentLength(int len) {
+ serverResponse.headers().set(HttpHeaders.CONTENT_LENGTH,
String.valueOf(len));
+ }
+
@Override
public void setStatus(int sc) {
@@ -100,7 +106,7 @@ public class VertxServerResponseToHttpServletResponse
extends AbstractHttpServle
}
@Override
- public void flushBuffer() {
+ public void endResponse() {
if (context == Vertx.currentContext()) {
internalFlushBuffer();
return;
@@ -110,12 +116,7 @@ public class VertxServerResponseToHttpServletResponse
extends AbstractHttpServle
}
public void internalFlushBuffer() {
- if (bodyBuffer == null) {
- serverResponse.end();
- return;
- }
-
- serverResponse.end(bodyBuffer);
+ serverResponse.end();
}
@Override
@@ -127,6 +128,19 @@ public class VertxServerResponseToHttpServletResponse
extends AbstractHttpServle
return new PumpFromPart(context, part).toWriteStream(serverResponse, null);
}
+ @Override
+ public CompletableFuture<Void> sendBuffer(Buffer buffer) {
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ serverResponse.write(buffer).onComplete(result -> {
+ if (result.failed()) {
+ future.completeExceptionally(result.cause());
+ } else {
+ future.complete(null);
+ }
+ });
+ return future;
+ }
+
@Override
public void setChunked(boolean chunked) {
serverResponse.setChunked(chunked);
diff --git
a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/http/TestStandardHttpServletResponseEx.java
b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/http/TestStandardHttpServletResponseEx.java
index c672fb288..289d5b152 100644
---
a/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/http/TestStandardHttpServletResponseEx.java
+++
b/foundations/foundation-vertx/src/test/java/org/apache/servicecomb/foundation/vertx/http/TestStandardHttpServletResponseEx.java
@@ -103,7 +103,7 @@ public class TestStandardHttpServletResponseEx {
Buffer body = Buffer.buffer().appendString("body");
responseEx.setBodyBuffer(body);
responseEx.flushBuffer();
- Assertions.assertEquals("body", buffer.toString());
+ Assertions.assertEquals(0, buffer.length());
}
@Test
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 66c675475..1ab6d81bc 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
@@ -248,7 +248,7 @@ public class TestVertxServerResponseToHttpServletResponse {
@Test
public void flushBuffer_sameContext() throws IOException {
- response.flushBuffer();
+ response.endResponse();
Assertions.assertFalse(runOnContextInvoked);
}
@@ -261,7 +261,7 @@ public class TestVertxServerResponseToHttpServletResponse {
result = null;
}
};
- response.flushBuffer();
+ response.endResponse();
Assertions.assertTrue(runOnContextInvoked);
}
@@ -278,7 +278,7 @@ public class TestVertxServerResponseToHttpServletResponse {
response.setBodyBuffer(Buffer.buffer());
response.internalFlushBuffer();
- Assertions.assertTrue(flushWithBody);
+ Assertions.assertFalse(flushWithBody);
}
@Test
diff --git
a/transports/transport-highway/src/main/java/org/apache/servicecomb/transport/highway/HighwayProducerInvocationFlow.java
b/transports/transport-highway/src/main/java/org/apache/servicecomb/transport/highway/HighwayProducerInvocationFlow.java
index f973f9e1c..3f38fd760 100644
---
a/transports/transport-highway/src/main/java/org/apache/servicecomb/transport/highway/HighwayProducerInvocationFlow.java
+++
b/transports/transport-highway/src/main/java/org/apache/servicecomb/transport/highway/HighwayProducerInvocationFlow.java
@@ -56,7 +56,7 @@ public class HighwayProducerInvocationFlow extends
ProducerInvocationFlow {
}
@Override
- protected void sendResponse(Invocation invocation, Response response) {
+ protected void endResponse(Invocation invocation, Response response) {
HighwayTransportContext transportContext =
invocation.getTransportContext();
connection.write(transportContext.getResponseBuffer());
}
