[#4873] Support for SSE interface RPC calls #4875

chengyouling · 2025-07-09T07:54:57Z

No description provided.

…hema

… add SseEventResponseEntity<T> adapt setting retry/event attributes

liubao68 · 2025-07-23T09:26:02Z

.../consumer/src/main/java/org/apache/servicecomb/samples/ConsumerReactiveStreamController.java

+ * 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.samples;
+
+


Test cases here will not run as integration tests. Can you add tests in some other module that run as integration tests?

yes, will add in spring-mvc demo

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

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

dependencies/default/pom.xml

...src/main/java/org/apache/servicecomb/transport/rest/client/http/DefaultHttpClientFilter.java

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

liubao68 · 2025-08-05T11:37:00Z

...-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/HttpServletResponseEx.java

+  default void setChunkedForEvent(boolean chunked) {
+    // not set header transfer-encoding=chunked in Rest Over Servlet, or will have Multiple in response.
+  }


Maybe change setChunked implementation to first check if header exists and then add it?

The chunked header is empty when constructing the response in Rest Over Servlet model, so it be always add chunked header information, and tomcat itself will actively construct the chunked response header in the streaming scenario.

yhs0092 · 2025-08-09T08:53:01Z

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

+  public Object doDecodeResponse(InputStream input, JavaType type) throws Exception {
+    String buffer = new String(input.readAllBytes(), StandardCharsets.UTF_8);
+    SseEventResponseEntity<?> responseEntity = new SseEventResponseEntity<>();
+    for (String line : buffer.split("\n")) {


这里可能要识别一下\n, \r, \r\n 三种场景, 对SSE协议而言, 这三种场景都是合法的行分隔方式. 参考RFC8895 3.4 节的描述:

The server continually sends messages. Each message has one or more lines, where a line is terminated by a carriage return immediately followed by a new line, a carriage return not immediately followed by a new line, or a new line not immediately preceded by a carriage return. A message is terminated by a blank line (two line terminators in a row).

补充文档链接: https://datatracker.ietf.org/doc/html/rfc8895

yhs0092 · 2025-08-09T08:56:05Z

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

+        responseEntity.data(RestObjectMapperFactory.getRestObjectMapper()
+            .readValue(line.substring(6), type));
+      }
+    }


这里要不要加个消息合法性校验, 比如之前集成测试遇到过的消息粘包导致的解析问题. 如果字段重复了, 打一个告警日志会好分析很多.

yhs0092 · 2025-08-09T09:05:43Z

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

    }
  }
+
+  default void refreshEventId(int index) {


refreshEventId 是 SSE 场景特有的方法, 但我们把它加在 ProduceProcessor 里面去了, 这通常并不合适.
建议考虑把它放在 ProduceEventStreamProcessor 中.

yhs0092 · 2025-08-09T09:23:03Z

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

  }

+  private boolean checkServerSendEvents() {
+    return !CollectionUtils.isEmpty(produces) && produces.contains(EVENTS_MEDIA_TYPE);


MIME type 取值为 text/event-stream;charset=utf-8 也是合法的. 同样是这个类里的 doCreateProduceProcessors 方法在处理 produces 属性值的时候就考虑到了 MIME type 带分号的情况. 这里建议也优化一下实现细节.

yhs0092 · 2025-08-09T10:09:52Z

...nt/src/main/java/org/apache/servicecomb/transport/rest/client/http/RestClientInvocation.java

    }

+    if (restOperationMeta != null && restOperationMeta.isServerSendEvents()) {
+      processFlowableResponseBody(FlowableHelper.toFlowable(httpClientResponse));


这里可能需要再确认一下, 如果微服务客户端调用服务端的SSE接口方法, 但是服务端报错了, 比如返回 403 状态码的 application/json 响应, 那么这里是否就不应该以 SSE 的形式去处理 response body 了呢?

换句话说, 这里的 if 判断条件应该以服务端返回的实际 Content-Type 为准会更合适一些?

…esponseMapper

yhs0092 · 2025-08-13T07:16:19Z

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

+      appendData(bufferBuilder, responseEntity.getData());
+      bufferBuilder.append("\n");
+      output.write(bufferBuilder.toString().getBytes(StandardCharsets.UTF_8));
+    }


建议保留 else 分支, 打印告警日志 ---- 如果 result 类型不是 SseEventResponseEntity, 我们不应该悄无声息地把错误放过去了.

yhs0092 · 2025-08-14T08:25:51Z

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

+    if (eventId == null) {
+      return;
+    }
+    eventBuilder.append("eventId: ").append(eventId.intValue()).append("\n");


协议规定的消息ID的名字是"id", 不是"eventId", 这里需要改一下

yhs0092 · 2025-08-14T08:29:34Z

...apache/servicecomb/swagger/invocation/response/producer/PublisherProducerResponseMapper.java

+public class PublisherProducerResponseMapper implements ProducerResponseMapper {
+  private final boolean shouldConstructEntity;
+
+  private int writeIndex = 0;


PublisherProducerResponseMapper 是全局共享的实例. 消息ID如果是通过这个对象的属性累加得到的, 就会存在不同SSE消息流之间 ID 串线的问题.

yhs0092 · 2025-08-14T14:41:46Z

...core/src/main/java/org/apache/servicecomb/swagger/invocation/sse/SseEventResponseEntity.java

+    if (this.data != null) {
+      LOGGER.warn("origin data: [{}] is exists, overridden by the current value: [{}]", this.data, data);
+    }
+    this.data = (T) data;


同一条消息里, 可以有多行data, 所以这里还不能直接打告警日志再覆盖.

还是参考 https://datatracker.ietf.org/doc/html/rfc8895

If a message has more than one "data" line, the value of the data field is the concatenation of the values on those lines. There can be only one "event" and "id" line per message. The "data" field is required; the others are optional.

下图是RFC文档给出来的样例(注意红框部分):

这里建议参考一下 Spring 的 org.springframework.web.servlet.mvc.method.annotation.SseEmitter.SseEventBuilderImpl#data(java.lang.Object, org.springframework.http.MediaType) 方法. 它在处理 SSE 的消息的时候允许业务在同一条消息中添加多个 data 字段, 并且如果消息中包含换行符时, 自动将一个data字段拆分成多个data字段存放在消息中.
(当然这里是服务端的逻辑, 客户端应该也有类似的处理措施).

我觉得这里刚好顺带解答了另外一个问题, 就是如果\r\n, \r, \n都是 SSE 消息的本身的换行符, 那么消息内容中的换行符应该如何表示? 从 Spring 给的解决方案来看, 大概率就是一条消息中包含多个 data 字段, 就表示这条消息有多行.

…sistent object states.

…arios

[apache#4873] Support for SSE interface RPC calls and register SSE sc…

40d7c41

…hema

chengyouling self-assigned this Jul 9, 2025

chengyouling added 3 commits July 23, 2025 11:18

When reconnecting, refresh the eventId based on the last-event-id and…

14c05e7

… add SseEventResponseEntity<T> adapt setting retry/event attributes

fixed test problems

7eda5d9

fixed style problem

517480b

liubao68 reviewed Jul 23, 2025

View reviewed changes

chengyouling added 6 commits July 24, 2025 14:51

add testcase

500e8b3

add testcase

5f3a593

only test rest

a7135bd

test reactive testcase

58050a5

move testcase to spring-boot-springmvc

45625a0

for test

af92830

chengyouling force-pushed the 2.9.x-sse branch from 7de4a7c to af92830 Compare July 29, 2025 02:34

chengyouling added 2 commits July 29, 2025 11:24

adjust Flowable result

0332c62

fixed testcase

0cd0c47

yanghao605 reviewed Jul 29, 2025

View reviewed changes

.../main/java/org/apache/servicecomb/common/rest/codec/produce/ProduceEventStreamProcessor.java Outdated Show resolved Hide resolved

yanghao605 reviewed Jul 29, 2025

View reviewed changes

...rest/src/main/java/org/apache/servicecomb/common/rest/filter/inner/ServerRestArgsFilter.java Outdated Show resolved Hide resolved

yanghao605 reviewed Jul 29, 2025

View reviewed changes

dependencies/default/pom.xml Show resolved Hide resolved

yanghao605 reviewed Jul 29, 2025

View reviewed changes

...src/main/java/org/apache/servicecomb/transport/rest/client/http/DefaultHttpClientFilter.java Outdated Show resolved Hide resolved

yanghao605 reviewed Jul 29, 2025

View reviewed changes

.../main/java/org/apache/servicecomb/common/rest/codec/produce/ProduceEventStreamProcessor.java Outdated Show resolved Hide resolved

chengyouling added 7 commits July 30, 2025 10:03

adjust flowable response decode

0e3e646

for test

9f7a9dc

add flushBuffer

7076c84

change using flushStreamBuffer for test

76da23e

fixed testcase

2b32da4

add response parsing

255dee9

fixed Multiple transfer-encoding=chunked in Rest Over Servlet

7fb5f75

liubao68 reviewed Aug 5, 2025

View reviewed changes

yhs0092 reviewed Aug 9, 2025

View reviewed changes

chengyouling added 2 commits August 12, 2025 16:17

fixed review options，compatible use \r\n \n \r，add PublisherProducerR…

cfbe610

…esponseMapper

delete not use EVENTS_MEDIA_TYPE

10255be

yhs0092 reviewed Aug 13, 2025

View reviewed changes

add warn for encoding

1adfc0b

yhs0092 reviewed Aug 14, 2025

View reviewed changes

adjust eventId to id, resolution concurrency using writeIndex problem

4c2c34e

yhs0092 reviewed Aug 14, 2025

View reviewed changes

chengyouling added 4 commits August 15, 2025 16:59

Compatibility settings for multiple data

63e6c1a

Compatible with both client and server, addressing scenarios of incon…

0812b8b

…sistent object states.

add catch for subscribe

e38bb52

Compatible with both packet aggregation and packet fragmentation scen…

0da8280

…arios

yhs0092 approved these changes Aug 20, 2025

View reviewed changes

chengyouling merged commit cacfb79 into apache:2.9.x Aug 20, 2025
6 checks passed

chengyouling deleted the 2.9.x-sse branch January 6, 2026 07:29

[#4873] Support for SSE interface RPC calls #4875

[#4873] Support for SSE interface RPC calls #4875

Uh oh!

Conversation

chengyouling commented Jul 9, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

chengyouling Aug 6, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yhs0092 Aug 9, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

chengyouling Aug 6, 2025 •

edited

Loading

yhs0092 Aug 9, 2025 •

edited

Loading