@Test @SuppressWarnings("unchecked") public void invokesSchedulerForAllEnvironments() throws Exception { when(dataService.listEnvironments(ListEnvironmentsRequest.builder().cluster(CLUSTER).build())) .thenReturn( ListEnvironmentsResponse.builder() .environmentIds(Arrays.asList(FIRST_ENVIRONMENT_ID, SECOND_ENVIRONMENT_ID)) .build()); when(scheduler.callAsync(schedulerArgument.capture())) .thenReturn( CompletableFuture.completedFuture( new SchedulerOutput(CLUSTER_NAME, FIRST_ENVIRONMENT_ID, 1, 1))); ManagerHandler handler = new ManagerHandler(dataService, ecs, scheduler); handler.handleRequest(new ManagerInput(CLUSTER), null); assertThat( schedulerArgument.getAllValues(), contains( hasProperty("environmentId", is(FIRST_ENVIRONMENT_ID)), hasProperty("environmentId", is(SECOND_ENVIRONMENT_ID)))); }
@Override @SneakyThrows public ManagerOutput handleRequest(ManagerInput input, Context context) { log.debug("Manager request: {}", input); ListEnvironmentsResponse r = data.listEnvironments( ListEnvironmentsRequest.builder().cluster(input.getCluster()).build()); List<EnvironmentId> environments = r.getEnvironmentIds(); ClusterSnapshot state = ecs.snapshotState(input.getCluster().getClusterName()); Stream<CompletableFuture<SchedulerOutput>> pendingRequests = environments .stream() .map(environmentId -> scheduler.callAsync(new SchedulerInput(state, environmentId))); List<SchedulerOutput> outputs = pendingRequests.collect(CompletableFutures.joinList()).join(); return new ManagerOutput(input.getCluster(), outputs); }
ManagerHandler implements RequestHandler<ManagerInput, ManagerOutput> { @Override @SneakyThrows public ManagerOutput handleRequest(ManagerInput input, Context context) { log.debug("Manager request: {}", input); ListEnvironmentsResponse r = data.listEnvironments( ListEnvironmentsRequest.builder().cluster(input.getCluster()).build()); List<EnvironmentId> environments = r.getEnvironmentIds(); ClusterSnapshot state = ecs.snapshotState(input.getCluster().getClusterName()); Stream<CompletableFuture<SchedulerOutput>> pendingRequests = environments .stream() .map(environmentId -> scheduler.callAsync(new SchedulerInput(state, environmentId))); List<SchedulerOutput> outputs = pendingRequests.collect(CompletableFutures.joinList()).join(); return new ManagerOutput(input.getCluster(), outputs); } }
ManagerHandler implements RequestHandler<ManagerInput, ManagerOutput> { @Override @SneakyThrows public ManagerOutput handleRequest(ManagerInput input, Context context) { log.debug("Manager request: {}", input); ListEnvironmentsResponse r = data.listEnvironments( ListEnvironmentsRequest.builder().cluster(input.getCluster()).build()); List<EnvironmentId> environments = r.getEnvironmentIds(); ClusterSnapshot state = ecs.snapshotState(input.getCluster().getClusterName()); Stream<CompletableFuture<SchedulerOutput>> pendingRequests = environments .stream() .map(environmentId -> scheduler.callAsync(new SchedulerInput(state, environmentId))); List<SchedulerOutput> outputs = pendingRequests.collect(CompletableFutures.joinList()).join(); return new ManagerOutput(input.getCluster(), outputs); }  }
ManagerHandler implements RequestHandler<ManagerInput, ManagerOutput> { @Override @SneakyThrows public ManagerOutput handleRequest(ManagerInput input, Context context) { log.debug("Manager request: {}", input); ListEnvironmentsResponse r = data.listEnvironments( ListEnvironmentsRequest.builder().cluster(input.getCluster()).build()); List<EnvironmentId> environments = r.getEnvironmentIds(); ClusterSnapshot state = ecs.snapshotState(input.getCluster().getClusterName()); Stream<CompletableFuture<SchedulerOutput>> pendingRequests = environments .stream() .map(environmentId -> scheduler.callAsync(new SchedulerInput(state, environmentId))); List<SchedulerOutput> outputs = pendingRequests.collect(CompletableFutures.joinList()).join(); return new ManagerOutput(input.getCluster(), outputs); }  @Override @SneakyThrows // TODO add checked exception handling ManagerOutput handleRequest(ManagerInput input, Context context); }
ManagerHandler implements RequestHandler<ManagerInput, ManagerOutput> { @Override @SneakyThrows public ManagerOutput handleRequest(ManagerInput input, Context context) { log.debug("Manager request: {}", input); ListEnvironmentsResponse r = data.listEnvironments( ListEnvironmentsRequest.builder().cluster(input.getCluster()).build()); List<EnvironmentId> environments = r.getEnvironmentIds(); ClusterSnapshot state = ecs.snapshotState(input.getCluster().getClusterName()); Stream<CompletableFuture<SchedulerOutput>> pendingRequests = environments .stream() .map(environmentId -> scheduler.callAsync(new SchedulerInput(state, environmentId))); List<SchedulerOutput> outputs = pendingRequests.collect(CompletableFutures.joinList()).join(); return new ManagerOutput(input.getCluster(), outputs); }  @Override @SneakyThrows // TODO add checked exception handling ManagerOutput handleRequest(ManagerInput input, Context context);  }