You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@servicecomb.apache.org by "chengyouling (via GitHub)" <gi...@apache.org> on 2025/07/09 07:55:01 UTC
[PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
chengyouling opened a new pull request, #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "chengyouling (via GitHub)" <gi...@apache.org>.
chengyouling commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2255766252
##########
foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/HttpServletResponseEx.java:
##########
@@ -48,4 +48,8 @@ default Flowable<Buffer> getFlowableBuffer() {
}
void endResponse() throws IOException;
+
+ default void setChunkedForEvent(boolean chunked) {
+ // not set header transfer-encoding=chunked in Rest Over Servlet, or will have Multiple in response.
+ }
Review Comment:
add checked in setChunked method still have multiple headers of HttpServletResponseEx, because there header is null, code like this:
default void setChunked(boolean chunked) {
if (getHeader(HttpHeaders.TRANSFER_ENCODING.toString()) != null) {
return;
}
setHeader(HttpHeaders.TRANSFER_ENCODING.toString(), HttpHeaders.CHUNKED.toString());
}
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "chengyouling (via GitHub)" <gi...@apache.org>.
chengyouling commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2255766252
##########
foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/HttpServletResponseEx.java:
##########
@@ -48,4 +48,8 @@ default Flowable<Buffer> getFlowableBuffer() {
}
void endResponse() throws IOException;
+
+ default void setChunkedForEvent(boolean chunked) {
+ // not set header transfer-encoding=chunked in Rest Over Servlet, or will have Multiple in response.
+ }
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "yhs0092 (via GitHub)" <gi...@apache.org>.
yhs0092 commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2264611657
##########
common/common-rest/src/main/java/org/apache/servicecomb/common/rest/codec/produce/ProduceEventStreamProcessor.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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 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 {
+ private int writeIndex = 0;
+
+ @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());
+ } else {
+ appendEventId(bufferBuilder, writeIndex++);
+ appendData(bufferBuilder, result);
+ }
+ 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);
+ SseEventResponseEntity<?> responseEntity = new SseEventResponseEntity<>();
+ 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));
+ continue;
+ }
+ if (line.startsWith("retry: ")) {
+ responseEntity.retry(Long.parseLong(line.substring(7)));
+ continue;
+ }
+ if (line.startsWith("data: ")) {
+ responseEntity.data(RestObjectMapperFactory.getRestObjectMapper()
+ .readValue(line.substring(6), type));
+ }
+ }
Review Comment:
这里要不要加个消息合法性校验, 比如之前集成测试遇到过的消息粘包导致的解析问题. 如果字段重复了, 打一个告警日志会好分析很多.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "yanghao605 (via GitHub)" <gi...@apache.org>.
yanghao605 commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2238790675
##########
common/common-rest/src/main/java/org/apache/servicecomb/common/rest/codec/produce/ProduceEventStreamProcessor.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.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
+ 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());
+ } else {
+ appendEventId(bufferBuilder, writeIndex++);
+ appendData(bufferBuilder, result);
+ }
+ 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);
+ LOGGER.info("=========doDecodeResponse buffer===================>" + buffer + "stack: {}", Arrays.toString(
Review Comment:
这里直接日志打印buffer应该是不必要的吧？debug？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "yhs0092 (via GitHub)" <gi...@apache.org>.
yhs0092 commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2275893995
##########
swagger/swagger-invocation/invocation-core/src/main/java/org/apache/servicecomb/swagger/invocation/response/producer/PublisherProducerResponseMapper.java:
##########
@@ -0,0 +1,45 @@
+/*
+ * 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.producer;
+
+import org.apache.servicecomb.swagger.invocation.Response;
+import org.apache.servicecomb.swagger.invocation.sse.SseEventResponseEntity;
+
+import io.reactivex.rxjava3.core.Flowable;
+import jakarta.ws.rs.core.Response.StatusType;
+
+public class PublisherProducerResponseMapper implements ProducerResponseMapper {
+ private final boolean shouldConstructEntity;
+
+ private int writeIndex = 0;
Review Comment:
PublisherProducerResponseMapper 是全局共享的实例. 消息ID如果是通过这个对象的属性累加得到的, 就会存在不同SSE消息流之间 ID 串线的问题.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "chengyouling (via GitHub)" <gi...@apache.org>.
chengyouling commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2244629400
##########
transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/http/DefaultHttpClientFilter.java:
##########
@@ -103,7 +106,13 @@ protected Response extractResponse(Invocation invocation, HttpServletResponseEx
}
try {
- result = produceProcessor.decodeResponse(responseEx.getBodyBuffer(), responseType);
+ if (responseEx.getFlowableBuffer() == null) {
+ result = produceProcessor.decodeResponse(responseEx.getBodyBuffer(), responseType);
+ } else {
+ Flowable<Buffer> flowable = responseEx.getFlowableBuffer();
+ ProduceProcessor finalProduceProcessor = new ProduceEventStreamProcessor();
+ result = flowable.map(buffer -> extractFlowableBody(finalProduceProcessor, responseType, buffer));
Review Comment:
后期会考虑消息截断拼接的事项
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "yanghao605 (via GitHub)" <gi...@apache.org>.
yanghao605 commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2238809338
##########
dependencies/default/pom.xml:
##########
@@ -397,6 +399,21 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.reactivestreams</groupId>
+ <artifactId>reactive-streams</artifactId>
Review Comment:
reactive-streams已经很久没有新的版本了，是否考虑使用其它软件替代？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "chengyouling (via GitHub)" <gi...@apache.org>.
chengyouling commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2244623477
##########
dependencies/default/pom.xml:
##########
@@ -397,6 +399,21 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.reactivestreams</groupId>
+ <artifactId>reactive-streams</artifactId>
Review Comment:
暂时与3.x保持一致
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "chengyouling (via GitHub)" <gi...@apache.org>.
chengyouling commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2245116911
##########
common/common-rest/src/main/java/org/apache/servicecomb/common/rest/filter/inner/ServerRestArgsFilter.java:
##########
@@ -91,4 +105,75 @@ 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.setChunked(true);
Review Comment:
增加setChunkedForEvent方法独立处理sse场景
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "yhs0092 (via GitHub)" <gi...@apache.org>.
yhs0092 commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2264610884
##########
common/common-rest/src/main/java/org/apache/servicecomb/common/rest/codec/produce/ProduceEventStreamProcessor.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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 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 {
+ private int writeIndex = 0;
+
+ @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());
+ } else {
+ appendEventId(bufferBuilder, writeIndex++);
+ appendData(bufferBuilder, result);
+ }
+ 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);
+ SseEventResponseEntity<?> responseEntity = new SseEventResponseEntity<>();
+ for (String line : buffer.split("\n")) {
Review Comment:
这里可能要识别一下`\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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "chengyouling (via GitHub)" <gi...@apache.org>.
chengyouling commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2269073076
##########
common/common-rest/src/main/java/org/apache/servicecomb/common/rest/codec/produce/ProduceEventStreamProcessor.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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 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 {
+ private int writeIndex = 0;
+
+ @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());
+ } else {
+ appendEventId(bufferBuilder, writeIndex++);
+ appendData(bufferBuilder, result);
+ }
+ 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);
+ SseEventResponseEntity<?> responseEntity = new SseEventResponseEntity<>();
+ for (String line : buffer.split("\n")) {
Review Comment:
fixed
##########
common/common-rest/src/main/java/org/apache/servicecomb/common/rest/codec/produce/ProduceEventStreamProcessor.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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 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 {
+ private int writeIndex = 0;
+
+ @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());
+ } else {
+ appendEventId(bufferBuilder, writeIndex++);
+ appendData(bufferBuilder, result);
+ }
+ 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);
+ SseEventResponseEntity<?> responseEntity = new SseEventResponseEntity<>();
+ 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));
+ continue;
+ }
+ if (line.startsWith("retry: ")) {
+ responseEntity.retry(Long.parseLong(line.substring(7)));
+ continue;
+ }
+ if (line.startsWith("data: ")) {
+ responseEntity.data(RestObjectMapperFactory.getRestObjectMapper()
+ .readValue(line.substring(6), type));
+ }
+ }
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "chengyouling (via GitHub)" <gi...@apache.org>.
chengyouling commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2255766252
##########
foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/HttpServletResponseEx.java:
##########
@@ -48,4 +48,8 @@ default Flowable<Buffer> getFlowableBuffer() {
}
void endResponse() throws IOException;
+
+ default void setChunkedForEvent(boolean chunked) {
+ // not set header transfer-encoding=chunked in Rest Over Servlet, or will have Multiple in response.
+ }
Review Comment:
add checked in setChunked method still have multiple headers of HttpServletResponseEx, because there header is null, so I think it's better to keep using setChunkedForEvent method to avoid affecting other abilities, code like this:
default void setChunked(boolean chunked) {
if (getHeader(HttpHeaders.TRANSFER_ENCODING.toString()) != null) {
return;
}
setHeader(HttpHeaders.TRANSFER_ENCODING.toString(), HttpHeaders.CHUNKED.toString());
}
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "yhs0092 (via GitHub)" <gi...@apache.org>.
yhs0092 commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2272302223
##########
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));
+ }
Review Comment:
建议保留 else 分支, 打印告警日志 ---- 如果 result 类型不是 `SseEventResponseEntity`, 我们不应该悄无声息地把错误放过去了.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "yanghao605 (via GitHub)" <gi...@apache.org>.
yanghao605 commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2238818556
##########
transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/http/DefaultHttpClientFilter.java:
##########
@@ -103,7 +106,13 @@ protected Response extractResponse(Invocation invocation, HttpServletResponseEx
}
try {
- result = produceProcessor.decodeResponse(responseEx.getBodyBuffer(), responseType);
+ if (responseEx.getFlowableBuffer() == null) {
+ result = produceProcessor.decodeResponse(responseEx.getBodyBuffer(), responseType);
+ } else {
+ Flowable<Buffer> flowable = responseEx.getFlowableBuffer();
+ ProduceProcessor finalProduceProcessor = new ProduceEventStreamProcessor();
+ result = flowable.map(buffer -> extractFlowableBody(finalProduceProcessor, responseType, buffer));
Review Comment:
这里接口服务端响应的时候，有测试过单次响应时，数据流过大的场景吗？比如服务一次事件返回了10000个字符串，vert.x内部接收数据时存在buffer，貌似会出现截断的情况
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "chengyouling (via GitHub)" <gi...@apache.org>.
chengyouling commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2269074488
##########
transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/http/RestClientInvocation.java:
##########
@@ -218,6 +220,11 @@ protected void handleResponse(HttpClientResponse httpClientResponse) {
return;
}
+ if (restOperationMeta != null && restOperationMeta.isServerSendEvents()) {
+ processFlowableResponseBody(FlowableHelper.toFlowable(httpClientResponse));
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "yhs0092 (via GitHub)" <gi...@apache.org>.
yhs0092 commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2275878086
##########
common/common-rest/src/main/java/org/apache/servicecomb/common/rest/codec/produce/ProduceEventStreamProcessor.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * 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 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);
+
+ 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));
+ } else {
+ LOGGER.warn("Does not support encoding objects other than SseEventResponseEntity!");
+ }
+ }
+
+ @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");
Review Comment:
协议规定的消息ID的名字是"id", 不是"eventId", 这里需要改一下
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "yanghao605 (via GitHub)" <gi...@apache.org>.
yanghao605 commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2238825630
##########
common/common-rest/src/main/java/org/apache/servicecomb/common/rest/codec/produce/ProduceEventStreamProcessor.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.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
+ 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());
+ } else {
+ appendEventId(bufferBuilder, writeIndex++);
+ appendData(bufferBuilder, result);
+ }
+ 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);
+ LOGGER.info("=========doDecodeResponse buffer===================>" + buffer + "stack: {}", Arrays.toString(
+ new Exception().getStackTrace()));
+ SseEventResponseEntity<?> responseEntity = new SseEventResponseEntity<>();
+ boolean isResponseEntity = false;
+ for (String line : buffer.split("\n")) {
Review Comment:
这里的基于\n区分事件，如果服务端在吐数据时，内部也有\n如何处理？是sse格式上要求\n是用来区分事件的标识符吗？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "yanghao605 (via GitHub)" <gi...@apache.org>.
yanghao605 commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2238798928
##########
common/common-rest/src/main/java/org/apache/servicecomb/common/rest/filter/inner/ServerRestArgsFilter.java:
##########
@@ -91,4 +105,75 @@ 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.setChunked(true);
Review Comment:
这里如果是Rest Over Servlet模式下，走tomcat管理链接时，返回的响应头中会存在两个chunk的header，这里如果go的客户端直接调用的话会抛出异常
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "chengyouling (via GitHub)" <gi...@apache.org>.
chengyouling commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2255766252
##########
foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/HttpServletResponseEx.java:
##########
@@ -48,4 +48,8 @@ default Flowable<Buffer> getFlowableBuffer() {
}
void endResponse() throws IOException;
+
+ default void setChunkedForEvent(boolean chunked) {
+ // not set header transfer-encoding=chunked in Rest Over Servlet, or will have Multiple in response.
+ }
Review Comment:
add checked in setChunked method still have multiple headers of HttpServletResponseEx, code like this:
default void setChunked(boolean chunked) {
if (getHeader(HttpHeaders.TRANSFER_ENCODING.toString()) != null) {
return;
}
setHeader(HttpHeaders.TRANSFER_ENCODING.toString(), HttpHeaders.CHUNKED.toString());
}
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "yhs0092 (via GitHub)" <gi...@apache.org>.
yhs0092 commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2276839863
##########
swagger/swagger-invocation/invocation-core/src/main/java/org/apache/servicecomb/swagger/invocation/sse/SseEventResponseEntity.java:
##########
@@ -0,0 +1,97 @@
+/*
+ * 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.sse;
+
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import jakarta.validation.constraints.NotNull;
+
+public class SseEventResponseEntity<T> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(SseEventResponseEntity.class);
+
+ /**
+ * event id
+ */
+ private Integer id;
+
+ /**
+ * event type
+ */
+ private String event;
+
+ /**
+ * reconnection time
+ */
+ private Long retry;
+
+ /**
+ * business data
+ */
+ @NotNull
+ private T data;
+
+ public SseEventResponseEntity<T> id(int id) {
+ if (this.id != null) {
+ LOGGER.warn("origin id: [{}] is exists, overridden by the current value: [{}]", this.id, id);
+ }
+ this.id = id;
+ return this;
+ }
+
+ public SseEventResponseEntity<T> event(String event) {
+ if (!StringUtils.isEmpty(this.event)) {
+ LOGGER.warn("origin event: [{}] is exists, overridden by the current value: [{}]", this.event, event);
+ }
+ this.event = event;
+ return this;
+ }
+
+ public SseEventResponseEntity<T> retry(long retry) {
+ if (this.retry != null) {
+ LOGGER.warn("origin retry: [{}] is exists, overridden by the current value: [{}]", this.retry, retry);
+ }
+ this.retry = retry;
+ return this;
+ }
+
+ public SseEventResponseEntity<T> data(Object data) {
+ if (this.data != null) {
+ LOGGER.warn("origin data: [{}] is exists, overridden by the current value: [{}]", this.data, data);
+ }
+ this.data = (T) data;
Review Comment:
同一条消息里, 可以有多行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文档给出来的样例(注意红框部分):
<img width="672" height="546" alt="image" src="https://github.com/user-attachments/assets/204ba950-23d0-4e10-b011-92964170675e" />
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "yhs0092 (via GitHub)" <gi...@apache.org>.
yhs0092 commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2264610884
##########
common/common-rest/src/main/java/org/apache/servicecomb/common/rest/codec/produce/ProduceEventStreamProcessor.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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 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 {
+ private int writeIndex = 0;
+
+ @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());
+ } else {
+ appendEventId(bufferBuilder, writeIndex++);
+ appendData(bufferBuilder, result);
+ }
+ 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);
+ SseEventResponseEntity<?> responseEntity = new SseEventResponseEntity<>();
+ for (String line : buffer.split("\n")) {
Review Comment:
这里可能要识别一下`\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).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "yhs0092 (via GitHub)" <gi...@apache.org>.
yhs0092 commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2264615760
##########
common/common-rest/src/main/java/org/apache/servicecomb/common/rest/codec/produce/ProduceProcessor.java:
##########
@@ -76,4 +76,7 @@ default Object decodeResponse(Buffer buffer, JavaType type) throws Exception {
return doDecodeResponse(input, type);
}
}
+
+ default void refreshEventId(int index) {
Review Comment:
`refreshEventId` 是 SSE 场景特有的方法, 但我们把它加在 `ProduceProcessor` 里面去了, 这通常并不合适.
建议考虑把它放在 `ProduceEventStreamProcessor` 中.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "chengyouling (via GitHub)" <gi...@apache.org>.
chengyouling commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2278589839
##########
swagger/swagger-invocation/invocation-core/src/main/java/org/apache/servicecomb/swagger/invocation/sse/SseEventResponseEntity.java:
##########
@@ -0,0 +1,97 @@
+/*
+ * 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.sse;
+
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import jakarta.validation.constraints.NotNull;
+
+public class SseEventResponseEntity<T> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(SseEventResponseEntity.class);
+
+ /**
+ * event id
+ */
+ private Integer id;
+
+ /**
+ * event type
+ */
+ private String event;
+
+ /**
+ * reconnection time
+ */
+ private Long retry;
+
+ /**
+ * business data
+ */
+ @NotNull
+ private T data;
+
+ public SseEventResponseEntity<T> id(int id) {
+ if (this.id != null) {
+ LOGGER.warn("origin id: [{}] is exists, overridden by the current value: [{}]", this.id, id);
+ }
+ this.id = id;
+ return this;
+ }
+
+ public SseEventResponseEntity<T> event(String event) {
+ if (!StringUtils.isEmpty(this.event)) {
+ LOGGER.warn("origin event: [{}] is exists, overridden by the current value: [{}]", this.event, event);
+ }
+ this.event = event;
+ return this;
+ }
+
+ public SseEventResponseEntity<T> retry(long retry) {
+ if (this.retry != null) {
+ LOGGER.warn("origin retry: [{}] is exists, overridden by the current value: [{}]", this.retry, retry);
+ }
+ this.retry = retry;
+ return this;
+ }
+
+ public SseEventResponseEntity<T> data(Object data) {
+ if (this.data != null) {
+ LOGGER.warn("origin data: [{}] is exists, overridden by the current value: [{}]", this.data, data);
+ }
+ this.data = (T) data;
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "liubao68 (via GitHub)" <gi...@apache.org>.
liubao68 commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2254078651
##########
foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/http/HttpServletResponseEx.java:
##########
@@ -48,4 +48,8 @@ default Flowable<Buffer> getFlowableBuffer() {
}
void endResponse() throws IOException;
+
+ default void setChunkedForEvent(boolean chunked) {
+ // not set header transfer-encoding=chunked in Rest Over Servlet, or will have Multiple in response.
+ }
Review Comment:
Maybe change `setChunked` implementation to first check if header exists and then add it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "chengyouling (via GitHub)" <gi...@apache.org>.
chengyouling commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2244622335
##########
common/common-rest/src/main/java/org/apache/servicecomb/common/rest/codec/produce/ProduceEventStreamProcessor.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.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
+ 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());
+ } else {
+ appendEventId(bufferBuilder, writeIndex++);
+ appendData(bufferBuilder, result);
+ }
+ 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);
+ LOGGER.info("=========doDecodeResponse buffer===================>" + buffer + "stack: {}", Arrays.toString(
Review Comment:
just for test
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "chengyouling (via GitHub)" <gi...@apache.org>.
chengyouling commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2244632080
##########
common/common-rest/src/main/java/org/apache/servicecomb/common/rest/codec/produce/ProduceEventStreamProcessor.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.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
+ 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());
+ } else {
+ appendEventId(bufferBuilder, writeIndex++);
+ appendData(bufferBuilder, result);
+ }
+ 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);
+ LOGGER.info("=========doDecodeResponse buffer===================>" + buffer + "stack: {}", Arrays.toString(
+ new Exception().getStackTrace()));
+ SseEventResponseEntity<?> responseEntity = new SseEventResponseEntity<>();
+ boolean isResponseEntity = false;
+ for (String line : buffer.split("\n")) {
Review Comment:
sse格式上暂时用\n是用来区分事件的标识符
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "yhs0092 (via GitHub)" <gi...@apache.org>.
yhs0092 commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2264642082
##########
transports/transport-rest/transport-rest-client/src/main/java/org/apache/servicecomb/transport/rest/client/http/RestClientInvocation.java:
##########
@@ -218,6 +220,11 @@ protected void handleResponse(HttpClientResponse httpClientResponse) {
return;
}
+ if (restOperationMeta != null && restOperationMeta.isServerSendEvents()) {
+ processFlowableResponseBody(FlowableHelper.toFlowable(httpClientResponse));
Review Comment:
这里可能需要再确认一下, 如果微服务客户端调用服务端的SSE接口方法, 但是服务端报错了, 比如返回 403 状态码的 `application/json` 响应, 那么这里是否就不应该以 SSE 的形式去处理 response body 了呢?
换句话说, 这里的 if 判断条件应该以服务端返回的实际 `Content-Type` 为准会更合适一些?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "chengyouling (via GitHub)" <gi...@apache.org>.
chengyouling commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2275035001
##########
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));
+ }
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "yhs0092 (via GitHub)" <gi...@apache.org>.
yhs0092 commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2276859256
##########
swagger/swagger-invocation/invocation-core/src/main/java/org/apache/servicecomb/swagger/invocation/sse/SseEventResponseEntity.java:
##########
@@ -0,0 +1,97 @@
+/*
+ * 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.sse;
+
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import jakarta.validation.constraints.NotNull;
+
+public class SseEventResponseEntity<T> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(SseEventResponseEntity.class);
+
+ /**
+ * event id
+ */
+ private Integer id;
+
+ /**
+ * event type
+ */
+ private String event;
+
+ /**
+ * reconnection time
+ */
+ private Long retry;
+
+ /**
+ * business data
+ */
+ @NotNull
+ private T data;
+
+ public SseEventResponseEntity<T> id(int id) {
+ if (this.id != null) {
+ LOGGER.warn("origin id: [{}] is exists, overridden by the current value: [{}]", this.id, id);
+ }
+ this.id = id;
+ return this;
+ }
+
+ public SseEventResponseEntity<T> event(String event) {
+ if (!StringUtils.isEmpty(this.event)) {
+ LOGGER.warn("origin event: [{}] is exists, overridden by the current value: [{}]", this.event, event);
+ }
+ this.event = event;
+ return this;
+ }
+
+ public SseEventResponseEntity<T> retry(long retry) {
+ if (this.retry != null) {
+ LOGGER.warn("origin retry: [{}] is exists, overridden by the current value: [{}]", this.retry, retry);
+ }
+ this.retry = retry;
+ return this;
+ }
+
+ public SseEventResponseEntity<T> data(Object data) {
+ if (this.data != null) {
+ LOGGER.warn("origin data: [{}] is exists, overridden by the current value: [{}]", this.data, data);
+ }
+ this.data = (T) data;
Review Comment:
这里建议参考一下 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` 字段, 就表示这条消息有多行.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "chengyouling (via GitHub)" <gi...@apache.org>.
chengyouling commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2276154897
##########
common/common-rest/src/main/java/org/apache/servicecomb/common/rest/codec/produce/ProduceEventStreamProcessor.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * 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 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);
+
+ 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));
+ } else {
+ LOGGER.warn("Does not support encoding objects other than SseEventResponseEntity!");
+ }
+ }
+
+ @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");
Review Comment:
fixed
##########
swagger/swagger-invocation/invocation-core/src/main/java/org/apache/servicecomb/swagger/invocation/response/producer/PublisherProducerResponseMapper.java:
##########
@@ -0,0 +1,45 @@
+/*
+ * 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.producer;
+
+import org.apache.servicecomb.swagger.invocation.Response;
+import org.apache.servicecomb.swagger.invocation.sse.SseEventResponseEntity;
+
+import io.reactivex.rxjava3.core.Flowable;
+import jakarta.ws.rs.core.Response.StatusType;
+
+public class PublisherProducerResponseMapper implements ProducerResponseMapper {
+ private final boolean shouldConstructEntity;
+
+ private int writeIndex = 0;
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "liubao68 (via GitHub)" <gi...@apache.org>.
liubao68 commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2224958746
##########
demo/demo-cse-v2/consumer/src/main/java/org/apache/servicecomb/samples/ConsumerReactiveStreamController.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.samples;
+
+
Review Comment:
Test cases here will not run as integration tests. Can you add tests in some other module that run as integration tests?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "chengyouling (via GitHub)" <gi...@apache.org>.
chengyouling merged PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "chengyouling (via GitHub)" <gi...@apache.org>.
chengyouling commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2269073627
##########
common/common-rest/src/main/java/org/apache/servicecomb/common/rest/codec/produce/ProduceProcessor.java:
##########
@@ -76,4 +76,7 @@ default Object decodeResponse(Buffer buffer, JavaType type) throws Exception {
return doDecodeResponse(input, type);
}
}
+
+ default void refreshEventId(int index) {
Review Comment:
fixed
##########
common/common-rest/src/main/java/org/apache/servicecomb/common/rest/definition/RestOperationMeta.java:
##########
@@ -157,6 +167,10 @@ private boolean checkDownloadFileFlag() {
return false;
}
+ private boolean checkServerSendEvents() {
+ return !CollectionUtils.isEmpty(produces) && produces.contains(EVENTS_MEDIA_TYPE);
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "chengyouling (via GitHub)" <gi...@apache.org>.
chengyouling commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2227092662
##########
demo/demo-cse-v2/consumer/src/main/java/org/apache/servicecomb/samples/ConsumerReactiveStreamController.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.samples;
+
+
Review Comment:
yes, will add in spring-mvc demo
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [#4873] Support for SSE interface RPC calls [servicecomb-java-chassis]
Posted by "yhs0092 (via GitHub)" <gi...@apache.org>.
yhs0092 commented on code in PR #4875:
URL: https://github.com/apache/servicecomb-java-chassis/pull/4875#discussion_r2264628894
##########
common/common-rest/src/main/java/org/apache/servicecomb/common/rest/definition/RestOperationMeta.java:
##########
@@ -157,6 +167,10 @@ private boolean checkDownloadFileFlag() {
return false;
}
+ private boolean checkServerSendEvents() {
+ return !CollectionUtils.isEmpty(produces) && produces.contains(EVENTS_MEDIA_TYPE);
Review Comment:
MIME type 取值为 `text/event-stream;charset=utf-8` 也是合法的. 同样是这个类里的 `doCreateProduceProcessors` 方法在处理 `produces` 属性值的时候就考虑到了 MIME type 带分号的情况. 这里建议也优化一下实现细节.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@servicecomb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org