Triple stream support (issue#1332) #1360

namelessssssssssss · 2023-08-25T12:48:03Z

Motivation:

This PR can provides stream transport support for triple protocol, including server streaming and bidirectional/client streaming.

Modification:

Add interface StreamHandler and implements
Provide and update related test.

Change log (Chinese) :

概述

Stream 方式是一种异步的流式处理方式，可以在数据传输过程中逐个处理数据，避免一次性传输大量数据造成的性能问题。服务端Stream 是指服务端在处理请求时，将数据分成多个部分逐个返回给客户端的过程；客户端 Stream 是指客户端在请求服务器时，将请求参数分成多个部分逐个发送给服务器的过程。Stream 方式可以让我们在处理大量数据时更高效地使用资源，提高系统的性能和响应速度。

业务场景

流式传输适配下列业务场景：

接口希望发送大量数据，如图像、视频二进制等数据。在一次RPC请求或响应中传输大量数据可能导致性能或实现上的问题。而如果按照传统方式，多次发起普通RPC请求，则难以解决消息顺序和性能问题（想要保证有序则需要串行发送请求）
数据流场景，数据本身没有确定边界，需要按照其发送顺序进行处理
推送场景，希望在单次调用的上下文中处理多个消息

实现的流传输应具有以下语义保证：

提供消息边界，可以方便地对消息单独处理
严格有序，发送端的顺序和接收端顺序一致
全双工，发送不需要等待
支持取消和超时

流传输模型

对于流式传输，我们可以将Stream分为四种类型（借鉴于gRPC）：

Unary，传统请求-响应模型：客户端一次请求，服务端一次响应

ClientStream，客户端流：客户端多次请求，服务端最终返回一次响应，服务端接收请求期间可以返回已完成/取消信号来停止客户端继续发送请求，或者在客户端请求完成后返回响应。

ServerStream，服务端流：客户端单次请求，服务端返回多次响应，客户端接收请求期间可以返回已完成/取消信号来停止接受服务端的响应，或者等待服务端发送完成相应信号。

BidirectionalStream，双向流：客户端发送一次请求后，客户端-服务端均可相互多次乱序发送请求，直到任意一方发送已完成/取消信号。

流传输接口定义

我们先从RPC调用的最上层开始分析。首先，我们需要定义客户端/服务端用于接收/发送流式请求的接口，该接口定义了如何发送、接收、处理数据流，以及数据流的异常、结束操作。

考虑到目前传输层的实现，此处可以考虑直接使用 gRPC 中的 StreamObserver，这样对于 Triple 协议的传输层实现来说比较方便（TripleClientInvoker 直接依赖 gRPC 进行传输），但缺点是可能导致其它未来可能支持流传输的协议同时耦合了gRPC的API。

更好的方案是使用自定义的 StreamObserver（此处先叫做StreamHandler），将其和gRPC解耦，在 Triple 传输层通过适配器转换为gRPC StreamObserver。此处使用自定义的StreamHandler。

我们为该接口定义以下方法，实际和gRPC中的StreamObserver一致：

//泛型为消息类型
public interface StreamHandler<T> {
   /**
    * 接收者通过实现该方法,定义单条信息的处理逻辑
    * 发送者则通过调用该方法发送一条信息
    */
    void onMessage(T message);
    
    /**
    * 接收者通过该方法定义信息发送完成后的处理逻辑
    * 发送者通过调用该方法告知接收者所有信息发送完毕
    */
    void onFinish();
    
    /**
    * 接收者通过该方法定义处理某条消息发生异常时的处理逻辑 (onMessage方法抛出异常时)
    * 发送者不应调用该方法
    */
    void onException(Throwable throwable);
    
}

全局处理

**1，标记请求类型 **

首先是识别方法调用类型，并为 SofaRequest 打上流传输的标记。

对于客户端来说，SofaRequest在客户端代理中创建。其中标记传输类型的字段在DefaultClientProxyInvoker中的 decorateRequest方法中设置（对于泛型调用）。

//DefaultClientProxyInvoker 
@Override
    protected void decorateRequest(SofaRequest request) {
        // 公共的设置
        super.decorateRequest(request);

        // 缓存是为了加快速度
        request.setTargetServiceUniqueName(serviceName);
        request.setSerializeType(serializeType == null ? 0 : serializeType);

        if (!consumerConfig.isGeneric()) {
            // 找到调用类型， generic的时候类型在filter里进行判断
             request.setInvokeType(consumerConfig.getMethodInvokeType(request));
        }
        ......
    }

setInvokeType方法最终调用ConsumerConfig中的getMethodInvokeType，尝试从接口配置中获取已缓存的调用方式，否则使用默认值（一元调用）。可以将判断调用类型的操作添加在其中的 getMethodInvokeType 中，这样也可以将方法调用模式缓存，防止每次调用都要进行繁琐的判断操作。

修改后的方法：

  //ConsumerConfig
  public String getMethodInvokeType(SofaRequest sofaRequest) {
        String methodName = sofaRequest.getMethodName();
        String invokeType = (String) getMethodConfigValue(methodName, RpcConstants.CONFIG_KEY_INVOKE_TYPE,null);

        if(invokeType == null) {
            invokeType = getAndCacheCallType(sofaRequest);
        }

        return invokeType;
    }

     /**
     * Get and cache the call type of certain method
     * @param request RPC request
     * @return request call type
     */
    public String getAndCacheCallType(SofaRequest request){
         Method method  = request.getMethod();
         String callType = MethodConfig.mapStreamType(
                 method,
                 (String) getMethodConfigValue(request.getMethodName(), RpcConstants.CONFIG_KEY_INVOKE_TYPE, getInvokeType())
         );
         //Method level config
         updateAttribute(buildMethodConfigKey(request,RpcConstants.CONFIG_KEY_INVOKE_TYPE),callType,true);
         return callType;
    }

//MethodConfig
/**
     * Gets the stream call type of certain method
     * @param method the method
     * @return call type,server/client/bidirectional stream or default value. If not mapped to any stream call type, use the default value
     */
    public static String mapStreamType(Method method, String defaultValue){
        Class<?>[] paramClasses = method.getParameterTypes();
        Class<?> returnClass = method.getReturnType();

        int paramLen = paramClasses.length;
        String callType;

        //BidirectionalStream
        if(paramLen>0 && paramClasses[0].isAssignableFrom(StreamHandler.class) && returnClass.isAssignableFrom(StreamHandler.class)){

            if(paramLen > 1){
                throw new SofaRpcException(RpcErrorType.CLIENT_CALL_TYPE,"Bidirectional stream method parameters can be only one StreamHandler.");
            }

            callType = RpcConstants.INVOKER_TYPE_BI_STREAMING;
        }
        //ServerStream
        else if ( paramLen>1 && paramClasses[1].isAssignableFrom(StreamHandler.class) && returnClass.isAssignableFrom(StreamHandler.class)){
            callType = RpcConstants.INVOKER_TYPE_SERVER_STREAMING;
        }
        //ClientStream
        else if (returnClass.isAssignableFrom(StreamHandler.class) && Arrays.stream(paramClasses).noneMatch(clazz -> clazz.isAssignableFrom(StreamHandler.class))) {
            callType = RpcConstants.INVOKER_TYPE_CLIENT_STREAMING;
        }
        else if (returnClass.isAssignableFrom(StreamHandler.class) || Arrays.stream(paramClasses).anyMatch(clazz -> clazz.isAssignableFrom(StreamHandler.class))) {
            throw new SofaRpcException(RpcErrorType.CLIENT_CALL_TYPE, "StreamHandler can only appear at the specified location of parameter. Please check related docs.");
        }
        //Other call types
        else {
            callType = defaultValue;
        }

        return callType;
    }

主要完成以下三个操作：

如果方法参数中包含 StreamHandler，根据 StreamHandler 在方法参数中出现的情况，将SofaRequest 的 InvokeType 标记为 STREAM_CLIENT / STREAM_SERVER / STREAM_BI ,对应三种流传输方式。这三个常量可以添加到 RpcConstants 中。
同时对方法中 StreamHandler 参数位置进行校验，若位置不符合定义中的用法直接抛出异常，并日志提示用户不符的原因。以上方法的考虑较简单，之后可添加更多的提示信息。
缓存方法调用方式。

此处将调用方式缓存到了MethodConfig中，因为InterfaceConfig使用的是UnmodifiableMap，在其初始化完成后已不能再添加新的信息。由于 MethodConfig 本身没有保存 Method 引用，且不会默认创建，使得获取方法参数及返回值的具体类型有些困难，需要在实际请求时通过 SofaRequest 拿到具体的参数 Class，才可较简单的判断方法参数及返回值中StreamHandler的出现情况，判断请求类型。

2,修改AbstractCluster.doSendMsg

在该方法中通过获取SofaRequest的RequestType，判断是使用同步、异步、回调还是其它调用方法。对于流式调用，需要在其中添加新请求类型的处理逻辑。

 protected SofaResponse doSendMsg(ProviderInfo providerInfo, ClientTransport transport,sofaRequest request) throws SofaRpcException {
   ...
        try {
   ...
            // 同步调用
            if (RpcConstants.INVOKER_TYPE_SYNC.equals(invokeType)) {
    ...
            }
            // 单向调用
            else if (RpcConstants.INVOKER_TYPE_ONEWAY.equals(invokeType)) {
    ...
            }
            // Callback调用
            else if (RpcConstants.INVOKER_TYPE_CALLBACK.equals(invokeType)) {
    ...
            }
            // Future调用
            else if (RpcConstants.INVOKER_TYPE_FUTURE.equals(invokeType)) {
   ...
            }
            // 流式调用
            else if (RpcConstants.INVOKER_TYPE_STREAM_CLIENT.equals(invokeType)
                    || RpcConstants.INVOKER_TYPE_STREAM_BI.equals(invokeType)
                    || RpcConstants.INVOKER_TYPE_STREAM_SERVER.equals(invokeType)) { 
                response = transport.syncSend(request, Integer.MAX_VALUE);
            }
            else {
                throw new SofaRpcException(RpcErrorType.CLIENT_UNDECLARED_ERROR, "Unknown invoke type:" + invokeType);
            } 
            return response;
        } catch (SofaRpcException e) {
            throw e;
        } catch (Throwable e) { // 客户端其它异常
            throw new SofaRpcException(RpcErrorType.CLIENT_UNDECLARED_ERROR, e);
        }
    }

根据流式传输方法的定义，如果客户端希望通过Stream向服务端再次发送信息，它需要通过调用RPC方法返回的StreamHandler中的onMessage方法来向服务端发送下一条信息。

消费者对流传输方法的第一次调用实际是向提供者注册传输会话，且需要依赖返回的StreamHandler向提供者发送后续信息（且需要确保StreamHandler已初始化完毕，保证消费者直接使用其发送消息不会出错），此处选择同步调用。

协议处理

Triple协议

SofaRPC 的 Triple 协议传输层实现目前直接依赖 gRPC 进行。

Triple 目前借助 gRPC API 完成传输操作，因此添加流传输支持较为简单，复用 gRPC 的流传输实现即可，无须处理下层的连接问题。在Triple客户端传输层（TripleClientInvoker）中需要添加的处理主要是将上层传入的StreamHandler适配为gRPC的StreamObserver，并根据方法参数中 StreamObserver 的存在情况，判断当前调用方式，选择对应的gRPC API即可。对于服务端，在IDL中添加流传输方法的定义，修改GenericsService实现新方法即可（泛型调用时）。

泛型调用

当 TripleClientInvoker 中 useGeneric 字段为 false 时，表示消费者调用的服务存在IDL及对应的Stub，可以直接通过生成的Stub进行调用。当 useGeneric 为 true 时，将在运行时动态指定调用的服务及方法，借助底层的泛型服务完成传输。

因此，Triple协议的修改分为两个方面：非泛型调用时，通过服务接口生成的stub进行流式调用；以及泛型调用时，通过预生成的GenericService stub发起流式调用。

IDL修改

对于泛型调用，Triple协议通过将已序列化完成的请求统一封装为 Request，通过预生成的gRPC stub完成传输过程。

因此需要修改transformer.proto, 添加流式调用方法的定义。考虑客户端流和双向流使用可能使用相同的调用方法，此处没有单独定义客户端流的传输方法。

syntax = "proto3";
option java_multiple_files = true;
option java_package = "triple";
option java_outer_classname = "GenericProto";

service GenericService {
    rpc generic (Request) returns (Response) {}

    rpc genericBiStream (stream Request) returns (stream Response){}

    rpc genericServerStream(Request) returns (stream Response){}
}

message Request {
    string serializeType = 1;
    repeated bytes  args = 2;
    repeated string argTypes = 3;
}

message Response {
    string serializeType = 1;
    bytes  data = 2;
    string type = 3;
}

之后，服务端实现新的泛型服务方法

public class GenericServiceImpl extends SofaGenericServiceTriple.GenericServiceImplBase {
//双向流调用实现
   @Override
    public StreamObserver<Request> genericBiStream(StreamObserver<Response> responseObserver) {
        Method serviceMethod = getBidirectionalStreamRequestMethod();
        //通过上下文创建请求
        SofaRequest sofaRequest = TracingContextKey.getKeySofaRequest().get(Context.current());

        if (serviceMethod == null) {
            throw new SofaRpcException(RpcErrorType.SERVER_NOT_FOUND_INVOKER, "Cannot find invoke method " +
                sofaRequest.getMethodName());
        }
        String methodName = serviceMethod.getName();
        try {
            ResponseSerializeStreamHandler serverResponseHandler = new ResponseSerializeStreamHandler(responseObserver,
                getSerialization());

            setBidirectionalStreamRequestParams(sofaRequest, serviceMethod, serverResponseHandler);

            SofaResponse sofaResponse = invoker.invoke(sofaRequest);

            StreamHandler<Object> clientHandler = (StreamHandler<Object>) sofaResponse.getAppResponse();

            return new StreamObserver<Request>() {
                volatile Serializer serializer = null;

                volatile Class<?>[] argTypes   = null;

                @Override
                public void onNext(Request request) {
                    checkInitialize(request);
                    Object message = getInvokeArgs(request, argTypes, serializer, false)[0];
                    clientHandler.onMessage(message);
                }

                private void checkInitialize(Request request) {
                    if (serializer == null && argTypes == null) {
                        synchronized (this) {
                            if (serializer == null && argTypes == null) {
                                serializer = SerializerFactory.getSerializer(request.getSerializeType());
                                argTypes = getArgTypes(request, false);
                            }
                        }
                    }
                }

                @Override
                public void onError(Throwable t) {
                    clientHandler.onException(t);
                }

                @Override
                public void onCompleted() {
                    clientHandler.onFinish();
                }
            };
        } catch (Exception e) {
            LOGGER.error("Invoke " + methodName + " error:", e);
            throw new SofaRpcRuntimeException(e);
        } finally {
            Thread.currentThread().setContextClassLoader(Thread.currentThread().getContextClassLoader());
        }
    }

以上实现的主要难点在于，流调用方法传入及返回的均为StreamHandler，这意味着在实际请求到来之前无法得知具体的请求类型。因此以上的实现中，序列化及方法参数信息是在接收到第一次请求之后才被初始化的。

Triple传输层修改

*由于实际上三种流传输的底层流程差别不大，以下的修改示例均为双向流（BidirectionalStream）。

客户端

以下为对TripleClientInvoker相关的修改。

重构invoke方法，根据callType区分为三种调用流程

    @Override
    public SofaResponse invoke(SofaRequest sofaRequest, int timeout)
            throws Exception {

        MethodDescriptor.MethodType callType = mapCallType(sofaRequest);

        if(!useGeneric){
            //非泛型,根据生成的Stub调用
            return stubCall(sofaRequest,timeout);
        } else if (callType.equals(MethodDescriptor.MethodType.UNARY)) {
            return unaryCall(sofaRequest, timeout);
        } else {
            return streamCall(sofaRequest, timeout, callType);
        }
    }

然后，根据具体调用方式选择具体调用逻辑

private SofaResponse streamCall(SofaRequest sofaRequest, int timeout, boolean useGeneric,MethodDescriptor.MethodType callType) {
         switch (callType){
             case BIDI_STREAMING:
                 return binaryCall(sofaRequest,timeout,useGeneric);
             case CLIENT_STREAMING:
                 return clientStreamCall(sofaRequest,timeout,useGeneric);
             case SERVER_STREAMING:
                 return serverStreamCall(sofaRequest,timeout,useGeneric);
             default:
                 throw new SofaRpcException(RpcErrorType.CLIENT_CALL_TYPE,"Unknown stream call type:"+callType);
         }
    }

实现双向流传输

private SofaResponse binaryStreamCall(SofaRequest sofaRequest, int timeout) {
        StreamHandler streamHandler = (StreamHandler) sofaRequest.getMethodArgs()[0];

        MethodDescriptor<Request, Response> methodDescriptor = getMethodDescriptor(sofaRequest);
        ClientCall<Request, Response> call = channel.newCall(methodDescriptor, buildCustomCallOptions(sofaRequest, timeout));

        StreamObserver<Request> observer = ClientCalls.asyncBidiStreamingCall(
                call,
                new ClientStreamObserverAdapter(
                        streamHandler,
                        sofaRequest.getSerializeType()
                )
        );
        StreamHandler<Request> handler = new StreamHandler() {
            @Override
            public void onMessage(Object message) {
                SofaRequest request = MessageBuilder.copyEmptyRequest(sofaRequest);
                Object[] args = new Object[]{message};
                request.setMethodArgs(args);
                request.setMethodArgSigs(rebuildTrueRequestArgSigs(args));
                Request req = getRequest(request, serialization, serializer, 0);
                observer.onNext(req);
            }

            @Override
            public void onFinish() {
                observer.onCompleted();
            }

            @Override
            public void onException(Throwable throwable) {
                observer.onError(TripleExceptionUtils.asStatusRuntimeException(throwable));
            }
        };
        SofaResponse sofaResponse = new SofaResponse();
        sofaResponse.setAppResponse(handler);
        return sofaResponse;
    }

为MethodDescriptor设置新的调用类型

    private MethodDescriptor<Request,Response> getMethodDescriptor(SofaRequest sofaRequest) {
        String serviceName = sofaRequest.getInterfaceName();
        String methodName = sofaRequest.getMethodName();
        MethodDescriptor.Marshaller<?> requestMarshaller = ProtoUtils.marshaller(Request.getDefaultInstance());
        MethodDescriptor.Marshaller<?> responseMarshaller = ProtoUtils.marshaller(Response.getDefaultInstance());
        String fullMethodName = generateFullMethodName(serviceName, methodName);

        MethodDescriptor.Builder builder = MethodDescriptor
                .newBuilder()
                .setFullMethodName(fullMethodName)
                .setSampledToLocalTracing(true)
                .setRequestMarshaller((MethodDescriptor.Marshaller<Object>) requestMarshaller)
                .setResponseMarshaller((MethodDescriptor.Marshaller<Object>) responseMarshaller);

        MethodDescriptor.MethodType callType = SofaProtoUtils.mapGrpcCallType(sofaRequest.getInvokeType());
        builder.setType(callType);
        return builder.build();
    }

添加工具方法：

//SofaProtoUtils
    public static MethodDescriptor.MethodType mapGrpcCallType(String callType){
        switch (callType){
            case INVOKER_TYPE_ONEWAY:
            case INVOKER_TYPE_FUTURE:
            case INVOKER_TYPE_CALLBACK:
            case INVOKER_TYPE_SYNC:
                return MethodDescriptor.MethodType.UNARY;
            case INVOKER_TYPE_BI_STREAMING:
                return MethodDescriptor.MethodType.BIDI_STREAMING;
            case INVOKER_TYPE_CLIENT_STREAMING:
                return MethodDescriptor.MethodType.CLIENT_STREAMING;
            case INVOKER_TYPE_SERVER_STREAMING:
                return MethodDescriptor.MethodType.SERVER_STREAMING;
            default:
                throw new SofaRpcException(RpcErrorType.CLIENT_CALL_TYPE, "Unsupported invoke type:" + callType);
        }
    }

服务端

方法绑定问题

对于服务端，泛型调用的入口是 GenericService 的实现，即 GenericSerivceImpl。

在为泛型服务的IDL中添加流式方法的定义并重新编译后，发现一元调用时（UNARY）不能正确的选择GenericServiceImple 的 generic 方法，而选择调用了 genericBiStream 方法。

断点调试可以发现，底层 ServerMethodDefinition 和 CallHandler 的绑定错误，使得方法选择了ID错误的CallHandler，从而调用了错误的方法。ServerMethodDefinition 在 ServerImpl.runInternal() 中通过 registry.lookupMethod(methodName)获取。实际调用的是 MutableHandlerRegistry，通过其中的 services 获取 ServerServiceDefinition。而 service 中的 ServerServiceDefinition 由 TripleServer.registerProcessor 放入，自此回到了 SofaRPC 自己的实现。

//TripleServer.registerProcessor
...
ServerServiceDefinition serviceDefinition = getServerServiceDefinition(providerConfig, uniqueIdInvoker);
            this.serviceInfo.put(providerConfig, serviceDefinition);
...

**分析数据流，可以发现将客户端实际服务方法和泛型服务方法绑定起来的操作发生在TripleServer.buildSofaServiceDef方法中。**该流程实际在服务提供者启动，导出服务接口时进行。

private ServerServiceDefinition buildSofaServiceDef(GenericServiceImpl genericService,ProviderConfig providerConfig) {
        ServerServiceDefinition templateDefinition = genericService.bindService();
        ServerCallHandler<Request, Response> templateHandler = (ServerCallHandler<Request, Response>) templateDefinition
            .getMethods().iterator().next().getServerCallHandler(); //这里只拿到了泛型服务中的第一个方法，没有适配新的流调用方法。
        List<MethodDescriptor<Request, Response>> methodDescriptor = getMethodDescriptor(providerConfig);
        List<ServerMethodDefinition<Request, Response>> methodDefs = getMethodDefinitions(templateHandler,//此处使用的 TemplateHandler 应随方法调用方式变化
methodDescriptor);
        ServerServiceDefinition.Builder builder = ServerServiceDefinition.builder(getServiceDescriptor( //应该为将实际服务与泛型服务中的特定方法绑定起来
            templateDefinition, providerConfig, methodDescriptor));
        for (ServerMethodDefinition<Request, Response> methodDef : methodDefs) {
            builder.addMethod(methodDef);
        }
        return builder.build();
    }

修改后的实现：

   private ServerServiceDefinition buildSofaServiceDef(GenericServiceImpl genericService,
                                                        ProviderConfig providerConfig) {
        ServerServiceDefinition templateDefinition = genericService.bindService();
        List<MethodDescriptor<Request, Response>> methodDescriptor = getMethodDescriptor(providerConfig);//拿实际方法的 Descriptor
        List<ServerMethodDefinition<Request, Response>> methodDefs = mapMethodHandler(templateDefinition,methodDescriptor);
        ServerServiceDefinition.Builder builder = ServerServiceDefinition.builder(getServiceDescriptor( 
            templateDefinition, providerConfig, methodDescriptor));
        for (ServerMethodDefinition<Request, Response> methodDef : methodDefs) {
            builder.addMethod(methodDef);
        }
        return builder.build();
    }
//将实际服务与泛型服务中的特定方法绑定起来
   private List<ServerMethodDefinition<Request,Response>>createMethodDefainition(ServerServiceDefinition geneticServiceDefinition, List<MethodDescriptor<Request, Response>> serviceMethods){
            Collection<ServerMethodDefinition<?, ?>> genericServiceMethods = geneticServiceDefinition.getMethods();
            List<ServerMethodDefinition<Request,Response>> serverMethodDefinitions = new ArrayList<>();
            //Map ture service method to certain generic service method.
            for (ServerMethodDefinition<?,?> genericMethods : genericServiceMethods){
                for(MethodDescriptor<Request,Response> methodDescriptor : serviceMethods){

                    if(methodDescriptor.getType().equals(genericMethods.getMethodDescriptor().getType())){

                        ServerMethodDefinition<Request,Response> genericMeth = (ServerMethodDefinition<Request, Response>) genericMethods;

                        serverMethodDefinitions.add(
                                ServerMethodDefinition.create(methodDescriptor, genericMeth.getServerCallHandler())
                        );
                    }
                }
            }
            return serverMethodDefinitions;
    }

这样 ServerMethodDefinition 就和 callHandler 正确的绑定起来了。

方法定位问题

以上方法传入的 ProviderConfig 提供了实际服务接口的配置信息，包括这些方法的调用方式。但是调用方式默认被设置为UNARY。在上层需要附加一部分根据接口方法参数判断调用类型的逻辑。

//TripleServer
private List<MethodDescriptor<Request, Response>> getMethodDescriptor(ProviderConfig providerConfig) {
...
            MethodDescriptor<Request, Response> methodDescriptor = MethodDescriptor.<Request, Response>newBuilder()
                    .setType(MethodDescriptor.MethodType.UNARY) //默认固定为一元调用
                    .setFullMethodName(generateFullMethodName(providerConfig.getInterfaceId(), name))
                    .setSampledToLocalTracing(true)
                    .setRequestMarshaller(ProtoUtils.marshaller(
                            Request.getDefaultInstance()))
                    .setResponseMarshaller(ProtoUtils.marshaller(
                            Response.getDefaultInstance()))
                    .build();
            result.add(methodDescriptor);
...
        return result;
    }

修改后：

    private List<MethodDescriptor<Request, Response>> getMethodDescriptor(ProviderConfig providerConfig) {
        List<MethodDescriptor<Request, Response>> result = new ArrayList<>();
        Set<String> methodNames = SofaProtoUtils.getMethodNames(providerConfig.getInterfaceId());

        for (String name : methodNames) {
            //根据方法名获取调用模式
            MethodDescriptor.MethodType methodType = SofaProtoUtils.mapGrpcCallType(providerConfig.getMethodCallType(name));
            MethodDescriptor<Request, Response> methodDescriptor = MethodDescriptor.<Request, Response>newBuilder()
                    .setType(methodType)
                    .setFullMethodName(generateFullMethodName(providerConfig.getInterfaceId(), name))
                    .setSampledToLocalTracing(true)
                    .setRequestMarshaller(ProtoUtils.marshaller(
                            Request.getDefaultInstance()))
                    .setResponseMarshaller(ProtoUtils.marshaller(
                            Response.getDefaultInstance()))
                    .build();
            result.add(methodDescriptor);
        }
        return result;
    }

在接口配置（AbstractInterfaceConfig）中提供了按Class匹配并缓存方法调用模式的方法：

//AbstractInterfaceConfig
...
    /**
     * MethodCallType ( Map<method.getName(),callType> )
     */
    protected transient volatile Map<String,String> methodCallType = null;
...
     /**
     * Cache the call type of interface methods
     */
    protected void loadMethodCallType(Class<?> interfaceClass){
        Method[] methods =  interfaceClass.getDeclaredMethods();
        this.methodCallType = new ConcurrentHashMap<>();
        for(Method method :methods) {
            //此处调用的是全局处理中提到的 MethodConfig 的新方法，根据方法参数匹配调用方法
 methodCallType.put(method.getName(),MethodConfig.mapStreamType(method,RpcConstants.INVOKER_TYPE_SYNC));
        }
    }

    public String getMethodCallType(String methodName) {
        return methodCallType.get(methodName);
    }

*写在 AbstractInterfaceConfig 是考虑消费者是否能复用这部分逻辑

在 ProviderConfig 初始化时，设置服务引用时会尝试匹配服务方法的具体调用方式，并缓存：

//ProviderConfig
    /**
     * Sets ref.
     *
     * @param ref the ref
     * @return the ref
     */
    public ProviderConfig<T> setRef(T ref) {
        this.ref = ref;
        //加载方法调用方式、缓存
        loadMethodCallType(ref.getClass());
        return this;
    }

流信息序列化问题

对于双向流和客户端流，接口声明时的入参和返回值均仅为 StreamHandler。这导致无论是客户端请求和服务端响应都无法在调用开始就得知具体的消息类型。因此获取、缓存具体消息类型、序列化器的操作需要延迟在客户端/服务端第一次得到具体请求时再进行。

客户端向服务端发送消息使用的StreamHandler适配器，TripleClientInvoker中的匿名实现，将请求序列化为Request：

   StreamHandler<Request> handler = new StreamHandler() {
                    @Override
                    public void onMessage(Object message) {
                        SofaRequest request = MessageBuilder.copyEmptyRequest(sofaRequest);
                        Object[] args = new Object[]{message};
                        request.setMethodArgs(args);
                        request.setMethodArgSigs(rebuildTrueRequestArgSigs(args));
                        Request req = getRequest(request,serialization,serializer);
                        observer.onNext(req);
                    }

                    @Override
                    public void onFinish() {
                        observer.onCompleted();
                    }

                    @Override
                    public void onException(Throwable throwable) {
                        observer.onError(throwable);
                    }
                };

客户端处理服务端响应使用的StreamHandler适配器，将Response反序列化为指定类型：

public class ClientStreamObserverAdapter implements StreamObserver<triple.Response> {

    private final StreamHandler<Object> streamHandler;

    private final Serializer serializer;

    private volatile Class<?> returnType;

    public ClientStreamObserverAdapter(StreamHandler<Object> streamHandler, byte serializeType) {
        this.streamHandler = streamHandler;
        this.serializer = SerializerFactory.getSerializer(serializeType);
    }

    @Override
    public void onNext(triple.Response response) {
        byte[] responseDate = response.getData().toByteArray();
        Object appResponse = null;
        String returnTypeName = response.getType();
        if (responseDate != null && responseDate.length > 0) {
            if(returnType == null && !returnTypeName.isEmpty()) {
                try {
                    returnType = Class.forName(returnTypeName);
                }catch (ClassNotFoundException e){
                    throw new SofaRpcException(RpcErrorType.CLIENT_SERIALIZE,"Can not find return type :"+returnType);
                }
            }
            appResponse = serializer.decode(new ByteArrayWrapperByteBuf(responseDate), returnType, null);
        }

        streamHandler.onMessage(appResponse);
    }

    @Override
    public void onError(Throwable t) {
        streamHandler.onException(t);
    }

    @Override
    public void onCompleted() {
        streamHandler.onFinish();
    }
}

服务端发送响应时使用的StreamHandler适配器，将原始响应序列化为Response：

public class ResponseSerializeStreamHandler<T> implements StreamHandler<T> {

    private final StreamObserver<triple.Response> streamObserver;

    private Serializer serializer;

    private String serializeType;


    public ResponseSerializeStreamHandler(StreamObserver<triple.Response> streamObserver,String serializeType) {
        this.streamObserver = streamObserver;
        serializer = SerializerFactory.getSerializer(serializeType);
        this.serializeType = serializeType;
    }

    @Override
    public void onMessage(T message) {
        Response.Builder builder = Response.newBuilder();
        builder.setType(message.getClass().getName());
        builder.setSerializeType(serializeType);
        builder.setData(ByteString.copyFrom(serializer.encode(message, null).array()));

        streamObserver.onNext(builder.build());
    }

    @Override
    public void onFinish() {
        streamObserver.onCompleted();
    }

    @Override
    public void onException(Throwable throwable) {
        streamObserver.onError(throwable);
    }

}

服务端处理客户端请求使用的StreamHandler，TripleClientInvoker 中的匿名实现，将请求反序列化为指定类型：

return new StreamObserver<Request>() {
                volatile Serializer serializer = null;

                volatile Class<?>[] argTypes = null;

                @Override
                public void onNext(Request request) {
                    checkInitialize(request);
                    Object message = getInvokeArgs(request, argTypes, serializer)[0];
                    clientHandler.onMessage(message);
                }

                private void checkInitialize(Request request){
                    if (serializer == null && argTypes == null) {
                        synchronized (this) {
                            if (serializer == null && argTypes == null) {
                                serializer = SerializerFactory.getSerializer(request.getSerializeType());
                                argTypes = getArgTypes(request);
                            }
                        }
                    }
                }

                @Override
                public void onError(Throwable t) {
                    clientHandler.onException(t);
                }

                @Override
                public void onCompleted() {
                    clientHandler.onFinish();
                }
            };

Summary by CodeRabbit

New Features
- Added support for new RPC invocation types: unary, client streaming, server streaming, and bidirectional streaming.
- Introduced new methods for bidirectional and server streaming in the GenericService.
- Enhanced streaming capabilities in test scenarios with new methods and tests for bi-directional and server-side streaming.
Bug Fixes
- Updated method invocation logic to correctly handle different streaming scenarios, enhancing the robustness of streaming communication.
Documentation
- Added detailed comments and documentation related to new streaming functionalities and method invocation types.
Refactor
- Refactored method signature changes across multiple classes to support enhanced streaming functionalities.
- Improved error handling and response processing logic for different call types in streaming scenarios.
Tests
- Introduced comprehensive test cases for bi-directional and server streaming, ensuring functionality works as expected under various conditions.
Chores
- Updated import statements and cleaned up unused imports across several files to streamline dependencies and maintain code cleanliness.

sofastack-cla · 2023-08-25T12:49:10Z

Hi @namelessssssssssss, welcome to SOFAStack community, Please sign Contributor License Agreement!

After you signed CLA, we will automatically sync the status of this pull request in 3 minutes.

namelessssssssssss · 2023-08-25T13:19:26Z

@EvenLjj PTAL :)

codecov · 2023-08-25T14:40:03Z

Codecov Report

Attention: Patch coverage is 83.66013% with 50 lines in your changes are missing coverage. Please review.

Project coverage is 72.19%. Comparing base (e67ea54) to head (df27427).
Report is 8 commits behind head on master.

❗ Current head df27427 differs from pull request most recent head c7435df. Consider uploading reports for the commit c7435df to get more accurate results

Files	Patch %	Lines
...pay/sofa/rpc/server/triple/GenericServiceImpl.java	75.51%	18 Missing and 6 partials ⚠️
...sofa/rpc/transport/triple/TripleClientInvoker.java	89.41%	5 Missing and 4 partials ⚠️
.../java/com/alipay/sofa/rpc/config/MethodConfig.java	61.53%	2 Missing and 3 partials ⚠️
...age/triple/stream/ClientStreamObserverAdapter.java	80.00%	2 Missing and 2 partials ⚠️
...java/com/alipay/sofa/rpc/utils/SofaProtoUtils.java	50.00%	2 Missing and 1 partial ⚠️
...om/alipay/sofa/rpc/utils/TripleExceptionUtils.java	25.00%	2 Missing and 1 partial ⚠️
...va/com/alipay/sofa/rpc/client/AbstractCluster.java	50.00%	0 Missing and 2 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #1360      +/-   ##
============================================
+ Coverage     72.04%   72.19%   +0.15%     
- Complexity      791      793       +2     
============================================
  Files           422      425       +3     
  Lines         17814    18052     +238     
  Branches       2768     2801      +33     
============================================
+ Hits          12834    13033     +199     
- Misses         3565     3590      +25     
- Partials       1415     1429      +14

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

nobodyiam · 2023-08-26T00:31:47Z

This is interesting 😀

OrezzerO · 2023-11-12T15:24:02Z

更好的方案是使用自定义的 StreamObserver（此处先叫做StreamHandler），将其和gRPC解耦，在 Triple 传输层通过适配器转换为gRPC StreamObserver。此处使用自定义的StreamHandler。

确实是这样的, 因此方案中还需要考虑如何让 bolt 协议来支持流式处理. 我能想到两种方案:

在 bolt 协议中支持流式处理. 目前 bolt 协议用的是ping-pong 的方式只支持一来一回的应答模式,需要改动底层协议才能支持流式.
通讯层(grpc/bolt)都采用 ping-pong 的模式, 在 RPC 之中模拟流式处理. 这种方案的好处是可以兼容各种通讯层协议; 缺点是性能不是最优的, 有比较大的浪费.

我个人比较希望采用第一种方案,让bolt 原生支持 stream ,这能让使用bolt 的其他框架,也能享受到 stream 的优势.

回到这个 PR 上, 我建议统筹考虑 bolt/sofarpc 应该如何支持 stream ,再来合并这个PR.

coderabbitai · 2024-03-27T11:29:00Z

Walkthrough

The updates involve enriching RPC functionalities with added support for various streaming types (unary, client, server, and bidirectional). Changes include adjusting method signatures to accommodate streaming, introducing new utilities for handling streaming and exceptions, and expanding test cases to ensure robustness. This overhaul not only streamlines method invocations but also extends the RPC framework's capabilities to effectively handle complex streaming scenarios.

Changes

Files	Change Summary
`RpcConstants.java`, `AbstractInterfaceConfig.java`, `ConsumerConfig.java`, `ProviderConfig.java`	Added constants for new streaming types, caching method call types, and updated method configurations.
`MethodConfig.java`, `AbstractCluster.java`, `TripleClientInvoker.java`	Enhanced handling of different streaming scenarios, updated timeout behaviors, and improved error handling.
`MessageBuilder.java`, `StreamHandler.java`, `SofaStreamObserver.java`	Introduced new methods for message handling and stream observation.
`ConsumerGenericFilter.java`, `UniqueIdInvoker.java`, `TripleServer.java`	Updated method signatures to accept `SofaRequest`, improved method identification, and enhanced stream handling.
`GenericServiceImpl.java`, `SofaProtoUtils.java`	Expanded service functionalities, added utility methods for streaming, and updated method mapping utilities.
`transformer.proto`, `helloworld.proto`	Added new RPC methods supporting various streaming types.
`HelloService.java`, `HelloServiceImpl.java`, `ServerResponse.java`, `ClientRequest.java`, `ExtendClientRequest.java`	Defined new interfaces and classes for handling streaming in test scenarios.
`TripleGenericStreamTest.java`, `TripleStubStreamTest.java`	Introduced comprehensive test cases for bi-directional and server streaming.

🐇✨📜
O hark! New streams do flow through code,
With swift, smooth currents, they now bode.
From unary to bidirectional streams,
Through SOFA's veins, the data gleams.
A rabbit’s cheer for the tech brigade,
For each clever tweak and upgrade made!
🌟🐾

Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media?

Share

Tips

Chat

There are 3 ways to chat with CodeRabbit:

Review comments: Directly reply to a review comment made by CodeRabbit. Example:
- I pushed a fix in commit <commit_id>.
- Generate unit testing code for this file.
- Open a follow-up GitHub issue for this discussion.
Files and specific lines of code (under the "Files changed" tab): Tag @coderabbitai in a new review comment at the desired location with your query. Examples:
- @coderabbitai generate unit testing code for this file.
- @coderabbitai modularize this function.
PR comments: Tag @coderabbitai in a new PR comment to ask questions about the PR branch. For the best results, please provide a very specific query, as very limited context is provided in this mode. Examples:
- @coderabbitai generate interesting stats about this repository and render them as a table.
- @coderabbitai show all the console.log statements in this repository.
- @coderabbitai read src/utils.ts and generate unit testing code.
- @coderabbitai read the files in the src/scheduler package and generate a class diagram using mermaid and a README in the markdown format.

Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments.

CodeRabbit Commands (invoked as PR comments)

@coderabbitai pause to pause the reviews on a PR.
@coderabbitai resume to resume the paused reviews.
@coderabbitai review to trigger a review. This is useful when automatic reviews are disabled for the repository.
@coderabbitai resolve resolve all the CodeRabbit review comments.
@coderabbitai help to get help.

Additionally, you can add @coderabbitai ignore anywhere in the PR description to prevent this PR from being reviewed.

CodeRabbit Configration File (`.coderabbit.yaml`)

You can programmatically configure CodeRabbit by adding a .coderabbit.yaml file to the root of your repository.
Please see the configuration documentation for more information.
If your editor has YAML language server enabled, you can add the path at the top of this file to enable auto-completion and validation: # yaml-language-server: $schema=https://coderabbit.ai/integrations/schema.v2.json

Documentation and Community

Visit our Documentation for detailed information on how to use CodeRabbit.
Join our Discord Community to get help, request features, and share feedback.
Follow us on X/Twitter for updates and announcements.

coderabbitai

Review Status

Actionable comments generated: 14

Configuration used: CodeRabbit UI

Commits

Files that changed from the base of the PR and between e67ea54 and df27427.

Files ignored due to path filters (3)

remoting/remoting-triple/build/generated/source/proto/main/java/triple/GenericProto.java is excluded by: !**/generated/**
remoting/remoting-triple/build/generated/source/proto/main/java/triple/GenericServiceGrpc.java is excluded by: !**/generated/**
remoting/remoting-triple/build/generated/source/proto/main/java/triple/SofaGenericServiceTriple.java is excluded by: !**/generated/**

Files selected for processing (29)

bootstrap/bootstrap-api/src/main/java/com/alipay/sofa/rpc/bootstrap/DefaultClientProxyInvoker.java (1 hunks)
core-impl/client/src/main/java/com/alipay/sofa/rpc/client/AbstractCluster.java (1 hunks)
core/api/src/main/java/com/alipay/sofa/rpc/common/RpcConstants.java (1 hunks)
core/api/src/main/java/com/alipay/sofa/rpc/config/AbstractInterfaceConfig.java (3 hunks)
core/api/src/main/java/com/alipay/sofa/rpc/config/ConsumerConfig.java (2 hunks)
core/api/src/main/java/com/alipay/sofa/rpc/config/MethodConfig.java (3 hunks)
core/api/src/main/java/com/alipay/sofa/rpc/config/ProviderConfig.java (1 hunks)
core/api/src/main/java/com/alipay/sofa/rpc/message/MessageBuilder.java (1 hunks)
core/api/src/main/java/com/alipay/sofa/rpc/transport/StreamHandler.java (1 hunks)
core/exception/src/main/java/com/alipay/sofa/rpc/core/exception/RpcErrorType.java (1 hunks)
remoting/remoting-bolt/src/main/java/com/alipay/sofa/rpc/filter/ConsumerGenericFilter.java (1 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/message/triple/stream/ClientStreamObserverAdapter.java (1 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/message/triple/stream/ResponseSerializeStreamHandler.java (1 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/GenericServiceImpl.java (6 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/TripleServer.java (8 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/UniqueIdInvoker.java (3 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/transport/triple/TripleClientInvoker.java (8 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/utils/SofaProtoUtils.java (2 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/utils/TripleExceptionUtils.java (1 hunks)
remoting/remoting-triple/src/main/proto/transformer.proto (1 hunks)
remoting/remoting-triple/src/test/java/com/alipay/sofa/rpc/server/triple/GenericServiceImplTest.java (2 hunks)
test/test-integration/src/main/proto/helloworld.proto (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/GreeterImpl.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/TripleServerTest.java (2 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/triple/stream/ClientRequest.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/triple/stream/HelloService.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/triple/stream/HelloServiceImpl.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/triple/stream/ServerResponse.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/triple/stream/TripleGenericStreamTest.java (1 hunks)

Files not reviewed due to errors (1)

(no review received)

Additional comments: 53

remoting/remoting-triple/src/main/proto/transformer.proto (2)

13-13: The addition of genericBiStream method supports bidirectional streaming, aligning with the PR's objectives to enhance streaming capabilities.

15-15: The addition of genericServerStream method supports server streaming, which is consistent with the PR's goal to introduce various streaming models.

test/test-integration/src/test/java/com/alipay/sofa/rpc/triple/stream/ServerResponse.java (1)

19-35: The ServerResponse class is well-structured with appropriate fields and methods. Good job following best practices for encapsulation.

test/test-integration/src/main/proto/helloworld.proto (1)

29-29: The addition of SayHelloBinary method supports bidirectional streaming, aligning with the PR's objectives to enhance streaming capabilities.

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/utils/TripleExceptionUtils.java (1)

24-30: The asStatusRuntimeException method is well-implemented, correctly handling null cases and converting Throwable into a StatusRuntimeException. Good job following best practices for error handling.

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

21-38: The HelloService interface is well-designed, covering unary, bidirectional streaming, and server streaming use cases. The constants for triggering specific stream behaviors are a thoughtful addition.

core/api/src/main/java/com/alipay/sofa/rpc/transport/StreamHandler.java (1)

22-47: The StreamHandler interface provides a clear and concise contract for handling streaming messages, completion, and exceptions, aligning well with the PR's objectives and resembling gRPC's StreamObserver.

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/message/triple/stream/ResponseSerializeStreamHandler.java (1)

30-62: The ResponseSerializeStreamHandler class correctly implements the StreamHandler interface, effectively handling message serialization, stream completion, and exceptions. The dynamic serializer selection based on serializeType is a good practice.

core/exception/src/main/java/com/alipay/sofa/rpc/core/exception/RpcErrorType.java (1)

96-99: The addition of CLIENT_CALL_TYPE with the value 260 to represent an unsupported RPC call type exception is a good practice. It enhances error specificity and aids in debugging. However, it would be beneficial to ensure that this new error type is well-documented where it's used, to inform developers of its purpose and usage context.

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/message/triple/stream/ClientStreamObserverAdapter.java (1)

43-70: The implementation of ClientStreamObserverAdapter is well-structured and aligns with the objectives of integrating streaming capabilities. However, it's recommended to add error handling around the deserialization logic in the onNext method to gracefully handle potential serialization issues.

test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/GreeterImpl.java (1)

55-73: The sayHelloBinary method correctly implements a binary streaming pattern. For completeness and better test coverage, consider adding basic logging or handling in the onError and onCompleted methods, even if it's just logging the events.

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

71-87: The implementation of mapGrpcCallType method is well-done, providing clear mapping between SOFA RPC call types and gRPC method types. It's recommended to add documentation explaining the mapping logic and the scenarios in which the SofaRpcException will be thrown for unsupported invoke types.

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

21-89: The implementations of streaming methods in HelloServiceImpl are well-done and align with the expected streaming patterns. It would be beneficial to add comments explaining the purpose and usage of special commands like CMD_TRIGGER_STREAM_FINISH and CMD_TRIGGER_STEAM_ERROR within the test scenarios.

remoting/remoting-triple/src/test/java/com/alipay/sofa/rpc/server/triple/GenericServiceImplTest.java (1)

66-66: The addition of the providerConfig parameter to the GenericServiceImpl constructor and the extra argument in the TripleClientInvoker.getRequest call are significant changes that enhance integration with the streaming capabilities. Ensure that the impact of these changes is well-documented and thoroughly tested to confirm their effectiveness and to avoid potential regressions.

bootstrap/bootstrap-api/src/main/java/com/alipay/sofa/rpc/bootstrap/DefaultClientProxyInvoker.java (1)

95-95: The change from getMethodInvokeType(request.getMethodName()) to getMethodInvokeType(request) in the decorateRequest method likely enhances how the invokeType is determined by considering more information from the SofaRequest object. Ensure that the new method signature is supported everywhere it's used and that this change integrates well with the rest of the system.

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/UniqueIdInvoker.java (2)

19-19: The addition of imports for RpcConstants, StreamHandler, and Java utility classes supports the new functionality introduced in this file. These are standard and necessary for the changes made.

Also applies to: 31-31, 35-35, 37-37

194-194: Changing the access level of the getServiceUniqueName method to public is likely intended to make it accessible from outside the class. Ensure there are no unintended side effects of making this method public.

core/api/src/main/java/com/alipay/sofa/rpc/config/MethodConfig.java (2)

19-20: The addition of imports for RpcConstants, RpcErrorType, SofaRpcException, SofaResponseCallback, and StreamHandler supports the new functionality introduced in this file. These are standard and necessary for the changes made.

Also applies to: 21-21, 23-23

336-369: The mapStreamType method introduces logic to determine the stream call type for a given method, which is a crucial enhancement for supporting streaming. Consider refining the error messages thrown by this method to ensure they are clear and provide sufficient information for debugging.
core/api/src/main/java/com/alipay/sofa/rpc/config/ProviderConfig.java (1)
211-211: The addition of the loadMethodCallType call in the setRef method likely aims to enhance automatic configuration based on the reference class. Ensure that the loadMethodCallType method is implemented correctly and assess its impact on performance and behavior.

Verification successful

The loadMethodCallType method is indeed implemented in AbstractInterfaceConfig.java and used in ProviderConfig.java, confirming the initial observation. While the method's existence and purpose align with the expectations, assessing its implementation details and impact on performance and behavior directly is beyond the current scope. It's recommended to review the implementation of loadMethodCallType for correctness and efficiency.
Scripts Executed

The following scripts were executed for the analysis:

Script:
#!/bin/bash
# Verify the implementation of loadMethodCallType
grep -R "loadMethodCallType" .
# Assess performance impact if possible
Length of output: 291
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/GenericServiceImpl.java (7)

21-26: Adding imports for RpcConstants, ReflectCache, StringUtils, and ProviderConfig indicates that these classes are now being used within GenericServiceImpl. Ensure that these new dependencies are justified and optimally used within the implementation to avoid unnecessary coupling and to maintain modularity.

34-37: The addition of imports for ResponseSerializeStreamHandler and StreamHandler suggests the introduction of streaming capabilities within GenericServiceImpl. It's crucial to ensure that these handlers are correctly implemented and integrated, particularly focusing on error handling, stream lifecycle management, and performance considerations.

61-66: The introduction of a new private field providerConfig and its initialization within the constructor enhances the class's capability to access provider configuration details. This change should be carefully reviewed to ensure that it aligns with the class's responsibilities and does not introduce unnecessary dependencies or complexity.

77-78: The method setClassLoaderAndGetRequestMethod is being called with specific parameters. It's important to verify that this method correctly handles class loading and method retrieval, especially considering the dynamic nature of class loading in Java and potential classloader leaks.

84-84: The method setUnaryOrServerRequestParams is being utilized to set request parameters for unary or server streaming calls. Ensure that this method adequately validates its inputs and correctly handles the serialization and deserialization of request parameters to prevent issues such as data corruption or security vulnerabilities.

198-203: The method getBidirectionalStreamRequestMethod retrieves the method for bidirectional streaming. It's important to verify that this method accurately identifies the correct method to be used for bidirectional streaming, considering potential overloads and ensuring compatibility with the streaming protocol used.

325-335: The methods getSerialization and getDefaultSerialization are designed to retrieve the serialization type for the provider. It's important to ensure that these methods correctly determine the serialization type based on the provider configuration and fall back to a sensible default if necessary. Additionally, the impact of serialization choices on performance and compatibility should be considered.

test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/TripleServerTest.java (1)

36-43: Adding imports for StreamObserver, CountDownLatch, and TimeUnit indicates that the new test method testBiStream utilizes these classes for testing bidirectional streaming functionality. Ensure that these imports are optimally used within the test to maintain clarity and focus on testing objectives.

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/TripleServer.java (3)

60-62: The addition of imports for ArrayList and Collection suggests new collection manipulations within TripleServer. Ensure that these collections are used efficiently, considering their impact on performance, especially in concurrent environments or when handling large datasets.

117-117: The adjustment in the declaration of invokerMap to explicitly specify its type as ConcurrentHashMap is a good practice for clarity and ensuring thread safety in concurrent access scenarios. This change aligns with the need for thread-safe operations on the invokerMap.

301-301: The addition of providerConfig as a parameter in the instantiation of GenericServiceImpl within getServerServiceDefinition method enhances the ability to pass configuration details directly to the service implementation. Ensure that this change is consistently applied and that the providerConfig is used appropriately within GenericServiceImpl.

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/transport/triple/TripleClientInvoker.java (12)

41-50: New imports related to message handling and stream processing have been introduced. Ensure that these imports are used effectively throughout the file and that there are no unused imports to maintain code cleanliness.

74-89: The introduction of new class fields such as serializer and serialization is noted. It's important to ensure that these fields are properly encapsulated and accessed in a thread-safe manner if necessary, given the concurrent nature of network programming.

111-125: The getRequest method has been modified to support serialization of method arguments. Ensure that the serialization process is secure and efficient, especially for large data payloads. Consider handling potential serialization exceptions gracefully.

145-153: The invoke method now includes logic to handle different call types (unary, client streaming, server streaming, and bidirectional streaming). It's crucial to ensure that the mapping between SofaRequest call types and gRPC call types is accurate and that each case is handled appropriately.

156-167: The mapCallType method correctly maps SOFA RPC call types to gRPC call types. Verify that all supported call types are covered and that the default case (unary) is the intended fallback option.

170-180: The streamCall method's switch statement handles different streaming call types. Ensure that each case is implemented correctly and that the default case throws an exception for unknown call types, as this aids in debugging and maintainability.

184-200: The unaryCall method's implementation for unary calls looks correct. Pay attention to error handling and ensure that the response deserialization process is secure and efficient.

203-210: The stubCall method uses reflection to invoke methods on the stub. Ensure that this approach is safe and consider potential performance implications of using reflection.

213-250: The binaryStreamCall and related streaming call methods (clientStreamCall, serverStreamCall) introduce complex logic for handling streaming calls. It's important to thoroughly test these methods to ensure they handle stream lifecycle events correctly and efficiently.

318-318: The asyncInvoke method has been modified to support asynchronous invocation. Ensure that the asynchronous logic is correctly implemented and that callback and future mechanisms are properly handled.

345-355: The rebuildTrueRequestArgSigs method rebuilds argument signatures for stream calls. Verify that this method correctly handles null arguments and that the argument types are accurately reconstructed.

479-494: The getMethodDescriptor method constructs a MethodDescriptor for gRPC calls. Ensure that the method descriptor is correctly built, including the correct call type and marshaller for request and response types.

core/api/src/main/java/com/alipay/sofa/rpc/common/RpcConstants.java (1)

124-139: New constants for different invocation types (unary, client streaming, server streaming, and bidirectional streaming) have been added. Ensure that these constants are used consistently throughout the codebase and that their naming convention aligns with existing constants for clarity and maintainability.
core/api/src/main/java/com/alipay/sofa/rpc/config/ConsumerConfig.java (4)
30-30: Importing SofaRequest is necessary for the changes made in this file, specifically for the method getMethodInvokeType which now accepts a SofaRequest object instead of a method name. This change aligns with the PR's objective to enhance stream transport support.

939-952: The method getMethodInvokeType has been modified to accept a SofaRequest object instead of a method name. This change is crucial for supporting stream transport, as it allows the method to determine the invocation type based on the request's details, including the method name and potentially other request-specific information in the future. This approach enhances flexibility and aligns with the PR's objectives.

However, it's important to ensure that all calls to this method throughout the codebase have been updated to pass a SofaRequest object instead of just a method name to avoid runtime errors.

Verification successful

Based on the script execution results, which did not find any instances of getMethodInvokeType being called with a single string argument, it appears that all calls to this method have been correctly updated to pass a SofaRequest object. This aligns with the requirements for supporting stream transport and enhancing flexibility as mentioned in the review comment. Therefore, the changes have been successfully verified.
Scripts Executed

The following scripts were executed for the analysis:

Script:
#!/bin/bash
# Search for calls to getMethodInvokeType with a single string argument, which should now be updated.
ast-grep --lang java --pattern $'getMethodInvokeType($_)' | grep -v 'SofaRequest'
Length of output: 80
* 959-970: The `getAndCacheCallType` method is a new addition that dynamically determines and caches the call type for a given `SofaRequest`. This method is essential for the stream transport support, as it allows for efficient handling of different types of streaming requests by caching the call type at runtime. This caching mechanism can improve performance by avoiding repeated determination of the call type for the same method.
It's important to ensure that the caching mechanism is thread-safe and that the cache is properly invalidated or updated when necessary to prevent stale or incorrect call type information.

977-994: The buildMethodConfigKey and getMethodSignature methods are new additions that support the caching mechanism for method call types. buildMethodConfigKey constructs a unique key for each method configuration, which includes the method signature. getMethodSignature generates a string representation of a method's signature, which is used as part of the cache key.

These methods are crucial for ensuring that the call type for each unique method is correctly identified and cached. It's important to verify that the method signature generation correctly handles overloads and different parameter types to ensure uniqueness and accuracy in the cache keys.
core/api/src/main/java/com/alipay/sofa/rpc/config/AbstractInterfaceConfig.java (4)

215-218: The introduction of the methodCallType field is a significant addition that supports the caching of method call types. This field is crucial for the stream transport support, as it allows for efficient retrieval of the call type for each method, which is essential for handling different types of streaming requests.

It's important to ensure that access to this field is thread-safe, given that it may be accessed and modified by multiple threads concurrently.

258-264: The loadMethodCallType method is responsible for initializing the methodCallType cache with the default call type for each method of the provided interface class. This method is crucial for the stream transport support, as it pre-populates the cache with default values, which can then be overridden based on specific configurations or request types.

It's important to ensure that this method is called at the appropriate time during the initialization of the interface config to ensure that the cache is correctly populated before it's used.

266-268: The getMethodCallType method provides a way to retrieve the cached call type for a given method name. This method is essential for efficiently determining the call type for method invocations, supporting the stream transport enhancements introduced in the PR.

It's important to verify that the method name used as the key in the methodCallType map accurately reflects the method's signature, especially in cases where method overloading is used, to ensure the correct call type is retrieved.

1038-1038: The buildmkey method is used to construct a unique key for caching method-specific configurations, including the call type. This method is crucial for ensuring that configurations are correctly associated with the appropriate methods, especially in cases where methods are overloaded.

It's important to ensure that the key generation logic accurately reflects the method's signature to prevent collisions or incorrect configurations being applied to methods.

core-impl/client/src/main/java/com/alipay/sofa/rpc/client/AbstractCluster.java (1)

660-663: Using Integer.MAX_VALUE as a timeout for streaming requests effectively means "wait indefinitely." While this might be suitable for some streaming scenarios, it's important to consider if there are cases where a more finite timeout would be appropriate to avoid potential resource exhaustion or hanging connections. Additionally, the uniform handling of different streaming types (CLIENT_STREAMING, BI_STREAMING, SERVER_STREAMING) without differentiation might be intentional but ensure this aligns with the design goals and consider if any specific handling is needed for each type.

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/UniqueIdInvoker.java

...ting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/GenericServiceImpl.java

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

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/TripleServer.java

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

core/api/src/main/java/com/alipay/sofa/rpc/message/MessageBuilder.java

remoting/remoting-bolt/src/main/java/com/alipay/sofa/rpc/filter/ConsumerGenericFilter.java

coderabbitai

Actionable comments posted: 10

...ting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/GenericServiceImpl.java

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

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

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

coderabbitai

Actionable comments posted: 2

.../remoting-triple/src/main/java/com/alipay/sofa/rpc/transport/triple/TripleClientInvoker.java

...ting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/GenericServiceImpl.java

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/UniqueIdInvoker.java

.../remoting-triple/src/main/java/com/alipay/sofa/rpc/transport/triple/TripleClientInvoker.java

core/api/src/main/java/com/alipay/sofa/rpc/message/MessageBuilder.java

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

coderabbitai

Actionable comments posted: 3

Out of diff range and nitpick comments (3)

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

23-23: Ensure consistent naming for command constants.

Consider using a more descriptive name for CMD_TRIGGER_STREAM_FINISH to clarify its purpose, such as CMD_FINISH_STREAM.

25-25: Ensure consistent naming for command constants.

Consider using a more descriptive name for CMD_TRIGGER_STREAM_ERROR to clarify its purpose, such as CMD_ERROR_STREAM.

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/transport/triple/TripleClientInvoker.java (1)

483-498: Review the method descriptor construction for completeness.

Consider adding more detailed comments explaining each step in the getMethodDescriptor method. This will improve code readability and maintainability, especially for new developers.

Review Details

Configuration used: CodeRabbit UI
Review profile: CHILL

Commits

Files that changed from the base of the PR and between 266006d and 437ff33.

Files selected for processing (11)

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/GenericServiceImpl.java (5 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/TripleServer.java (5 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/UniqueIdInvoker.java (3 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/transport/triple/TripleClientInvoker.java (10 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/utils/SofaProtoUtils.java (2 hunks)
remoting/remoting-triple/src/test/java/com/alipay/sofa/rpc/server/triple/GenericServiceImplTest.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ClientRequest.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ExtendClientRequest.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/HelloService.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/HelloServiceImpl.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/TripleGenericStreamTest.java (1 hunks)

Files not reviewed due to errors (1)

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/TripleServer.java (no review received)

Files skipped from review due to trivial changes (2)

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

Additional comments not posted (14)

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

29-29: Define the method for bidirectional streaming.

Ensure that the StreamHandler types for request and response are correctly used and documented.

31-31: Define the method for server streaming.

Ensure that the StreamHandler for the response is correctly used and documented.

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

27-60: Review the implementation of bidirectional streaming.

Ensure that the method handles different command types (finish, error) and processes extended client requests correctly.

64-78: Review the implementation of server streaming.

Ensure that the method handles the error command correctly and finishes the stream appropriately.

remoting/remoting-triple/src/test/java/com/alipay/sofa/rpc/server/triple/GenericServiceImplTest.java (1)

139-139: Ensure correct building of requests in tests.

Verify that the TripleClientInvoker.buildRequest method is correctly used to construct requests based on SofaRequest.

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

79-93: Review the method for mapping gRPC call types.

Ensure that the method correctly translates custom call types into gRPC MethodDescriptor.MethodType.

95-105: Review the caching mechanism for stream call types.

Verify that the method correctly caches the stream type for each method based on its signature.

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/UniqueIdInvoker.java (1)

181-191: Review the method for getting declared methods based on call type.

Ensure that the method correctly adjusts the argument types list based on the server streaming call type.

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

87-158: Review the test cases for bidirectional streaming.

Verify that the tests correctly simulate different streaming scenarios and handle exceptions as expected.

160-214: Review the test cases for server streaming.

Ensure that the tests correctly handle finish and exception scenarios in server streaming.

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/GenericServiceImpl.java (2)

98-167: Review the implementation of bidirectional streaming.

Ensure that the method correctly initializes streaming, handles stream messages, and manages errors.

169-196: Review the implementation of server streaming.

Verify that the method correctly initializes the stream, sets up request parameters, and handles stream completion and errors.
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/transport/triple/TripleClientInvoker.java (2)
115-115: Ensure methodCallType is properly initialized.

This script checks that methodCallType is initialized before any usage to prevent runtime errors.

Verification successful

The verification process has confirmed that methodCallType is initialized before its usage in the TripleClientInvoker.java file. The variable is initially set to null and subsequently assigned a value from SofaProtoUtils.cacheStreamCallType(consumerConfig.getProxyClass()). It is then used to retrieve a value based on a method name. This sequence ensures that methodCallType is properly initialized before it is accessed, aligning with the concerns raised in the review comment.

Conclusion:

The initialization and usage of methodCallType are correctly implemented in the TripleClientInvoker.java.
Scripts Executed

The following scripts were executed for the analysis:

Script:
#!/bin/bash
# Description: Verify that `methodCallType` is initialized before use.

# Test: Search for the initialization and usage of `methodCallType`. Expect: Initialization before usage.
rg --type java 'methodCallType' -- 'TripleClientInvoker.java'
Length of output: 124

Script:
#!/bin/bash
# Description: Verify that `methodCallType` is initialized before use in TripleClientInvoker.java.

# Test: Search for the initialization and usage of `methodCallType` in TripleClientInvoker.java. Expect: Initialization before usage.
fd 'TripleClientInvoker.java' | xargs rg 'methodCallType'
Length of output: 275
199-239: Ensure proper error handling in streaming calls.

This script checks that errors in streaming calls are handled properly to ensure robustness.

Verification successful

The error handling in the streaming calls within the TripleClientInvoker.java file has been verified. The method onError is appropriately used to handle exceptions by converting them into StatusRuntimeException and passing them to the observer's onError method. This ensures that errors are properly propagated and handled in the streaming context.

The onError method in the StreamHandler interface is implemented to handle exceptions by calling observer.onError, which is a standard practice in handling streaming errors in gRPC (which is likely the underlying technology given the context).

This confirms that the error handling mechanisms are in place as expected.
Scripts Executed

The following scripts were executed for the analysis:

Script:
#!/bin/bash
# Description: Verify error handling in streaming calls.

# Test: Search for error handling patterns in streaming methods. Expect: Proper error handling.
rg --type java 'onError' -- 'TripleClientInvoker.java'
Length of output: 117

Script:
#!/bin/bash
# Description: Verify error handling in streaming calls in the correct file.

# Test: Search for error handling patterns in streaming methods. Expect: Proper error handling.
fd 'TripleClientInvoker.java' | xargs rg 'onError'
Length of output: 258

.../remoting-triple/src/main/java/com/alipay/sofa/rpc/transport/triple/TripleClientInvoker.java

coderabbitai

Actionable comments posted: 13

Out of diff range and nitpick comments (2)

test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ExtendServerResponse.java (1)

27-30: Ensure proper documentation for the constructor.

Consider adding Javadoc comments to the constructor to explain its parameters and purpose, enhancing code maintainability and readability.
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/GenericServiceImpl.java (1)
Line range hint 68-97: Ensure thread safety and context management.

The method modifies the thread's context class loader but may not properly restore it if an exception is thrown before the finally block. Consider using a try-finally block to ensure the original class loader is always restored.
+ finally {
+     Thread.currentThread().setContextClassLoader(oldClassLoader);
+ }

Review Details

Configuration used: CodeRabbit UI
Review profile: CHILL

Commits

Files that changed from the base of the PR and between df27427 and 3d00d9d.

Files selected for processing (18)

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/message/triple/stream/ClientStreamObserverAdapter.java (1 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/message/triple/stream/ResponseSerializeStreamHandler.java (1 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/GenericServiceImpl.java (5 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/TripleServer.java (5 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/UniqueIdInvoker.java (3 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/transport/triple/TripleClientInvoker.java (10 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/utils/SofaProtoUtils.java (2 hunks)
remoting/remoting-triple/src/test/java/com/alipay/sofa/rpc/server/triple/GenericServiceImplTest.java (1 hunks)
test/test-integration/src/main/proto/helloworld.proto (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/GreeterImpl.java (2 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ClientRequest.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ExtendClientRequest.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ExtendServerResponse.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/HelloService.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/HelloServiceImpl.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ServerResponse.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/TripleGenericStreamTest.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/TripleStubStreamTest.java (1 hunks)

Files skipped from review due to trivial changes (1)

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

Files skipped from review as they are similar to previous changes (9)

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/TripleServer.java
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/UniqueIdInvoker.java
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/utils/SofaProtoUtils.java
remoting/remoting-triple/src/test/java/com/alipay/sofa/rpc/server/triple/GenericServiceImplTest.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ClientRequest.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ExtendClientRequest.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/HelloService.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/HelloServiceImpl.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/TripleGenericStreamTest.java

Additional comments not posted (15)

test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ExtendServerResponse.java (2)

32-34: LGTM! The getter method for extendString is implemented correctly.

36-38: LGTM! The setter method for extendString is implemented correctly.

test/test-integration/src/main/proto/helloworld.proto (4)

28-28: LGTM! The unary RPC method SayHello is defined correctly.

29-29: LGTM! The bidirectional streaming RPC method SayHelloBinary is defined correctly.

31-31: LGTM! The client streaming RPC method SayHelloClientStream is defined correctly.

33-33: LGTM! The server streaming RPC method SayHelloServerStream is defined correctly.

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/message/triple/stream/ResponseSerializeStreamHandler.java (2)

58-60: LGTM! The onFinish method correctly completes the stream.

63-65: LGTM! The onException method correctly handles exceptions by converting them and passing to onError.

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/message/triple/stream/ClientStreamObserverAdapter.java (2)

62-65: LGTM! The onError method correctly forwards the error to the StreamHandler.

67-70: LGTM! The onCompleted method correctly signals the completion of the stream.

test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/GreeterImpl.java (1)

116-123: LGTM! The server streaming method sayHelloServerStream is implemented correctly.

test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/TripleStubStreamTest.java (1)

156-167: Verify correct message ordering in server streaming.

The test checks messages received from server streaming but does not verify the order of messages, which is crucial for certain applications. Consider adding checks to ensure that messages are received in the correct order.

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/transport/triple/TripleClientInvoker.java (3)

154-154: Handle unknown streaming call types more gracefully.

This change logs a warning and returns a default response instead of throwing an exception, which might be more graceful in a production environment.

187-221: Implementation of bidirectional streaming looks robust.

This method correctly sets up and handles bidirectional streaming, aligning with the PR's objectives for enhanced streaming capabilities.

475-477: Refactor to simplify the mapping of call types.

This refactoring uses Optional to simplify the control flow, making the code cleaner and easier to read.

.../remoting-triple/src/main/java/com/alipay/sofa/rpc/transport/triple/TripleClientInvoker.java

.../src/main/java/com/alipay/sofa/rpc/message/triple/stream/ResponseSerializeStreamHandler.java

...ple/src/main/java/com/alipay/sofa/rpc/message/triple/stream/ClientStreamObserverAdapter.java

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

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

...ting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/GenericServiceImpl.java

coderabbitai

Actionable comments posted: 8

Review Details

Configuration used: CodeRabbit UI
Review profile: CHILL

Commits

Files that changed from the base of the PR and between 3d00d9d and 9a7ab52.

Files selected for processing (19)

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/message/triple/stream/ClientStreamObserverAdapter.java (1 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/message/triple/stream/ResponseSerializeStreamHandler.java (1 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/GenericServiceImpl.java (5 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/TripleServer.java (5 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/UniqueIdInvoker.java (3 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/transport/triple/TripleClientInvoker.java (9 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/utils/SofaProtoUtils.java (2 hunks)
remoting/remoting-triple/src/test/java/com/alipay/sofa/rpc/server/triple/GenericServiceImplTest.java (2 hunks)
test/test-integration/src/main/proto/helloworld.proto (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/GreeterImpl.java (2 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/NativeGrpcGreeterImpl.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ClientRequest.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ExtendClientRequest.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ExtendServerResponse.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/HelloService.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/HelloServiceImpl.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ServerResponse.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/TripleGenericStreamTest.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/TripleStubStreamTest.java (1 hunks)

Files skipped from review as they are similar to previous changes (14)

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/message/triple/stream/ClientStreamObserverAdapter.java
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/message/triple/stream/ResponseSerializeStreamHandler.java
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/TripleServer.java
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/UniqueIdInvoker.java
remoting/remoting-triple/src/test/java/com/alipay/sofa/rpc/server/triple/GenericServiceImplTest.java
test/test-integration/src/main/proto/helloworld.proto
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/GreeterImpl.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ClientRequest.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ExtendClientRequest.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ExtendServerResponse.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/HelloService.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/HelloServiceImpl.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ServerResponse.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/TripleGenericStreamTest.java

Additional comments not posted (14)

test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/NativeGrpcGreeterImpl.java (2)

73-76: Potential duplicate responses per request.

Verify if sending two responses per request in sayHelloBinary is intended. If not, consider sending only one response or adjusting the logic accordingly.

119-126: Implementation of server streaming looks correct.

The method sayHelloServerStream correctly implements the server streaming pattern by sending multiple responses.

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

82-94: Proper handling of call type mapping.

The method mapGrpcCallType correctly maps custom call types to gRPC method types and handles unsupported types with an exception.

97-107: Efficient caching of method stream types.

The method cacheStreamCallType efficiently determines and caches stream call types for methods, ensuring that only valid types are stored.

115-135: Correct identification of stream call types.

The method mapStreamType accurately identifies different streaming call types based on method signatures and handles errors appropriately.

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/GenericServiceImpl.java (3)

68-73: Proper handling of method invocation for unary calls.

The method generic correctly handles unary call invocations, including class loader management and exception handling.

98-167: Efficient handling of bidirectional streaming.

The method genericBiStream efficiently handles bidirectional streaming calls with proper synchronization for initialization and robust error management.

169-196: Correct implementation of server streaming.

The method genericServerStream correctly implements server streaming functionality, including parameter setup, error handling, and class loader management.

test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/TripleStubStreamTest.java (5)

124-150: Potential mismatch in expected response count.

Verify if the expected count of responses in testTripleStubBiStream matches the actual responses sent by the server. Adjust the CountDownLatch or the test logic as necessary to prevent test hangs.

155-182: Verify response message format in client streaming test.

Ensure that the response message in testTripleStubClientStream correctly reflects the number of requests received. Adjust the test or server logic if there's a mismatch.

185-195: Correct testing of server streaming functionality.

The method testTripleStubServerStream correctly tests the server streaming functionality with appropriate synchronization and response validation.

199-289: Comprehensive testing of protocol interactions.

The method testGrpcClientToTripleServer comprehensively tests interactions between a gRPC client and a Triple server across various streaming scenarios, ensuring correct behavior and synchronization.

291-361: Effective testing of client-server interactions across protocols.

The method testTripleClientToGrpcServer effectively tests interactions between a Triple client and a gRPC server, covering various streaming types and ensuring synchronization and correct behavior.

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/transport/triple/TripleClientInvoker.java (1)

449-465: The getMethodDescriptor method constructs a MethodDescriptor for gRPC calls. It's crucial to ensure that the method correctly handles all types of methods, especially new streaming types. Also, consider enhancing the method to handle more complex scenarios as streaming evolves.

.../remoting-triple/src/main/java/com/alipay/sofa/rpc/transport/triple/TripleClientInvoker.java

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

coderabbitai

Actionable comments posted: 3

Review Details

Configuration used: CodeRabbit UI
Review profile: CHILL

Commits

Files that changed from the base of the PR and between 9a7ab52 and d222c55.

Files selected for processing (19)

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/message/triple/stream/ClientStreamObserverAdapter.java (1 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/message/triple/stream/ResponseSerializeStreamHandler.java (1 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/GenericServiceImpl.java (5 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/TripleServer.java (5 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/UniqueIdInvoker.java (3 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/transport/triple/TripleClientInvoker.java (9 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/utils/SofaProtoUtils.java (2 hunks)
remoting/remoting-triple/src/test/java/com/alipay/sofa/rpc/server/triple/GenericServiceImplTest.java (2 hunks)
test/test-integration/src/main/proto/helloworld.proto (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/GreeterImpl.java (2 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/NativeGrpcGreeterImpl.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ClientRequest.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ExtendClientRequest.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ExtendServerResponse.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/HelloService.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/HelloServiceImpl.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ServerResponse.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/TripleGenericStreamTest.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/TripleStubStreamTest.java (1 hunks)

Files skipped from review as they are similar to previous changes (16)

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/message/triple/stream/ClientStreamObserverAdapter.java
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/message/triple/stream/ResponseSerializeStreamHandler.java
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/TripleServer.java
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/UniqueIdInvoker.java
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/utils/SofaProtoUtils.java
remoting/remoting-triple/src/test/java/com/alipay/sofa/rpc/server/triple/GenericServiceImplTest.java
test/test-integration/src/main/proto/helloworld.proto
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/NativeGrpcGreeterImpl.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ClientRequest.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ExtendClientRequest.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ExtendServerResponse.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/HelloService.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/HelloServiceImpl.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ServerResponse.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/TripleGenericStreamTest.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/TripleStubStreamTest.java

Additional comments not posted (8)

test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/GreeterImpl.java (1)

117-123: Implementation of server streaming looks good.

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/GenericServiceImpl.java (3)

68-73: Proper handling of class loader and error management in unary call handling.

98-167: Well-implemented bidirectional streaming with robust error handling and resource management.

169-196: Server streaming implementation is robust with appropriate error handling and class loader management.

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/transport/triple/TripleClientInvoker.java (4)

142-156: Effective management of different streaming call types with appropriate error handling.

168-185: Proper handling of unary gRPC calls with correct serialization and response processing.

187-221: Bidirectional streaming setup is robust with proper message handling and exception management.

227-240: Server streaming call handling is correctly implemented with appropriate setup and response management.

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

EvenLjj

LGTM

coderabbitai

Actionable comments posted: 12

Out of diff range and nitpick comments (3)

core/api/src/main/java/com/alipay/sofa/rpc/transport/SofaStreamObserver.java (2)

47-47: Ensure proper documentation for onError method.

The documentation for the onError method could be improved by explaining the expected behavior in more detail, especially regarding the state of the stream after an error occurs.

47-47: Consider enhancing the documentation for onError.

It would be beneficial to include more details about how errors should be handled and the state of the stream after an error is reported.
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/GenericServiceImpl.java (1)
Line range hint 73-86: Improve exception handling in generic method.

The method should catch specific exceptions such as SerializationException and handle them appropriately, instead of throwing a generic SofaRpcRuntimeException.
+ } catch (SerializationException se) {
+     LOGGER.error("Serialization error", se);
+     throw new SofaRpcException("Error during serialization", se);
+ }

Review Details

Configuration used: CodeRabbit UI
Review profile: CHILL

Commits

Files that changed from the base of the PR and between d222c55 and c7435df.

Files selected for processing (20)

core/api/src/main/java/com/alipay/sofa/rpc/transport/SofaStreamObserver.java (1 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/message/triple/stream/ClientStreamObserverAdapter.java (1 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/message/triple/stream/ResponseSerializeSofaStreamObserver.java (1 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/GenericServiceImpl.java (5 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/TripleServer.java (5 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/UniqueIdInvoker.java (3 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/transport/triple/TripleClientInvoker.java (9 hunks)
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/utils/SofaProtoUtils.java (2 hunks)
remoting/remoting-triple/src/test/java/com/alipay/sofa/rpc/server/triple/GenericServiceImplTest.java (2 hunks)
test/test-integration/src/main/proto/helloworld.proto (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/GreeterImpl.java (2 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/NativeGrpcGreeterImpl.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ClientRequest.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ExtendClientRequest.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ExtendServerResponse.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/HelloService.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/HelloServiceImpl.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ServerResponse.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/TripleGenericStreamTest.java (1 hunks)
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/TripleStubStreamTest.java (1 hunks)

Files skipped from review as they are similar to previous changes (16)

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/message/triple/stream/ClientStreamObserverAdapter.java
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/TripleServer.java
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/UniqueIdInvoker.java
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/utils/SofaProtoUtils.java
remoting/remoting-triple/src/test/java/com/alipay/sofa/rpc/server/triple/GenericServiceImplTest.java
test/test-integration/src/main/proto/helloworld.proto
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/GreeterImpl.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/NativeGrpcGreeterImpl.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ClientRequest.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ExtendClientRequest.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ExtendServerResponse.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/HelloService.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/HelloServiceImpl.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/ServerResponse.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/TripleGenericStreamTest.java
test/test-integration/src/test/java/com/alipay/sofa/rpc/test/triple/stream/TripleStubStreamTest.java

Additional comments not posted (8)

core/api/src/main/java/com/alipay/sofa/rpc/transport/SofaStreamObserver.java (2)

28-28: Method onNext implementation looks good.

36-36: Method onCompleted implementation is correct.
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/message/triple/stream/ResponseSerializeSofaStreamObserver.java (2)
59-59: Method onCompleted correctly delegates to the underlying observer.

63-65: Verify the implementation of TripleExceptionUtils.asStatusRuntimeException.

Verification successful

The implementation of TripleExceptionUtils.asStatusRuntimeException correctly handles both null and non-null exceptions, converting them into StatusRuntimeException with appropriate descriptions and causes. This method is well-suited for its usage in the onError method of ResponseSerializeSofaStreamObserver, ensuring that all types of exceptions are properly formatted for gRPC communication.
Scripts Executed

The following scripts were executed for the analysis:

Script:
#!/bin/bash
# Description: Verify the implementation of `TripleExceptionUtils.asStatusRuntimeException` to ensure it handles all exception types correctly.

# Test: Search for the method implementation. Expect: Proper exception handling logic.
rg --type java 'asStatusRuntimeException' -- $(fd TripleExceptionUtils.java)
Length of output: 248

Script:
#!/bin/bash
# Description: Fetch the complete implementation of `asStatusRuntimeException` to verify its exception handling capabilities.

# Using ast-grep to retrieve the method implementation
ast-grep --lang java --pattern $'public static StatusRuntimeException asStatusRuntimeException(Throwable $_) {
  $$$
}'
Length of output: 1150
remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/GenericServiceImpl.java (3)

170-197: Method genericServerStream is implemented correctly.

234-234: Method getAppResponse is implemented correctly.

260-273: Method getArgTypes is implemented correctly.

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/transport/triple/TripleClientInvoker.java (1)

157-165: Method stubCall is implemented correctly.

...main/java/com/alipay/sofa/rpc/message/triple/stream/ResponseSerializeSofaStreamObserver.java

...ting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/GenericServiceImpl.java

.../remoting-triple/src/main/java/com/alipay/sofa/rpc/transport/triple/TripleClientInvoker.java

sofastack-cla bot added cla:no Need sign CLA First-time contributor First-time contributor size/XXL cla:yes CLA is ok and removed cla:no Need sign CLA labels Aug 25, 2023

EvenLjj added the later This will be worked on in later version label Oct 23, 2023

Add bi-stream call support for triple

df27427

EvenLjj force-pushed the triple-stream-support branch from 0a14a70 to df27427 Compare March 27, 2024 11:28

coderabbitai bot reviewed Mar 27, 2024

View reviewed changes

coderabbitai bot reviewed Apr 15, 2024

View reviewed changes

EvenLjj force-pushed the triple-stream-support branch from 837b208 to 0103a31 Compare April 15, 2024 08:46

coderabbitai bot reviewed Apr 15, 2024

View reviewed changes

.../remoting-triple/src/main/java/com/alipay/sofa/rpc/transport/triple/TripleClientInvoker.java Outdated Show resolved Hide resolved

.../remoting-triple/src/main/java/com/alipay/sofa/rpc/transport/triple/TripleClientInvoker.java Outdated Show resolved Hide resolved

Lo1nt reviewed Apr 15, 2024

View reviewed changes

EvenLjj force-pushed the triple-stream-support branch from bd169a3 to 266006d Compare April 22, 2024 02:12

coderabbitai bot reviewed May 7, 2024

View reviewed changes

.../remoting-triple/src/main/java/com/alipay/sofa/rpc/transport/triple/TripleClientInvoker.java Outdated Show resolved Hide resolved

.../remoting-triple/src/main/java/com/alipay/sofa/rpc/transport/triple/TripleClientInvoker.java Outdated Show resolved Hide resolved

EvenLjj force-pushed the triple-stream-support branch from 437ff33 to 3d00d9d Compare May 8, 2024 04:13

coderabbitai bot reviewed May 8, 2024

View reviewed changes

EvenLjj force-pushed the triple-stream-support branch from 3d00d9d to 9a7ab52 Compare May 9, 2024 09:36

coderabbitai bot reviewed May 9, 2024

View reviewed changes

EvenLjj force-pushed the triple-stream-support branch from 9a7ab52 to d222c55 Compare May 9, 2024 09:49

coderabbitai bot reviewed May 9, 2024

View reviewed changes

Lo1nt approved these changes May 10, 2024

View reviewed changes

EvenLjj approved these changes May 10, 2024

View reviewed changes

triple pojo mode support stream

c7435df

EvenLjj force-pushed the triple-stream-support branch from d222c55 to c7435df Compare May 13, 2024 15:33

coderabbitai bot reviewed May 13, 2024

View reviewed changes

EvenLjj merged commit 56ed259 into sofastack:master May 14, 2024
4 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Triple stream support (issue#1332) #1360

Triple stream support (issue#1332) #1360

namelessssssssssss commented Aug 25, 2023 •

edited by coderabbitai bot

Loading

sofastack-cla bot commented Aug 25, 2023

namelessssssssssss commented Aug 25, 2023

codecov bot commented Aug 25, 2023 •

edited

Loading

nobodyiam commented Aug 26, 2023

OrezzerO commented Nov 12, 2023

coderabbitai bot commented Mar 27, 2024 •

edited

Loading

Chat

CodeRabbit Commands (invoked as PR comments)

CodeRabbit Configration File (`.coderabbit.yaml`)

Documentation and Community

coderabbitai bot left a comment

coderabbitai bot left a comment

coderabbitai bot left a comment

coderabbitai bot left a comment

coderabbitai bot left a comment

coderabbitai bot left a comment

coderabbitai bot left a comment

EvenLjj left a comment

coderabbitai bot left a comment

Triple stream support (issue#1332) #1360

Triple stream support (issue#1332) #1360

Conversation

namelessssssssssss commented Aug 25, 2023 • edited by coderabbitai bot Loading

Motivation:

Modification:

Change log (Chinese) :

概述

业务场景

流传输模型

流传输接口定义

全局处理

协议处理

Triple协议

泛型调用

IDL修改

Triple传输层修改

客户端

服务端

流信息序列化问题

Summary by CodeRabbit

sofastack-cla bot commented Aug 25, 2023

namelessssssssssss commented Aug 25, 2023

codecov bot commented Aug 25, 2023 • edited Loading

Codecov Report

nobodyiam commented Aug 26, 2023

OrezzerO commented Nov 12, 2023

coderabbitai bot commented Mar 27, 2024 • edited Loading

Walkthrough

Changes

Chat

CodeRabbit Commands (invoked as PR comments)

CodeRabbit Configration File (.coderabbit.yaml)

Documentation and Community

coderabbitai bot left a comment

Choose a reason for hiding this comment

coderabbitai bot left a comment

Choose a reason for hiding this comment

coderabbitai bot left a comment

Choose a reason for hiding this comment

coderabbitai bot left a comment

Choose a reason for hiding this comment

coderabbitai bot left a comment

Choose a reason for hiding this comment

coderabbitai bot left a comment

Choose a reason for hiding this comment

coderabbitai bot left a comment

Choose a reason for hiding this comment

EvenLjj left a comment

Choose a reason for hiding this comment

coderabbitai bot left a comment

Choose a reason for hiding this comment

namelessssssssssss commented Aug 25, 2023 •

edited by coderabbitai bot

Loading

codecov bot commented Aug 25, 2023 •

edited

Loading

coderabbitai bot commented Mar 27, 2024 •

edited

Loading

CodeRabbit Configration File (`.coderabbit.yaml`)