add response parsing

chengyouling · chengyouling · commit 255dee90ef25 · 2025-07-31T15:46:25.000+08:00
diff --git a/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/AbstractRestInvocation.java b/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/AbstractRestInvocation.java
@@ -351,7 +351,7 @@ protected void onExecuteHttpServerFiltersFinish(Response response, Throwable e)
 
     if (!(response.getResult() instanceof ServerWebSocket)) {
       try {
-        responseEx.flushBuffer();
+        responseEx.endResponse();
       } catch (Throwable flushException) {
         LOGGER.error("Failed to flush rest response, operation:{}, request uri:{}",
             getMicroserviceQualifiedName(), requestEx.getRequestURI(), flushException);
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
@@ -53,24 +53,24 @@ protected Invocation sendCreateInvocationException(Throwable throwable) {
           requestEx.getRequestURI(), e);
     }
 
-    flushResponse("UNKNOWN_OPERATION");
+    endResponse("UNKNOWN_OPERATION");
     return null;
   }
 
   @Override
   protected void sendResponse(Invocation invocation, Response response) {
     if (isDownloadFileResponseType(invocation, response)) {
       responseEx.sendPart(PartUtils.getSinglePart(null, response.getResult()))
-          .whenComplete((r, e) -> flushResponse(invocation.getMicroserviceQualifiedName()));
+          .whenComplete((r, e) -> endResponse(invocation.getMicroserviceQualifiedName()));
       return;
     }
 
-    flushResponse(invocation.getMicroserviceQualifiedName());
+    endResponse(invocation.getMicroserviceQualifiedName());
   }
 
-  private void flushResponse(String operationName) {
+  private void endResponse(String operationName) {
     try {
-      responseEx.flushBuffer();
+      responseEx.endResponse();
     } catch (Throwable flushException) {
       LOGGER.error("Failed to flush rest response, operation:{}, request uri:{}",
           operationName, requestEx.getRequestURI(), flushException);
diff --git a/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/codec/produce/ProduceEventStreamProcessor.java b/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/codec/produce/ProduceEventStreamProcessor.java
@@ -20,20 +20,16 @@
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
 
 import org.apache.commons.lang3.StringUtils;
 import org.apache.servicecomb.common.rest.codec.RestObjectMapperFactory;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.servicecomb.swagger.invocation.sse.SseEventResponseEntity;
 
 import com.fasterxml.jackson.databind.JavaType;
 
 import jakarta.ws.rs.core.MediaType;
 
 public class ProduceEventStreamProcessor implements ProduceProcessor {
-  private static final Logger LOGGER = LoggerFactory.getLogger(ProduceEventStreamProcessor.class);
-
   private int writeIndex = 0;
 
   @Override
@@ -65,15 +61,6 @@ public void doEncodeResponse(OutputStream output, Object result) throws Exceptio
   @Override
   public Object doDecodeResponse(InputStream input, JavaType type) throws Exception {
     String buffer = new String(input.readAllBytes(), StandardCharsets.UTF_8);
-    LOGGER.info("=========doDecodeResponse buffer===================>" + buffer + "stack: {}", Arrays.toString(
-        new Exception().getStackTrace()));
-    if (isResponseEntity(type)) {
-      return parseAsSseEventResponseEntity(buffer, type);
-    }
-    return parseOriginObject(buffer, type);
-  }
-
-  private Object parseAsSseEventResponseEntity(String buffer, JavaType type) throws Exception {
     SseEventResponseEntity<?> responseEntity = new SseEventResponseEntity<>();
     for (String line : buffer.split("\n")) {
       if (line.startsWith("eventId: ")) {
@@ -90,30 +77,12 @@ private Object parseAsSseEventResponseEntity(String buffer, JavaType type) throw
       }
       if (line.startsWith("data: ")) {
         responseEntity.data(RestObjectMapperFactory.getRestObjectMapper()
-            .readValue(line.substring(6), getParsObjectType(type)));
+            .readValue(line.substring(6), type));
       }
     }
     return responseEntity;
   }
 
-  private Object parseOriginObject(String buffer, JavaType type) throws Exception {
-    for (String line : buffer.split("\n")) {
-      if (line.startsWith("data: ")) {
-        return RestObjectMapperFactory.getRestObjectMapper()
-            .readValue(line.substring(6), type);
-      }
-    }
-    return null;
-  }
-
-  private boolean isResponseEntity(JavaType type) {
-    return type.getRawClass().getName().equals(SseEventResponseEntity.class.getName());
-  }
-
-  private JavaType getParsObjectType(JavaType type) {
-    return type.getBindings().getBoundType(0);
-  }
-
   @Override
   public void refreshEventId(int index) {
     this.writeIndex = index;
diff --git a/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/filter/inner/ServerRestArgsFilter.java b/common/common-rest/src/main/java/org/apache/servicecomb/common/rest/filter/inner/ServerRestArgsFilter.java
@@ -169,7 +169,7 @@ private static CompletableFuture<Response> writeResponse(
           result.completeExceptionally(e);
         }
         try {
-          responseEx.flushStreamBuffer();
+          responseEx.flushBuffer();
         } catch (IOException ex) {
           LOGGER.warn("Failed to flush buffer for Server Send Events", ex);
         }
diff --git a/common/common-rest/src/test/java/org/apache/servicecomb/common/rest/TestAbstractRestInvocation.java b/common/common-rest/src/test/java/org/apache/servicecomb/common/rest/TestAbstractRestInvocation.java
@@ -916,7 +916,7 @@ public void setStatus(int sc) {
       }
 
       @Override
-      public void flushBuffer() {
+      public void endResponse() {
         endCount.value = endCount.value + 1;
       }
 
@@ -956,7 +956,7 @@ public void setStatus(int sc) {
       }
 
       @Override
-      public void flushBuffer() {
+      public void endResponse() {
         endCount.value = endCount.value + 1;
       }
 
diff --git a/demo/demo-spring-boot-transport/demo-spring-boot-springmvc-client/src/main/java/org/apache/servicecomb/springboot/springmvc/client/ReactiveStreamIT.java b/demo/demo-spring-boot-transport/demo-spring-boot-springmvc-client/src/main/java/org/apache/servicecomb/springboot/springmvc/client/ReactiveStreamIT.java
@@ -21,7 +21,7 @@
 import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.lang3.StringUtils;
-import org.apache.servicecomb.common.rest.codec.produce.SseEventResponseEntity;
+import org.apache.servicecomb.swagger.invocation.sse.SseEventResponseEntity;
 import org.apache.servicecomb.demo.CategorizedTestCase;
 import org.apache.servicecomb.demo.TestMgr;
 import org.apache.servicecomb.demo.model.Model;
@@ -75,21 +75,18 @@ public void onSubscribe(Subscription s) {
 
       @Override
       public void onNext(String s) {
-        LOGGER.info("=========buildBufferString result===================>" + s);
         buffer.append(s);
         subscription.request(1);
       }
 
       @Override
       public void onError(Throwable t) {
-        LOGGER.info("=========buildBufferString onError===================>");
         subscription.cancel();
         countDownLatch.countDown();
       }
 
       @Override
       public void onComplete() {
-        LOGGER.info("=========buildBufferString onComplete===================>");
         countDownLatch.countDown();
       }
     });
@@ -113,20 +110,17 @@ public void onSubscribe(Subscription s) {
       @Override
       public void onNext(Model s) {
         buffer.append(s.getName()).append(s.getAge());
-        LOGGER.info("=========testSseModel result===================>" + buffer);
         subscription.request(1);
       }
 
       @Override
       public void onError(Throwable t) {
-        LOGGER.info("=========testSseModel error===================>");
         subscription.cancel();
         countDownLatch.countDown();
       }
 
       @Override
       public void onComplete() {
-        LOGGER.info("=========testSseModel onComplete===================>");
         countDownLatch.countDown();
       }
     });
diff --git a/demo/demo-spring-boot-transport/demo-spring-boot-springmvc-client/src/main/java/org/apache/servicecomb/springboot/springmvc/client/ThirdSvcConfiguration.java b/demo/demo-spring-boot-transport/demo-spring-boot-springmvc-client/src/main/java/org/apache/servicecomb/springboot/springmvc/client/ThirdSvcConfiguration.java
@@ -17,7 +17,7 @@
 
 package org.apache.servicecomb.springboot.springmvc.client;
 
-import org.apache.servicecomb.common.rest.codec.produce.SseEventResponseEntity;
+import org.apache.servicecomb.swagger.invocation.sse.SseEventResponseEntity;
 import org.apache.servicecomb.demo.model.Model;
 import org.apache.servicecomb.provider.pojo.Invoker;
 import org.reactivestreams.Publisher;
diff --git a/demo/demo-spring-boot-transport/demo-spring-boot-springmvc-server/src/main/java/org/apache/servicecomb/springboot/springmvc/server/ReactiveStreamController.java b/demo/demo-spring-boot-transport/demo-spring-boot-springmvc-server/src/main/java/org/apache/servicecomb/springboot/springmvc/server/ReactiveStreamController.java
@@ -19,7 +19,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import org.apache.servicecomb.common.rest.codec.produce.SseEventResponseEntity;
+import org.apache.servicecomb.swagger.invocation.sse.SseEventResponseEntity;
 import org.apache.servicecomb.demo.model.Model;
 import org.apache.servicecomb.provider.rest.common.RestSchema;
 import org.reactivestreams.Publisher;
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
@@ -228,4 +228,9 @@ public CompletableFuture<Void> sendPart(Part body) {
   public CompletableFuture<Void> sendBuffer(Buffer buffer) {
     throw new Error("not supported method");
   }
+
+  @Override
+  public void endResponse() throws IOException {
+    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
@@ -47,5 +47,5 @@ default Flowable<Buffer> getFlowableBuffer() {
     return null;
   }
 
-  default void flushStreamBuffer() throws IOException {};
+  void endResponse() throws IOException;
 }
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
@@ -83,7 +83,7 @@ public StatusType getStatusType() {
   }
 
   @Override
-  public void flushBuffer() throws IOException {
+  public void endResponse() throws IOException {
     byte[] bytes = getBodyBytes();
     if (bytes != null) {
       getOutputStream().write(bytes, 0, getBodyBytesLength());
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
@@ -102,7 +102,7 @@ public Collection<String> getHeaderNames() {
   }
 
   @Override
-  public void flushBuffer() {
+  public void endResponse() {
     if (context == Vertx.currentContext()) {
       internalFlushBuffer();
       return;
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
@@ -74,7 +74,7 @@ public void getBodyBytesLength() {
   }
 
   @Test
-  public void flushBuffer() throws IOException {
+  public void endResponse() throws IOException {
     Buffer buffer = Buffer.buffer();
     ServletOutputStream output = new ServletOutputStream() {
       @Override
@@ -97,12 +97,12 @@ public void write(int b) {
     responseEx = new StandardHttpServletResponseEx(response);
 
     // no body
-    responseEx.flushBuffer();
+    responseEx.endResponse();
     Assertions.assertEquals(0, buffer.length());
 
     Buffer body = Buffer.buffer().appendString("body");
     responseEx.setBodyBuffer(body);
-    responseEx.flushBuffer();
+    responseEx.endResponse();
     Assertions.assertEquals("body", buffer.toString());
   }
 
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
@@ -253,21 +253,21 @@ public void getHeaderNames() {
   }
 
   @Test
-  public void flushBuffer_sameContext() throws IOException {
-    response.flushBuffer();
+  public void endResponse_sameContext() throws IOException {
+    response.endResponse();
 
     Assertions.assertFalse(runOnContextInvoked);
   }
 
   @Test
-  public void flushBuffer_diffContext() throws IOException {
+  public void endResponse_diffContext() throws IOException {
     new Expectations() {
       {
         Vertx.currentContext();
         result = null;
       }
     };
-    response.flushBuffer();
+    response.endResponse();
 
     Assertions.assertTrue(runOnContextInvoked);
   }
diff --git a/swagger/swagger-invocation/invocation-core/pom.xml b/swagger/swagger-invocation/invocation-core/pom.xml
@@ -31,6 +31,14 @@
       <groupId>org.apache.servicecomb</groupId>
       <artifactId>swagger-generator-core</artifactId>
     </dependency>
+    <dependency>
+      <groupId>io.vertx</groupId>
+      <artifactId>vertx-rx-java3</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.vertx</groupId>
+      <artifactId>vertx-core</artifactId>
+    </dependency>
 
     <dependency>
       <groupId>org.apache.servicecomb</groupId>
diff --git a/swagger/swagger-invocation/invocation-core/src/main/java/org/apache/servicecomb/swagger/invocation/response/consumer/PublisherConsumerResponseMapper.java b/swagger/swagger-invocation/invocation-core/src/main/java/org/apache/servicecomb/swagger/invocation/response/consumer/PublisherConsumerResponseMapper.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.servicecomb.swagger.invocation.response.consumer;
+
+import org.apache.servicecomb.swagger.invocation.Response;
+import org.apache.servicecomb.swagger.invocation.sse.SseEventResponseEntity;
+
+import io.reactivex.rxjava3.core.Flowable;
+
+public class PublisherConsumerResponseMapper implements ConsumerResponseMapper {
+  private final boolean shouldExtractEntity;
+
+  public PublisherConsumerResponseMapper(boolean shouldExtractEntity) {
+    this.shouldExtractEntity = shouldExtractEntity;
+  }
+
+  @Override
+  public Object mapResponse(Response response) {
+    Flowable<SseEventResponseEntity<?>> flowable = response.getResult();
+    if (shouldExtractEntity) {
+      return flowable.map(SseEventResponseEntity::getData);
+    }
+    return flowable;
+  }
+}
diff --git a/swagger/swagger-invocation/invocation-core/src/main/java/org/apache/servicecomb/swagger/invocation/response/consumer/PublisherConsumerResponseMapperFactory.java b/swagger/swagger-invocation/invocation-core/src/main/java/org/apache/servicecomb/swagger/invocation/response/consumer/PublisherConsumerResponseMapperFactory.java
@@ -0,0 +1,43 @@
+/*
+ * 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.swagger.invocation.response.consumer;
+
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+
+import org.apache.servicecomb.swagger.invocation.response.ResponseMapperFactorys;
+import org.apache.servicecomb.swagger.invocation.sse.SseEventResponseEntity;
+import org.reactivestreams.Publisher;
+
+public class PublisherConsumerResponseMapperFactory implements ConsumerResponseMapperFactory {
+  @Override
+  public boolean isMatch(Type consumerType) {
+    if (!ParameterizedType.class.isAssignableFrom(consumerType.getClass())) {
+      return false;
+    }
+
+    return ((ParameterizedType) consumerType).getRawType().equals(Publisher.class);
+  }
+
+  @Override
+  public ConsumerResponseMapper createResponseMapper(ResponseMapperFactorys<ConsumerResponseMapper> factorys,
+      Type consumerType) {
+    Type realConsumerType = ((ParameterizedType) consumerType).getActualTypeArguments()[0];
+    return new PublisherConsumerResponseMapper(
+        !realConsumerType.getTypeName().contains(SseEventResponseEntity.class.getName()));
+  }
+}
diff --git a/swagger/swagger-invocation/invocation-core/src/main/java/org/apache/servicecomb/swagger/invocation/sse/SseEventResponseEntity.java b/swagger/swagger-invocation/invocation-core/src/main/java/org/apache/servicecomb/swagger/invocation/sse/SseEventResponseEntity.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.servicecomb.common.rest.codec.produce;
+package org.apache.servicecomb.swagger.invocation.sse;
 
 import jakarta.validation.constraints.NotNull;
 
diff --git a/swagger/swagger-invocation/invocation-core/src/main/java/org/apache/servicecomb/swagger/invocation/sse/SseEventResponseEntityProcessor.java b/swagger/swagger-invocation/invocation-core/src/main/java/org/apache/servicecomb/swagger/invocation/sse/SseEventResponseEntityProcessor.java
diff --git a/swagger/swagger-invocation/invocation-core/src/main/resources/META-INF/services/org.apache.servicecomb.swagger.generator.ResponseTypeProcessor b/swagger/swagger-invocation/invocation-core/src/main/resources/META-INF/services/org.apache.servicecomb.swagger.generator.ResponseTypeProcessor
diff --git a/swagger/swagger-invocation/invocation-core/src/main/resources/META-INF/services/org.apache.servicecomb.swagger.invocation.response.consumer.ConsumerResponseMapperFactory b/swagger/swagger-invocation/invocation-core/src/main/resources/META-INF/services/org.apache.servicecomb.swagger.invocation.response.consumer.ConsumerResponseMapperFactory
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
diff --git a/transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/http/RestClientInvocation.java b/transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/http/RestClientInvocation.java

Original file line number	Diff line number	Diff line change
`@@ -169,7 +169,7 @@ private static CompletableFuture<Response> writeResponse(`
`169`	`169`	`result.completeExceptionally(e);`
`170`	`170`	`}`
`171`	`171`	`try {`
`172`		`- responseEx.flushStreamBuffer();`
	`172`	`+ responseEx.flushBuffer();`
`173`	`173`	`} catch (IOException ex) {`
`174`	`174`	`LOGGER.warn("Failed to flush buffer for Server Send Events", ex);`
`175`	`175`	`}`
Original file line number	Diff line number	Diff line change
`@@ -916,7 +916,7 @@ public void setStatus(int sc) {`
`916`	`916`	`}`
`917`	`917`
`918`	`918`	`@Override`
`919`		`- public void flushBuffer() {`
	`919`	`+ public void endResponse() {`
`920`	`920`	`endCount.value = endCount.value + 1;`
`921`	`921`	`}`
`922`	`922`
`@@ -956,7 +956,7 @@ public void setStatus(int sc) {`
`956`	`956`	`}`
`957`	`957`
`958`	`958`	`@Override`
`959`		`- public void flushBuffer() {`
	`959`	`+ public void endResponse() {`
`960`	`960`	`endCount.value = endCount.value + 1;`
`961`	`961`	`}`
`962`	`962`
Original file line number	Diff line number	Diff line change
`@@ -228,4 +228,9 @@ public CompletableFuture<Void> sendPart(Part body) {`
`228`	`228`	`public CompletableFuture<Void> sendBuffer(Buffer buffer) {`
`229`	`229`	`throw new Error("not supported method");`
`230`	`230`	`}`
	`231`	`+`
	`232`	`+ @Override`
	`233`	`+ public void endResponse() throws IOException {`
	`234`	`+ throw new Error("not supported method");`
	`235`	`+ }`
`231`	`236`	`}`
Original file line number	Diff line number	Diff line change
`@@ -47,5 +47,5 @@ default Flowable<Buffer> getFlowableBuffer() {`
`47`	`47`	`return null;`
`48`	`48`	`}`
`49`	`49`
`50`		`- default void flushStreamBuffer() throws IOException {};`
	`50`	`+ void endResponse() throws IOException;`
`51`	`51`	`}`