@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); }
@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() ); }
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() ); } }
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); }
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); }
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);  }