AggregatingJobServiceGateway implements JobServiceGateway { private Observable<JobQueryResult> findJobsWithCursorPagination(JobQuery request, Set<String> fields, CallMetadata callMetadata) { return aggregatingClient.call(JobManagementServiceGrpc::newStub, findJobsInCell(request, callMetadata)) .map(CellResponse::getResult) .map(this::addStackName) .reduce(this::combineJobResults) .map(combinedResults -> { Pair<List<Job>, Pagination> combinedPage = takeCombinedPage( request.getPage(), combinedResults.getItemsList(), combinedResults.getPagination(), JobManagerCursors.jobCursorOrderComparator(), JobManagerCursors::newCursorFrom ); if (!CollectionsExt.isNullOrEmpty(fields)) { combinedPage = combinedPage.mapLeft(jobs -> jobs.stream() .map(job -> ProtobufExt.copy(job, fields)) .collect(Collectors.toList()) ); } return JobQueryResult.newBuilder() .addAllItems(combinedPage.getLeft()) .setPagination(combinedPage.getRight()) .build(); }); } @Inject  AggregatingJobServiceGateway(GrpcConfiguration grpcConfiguration,
                                        TitusFederationConfiguration federationConfiguration,
                                        CellConnector connector,
                                        CellRouter router,
                                        AggregatingCellClient aggregatingClient,
                                        AggregatingJobManagementServiceHelper jobManagementServiceHelper); @Override Observable<String> createJob(JobDescriptor jobDescriptor, CallMetadata callMetadata); @Override Completable updateJobCapacity(JobCapacityUpdate request, CallMetadata callMetadata); @Override Completable updateJobCapacityWithOptionalAttributes(JobCapacityUpdateWithOptionalAttributes request, CallMetadata callMetadata); @Override Completable updateJobProcesses(JobProcessesUpdate request, CallMetadata callMetadata); @Override Completable updateJobStatus(JobStatusUpdate request, CallMetadata callMetadata); @Override Mono<Void> updateJobDisruptionBudget(JobDisruptionBudgetUpdate request, CallMetadata callMetadata); @Override Mono<Void> updateJobAttributes(JobAttributesUpdate request, CallMetadata callMetadata); @Override Mono<Void> deleteJobAttributes(JobAttributesDeleteRequest request, CallMetadata callMetadata); @Override Observable<Job> findJob(String jobId, CallMetadata callMetadata); @Override Observable<JobQueryResult> findJobs(JobQuery request, CallMetadata callMetadata); @Override Observable<JobChangeNotification> observeJob(String jobId, CallMetadata callMetadata); @Override Observable<JobChangeNotification> observeJobs(ObserveJobsQuery query, CallMetadata callMetadata); @Override Completable killJob(String jobId, CallMetadata callMetadata); @Override Observable<Task> findTask(String taskId, CallMetadata callMetadata); @Override Observable<TaskQueryResult> findTasks(TaskQuery request, CallMetadata callMetadata); @Override Completable killTask(TaskKillRequest request, CallMetadata callMetadata); @Override Completable updateTaskAttributes(TaskAttributesUpdate attributesUpdate, CallMetadata callMetadata); @Override Completable deleteTaskAttributes(TaskAttributesDeleteRequest deleteRequest, CallMetadata callMetadata); @Override Completable moveTask(TaskMoveRequest taskMoveRequest, CallMetadata callMetadata);  }
@Test public void findJobsWithCursorPagination() { Pair<List<Job>, List<Job>> cellSnapshots = generateTestJobs(); List<Job> allJobs = walkAllFindJobsPages(10); assertThat(allJobs).hasSize(cellSnapshots.getLeft().size() + cellSnapshots.getRight().size()); List<Job> expected = Stream.concat(cellSnapshots.getLeft().stream(), cellSnapshots.getRight().stream()) .sorted(JobManagerCursors.jobCursorOrderComparator()) .map(this::withStackName) .collect(Collectors.toList()); assertThat(allJobs).containsExactlyElementsOf(expected); }