AggregatingAutoScalingService implements AutoScalingService { @Override public Observable<GetPolicyResult> getAllScalingPolicies(CallMetadata callMetadata) { return aggregatingClient.call(AutoScalingServiceGrpc::newStub, getAllScalingPoliciesInCell(callMetadata)) .map(CellResponse::getResult) .reduce((acc, next) -> GetPolicyResult.newBuilder() .addAllItems(acc.getItemsList()) .addAllItems(next.getItemsList()) .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 getPoliciesFromTwoCells() { ScalingPolicyID policy1 = ScalingPolicyID.newBuilder().setId(POLICY_1).build(); ScalingPolicyID policy2 = ScalingPolicyID.newBuilder().setId(POLICY_2).build(); ScalingPolicyResult policyOneResult = ScalingPolicyResult.newBuilder().setId(policy1).build(); ScalingPolicyResult policyTwoResult = ScalingPolicyResult.newBuilder().setId(policy2).build(); CellWithPolicies cellOneService = new CellWithPolicies(Collections.singletonList(policyOneResult)); CellWithPolicies cellTwoService = new CellWithPolicies(Collections.singletonList(policyTwoResult)); cellOne.getServiceRegistry().addService(cellOneService); cellTwo.getServiceRegistry().addService(cellTwoService); final AssertableSubscriber<GetPolicyResult> testSubscriber = service.getAllScalingPolicies(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(2); }
@Test public void getPoliciesFromTwoCellsWithOneFailing() { ScalingPolicyID policy1 = ScalingPolicyID.newBuilder().setId(POLICY_1).build(); ScalingPolicyResult policyOneResult = ScalingPolicyResult.newBuilder().setId(policy1).build(); CellWithPolicies cellOneService = new CellWithPolicies(Collections.singletonList(policyOneResult)); CellWithFailingAutoscalingService badCell = new CellWithFailingAutoscalingService(); cellOne.getServiceRegistry().addService(cellOneService); cellTwo.getServiceRegistry().addService(badCell); final AssertableSubscriber<GetPolicyResult> testSubscriber = service.getAllScalingPolicies(JUNIT_REST_CALL_METADATA).test(); testSubscriber.awaitTerminalEvent(1, TimeUnit.SECONDS); testSubscriber.assertNoValues(); testSubscriber.assertError(StatusRuntimeException.class); List<Throwable> onErrorEvents = testSubscriber.getOnErrorEvents(); assertThat(onErrorEvents.size()).isEqualTo(1); }