for test

Author: chengyouling

common/common-rest/src/main/java/org/apache/servicecomb/common/rest/codec/produce/ProduceEventStreamProcessor.java

@@ -20,16 +20,21 @@
 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.apache.servicecomb.swagger.sse.SseEventResponseEntity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 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
@@ -61,37 +66,35 @@ 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()));
     SseEventResponseEntity<?> responseEntity = new SseEventResponseEntity<>();
+    boolean isResponseEntity = false;
     for (String line : buffer.split("\n")) {
       if (line.startsWith("eventId: ")) {
         responseEntity.eventId(Integer.parseInt(line.substring(9)));
         continue;
       }
       if (line.startsWith("event: ")) {
         responseEntity.event(line.substring(7));
+        isResponseEntity = true;
         continue;
       }
       if (line.startsWith("retry: ")) {
         responseEntity.retry(Long.parseLong(line.substring(7)));
+        isResponseEntity = true;
         continue;
       }
       if (line.startsWith("data: ")) {
         responseEntity.data(RestObjectMapperFactory.getRestObjectMapper().readValue(line.substring(6), type));
       }
     }
-    if (isNotResponseEntity(responseEntity)) {
-      writeIndex++;
+    if (!isResponseEntity) {
       return responseEntity.getData();
     }
     return responseEntity;
   }
 
-  private boolean isNotResponseEntity(SseEventResponseEntity<?> responseEntity) {
-    return StringUtils.isEmpty(responseEntity.getEvent())
-        && responseEntity.getRetry() == null
-        && (responseEntity.getEventId() != null && responseEntity.getEventId() == writeIndex);
-  }
-
   @Override
   public void refreshEventId(int index) {
     this.writeIndex = index;

demo/demo-spring-boot-transport/demo-spring-boot-springmvc-client/src/main/java/org/apache/servicecomb/springboot/springmvc/client/ReactiveStreamIT.java

@@ -29,12 +29,16 @@
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Component;
 
 @Component
 public class ReactiveStreamIT implements CategorizedTestCase {
+  private static final Logger LOGGER = LoggerFactory.getLogger(ReactiveStreamIT.class);
+
   @Autowired
   @Qualifier("reactiveStreamProvider")
   ReactiveStreamClient reactiveStreamProvider;
@@ -58,8 +62,8 @@ private void testSseStringWithParam(ReactiveStreamClient client) throws Exceptio
   }
 
   private String buildBufferString(Publisher<String> result) throws Exception {
-    StringBuilder buffer = new StringBuilder();
     CountDownLatch countDownLatch = new CountDownLatch(1);
+    StringBuilder buffer = new StringBuilder();
     result.subscribe(new Subscriber<>() {
       Subscription subscription;
 
@@ -71,18 +75,21 @@ 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();
       }
     });
@@ -106,17 +113,20 @@ 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();
       }
     });
@@ -144,17 +154,20 @@ public void onNext(SseEventResponseEntity<Model> responseEntity) {
         }
         buffer.append(((Model) responseEntity.getData()).getName())
             .append(((Model) responseEntity.getData()).getAge());
+        LOGGER.info("=========testSseResponseEntity result===================>" + buffer);
         subscription.request(1);
       }
 
       @Override
       public void onError(Throwable t) {
+        LOGGER.info("=========testSseResponseEntity error===================>");
         subscription.cancel();
         countDownLatch.countDown();
       }
 
       @Override
       public void onComplete() {
+        LOGGER.info("=========testSseResponseEntity onComplete===================>");
         countDownLatch.countDown();
       }
     });

demo/demo-spring-boot-transport/demo-spring-boot-springmvc-server/src/main/java/org/apache/servicecomb/springboot/springmvc/server/ReactiveStreamController.java

@@ -16,6 +16,8 @@
  */
 package org.apache.servicecomb.springboot.springmvc.server;
 
+import java.util.Arrays;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -34,24 +36,28 @@
 public class ReactiveStreamController {
   @GetMapping("/sseString")
   public Publisher<String> sseString() {
-    return Flowable.fromArray("a", "b", "c");
+    List<String> data = Arrays.asList("a", "b", "c");
+    return Flowable.intervalRange(0, 3, 5, 1, TimeUnit.SECONDS)
+        .map(item -> data.get(item.intValue()));
   }
 
   @GetMapping("/sseStringWithParam")
   public Publisher<String> sseStringWithParam(@RequestParam(name = "name") String name) {
-    return Flowable.fromArray("a", "b", "c", name);
+    List<String> data = Arrays.asList("a", "b", "c", name);
+    return Flowable.intervalRange(0, 4, 5, 1, TimeUnit.SECONDS)
+        .map(item -> data.get(item.intValue()));
   };
 
   @GetMapping("/sseModel")
   public Publisher<Model> sseModel() {
-    return Flowable.intervalRange(0, 5, 0, 1, TimeUnit.SECONDS)
+    return Flowable.intervalRange(0, 5, 5, 1, TimeUnit.SECONDS)
         .map(item -> new Model("jack", item.intValue()));
   }
 
   @GetMapping("/sseResponseEntity")
   public Publisher<SseEventResponseEntity<Model>> sseResponseEntity() {
     AtomicInteger index = new AtomicInteger(0);
-    return Flowable.intervalRange(0, 3, 0, 1, TimeUnit.SECONDS)
+    return Flowable.intervalRange(0, 3, 5, 1, TimeUnit.SECONDS)
         .map(item -> new SseEventResponseEntity<Model>()
             .event("test" + index)
             .eventId(index.getAndIncrement())