Java 类java.util.concurrent.CompletableFuture 实例源码

@Override
public <M, R> CompletableFuture<R> sendAndReceive(M message,
                                                  MessageSubject subject,
                                                  Function<M, byte[]> encoder,
                                                  Function<byte[], R> decoder,
                                                  NodeId toNodeId) {
    checkPermission(CLUSTER_WRITE);
    try {
        ClusterMessage envelope = new ClusterMessage(
                clusterService.getLocalNode().id(),
                subject,
                timeFunction(encoder, subjectMeteringAgent, SERIALIZING).
                        apply(message));
        return sendAndReceive(subject, envelope.getBytes(), toNodeId).
                thenApply(bytes -> timeFunction(decoder, subjectMeteringAgent, DESERIALIZING).apply(bytes));
    } catch (Exception e) {
        return Tools.exceptionalFuture(e);
    }
}

@Override
public Future<Long> getAmountOfFinishedProcessesBetweenForUser(final LocalDateTime start,
    final LocalDateTime end, final Long userId) {
  final CompletableFuture<Long> future = new CompletableFuture<>();

  final ActorRef analysisActor = getAnalysisActor();

  PatternsCS
      .ask(analysisActor, new FinishedProcessesInRangeForUserMessage.Request(start, end, userId),
          Global.TIMEOUT)
      .toCompletableFuture()
      .thenApply(obj -> (FinishedProcessesInRangeForUserMessage.Response) obj)
      .whenComplete((msg, exc) -> future.complete(msg.getAmount()));

  return future;
}

private ProductLoader mockLoader() {
    ProductLoader loader = mock(ProductLoader.class);
    CompletableFuture<Collection<ProductLinkages>> f = new CompletableFuture<>();
    when(loader.getAllProducts()).thenReturn(f);
    Thread t = new Thread(() -> {
        try {
            Thread.sleep(50);
            f.complete(ImmutableList.of(
                    mockProductLinkages(),
                    mockProductLinkages()
            ));
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    });
    t.start();
    return loader;
}

/**
 * Verifies that continuations scheduled on a future can be executed inline with the specified completing action.
 *
 * @param antecedent The future to test.
 * @param completingAction The action that results in the synchronous completion of the future.
 */
protected static void verifyCanInlineContinuations(@NotNull CompletableFuture<?> antecedent, @NotNull Runnable completingAction) {
    Requires.notNull(antecedent, "antecedent");
    Requires.notNull(completingAction, "completingAction");

    Thread callingThread = Thread.currentThread();
    CompletableFuture<Void> continuation = antecedent.handle((result, exception) -> {
        Assert.assertSame(callingThread, Thread.currentThread());
        return null;
    });

    completingAction.run();
    Assert.assertTrue(continuation.isDone());

    // Rethrow any exceptions.
    continuation.join();
}

@NotNull
final CompletableFuture<Void> stopAsync(@NotNull CompletableFuture<Void> operation) {
    Requires.notNull(operation, "operation");

    return Async.runAsync(() -> {
        CompletableFuture<Void> dependentOperation = Futures.completedNull();
        if (dependentService != null) {
            dependentOperation = dependentService.stopAsync(dependentTask);
        }

        return Async.awaitAsync(
            dependentOperation,
            () -> {
                stopRequested.set();
                return Async.usingAsync(
                    joinableCollection.join(),
                    () -> Async.awaitAsync(operation));
            });
    });
}

@Override
public CompletionStage<NodeConnectionReport> closeConnectionAsync(
    SocketAddress connection, CloseType type) {
  Optional<Channel> channel =
      this.clientChannelGroup
          .stream()
          .filter(c -> c.remoteAddress().equals(connection))
          .findFirst();

  if (channel.isPresent()) {
    ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    channelGroup.add(channel.get());
    ClusterConnectionReport clusterReport = new ClusterConnectionReport(getCluster().getId());
    NodeConnectionReport report =
        clusterReport.addNode(this, Collections.singletonList(connection), getAddress());

    return closeChannelGroup(channelGroup, type).thenApply(f -> report);
  } else {
    CompletableFuture<NodeConnectionReport> failedFuture = new CompletableFuture<>();
    failedFuture.completeExceptionally(new IllegalArgumentException("Not found"));
    return failedFuture;
  }
}

@Override
public CompletableFuture<PlcWriteResponse> write(PlcWriteRequest writeRequest) {
  curWriteCnt++;
  if (writeExceptionTriggerCount > 0 && curWriteCnt == writeExceptionTriggerCount) {
    curWriteCnt = 0;
    CompletableFuture<PlcWriteResponse> cf = new CompletableFuture<>();
    cf.completeExceptionally(new PlcIoException(writeExceptionMsg));
    return cf;
  }
   List<WriteResponseItem> responseItems = new LinkedList<>();
    for (WriteRequestItem requestItem : writeRequest.getRequestItems()) {
        setDataValue(requestItem.getAddress(), requestItem.getValues());
        WriteResponseItem responseItem = new WriteResponseItem(requestItem, ResponseCode.OK);
      responseItems.add(responseItem);
    }
  PlcWriteResponse response = new PlcWriteResponse(writeRequest, responseItems);
  return CompletableFuture.completedFuture(response);
}

@Override
public CompletableFuture<MastershipEvent> relinquishRole(NodeId nodeId, DeviceId deviceId) {
    checkArgument(nodeId != null, NODE_ID_NULL);
    checkArgument(deviceId != null, DEVICE_ID_NULL);

    if (nodeId.equals(localNodeId)) {
        return relinquishLocalRole(deviceId);
    }

    log.debug("Forwarding request to relinquish "
            + "role for device {} to {}", deviceId, nodeId);
    return clusterCommunicator.sendAndReceive(
            deviceId,
            ROLE_RELINQUISH_SUBJECT,
            SERIALIZER::encode,
            SERIALIZER::decode,
            nodeId);
}

@Test
void testMultipleActsResultInAsynchronousActionsFailure() {
    int failuresUntilSuccess = 2;
    int numberOfItemsToProcess = 3;

    objectUnderTest.failuresUntilSuccess(failuresUntilSuccess);

    List<CompletableFuture<Object>> completableFutures = new ArrayList<>();
    for (int currentItemIndex = 0; currentItemIndex < numberOfItemsToProcess; currentItemIndex++) {
        completableFutures.add(objectUnderTest.actOnItem(new Object()));
    }

    CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture<?>[0])).join();

    assertThat(objectUnderTest.timesAsynchronousActionCalled.get())
            .isEqualTo((failuresUntilSuccess + 1) * numberOfItemsToProcess);
}

@Test
public void testDoubleDirectClientChooseBadDirectClient(final TestContext testContext) throws Exception {
    final Clients<AsyncClient<? extends Credentials, ? extends CommonProfile>, AsyncAuthorizationGenerator<CommonProfile>> clients = doubleDirectClients();
    when(config.getClients()).thenReturn(clients);
    final String clientNames = NAME;
    when(webContext.getRequestParameter(eq(Clients.DEFAULT_CLIENT_NAME_PARAMETER))).thenReturn(VALUE);
    asyncSecurityLogic = new DefaultAsyncSecurityLogic<>(true, true, config, httpActionAdapter);
    final Async async = testContext.async();

    exception.expect(CompletionException.class);
    exception.expectCause(allOf(IsInstanceOf.instanceOf(TechnicalException.class),
            hasProperty("message", is("Client not allowed: " + VALUE))));
    final CompletableFuture<Object> result = asyncSecurityLogic.perform(webContext, accessGrantedAdapter, clientNames, null, null);
    assertSuccessfulEvaluation(result, ExceptionSoftener.softenConsumer(o -> {
        assertThat(o, is(nullValue()));
        verify(accessGrantedAdapter, times(0)).adapt(webContext);
    }), async);

}

/**
 * This method blocks to create the supplied {@code categoryDrafts} in the CTP project defined by the supplied
 * {@code ctpClient},
 *
 * <p>Note: the method creates the given categories in parallel. So it expects them all to be in the same hierarchy
 * level.
 *
 * @param ctpClient      defines the CTP project to create the categories on.
 * @param categoryDrafts the drafts to build the categories from.
 */
public static List<Category> createCategories(@Nonnull final SphereClient ctpClient,
                                              @Nonnull final List<CategoryDraft> categoryDrafts) {
    final List<CompletableFuture<Category>> futures = new ArrayList<>();
    for (CategoryDraft categoryDraft : categoryDrafts) {
        final CategoryCreateCommand categoryCreateCommand = CategoryCreateCommand.of(categoryDraft);
        final CompletableFuture<Category> categoryCompletableFuture =
            ctpClient.execute(categoryCreateCommand).toCompletableFuture();
        futures.add(categoryCompletableFuture);

    }
    CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]))
                     .toCompletableFuture().join();
    return futures.stream()
                  .map(CompletableFuture::toCompletableFuture)
                  .map(CompletableFuture::join)
                  .collect(Collectors.toList());
}

@Test
public void test_convertEdgeElementToEdgeMessage_write() throws Exception {
  logger.info("[RUN] test_convertEdgeElementToEdgeMessage_write");

  String operation = EdgeCommandType.CMD_WRITE.getValue();
  EdgeElement element = new EdgeElement(operation);
  element.getEdgeAttributeList().add(new EdgeAttribute("value_descriptor",
      EdgeFormatIdentifier.STRING_TYPE.getValue(), providerKey));
  element.getEdgeAttributeList().add(
      new EdgeAttribute("input_argument", EdgeFormatIdentifier.INTEGER_TYPE.getValue(), 100));

  CompletableFuture<String> future = null;
  EdgeMessage msg = OPCUAMessageHandler.getInstance().convertEdgeElementToEdgeMessage(element,
      operation, providerKey, addressable, future);
  assertNotNull(msg);
  logger.info("[PASS] test_convertEdgeElementToEdgeMessage_write");
}

@Override
protected void removeNodeSpi() throws BackingStoreException {
    log.debug("removeNodeSpi()");
    // We need to make this sync. Use doneFuture to sync
    CompletableFuture<Void> doneFuture = new CompletableFuture<>();
    service.findByNameLike(absolutePath())
            .thenAccept(nodes -> {
                List<CompletableFuture<Void>> fs = nodes.stream()
                        .map(service::delete)
                        .collect(Collectors.toList());

                CompletableFuture[] fa = fs.toArray(new CompletableFuture[fs.size()]);
                CompletableFuture.allOf(fa)
                    .thenAccept(v -> doneFuture.complete(null)); // Async
            });
    try {
        doneFuture.get(timeout.toMillis(), TimeUnit.MILLISECONDS);
    }
    catch (Exception e) {
        log.warn("Failed to call removeNodeSpi()", e);
    }
}

@Test
public void should_Cache_failed_fetches() {
    List<Collection<Integer>> loadCalls = new ArrayList<>();
    DataLoader<Integer, Object> errorLoader = idLoaderAllExceptions(new DataLoaderOptions(), loadCalls);

    CompletableFuture<Object> future1 = errorLoader.load(1);
    errorLoader.dispatch();

    await().until(future1::isDone);
    assertThat(future1.isCompletedExceptionally(), is(true));
    assertThat(cause(future1), instanceOf(IllegalStateException.class));

    CompletableFuture<Object> future2 = errorLoader.load(1);
    errorLoader.dispatch();

    await().until(future2::isDone);
    assertThat(future2.isCompletedExceptionally(), is(true));
    assertThat(cause(future2), instanceOf(IllegalStateException.class));

    assertThat(loadCalls, equalTo(singletonList(singletonList(1))));
}

@Test
public void shouldCallTrustWithoutOptions() throws Exception {
  final TrustOutput output = TrustOutput.builder()
      .addTrustedPubkey(TrustedPubkey.builder()
                            .prefix("")
                            .key("pubkey1")
                            .location("")
                            .build())
      .build();
  final Response<ByteString> responsePayload =
      Response.forPayload(ByteString.of(Json.serialize(output)));
  when(client.send(
      Request.forUri("http://localhost:8080/api/v0/rkt/trust?pubkey=http://example.com/pubkey1",
                     DEFAULT_HTTP_METHOD)))
      .thenReturn(CompletableFuture.completedFuture(responsePayload));
  final TrustOutput response =
      rktLauncherRemote.trust("http://example.com/pubkey1").toCompletableFuture().get();
  assertEquals(output, response);
}

@Test
public void shouldResponseWithEmptyAckAndSeparateResponse() throws Exception {
    //empty ack
    transport.when(newCoapPacket(1).token(123).get().uriPath("/path1").build())
            .then(newCoapPacket(1).ack(null).build());

    CompletableFuture<CoapPacket> futResp = client.resource("/path1").token(123).get();

    //separate response
    transport.receive(newCoapPacket(2).token(123).non(Code.C205_CONTENT).payload("dupa").build(), SERVER_ADDRESS);

    assertEquals("dupa", futResp.get().getPayloadString());
}

@Test
public void testCallbackWithSessionRenewal(final TestContext testContext) throws Exception {
    final TestProfile expectedProfile = TestProfile.from(TEST_CREDENTIALS);
    when(webContext.getRequestParameter(Clients.DEFAULT_CLIENT_NAME_PARAMETER)).thenReturn(NAME);
    final String originalSessionId = UUID.randomUUID().toString();
    final String renewedSessionId = UUID.randomUUID().toString();
    when(sessionStore.getOrCreateSessionId(eq(webContext)))
            .thenReturn(CompletableFuture.completedFuture(originalSessionId))
            .thenReturn(CompletableFuture.completedFuture(renewedSessionId));
    final Clients<AsyncClient<? extends Credentials, ? extends TestProfile>, AsyncAuthorizationGenerator<TestProfile>> clients = clientsWithOneIndirectClient();
    when(config.getClients()).thenReturn(clients);
    asyncCallbackLogic = new DefaultAsyncCallbackLogic<>(false, true, config, httpActionAdapter);
    final Async async = testContext.async();
    final CompletableFuture<Object> future = asyncCallbackLogic.perform(webContext, null);
    final CompletableFuture<Map<String, TestProfile>> profilesFuture = future.thenAccept(o -> {
        assertThat(o, is(nullValue()));
        assertThat(status.get(), is(302));
        assertThat(responseHeaders.get(LOCATION_HEADER), is(Pac4jConstants.DEFAULT_URL_VALUE));
        verify(sessionStore, times(1)).renewSession(any(AsyncWebContext.class));
    }).thenCompose((Void v) ->  webContext.getSessionStore().get(webContext, Pac4jConstants.USER_PROFILES));

    assertSuccessfulEvaluation(profilesFuture, profiles -> {
        assertThat(profiles.containsValue(expectedProfile), is(true));
        assertThat(profiles.size(), is(1));
    }, async);

}

/**
 * Synchronously blocks the calling thread for the completion of the specified future. If running on the main
 * thread, any applicable message pump is suppressed while the thread sleeps.
 *
 * <p>Implementations should take care that exceptions from faulted or canceled futures not be thrown back to the
 * caller.</p>
 *
 * @param task The future whose completion is being waited on.
 */
@SuppressWarnings(Suppressions.TRY_SCOPE)
protected void waitSynchronously(CompletableFuture<?> future) {
    if (getContext().isOnMainThread()) {
        // Suppress any reentrancy by causing this synchronously blocking wait to not pump any messages at all.
        try (SpecializedSyncContext syncContext = SpecializedSyncContext.apply(getContext().getNoMessagePumpSynchronizationContext())) {
            waitSynchronouslyCore(future);
        }
    } else {
        waitSynchronouslyCore(future);
    }
}

@Override
public CompletableFuture<TaggedResponse> continueAfterResponse(ImapResponse imapResponse, Throwable throwable) {
  if (throwable != null || !(imapResponse instanceof ContinuationResponse)) {
    throw new UnexpectedAppendResponseException(throwable);
  }

  return imapClient.send(getStringLiteralCommand());
}

public static void main(String[] args) throws Exception{
        final CompletableFuture<Integer> future=new CompletableFuture<>();
        new Thread(new AskThread(future)).start();
        //模拟主线程的计算，然后触发re.get()
        Thread.sleep(100);
        System.out.println("main thread:before complete"+Thread.currentThread().getName()+"  status:"+future);
        future.complete(60);
//        System.out.println("future.get():"+future.get());
        System.out.println("main thread:after complete"+Thread.currentThread().getName()+"  status:"+future);
    }

/**
 * Apply a function that returns a CompletableFuture on each node.
 *
 * @param fun Function to apply
 * @return future result of applying function on each node.
 */
default CompletionStage<Void> forEachNode(Function<BoundNode, CompletionStage<Void>> fun) {
  return CompletableFuture.allOf(
          this.getNodes()
              .stream()
              .map(i -> fun.apply(i).toCompletableFuture())
              .collect(Collectors.toList())
              .toArray(new CompletableFuture[] {}))
      .thenApply(v -> null);
}

@Test
void testProcessFileByIdSuccess() {
    Metadata returnedData = buildFakeMetadata(true, false);
    when(dbService.getMetadataById(anyString())).thenReturn(returnedData);
    when(dbService.write(any(Metadata.class))).thenReturn(CompletableFuture.completedFuture(returnedData));

    String message = objectUnderTest.processFileById(MEEP);

    verify(dbService, times(1)).getMetadataById(MEEP);
    verify(dbService, times(1)).write(returnedData);

    assertThat(message).isEqualTo(CpcFileServiceImpl.FILE_FOUND_PROCESSED);
}

/**
 * Get method command message from EdgeElement format<br>
 * Use {@link org.command.json.format.EdgeJsonFormatter#getStringValueByName(List, String)} to get
 * value from EdgeAttributes<br>
 * Use {@link #getEndpointUrifromAddressable(Addressable)} to get addressable
 * 
 * @param element element object of json format
 * @param providerKey provider key which node
 * @param addr addressable object
 * @param future CompletableFuture object
 * @return command message format is {@link org.edge.protocol.opcua.api.common.EdgeMessage}
 */
private EdgeMessage getMethodMessage(EdgeElement element, String providerKey, Addressable addr,
    CompletableFuture<String> future) throws Exception {
  Object inputValue = EdgeJsonFormatter.getObjectValueByName(element.getEdgeAttributeList(),
      OPCUAMessageKeyIdentifier.INPUT_ARGUMENT.getValue());
  EdgeEndpointInfo epInfo =
      new EdgeEndpointInfo.Builder(getEndpointUrifromAddressable(addr)).setFuture(future).build();
  EdgeNodeInfo ep = new EdgeNodeInfo.Builder().setValueAlias(providerKey).build();
  EdgeMessage msg = new EdgeMessage.Builder(epInfo).setCommand(EdgeCommandType.CMD_METHOD)
      .setRequest(new EdgeRequest.Builder(ep)
          .setMessage(new EdgeVersatility.Builder(inputValue).build()).build())
      .build();
  return msg;
}

public static CompletableFuture<List<DataValue>> read(
        final OpcUaClient client,
        final AttributeId attributeId,
        final NodeId... nodeIds) {

    return client
            .read(
                    0,
                    Both,
                    asList(nodeIds),
                    nCopies(nodeIds.length, attributeId.uid()));
}

@Override
public void balanceRoles() {
    List<ControllerNode> nodes = newArrayList(clusterService.getNodes());
    Map<ControllerNode, Set<DeviceId>> controllerDevices = new HashMap<>();
    int deviceCount = 0;

    // Create buckets reflecting current ownership.
    for (ControllerNode node : nodes) {
        if (clusterService.getState(node.id()).isActive()) {
            Set<DeviceId> devicesOf = new HashSet<>(getDevicesOf(node.id()));
            deviceCount += devicesOf.size();
            controllerDevices.put(node, devicesOf);
            log.info("Node {} has {} devices.", node.id(), devicesOf.size());
        }
    }

    if (useRegionForBalanceRoles && balanceRolesUsingRegions(controllerDevices)) {
        return;
    }

    // Now re-balance the buckets until they are roughly even.
    List<CompletableFuture<Void>> balanceBucketsFutures = balanceControllerNodes(controllerDevices, deviceCount);

    CompletableFuture<Void> balanceRolesFuture = CompletableFuture.allOf(
            balanceBucketsFutures.toArray(new CompletableFuture[balanceBucketsFutures.size()]));

    Futures.getUnchecked(balanceRolesFuture);
}

@Test
public void test_convertEdgeElementToEdgeMessage_sub() throws Exception {
  logger.info("[RUN] test_convertEdgeElementToEdgeMessage_sub");

  String operation = EdgeCommandType.CMD_SUB.getValue();
  EdgeElement element = new EdgeElement(operation);
  element.getEdgeAttributeList().add(
      new EdgeAttribute("sampling_interval", EdgeFormatIdentifier.DOUBLE_TYPE.getValue(), 1000.0));

  CompletableFuture<String> future = null;
  EdgeMessage msg = OPCUAMessageHandler.getInstance().convertEdgeElementToEdgeMessage(element,
      operation, providerKey, addressable, future);
  assertNotNull(msg);
  logger.info("[PASS] test_convertEdgeElementToEdgeMessage_sub");
}

private static CompletableFuture<Double> calculatePriceAsync2(String product) {
    CompletableFuture<Double> promise = CompletableFuture.supplyAsync(() -> {
        delay(3000);
        if(product.length() > 4 ) throw new RuntimeException("Illegal Product " + product);
        double res = random.nextDouble() * product.charAt(0) + product.charAt(1);
        double price = ((int) (res * 100)) / 100.0;
        return price;
    });
    return promise;
}

@Test
public void shouldMakeNonBlockingRequest() throws Exception {
    CoapPacket req = newCoapPacket(LOCAL_5683).mid(1).get().uriPath("/test").build();

    CompletableFuture<CoapPacket> resp = server.makeRequest(req);

    verify(msg).makeRequest(eq(req), any(), any());
    assertFalse(resp.isDone());
}

@Test
public void noClientsSuccessfullyAuthenticate(final TestContext testContext) {
    final AsyncDirectClientAuthenticator<TestProfile, AsyncWebContext> authenticator = new AsyncDirectClientAuthenticator<>(AsyncProfileSave.SINGLE_PROFILE_SAVE,
            new AsyncSaveProfileToSessionDecision(true), new AsyncLoadProfileFromSessionDecision());
    final AsyncWebContext webContext = webContextBuilder.build();
    final AsyncClient<TestCredentials, TestProfile> directClient = getClient(false, false);
    final Async async = testContext.async();

    final CompletableFuture<List<TestProfile>>authResultFuture = authenticator.authenticate(Arrays.asList(directClient), webContext, new AsyncProfileManager<>(webContext));

    assertSuccessfulEvaluation(authResultFuture,
            profiles -> assertThat(profiles, is(new ArrayList<>())),
            async);

}

@Override
public CompletableFuture<ProductSet> getProductSet() {
    CompletableFuture<ProductSet> getFuture = new CompletableFuture<>();
    CompletableFuture<Void> allCompleted = CompletableFuture
        .allOf(loaders.toArray(new CompletableFuture[loaders.size()]));
    allCompleted.thenRun(() -> {
        loaders.forEach(loader -> loader.join().forEach(loaded -> products.addProductLinkage(loaded)));
        getFuture.complete(products);
    }).exceptionally(t -> {
        LOG.error("unexpected error loading products", t);
        return null;
    });
    return getFuture;
}

private Client aClientWithConsecutiveContent(final String firstContent, final String... other) {
    final Client client = mock(Client.class);
    @SuppressWarnings("unchecked")
    final CompletableFuture<Response<ByteString>>[] otherResponses = asList(other)
        .stream()
        .map(c -> completedFuture(contentResponse(c, "")))
        .collect(Collectors.toList()).toArray(new CompletableFuture[0]);

    when(client.send(any())).thenReturn(completedFuture(contentResponse(firstContent, "")),
        otherResponses);
    return client;
}

/**
 * Returns the latest value of {@code watchFile()} result.
 *
 * @param defaultValue the default value which is returned when the value is not available yet
 */
@Nullable
default T latestValue(@Nullable T defaultValue) {
    final CompletableFuture<Latest<T>> initialValueFuture = initialValueFuture();
    if (initialValueFuture.isDone() && !initialValueFuture.isCompletedExceptionally()) {
        return latest().value();
    } else {
        return defaultValue;
    }
}

/**
 * Given an inventory entry {@code draft}, issues a request to the CTP project to create a corresponding Inventory
 * Entry.
 *
 * <p>The {@code statistics} instance is updated accordingly to whether the CTP request was carried
 * out successfully or not. If an exception was thrown on executing the request to CTP, the error handling method
 * is called.
 *
 * @param draft the inventory entry draft to create the inventory entry from.
 * @return a future which contains an empty result after execution of the create.
 */
private CompletionStage<Void> create(@Nonnull final InventoryEntryDraft draft) {
    return syncOptions.applyBeforeCreateCallBack(draft)
            .map(inventoryService::createInventoryEntry)
            .map(creationFuture -> creationFuture
                            .thenAccept(createdInventory -> statistics.incrementCreated())
                            .exceptionally(exception -> {
                                final Reference<Channel> supplyChannel = draft.getSupplyChannel();
                                final String errorMessage = format(CTP_INVENTORY_ENTRY_CREATE_FAILED,
                                        draft.getSku(), supplyChannel != null ? supplyChannel.getId() : null);
                                handleError(errorMessage, exception, 1);
                                return null;
                            }))
            .orElseGet(() -> CompletableFuture.completedFuture(null));
}

CompletableFuture<Void> modify(StorageServer server, FileId fileId, File file, int offset) {
    return executor.execute(
            server.toInetAddress(),
            new FileModifyEncoder(fileId, file, offset),
            Replier.NOPDecoder.INSTANCE
    );
}

@Test
public void canDeleteALoanPolicy()
  throws InterruptedException,
  MalformedURLException,
  TimeoutException,
  ExecutionException,
  UnsupportedEncodingException {

  CompletableFuture<JsonResponse> deleteCompleted = new CompletableFuture<>();

  UUID id = UUID.randomUUID();

  createLoanPolicy(new LoanPolicyRequestBuilder().withId(id).create());

  client.delete(loanPolicyStorageUrl(String.format("/%s", id.toString())),
    StorageTestSuite.TENANT_ID,
    ResponseHandler.json(deleteCompleted));

  JsonResponse createResponse = deleteCompleted.get(5, TimeUnit.SECONDS);

  assertThat(String.format("Failed to delete loan policy: %s", createResponse.getBody()),
    createResponse.getStatusCode(), is(HttpURLConnection.HTTP_NO_CONTENT));

  JsonResponse getResponse = getById(id);

  assertThat(String.format("Found a deleted loan policy: %s", getResponse.getBody()),
    getResponse.getStatusCode(), is(HttpURLConnection.HTTP_NOT_FOUND));
}

public static CompletionStage<Stage> stageOf(final String title, final Pane rootPane) {
    final CompletableFuture<Stage> upcomingStage = new CompletableFuture<>();
    Platform.runLater(() -> {
        final Stage stage = new Stage(StageStyle.DECORATED);
        stage.setTitle(title);
        stage.setScene(new Scene(rootPane));
        upcomingStage.complete(stage);
    });
    return upcomingStage;
}

protected void readContent(ByteBuf in, CompletableFuture<T> promise) {

        if (in.readableBytes() < length) {
            return;
        }

        ByteBuf buf = in.readSlice((int) length);
        T result = decoder.decode(buf);
        promise.complete(result);
        atHead = true;
    }

@Test
public void it_should_invoke_only_first_attempt_in_invocation_thread() throws Exception {
  try (AsyncRetryExecutor executor = createBuilder(createExecutorService()).build()) {
    Thread currentThread = Thread.currentThread();
    AtomicBoolean shouldThrowException = new AtomicBoolean(true);
    CompletableFuture<Thread> future = executor.submit(() -> {
      if (shouldThrowException.compareAndSet(true, false)) {
        throw new Exception();
      }
      return Thread.currentThread();
    });
    assertCompletedNotWith(future, currentThread);
  }
}

public static String syncQuery(String id) {
  CompletableFuture<String> future = doQuery(id, true);
  try {
    return future.get();
  } catch (InterruptedException | ExecutionException e) {
    throw new InvocationException(Status.INTERNAL_SERVER_ERROR.getStatusCode(),
        Status.INTERNAL_SERVER_ERROR.getReasonPhrase(), (Object) "Failed to query from redis.", e);
  }
}

@Test(timeout=1000, expected= IntentionalException.class)
public void testUnexpectedExceptionBehaviour(final TestContext testContext) throws Exception {
    doThrow(new IntentionalException()).when(authenticator).validate(eq(TEST_CREDENTIALS), eq(webContext));
    final Async async = testContext.async();
    final CompletableFuture<Void> authFuture = asyncAuthenticator.validate(TEST_CREDENTIALS, webContext);
    assertSuccessfulEvaluation(authFuture, res   -> {}, async);
}