feat: server stream supports requests without parameters

dubbo-common/src/main/java/org/apache/dubbo/rpc/model/ReflectionMethodDescriptor.java

@@ -22,8 +22,11 @@
 import org.apache.dubbo.common.utils.ReflectUtils;
 
 import java.lang.reflect.Method;
+import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
@@ -48,6 +51,8 @@ public class ReflectionMethodDescriptor implements MethodDescriptor {
     private final Method method;
     private final boolean generic;
     private final RpcType rpcType;
+    private Class<?>[] actualRequestTypes;
+    private Class<?> actualResponseType;
 
     public ReflectionMethodDescriptor(Method method) {
         this.method = method;
@@ -82,13 +87,17 @@ private RpcType determineRpcType() {
         if (parameterClasses.length > 2) {
             return RpcType.UNARY;
         }
+        Type[] genericParameterTypes = method.getGenericParameterTypes();
         if (parameterClasses.length == 1 && isStreamType(parameterClasses[0]) && isStreamType(returnClass)) {
+            this.actualRequestTypes = new Class<?>[] {
+                obtainActualTypeInStreamObserver(
+                        ((ParameterizedType) method.getGenericReturnType()).getActualTypeArguments()[0])
+            };
+            actualResponseType = obtainActualTypeInStreamObserver(
+                    ((ParameterizedType) genericParameterTypes[0]).getActualTypeArguments()[0]);
             return RpcType.BI_STREAM;
         }
-        if (parameterClasses.length == 2
-                && !isStreamType(parameterClasses[0])
-                && isStreamType(parameterClasses[1])
-                && returnClass.getName().equals(void.class.getName())) {
+        if (parameterClasses.length >= 1 && hasStreamType(parameterClasses)) {
             return RpcType.SERVER_STREAM;
         }
         if (Arrays.stream(parameterClasses).anyMatch(this::isStreamType) || isStreamType(returnClass)) {
@@ -99,10 +108,30 @@ && isStreamType(parameterClasses[1])
         return RpcType.UNARY;
     }
 
+    private boolean hasStreamType(Class<?>[] parameterClasses) {
+        List<Class<?>> parameterClassList = new ArrayList<>();
+        for (Class<?> parameterClass : parameterClasses) {
+            if (isStreamType(parameterClass)) {
+                actualResponseType = parameterClass;
+            } else {
+                parameterClassList.add(parameterClass);
+            }
+        }
+        actualRequestTypes = parameterClassList.toArray(new Class<?>[0]);
+        return actualResponseType != null;
+    }
+
     private boolean isStreamType(Class<?> classType) {
         return StreamObserver.class.isAssignableFrom(classType);
     }
 
+    private static Class<?> obtainActualTypeInStreamObserver(Type typeInStreamObserver) {
+        return (Class<?>)
+                (typeInStreamObserver instanceof ParameterizedType
+                        ? ((ParameterizedType) typeInStreamObserver).getRawType()
+                        : typeInStreamObserver);
+    }
+
     @Override
     public String getMethodName() {
         return methodName;
@@ -156,6 +185,14 @@ public Object getAttribute(String key) {
         return this.attributeMap.get(key);
     }
 
+    public Class<?>[] getActualRequestTypes() {
+        return actualRequestTypes;
+    }
+
+    public Class<?> getActualResponseType() {
+        return actualResponseType;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {

dubbo-demo/dubbo-demo-spring-boot/dubbo-demo-spring-boot-servlet/src/main/java/org/apache/dubbo/springboot/demo/servlet/GreeterService.java

@@ -37,6 +37,8 @@ public interface GreeterService {
      */
     void sayHelloServerStream(HelloRequest request, StreamObserver<HelloReply> responseObserver);
 
+    void sayHelloServerStreamNoParameter(StreamObserver<HelloReply> responseObserver);
+
     /**
      * Sends greetings with bi streaming
      */

dubbo-demo/dubbo-demo-spring-boot/dubbo-demo-spring-boot-servlet/src/main/java/org/apache/dubbo/springboot/demo/servlet/GreeterServiceImpl.java

@@ -52,6 +52,17 @@ public void sayHelloServerStream(HelloRequest request, StreamObserver<HelloReply
         responseObserver.onCompleted();
     }
 
+    @Override
+    public void sayHelloServerStreamNoParameter(StreamObserver<HelloReply> responseObserver) {
+        LOGGER.info("Received sayHelloServerStream request");
+        for (int i = 1; i < 6; i++) {
+            LOGGER.info("sayHelloServerStream onNext:  {} times", i);
+            responseObserver.onNext(toReply("Hello " + ' ' + i + " times"));
+        }
+        LOGGER.info("sayHelloServerStream onCompleted");
+        responseObserver.onCompleted();
+    }
+
     @Override
     public StreamObserver<HelloRequest> sayHelloBiStream(StreamObserver<HelloReply> responseObserver) {
         LOGGER.info("Received sayHelloBiStream request");

dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/MethodMetadata.java

@@ -20,9 +20,6 @@
 import org.apache.dubbo.rpc.model.ReflectionMethodDescriptor;
 import org.apache.dubbo.rpc.model.StubMethodDescriptor;
 
-import java.lang.reflect.ParameterizedType;
-import java.lang.reflect.Type;
-
 public class MethodMetadata {
 
     private final Class<?>[] actualRequestTypes;
@@ -64,19 +61,9 @@ private static MethodMetadata doResolveReflection(ReflectionMethodDescriptor met
         switch (method.getRpcType()) {
             case CLIENT_STREAM:
             case BI_STREAM:
-                actualRequestTypes = new Class<?>[] {
-                    obtainActualTypeInStreamObserver(
-                            ((ParameterizedType) method.getMethod().getGenericReturnType()).getActualTypeArguments()[0])
-                };
-                actualResponseType = obtainActualTypeInStreamObserver(
-                        ((ParameterizedType) method.getMethod().getGenericParameterTypes()[0])
-                                .getActualTypeArguments()[0]);
-                return new MethodMetadata(actualRequestTypes, actualResponseType);
             case SERVER_STREAM:
-                actualRequestTypes = new Class[] {method.getMethod().getParameterTypes()[0]};
-                actualResponseType = obtainActualTypeInStreamObserver(
-                        ((ParameterizedType) method.getMethod().getGenericParameterTypes()[1])
-                                .getActualTypeArguments()[0]);
+                actualRequestTypes = method.getActualRequestTypes();
+                actualResponseType = method.getActualResponseType();
                 return new MethodMetadata(actualRequestTypes, actualResponseType);
             case UNARY:
                 actualRequestTypes = method.getParameterClasses();
@@ -85,11 +72,4 @@ private static MethodMetadata doResolveReflection(ReflectionMethodDescriptor met
         }
         throw new IllegalStateException("Can not reach here");
     }
-
-    static Class<?> obtainActualTypeInStreamObserver(Type typeInStreamObserver) {
-        return (Class<?>)
-                (typeInStreamObserver instanceof ParameterizedType
-                        ? ((ParameterizedType) typeInStreamObserver).getRawType()
-                        : typeInStreamObserver);
-    }
 }

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

@@ -33,6 +33,13 @@ public void onReturn(Object value) {}
 
     @Override
     public void onMessage(Object message) {
+        Class<?>[] params = invocation.getParameterTypes();
+        if (params.length == 1) {
+            if (params[0].isInstance(responseObserver)) {
+                invocation.setArguments(new Object[] {responseObserver});
+                return;
+            }
+        }
         if (message instanceof Object[]) {
             message = ((Object[]) message)[0];
         }