class 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 public  AggregatingAutoScalingService(CellConnector connector,
                                         GrpcConfiguration configuration,
                                         AggregatingJobManagementServiceHelper jobManagementServiceHelper,
                                         AggregatingCellClient aggregatingClient);

}