@Test public void invokeFunctionWithInvalidFunctionId() throws Exception { String testFunctionId = "INVALIDFUNCTIONID"; byte[] invokeBody = "INPUTDATA".getBytes(); String contentType = "text/plain"; BlobResponse blobResponse = makeBlobResponse(invokeBody, contentType, "BLOBID"); when(blobStoreClient.writeBlob(testFlowId, invokeBody, contentType)).thenReturn(blobResponse); when(mockHttpClient.execute(requestContainingAddInvokeFunctionStageRequest(testFunctionId, APIModel.HTTPMethod.Post))).thenReturn(new HttpClient.HttpResponse(400)); thrown.expect(PlatformCommunicationException.class); thrown.expectMessage("Failed to add stage"); Map headersMap = Collections.singletonMap("Content-type", contentType); Headers headers = Headers.fromMap(headersMap); completerClient.invokeFunction(new FlowId(testFlowId), testFunctionId, invokeBody, HttpMethod.POST, headers, locationFn()); }
@Override public CompletionId invokeFunction(FlowId flowId, String functionId, byte[] data, HttpMethod method, Headers headers, CodeLocation codeLocation) { APIModel.HTTPReq httpReq = new APIModel.HTTPReq(); if (headers != null) { if (data.length > 0) { BlobResponse blobResponse = blobStoreClient.writeBlob(flowId.getId(), data, headers.get(CONTENT_TYPE_HEADER).orElse(CONTENT_TYPE_OCTET_STREAM)); httpReq.body = APIModel.Blob.fromBlobResponse(blobResponse); } httpReq.headers = new ArrayList<>(); headers.asMap().forEach((k, vs) -> vs.forEach(v -> httpReq.headers.add(APIModel.HTTPHeader.create(k, v)))); Map<String, List<String>> headersMap = headers.asMap(); headersMap.forEach((key, values) -> values.forEach(value -> httpReq.headers.add(APIModel.HTTPHeader.create(key, value)))); } httpReq.method = APIModel.HTTPMethod.fromFlow(method); APIModel.AddInvokeFunctionStageRequest addInvokeFunctionStageRequest = new APIModel.AddInvokeFunctionStageRequest(); addInvokeFunctionStageRequest.arg = httpReq; addInvokeFunctionStageRequest.codeLocation = codeLocation.getLocation(); addInvokeFunctionStageRequest.callerId = FlowRuntimeGlobals.getCurrentCompletionId().map(CompletionId::getId).orElse(null); addInvokeFunctionStageRequest.functionId = functionId; try { return executeAddInvokeFunctionStageRequest(flowId, addInvokeFunctionStageRequest); } catch (IOException e) { throw new PlatformCommunicationException("Failed to create invokeFunction stage", e); } }
RemoteFlowApiClient implements CompleterClient { @Override public CompletionId invokeFunction(FlowId flowId, String functionId, byte[] data, HttpMethod method, Headers headers, CodeLocation codeLocation) { APIModel.HTTPReq httpReq = new APIModel.HTTPReq(); if (headers != null) { if (data.length > 0) { BlobResponse blobResponse = blobStoreClient.writeBlob(flowId.getId(), data, headers.get(CONTENT_TYPE_HEADER).orElse(CONTENT_TYPE_OCTET_STREAM)); httpReq.body = APIModel.Blob.fromBlobResponse(blobResponse); } httpReq.headers = new ArrayList<>(); headers.asMap().forEach((k, vs) -> vs.forEach(v -> httpReq.headers.add(APIModel.HTTPHeader.create(k, v)))); Map<String, List<String>> headersMap = headers.asMap(); headersMap.forEach((key, values) -> values.forEach(value -> httpReq.headers.add(APIModel.HTTPHeader.create(key, value)))); } httpReq.method = APIModel.HTTPMethod.fromFlow(method); APIModel.AddInvokeFunctionStageRequest addInvokeFunctionStageRequest = new APIModel.AddInvokeFunctionStageRequest(); addInvokeFunctionStageRequest.arg = httpReq; addInvokeFunctionStageRequest.codeLocation = codeLocation.getLocation(); addInvokeFunctionStageRequest.callerId = FlowRuntimeGlobals.getCurrentCompletionId().map(CompletionId::getId).orElse(null); addInvokeFunctionStageRequest.functionId = functionId; try { return executeAddInvokeFunctionStageRequest(flowId, addInvokeFunctionStageRequest); } catch (IOException e) { throw new PlatformCommunicationException("Failed to create invokeFunction stage", e); } } }
RemoteFlowApiClient implements CompleterClient { @Override public CompletionId invokeFunction(FlowId flowId, String functionId, byte[] data, HttpMethod method, Headers headers, CodeLocation codeLocation) { APIModel.HTTPReq httpReq = new APIModel.HTTPReq(); if (headers != null) { if (data.length > 0) { BlobResponse blobResponse = blobStoreClient.writeBlob(flowId.getId(), data, headers.get(CONTENT_TYPE_HEADER).orElse(CONTENT_TYPE_OCTET_STREAM)); httpReq.body = APIModel.Blob.fromBlobResponse(blobResponse); } httpReq.headers = new ArrayList<>(); headers.asMap().forEach((k, vs) -> vs.forEach(v -> httpReq.headers.add(APIModel.HTTPHeader.create(k, v)))); Map<String, List<String>> headersMap = headers.asMap(); headersMap.forEach((key, values) -> values.forEach(value -> httpReq.headers.add(APIModel.HTTPHeader.create(key, value)))); } httpReq.method = APIModel.HTTPMethod.fromFlow(method); APIModel.AddInvokeFunctionStageRequest addInvokeFunctionStageRequest = new APIModel.AddInvokeFunctionStageRequest(); addInvokeFunctionStageRequest.arg = httpReq; addInvokeFunctionStageRequest.codeLocation = codeLocation.getLocation(); addInvokeFunctionStageRequest.callerId = FlowRuntimeGlobals.getCurrentCompletionId().map(CompletionId::getId).orElse(null); addInvokeFunctionStageRequest.functionId = functionId; try { return executeAddInvokeFunctionStageRequest(flowId, addInvokeFunctionStageRequest); } catch (IOException e) { throw new PlatformCommunicationException("Failed to create invokeFunction stage", e); } } RemoteFlowApiClient(String apiUrlBase, BlobStoreClient blobClient, HttpClient httpClient); }
RemoteFlowApiClient implements CompleterClient { @Override public CompletionId invokeFunction(FlowId flowId, String functionId, byte[] data, HttpMethod method, Headers headers, CodeLocation codeLocation) { APIModel.HTTPReq httpReq = new APIModel.HTTPReq(); if (headers != null) { if (data.length > 0) { BlobResponse blobResponse = blobStoreClient.writeBlob(flowId.getId(), data, headers.get(CONTENT_TYPE_HEADER).orElse(CONTENT_TYPE_OCTET_STREAM)); httpReq.body = APIModel.Blob.fromBlobResponse(blobResponse); } httpReq.headers = new ArrayList<>(); headers.asMap().forEach((k, vs) -> vs.forEach(v -> httpReq.headers.add(APIModel.HTTPHeader.create(k, v)))); Map<String, List<String>> headersMap = headers.asMap(); headersMap.forEach((key, values) -> values.forEach(value -> httpReq.headers.add(APIModel.HTTPHeader.create(key, value)))); } httpReq.method = APIModel.HTTPMethod.fromFlow(method); APIModel.AddInvokeFunctionStageRequest addInvokeFunctionStageRequest = new APIModel.AddInvokeFunctionStageRequest(); addInvokeFunctionStageRequest.arg = httpReq; addInvokeFunctionStageRequest.codeLocation = codeLocation.getLocation(); addInvokeFunctionStageRequest.callerId = FlowRuntimeGlobals.getCurrentCompletionId().map(CompletionId::getId).orElse(null); addInvokeFunctionStageRequest.functionId = functionId; try { return executeAddInvokeFunctionStageRequest(flowId, addInvokeFunctionStageRequest); } catch (IOException e) { throw new PlatformCommunicationException("Failed to create invokeFunction stage", e); } } 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 invokeFunction(FlowId flowId, String functionId, byte[] data, HttpMethod method, Headers headers, CodeLocation codeLocation) { APIModel.HTTPReq httpReq = new APIModel.HTTPReq(); if (headers != null) { if (data.length > 0) { BlobResponse blobResponse = blobStoreClient.writeBlob(flowId.getId(), data, headers.get(CONTENT_TYPE_HEADER).orElse(CONTENT_TYPE_OCTET_STREAM)); httpReq.body = APIModel.Blob.fromBlobResponse(blobResponse); } httpReq.headers = new ArrayList<>(); headers.asMap().forEach((k, vs) -> vs.forEach(v -> httpReq.headers.add(APIModel.HTTPHeader.create(k, v)))); Map<String, List<String>> headersMap = headers.asMap(); headersMap.forEach((key, values) -> values.forEach(value -> httpReq.headers.add(APIModel.HTTPHeader.create(key, value)))); } httpReq.method = APIModel.HTTPMethod.fromFlow(method); APIModel.AddInvokeFunctionStageRequest addInvokeFunctionStageRequest = new APIModel.AddInvokeFunctionStageRequest(); addInvokeFunctionStageRequest.arg = httpReq; addInvokeFunctionStageRequest.codeLocation = codeLocation.getLocation(); addInvokeFunctionStageRequest.callerId = FlowRuntimeGlobals.getCurrentCompletionId().map(CompletionId::getId).orElse(null); addInvokeFunctionStageRequest.functionId = functionId; try { return executeAddInvokeFunctionStageRequest(flowId, addInvokeFunctionStageRequest); } catch (IOException e) { throw new PlatformCommunicationException("Failed to create invokeFunction stage", e); } } 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; }