反射服务

调用一个gRPC服务需要客户端持有服务定义的proto文件, 通过proto文件生成本地调用的stub. 如何在没有proto定义的情况下也能调用服务端提供的服务呢. gRPC 提供了 grpc.reflection.v1alpha.ServerReflection 服务，在 Server 端添加后可以通过该服务获取所有服务的信息，包括服务定义，方法，属性等； 根据获取到的服务信息实现泛化调用

// Copyright 2016 The gRPC Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// Service exported by server reflection
// Warning: this entire file is deprecated. Use this instead:
// https://github.com/grpc/grpc-proto/blob/master/grpc/reflection/v1/reflection.proto
syntax = "proto3";
package grpc.reflection.v1alpha;
option deprecated = true;
option java_multiple_files = true;
option java_package = "io.grpc.reflection.v1alpha";
option java_outer_classname = "ServerReflectionProto";
service ServerReflection {
  // The reflection service is structured as a bidirectional stream, ensuring
  // all related requests go to a single server.
  rpc ServerReflectionInfo(stream ServerReflectionRequest)
      returns (stream ServerReflectionResponse);
}
// The message sent by the client when calling ServerReflectionInfo method.
message ServerReflectionRequest {
  string host = 1;
  // To use reflection service, the client should set one of the following
  // fields in message_request. The server distinguishes requests by their
  // defined field and then handles them using corresponding methods.
  oneof message_request {
    // Find a proto file by the file name.
    string file_by_filename = 3;
    // Find the proto file that declares the given fully-qualified symbol name.
    // This field should be a fully-qualified symbol name
    // (e.g. <package>.<service>[.<method>] or <package>.<type>).
    string file_containing_symbol = 4;
    // Find the proto file which defines an extension extending the given
    // message type with the given field number.
    ExtensionRequest file_containing_extension = 5;
    // Finds the tag numbers used by all known extensions of extendee_type, and
    // appends them to ExtensionNumberResponse in an undefined order.
    // Its corresponding method is best-effort: it's not guaranteed that the
    // reflection service will implement this method, and it's not guaranteed
    // that this method will provide all extensions. Returns
    // StatusCode::UNIMPLEMENTED if it's not implemented.
    // This field should be a fully-qualified type name. The format is
    // <package>.<type>
    string all_extension_numbers_of_type = 6;
    // List the full names of registered services. The content will not be
    // checked.
    string list_services = 7;
  }
}
// The type name and extension number sent by the client when requesting
// file_containing_extension.
message ExtensionRequest {
  // Fully-qualified type name. The format should be <package>.<type>
  string containing_type = 1;
  int32 extension_number = 2;
}
// The message sent by the server to answer ServerReflectionInfo method.
message ServerReflectionResponse {
  string valid_host = 1;
  ServerReflectionRequest original_request = 2;
  // The server set one of the following fields accroding to the message_request
  // in the request.
  oneof message_response {
    // This message is used to answer file_by_filename, file_containing_symbol,
    // file_containing_extension requests with transitive dependencies. As
    // the repeated label is not allowed in oneof fields, we use a
    // FileDescriptorResponse message to encapsulate the repeated fields.
    // The reflection service is allowed to avoid sending FileDescriptorProtos
    // that were previously sent in response to earlier requests in the stream.
    FileDescriptorResponse file_descriptor_response = 4;
    // This message is used to answer all_extension_numbers_of_type requst.
    ExtensionNumberResponse all_extension_numbers_response = 5;
    // This message is used to answer list_services request.
    ListServiceResponse list_services_response = 6;
    // This message is used when an error occurs.
    ErrorResponse error_response = 7;
  }
}
// Serialized FileDescriptorProto messages sent by the server answering
// a file_by_filename, file_containing_symbol, or file_containing_extension
// request.
message FileDescriptorResponse {
  // Serialized FileDescriptorProto messages. We avoid taking a dependency on
  // descriptor.proto, which uses proto2 only features, by making them opaque
  // bytes instead.
  repeated bytes file_descriptor_proto = 1;
}
// A list of extension numbers sent by the server answering
// all_extension_numbers_of_type request.
message ExtensionNumberResponse {
  // Full name of the base type, including the package name. The format
  // is <package>.<type>
  string base_type_name = 1;
  repeated int32 extension_number = 2;
}
// A list of ServiceResponse sent by the server answering list_services request.
message ListServiceResponse {
  // The information of each service may be expanded in the future, so we use
  // ServiceResponse message to encapsulate it.
  repeated ServiceResponse service = 1;
}
// The information of a single service used by ListServiceResponse to answer
// list_services request.
message ServiceResponse {
  // Full name of a registered service, including its package name. The format
  // is <package>.<service>
  string name = 1;
}
// The error code and error message sent by the server when an error occurs.
message ErrorResponse {
  // This field uses the error codes defined in grpc::StatusCode.
  int32 error_code = 1;
  string error_message = 2;
}

该服务只有一个双向流的方法 ServerReflectionInfo，调用时根据请求参数不同，调用不同的方法进行处理，并返回响应；该方法的流控是非自动的，只有当一个请求完成之后才会获取下一个请求

使用方式

添加maven依赖:

...
<dependency>
  <groupId>io.grpc</groupId>
  <artifactId>grpc-services</artifactId>
  <version>${VERSION}</version>
</dependency>
...

Server端:

public static void main(String[] args) {
    // 构建 Server
    int port = 50051;
    server = ServerBuilder.forPort(port)
        .addService(new GreeterImpl())
        // 添加反射服务
        .addService(ProtoReflectionService.newInstance())
        .build()
        .start();
    logger.info("Server started, listening on " + port);
    Runtime.getRuntime().addShutdownHook(new Thread() {
        @Override
        public void run() {
            // Use stderr here since the logger may have been reset by its JVM shutdown hook.
            System.err.println("*** shutting down gRPC server since JVM is shutting down");
            try {
                HelloWorldServer.this.stop();
            } catch (InterruptedException e) {
                e.printStackTrace(System.err);
            }
            System.err.println("*** server shut down");
        }
    });
    server.awaitTermination();
}

通过反射服务动态调用

凡事涉及到反射, 处理过程都是很复杂的, 以java为例, 处理流程如下:

1. 使用完整的服务名, 通过反射服务获取服务的proto定义, java中以 DescriptorProtos.FileDescriptorSet 对象描述
2. 通过获取到的 DescriptorProtos.FileDescriptorSet 解析出方法描述, java中以 Descriptors.MethodDescriptor 对象描述
3. 根据方法描述, 创建动态message builder, DynamicMessage.newBuilder(methodDescriptor.getInputType())
4. 使用 JsonFormat.parser().merge(“json对象”, messageBuilder), 将json格式的参数转换成protobuf message对象
5. 根据方法的类型 (Unary, Server streaming, Client streaming, Bidirectional streaming), 调用不同的 ClientCalls方法(同步/异步)
   1. ClientCalls.asyncUnaryCall
   2. ClientCalls.asyncServerStreamingCall
   3. ClientCalls.asyncClientStreamingCall
   4. ClientCalls.asyncBidiStreamingCall
   5. ClientCalls.blockingUnaryCall
   6. ClientCalls.blockingServerStreamingCall
6. 通过JsonFormat.printer().print(response), 输出为json

完全示例代码如下:

import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import com.google.protobuf.*;
import com.google.protobuf.util.JsonFormat;
import io.grpc.*;
import io.grpc.reflection.v1alpha.ServerReflectionGrpc;
import io.grpc.reflection.v1alpha.ServerReflectionRequest;
import io.grpc.reflection.v1alpha.ServerReflectionResponse;
import io.grpc.reflection.v1alpha.ServiceResponse;
import io.grpc.stub.StreamObserver;

import java.io.IOException;
import java.io.InputStream;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

import static io.grpc.MethodDescriptor.generateFullMethodName;
import static io.grpc.stub.ClientCalls.*;

public class HelloWorldReflectionClient {
    private static final Logger logger = Logger.getLogger(HelloWorldClient.class.getName());

    private final ServerReflectionGrpc.ServerReflectionStub stub;

    public HelloWorldReflectionClient(Channel channel) {
        this.stub = ServerReflectionGrpc.newStub(channel);
    }

    /**
     * 获取服务列表
     */
    public void listServices() {
        StreamObserver<ServerReflectionRequest> requestStreamObserver = stub.serverReflectionInfo(new StreamObserver<ServerReflectionResponse>() {

            @Override
            public void onNext(ServerReflectionResponse value) {
                for (ServiceResponse serviceResponse : value.getListServicesResponse().getServiceList()) {
                    logger.log(Level.INFO, "reflection service: {0}", serviceResponse.getName());
                }
            }

            @Override
            public void onError(Throwable t) {

            }

            @Override
            public void onCompleted() {

            }
        });
        ServerReflectionRequest request = ServerReflectionRequest.newBuilder()
                .setListServices("")
                .build();
        requestStreamObserver.onNext(request);
        requestStreamObserver.onCompleted();
    }

    /**
     * 反射调用
     * @param serviceName 完整服务名
     * @param methodName 方法名
     * @param name
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public void reflectionCall(String serviceName, String methodName, String name) throws ExecutionException, InterruptedException {
        LookupServiceHandler                    rpcHandler            = new LookupServiceHandler(serviceName);
        StreamObserver<ServerReflectionRequest> requestStreamObserver = stub.serverReflectionInfo(rpcHandler);
        // 获取proto文件描述
        DescriptorProtos.FileDescriptorSet      fileDescriptorSet     = rpcHandler.start(requestStreamObserver).get();
        if (fileDescriptorSet == null) {
            logger.log(Level.WARNING, "Service: " + serviceName + " not found.");
            return;
        }
        ServiceResolver serviceResolver = ServiceResolver.fromFileDescriptorSet(fileDescriptorSet);
        // 通过方法名, 获取方法描述
        Descriptors.MethodDescriptor methodDescriptor = serviceResolver.resolveServiceMethod(serviceName, methodName);
        // 构建动态消息对象
        DynamicMessage.Builder       messageBuilder   = DynamicMessage.newBuilder(methodDescriptor.getInputType());
        Message message = null;
        try {
            // 通过json转换
            serviceResolver.getParser().merge("{\"name\": \""+name+"\"}", messageBuilder);
            message = messageBuilder.build();
        } catch (InvalidProtocolBufferException e) {
            logger.log(Level.WARNING, "invalid protocol buffer, ", e);
            return;
        }
        JsonFormat.Printer printer = JsonFormat.printer().usingTypeRegistry(serviceResolver.getRegistry());
        MethodDescriptor.MethodType methodType = fetchMethodType(methodDescriptor);
        StreamObserver<Message> responseStreamObserver = new StreamObserver<Message>() {
            @Override
            public void onNext(Message value) {
                try {
                    logger.info("Reflection call response: " + printer.print(value));
                } catch (InvalidProtocolBufferException e) {
                }
            }

            @Override
            public void onError(Throwable t) {

            }

            @Override
            public void onCompleted() {

            }
        };

        // 根据不同的方法类型, 进行调用
        switch (methodType) {
            case UNARY:
                asyncUnaryCall(stub.getChannel().newCall(createGrpcMethodDescriptor(methodDescriptor)
                        , CallOptions.DEFAULT), message, responseStreamObserver);
                break;
            case SERVER_STREAMING:
                asyncServerStreamingCall(stub.getChannel().newCall(createGrpcMethodDescriptor(methodDescriptor)
                        , CallOptions.DEFAULT), message, responseStreamObserver);
                break;
            case CLIENT_STREAMING:
                StreamObserver<Message> reqStreamObserver = asyncClientStreamingCall(stub.getChannel().newCall(createGrpcMethodDescriptor(methodDescriptor)
                        , CallOptions.DEFAULT), responseStreamObserver);
                reqStreamObserver.onNext(message);
                reqStreamObserver.onCompleted();
            case BIDI_STREAMING:
                StreamObserver<Message> biReqStreamObserver = asyncBidiStreamingCall(stub.getChannel().newCall(createGrpcMethodDescriptor(methodDescriptor)
                        , CallOptions.DEFAULT), responseStreamObserver);
                biReqStreamObserver.onNext(message);
                biReqStreamObserver.onCompleted();
                break;
            default:
                logger.info("Unknown methodType: " + methodType);
        }

    }

    private io.grpc.MethodDescriptor<Message, Message> createGrpcMethodDescriptor(final Descriptors.MethodDescriptor descriptor) {
        return io.grpc.MethodDescriptor.<Message, Message>newBuilder()
                .setType(fetchMethodType(descriptor))
                .setFullMethodName(fetchFullMethodName(descriptor))
                .setRequestMarshaller(new DynamicMessageMarshaller(descriptor.getInputType()))
                .setResponseMarshaller(new DynamicMessageMarshaller(descriptor.getOutputType()))
                .build();
    }

    public String fetchFullMethodName(final Descriptors.MethodDescriptor methodDescriptor) {
        String serviceName = methodDescriptor.getService().getFullName();
        String methodName  = methodDescriptor.getName();
        return generateFullMethodName(serviceName, methodName);
    }

    public MethodDescriptor.MethodType fetchMethodType(final Descriptors.MethodDescriptor methodDescriptor) {
        boolean clientStreaming = methodDescriptor.toProto().getClientStreaming();
        boolean serverStreaming = methodDescriptor.toProto().getServerStreaming();
        if (clientStreaming && serverStreaming) {
            return MethodDescriptor.MethodType.BIDI_STREAMING;
        } else if (!clientStreaming && !serverStreaming) {
            return MethodDescriptor.MethodType.UNARY;
        } else if (!clientStreaming) {
            return MethodDescriptor.MethodType.SERVER_STREAMING;
        } else {
            return MethodDescriptor.MethodType.SERVER_STREAMING;
        }
    }

    public static void main(String[] args) throws InterruptedException {
        String target = "localhost:50051";
        ManagedChannel channel = ManagedChannelBuilder.forTarget(target)
                .usePlaintext()
                .build();
        try {
            HelloWorldReflectionClient client = new HelloWorldReflectionClient(channel);
            client.listServices();
            client.reflectionCall("helloworld.Greeter", "SayHello", "test reflection call");
            // 等待异步执行完成
            Thread.sleep(5000);
        } catch (ExecutionException e) {
            e.printStackTrace();
        } finally {
            // ManagedChannels use resources like threads and TCP connections. To prevent leaking these
            // resources the channel should be shut down when it will no longer be used. If it may be used
            // again leave it running.
            channel.shutdownNow().awaitTermination(5, TimeUnit.SECONDS);
        }
    }

    public static class LookupServiceHandler implements StreamObserver<ServerReflectionResponse> {

        private final String serviceName;

        private final Set<String> requestedDescriptors;

        private final SettableFuture<DescriptorProtos.FileDescriptorSet> resultFuture;

        private final Map<String, DescriptorProtos.FileDescriptorProto> resolvedDescriptors;

        private StreamObserver<ServerReflectionRequest> requestStream;

        private int outstandingRequests;

        public LookupServiceHandler(final String serviceName) {
            this.serviceName = serviceName;
            this.resultFuture = SettableFuture.create();
            this.resolvedDescriptors = new HashMap<>();
            this.requestedDescriptors = new HashSet<>();
            this.outstandingRequests = 0;
        }

        /**
         * Start the handler.
         *
         * @param requestStream stream
         * @return ListenableFuture future
         */
        public ListenableFuture<DescriptorProtos.FileDescriptorSet> start(final StreamObserver<ServerReflectionRequest> requestStream) {
            this.requestStream = requestStream;
            requestStream.onNext(requestForSymbol(serviceName));
            ++outstandingRequests;
            return resultFuture;
        }

        @Override
        public void onNext(final ServerReflectionResponse response) {
            ServerReflectionResponse.MessageResponseCase responseCase = response.getMessageResponseCase();
            if (responseCase == ServerReflectionResponse.MessageResponseCase.FILE_DESCRIPTOR_RESPONSE) {
                ImmutableSet<DescriptorProtos.FileDescriptorProto> descriptors =
                        parseDescriptors(response.getFileDescriptorResponse().getFileDescriptorProtoList());
                descriptors.forEach(d -> resolvedDescriptors.put(d.getName(), d));
                descriptors.forEach(this::processDependencies);
            }
        }

        @Override
        public void onError(final Throwable t) {
            resultFuture.setException(new RuntimeException("Reflection lookup rpc failed for: " + serviceName, t));
        }

        @Override
        public void onCompleted() {
            if (!resultFuture.isDone()) {
                resultFuture.setException(new RuntimeException("Unexpected end of rpc"));
            }
        }

        private ImmutableSet<DescriptorProtos.FileDescriptorProto> parseDescriptors(final List<ByteString> descriptorBytes) {
            ImmutableSet.Builder<DescriptorProtos.FileDescriptorProto> resultBuilder = ImmutableSet.builder();
            for (ByteString fileDescriptorBytes : descriptorBytes) {
                try {
                    resultBuilder.add(DescriptorProtos.FileDescriptorProto.parseFrom(fileDescriptorBytes));
                } catch (InvalidProtocolBufferException e) {
                    throw new RuntimeException(e);
                }
            }
            return resultBuilder.build();
        }

        private void processDependencies(final DescriptorProtos.FileDescriptorProto fileDescriptor) {
            fileDescriptor.getDependencyList().forEach(dep -> {
                if (!resolvedDescriptors.containsKey(dep) && !requestedDescriptors.contains(dep)) {
                    requestedDescriptors.add(dep);
                    ++outstandingRequests;
                    requestStream.onNext(requestForDescriptor(dep));
                }
            });

            --outstandingRequests;
            if (outstandingRequests == 0) {
                resultFuture.set(DescriptorProtos.FileDescriptorSet.newBuilder()
                        .addAllFile(resolvedDescriptors.values())
                        .build());
                requestStream.onCompleted();
            }
        }

        private static ServerReflectionRequest requestForDescriptor(final String name) {
            return ServerReflectionRequest.newBuilder()
                    .setFileByFilename(name)
                    .build();
        }

        private static ServerReflectionRequest requestForSymbol(final String symbol) {
            return ServerReflectionRequest.newBuilder()
                    .setFileContainingSymbol(symbol)
                    .build();
        }
    }

    public static class ServiceResolver {

        private final ImmutableList<Descriptors.FileDescriptor> fileDescriptors;
        private final JsonFormat.TypeRegistry                   registry;
        private final JsonFormat.Parser                         parser;

        private ServiceResolver(final Iterable<Descriptors.FileDescriptor> fileDescriptors) {
            this.fileDescriptors = ImmutableList.copyOf(fileDescriptors);
            this.registry = JsonFormat.TypeRegistry.newBuilder().add(listMessageTypes()).build();
            this.parser = JsonFormat.parser().usingTypeRegistry(registry).ignoringUnknownFields();
        }

        /**
         * Creates a resolver.
         *
         * @param descriptorSet descriptorSet
         * @return ServiceResolver serviceResolver
         */
        public static ServiceResolver fromFileDescriptorSet(final DescriptorProtos.FileDescriptorSet descriptorSet) {
            ImmutableMap<String, DescriptorProtos.FileDescriptorProto> descriptorProtoIndex =
                    computeDescriptorProtoIndex(descriptorSet);
            Map<String, Descriptors.FileDescriptor> descriptorCache = new HashMap<>(8);

            ImmutableList.Builder<Descriptors.FileDescriptor> result = ImmutableList.builder();
            for (DescriptorProtos.FileDescriptorProto descriptorProto : descriptorSet.getFileList()) {
                try {
                    result.add(descriptorFromProto(descriptorProto, descriptorProtoIndex, descriptorCache));
                } catch (Descriptors.DescriptorValidationException e) {
                    logger.log(Level.WARNING, "Skipped descriptor " + descriptorProto.getName() + " due to error", e);
                }
            }
            return new ServiceResolver(result.build());
        }

        /**
         * Lists all the known message types.
         *
         * @return ImmutableSet set
         */
        public ImmutableSet<Descriptors.Descriptor> listMessageTypes() {
            ImmutableSet.Builder<Descriptors.Descriptor> resultBuilder = ImmutableSet.builder();
            fileDescriptors.forEach(d -> resultBuilder.addAll(d.getMessageTypes()));
            return resultBuilder.build();
        }

        /**
         * Resolve service method.
         *
         * @return MethodDescriptor
         */
        public Descriptors.MethodDescriptor resolveServiceMethod(final String fullServiceName, final String methodName) {
            String                        serviceName = fullServiceName.substring(fullServiceName.lastIndexOf(".") + 1);
            String                        packageName = fullServiceName.substring(0, fullServiceName.lastIndexOf("."));
            Descriptors.ServiceDescriptor service     = findService(packageName, serviceName);
            Descriptors.MethodDescriptor  method      = service.findMethodByName(methodName);
            if (method == null) {
                throw new IllegalArgumentException("Unable to find method " + methodName + " in service " + serviceName);
            }
            return method;
        }

        public Descriptors.ServiceDescriptor findService(String fullServiceName) {
            String serviceName = fullServiceName.substring(fullServiceName.lastIndexOf(".") + 1);
            String packageName = fullServiceName.substring(0, fullServiceName.lastIndexOf("."));
            return findService(packageName, serviceName);
        }

        private Descriptors.ServiceDescriptor findService(final String packageName, final String serviceName) {
            for (Descriptors.FileDescriptor fileDescriptor : fileDescriptors) {
                if (!fileDescriptor.getPackage().equals(packageName)) {
                    continue;
                }
                Descriptors.ServiceDescriptor serviceDescriptor = fileDescriptor.findServiceByName(serviceName);
                if (serviceDescriptor != null) {
                    return serviceDescriptor;
                }
            }
            throw new IllegalArgumentException("Unable to find service with name: " + serviceName);
        }

        private static ImmutableMap<String, DescriptorProtos.FileDescriptorProto> computeDescriptorProtoIndex(final DescriptorProtos.FileDescriptorSet fileDescriptorSet) {
            ImmutableMap.Builder<String, DescriptorProtos.FileDescriptorProto> resultBuilder = ImmutableMap.builder();
            for (DescriptorProtos.FileDescriptorProto descriptorProto : fileDescriptorSet.getFileList()) {
                resultBuilder.put(descriptorProto.getName(), descriptorProto);
            }
            return resultBuilder.build();
        }

        private static Descriptors.FileDescriptor descriptorFromProto(
                final DescriptorProtos.FileDescriptorProto descriptorProto,
                final ImmutableMap<String, DescriptorProtos.FileDescriptorProto> descriptorProtoIndex,
                final Map<String, Descriptors.FileDescriptor> descriptorCache) throws Descriptors.DescriptorValidationException {
            // First check the cache.
            String descriptorName = descriptorProto.getName();
            if (descriptorCache.containsKey(descriptorName)) {
                return descriptorCache.get(descriptorName);
            }

            // Then fetch all the required dependencies recursively.
            ImmutableList.Builder<Descriptors.FileDescriptor> dependencies = ImmutableList.builder();
            for (String dependencyName : descriptorProto.getDependencyList()) {
                if (!descriptorProtoIndex.containsKey(dependencyName)) {
                    throw new IllegalArgumentException("Could not find dependency: " + dependencyName);
                }
                DescriptorProtos.FileDescriptorProto dependencyProto      = descriptorProtoIndex.get(dependencyName);
                Descriptors.FileDescriptor           dependencyDescriptor = descriptorFromProto(dependencyProto, descriptorProtoIndex, descriptorCache);
                dependencies.add(dependencyDescriptor);
            }
            // Finally construct the actual descriptor.
            Descriptors.FileDescriptor[] empty          = new Descriptors.FileDescriptor[0];
            Descriptors.FileDescriptor   fileDescriptor = Descriptors.FileDescriptor.buildFrom(descriptorProto, dependencies.build().toArray(empty));
            descriptorCache.put(descriptorName, fileDescriptor);
            return fileDescriptor;
        }

        public JsonFormat.TypeRegistry getRegistry() {
            return registry;
        }

        public JsonFormat.Parser getParser() {
            return parser;
        }
    }

    public static class DynamicMessageMarshaller implements MethodDescriptor.Marshaller<Message> {

        private final Descriptors.Descriptor messageDescriptor;

        public DynamicMessageMarshaller(final Descriptors.Descriptor messageDescriptor) {
            this.messageDescriptor = messageDescriptor;
        }

        @Override
        public Message parse(final InputStream inputStream) {
            try {
                return DynamicMessage.newBuilder(messageDescriptor)
                        .mergeFrom(inputStream, ExtensionRegistryLite.getEmptyRegistry())
                        .build();
            } catch (IOException e) {
                throw new RuntimeException("Unable to merge from the supplied input stream", e);
            }
        }

        @Override
        public InputStream stream(final Message abstractMessage) {
            return abstractMessage.toByteString().newInput();
        }
    }

}

基于反射服务的gRPC调试工具

• Evans https://github.com/ktr0731/evans
• BloomRPC https://github.com/uw-labs/bloomrpc
• grpcurl https://github.com/fullstorydev/grpcurl
• grpcui https://github.com/fullstorydev/grpcui
• grpc-swagger https://github.com/grpc-swagger/grpc-swagger