[#4873] Support for SSE interface RPC calls

.github/workflows/maven.yml

@@ -36,7 +36,7 @@ jobs:
     - name: Set up jdk
       uses: actions/setup-java@v3
       with:
-        java-version: '21'
+        java-version: '17'
         distribution: 'temurin'
     - name: Set up Maven
       uses: stCarolas/[email protected]

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);

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);

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

@@ -0,0 +1,139 @@
+/*
+ * 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.common.rest.codec.produce;
+
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.servicecomb.common.rest.codec.RestObjectMapperFactory;
+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 {
+  public static final List<String> DEFAULT_DELIMITERS = Arrays.asList("\r\n", "\n", "\r");
+
+  @Override
+  public String getName() {
+    return MediaType.SERVER_SENT_EVENTS;
+  }
+
+  @Override
+  public int getOrder() {
+    return 0;
+  }
+
+  @Override
+  public void doEncodeResponse(OutputStream output, Object result) throws Exception {
+    StringBuilder bufferBuilder = new StringBuilder();
+    if (result instanceof SseEventResponseEntity<?> responseEntity) {
+      appendEventId(bufferBuilder, responseEntity.getEventId());
+      appendEvent(bufferBuilder, responseEntity.getEvent());
+      appendRetry(bufferBuilder, responseEntity.getRetry());
+      appendData(bufferBuilder, responseEntity.getData());
+      bufferBuilder.append("\n");
+      output.write(bufferBuilder.toString().getBytes(StandardCharsets.UTF_8));
+    }
+  }
+
+  @Override
+  public Object doDecodeResponse(InputStream input, JavaType type) throws Exception {
+    String buffer = new String(input.readAllBytes(), StandardCharsets.UTF_8);
+    List<String> lines = new ArrayList<>();
+    splitStringByDelimiters(buffer, lines);
+    SseEventResponseEntity<?> responseEntity = new SseEventResponseEntity<>();
+    for (String line : lines) {
+      if (line.startsWith("eventId:")) {
+        responseEntity.eventId(Integer.parseInt(line.substring("eventId:".length()).trim()));
+        continue;
+      }
+      if (line.startsWith("event:")) {
+        responseEntity.event(line.substring("event:".length()).trim());
+        continue;
+      }
+      if (line.startsWith("retry:")) {
+        responseEntity.retry(Long.parseLong(line.substring("retry:".length()).trim()));
+        continue;
+      }
+      if (line.startsWith("data:")) {
+        responseEntity.data(RestObjectMapperFactory.getRestObjectMapper()
+            .readValue(line.substring("data:".length()).trim(), type));
+      }
+    }
+    return responseEntity;
+  }
+
+  private void splitStringByDelimiters(String str, List<String> lines) {
+    boolean isContainsDelimiters = false;
+    for (String split : DEFAULT_DELIMITERS) {
+      if (str.contains(split)) {
+        isContainsDelimiters = true;
+        splitStrings(str.split(split), lines);
+      }
+    }
+    if (!isContainsDelimiters) {
+      lines.add(str);
+    }
+  }
+
+  private void splitStrings(String[] strings, List<String> lines) {
+    for (String str : strings) {
+      if (StringUtils.isEmpty(str)) {
+        continue;
+      }
+      splitStringByDelimiters(str, lines);
+    }
+  }
+
+  private void appendEventId(StringBuilder eventBuilder, Integer eventId) {
+    if (eventId == null) {
+      return;
+    }
+    eventBuilder.append("eventId: ").append(eventId.intValue()).append("\n");
+  }
+
+  private void appendEvent(StringBuilder eventBuilder, String event) {
+    if (StringUtils.isEmpty(event)) {
+      return;
+    }
+    eventBuilder.append("event: ").append(event).append("\n");
+  }
+
+  private void appendRetry(StringBuilder eventBuilder, Long retry) {
+    if (retry == null) {
+      return;
+    }
+    eventBuilder.append("retry: ").append(retry.longValue()).append("\n");
+  }
+
+  private void appendData(StringBuilder eventBuilder, Object data) throws Exception {
+    if (data == null) {
+      throw new Exception("sse response data is null!");
+    }
+    eventBuilder.append("data: ")
+        .append(RestObjectMapperFactory.getRestObjectMapper().writeValueAsString(data))
+        .append("\n");
+  }
+}

common/common-rest/src/main/java/org/apache/servicecomb/common/rest/definition/RestOperationMeta.java

@@ -88,6 +88,8 @@ public class RestOperationMeta {
   // 快速构建URL path
   private URLPathBuilder pathBuilder;
 
+  protected static final String EVENTS_MEDIA_TYPE = MediaType.SERVER_SENT_EVENTS;
+
   public void init(OperationMeta operationMeta) {
     this.operationMeta = operationMeta;
 
@@ -258,6 +260,12 @@ protected void doCreateProduceProcessors(Class<?> serialViewClass) {
           ProduceProcessorManager.INSTANCE.getOrCreateAcceptMap(serialViewClass));
     } else {
       for (String produce : produces) {
+        if (produce.equals(EVENTS_MEDIA_TYPE)) {
+          // When the produce type is event-stream, the ProduceEventStreamProcessor implementation class corresponding
+          // to event-stream is not added, and it is set to the default type ProduceJsonProcessor.
+          // In case of an exception, the response result is parsed.
+          continue;
+        }
         if (produce.contains(";")) {
           produce = produce.substring(0, produce.indexOf(";"));
         }

common/common-rest/src/main/java/org/apache/servicecomb/common/rest/filter/inner/RestServerCodecFilter.java

@@ -44,8 +44,8 @@
 import org.apache.servicecomb.swagger.invocation.Response;
 import org.springframework.stereotype.Component;
 
-import io.netty.buffer.Unpooled;
 import io.vertx.core.MultiMap;
+import io.vertx.core.buffer.Buffer;
 
 @Component
 public class RestServerCodecFilter implements ProducerFilter {
@@ -103,7 +103,7 @@ public static CompletableFuture<Response> encodeResponse(Response response, bool
     }
 
     responseEx.setContentType(produceProcessor.getName() + "; charset=utf-8");
-    try (BufferOutputStream output = new BufferOutputStream(Unpooled.compositeBuffer())) {
+    try (BufferOutputStream output = new BufferOutputStream(Buffer.buffer())) {
       produceProcessor.encodeResponse(output, response.getResult());
 
       responseEx.setBodyBuffer(output.getBuffer());

common/common-rest/src/main/java/org/apache/servicecomb/common/rest/filter/inner/ServerRestArgsFilter.java

@@ -19,11 +19,13 @@
 
 import static org.apache.servicecomb.common.rest.filter.inner.RestServerCodecFilter.isDownloadFileResponseType;
 
+import java.io.IOException;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 
 import org.apache.servicecomb.common.rest.RestConst;
 import org.apache.servicecomb.common.rest.codec.RestCodec;
+import org.apache.servicecomb.common.rest.codec.produce.ProduceEventStreamProcessor;
 import org.apache.servicecomb.common.rest.codec.produce.ProduceProcessor;
 import org.apache.servicecomb.common.rest.definition.RestOperationMeta;
 import org.apache.servicecomb.common.rest.filter.HttpServerFilter;
@@ -36,15 +38,22 @@
 import org.apache.servicecomb.swagger.invocation.Response;
 import org.apache.servicecomb.swagger.invocation.exception.ExceptionFactory;
 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;
 
 import com.netflix.config.DynamicPropertyFactory;
 
-import io.netty.buffer.Unpooled;
+import io.vertx.core.buffer.Buffer;
 
 public class ServerRestArgsFilter implements HttpServerFilter {
   private static final boolean enabled = DynamicPropertyFactory.getInstance().getBooleanProperty
       ("servicecomb.http.filter.server.serverRestArgs.enabled", true).get();
 
+  private static final Logger LOGGER = LoggerFactory.getLogger(ServerRestArgsFilter.class);
+
   @Override
   public int getOrder() {
     return -100;
@@ -74,10 +83,15 @@ public CompletableFuture<Void> beforeSendResponseAsync(Invocation invocation, Ht
       return responseEx.sendPart(PartUtils.getSinglePart(null, response.getResult()));
     }
 
-    responseEx.setContentType(produceProcessor.getName() + "; charset=utf-8");
+    if (isServerSendEvent(response)) {
+      produceProcessor = new ProduceEventStreamProcessor();
+      responseEx.setContentType(produceProcessor.getName() + "; charset=utf-8");
+      return writeServerSendEvent(invocation, response, produceProcessor, responseEx);
+    }
 
+    responseEx.setContentType(produceProcessor.getName() + "; charset=utf-8");
     CompletableFuture<Void> future = new CompletableFuture<>();
-    try (BufferOutputStream output = new BufferOutputStream(Unpooled.compositeBuffer())) {
+    try (BufferOutputStream output = new BufferOutputStream(Buffer.buffer())) {
       if (failed) {
         produceProcessor.encodeResponse(output, ((InvocationException) response.getResult()).getErrorData());
       } else {
@@ -91,4 +105,71 @@ public CompletableFuture<Void> beforeSendResponseAsync(Invocation invocation, Ht
     }
     return future;
   }
+
+  public static boolean isServerSendEvent(Response response) {
+    return response.getResult() instanceof Publisher<?>;
+  }
+
+  private static CompletableFuture<Void> writeServerSendEvent(Invocation invocation, Response response,
+      ProduceProcessor produceProcessor, HttpServletResponseEx responseEx) {
+    responseEx.setChunkedForEvent(true);
+    CompletableFuture<Void> 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;
+      }
+
+      @Override
+      public void onNext(Object o) {
+        writeResponse(responseEx, produceProcessor, o, response).whenComplete((r, e) -> {
+          if (e != null) {
+            subscription.cancel();
+            result.completeExceptionally(e);
+            return;
+          }
+          subscription.request(1);
+        });
+      }
+
+      @Override
+      public void onError(Throwable t) {
+        result.completeExceptionally(t);
+      }
+
+      @Override
+      public void onComplete() {
+        result.complete(null);
+      }
+    });
+    return result;
+  }
+
+  private static CompletableFuture<Response> writeResponse(
+      HttpServletResponseEx responseEx, ProduceProcessor produceProcessor, Object data, Response response) {
+    try (BufferOutputStream output = new BufferOutputStream(Buffer.buffer())) {
+      produceProcessor.encodeResponse(output, data);
+      CompletableFuture<Response> result = new CompletableFuture<>();
+      responseEx.sendBuffer(output.getBuffer()).whenComplete((v, e) -> {
+        if (e != null) {
+          result.completeExceptionally(e);
+        }
+        try {
+          responseEx.flushBuffer();
+        } catch (IOException ex) {
+          LOGGER.warn("Failed to flush buffer for Server Send Events", ex);
+        }
+      });
+      result.complete(response);
+      return result;
+    } catch (Throwable e) {
+      LOGGER.error("internal service error must be fixed.", e);
+      responseEx.setStatus(500);
+      return CompletableFuture.failedFuture(e);
+    }
+  }
 }

common/common-rest/src/main/resources/META-INF/services/org.apache.servicecomb.common.rest.codec.produce.ProduceProcessor

@@ -16,4 +16,5 @@
 #
 
 org.apache.servicecomb.common.rest.codec.produce.ProduceJsonProcessor
-org.apache.servicecomb.common.rest.codec.produce.ProduceTextPlainProcessor
+org.apache.servicecomb.common.rest.codec.produce.ProduceTextPlainProcessor
+org.apache.servicecomb.common.rest.codec.produce.ProduceEventStreamProcessor

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;
       }