### Question:
GrpcRootRangeService extends RootRangeServiceImplBase { @Override public void getStream(GetStreamRequest request, StreamObserver<GetStreamResponse> responseObserver) { rs.getStream(request).whenComplete( new ResponseHandler<GetStreamResponse>(responseObserver) { @Override protected GetStreamResponse createErrorResp(Throwable cause) { return GetStreamResponse.newBuilder() .setCode(StatusCode.INTERNAL_SERVER_ERROR) .build(); } }); } GrpcRootRangeService(RangeStoreService rs); @Override void createNamespace(CreateNamespaceRequest request,
                                StreamObserver<CreateNamespaceResponse> responseObserver); @Override void deleteNamespace(DeleteNamespaceRequest request,
                                StreamObserver<DeleteNamespaceResponse> responseObserver); @Override void getNamespace(GetNamespaceRequest request,
                             StreamObserver<GetNamespaceResponse> responseObserver); @Override void createStream(CreateStreamRequest request,
                             StreamObserver<CreateStreamResponse> responseObserver); @Override void deleteStream(DeleteStreamRequest request,
                             StreamObserver<DeleteStreamResponse> responseObserver); @Override void getStream(GetStreamRequest request,
                          StreamObserver<GetStreamResponse> responseObserver);  }### Answer:
@Test public void testGetStreamSuccess() throws Exception { RangeStoreService rangeService = mock(RangeStoreService.class); GrpcRootRangeService grpcService = new GrpcRootRangeService(rangeService); GetStreamResponse getResp = GetStreamResponse.newBuilder() .setCode(StatusCode.SUCCESS) .setStreamProps(streamProps) .build(); GetStreamRequest getReq = createGetStreamRequest(nsName, streamName); when(rangeService.getStream(getReq)).thenReturn( CompletableFuture.completedFuture(getResp)); AtomicReference<GetStreamResponse> resultHolder = new AtomicReference<>(); AtomicReference<Throwable> exceptionHolder = new AtomicReference<>(); CountDownLatch latch = new CountDownLatch(1); StreamObserver<GetStreamResponse> streamObserver = new StreamObserver<GetStreamResponse>() { @Override public void onNext(GetStreamResponse resp) { resultHolder.set(resp); } @Override public void onError(Throwable t) { exceptionHolder.set(t); latch.countDown(); } @Override public void onCompleted() { latch.countDown(); } }; grpcService.getStream( GetStreamRequest.newBuilder() .setStreamName(StreamName.newBuilder() .setNamespaceName(nsName) .setStreamName(streamName)) .build(), streamObserver); latch.await(); assertNull(exceptionHolder.get()); assertNotNull(resultHolder.get()); assertTrue(getResp == resultHolder.get()); verify(rangeService, times(1)).getStream(getReq); }
@Test public void testGetStreamFailure() throws Exception { RangeStoreService rangeService = mock(RangeStoreService.class); GrpcRootRangeService grpcService = new GrpcRootRangeService(rangeService); GetStreamResponse getResp = GetStreamResponse.newBuilder() .setCode(StatusCode.INTERNAL_SERVER_ERROR) .build(); GetStreamRequest getReq = createGetStreamRequest(nsName, streamName); when(rangeService.getStream(getReq)).thenReturn( FutureUtils.exception(CAUSE)); AtomicReference<GetStreamResponse> resultHolder = new AtomicReference<>(); AtomicReference<Throwable> exceptionHolder = new AtomicReference<>(); CountDownLatch latch = new CountDownLatch(1); StreamObserver<GetStreamResponse> streamObserver = new StreamObserver<GetStreamResponse>() { @Override public void onNext(GetStreamResponse resp) { resultHolder.set(resp); } @Override public void onError(Throwable t) { exceptionHolder.set(t); latch.countDown(); } @Override public void onCompleted() { latch.countDown(); } }; grpcService.getStream( GetStreamRequest.newBuilder() .setStreamName(StreamName.newBuilder() .setNamespaceName(nsName) .setStreamName(streamName)) .build(), streamObserver); latch.await(); assertNull(exceptionHolder.get()); assertNotNull(resultHolder.get()); assertEquals(getResp, resultHolder.get()); verify(rangeService, times(1)).getStream(getReq); }