io.grpc.MethodDescriptor#Marshaller ( )源码实例Demo

ByteSizeMarshaller(MethodDescriptor.Marshaller<T> delegate) {
  this.delegate = delegate;
}
 

@Override
protected MethodDescriptor.Marshaller<Object> generateMarshaller(Type type) {
    return new JacksonMarshaller(type, objectMapper);
}
 

@Override
protected MethodDescriptor.Marshaller<Object> generateMarshaller(Type type) {
    return new ProtoMarshaller(type);
}
 

@Override
public MethodDescriptor.Marshaller<Object> emptyMarshaller() {
    return generateMarshaller(null);
}
 

/**
 * 获取返回类型的解析器
 *
 * @param methodCallProperty methodCallProperty
 * @return 解析器
 */
@Override
public MethodDescriptor.Marshaller<Object> parseReturnMarshaller(MethodCallProperty methodCallProperty) {
    Method method = methodCallProperty.getMethod();
    Type[] types;
    boolean existParam;
    switch (methodCallProperty.getMethodType()) {
        case UNARY:
            if (method.getReturnType() == ListenableFuture.class) {
                //等于ClientCalls.futureUnaryCall()
                //获取ListenableFuture的泛型
                types = Utils.reflectMethodReturnTypes(method);
                return generateMarshaller(Utils.safeElement(types, 0));
            } else if (method.getReturnType().getName().equals("void")) {
                //判断参数中是否存在StreamObserver泛型
                //参数中为StreamObserver的泛型
                types = Utils.reflectMethodParameterTypes(method, StreamObserver.class);
                //说明参数中不含有StreamObserver参数
                if (types == null) {
                    //返回普通方式
                    return generateMarshaller(method.getGenericReturnType());
                }
                //存在，相当于ClientCalls.asyncUnaryCall();
                //检查是否存在两个参数，一个为业务参数，一个为StreamObserver，并且顺序一致
                checkTwoParamHasStreamObServer(methodCallProperty);
                return generateMarshaller(Utils.safeElement(types, 0));
            }
            //直接返回方法的返回类型，等于ClientCalls.blockingUnaryCall
            return generateMarshaller(method.getGenericReturnType());
        case BIDI_STREAMING://双向流，相当于asyncBidiStreamingCall
            //检查是否存在一个参数，为StreamObserver
            checkOneParamHasStreamObServer(methodCallProperty);
            //判断方法返回类型是否为StreamObserver（此为客户端传输数据所用，服务端响应在参数的StreamObserver中）
            if (method.getReturnType() != StreamObserver.class) {
                throw new GRpcMethodNoMatchException(method.getDeclaringClass().getName(), method.getName(), methodCallProperty.getMethodType().name(),
                        "You should use [StreamObserver] for your return value.Please check it.");
            }
            //检验参数是否为StreamObserver，获取服务端响应泛型
            existParam = Utils.checkMethodHasParamClass(method, StreamObserver.class);
            if (!existParam) {
                throw new GRpcMethodNoMatchException(method.getDeclaringClass().getName(), method.getName(), methodCallProperty.getMethodType().name(),
                        "You should use [StreamObserver] for your param value.Please check it.");
            }
            //获取返回类型的泛型
            types = Utils.reflectMethodReturnTypes(method);
            return generateMarshaller(Utils.safeElement(types, 0));
        case CLIENT_STREAMING: //客户端流。等于ClientCalls.asyncClientStreamingCall()
            //检查是否存在一个参数，为StreamObserver
            checkOneParamHasStreamObServer(methodCallProperty);
            //方法返回类型不为空时，必须为StreamObserver，检验（此为客户端传输数据所用，服务端响应在参数的StreamObserver中）
            if (!method.getReturnType().getName().equals("void") && method.getReturnType() != StreamObserver.class) {
                throw new GRpcMethodNoMatchException(method.getDeclaringClass().getName(), method.getName(), methodCallProperty.getMethodType().name(),
                        "You should use [StreamObserver] for your return value.Please check it.");
            }
            //检验参数是否为StreamObserver，获取服务端响应泛型
            existParam = Utils.checkMethodHasParamClass(method, StreamObserver.class);
            if (!existParam) {
                throw new GRpcMethodNoMatchException(method.getDeclaringClass().getName(), method.getName(), methodCallProperty.getMethodType().name(),
                        "You should use [StreamObserver] for your param value.Please check it.");
            }
            //获取返回类型的泛型
            types = Utils.reflectMethodReturnTypes(method);
            return generateMarshaller(Utils.safeElement(types, 0));
        case SERVER_STREAMING://等于ClientCalls.blockingServerStreamingCall
            if (method.getReturnType() != Iterator.class) {
                throw new GRpcMethodNoMatchException(method.getDeclaringClass().getName(), method.getName(), methodCallProperty.getMethodType().name(),
                        "You should use [Iterator] for your return value.Please check it.");
            }
            //获取返回类型的泛型
            types = Utils.reflectMethodReturnTypes(method);
            return generateMarshaller(Utils.safeElement(types, 0));
    }
    throw new GRpcMethodNoMatchException(method.getDeclaringClass().getName(), method.getName(), methodCallProperty.getMethodType().name(),
            "Return value type no match.Please check your configuration.");
}
 

/**
 * 获取请求类型的解析器
 *
 * @param methodCallProperty methodCallProperty
 * @return 解析器
 */
@Override
public MethodDescriptor.Marshaller<Object> parseRequestMarshaller(MethodCallProperty methodCallProperty) {
    Method method = methodCallProperty.getMethod();
    Type[] types;
    switch (methodCallProperty.getMethodType()) {
        case UNARY: //一对一，等于asyncUnaryCall()
            //检验是否两个参数，包含StreamObserver，并且顺序一致
            if (method.getGenericParameterTypes().length == 2) {
                checkTwoParamHasStreamObServer(methodCallProperty);
            }
            //获取获取请求参数类型，第一个为业务实体
            types = method.getGenericParameterTypes();
            return generateMarshaller(Utils.safeElement(types, 0));
        case BIDI_STREAMING://双向流，等于asyncBidiStreamingCall()
            //检验是否一个参数，为StreamObserver
            checkOneParamHasStreamObServer(methodCallProperty);
            //检查方法的返回值是否为StreamObserver类型
            if (method.getReturnType() != StreamObserver.class) {
                throw new GRpcMethodNoMatchException(method.getDeclaringClass().getName(), method.getName(), methodCallProperty.getMethodType().name(),
                        "You should use [StreamObserver] for your return value.Please check it.");
            }
            //获取方法参数类型为StreamObserver的泛型
            types = Utils.reflectMethodParameterTypes(method, StreamObserver.class);
            if (types == null) {
                throw new GRpcMethodNoMatchException(method.getDeclaringClass().getName(), method.getName(), methodCallProperty.getMethodType().name(),
                        "You should use [StreamObserver] for your param value.Please check it.");
            }
            //获取返回类型的泛型
            types = Utils.reflectMethodReturnTypes(method);
            return generateMarshaller(Utils.safeElement(types, 0));
        case CLIENT_STREAMING: //客户端流。等于asyncClientStreamingCall()
            //检查方法的返回值是否为StreamObserver类型
            if (method.getReturnType() != StreamObserver.class) {
                throw new GRpcMethodNoMatchException(method.getDeclaringClass().getName(), method.getName(), methodCallProperty.getMethodType().name(),
                        "You should use [StreamObserver] for your return value.Please check it.");
            }
            //检验返回流StreamObserver是否存在，并且唯一。
            checkOneParamHasStreamObServer(methodCallProperty);
            //获取返回类型的泛型，为请求参数的泛型
            types = Utils.reflectMethodReturnTypes(method);
            return generateMarshaller(Utils.safeElement(types, 0));
        case SERVER_STREAMING://等于asyncServerStreamingCall()
            //检验是否两个参数，并且顺序一致
            if (method.getGenericParameterTypes().length == 2) {
                checkTwoParamHasStreamObServer(methodCallProperty);
            } else {
                checkOneParamHasStreamObServer(methodCallProperty);
            }
            //获取获取请求参数类型，第一个为业务实体
            types = method.getGenericParameterTypes();
            return generateMarshaller(Utils.safeElement(types, 0));
    }
    throw new GRpcMethodNoMatchException(method.getDeclaringClass().getName(), method.getName(), methodCallProperty.getMethodType().name(),
            "Request value type no match.Please check your configuration.");
}
 

@Test
@SuppressWarnings("unchecked")
public void testInterceptors() {
    String requestDescriptor = "createStream-myScope-myStream";
    long requestId = 1234L;
    ClientInterceptor clientInterceptor = RPCTracingHelpers.getClientInterceptor();
    RequestTracker requestTracker = new RequestTracker(true);

    // Mocking RPC elements.
    MethodDescriptor.Marshaller<Object> mockMarshaller = Mockito.mock(MethodDescriptor.Marshaller.class);
    ClientCall.Listener<Object> listener = Mockito.mock(ClientCall.Listener.class);
    ServerCall serverCall = Mockito.mock(ServerCall.class);
    ServerCallHandler serverCallHandler = Mockito.mock(ServerCallHandler.class);
    ManagedChannel channel = NettyChannelBuilder.forTarget("localhost").build();
    MethodDescriptor method = MethodDescriptor.newBuilder()
                                              .setFullMethodName("createStream")
                                              .setType(MethodDescriptor.MethodType.UNARY)
                                              .setRequestMarshaller(mockMarshaller)
                                              .setResponseMarshaller(mockMarshaller)
                                              .build();
    Mockito.when(serverCall.getMethodDescriptor()).thenReturn(method);

    // Actual elements to work with.
    CallOptions callOptions = CallOptions.DEFAULT;
    Metadata headers = new Metadata();

    // Test that headers do not contain tracing-related key/values, as call options are not set.
    clientInterceptor.interceptCall(method, callOptions, channel).start(listener, headers);
    assertFalse(headers.containsKey(RPCTracingHelpers.DESCRIPTOR_HEADER));
    assertFalse(headers.containsKey(RPCTracingHelpers.ID_HEADER));

    // Check that the server interceptor handles clients not sending tracing headers and that the cache remains clean.
    ServerInterceptor serverInterceptor = RPCTracingHelpers.getServerInterceptor(requestTracker);
    serverInterceptor.interceptCall(serverCall, headers, serverCallHandler);
    assertEquals(0, requestTracker.getNumDescriptors());

    // Add call options and check that headers are correctly set.
    callOptions = callOptions.withOption(RPCTracingHelpers.REQUEST_DESCRIPTOR_CALL_OPTION, requestDescriptor)
                             .withOption(RPCTracingHelpers.REQUEST_ID_CALL_OPTION, String.valueOf(requestId));
    clientInterceptor.interceptCall(method, callOptions, channel).start(listener, headers);
    assertEquals(requestDescriptor, headers.get(RPCTracingHelpers.DESCRIPTOR_HEADER));
    assertEquals(requestId, Long.parseLong(headers.get(RPCTracingHelpers.ID_HEADER)));

    // Test that the server interceptor correctly sets these headers in the cache for further tracking.
    serverInterceptor.interceptCall(serverCall, headers, serverCallHandler);
    assertEquals(1, requestTracker.getNumDescriptors());
    assertEquals(requestId, requestTracker.getRequestIdFor(requestDescriptor));
}
 

ByteSizeMarshaller(MethodDescriptor.Marshaller<T> delegate) {
  this.delegate = delegate;
}
 

/**
 * Creates a new marshaller that does nothing.
 *
 * @since 1.1.0
 */
@ExperimentalApi("https://github.com/grpc/grpc-java/issues/2600")
public static MethodDescriptor.Marshaller<Void> voidMarshaller() {
  return new NoopMarshaller();
}
 

/**
 * 生成装配器
 *
 * @param type 泛型类型
 * @return 装配器
 */
protected abstract MethodDescriptor.Marshaller<Object> generateMarshaller(Type type);
 

/**
 * Creates a new marshaller that does nothing.
 *
 * @since 1.1.0
 */
@ExperimentalApi("https://github.com/grpc/grpc-java/issues/2600")
public static MethodDescriptor.Marshaller<Void> voidMarshaller() {
  return new NoopMarshaller();
}
 

abstract MethodDescriptor.Marshaller<T> delegate(); 

MethodDescriptor.Marshaller<Object> emptyMarshaller(); 

MethodDescriptor.Marshaller<Object> parseReturnMarshaller(MethodCallProperty methodCallProperty); 

MethodDescriptor.Marshaller<Object> parseRequestMarshaller(MethodCallProperty methodCallProperty); 

abstract MethodDescriptor.Marshaller<T> delegate();