import com.spotify.futures.CompletableFutures; //导入依赖的package包/类
@Override
public CompletionStage<WorkflowState> updateWorkflowState(String componentId, String workflowId,
                                                          WorkflowState workflowState) {
  final HttpUrl.Builder urlBuilder = getUrlBuilder()
      .addPathSegment("workflows")
      .addPathSegment(componentId)
      .addPathSegment(workflowId)
      .addPathSegment("state");
  final ByteString payload;
  try {
    payload = serialize(workflowState);
  } catch (JsonProcessingException e) {
    return CompletableFutures.exceptionallyCompletedFuture(new RuntimeException(e));
  }
  return executeRequest(Request.forUri(urlBuilder.build().toString(), "PATCH")
                            .withPayload(payload), WorkflowState.class);
}
 

import com.spotify.futures.CompletableFutures; //导入依赖的package包/类
@Override
public CompletionStage<Void> triggerWorkflowInstance(String componentId,
                                                     String workflowId,
                                                     String parameter) {
  final HttpUrl.Builder urlBuilder = getUrlBuilder()
      .addPathSegment("scheduler")
      .addPathSegment("trigger");
  final WorkflowInstance workflowInstance = WorkflowInstance.create(
      WorkflowId.create(componentId, workflowId),
      parameter);
  try {
    final ByteString payload = serialize(workflowInstance);
    return executeRequest(
        Request.forUri(urlBuilder.build().toString(), "POST").withPayload(payload))
        .thenApply(response -> (Void) null);
  } catch (JsonProcessingException e) {
    return CompletableFutures.exceptionallyCompletedFuture(new RuntimeException(e));
  }
}
 

import com.spotify.futures.CompletableFutures; //导入依赖的package包/类
@Override
public CompletionStage<Void> haltWorkflowInstance(String componentId,
                                                  String workflowId,
                                                  String parameter) {
  final HttpUrl.Builder urlBuilder = getUrlBuilder()
      .addPathSegment("scheduler")
      .addPathSegment("halt");
  final WorkflowInstance workflowInstance = WorkflowInstance.create(
      WorkflowId.create(componentId, workflowId),
      parameter);
  try {
    final ByteString payload = serialize(workflowInstance);
    return executeRequest(
        Request.forUri(urlBuilder.build().toString(), "POST").withPayload(payload))
        .thenApply(response -> (Void) null);
  } catch (JsonProcessingException e) {
    return CompletableFutures.exceptionallyCompletedFuture(new RuntimeException(e));
  }
}
 

import com.spotify.futures.CompletableFutures; //导入依赖的package包/类
@Override
public CompletionStage<Void> retryWorkflowInstance(String componentId,
                                                   String workflowId,
                                                   String parameter) {
  final HttpUrl.Builder urlBuilder = getUrlBuilder()
      .addPathSegment("scheduler")
      .addPathSegment("retry");
  final WorkflowInstance workflowInstance = WorkflowInstance.create(
      WorkflowId.create(componentId, workflowId),
      parameter);
  try {
    final ByteString payload = serialize(workflowInstance);
    return executeRequest(
        Request.forUri(urlBuilder.build().toString(), "POST").withPayload(payload))
        .thenApply(response -> (Void) null);
  } catch (JsonProcessingException e) {
    return CompletableFutures.exceptionallyCompletedFuture(new RuntimeException(e));
  }
}
 

import com.spotify.futures.CompletableFutures; //导入依赖的package包/类
@Override
public CompletionStage<Backfill> backfillCreate(String componentId, String workflowId,
                                                String start, String end,
                                                int concurrency,
                                                String description) {
  final HttpUrl.Builder urlBuilder = getUrlBuilder().addPathSegment("backfills");
  try {
    final ByteString payload = serialize(BackfillInput.create(
        Instant.parse(start), Instant.parse(end), componentId, workflowId, concurrency,
        Optional.ofNullable(description)));
    return executeRequest(Request.forUri(urlBuilder.build().toString(), "POST")
        .withPayload(payload), Backfill.class);
  } catch (JsonProcessingException e) {
    return CompletableFutures.exceptionallyCompletedFuture(new RuntimeException(e));
  }
}
 

import com.spotify.futures.CompletableFutures; //导入依赖的package包/类
@Override
public CompletionStage<Backfill> backfillEditConcurrency(String backfillId, int concurrency) {
  return backfill(backfillId, false).thenCompose(backfillPayload -> {
    final Backfill editedBackfill = backfillPayload.backfill().builder()
        .concurrency(concurrency)
        .build();
    final HttpUrl.Builder urlBuilder = getUrlBuilder()
        .addPathSegment("backfills")
        .addPathSegment(backfillId);
    try {
      final ByteString payload = serialize(editedBackfill);
      return executeRequest(Request.forUri(urlBuilder.build().toString(), "PUT")
          .withPayload(payload), Backfill.class);
    } catch (JsonProcessingException e) {
      return CompletableFutures.exceptionallyCompletedFuture(new RuntimeException(e));
    }
  });
}
 

import com.spotify.futures.CompletableFutures; //导入依赖的package包/类
private CompletionStage<Response<ByteString>> haltActiveBackfillInstances(Backfill backfill, Client client) {
  return CompletableFutures.allAsList(
      retrieveBackfillStatuses(backfill).stream()
          .filter(BackfillResource::isActiveState)
          .map(RunStateData::workflowInstance)
          .map(workflowInstance -> haltActiveBackfillInstance(workflowInstance, client))
          .collect(toList()))
      .handle((result, throwable) -> {
        if (throwable != null || result.contains(Boolean.FALSE)) {
          return Response.forStatus(
              Status.INTERNAL_SERVER_ERROR
                  .withReasonPhrase(
                      "some active instances cannot be halted, however no new ones will be triggered"));
        } else {
          return Response.ok();
        }
      });
}
 

import com.spotify.futures.CompletableFutures; //导入依赖的package包/类
private CompletionStage<List<GetResult<V>>> multiget(List<byte[]> keys, int ttl) {
  final int size = keys.size();
  if (size == 0) {
    return CompletableFuture.completedFuture(Collections.<GetResult<V>>emptyList());
  }

  final List<List<byte[]>> keyPartition =
        Lists.partition(keys, MemcacheEncoder.MAX_MULTIGET_SIZE);
  final List<CompletionStage<List<GetResult<byte[]>>>> futureList =
        new ArrayList<>(keyPartition.size());

  for (final List<byte[]> part : keyPartition) {
    MultigetRequest request = MultigetRequest.create(part, ttl);
    futureList.add(rawMemcacheClient.send(request));
  }

  final CompletionStage<List<GetResult<byte[]>>> future =
      CompletableFutures.allAsList(futureList)
          .thenApply(Utils.flatten());

  metrics.measureMultigetFuture(future);
  return transformerUtil.decodeList(future);
}
 

import com.spotify.futures.CompletableFutures; //导入依赖的package包/类
private CompletionStage<List<GetResult<V>>> multiget(List<byte[]> keys, boolean withCas) {
  final int size = keys.size();
  if (size == 0) {
    return CompletableFuture.completedFuture(Collections.<GetResult<V>>emptyList());
  }

  final List<List<byte[]>> keyPartition =
          Lists.partition(keys, MemcacheEncoder.MAX_MULTIGET_SIZE);
  final List<CompletionStage<List<GetResult<byte[]>>>> futureList =
          new ArrayList<>(keyPartition.size());

  for (final List<byte[]> part : keyPartition) {
    MultigetRequest request = MultigetRequest.create(part, withCas);
    futureList.add(rawMemcacheClient.send(request));
  }

  final CompletionStage<List<GetResult<byte[]>>> future =
      ((CompletionStage<List<List<GetResult<byte[]>>>>) CompletableFutures.allAsList(futureList))
          .thenApply(Utils.flatten());

  metrics.measureMultigetFuture(future);
  return transformerUtil.decodeList(future);
}
 

import com.spotify.futures.CompletableFutures; //导入依赖的package包/类
private CompletableFuture<Void> checkAndUpdateHealthyServers() {
    List<ServerConnection> checkedServers = updateServerList();

    CompletableFuture<List<Boolean>> healthCheckResults = CompletableFutures.successfulAsList(
            checkedServers.stream()
                          .map(connection -> connection.healthChecker.isHealthy(connection.endpoint()))
                          .collect(toImmutableList()),
            t -> false);
    return healthCheckResults.handle(voidFunction((result, thrown) -> {
        ImmutableList.Builder<Endpoint> newHealthyEndpoints = ImmutableList.builder();
        for (int i = 0; i < result.size(); i++) {
            if (result.get(i)) {
                newHealthyEndpoints.add(checkedServers.get(i).endpoint());
            }
        }
        setEndpoints(newHealthyEndpoints.build());
    }));
}
 

import com.spotify.futures.CompletableFutures; //导入依赖的package包/类
static <T extends Options, S extends Output> CompletionStage<S> sendRequest(
    final Client client,
    final String uri,
    final T options,
    final Class<S> cls) throws RktLauncherRemoteException {
  final ByteString payload;
  try {
    payload = ByteString.of(Json.serialize(options));
  } catch (JsonProcessingException e) {
    return CompletableFutures.exceptionallyCompletedFuture(
        new RktLauncherRemoteException("failed to serialize payload", e));
  }
  return sendRequest(client, Request.forUri(uri, DEFAULT_HTTP_METHOD).withPayload(payload), cls);
}
 

import com.spotify.futures.CompletableFutures; //导入依赖的package包/类
@Override
public CompletableFuture<Revision> normalize(Revision revision) {
    try {
        return CompletableFuture.completedFuture(blockingNormalize(revision));
    } catch (RevisionNotFoundException e) {
        return CompletableFutures.exceptionallyCompletedFuture(e);
    }
}
 

import com.spotify.futures.CompletableFutures; //导入依赖的package包/类
public CompletableFuture<TaggedResponse> continueAfterResponse(ImapResponse imapResponse, Throwable throwable) {
  if (throwable != null) {
    return CompletableFutures.exceptionallyCompletedFuture(throwable);
  }

  String toEncode = NUL + user + NUL + password;
  String base64EncodedAuthRequest = BaseEncoding.base64().encode(toEncode.getBytes(StandardCharsets.UTF_8));
  return imapClient.send(new SimpleStringCommand(base64EncodedAuthRequest));
}
 

import com.spotify.futures.CompletableFutures; //导入依赖的package包/类
@Override
public CompletionStage<Workflow> createOrUpdateWorkflow(String componentId, WorkflowConfiguration workflowConfig) {
  final HttpUrl.Builder urlBuilder = getUrlBuilder()
      .addPathSegment("workflows")
      .addPathSegment(componentId);
  final ByteString payload;
  try {
    payload = serialize(workflowConfig);
  } catch (JsonProcessingException e) {
    return CompletableFutures.exceptionallyCompletedFuture(new RuntimeException(e));
  }
  final Request request = Request.forUri(urlBuilder.build().toString(), "POST").withPayload(payload);
  return executeRequest(request, Workflow.class);
}
 

import com.spotify.futures.CompletableFutures; //导入依赖的package包/类
@Override
public CompletionStage<Resource> resourceCreate(String resourceId, int concurrency) {
  final HttpUrl.Builder urlBuilder = getUrlBuilder().addPathSegment("resources");
  try {
    final ByteString payload = serialize(Resource.create(resourceId, concurrency));
    return executeRequest(Request.forUri(urlBuilder.build().toString(), "POST")
        .withPayload(payload), Resource.class);
  } catch (JsonProcessingException e) {
    return CompletableFutures.exceptionallyCompletedFuture(new RuntimeException(e));
  }
}
 

import com.spotify.futures.CompletableFutures; //导入依赖的package包/类
@Override
public CompletionStage<Resource> resourceEdit(String resourceId, int concurrency) {
  final HttpUrl.Builder urlBuilder = getUrlBuilder()
      .addPathSegment("resources")
      .addPathSegment(resourceId);
  try {
    final ByteString payload = serialize(Resource.create(resourceId, concurrency));
    return executeRequest(Request.forUri(urlBuilder.build().toString(), "PUT")
        .withPayload(payload), Resource.class);
  } catch (JsonProcessingException e) {
    return CompletableFutures.exceptionallyCompletedFuture(new RuntimeException(e));
  }
}
 

import com.spotify.futures.CompletableFutures; //导入依赖的package包/类
private CompletionStage<Response<ByteString>> executeRequest(final Request request) {
  final Optional<String> authToken;
  try {
    authToken = auth.getToken(this.apiHost.toString());
  } catch (IOException | GeneralSecurityException e) {
    // Credential probably invalid, configured wrongly or the token request failed.
    return CompletableFutures.exceptionallyCompletedFuture(
        new ClientErrorException("Authentication failure: " + e.getMessage(), e));
  }
  return client.send(decorateRequest(request, authToken)).handle((response, e) -> {
    if (e != null) {
      final Throwable rootCause = Throwables.getRootCause(e);
      if (rootCause instanceof SocketTimeoutException) {
        throw new ClientErrorException("Connection failed: " + rootCause.getMessage() + ": " + apiHost, e);
      } else {
        throw new ClientErrorException("Request failed: " + request, e);
      }
    } else {
      switch (response.status().family()) {
        case SUCCESSFUL:
          return response;
        default:
          final String message = response.status().code() + " " + response.status().reasonPhrase();
          throw new ApiErrorException(message, response.status().code(), authToken.isPresent());
      }
    }
  });
}
 

import com.spotify.futures.CompletableFutures; //导入依赖的package包/类
@Test
public void testClientUnknownError() {
  final NullPointerException exception = new NullPointerException();
  when(client.triggerWorkflowInstance(any(), any(), any()))
      .thenReturn(CompletableFutures.exceptionallyCompletedFuture(exception));

  try {
    CliMain.run(cliContext, "t", "foo", "bar", "2017-01-02");
    fail();
  } catch (CliExitException e) {
    assertThat(e.status(), is(ExitStatus.ClientError));
  }

  verify(cliOutput).printError(Throwables.getStackTraceAsString(exception));
}
 

import com.spotify.futures.CompletableFutures; //导入依赖的package包/类
CompletableFuture<Chunk> createCompletableFuture() {
    LoadState currentState = state.get();
    if (currentState == LoadState.COMPLETED) {
        return CompletableFuture.completedFuture(loaded.get());
    } else if (currentState == LoadState.EXCEPTIONAL) {
        return CompletableFutures.exceptionallyCompletedFuture(loadException.get());
    }

    CompletableFuture<Chunk> completableFuture = new CompletableFuture<>();
    futuresToComplete.add(completableFuture);
    return completableFuture;
}
 

import com.spotify.futures.CompletableFutures; //导入依赖的package包/类
private CompletableFuture<List<Chunk>> getChunksForRadius(int radius) {
    // Get current player's position in chunk coordinates.
    int chunkX = getPosition().getFloorX() >> 4;
    int chunkZ = getPosition().getFloorZ() >> 4;

    // Now get and send chunk data.
    Set<Vector2i> chunksForRadius = new HashSet<>();
    List<CompletableFuture<Chunk>> completableFutures = new ArrayList<>();

    for (int x = -radius; x <= radius; x++) {
        for (int z = -radius; z <= radius; z++) {
            int newChunkX = chunkX + x, newChunkZ = chunkZ + z;
            Vector2i chunkCoords = new Vector2i(newChunkX, newChunkZ);
            chunksForRadius.add(chunkCoords);

            if (!sentChunks.add(chunkCoords)) {
                // Already sent, don't need to resend.
                continue;
            }

            completableFutures.add(getLevel().getChunk(newChunkX, newChunkZ));
        }
    }

    sentChunks.retainAll(chunksForRadius);

    return CompletableFutures.allAsList(completableFutures);
}