@Test public void waitForCompletionOnExceptionallyCompletedStage() throws Exception { String testBlobId = "BLOBID"; String testContentType = "application/java-serialized-object"; APIModel.Blob blob = new APIModel.Blob(); blob.blobId = testBlobId; blob.contentType = testContentType; APIModel.CompletionResult completionResult = APIModel.CompletionResult.failure(APIModel.BlobDatum.fromBlob(blob)); when(mockHttpClient.execute(requestForAwaitStageResult())).thenReturn(responseWithAwaitStageResponse(completionResult)); class MyException extends RuntimeException {} MyException myException = new MyException(); when(blobStoreClient.readBlob(eq(testFlowId), eq(testBlobId), any(), eq(testContentType))).thenReturn(myException); thrown.expect(new BaseMatcher<Object>() { @Override public void describeTo(Description description) { description.appendText("an exception of type FlowCompletionException, containing a cause of type MyException"); } @Override public boolean matches(Object o) { if(o instanceof FlowCompletionException) { FlowCompletionException flowCompletionException = (FlowCompletionException) o; if(flowCompletionException.getCause().equals(myException)) { return true; } } return false; } }); completerClient.waitForCompletion(new FlowId(testFlowId), new CompletionId(testStageId), null); }
@Override public Object waitForCompletion(FlowId flowId, CompletionId id, ClassLoader ignored, long timeout, TimeUnit unit) throws TimeoutException { long msTimeout = TimeUnit.MILLISECONDS.convert(timeout, unit); long start = System.currentTimeMillis(); do { long lastStart = System.currentTimeMillis(); long remainingTimeout = Math.max(1, start + msTimeout - lastStart); try (HttpClient.HttpResponse response = httpClient.execute(prepareGet(apiUrlBase + "/flows/" + flowId.getId() + "/stages/" + id.getId() + "/await?timeout_ms=" + remainingTimeout))) { if (response.getStatusCode() == 200) { APIModel.AwaitStageResponse resp = FlowRuntimeGlobals.getObjectMapper().readValue(response.getContentStream(), APIModel.AwaitStageResponse.class); if (resp.result.successful) { return resp.result.toJava(flowId, blobStoreClient, getClass().getClassLoader()); } else { throw new FlowCompletionException((Throwable) resp.result.toJava(flowId, blobStoreClient, getClass().getClassLoader())); } } else if (response.getStatusCode() == 408) { } else { throw asError(response); } try { Thread.sleep(Math.max(0, 500 - (System.currentTimeMillis() - lastStart))); } catch (InterruptedException e) { throw new PlatformCommunicationException("Interrupted", e); } } catch (IOException e) { throw new PlatformCommunicationException("Error fetching result", e); } } while (System.currentTimeMillis() - start < msTimeout); throw new TimeoutException("Stage did not completed before timeout "); }
RemoteFlowApiClient implements CompleterClient { @Override public Object waitForCompletion(FlowId flowId, CompletionId id, ClassLoader ignored, long timeout, TimeUnit unit) throws TimeoutException { long msTimeout = TimeUnit.MILLISECONDS.convert(timeout, unit); long start = System.currentTimeMillis(); do { long lastStart = System.currentTimeMillis(); long remainingTimeout = Math.max(1, start + msTimeout - lastStart); try (HttpClient.HttpResponse response = httpClient.execute(prepareGet(apiUrlBase + "/flows/" + flowId.getId() + "/stages/" + id.getId() + "/await?timeout_ms=" + remainingTimeout))) { if (response.getStatusCode() == 200) { APIModel.AwaitStageResponse resp = FlowRuntimeGlobals.getObjectMapper().readValue(response.getContentStream(), APIModel.AwaitStageResponse.class); if (resp.result.successful) { return resp.result.toJava(flowId, blobStoreClient, getClass().getClassLoader()); } else { throw new FlowCompletionException((Throwable) resp.result.toJava(flowId, blobStoreClient, getClass().getClassLoader())); } } else if (response.getStatusCode() == 408) { } else { throw asError(response); } try { Thread.sleep(Math.max(0, 500 - (System.currentTimeMillis() - lastStart))); } catch (InterruptedException e) { throw new PlatformCommunicationException("Interrupted", e); } } catch (IOException e) { throw new PlatformCommunicationException("Error fetching result", e); } } while (System.currentTimeMillis() - start < msTimeout); throw new TimeoutException("Stage did not completed before timeout "); } }
RemoteFlowApiClient implements CompleterClient { @Override public Object waitForCompletion(FlowId flowId, CompletionId id, ClassLoader ignored, long timeout, TimeUnit unit) throws TimeoutException { long msTimeout = TimeUnit.MILLISECONDS.convert(timeout, unit); long start = System.currentTimeMillis(); do { long lastStart = System.currentTimeMillis(); long remainingTimeout = Math.max(1, start + msTimeout - lastStart); try (HttpClient.HttpResponse response = httpClient.execute(prepareGet(apiUrlBase + "/flows/" + flowId.getId() + "/stages/" + id.getId() + "/await?timeout_ms=" + remainingTimeout))) { if (response.getStatusCode() == 200) { APIModel.AwaitStageResponse resp = FlowRuntimeGlobals.getObjectMapper().readValue(response.getContentStream(), APIModel.AwaitStageResponse.class); if (resp.result.successful) { return resp.result.toJava(flowId, blobStoreClient, getClass().getClassLoader()); } else { throw new FlowCompletionException((Throwable) resp.result.toJava(flowId, blobStoreClient, getClass().getClassLoader())); } } else if (response.getStatusCode() == 408) { } else { throw asError(response); } try { Thread.sleep(Math.max(0, 500 - (System.currentTimeMillis() - lastStart))); } catch (InterruptedException e) { throw new PlatformCommunicationException("Interrupted", e); } } catch (IOException e) { throw new PlatformCommunicationException("Error fetching result", e); } } while (System.currentTimeMillis() - start < msTimeout); throw new TimeoutException("Stage did not completed before timeout "); } RemoteFlowApiClient(String apiUrlBase, BlobStoreClient blobClient, HttpClient httpClient); }
RemoteFlowApiClient implements CompleterClient { @Override public Object waitForCompletion(FlowId flowId, CompletionId id, ClassLoader ignored, long timeout, TimeUnit unit) throws TimeoutException { long msTimeout = TimeUnit.MILLISECONDS.convert(timeout, unit); long start = System.currentTimeMillis(); do { long lastStart = System.currentTimeMillis(); long remainingTimeout = Math.max(1, start + msTimeout - lastStart); try (HttpClient.HttpResponse response = httpClient.execute(prepareGet(apiUrlBase + "/flows/" + flowId.getId() + "/stages/" + id.getId() + "/await?timeout_ms=" + remainingTimeout))) { if (response.getStatusCode() == 200) { APIModel.AwaitStageResponse resp = FlowRuntimeGlobals.getObjectMapper().readValue(response.getContentStream(), APIModel.AwaitStageResponse.class); if (resp.result.successful) { return resp.result.toJava(flowId, blobStoreClient, getClass().getClassLoader()); } else { throw new FlowCompletionException((Throwable) resp.result.toJava(flowId, blobStoreClient, getClass().getClassLoader())); } } else if (response.getStatusCode() == 408) { } else { throw asError(response); } try { Thread.sleep(Math.max(0, 500 - (System.currentTimeMillis() - lastStart))); } catch (InterruptedException e) { throw new PlatformCommunicationException("Interrupted", e); } } catch (IOException e) { throw new PlatformCommunicationException("Error fetching result", e); } } while (System.currentTimeMillis() - start < msTimeout); throw new TimeoutException("Stage did not completed before timeout "); } 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 Object waitForCompletion(FlowId flowId, CompletionId id, ClassLoader ignored, long timeout, TimeUnit unit) throws TimeoutException { long msTimeout = TimeUnit.MILLISECONDS.convert(timeout, unit); long start = System.currentTimeMillis(); do { long lastStart = System.currentTimeMillis(); long remainingTimeout = Math.max(1, start + msTimeout - lastStart); try (HttpClient.HttpResponse response = httpClient.execute(prepareGet(apiUrlBase + "/flows/" + flowId.getId() + "/stages/" + id.getId() + "/await?timeout_ms=" + remainingTimeout))) { if (response.getStatusCode() == 200) { APIModel.AwaitStageResponse resp = FlowRuntimeGlobals.getObjectMapper().readValue(response.getContentStream(), APIModel.AwaitStageResponse.class); if (resp.result.successful) { return resp.result.toJava(flowId, blobStoreClient, getClass().getClassLoader()); } else { throw new FlowCompletionException((Throwable) resp.result.toJava(flowId, blobStoreClient, getClass().getClassLoader())); } } else if (response.getStatusCode() == 408) { } else { throw asError(response); } try { Thread.sleep(Math.max(0, 500 - (System.currentTimeMillis() - lastStart))); } catch (InterruptedException e) { throw new PlatformCommunicationException("Interrupted", e); } } catch (IOException e) { throw new PlatformCommunicationException("Error fetching result", e); } } while (System.currentTimeMillis() - start < msTimeout); throw new TimeoutException("Stage did not completed before timeout "); } 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; }