@Test public void getAllLoadBalancersWithPagination() { final List<JobLoadBalancer> jobLoadBalancersOne = Observable.range(1, 5) .map(i -> new JobLoadBalancer(JOB_1, String.format("LB_1_%d", i.intValue()))) .toList().toBlocking().first(); final List<JobLoadBalancer> jobLoadBalancersTwo = Observable.range(1, 5) .map(i -> new JobLoadBalancer(JOB_2, String.format("LB_2_%d", i.intValue()))) .toList().toBlocking().first(); final CellWithLoadBalancers cellWithLoadBalancersOne = new CellWithLoadBalancers(jobLoadBalancersOne); final CellWithLoadBalancers cellWithLoadBalancersTwo = new CellWithLoadBalancers(jobLoadBalancersTwo); cellOne.getServiceRegistry().addService(cellWithLoadBalancersOne); cellTwo.getServiceRegistry().addService(cellWithLoadBalancersTwo); final List<JobLoadBalancer> results = ServiceTests.walkAllPages( 2, request -> service.getAllLoadBalancers(request, JUNIT_REST_CALL_METADATA), page -> GetAllLoadBalancersRequest.newBuilder().setPage(page).build(), GetAllLoadBalancersResult::getPagination, this::buildJobLoadBalancerList ); assertThat(results).hasSize(10); }
@Override public Observable<GetAllLoadBalancersResult> getAllLoadBalancers(GetAllLoadBalancersRequest request, CallMetadata callMetadata) { if (request.getPage().getPageSize() <= 0) { return Observable.just(GetAllLoadBalancersResult.newBuilder() .setPagination(emptyGrpcPagination(request.getPage())) .build()); } if (StringExt.isNotEmpty(request.getPage().getCursor()) || request.getPage().getPageNumber() == 0) { return findLoadBalancersWithCursorPagination(request, callMetadata); } return Observable.error(TitusServiceException.invalidArgument("pageNumbers are not supported, please use cursors")); }
AggregatingLoadbalancerService implements LoadBalancerService { @Override public Observable<GetAllLoadBalancersResult> getAllLoadBalancers(GetAllLoadBalancersRequest request, CallMetadata callMetadata) { if (request.getPage().getPageSize() <= 0) { return Observable.just(GetAllLoadBalancersResult.newBuilder() .setPagination(emptyGrpcPagination(request.getPage())) .build()); } if (StringExt.isNotEmpty(request.getPage().getCursor()) || request.getPage().getPageNumber() == 0) { return findLoadBalancersWithCursorPagination(request, callMetadata); } return Observable.error(TitusServiceException.invalidArgument("pageNumbers are not supported, please use cursors")); } }
AggregatingLoadbalancerService implements LoadBalancerService { @Override public Observable<GetAllLoadBalancersResult> getAllLoadBalancers(GetAllLoadBalancersRequest request, CallMetadata callMetadata) { if (request.getPage().getPageSize() <= 0) { return Observable.just(GetAllLoadBalancersResult.newBuilder() .setPagination(emptyGrpcPagination(request.getPage())) .build()); } if (StringExt.isNotEmpty(request.getPage().getCursor()) || request.getPage().getPageNumber() == 0) { return findLoadBalancersWithCursorPagination(request, callMetadata); } return Observable.error(TitusServiceException.invalidArgument("pageNumbers are not supported, please use cursors")); } @Inject  AggregatingLoadbalancerService(CellConnector connector,
                                          CallMetadataResolver callMetadataResolver,
                                          GrpcConfiguration grpcConfiguration,
                                          AggregatingCellClient aggregatingClient,
                                          AggregatingJobManagementServiceHelper jobManagementServiceHelper); }
AggregatingLoadbalancerService implements LoadBalancerService { @Override public Observable<GetAllLoadBalancersResult> getAllLoadBalancers(GetAllLoadBalancersRequest request, CallMetadata callMetadata) { if (request.getPage().getPageSize() <= 0) { return Observable.just(GetAllLoadBalancersResult.newBuilder() .setPagination(emptyGrpcPagination(request.getPage())) .build()); } if (StringExt.isNotEmpty(request.getPage().getCursor()) || request.getPage().getPageNumber() == 0) { return findLoadBalancersWithCursorPagination(request, callMetadata); } return Observable.error(TitusServiceException.invalidArgument("pageNumbers are not supported, please use cursors")); } @Inject  AggregatingLoadbalancerService(CellConnector connector,
                                          CallMetadataResolver callMetadataResolver,
                                          GrpcConfiguration grpcConfiguration,
                                          AggregatingCellClient aggregatingClient,
                                          AggregatingJobManagementServiceHelper jobManagementServiceHelper); @Override Observable<GetAllLoadBalancersResult> getAllLoadBalancers(GetAllLoadBalancersRequest request, CallMetadata callMetadata); @Override Observable<GetJobLoadBalancersResult> getLoadBalancers(JobId jobId, CallMetadata callMetadata); @Override Completable addLoadBalancer(AddLoadBalancerRequest request, CallMetadata callMetadata); @Override Completable removeLoadBalancer(RemoveLoadBalancerRequest removeLoadBalancerRequest, CallMetadata callMetadata); }
AggregatingLoadbalancerService implements LoadBalancerService { @Override public Observable<GetAllLoadBalancersResult> getAllLoadBalancers(GetAllLoadBalancersRequest request, CallMetadata callMetadata) { if (request.getPage().getPageSize() <= 0) { return Observable.just(GetAllLoadBalancersResult.newBuilder() .setPagination(emptyGrpcPagination(request.getPage())) .build()); } if (StringExt.isNotEmpty(request.getPage().getCursor()) || request.getPage().getPageNumber() == 0) { return findLoadBalancersWithCursorPagination(request, callMetadata); } return Observable.error(TitusServiceException.invalidArgument("pageNumbers are not supported, please use cursors")); } @Inject  AggregatingLoadbalancerService(CellConnector connector,
                                          CallMetadataResolver callMetadataResolver,
                                          GrpcConfiguration grpcConfiguration,
                                          AggregatingCellClient aggregatingClient,
                                          AggregatingJobManagementServiceHelper jobManagementServiceHelper); @Override Observable<GetAllLoadBalancersResult> getAllLoadBalancers(GetAllLoadBalancersRequest request, CallMetadata callMetadata); @Override Observable<GetJobLoadBalancersResult> getLoadBalancers(JobId jobId, CallMetadata callMetadata); @Override Completable addLoadBalancer(AddLoadBalancerRequest request, CallMetadata callMetadata); @Override Completable removeLoadBalancer(RemoveLoadBalancerRequest removeLoadBalancerRequest, CallMetadata callMetadata);  }