How to create GRPC client directly from protobuf without compiling it into java code

Question

When working with gRPC, we need to generate the gRPC client and server interfaces from our .proto service definition via protocol buffer compiler (protoc) or using Gradle or Maven protoc build plugin.

Flow now: protobuf file -> java code -> gRPC client.

So, is there any way to skip this step?

How to create a generic gRPC client that can call the server directly from the protobuf file without compile into java code? Or, is there a way to Generated Code at runtime?

Flow expect: protobuf file -> gRPC client.

I want to build a generic gRPC client system with the input are protobuf files along with description of method, package, message request ... without having to compile again for each protobuf.

Thank you very much.

Answer 1

Protobuf systems really need protoc to be run. However, the generated code could be skipped. Instead of passing something like --java_out and --grpc_java_out to protoc you can pass --descriptor_set_out=FILE which will parse the .proto file into a descriptor file. A descriptor file is a proto-encoded FileDescriptorSet. This is the same basic format as used with the reflection service.

Once you have a descriptor, you can load it a FileDescriptor at a time and create a DynamicMessage.

Then for the gRPC piece, you need to create a gRPC MethodDescriptor.

static MethodDescriptor from(
  Descriptors.MethodDescriptor methodDesc
) {
  return MethodDescriptor.<DynamicMessage, DynamicMessage>newBuilder()
    // UNKNOWN is fine, but the "correct" value can be computed from
    // methodDesc.toProto().getClientStreaming()/getServerStreaming()
    .setType(getMethodTypeFromDesc(methodDesc))
    .setFullMethodName(MethodDescriptor.generateFullMethodName(
        serviceDesc.getFullName(), methodDesc.getName()))
    .setRequestMarshaller(ProtoUtils.marshaller(
        DynamicMessage.getDefaultInstance(methodDesc.getInputType())))
    .setResponseMarshaller(ProtoUtils.marshaller(
        DynamicMessage.getDefaultInstance(methodDesc.getOutputType())))
    .build();

static MethodDescriptor.MethodType getMethodTypeFromDesc(
  Descriptors.MethodDescriptor methodDesc
) {
  if (!methodDesc.isServerStreaming()
    && !methodDesc.isClientStreaming()) {
    return MethodDescriptor.MethodType.UNARY;
  } else if (methodDesc.isServerStreaming()
        && !methodDesc.isClientStreaming()) {
    return MethodDescriptor.MethodType.SERVER_STREAMING;
  } else if (!methodDesc.isServerStreaming()) {
    return MethodDescriptor.MethodType.CLIENT_STREAMING);
  } else {
    return MethodDescriptor.MethodType.BIDI_STREAMING);
  }
}

At that point you have everything you need and can call Channel.newCall(method, CallOptions.DEFAULT) in gRPC. You're also free to use ClientCalls to use something more similar to the stub APIs.

So dynamic calls are definitely possible, and is used for things like grpcurl. But it also is not easy and so is generally only done when necessary.

Answer 2

I did it in Java, and the step is:

1. Call reflection service to get FileDescriptorProto list by method name
2. Get FileDescriptor of method from FileDescriptorProto list by package name, service name
3. Get MethodDescriptor from ServiceDescriptor which get from the FileDescriptor
4. Generate a MethodDescriptor<DynamicMessage, DynamicMessage> by MethodDescriptor
5. Build request DynamicMessage from content like JSON or others
6. Call method
7. Parse response content to JSON from DynamicMessage response

You can reference the full sample in project helloworlde/grpc-java-sample#reflection

---

And proto is:

syntax = "proto3";

package io.github.helloworlde.grpc;

option go_package = "api;grpc_gateway";
option java_package = "io.github.helloworlde.grpc";
option java_multiple_files = true;
option java_outer_classname = "HelloWorldGrpc";

service HelloService{
  rpc SayHello(HelloMessage) returns (HelloResponse){
  }
}

message HelloMessage {
  string message = 2;
}

message HelloResponse {
  string message = 1;
}

Start server for this proto by yourself, and the full code in Java just like:

import com.google.protobuf.ByteString;
import com.google.protobuf.DescriptorProtos;
import com.google.protobuf.Descriptors;
import com.google.protobuf.DynamicMessage;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.TypeRegistry;
import com.google.protobuf.util.JsonFormat;
import io.grpc.CallOptions;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.MethodDescriptor;
import io.grpc.protobuf.ProtoUtils;
import io.grpc.reflection.v1alpha.ServerReflectionGrpc;
import io.grpc.reflection.v1alpha.ServerReflectionRequest;
import io.grpc.reflection.v1alpha.ServerReflectionResponse;
import io.grpc.stub.ClientCalls;
import io.grpc.stub.StreamObserver;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
public class ReflectionCall {

    public static void main(String[] args) throws InterruptedException {
        // 反射方法的格式只支持 package.service.method 或者 package.service
        String methodSymbol = "io.github.helloworlde.grpc.HelloService.SayHello";
        String requestContent = "{\"message\": \"Reflection\"}";

        // 构建 Channel
        ManagedChannel channel = ManagedChannelBuilder.forAddress("127.0.0.1", 9090)
                                                      .usePlaintext()
                                                      .build();
        // 使用 Channel 构建 BlockingStub
        ServerReflectionGrpc.ServerReflectionStub reflectionStub = ServerReflectionGrpc.newStub(channel);
        // 响应观察器
        StreamObserver<ServerReflectionResponse> streamObserver = new StreamObserver<ServerReflectionResponse>() {
            @Override
            public void onNext(ServerReflectionResponse response) {
                try {
                    // 只需要关注文件描述类型的响应
                    if (response.getMessageResponseCase() == ServerReflectionResponse.MessageResponseCase.FILE_DESCRIPTOR_RESPONSE) {
                        List<ByteString> fileDescriptorProtoList = response.getFileDescriptorResponse().getFileDescriptorProtoList();
                        handleResponse(fileDescriptorProtoList, channel, methodSymbol, requestContent);
                    } else {
                        log.warn("未知响应类型: " + response.getMessageResponseCase());
                    }
                } catch (Exception e) {
                    log.error("处理响应失败: {}", e.getMessage(), e);
                }
            }

            @Override
            public void onError(Throwable t) {

            }

            @Override
            public void onCompleted() {
                log.info("Complete");
            }
        };
        // 请求观察器
        StreamObserver<ServerReflectionRequest> requestStreamObserver = reflectionStub.serverReflectionInfo(streamObserver);

        // 构建并发送获取方法文件描述请求
        ServerReflectionRequest getFileContainingSymbolRequest = ServerReflectionRequest.newBuilder()
                                                                                        .setFileContainingSymbol(methodSymbol)
                                                                                        .build();
        requestStreamObserver.onNext(getFileContainingSymbolRequest);
        channel.awaitTermination(10, TimeUnit.SECONDS);
    }

    /**
     * 处理响应
     */
    private static void handleResponse(List<ByteString> fileDescriptorProtoList,
                                       ManagedChannel channel,
                                       String methodFullName,
                                       String requestContent) {
        try {
            // 解析方法和服务名称
            String fullServiceName = extraPrefix(methodFullName);
            String methodName = extraSuffix(methodFullName);
            String packageName = extraPrefix(fullServiceName);
            String serviceName = extraSuffix(fullServiceName);

            // 根据响应解析 FileDescriptor
            Descriptors.FileDescriptor fileDescriptor = getFileDescriptor(fileDescriptorProtoList, packageName, serviceName);

            // 查找服务描述
            Descriptors.ServiceDescriptor serviceDescriptor = fileDescriptor.getFile().findServiceByName(serviceName);
            // 查找方法描述
            Descriptors.MethodDescriptor methodDescriptor = serviceDescriptor.findMethodByName(methodName);

            // 发起请求
            executeCall(channel, fileDescriptor, methodDescriptor, requestContent);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    /**
     * 解析并查找方法对应的文件描述
     */
    private static Descriptors.FileDescriptor getFileDescriptor(List<ByteString> fileDescriptorProtoList,
                                                                String packageName,
                                                                String serviceName) throws Exception {

        Map<String, DescriptorProtos.FileDescriptorProto> fileDescriptorProtoMap =
                fileDescriptorProtoList.stream()
                                       .map(bs -> {
                                           try {
                                               return DescriptorProtos.FileDescriptorProto.parseFrom(bs);
                                           } catch (InvalidProtocolBufferException e) {
                                               e.printStackTrace();
                                           }
                                           return null;
                                       })
                                       .filter(Objects::nonNull)
                                       .collect(Collectors.toMap(DescriptorProtos.FileDescriptorProto::getName, f -> f));


        if (fileDescriptorProtoMap.isEmpty()) {
            log.error("服务不存在");
            throw new IllegalArgumentException("方法的文件描述不存在");
        }

        // 查找服务对应的 Proto 描述
        DescriptorProtos.FileDescriptorProto fileDescriptorProto = findServiceFileDescriptorProto(packageName, serviceName, fileDescriptorProtoMap);

        // 获取这个 Proto 的依赖
        Descriptors.FileDescriptor[] dependencies = getDependencies(fileDescriptorProto, fileDescriptorProtoMap);

        // 生成 Proto 的 FileDescriptor
        return Descriptors.FileDescriptor.buildFrom(fileDescriptorProto, dependencies);
    }


    /**
     * 根据包名和服务名查找相应的文件描述
     */
    private static DescriptorProtos.FileDescriptorProto findServiceFileDescriptorProto(String packageName,
                                                                                       String serviceName,
                                                                                       Map<String, DescriptorProtos.FileDescriptorProto> fileDescriptorProtoMap) {
        for (DescriptorProtos.FileDescriptorProto proto : fileDescriptorProtoMap.values()) {
            if (proto.getPackage().equals(packageName)) {
                boolean exist = proto.getServiceList()
                                     .stream()
                                     .anyMatch(s -> serviceName.equals(s.getName()));
                if (exist) {
                    return proto;
                }
            }
        }

        throw new IllegalArgumentException("服务不存在");
    }

    /**
     * 获取前缀
     */
    private static String extraPrefix(String content) {
        int index = content.lastIndexOf(".");
        return content.substring(0, index);
    }

    /**
     * 获取后缀
     */
    private static String extraSuffix(String content) {
        int index = content.lastIndexOf(".");
        return content.substring(index + 1);
    }

    /**
     * 获取依赖类型
     */
    private static Descriptors.FileDescriptor[] getDependencies(DescriptorProtos.FileDescriptorProto proto,
                                                                Map<String, DescriptorProtos.FileDescriptorProto> finalDescriptorProtoMap) {
        return proto.getDependencyList()
                    .stream()
                    .map(finalDescriptorProtoMap::get)
                    .map(f -> toFileDescriptor(f, getDependencies(f, finalDescriptorProtoMap)))
                    .toArray(Descriptors.FileDescriptor[]::new);
    }

    /**
     * 将 FileDescriptorProto 转为 FileDescriptor
     */
    @SneakyThrows
    private static Descriptors.FileDescriptor toFileDescriptor(DescriptorProtos.FileDescriptorProto fileDescriptorProto,
                                                               Descriptors.FileDescriptor[] dependencies) {
        return Descriptors.FileDescriptor.buildFrom(fileDescriptorProto, dependencies);
    }


    /**
     * 执行方法调用
     */
    private static void executeCall(ManagedChannel channel,
                                    Descriptors.FileDescriptor fileDescriptor,
                                    Descriptors.MethodDescriptor originMethodDescriptor,
                                    String requestContent) throws Exception {

        // 重新生成 MethodDescriptor
        MethodDescriptor<DynamicMessage, DynamicMessage> methodDescriptor = generateMethodDescriptor(originMethodDescriptor);

        CallOptions callOptions = CallOptions.DEFAULT;

        TypeRegistry registry = TypeRegistry.newBuilder()
                                            .add(fileDescriptor.getMessageTypes())
                                            .build();

        // 将请求内容由 JSON 字符串转为相应的类型
        JsonFormat.Parser parser = JsonFormat.parser().usingTypeRegistry(registry);
        DynamicMessage.Builder messageBuilder = DynamicMessage.newBuilder(originMethodDescriptor.getInputType());
        parser.merge(requestContent, messageBuilder);
        DynamicMessage requestMessage = messageBuilder.build();

        // 调用，调用方式可以通过 originMethodDescriptor.isClientStreaming() 和 originMethodDescriptor.isServerStreaming() 推断
        DynamicMessage response = ClientCalls.blockingUnaryCall(channel, methodDescriptor, callOptions, requestMessage);

        // 将响应解析为 JSON 字符串
        JsonFormat.Printer printer = JsonFormat.printer()
                                               .usingTypeRegistry(registry)
                                               .includingDefaultValueFields();
        String responseContent = printer.print(response);

        log.info("响应: {}", responseContent);
    }

    /**
     * 重新生成方法描述
     */
    private static MethodDescriptor<DynamicMessage, DynamicMessage> generateMethodDescriptor(Descriptors.MethodDescriptor originMethodDescriptor) {
        // 生成方法全名
        String fullMethodName = MethodDescriptor.generateFullMethodName(originMethodDescriptor.getService().getFullName(), originMethodDescriptor.getName());
        // 请求和响应类型
        MethodDescriptor.Marshaller<DynamicMessage> inputTypeMarshaller = ProtoUtils.marshaller(DynamicMessage.newBuilder(originMethodDescriptor.getInputType())
                                                                                                              .buildPartial());
        MethodDescriptor.Marshaller<DynamicMessage> outputTypeMarshaller = ProtoUtils.marshaller(DynamicMessage.newBuilder(originMethodDescriptor.getOutputType())
                                                                                                               .buildPartial());

        // 生成方法描述, originMethodDescriptor 的 fullMethodName 不正确
        return MethodDescriptor.<DynamicMessage, DynamicMessage>newBuilder()
                .setFullMethodName(fullMethodName)
                .setRequestMarshaller(inputTypeMarshaller)
                .setResponseMarshaller(outputTypeMarshaller)
                // 使用 UNKNOWN，自动修改
                .setType(MethodDescriptor.MethodType.UNKNOWN)
                .build();
    }
}