Support returning JSON content when using SSE (#15464)

* Support returning JSON content when using SSE

* Update Http2SseServerChannelObserver.java

* Update Http2SseServerChannelObserver.java

---------

Co-authored-by: zrlw <zrlw@sina.com>

Author: oxsean

dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/AbstractServerHttpChannelObserver.java

@@ -166,13 +166,13 @@ protected final HttpMetadata buildMetadata(
         if (message != null) {
             headers.set(HttpHeaderNames.CONTENT_TYPE.getKey(), responseEncoder.contentType());
         }
+        customizeHeaders(headers, throwable, message);
         if (data instanceof HttpResult) {
             HttpResult<?> result = (HttpResult<?>) data;
             if (result.getHeaders() != null) {
                 headers.set(result.getHeaders());
             }
         }
-        customizeHeaders(headers, throwable, message);
         return metadata;
     }
 
@@ -198,7 +198,7 @@ protected final void sendMetadata(HttpMetadata metadata) {
         }
     }
 
-    protected final HttpOutputMessage buildMessage(int statusCode, Object data) throws Throwable {
+    protected HttpOutputMessage buildMessage(int statusCode, Object data) throws Throwable {
         if (statusCode < 200 || statusCode == 204 || statusCode == 304) {
             return null;
         }

dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/ServerStreamServerCallListener.java

@@ -17,6 +17,7 @@
 package org.apache.dubbo.rpc.protocol.tri.h12;
 
 import org.apache.dubbo.common.stream.StreamObserver;
+import org.apache.dubbo.remoting.http12.HttpResult;
 import org.apache.dubbo.remoting.http12.exception.HttpStatusException;
 import org.apache.dubbo.rpc.Invoker;
 import org.apache.dubbo.rpc.RpcInvocation;
@@ -29,7 +30,12 @@ public ServerStreamServerCallListener(
     }
 
     @Override
-    public void onReturn(Object value) {}
+    public void onReturn(Object value) {
+        if (value instanceof HttpResult) {
+            responseObserver.onNext(value);
+            responseObserver.onCompleted();
+        }
+    }
 
     @Override
     public void onMessage(Object message) {

dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http1/Http1SseServerChannelObserver.java

@@ -20,19 +20,24 @@
 import org.apache.dubbo.remoting.http12.HttpConstants;
 import org.apache.dubbo.remoting.http12.HttpHeaderNames;
 import org.apache.dubbo.remoting.http12.HttpMetadata;
+import org.apache.dubbo.remoting.http12.HttpOutputMessage;
+import org.apache.dubbo.remoting.http12.HttpResult;
 import org.apache.dubbo.remoting.http12.h1.Http1ServerChannelObserver;
 import org.apache.dubbo.remoting.http12.message.HttpMessageEncoder;
 import org.apache.dubbo.remoting.http12.message.ServerSentEventEncoder;
 
 public class Http1SseServerChannelObserver extends Http1ServerChannelObserver {
 
+    private HttpMessageEncoder originalResponseEncoder;
+
     public Http1SseServerChannelObserver(HttpChannel httpChannel) {
         super(httpChannel);
     }
 
     @Override
     public void setResponseEncoder(HttpMessageEncoder responseEncoder) {
         super.setResponseEncoder(new ServerSentEventEncoder(responseEncoder));
+        this.originalResponseEncoder = responseEncoder;
     }
 
     @Override
@@ -49,4 +54,25 @@ protected HttpMetadata encodeHttpMetadata(boolean endStream) {
                 .header(HttpHeaderNames.TRANSFER_ENCODING.getKey(), HttpConstants.CHUNKED)
                 .header(HttpHeaderNames.CACHE_CONTROL.getKey(), HttpConstants.NO_CACHE);
     }
+
+    @Override
+    protected HttpOutputMessage buildMessage(int statusCode, Object data) throws Throwable {
+        if (data instanceof HttpResult) {
+            data = ((HttpResult<?>) data).getBody();
+
+            if (data == null && statusCode != 200) {
+                return null;
+            }
+
+            HttpOutputMessage message = encodeHttpOutputMessage(data);
+            try {
+                originalResponseEncoder.encode(message.getBody(), data);
+            } catch (Throwable t) {
+                message.close();
+                throw t;
+            }
+            return message;
+        }
+        return super.buildMessage(statusCode, data);
+    }
 }

dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/Http2SseServerChannelObserver.java

@@ -19,20 +19,25 @@
 import org.apache.dubbo.remoting.http12.HttpConstants;
 import org.apache.dubbo.remoting.http12.HttpHeaderNames;
 import org.apache.dubbo.remoting.http12.HttpMetadata;
+import org.apache.dubbo.remoting.http12.HttpOutputMessage;
+import org.apache.dubbo.remoting.http12.HttpResult;
 import org.apache.dubbo.remoting.http12.h2.H2StreamChannel;
 import org.apache.dubbo.remoting.http12.message.HttpMessageEncoder;
 import org.apache.dubbo.remoting.http12.message.ServerSentEventEncoder;
 import org.apache.dubbo.rpc.model.FrameworkModel;
 
 public final class Http2SseServerChannelObserver extends Http2StreamServerChannelObserver {
 
+    private HttpMessageEncoder originalResponseEncoder;
+
     public Http2SseServerChannelObserver(FrameworkModel frameworkModel, H2StreamChannel h2StreamChannel) {
         super(frameworkModel, h2StreamChannel);
     }
 
     @Override
     public void setResponseEncoder(HttpMessageEncoder responseEncoder) {
         super.setResponseEncoder(new ServerSentEventEncoder(responseEncoder));
+        this.originalResponseEncoder = responseEncoder;
     }
 
     @Override
@@ -41,6 +46,27 @@ protected HttpMetadata encodeHttpMetadata(boolean endStream) {
                 .header(HttpHeaderNames.CACHE_CONTROL.getKey(), HttpConstants.NO_CACHE);
     }
 
+    @Override
+    protected HttpOutputMessage buildMessage(int statusCode, Object data) throws Throwable {
+        if (data instanceof HttpResult) {
+            data = ((HttpResult<?>) data).getBody();
+
+            if (data == null && statusCode != 200) {
+                return null;
+            }
+
+            HttpOutputMessage message = encodeHttpOutputMessage(data);
+            try {
+                originalResponseEncoder.encode(message.getBody(), data);
+            } catch (Throwable t) {
+                message.close();
+                throw t;
+            }
+            return message;
+        }
+        return super.buildMessage(statusCode, data);
+    }
+
     @Override
     protected void doOnCompleted(Throwable throwable) {
         // if throwable is not null, the header will be flushed by super.doOnCompleted(throwable)