fix triple POJO stream parent interface method not cache stream call type issue (#1481)

Co-authored-by: liujianjun.ljj <[email protected]>

Author: EvenLjj

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/utils/SofaProtoUtils.java

@@ -96,7 +96,7 @@ public static MethodDescriptor.MethodType mapGrpcCallType(String callType) {
 
     public static Map<String, String> cacheStreamCallType(Class proxyClass) {
         Map<String, String> methodCallType = new ConcurrentHashMap<>();
-        Method[] declaredMethods = proxyClass.getDeclaredMethods();
+        Method[] declaredMethods = proxyClass.getMethods();
         for (Method method : declaredMethods) {
             String streamType = mapStreamType(method);
             if (StringUtils.isNotBlank(streamType)) {

test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/HelloService.java

@@ -18,7 +18,7 @@
 
 import com.alipay.sofa.rpc.transport.SofaStreamObserver;
 
-public interface HelloService {
+public interface HelloService extends ParentService {
 
     String CMD_TRIGGER_STREAM_FINISH = "finish";
 

test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/HelloServiceImpl.java

@@ -78,4 +78,15 @@ public void sayHelloServerStream(ClientRequest clientRequest, SofaStreamObserver
         }
     }
 
+    @Override
+    public ServerResponse sayHello(ClientRequest clientRequest) {
+        return new ServerResponse(clientRequest.getMsg(), clientRequest.getCount());
+    }
+
+    @Override
+    public void parentSayHelloServerStream(ClientRequest clientRequest,
+                                           SofaStreamObserver<ServerResponse> sofaStreamObserver) {
+        sayHelloServerStream(clientRequest, sofaStreamObserver);
+    }
+
 }

test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ParentService.java

@@ -0,0 +1,31 @@
+/*
+ * 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 com.alipay.sofa.rpc.test.triple.stream;
+
+import com.alipay.sofa.rpc.transport.SofaStreamObserver;
+
+/**
+ * @author Even
+ * @date 2025/3/4 23:49
+ */
+public interface ParentService {
+
+    ServerResponse sayHello(ClientRequest clientRequest);
+
+    void parentSayHelloServerStream(ClientRequest clientRequest, SofaStreamObserver<ServerResponse> sofaStreamObserver);
+
+}

test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/TripleGenericStreamTest.java

@@ -73,6 +73,7 @@ public static void beforeClass() throws InterruptedException {
             .setProtocol("tri")
             .setDirectUrl("triple://127.0.0.1:50066");
         helloServiceRef = consumerConfig.refer();
+        Thread.sleep(5000);
     }
 
     @AfterClass
@@ -84,6 +85,39 @@ public static void afterClass() {
         RpcInvokeContext.removeContext();
     }
 
+    @Test
+    public void testTripleParentCall() throws InterruptedException {
+        ClientRequest clientRequest = new ClientRequest("hello world", 5);
+        ServerResponse serverResponse = helloServiceRef.sayHello(clientRequest);
+        Assert.assertEquals("hello world", serverResponse.getMsg());
+
+        CountDownLatch countDownLatch = new CountDownLatch(6);
+        AtomicBoolean receivedFinish = new AtomicBoolean(false);
+        List<ServerResponse> list = new ArrayList<>();
+        helloServiceRef.parentSayHelloServerStream(clientRequest, new SofaStreamObserver<ServerResponse>() {
+            @Override
+            public void onNext(ServerResponse message) {
+                list.add(message);
+                countDownLatch.countDown();
+            }
+
+            @Override
+            public void onCompleted() {
+                receivedFinish.set(true);
+                countDownLatch.countDown();
+            }
+
+            @Override
+            public void onError(Throwable throwable) {
+                countDownLatch.countDown();
+            }
+        });
+
+        Assert.assertTrue(countDownLatch.await(20, TimeUnit.SECONDS));
+        Assert.assertEquals(5, list.size());
+        Assert.assertTrue(receivedFinish.get());
+    }
+
     @Test
     public void testTripleBiStreamFinish() throws InterruptedException {
         testTripleBiStream(false);