fix: ignore params type of StreamObserver (#15450)

* fix: ignore params type of StreamObserver

* code format

* rm test file

* code format

* fix: add the match logic of proto service with params type of StreamObserver

* fix: support server stream only

* fix: support server stream only

* code format

Author: fantiq

dubbo-demo/dubbo-demo-spring-boot-idl/dubbo-demo-spring-boot-idl-provider/README.md

@@ -0,0 +1,24 @@
+## 1. startup
+
+1. run the maven plugin `dubbo:compile`, generate protobuf java file.
+2. run `ProviderApplication`
+
+## 2. http request
+
+### 2.1 sample request
+
+```shell
+curl -v -d '{"name":"dubbo"}' -H 'Content-Type: application/json' http://127.0.0.1:50051/org.apache.dubbo.demo.hello.GreeterService/sayHello
+```
+
+### 2.2 request async
+
+```shell
+curl -v -d '{"name":"dubbo async"}' -H 'Content-Type: application/json' http://127.0.0.1:50051/org.apache.dubbo.demo.hello.GreeterService/sayHelloAsync
+```
+
+### 2.3 server stream
+
+```shell
+curl -v -d '{"name":"dubbo"}' -H 'Content-Type: application/json' http://127.0.0.1:50051/org.apache.dubbo.demo.hello.GreeterService/sayHelloStream
+```

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

@@ -16,12 +16,14 @@
  */
 package org.apache.dubbo.springboot.idl.demo.provider;
 
+import org.apache.dubbo.common.stream.StreamObserver;
 import org.apache.dubbo.config.annotation.DubboService;
 import org.apache.dubbo.demo.hello.GreeterService;
 import org.apache.dubbo.demo.hello.HelloReply;
 import org.apache.dubbo.demo.hello.HelloRequest;
 
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -45,6 +47,22 @@ public CompletableFuture<HelloReply> sayHelloAsync(HelloRequest request) {
                 HelloReply.newBuilder().setMessage("Hello " + request.getName()).build());
     }
 
+    @Override
+    public void sayHelloStream(HelloRequest request, StreamObserver<HelloReply> responseObserver) {
+        LOGGER.info("Received sayHelloStream request: {}", request.getName());
+        for (int i = 0; i < 5; i++) {
+            try {
+                TimeUnit.SECONDS.sleep(1);
+            } catch (InterruptedException e) {
+                responseObserver.onError(e);
+            }
+            responseObserver.onNext(HelloReply.newBuilder()
+                    .setMessage(i + "# Hello " + request.getName())
+                    .build());
+        }
+        responseObserver.onCompleted();
+    }
+
     private static HelloReply toReply(String message) {
         return HelloReply.newBuilder().setMessage(message).build();
     }

dubbo-demo/dubbo-demo-spring-boot-idl/dubbo-demo-spring-boot-idl-provider/src/main/proto/helloworld.proto

@@ -32,4 +32,5 @@ message HelloReply {
 service GreeterService {
     // Sends a greeting.
     rpc sayHello(HelloRequest) returns (HelloReply);
+    rpc sayHelloStream(HelloRequest) returns (stream HelloReply);
 }

dubbo-demo/dubbo-demo-spring-boot-idl/pom.xml

@@ -77,6 +77,10 @@
       <groupId>com.google.protobuf</groupId>
       <artifactId>protobuf-java</artifactId>
     </dependency>
+    <dependency>
+      <groupId>com.google.protobuf</groupId>
+      <artifactId>protobuf-java-util</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.apache.logging.log4j</groupId>
       <artifactId>log4j-slf4j-impl</artifactId>

dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/rest/mapping/DefaultRequestMappingRegistry.java

@@ -18,6 +18,7 @@
 
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.logger.FluentLogger;
+import org.apache.dubbo.common.stream.StreamObserver;
 import org.apache.dubbo.common.utils.ClassUtils;
 import org.apache.dubbo.config.context.ConfigManager;
 import org.apache.dubbo.config.nested.RestConfig;
@@ -31,6 +32,7 @@
 import org.apache.dubbo.rpc.model.ReflectionMethodDescriptor;
 import org.apache.dubbo.rpc.model.ReflectionServiceDescriptor;
 import org.apache.dubbo.rpc.model.ServiceDescriptor;
+import org.apache.dubbo.rpc.model.StubServiceDescriptor;
 import org.apache.dubbo.rpc.protocol.tri.DescriptorUtils;
 import org.apache.dubbo.rpc.protocol.tri.TripleProtocol;
 import org.apache.dubbo.rpc.protocol.tri.rest.Messages;
@@ -48,6 +50,7 @@
 
 import java.lang.reflect.Method;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.IdentityHashMap;
 import java.util.LinkedList;
@@ -129,7 +132,14 @@ public void register(Invoker<?> invoker) {
                 RequestMapping classMapping = resolver.resolve(serviceMeta);
                 consumer.accept((methods) -> {
                     Method method = methods.get(0);
-                    MethodDescriptor md = sd.getMethod(method.getName(), method.getParameterTypes());
+                    Class<?>[] paramTypes = method.getParameterTypes();
+                    MethodDescriptor md = sd.getMethod(method.getName(), paramTypes);
+                    if (md == null && sd instanceof StubServiceDescriptor) {
+                        int len = paramTypes.length;
+                        if (len > 0 && StreamObserver.class.isAssignableFrom(paramTypes[len - 1])) {
+                            md = sd.getMethod(method.getName(), Arrays.copyOf(paramTypes, len - 1));
+                        }
+                    }
                     MethodMeta methodMeta = new MethodMeta(methods, md, serviceMeta);
                     if (!resolver.accept(methodMeta)) {
                         return;