AggregatingAutoScalingService implements AutoScalingService { @Override public Observable<GetPolicyResult> getScalingPolicy(ScalingPolicyID request, CallMetadata callMetadata) { return getScalingPolicyInAllCells(request, callMetadata).map(CellResponse::getResult); } @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 getPolicyByIdFromTwoCells() { 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); AssertableSubscriber<GetPolicyResult> testSubscriber = service.getScalingPolicy( ScalingPolicyID.newBuilder().setId(POLICY_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); testSubscriber = service.getScalingPolicy(ScalingPolicyID.newBuilder().setId("badID").build(), JUNIT_REST_CALL_METADATA).test(); testSubscriber.awaitTerminalEvent(1, TimeUnit.SECONDS); testSubscriber.assertError(StatusRuntimeException.class); testSubscriber.assertNoValues(); List<Throwable> onErrorEvents = testSubscriber.getOnErrorEvents(); assertThat(onErrorEvents).isNotNull(); assertThat(onErrorEvents).hasSize(1); assertThat(Status.fromThrowable(onErrorEvents.get(0)).getCode()).isEqualTo(Status.CANCELLED.getCode()); }