AggregatingAutoScalingService implements AutoScalingService { @Override public Observable<GetPolicyResult> getJobScalingPolicies(JobId jobId, CallMetadata callMetadata) { return aggregatingClient.callExpectingErrors(AutoScalingServiceGrpc::newStub, getJobScalingPoliciesInCell(jobId, callMetadata)) .filter(response -> response.getResult().hasError() || response.getResult().getValue().getItemsCount() > 0) .reduce(ResponseMerger.emptyResponseMarker(), ResponseMerger.singleValue()) .filter(ResponseMerger::isNotEmptyResponseMarker) .flatMap(response -> response.getResult() .map(Observable::just) .onErrorGet(Observable::error) ) .switchIfEmpty(Observable.just(GetPolicyResult.newBuilder().build())); } @Inject  AggregatingAutoScalingService(CellConnector connector,
                                         GrpcConfiguration configuration,
                                         AggregatingJobManagementServiceHelper jobManagementServiceHelper,
                                         AggregatingCellClient aggregatingClient); @Override Observable<GetPolicyResult> getJobScalingPolicies(JobId jobId, CallMetadata callMetadata); @Override Observable<ScalingPolicyID> setAutoScalingPolicy(PutPolicyRequest request, CallMetadata callMetadata); @Override Observable<GetPolicyResult> getScalingPolicy(ScalingPolicyID request, CallMetadata callMetadata); @Override Observable<GetPolicyResult> getAllScalingPolicies(CallMetadata callMetadata); @Override Completable deleteAutoScalingPolicy(DeletePolicyRequest request, CallMetadata callMetadata); @Override Completable updateAutoScalingPolicy(UpdatePolicyRequest request, CallMetadata callMetadata);  }
@Test public void getPoliciesForJobFromTwoCells() { ScalingPolicyID policy1 = ScalingPolicyID.newBuilder().setId(POLICY_1).build(); ScalingPolicyID policy2 = ScalingPolicyID.newBuilder().setId(POLICY_2).build(); ScalingPolicyResult policyOneResult = ScalingPolicyResult.newBuilder().setId(policy1).setJobId(JOB_1).build(); ScalingPolicyResult policyTwoResult = ScalingPolicyResult.newBuilder().setId(policy2).setJobId(JOB_2).build(); CellWithPolicies cellOneService = new CellWithPolicies(Collections.singletonList(policyOneResult)); CellWithPolicies cellTwoService = new CellWithPolicies(Collections.singletonList(policyTwoResult)); cellOne.getServiceRegistry().addService(cellOneService); cellTwo.getServiceRegistry().addService(cellTwoService); AssertableSubscriber<GetPolicyResult> testSubscriber = service.getJobScalingPolicies(JobId.newBuilder().setId(JOB_2).build(), JUNIT_REST_CALL_METADATA).test(); testSubscriber.awaitValueCount(1, 1, TimeUnit.SECONDS); List<GetPolicyResult> onNextEvents = testSubscriber.getOnNextEvents(); assertThat(onNextEvents).isNotNull(); assertThat(onNextEvents.size()).isEqualTo(1); assertThat(onNextEvents.get(0).getItemsCount()).isEqualTo(1); assertThat(onNextEvents.get(0).getItems(0).getJobId()).isEqualTo(JOB_2); testSubscriber = service.getJobScalingPolicies(JobId.newBuilder().setId("badID").build(), JUNIT_REST_CALL_METADATA).test(); testSubscriber.awaitTerminalEvent(1, TimeUnit.SECONDS); testSubscriber.assertNoErrors(); onNextEvents = testSubscriber.getOnNextEvents(); assertThat(onNextEvents).isNotNull(); assertThat(onNextEvents.size()).isEqualTo(1); assertThat(onNextEvents.get(0).getItemsCount()).isEqualTo(0); }