Initial implementation

grpc-json-bridge-server/build.gradle

@@ -0,0 +1,6 @@
+apply from: rootProject.file('gradle/java.gradle')
+apply from: rootProject.file('gradle/junit5.gradle')
+
+dependencies {
+    compile 'io.dropwizard:dropwizard-core'
+}

grpc-json-bridge/build.gradle

@@ -2,9 +2,15 @@ apply from: rootProject.file('gradle/java.gradle')
 apply from: rootProject.file('gradle/junit5.gradle')
 
 dependencies {
+    processor 'org.immutables:value'
+
     compile 'io.grpc:grpc-netty'
     compile 'io.grpc:grpc-services'
+    compile 'org.slf4j:slf4j-api'
+    compile 'com.github.ben-manes.caffeine:caffeine'
 
     testCompile project(':test-grpc-service')
     testCompile 'org.assertj:assertj-core'
+    testCompile 'org.mockito:mockito-core'
+    testCompile 'org.slf4j:slf4j-simple'
 }

grpc-json-bridge/src/main/java/com/github/jared2501/grpc/bridge/GrpcBridge.java

@@ -0,0 +1,33 @@
+/*
+ * (c) Copyright 2018 Palantir Technologies Inc. All rights reserved.
+ */
+
+package com.github.jared2501.grpc.bridge;
+
+import javax.annotation.Nullable;
+
+public interface GrpcBridge {
+    interface InvocationHandle {
+        void start();
+        void cancel();
+    }
+
+    enum InvocationErrorType {
+        SERVICE_NOT_FOUND,
+        SERVICE_UNAVAILABLE,
+        METHOD_NOT_FOUND,
+        NON_UNARY_RESPONSE,
+        UNKNOWN
+    }
+
+    interface InvocationObserver {
+        void onError(InvocationErrorType type, @Nullable Throwable error);
+        void onResult(String jsonOutput);
+    }
+
+    /**
+     * Invokes the specified method on the specified service with the given JSON input, returns the JSON output of the
+     * result. Note that streaming methods will result in an error state for the returned future.
+     */
+    InvocationHandle invoke(String serviceName, String fullMethodName, String jsonInput, InvocationObserver observer);
+}

grpc-json-bridge/src/main/java/com/github/jared2501/grpc/bridge/GrpcBridgeImpl.java

@@ -0,0 +1,206 @@
+/*
+ * (c) Copyright 2018 Palantir Technologies Inc. All rights reserved.
+ */
+
+package com.github.jared2501.grpc.bridge;
+
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Maps;
+import com.google.common.io.ByteStreams;
+import com.google.protobuf.DynamicMessage;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.util.JsonFormat;
+import io.grpc.CallOptions;
+import io.grpc.Channel;
+import io.grpc.ClientCall;
+import io.grpc.Metadata;
+import io.grpc.MethodDescriptor;
+import io.grpc.Status;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Map;
+import java.util.Optional;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+final class GrpcBridgeImpl implements GrpcBridge {
+
+    private static final Logger log = LoggerFactory.getLogger(GrpcBridgeImpl.class);
+
+    private final Map<String, Channel> serviceChannels = Maps.newConcurrentMap();
+    private final Map<String, ServiceIndex> serviceIndices = Maps.newConcurrentMap();
+
+    @Override
+    public InvocationHandle invoke(
+            String serverName, String fullMethodName, String jsonInput, InvocationObserver observer) {
+        return new InvocationHandle() {
+
+            private final Object lock = new Object();
+            private boolean cancelled = false;
+            private ClientCall<byte[], byte[]> call;
+
+            @Override
+            public void start() {
+                Channel channel = serviceChannels.get(serverName);
+                ServiceIndex serviceIndex = serviceIndices.get(serverName);
+
+                if (channel == null) {
+                    observer.onError(InvocationErrorType.SERVICE_NOT_FOUND, null);
+                    return;
+                } else if (serviceIndex == null) {
+                    observer.onError(InvocationErrorType.SERVICE_UNAVAILABLE, null);
+                    return;
+                }
+
+                Optional<AvailableMethod> maybeMethod = serviceIndex.getMethod(fullMethodName);
+                if (!maybeMethod.isPresent()) {
+                    observer.onError(InvocationErrorType.METHOD_NOT_FOUND, null);
+                    return;
+                }
+                AvailableMethod method = maybeMethod.get();
+
+                // Generate the request
+                DynamicMessage.Builder request = DynamicMessage.newBuilder(method.methodDescriptor().getInputType());
+                JsonFormat.Parser parser = JsonFormat.parser().usingTypeRegistry(method.typeRegistry());
+                try {
+                    parser.merge(jsonInput, request);
+                } catch (InvalidProtocolBufferException e) {
+                    observer.onError(
+                            InvocationErrorType.UNKNOWN,
+                            new RuntimeException("Exception encountered when converting JSON to proto", e));
+                    return;
+                }
+
+                // Start the call
+                synchronized (lock) {
+                    if (cancelled) {
+                        return;
+                    }
+                    call = channel.newCall(getMethodDescriptor(fullMethodName), CallOptions.DEFAULT);
+                    call.start(
+                            new ClientCall.Listener<byte[]>() {
+
+                                private byte[] firstMessage;
+
+                                @Override
+                                public void onHeaders(Metadata headers) {}
+
+                                @Override
+                                public void onMessage(byte[] messageBytes) {
+                                    if (firstMessage == null) {
+                                        firstMessage = messageBytes;
+                                    } else {
+                                        observer.onError(InvocationErrorType.NON_UNARY_RESPONSE, null);
+                                        call.halfClose();
+                                        call.cancel("Expected unary response", new RuntimeException());
+                                    }
+                                }
+
+                                @Override
+                                public void onClose(Status status, Metadata trailers) {
+                                    if (status == Status.OK) {
+                                        deliverMessage();
+                                    } else {
+                                        // TODO: deliver error
+                                    }
+                                }
+
+                                @Override
+                                public void onReady() {}
+
+                                private void deliverMessage() {
+                                    DynamicMessage message;
+                                    try {
+                                        message = DynamicMessage.parseFrom(
+                                                method.methodDescriptor().getOutputType(), firstMessage);
+                                    } catch (InvalidProtocolBufferException e) {
+                                        observer.onError(
+                                                InvocationErrorType.UNKNOWN,
+                                                new RuntimeException("Exception encountered when parsing message", e));
+                                        return;
+                                    }
+
+                                    StringBuilder jsonOutput = new StringBuilder();
+                                    try {
+                                        JsonFormat.printer()
+                                                .usingTypeRegistry(method.typeRegistry())
+                                                .appendTo(message, jsonOutput);
+                                    } catch (IOException e) {
+                                        observer.onError(
+                                                InvocationErrorType.UNKNOWN,
+                                                new RuntimeException(
+                                                        "Exception encountered when printing response", e));
+                                        return;
+                                    }
+
+                                    observer.onResult(jsonOutput.toString());
+                                }
+                            },
+                            new Metadata());
+                    call.sendMessage(request.build().toByteArray());
+                    call.halfClose();
+                    call.request(2); // Request two messages to detect if method is unary or not
+                }
+            }
+
+            @Override
+            public void cancel() {
+                synchronized (lock) {
+                    cancelled = true;
+                    if (call != null) {
+                        call.halfClose();
+                        call.cancel("Cancellation requested", new RuntimeException());
+                        call = null;
+                    }
+                }
+            }
+        };
+    }
+
+    private void updateServices(Map<String, Channel> newServices) {
+        Iterables.removeIf(serviceChannels.keySet(), serverName -> !newServices.containsKey(serverName));
+        Iterables.removeIf(serviceIndices.keySet(), serverName -> !newServices.containsKey(serverName));
+
+        serviceChannels.putAll(newServices);
+        for (Map.Entry<String, Channel> newServer : newServices.entrySet()) {
+            if (!serviceIndices.containsKey(newServer.getKey())) {
+                //                ServiceIndex index = serviceIndexFactory.create();
+                //                serviceIndices.put(newServer.getKey(), index);
+                //                updateService(newServer.getKey(), newServer.getValue(), index);
+            }
+        }
+    }
+
+    private void updateService(String serverName, Channel channel, ServiceIndex index) {
+        // TODO handle failure case
+    }
+
+    private static MethodDescriptor<byte[], byte[]> getMethodDescriptor(String fullMethodName) {
+        return MethodDescriptor.newBuilder(ByteMarshaller.INSTANCE, ByteMarshaller.INSTANCE)
+                .setType(MethodDescriptor.MethodType.UNARY)
+                .setFullMethodName(fullMethodName)
+                .setIdempotent(false)
+                .setSafe(false)
+                .build();
+    }
+
+    private static class ByteMarshaller implements MethodDescriptor.Marshaller<byte[]> {
+        static final ByteMarshaller INSTANCE = new ByteMarshaller();
+
+        @Override
+        public InputStream stream(byte[] value) {
+            return new ByteArrayInputStream(value);
+        }
+
+        @Override
+        public byte[] parse(InputStream stream) {
+            try {
+                return ByteStreams.toByteArray(stream);
+            } catch (IOException e) {
+                throw new RuntimeException(e);
+            }
+        }
+    }
+
+}

grpc-json-bridge/src/main/java/com/github/jared2501/grpc/bridge/ReflectionChannel.java

@@ -0,0 +1,39 @@
+/*
+ * (c) Copyright 2018 Palantir Technologies Inc. All rights reserved.
+ */
+
+package com.github.jared2501.grpc.bridge;
+
+import com.google.protobuf.DescriptorProtos;
+
+public interface ReflectionChannel {
+    interface ReflectionObserver {
+        /** Invoked for every available gRPC service. */
+        void onAvailableService(String serviceName);
+
+        /** Invoked for every proto file. */
+        void onProtoFile(String fileName, DescriptorProtos.FileDescriptorProto proto);
+
+        /**
+         * Invoked once all services and proto files have been discovered. No further methods will be invoked on the
+         * observer if this method is invoked.
+         */
+        void onComplete();
+
+        /**
+         * Invoked if there are any expected errors.No further methods will be invoked on the observer if this method
+         * is invoked.
+         */
+        void onError(Throwable error);
+    }
+
+    interface ReflectionCall {
+        /**
+         * Cancels the reflection call with the given error. If the call is not complete then the
+         * {@link ReflectionObserver} will be notified via {@link ReflectionObserver#onError}.
+         */
+        void cancel(Throwable reason);
+    }
+
+    ReflectionCall startCall(ReflectionObserver observer);
+}