### Question:
GrpcRootRangeService extends RootRangeServiceImplBase { @Override public void createStream(CreateStreamRequest request, StreamObserver<CreateStreamResponse> responseObserver) { rs.createStream(request).whenComplete( new ResponseHandler<CreateStreamResponse>(responseObserver) { @Override protected CreateStreamResponse createErrorResp(Throwable cause) { return CreateStreamResponse.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 testCreateStreamSuccess() throws Exception { RangeStoreService rangeService = mock(RangeStoreService.class); GrpcRootRangeService grpcService = new GrpcRootRangeService(rangeService); CreateStreamResponse createResp = CreateStreamResponse.newBuilder() .setCode(StatusCode.SUCCESS) .setStreamProps(streamProps) .build(); CreateStreamRequest createReq = createCreateStreamRequest(nsName, streamName, DEFAULT_STREAM_CONF); when(rangeService.createStream(createReq)).thenReturn( CompletableFuture.completedFuture(createResp)); AtomicReference<CreateStreamResponse> resultHolder = new AtomicReference<>(); AtomicReference<Throwable> exceptionHolder = new AtomicReference<>(); CountDownLatch latch = new CountDownLatch(1); StreamObserver<CreateStreamResponse> streamObserver = new StreamObserver<CreateStreamResponse>() { @Override public void onNext(CreateStreamResponse resp) { resultHolder.set(resp); } @Override public void onError(Throwable t) { exceptionHolder.set(t); latch.countDown(); } @Override public void onCompleted() { latch.countDown(); } }; grpcService.createStream( CreateStreamRequest.newBuilder() .setNsName(nsName) .setName(streamName) .setStreamConf(DEFAULT_STREAM_CONF) .build(), streamObserver); latch.await(); assertNull(exceptionHolder.get()); assertNotNull(resultHolder.get()); assertTrue(createResp == resultHolder.get()); verify(rangeService, times(1)).createStream(createReq); }
@Test public void testCreateStreamFailure() throws Exception { RangeStoreService rangeService = mock(RangeStoreService.class); GrpcRootRangeService grpcService = new GrpcRootRangeService(rangeService); CreateStreamResponse createResp = CreateStreamResponse.newBuilder() .setCode(StatusCode.INTERNAL_SERVER_ERROR) .build(); CreateStreamRequest createReq = createCreateStreamRequest(nsName, streamName, DEFAULT_STREAM_CONF); when(rangeService.createStream(createReq)).thenReturn( FutureUtils.exception(CAUSE)); AtomicReference<CreateStreamResponse> resultHolder = new AtomicReference<>(); AtomicReference<Throwable> exceptionHolder = new AtomicReference<>(); CountDownLatch latch = new CountDownLatch(1); StreamObserver<CreateStreamResponse> streamObserver = new StreamObserver<CreateStreamResponse>() { @Override public void onNext(CreateStreamResponse resp) { resultHolder.set(resp); } @Override public void onError(Throwable t) { exceptionHolder.set(t); latch.countDown(); } @Override public void onCompleted() { latch.countDown(); } }; grpcService.createStream( CreateStreamRequest.newBuilder() .setNsName(nsName) .setName(streamName) .setStreamConf(DEFAULT_STREAM_CONF) .build(), streamObserver); latch.await(); assertNull(exceptionHolder.get()); assertNotNull(resultHolder.get()); assertEquals(createResp, resultHolder.get()); verify(rangeService, times(1)).createStream(createReq); }