diff --git a/grpc-json-bridge-server/build.gradle b/grpc-json-bridge-server/build.gradle new file mode 100644 index 0000000..59c5420 --- /dev/null +++ b/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' +} diff --git a/grpc-json-bridge/build.gradle b/grpc-json-bridge/build.gradle index 382c873..d60ad40 100644 --- a/grpc-json-bridge/build.gradle +++ b/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' } diff --git a/grpc-json-bridge/src/main/java/com/github/jared2501/grpc/bridge/GrpcBridge.java b/grpc-json-bridge/src/main/java/com/github/jared2501/grpc/bridge/GrpcBridge.java new file mode 100644 index 0000000..e8d1c89 --- /dev/null +++ b/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); +} diff --git a/grpc-json-bridge/src/main/java/com/github/jared2501/grpc/bridge/GrpcBridgeImpl.java b/grpc-json-bridge/src/main/java/com/github/jared2501/grpc/bridge/GrpcBridgeImpl.java new file mode 100644 index 0000000..b8016c3 --- /dev/null +++ b/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 serviceChannels = Maps.newConcurrentMap(); + private final Map 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 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 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() { + + 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 newServices) { + Iterables.removeIf(serviceChannels.keySet(), serverName -> !newServices.containsKey(serverName)); + Iterables.removeIf(serviceIndices.keySet(), serverName -> !newServices.containsKey(serverName)); + + serviceChannels.putAll(newServices); + for (Map.Entry 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 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 { + 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); + } + } + } + +} diff --git a/grpc-json-bridge/src/main/java/com/github/jared2501/grpc/bridge/ReflectionChannel.java b/grpc-json-bridge/src/main/java/com/github/jared2501/grpc/bridge/ReflectionChannel.java new file mode 100644 index 0000000..5b9dddc --- /dev/null +++ b/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); +} diff --git a/grpc-json-bridge/src/main/java/com/github/jared2501/grpc/bridge/ReflectionChannelImpl.java b/grpc-json-bridge/src/main/java/com/github/jared2501/grpc/bridge/ReflectionChannelImpl.java new file mode 100644 index 0000000..b0049bb --- /dev/null +++ b/grpc-json-bridge/src/main/java/com/github/jared2501/grpc/bridge/ReflectionChannelImpl.java @@ -0,0 +1,138 @@ +/* + * (c) Copyright 2018 Palantir Technologies Inc. All rights reserved. + */ + +package com.github.jared2501.grpc.bridge; + +import com.google.common.collect.Sets; +import com.google.protobuf.ByteString; +import com.google.protobuf.DescriptorProtos; +import com.google.protobuf.InvalidProtocolBufferException; +import io.grpc.Context; +import io.grpc.Status; +import io.grpc.reflection.v1alpha.FileDescriptorResponse; +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.util.Set; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public final class ReflectionChannelImpl implements ReflectionChannel { + + private static final Logger log = LoggerFactory.getLogger(ReflectionChannelImpl.class); + + private final ServerReflectionGrpc.ServerReflectionStub reflectionStub; + private final String serviceName; + + public ReflectionChannelImpl(ServerReflectionGrpc.ServerReflectionStub reflectionStub, String serviceName) { + this.reflectionStub = reflectionStub; + this.serviceName = serviceName; + } + + @Override + public ReflectionCall startCall(ReflectionObserver observer) { + Context.CancellableContext context = Context.CancellableContext.current().withCancellation(); + context.run(() -> { + ReflectionResponseObserver streamObs = new ReflectionResponseObserver(serviceName, observer); + StreamObserver reqStream = reflectionStub.serverReflectionInfo(streamObs); + streamObs.start(reqStream); + }); + return context::cancel; + } + + private static class ReflectionResponseObserver implements StreamObserver { + + private final String serviceName; + private final ReflectionObserver observer; + private final Set observedProtoFiles = Sets.newHashSet(); + private final Set outstandingRequests = Sets.newHashSet(); + + private StreamObserver reqStream; + + ReflectionResponseObserver(String serviceName, ReflectionObserver observer) { + this.serviceName = serviceName; + this.observer = observer; + } + + public void start(StreamObserver requestStreamToRun) { + reqStream = requestStreamToRun; + reqStream.onNext(ServerReflectionRequest.newBuilder() + .setListServices("true") + .build()); + } + + @Override + public void onNext(ServerReflectionResponse response) { + outstandingRequests.remove(response.getOriginalRequest()); + + switch (response.getMessageResponseCase()) { + case LIST_SERVICES_RESPONSE: + requestAllFilesForServices(response); + break; + case FILE_DESCRIPTOR_RESPONSE: + requestUnseenDependencyProtos(response.getFileDescriptorResponse()); + break; + default: + log.error("Unexpected response case: {}", response.getMessageResponseCase()); + break; + } + + if (outstandingRequests.isEmpty()) { + reqStream.onCompleted(); + observer.onComplete(); + } + } + + @Override + public void onError(Throwable error) { + log.debug("Throwable encountered when streaming handling reflecting for service {}", serviceName, error); + observer.onError(error); + } + + @Override + public void onCompleted() { + log.debug("Reflection complete, service {} likely shutting down.", serviceName); + observer.onError(Status.UNAVAILABLE.asException()); + } + + private void requestAllFilesForServices(ServerReflectionResponse response) { + for (ServiceResponse service : response.getListServicesResponse().getServiceList()) { + observer.onAvailableService(service.getName()); + makeRequest(ServerReflectionRequest.newBuilder() + .setFileContainingSymbol(service.getName()) + .build()); + } + } + + private void requestUnseenDependencyProtos(FileDescriptorResponse response) { + for (ByteString protoBytes : response.getFileDescriptorProtoList()) { + DescriptorProtos.FileDescriptorProto protoDescriptor; + try { + protoDescriptor = DescriptorProtos.FileDescriptorProto.parseFrom(protoBytes); + } catch (InvalidProtocolBufferException e) { + log.warn("InvalidProtocolBufferException when parsing proto bytes... skipping", e); + continue; + } + + observer.onProtoFile(protoDescriptor.getName(), protoDescriptor); + observedProtoFiles.add(protoDescriptor.getName()); + + for (String dependencyFileName : protoDescriptor.getDependencyList()) { + if (!observedProtoFiles.contains(dependencyFileName)) { + makeRequest(ServerReflectionRequest.newBuilder() + .setFileByFilename(dependencyFileName) + .build()); + } + } + } + } + + private void makeRequest(ServerReflectionRequest request) { + outstandingRequests.add(request); + reqStream.onNext(request); + } + } +} diff --git a/grpc-json-bridge/src/main/java/com/github/jared2501/grpc/bridge/ServiceIndex.java b/grpc-json-bridge/src/main/java/com/github/jared2501/grpc/bridge/ServiceIndex.java new file mode 100644 index 0000000..6d123dd --- /dev/null +++ b/grpc-json-bridge/src/main/java/com/github/jared2501/grpc/bridge/ServiceIndex.java @@ -0,0 +1,14 @@ +/* + * (c) Copyright 2018 Palantir Technologies Inc. All rights reserved. + */ + +package com.github.jared2501.grpc.bridge; + +import com.google.protobuf.Descriptors; +import java.util.Optional; + +public interface ServiceIndex { + + Optional getMethod(String fullMethodName); + +} diff --git a/grpc-json-bridge/src/main/java/com/github/jared2501/grpc/bridge/ServiceIndexImpl.java b/grpc-json-bridge/src/main/java/com/github/jared2501/grpc/bridge/ServiceIndexImpl.java new file mode 100644 index 0000000..6986aba --- /dev/null +++ b/grpc-json-bridge/src/main/java/com/github/jared2501/grpc/bridge/ServiceIndexImpl.java @@ -0,0 +1,175 @@ +/* + * (c) Copyright 2018 Palantir Technologies Inc. All rights reserved. + */ + +package com.github.jared2501.grpc.bridge; + +import com.google.common.collect.ImmutableMap; +import com.google.common.collect.Maps; +import com.google.common.collect.Sets; +import com.google.common.collect.TreeTraverser; +import com.google.common.util.concurrent.AbstractService; +import com.google.protobuf.DescriptorProtos; +import com.google.protobuf.Descriptors; +import com.google.protobuf.util.JsonFormat; +import java.time.Duration; +import java.util.Collection; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +// TODO: re-index when the channel changes state from closed -> open +public final class ServiceIndexImpl extends AbstractService implements ServiceIndex { + + private static final Logger log = LoggerFactory.getLogger(ServiceIndexImpl.class); + + private final ReflectionChannel reflectionChannel; + private final ScheduledExecutorService executorService; + private final Duration failureDelay; + + private AtomicReference currentCall = + new AtomicReference<>(); + private AtomicReference> index = + new AtomicReference<>(ImmutableMap.of()); + + public ServiceIndexImpl( + ReflectionChannel reflectionChannel, ScheduledExecutorService executorService, Duration failureDelay) { + this.reflectionChannel = reflectionChannel; + this.executorService = executorService; + this.failureDelay = failureDelay; + } + + @Override + protected void doStart() { + scheduleReflection(Duration.ZERO); + notifyStarted(); + } + + @Override + protected void doStop() { + ReflectionChannel.ReflectionCall call = currentCall.get(); + if (call != null) { + call.cancel(new RuntimeException()); + } + notifyStopped(); + } + + @Override + public Optional getMethod(String fullMethodName) { + return Optional.ofNullable(index.get().get(fullMethodName)); + } + + private void scheduleReflection(Duration scheduleDelay) { + executorService.schedule( + () -> { + if (!isRunning()) { + return; + } + ReflectionChannel.ReflectionCall lastCall = currentCall.getAndSet( + reflectionChannel.startCall(new IndexFromReflectionObserver())); + lastCall.cancel(new RuntimeException()); + }, + scheduleDelay.toMillis(), + TimeUnit.MILLISECONDS); + } + + private class IndexFromReflectionObserver implements ReflectionChannel.ReflectionObserver { + + private Set availableServices = Sets.newHashSet(); + private Map protoFiles = Maps.newHashMap(); + + @Override + public void onAvailableService(String serviceName) { + availableServices.add(serviceName); + } + + @Override + public void onProtoFile(String fileName, DescriptorProtos.FileDescriptorProto proto) { + protoFiles.put(fileName, proto); + } + + @Override + public void onComplete() { + index.set(buildIndex(compileProtos())); + } + + @Override + public void onError(Throwable error) { + log.warn("Error encountered receiving server reflection. Restart reflection. delay={}", + failureDelay, error); + scheduleReflection(failureDelay); + } + + + private Collection compileProtos() { + // Find all "roots", where a root is a proto file for which another proto file does not depend on it + Map rootsByFileName = Maps.newHashMap(protoFiles); + for (DescriptorProtos.FileDescriptorProto proto : protoFiles.values()) { + for (String dependencyFileName : proto.getDependencyList()) { + rootsByFileName.remove(dependencyFileName); + } + } + + Map compiledProtosByFileName = Maps.newHashMap(); + + // Perform a postorder traversal (i.e. visit children first) from every root, compiling and storing the proto + // file if it has not already been compiled + TreeTraverser treeTraverser = new TreeTraverser() { + @Override + public Iterable children( + DescriptorProtos.FileDescriptorProto root) { + return root.getDependencyList() + .stream() + // Note: skip visiting dependencies if they have already been compiled + .filter(dependencyFileName -> !compiledProtosByFileName.containsKey(dependencyFileName)) + .map(protoFiles::get) + .collect(Collectors.toSet()); + } + }; + for (DescriptorProtos.FileDescriptorProto root : rootsByFileName.values()) { + for (DescriptorProtos.FileDescriptorProto proto : treeTraverser.postOrderTraversal(root)) { + Descriptors.FileDescriptor[] dependencies = proto.getDependencyList() + .stream() + .map(compiledProtosByFileName::get) + .toArray(Descriptors.FileDescriptor[]::new); + + Descriptors.FileDescriptor compiledProto; + try { + compiledProto = Descriptors.FileDescriptor.buildFrom(proto, dependencies); + } catch (Descriptors.DescriptorValidationException e) { + log.warn("Exception encountered when building proto... skipping", e); + continue; + } + + compiledProtosByFileName.put(proto.getName(), compiledProto); + } + } + + return compiledProtosByFileName.values(); + } + + private Map buildIndex( + Collection compiledProtos) { + Map index = Maps.newHashMap(); + JsonFormat.TypeRegistry.Builder typeRegistryBuilder = JsonFormat.TypeRegistry.newBuilder(); + for (Descriptors.FileDescriptor compiledProto : compiledProtos) { + typeRegistryBuilder.add(compiledProto.getMessageTypes()); + for (Descriptors.ServiceDescriptor service : compiledProto.getServices()) { + if (availableServices.contains(service.getFullName())) { + for (Descriptors.MethodDescriptor method : service.getMethods()) { + String methodFullName = String.format("%s/%s", service.getFullName(), method.getName()); + index.put(methodFullName, method); + } + } + } + } + return index; + } + } +} diff --git a/grpc-json-bridge/src/test/java/com/github/jared2501/grpc/bridge/GrpcBridgeImplIntegrationTest.java b/grpc-json-bridge/src/test/java/com/github/jared2501/grpc/bridge/GrpcBridgeImplIntegrationTest.java new file mode 100644 index 0000000..e154edf --- /dev/null +++ b/grpc-json-bridge/src/test/java/com/github/jared2501/grpc/bridge/GrpcBridgeImplIntegrationTest.java @@ -0,0 +1,72 @@ +/* + * (c) Copyright 2018 Palantir Technologies Inc. All rights reserved. + */ + +package com.github.jared2501.grpc.bridge; + +import com.github.jared2501.grpc.bridge.test.TestServiceImpl; +import io.grpc.ManagedChannel; +import io.grpc.Server; +import io.grpc.inprocess.InProcessChannelBuilder; +import io.grpc.inprocess.InProcessServerBuilder; +import io.grpc.protobuf.services.ProtoReflectionService; +import java.io.IOException; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +class GrpcBridgeImplIntegrationTest { + + private Server server; + private ManagedChannel channel; + + @BeforeEach + void setUp() throws IOException { + server = InProcessServerBuilder.forName("test") + .addService(TestServiceImpl.INSTANCE) + .addService(ProtoReflectionService.newInstance()) + .build() + .start(); + channel = InProcessChannelBuilder.forName("test").build(); + } + + @AfterEach + void tearDown() throws InterruptedException { + channel.shutdown(); + channel.awaitTermination(5, TimeUnit.HOURS); + server.shutdown(); + server.awaitTermination(); + } + + @Test + void norfquix() throws ExecutionException, InterruptedException { + // GrpcBridge bridge = new GrpcBridgeImpl(serviceName -> channel); + // + // GrpcBridge.InvocationHandle handle = bridge.invoke( + // "foo", + // "com.github.jared2501.grpc.bridge.test.TestService/UnaryReqUnaryResp", + // "{\"message\": \"message\", \"empty\": {}}", + // new GrpcBridge.InvocationObserver() { + // @Override + // public void onResult(String jsonOutput) { + // System.out.println("output: " + jsonOutput); + // } + // + // @Override + // public void onMethodNotFound() { + // System.out.println("method not found!"); + // } + // + // @Override + // public void onError(Throwable error) { + // error.printStackTrace(); + // } + // }); + // + // handle.start(); + // + // Thread.sleep(5000); + } +} diff --git a/grpc-json-bridge/src/test/java/com/github/jared2501/grpc/bridge/ReflectionChannelImplTest.java b/grpc-json-bridge/src/test/java/com/github/jared2501/grpc/bridge/ReflectionChannelImplTest.java new file mode 100644 index 0000000..14fab9e --- /dev/null +++ b/grpc-json-bridge/src/test/java/com/github/jared2501/grpc/bridge/ReflectionChannelImplTest.java @@ -0,0 +1,77 @@ +/* + * (c) Copyright 2018 Palantir Technologies Inc. All rights reserved. + */ + +package com.github.jared2501.grpc.bridge; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.github.jared2501.grpc.bridge.test.TestServiceImpl; +import io.grpc.ManagedChannel; +import io.grpc.Server; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; +import io.grpc.inprocess.InProcessChannelBuilder; +import io.grpc.inprocess.InProcessServerBuilder; +import io.grpc.protobuf.services.ProtoReflectionService; +import io.grpc.reflection.v1alpha.ServerReflectionGrpc; +import java.io.IOException; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +class ReflectionChannelImplTest { + + private Server server; + private ManagedChannel channel; + private ReflectionChannel reflection; + + @BeforeEach + void setUp() throws IOException { + server = InProcessServerBuilder.forName("test") + .addService(TestServiceImpl.INSTANCE) + .addService(ProtoReflectionService.newInstance()) + .build() + .start(); + channel = InProcessChannelBuilder.forName("test").build(); + reflection = new ReflectionChannelImpl(ServerReflectionGrpc.newStub(channel), "test"); + } + + @AfterEach + void tearDown() throws InterruptedException { + channel.shutdown(); + channel.awaitTermination(5, TimeUnit.HOURS); + server.shutdown(); + server.awaitTermination(); + } + + @Test + void successfulRoundTrip() { + TestReflectionObserver observer = new TestReflectionObserver(); + reflection.startCall(observer); + + observer.waitUntilComplete(); + assertThat(observer.getError()).isNull(); + assertThat(observer.getAvailableService()).containsExactly( + "com.github.jared2501.grpc.bridge.test.TestService", + "grpc.reflection.v1alpha.ServerReflection"); + assertThat(observer.getProtoFiles().keySet()).containsExactly( + "test.proto", + "google/protobuf/empty.proto", + "transitive.proto", + "io/grpc/reflection/v1alpha/reflection.proto"); + } + + @Test + void unavailableChannel() throws InterruptedException { + channel.shutdown(); + channel.awaitTermination(5, TimeUnit.HOURS); + TestReflectionObserver observer = new TestReflectionObserver(); + reflection.startCall(observer); + observer.waitUntilComplete(); + assertThat(observer.getError()) + .isInstanceOf(StatusRuntimeException.class) + .matches(error -> ((StatusRuntimeException) error).getStatus().getCode() == Status.Code.UNAVAILABLE); + } +} diff --git a/grpc-json-bridge/src/test/java/com/github/jared2501/grpc/bridge/ServiceIndexImplTest.java b/grpc-json-bridge/src/test/java/com/github/jared2501/grpc/bridge/ServiceIndexImplTest.java new file mode 100644 index 0000000..5ad14af --- /dev/null +++ b/grpc-json-bridge/src/test/java/com/github/jared2501/grpc/bridge/ServiceIndexImplTest.java @@ -0,0 +1,16 @@ +/* + * (c) Copyright 2018 Palantir Technologies Inc. All rights reserved. + */ + +package com.github.jared2501.grpc.bridge; + +import static org.assertj.core.api.Assertions.assertThat; + +import org.junit.jupiter.api.Test; + +class ServiceIndexImplTest { + @Test + void getMethod() { + assertThat(true).isFalse(); + } +} diff --git a/grpc-json-bridge/src/test/java/com/github/jared2501/grpc/bridge/TestReflectionObserver.java b/grpc-json-bridge/src/test/java/com/github/jared2501/grpc/bridge/TestReflectionObserver.java new file mode 100644 index 0000000..e0aa8f0 --- /dev/null +++ b/grpc-json-bridge/src/test/java/com/github/jared2501/grpc/bridge/TestReflectionObserver.java @@ -0,0 +1,57 @@ +/* + * (c) Copyright 2018 Palantir Technologies Inc. All rights reserved. + */ + +package com.github.jared2501.grpc.bridge; + +import com.google.common.collect.Maps; +import com.google.protobuf.DescriptorProtos; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.Semaphore; +import org.assertj.core.util.Sets; + +public final class TestReflectionObserver implements ReflectionChannel.ReflectionObserver { + + private Set availableService = Sets.newHashSet(); + private Map protoFiles = Maps.newHashMap(); + private Throwable error; + private final Semaphore done = new Semaphore(0); + + @Override + public void onAvailableService(String serviceName) { + availableService.add(serviceName); + } + + @Override + public void onProtoFile(String fileName, DescriptorProtos.FileDescriptorProto proto) { + protoFiles.put(fileName, proto); + } + + @Override + public void onComplete() { + done.release(1); + } + + @Override + public void onError(Throwable err) { + error = err; + done.release(1); + } + + public void waitUntilComplete() { + done.acquireUninterruptibly(1); + } + + public Throwable getError() { + return error; + } + + public Set getAvailableService() { + return availableService; + } + + public Map getProtoFiles() { + return protoFiles; + } +} diff --git a/settings.gradle b/settings.gradle index 9747cca..f5796e5 100644 --- a/settings.gradle +++ b/settings.gradle @@ -1,4 +1,5 @@ rootProject.name = 'grpc-json-bridge' include 'grpc-json-bridge' +include 'grpc-json-bridge-server' include 'test-grpc-service' diff --git a/test-grpc-service/src/grpc/proto/test.proto b/test-grpc-service/src/grpc/proto/test.proto index 605ce3a..5ee8ffb 100644 --- a/test-grpc-service/src/grpc/proto/test.proto +++ b/test-grpc-service/src/grpc/proto/test.proto @@ -2,11 +2,14 @@ syntax = "proto3"; package com.github.jared2501.grpc.bridge.test; +import "transitive.proto"; + option java_package = "com.github.jared2501.grpc.bridge.test"; option java_multiple_files = true; message TestMessage { string message = 1; + TransitiveEmpty empty = 2; // Tests transitive imports } // A simple test service that will echo a TestMessage. diff --git a/test-grpc-service/src/grpc/proto/transitive.proto b/test-grpc-service/src/grpc/proto/transitive.proto new file mode 100644 index 0000000..1337250 --- /dev/null +++ b/test-grpc-service/src/grpc/proto/transitive.proto @@ -0,0 +1,12 @@ +syntax = "proto3"; + +package com.github.jared2501.grpc.bridge.test; + +import "google/protobuf/empty.proto"; + +option java_package = "com.github.jared2501.grpc.bridge.test"; +option java_multiple_files = true; + +message TransitiveEmpty { + google.protobuf.Empty empty = 1; +} diff --git a/test-grpc-service/src/main/java/com/github/jared2501/grpc/bridge/test/TestServiceImpl.java b/test-grpc-service/src/main/java/com/github/jared2501/grpc/bridge/test/TestServiceImpl.java index 18db38d..dc573d7 100644 --- a/test-grpc-service/src/main/java/com/github/jared2501/grpc/bridge/test/TestServiceImpl.java +++ b/test-grpc-service/src/main/java/com/github/jared2501/grpc/bridge/test/TestServiceImpl.java @@ -7,6 +7,11 @@ import io.grpc.stub.StreamObserver; public final class TestServiceImpl extends TestServiceGrpc.TestServiceImplBase { + + public static final TestServiceImpl INSTANCE = new TestServiceImpl(); + + private TestServiceImpl() {} + @Override public void unaryReqUnaryResp(TestMessage request, StreamObserver responseObserver) { responseObserver.onNext(request); diff --git a/test-grpc-service/src/test/java/com/github/jared2501/grpc/bridge/test/TestServiceImplTest.java b/test-grpc-service/src/test/java/com/github/jared2501/grpc/bridge/test/TestServiceImplTest.java index 4e2e2f2..1fe4272 100644 --- a/test-grpc-service/src/test/java/com/github/jared2501/grpc/bridge/test/TestServiceImplTest.java +++ b/test-grpc-service/src/test/java/com/github/jared2501/grpc/bridge/test/TestServiceImplTest.java @@ -24,7 +24,7 @@ class TestServiceImplTest { @BeforeEach void before() { MockitoAnnotations.initMocks(this); - testService = new TestServiceImpl(); + testService = TestServiceImpl.INSTANCE; } @Test diff --git a/versions.props b/versions.props index f94c40a..7ab67a3 100644 --- a/versions.props +++ b/versions.props @@ -1,6 +1,10 @@ +com.github.ben-manes.caffeine:caffeine = 2.6.1 com.google.protobuf:* = 3.5.0 +io.dropwizard:* = 1.2.3 io.grpc:* = 1.7.1 org.apiguardian:apiguardian-api = 1.0.0 org.assertj:assertj-core = 3.6.2 +org.immutables:* = 2.5.6 org.junit.jupiter:* = 5.0.2 org.mockito:* = 2.13.0 +org.slf4j:* = 1.7.25