@Test public void getLoadBalancersForJob() { JobLoadBalancer jobLoadBalancer1 = new JobLoadBalancer(JOB_1, LB_1); JobLoadBalancer jobLoadBalancer2 = new JobLoadBalancer(JOB_2, LB_2); final CellWithLoadBalancers cellWithLoadBalancersOne = new CellWithLoadBalancers(singletonList(jobLoadBalancer1)); final CellWithLoadBalancers cellWithLoadBalancersTwo = new CellWithLoadBalancers(singletonList(jobLoadBalancer2)); cellOne.getServiceRegistry().addService(cellWithLoadBalancersOne); cellTwo.getServiceRegistry().addService(cellWithLoadBalancersTwo); final AssertableSubscriber<GetJobLoadBalancersResult> resultSubscriber = service.getLoadBalancers(JobId.newBuilder().setId(JOB_2).build(), JUNIT_REST_CALL_METADATA).test(); resultSubscriber.awaitValueCount(1, 1, TimeUnit.SECONDS); resultSubscriber.assertNoErrors(); final List<GetJobLoadBalancersResult> onNextEvents = resultSubscriber.getOnNextEvents(); assertThat(onNextEvents.size()).isEqualTo(1); assertThat(onNextEvents.get(0).getJobId()).isEqualTo(JOB_2); assertThat(onNextEvents.get(0).getLoadBalancersCount()).isEqualTo(1); assertThat(onNextEvents.get(0).getLoadBalancers(0).getId()).isEqualTo(LB_2); }
@Override public Observable<GetJobLoadBalancersResult> getLoadBalancers(JobId jobId, CallMetadata callMetadata) { ClientCall<GetJobLoadBalancersResult> getLoadBalancersForJob = (client, responseObserver) -> wrap(client, callMetadata).getJobLoadBalancers(jobId, responseObserver); return aggregatingClient.callExpectingErrors(LoadBalancerServiceGrpc::newStub, getLoadBalancersForJob) .filter(response -> response.getResult().hasError() || response.getResult().getValue().getLoadBalancersCount() > 0) .reduce(ResponseMerger.emptyResponseMarker(), ResponseMerger.singleValue()) .filter(ResponseMerger::isNotEmptyResponseMarker) .flatMap(response -> response.getResult().map(Observable::just).onErrorGet(Observable::error)) .switchIfEmpty(Observable.just(GetJobLoadBalancersResult.newBuilder().setJobId(jobId.getId()).build())); }
AggregatingLoadbalancerService implements LoadBalancerService { @Override public Observable<GetJobLoadBalancersResult> getLoadBalancers(JobId jobId, CallMetadata callMetadata) { ClientCall<GetJobLoadBalancersResult> getLoadBalancersForJob = (client, responseObserver) -> wrap(client, callMetadata).getJobLoadBalancers(jobId, responseObserver); return aggregatingClient.callExpectingErrors(LoadBalancerServiceGrpc::newStub, getLoadBalancersForJob) .filter(response -> response.getResult().hasError() || response.getResult().getValue().getLoadBalancersCount() > 0) .reduce(ResponseMerger.emptyResponseMarker(), ResponseMerger.singleValue()) .filter(ResponseMerger::isNotEmptyResponseMarker) .flatMap(response -> response.getResult().map(Observable::just).onErrorGet(Observable::error)) .switchIfEmpty(Observable.just(GetJobLoadBalancersResult.newBuilder().setJobId(jobId.getId()).build())); } }
AggregatingLoadbalancerService implements LoadBalancerService { @Override public Observable<GetJobLoadBalancersResult> getLoadBalancers(JobId jobId, CallMetadata callMetadata) { ClientCall<GetJobLoadBalancersResult> getLoadBalancersForJob = (client, responseObserver) -> wrap(client, callMetadata).getJobLoadBalancers(jobId, responseObserver); return aggregatingClient.callExpectingErrors(LoadBalancerServiceGrpc::newStub, getLoadBalancersForJob) .filter(response -> response.getResult().hasError() || response.getResult().getValue().getLoadBalancersCount() > 0) .reduce(ResponseMerger.emptyResponseMarker(), ResponseMerger.singleValue()) .filter(ResponseMerger::isNotEmptyResponseMarker) .flatMap(response -> response.getResult().map(Observable::just).onErrorGet(Observable::error)) .switchIfEmpty(Observable.just(GetJobLoadBalancersResult.newBuilder().setJobId(jobId.getId()).build())); } @Inject  AggregatingLoadbalancerService(CellConnector connector,
                                          CallMetadataResolver callMetadataResolver,
                                          GrpcConfiguration grpcConfiguration,
                                          AggregatingCellClient aggregatingClient,
                                          AggregatingJobManagementServiceHelper jobManagementServiceHelper); }
AggregatingLoadbalancerService implements LoadBalancerService { @Override public Observable<GetJobLoadBalancersResult> getLoadBalancers(JobId jobId, CallMetadata callMetadata) { ClientCall<GetJobLoadBalancersResult> getLoadBalancersForJob = (client, responseObserver) -> wrap(client, callMetadata).getJobLoadBalancers(jobId, responseObserver); return aggregatingClient.callExpectingErrors(LoadBalancerServiceGrpc::newStub, getLoadBalancersForJob) .filter(response -> response.getResult().hasError() || response.getResult().getValue().getLoadBalancersCount() > 0) .reduce(ResponseMerger.emptyResponseMarker(), ResponseMerger.singleValue()) .filter(ResponseMerger::isNotEmptyResponseMarker) .flatMap(response -> response.getResult().map(Observable::just).onErrorGet(Observable::error)) .switchIfEmpty(Observable.just(GetJobLoadBalancersResult.newBuilder().setJobId(jobId.getId()).build())); } @Inject  AggregatingLoadbalancerService(CellConnector connector,
                                          CallMetadataResolver callMetadataResolver,
                                          GrpcConfiguration grpcConfiguration,
                                          AggregatingCellClient aggregatingClient,
                                          AggregatingJobManagementServiceHelper jobManagementServiceHelper); @Override Observable<GetAllLoadBalancersResult> getAllLoadBalancers(GetAllLoadBalancersRequest request, CallMetadata callMetadata); @Override Observable<GetJobLoadBalancersResult> getLoadBalancers(JobId jobId, CallMetadata callMetadata); @Override Completable addLoadBalancer(AddLoadBalancerRequest request, CallMetadata callMetadata); @Override Completable removeLoadBalancer(RemoveLoadBalancerRequest removeLoadBalancerRequest, CallMetadata callMetadata); }
AggregatingLoadbalancerService implements LoadBalancerService { @Override public Observable<GetJobLoadBalancersResult> getLoadBalancers(JobId jobId, CallMetadata callMetadata) { ClientCall<GetJobLoadBalancersResult> getLoadBalancersForJob = (client, responseObserver) -> wrap(client, callMetadata).getJobLoadBalancers(jobId, responseObserver); return aggregatingClient.callExpectingErrors(LoadBalancerServiceGrpc::newStub, getLoadBalancersForJob) .filter(response -> response.getResult().hasError() || response.getResult().getValue().getLoadBalancersCount() > 0) .reduce(ResponseMerger.emptyResponseMarker(), ResponseMerger.singleValue()) .filter(ResponseMerger::isNotEmptyResponseMarker) .flatMap(response -> response.getResult().map(Observable::just).onErrorGet(Observable::error)) .switchIfEmpty(Observable.just(GetJobLoadBalancersResult.newBuilder().setJobId(jobId.getId()).build())); } @Inject  AggregatingLoadbalancerService(CellConnector connector,
                                          CallMetadataResolver callMetadataResolver,
                                          GrpcConfiguration grpcConfiguration,
                                          AggregatingCellClient aggregatingClient,
                                          AggregatingJobManagementServiceHelper jobManagementServiceHelper); @Override Observable<GetAllLoadBalancersResult> getAllLoadBalancers(GetAllLoadBalancersRequest request, CallMetadata callMetadata); @Override Observable<GetJobLoadBalancersResult> getLoadBalancers(JobId jobId, CallMetadata callMetadata); @Override Completable addLoadBalancer(AddLoadBalancerRequest request, CallMetadata callMetadata); @Override Completable removeLoadBalancer(RemoveLoadBalancerRequest removeLoadBalancerRequest, CallMetadata callMetadata);  }