@Test public void supply() throws Exception { String contentType = "application/java-serialized-object"; String testBlobId = "BLOBID"; BlobResponse blobResponse = makeBlobResponse(lambdaBytes, contentType, testBlobId); when(blobStoreClient.writeBlob(testFlowId, lambdaBytes, contentType)).thenReturn(blobResponse); HttpClient.HttpResponse response = responseWithAddStageResponse(testFlowId, testStageId); when(mockHttpClient.execute(requestContainingAddStageRequest(blobResponse.blobId, Collections.emptyList()))).thenReturn(response); CompletionId completionId = completerClient.supply(new FlowId(testFlowId), serializableLambda, codeLocation); assertNotNull(completionId); assertEquals(completionId.getId(), testStageId); }
@Override public CompletionId supply(FlowId flowId, Serializable supplier, CodeLocation codeLocation) { return addStageWithClosure(APIModel.CompletionOperation.SUPPLY, flowId, supplier, codeLocation, Collections.emptyList()); }
RemoteFlowApiClient implements CompleterClient { @Override public CompletionId supply(FlowId flowId, Serializable supplier, CodeLocation codeLocation) { return addStageWithClosure(APIModel.CompletionOperation.SUPPLY, flowId, supplier, codeLocation, Collections.emptyList()); } }
RemoteFlowApiClient implements CompleterClient { @Override public CompletionId supply(FlowId flowId, Serializable supplier, CodeLocation codeLocation) { return addStageWithClosure(APIModel.CompletionOperation.SUPPLY, flowId, supplier, codeLocation, Collections.emptyList()); } RemoteFlowApiClient(String apiUrlBase, BlobStoreClient blobClient, HttpClient httpClient); }
RemoteFlowApiClient implements CompleterClient { @Override public CompletionId supply(FlowId flowId, Serializable supplier, CodeLocation codeLocation) { return addStageWithClosure(APIModel.CompletionOperation.SUPPLY, flowId, supplier, codeLocation, Collections.emptyList()); } RemoteFlowApiClient(String apiUrlBase, BlobStoreClient blobClient, HttpClient httpClient); @Override FlowId createFlow(String functionId); @Override CompletionId supply(FlowId flowId, Serializable supplier, CodeLocation codeLocation); @Override CompletionId thenApply(FlowId flowId, CompletionId completionId, Serializable fn, CodeLocation codeLocation); @Override CompletionId thenCompose(FlowId flowId, CompletionId completionId, Serializable fn, CodeLocation codeLocation); @Override CompletionId whenComplete(FlowId flowId, CompletionId completionId, Serializable fn, CodeLocation codeLocation); @Override CompletionId thenAccept(FlowId flowId, CompletionId completionId, Serializable fn, CodeLocation codeLocation); @Override CompletionId thenRun(FlowId flowId, CompletionId completionId, Serializable fn, CodeLocation codeLocation); @Override CompletionId acceptEither(FlowId flowId, CompletionId completionId, CompletionId alternate, Serializable fn, CodeLocation codeLocation); @Override CompletionId applyToEither(FlowId flowId, CompletionId completionId, CompletionId alternate, Serializable fn, CodeLocation codeLocation); @Override CompletionId thenAcceptBoth(FlowId flowId, CompletionId completionId, CompletionId alternate, Serializable fn, CodeLocation codeLocation); @Override CompletionId createCompletion(FlowId flowId, CodeLocation codeLocation); @Override CompletionId invokeFunction(FlowId flowId, String functionId, byte[] data, HttpMethod method, Headers headers, CodeLocation codeLocation); @Override CompletionId completedValue(FlowId flowId, boolean success, Object value, CodeLocation codeLocation); @Override CompletionId allOf(FlowId flowId, List<CompletionId> cids, CodeLocation codeLocation); @Override CompletionId handle(FlowId flowId, CompletionId completionId, Serializable fn, CodeLocation codeLocation); @Override CompletionId exceptionally(FlowId flowId, CompletionId completionId, Serializable fn, CodeLocation codeLocation); @Override CompletionId exceptionallyCompose(FlowId flowId, CompletionId completionId, Serializable fn, CodeLocation codeLocation); @Override CompletionId thenCombine(FlowId flowId, CompletionId completionId, Serializable fn, CompletionId alternate, CodeLocation codeLocation); @Override boolean complete(FlowId flowId, CompletionId completionId, Object value, CodeLocation codeLocation); @Override boolean completeExceptionally(FlowId flowId, CompletionId completionId, Throwable value, CodeLocation codeLocation); @Override CompletionId anyOf(FlowId flowId, List<CompletionId> cids, CodeLocation codeLocation); @Override CompletionId delay(FlowId flowId, long l, CodeLocation codeLocation); @Override Object waitForCompletion(FlowId flowId, CompletionId id, ClassLoader ignored, long timeout, TimeUnit unit); @Override Object waitForCompletion(FlowId flowId, CompletionId id, ClassLoader ignored); void commit(FlowId flowId); @Override void addTerminationHook(FlowId flowId, Serializable code, CodeLocation codeLocation); }
RemoteFlowApiClient implements CompleterClient { @Override public CompletionId supply(FlowId flowId, Serializable supplier, CodeLocation codeLocation) { return addStageWithClosure(APIModel.CompletionOperation.SUPPLY, flowId, supplier, codeLocation, Collections.emptyList()); } RemoteFlowApiClient(String apiUrlBase, BlobStoreClient blobClient, HttpClient httpClient); @Override FlowId createFlow(String functionId); @Override CompletionId supply(FlowId flowId, Serializable supplier, CodeLocation codeLocation); @Override CompletionId thenApply(FlowId flowId, CompletionId completionId, Serializable fn, CodeLocation codeLocation); @Override CompletionId thenCompose(FlowId flowId, CompletionId completionId, Serializable fn, CodeLocation codeLocation); @Override CompletionId whenComplete(FlowId flowId, CompletionId completionId, Serializable fn, CodeLocation codeLocation); @Override CompletionId thenAccept(FlowId flowId, CompletionId completionId, Serializable fn, CodeLocation codeLocation); @Override CompletionId thenRun(FlowId flowId, CompletionId completionId, Serializable fn, CodeLocation codeLocation); @Override CompletionId acceptEither(FlowId flowId, CompletionId completionId, CompletionId alternate, Serializable fn, CodeLocation codeLocation); @Override CompletionId applyToEither(FlowId flowId, CompletionId completionId, CompletionId alternate, Serializable fn, CodeLocation codeLocation); @Override CompletionId thenAcceptBoth(FlowId flowId, CompletionId completionId, CompletionId alternate, Serializable fn, CodeLocation codeLocation); @Override CompletionId createCompletion(FlowId flowId, CodeLocation codeLocation); @Override CompletionId invokeFunction(FlowId flowId, String functionId, byte[] data, HttpMethod method, Headers headers, CodeLocation codeLocation); @Override CompletionId completedValue(FlowId flowId, boolean success, Object value, CodeLocation codeLocation); @Override CompletionId allOf(FlowId flowId, List<CompletionId> cids, CodeLocation codeLocation); @Override CompletionId handle(FlowId flowId, CompletionId completionId, Serializable fn, CodeLocation codeLocation); @Override CompletionId exceptionally(FlowId flowId, CompletionId completionId, Serializable fn, CodeLocation codeLocation); @Override CompletionId exceptionallyCompose(FlowId flowId, CompletionId completionId, Serializable fn, CodeLocation codeLocation); @Override CompletionId thenCombine(FlowId flowId, CompletionId completionId, Serializable fn, CompletionId alternate, CodeLocation codeLocation); @Override boolean complete(FlowId flowId, CompletionId completionId, Object value, CodeLocation codeLocation); @Override boolean completeExceptionally(FlowId flowId, CompletionId completionId, Throwable value, CodeLocation codeLocation); @Override CompletionId anyOf(FlowId flowId, List<CompletionId> cids, CodeLocation codeLocation); @Override CompletionId delay(FlowId flowId, long l, CodeLocation codeLocation); @Override Object waitForCompletion(FlowId flowId, CompletionId id, ClassLoader ignored, long timeout, TimeUnit unit); @Override Object waitForCompletion(FlowId flowId, CompletionId id, ClassLoader ignored); void commit(FlowId flowId); @Override void addTerminationHook(FlowId flowId, Serializable code, CodeLocation codeLocation); static final String CONTENT_TYPE_HEADER; static final String CONTENT_TYPE_JAVA_OBJECT; static final String CONTENT_TYPE_OCTET_STREAM; }