ServerCalls { public static <ReqT, RespT> ServerCallHandler<ReqT, RespT> asyncBidiStreamingCall( final BidiStreamingMethod<ReqT, RespT> method) { return asyncStreamingRequestCall(method); } private  ServerCalls(); static ServerCallHandler<ReqT, RespT> asyncUnaryCall(
      final UnaryMethod<ReqT, RespT> method); static ServerCallHandler<ReqT, RespT> asyncServerStreamingCall(
      final ServerStreamingMethod<ReqT, RespT> method); static ServerCallHandler<ReqT, RespT> asyncClientStreamingCall(
      final ClientStreamingMethod<ReqT, RespT> method); static ServerCallHandler<ReqT, RespT> asyncBidiStreamingCall(
      final BidiStreamingMethod<ReqT, RespT> method); static void asyncUnimplementedUnaryCall(MethodDescriptor<?, ?> methodDescriptor,
      StreamObserver<?> responseObserver); static StreamObserver<T> asyncUnimplementedStreamingCall(
      MethodDescriptor<?, ?> methodDescriptor, StreamObserver<?> responseObserver);  }
@Test public void runtimeStreamObserverIsServerCallStreamObserver() throws Exception { final AtomicBoolean invokeCalled = new AtomicBoolean(); final AtomicBoolean onCancelCalled = new AtomicBoolean(); final AtomicBoolean onReadyCalled = new AtomicBoolean(); final AtomicReference<ServerCallStreamObserver<Integer>> callObserver = new AtomicReference<ServerCallStreamObserver<Integer>>(); ServerCallHandler<Integer, Integer> callHandler = ServerCalls.asyncBidiStreamingCall( new ServerCalls.BidiStreamingMethod<Integer, Integer>() { @Override public StreamObserver<Integer> invoke(StreamObserver<Integer> responseObserver) { assertTrue(responseObserver instanceof ServerCallStreamObserver); ServerCallStreamObserver<Integer> serverCallObserver = (ServerCallStreamObserver<Integer>) responseObserver; callObserver.set(serverCallObserver); serverCallObserver.setOnCancelHandler(new Runnable() { @Override public void run() { onCancelCalled.set(true); } }); serverCallObserver.setOnReadyHandler(new Runnable() { @Override public void run() { onReadyCalled.set(true); } }); invokeCalled.set(true); return new ServerCalls.NoopStreamObserver<Integer>(); } }); ServerCall.Listener<Integer> callListener = callHandler.startCall(STREAMING_METHOD, serverCall, new Metadata()); Mockito.when(serverCall.isReady()).thenReturn(true).thenReturn(false); Mockito.when(serverCall.isCancelled()).thenReturn(false).thenReturn(true); assertTrue(callObserver.get().isReady()); assertFalse(callObserver.get().isCancelled()); callListener.onReady(); callListener.onMessage(1); callListener.onCancel(); assertTrue(invokeCalled.get()); assertTrue(onReadyCalled.get()); assertTrue(onCancelCalled.get()); assertFalse(callObserver.get().isReady()); assertTrue(callObserver.get().isCancelled()); Mockito.verify(serverCall, times(2)).request(1); }
@Test public void cannotSetOnCancelHandlerAfterServiceInvocation() throws Exception { final AtomicReference<ServerCallStreamObserver<Integer>> callObserver = new AtomicReference<ServerCallStreamObserver<Integer>>(); ServerCallHandler<Integer, Integer> callHandler = ServerCalls.asyncBidiStreamingCall( new ServerCalls.BidiStreamingMethod<Integer, Integer>() { @Override public StreamObserver<Integer> invoke(StreamObserver<Integer> responseObserver) { callObserver.set((ServerCallStreamObserver<Integer>) responseObserver); return new ServerCalls.NoopStreamObserver<Integer>(); } }); ServerCall.Listener<Integer> callListener = callHandler.startCall(STREAMING_METHOD, serverCall, new Metadata()); callListener.onMessage(1); try { callObserver.get().setOnCancelHandler(new Runnable() { @Override public void run() { } }); fail("Cannot set onCancel handler after service invocation"); } catch (IllegalStateException expected) { } }
@Test public void cannotSetOnReadyHandlerAfterServiceInvocation() throws Exception { final AtomicReference<ServerCallStreamObserver<Integer>> callObserver = new AtomicReference<ServerCallStreamObserver<Integer>>(); ServerCallHandler<Integer, Integer> callHandler = ServerCalls.asyncBidiStreamingCall( new ServerCalls.BidiStreamingMethod<Integer, Integer>() { @Override public StreamObserver<Integer> invoke(StreamObserver<Integer> responseObserver) { callObserver.set((ServerCallStreamObserver<Integer>) responseObserver); return new ServerCalls.NoopStreamObserver<Integer>(); } }); ServerCall.Listener<Integer> callListener = callHandler.startCall(STREAMING_METHOD, serverCall, new Metadata()); callListener.onMessage(1); try { callObserver.get().setOnReadyHandler(new Runnable() { @Override public void run() { } }); fail("Cannot set onReady after service invocation"); } catch (IllegalStateException expected) { } }
@Test public void cannotDisableAutoFlowControlAfterServiceInvocation() throws Exception { final AtomicReference<ServerCallStreamObserver<Integer>> callObserver = new AtomicReference<ServerCallStreamObserver<Integer>>(); ServerCallHandler<Integer, Integer> callHandler = ServerCalls.asyncBidiStreamingCall( new ServerCalls.BidiStreamingMethod<Integer, Integer>() { @Override public StreamObserver<Integer> invoke(StreamObserver<Integer> responseObserver) { callObserver.set((ServerCallStreamObserver<Integer>) responseObserver); return new ServerCalls.NoopStreamObserver<Integer>(); } }); ServerCall.Listener<Integer> callListener = callHandler.startCall(STREAMING_METHOD, serverCall, new Metadata()); callListener.onMessage(1); try { callObserver.get().disableAutoInboundFlowControl(); fail("Cannot set onCancel handler after service invocation"); } catch (IllegalStateException expected) { } }
@Test public void disablingInboundAutoFlowControlSuppressesRequestsForMoreMessages() throws Exception { ServerCallHandler<Integer, Integer> callHandler = ServerCalls.asyncBidiStreamingCall( new ServerCalls.BidiStreamingMethod<Integer, Integer>() { @Override public StreamObserver<Integer> invoke(StreamObserver<Integer> responseObserver) { ServerCallStreamObserver<Integer> serverCallObserver = (ServerCallStreamObserver<Integer>) responseObserver; serverCallObserver.disableAutoInboundFlowControl(); return new ServerCalls.NoopStreamObserver<Integer>(); } }); ServerCall.Listener<Integer> callListener = callHandler.startCall(STREAMING_METHOD, serverCall, new Metadata()); callListener.onReady(); callListener.onMessage(1); Mockito.verify(serverCall, times(0)).request(1); }
@Test public void inprocessTransportManualFlow() throws Exception { final Semaphore semaphore = new Semaphore(1); ServerServiceDefinition service = ServerServiceDefinition.builder("some") .addMethod(STREAMING_METHOD, ServerCalls.asyncBidiStreamingCall( new ServerCalls.BidiStreamingMethod<Integer, Integer>() { int iteration; @Override public StreamObserver<Integer> invoke(StreamObserver<Integer> responseObserver) { final ServerCallStreamObserver<Integer> serverCallObserver = (ServerCallStreamObserver<Integer>) responseObserver; serverCallObserver.setOnReadyHandler(new Runnable() { @Override public void run() { while (serverCallObserver.isReady()) { serverCallObserver.onNext(iteration); } iteration++; semaphore.release(); } }); return new ServerCalls.NoopStreamObserver<Integer>() { @Override public void onCompleted() { serverCallObserver.onCompleted(); } }; } })) .build(); long tag = System.nanoTime(); InProcessServerBuilder.forName("go-with-the-flow" + tag).addService(service).build().start(); ManagedChannelImpl channel = InProcessChannelBuilder.forName("go-with-the-flow" + tag).build(); final ClientCall<Integer, Integer> clientCall = channel.newCall(STREAMING_METHOD, CallOptions.DEFAULT); final CountDownLatch latch = new CountDownLatch(1); final int[] receivedMessages = new int[6]; clientCall.start(new ClientCall.Listener<Integer>() { int index; @Override public void onMessage(Integer message) { receivedMessages[index++] = message; } @Override public void onClose(Status status, Metadata trailers) { latch.countDown(); } }, new Metadata()); semaphore.acquire(); clientCall.request(1); semaphore.acquire(); clientCall.request(2); semaphore.acquire(); clientCall.request(3); clientCall.halfClose(); latch.await(5, TimeUnit.SECONDS); assertArrayEquals(new int[]{0, 1, 1, 2, 2, 2}, receivedMessages); }