HttpTask extends WorkflowSystemTask { @Override public void start(Workflow workflow, Task task, WorkflowExecutor executor) { Object request = task.getInputData().get(requestParameter); task.setWorkerId(config.getServerId()); if(request == null) { task.setReasonForIncompletion(MISSING_REQUEST); task.setStatus(Status.FAILED); return; } Input input = objectMapper.convertValue(request, Input.class); if(input.getUri() == null) { String reason = "Missing HTTP URI. See documentation for HttpTask for required input parameters"; task.setReasonForIncompletion(reason); task.setStatus(Status.FAILED); return; } if(input.getMethod() == null) { String reason = "No HTTP method specified"; task.setReasonForIncompletion(reason); task.setStatus(Status.FAILED); return; } try { HttpResponse response = httpCall(input); logger.debug("Response: {}, {}, task:{}", response.statusCode, response.body, task.getTaskId()); if(response.statusCode > 199 && response.statusCode < 300) { if (isAsyncComplete(task)) { task.setStatus(Status.IN_PROGRESS); } else { task.setStatus(Status.COMPLETED); } } else { if(response.body != null) { task.setReasonForIncompletion(response.body.toString()); } else { task.setReasonForIncompletion("No response from the remote service"); } task.setStatus(Status.FAILED); } if(response != null) { task.getOutputData().put("response", response.asMap()); } }catch(Exception e) { logger.error("Failed to invoke http task: {} - uri: {}, vipAddress: {} in workflow: {}", task.getTaskId(), input.getUri(), input.getVipAddress(), task.getWorkflowInstanceId(), e); task.setStatus(Status.FAILED); task.setReasonForIncompletion("Failed to invoke http task due to: " + e.toString()); task.getOutputData().put("response", e.toString()); } } @Inject  HttpTask(RestClientManager restClientManager,
					Configuration config,
					ObjectMapper objectMapper);  HttpTask(String name,
					RestClientManager restClientManager,
					Configuration config,
					ObjectMapper objectMapper); @Override void start(Workflow workflow, Task task, WorkflowExecutor executor); @Override boolean execute(Workflow workflow, Task task, WorkflowExecutor executor); @Override void cancel(Workflow workflow, Task task, WorkflowExecutor executor); @Override boolean isAsync(); @Override int getRetryTimeInSecond(); static final String REQUEST_PARAMETER_NAME; static final String NAME; }
@Test public void testPost() { Task task = new Task(); Input input = new Input(); input.setUri("http: Map<String, Object> body = new HashMap<>(); body.put("input_key1", "value1"); body.put("input_key2", 45.3d); body.put("someKey", null); input.setBody(body); input.setMethod("POST"); task.getInputData().put(HttpTask.REQUEST_PARAMETER_NAME, input); httpTask.start(workflow, task, workflowExecutor); assertEquals(task.getReasonForIncompletion(), Task.Status.COMPLETED, task.getStatus()); Map<String, Object> hr = (Map<String, Object>) task.getOutputData().get("response"); Object response = hr.get("body"); assertEquals(Task.Status.COMPLETED, task.getStatus()); assertTrue("response is: " + response, response instanceof Map); Map<String, Object> map = (Map<String, Object>) response; Set<String> inputKeys = body.keySet(); Set<String> responseKeys = map.keySet(); inputKeys.containsAll(responseKeys); responseKeys.containsAll(inputKeys); }
@Test public void testPostNoContent() { Task task = new Task(); Input input = new Input(); input.setUri("http: Map<String, Object> body = new HashMap<>(); body.put("input_key1", "value1"); body.put("input_key2", 45.3d); input.setBody(body); input.setMethod("POST"); task.getInputData().put(HttpTask.REQUEST_PARAMETER_NAME, input); httpTask.start(workflow, task, workflowExecutor); assertEquals(task.getReasonForIncompletion(), Task.Status.COMPLETED, task.getStatus()); Map<String, Object> hr = (Map<String, Object>) task.getOutputData().get("response"); Object response = hr.get("body"); assertEquals(Task.Status.COMPLETED, task.getStatus()); assertNull("response is: " + response, response); }
@Test public void testFailure() { Task task = new Task(); Input input = new Input(); input.setUri("http: input.setMethod("GET"); task.getInputData().put(HttpTask.REQUEST_PARAMETER_NAME, input); httpTask.start(workflow, task, workflowExecutor); assertEquals("Task output: " + task.getOutputData(), Task.Status.FAILED, task.getStatus()); assertEquals(ERROR_RESPONSE, task.getReasonForIncompletion()); task.setStatus(Status.SCHEDULED); task.getInputData().remove(HttpTask.REQUEST_PARAMETER_NAME); httpTask.start(workflow, task, workflowExecutor); assertEquals(Task.Status.FAILED, task.getStatus()); assertEquals(HttpTask.MISSING_REQUEST, task.getReasonForIncompletion()); }
@Test public void testPostAsyncComplete() { Task task = new Task(); Input input = new Input(); input.setUri("http: Map<String, Object> body = new HashMap<>(); body.put("input_key1", "value1"); body.put("input_key2", 45.3d); input.setBody(body); input.setMethod("POST"); task.getInputData().put(HttpTask.REQUEST_PARAMETER_NAME, input); task.getInputData().put("asyncComplete", true); httpTask.start(workflow, task, workflowExecutor); assertEquals(task.getReasonForIncompletion(), Status.IN_PROGRESS, task.getStatus()); Map<String, Object> hr = (Map<String, Object>) task.getOutputData().get("response"); Object response = hr.get("body"); assertEquals(Status.IN_PROGRESS, task.getStatus()); assertTrue("response is: " + response, response instanceof Map); Map<String, Object> map = (Map<String, Object>) response; Set<String> inputKeys = body.keySet(); Set<String> responseKeys = map.keySet(); inputKeys.containsAll(responseKeys); responseKeys.containsAll(inputKeys); }
@Test public void testTextGET() { Task task = new Task(); Input input = new Input(); input.setUri("http: input.setMethod("GET"); task.getInputData().put(HttpTask.REQUEST_PARAMETER_NAME, input); httpTask.start(workflow, task, workflowExecutor); Map<String, Object> hr = (Map<String, Object>) task.getOutputData().get("response"); Object response = hr.get("body"); assertEquals(Task.Status.COMPLETED, task.getStatus()); assertEquals(TEXT_RESPONSE, response); }
@Test public void testNumberGET() { Task task = new Task(); Input input = new Input(); input.setUri("http: input.setMethod("GET"); task.getInputData().put(HttpTask.REQUEST_PARAMETER_NAME, input); httpTask.start(workflow, task, workflowExecutor); Map<String, Object> hr = (Map<String, Object>) task.getOutputData().get("response"); Object response = hr.get("body"); assertEquals(Task.Status.COMPLETED, task.getStatus()); assertEquals(NUM_RESPONSE, response); assertTrue(response instanceof Number); }
@Test public void testJsonGET() throws JsonProcessingException { Task task = new Task(); Input input = new Input(); input.setUri("http: input.setMethod("GET"); task.getInputData().put(HttpTask.REQUEST_PARAMETER_NAME, input); httpTask.start(workflow, task, workflowExecutor); Map<String, Object> hr = (Map<String, Object>) task.getOutputData().get("response"); Object response = hr.get("body"); assertEquals(Task.Status.COMPLETED, task.getStatus()); assertTrue(response instanceof Map); Map<String, Object> map = (Map<String, Object>) response; assertEquals(JSON_RESPONSE, objectMapper.writeValueAsString(map)); }
@Test public void testHTTPGetConnectionTimeOut() throws Exception{ Task task = new Task(); Input input = new Input(); Instant start = Instant.now(); input.setConnectionTimeOut(110); input.setMethod("GET"); input.setUri("http: task.getInputData().put(HttpTask.REQUEST_PARAMETER_NAME, input); task.setStatus(Status.SCHEDULED); task.setScheduledTime(0); httpTask.start(workflow,task,workflowExecutor); Instant end = Instant.now(); long diff = end.toEpochMilli()-start.toEpochMilli(); Assert.assertEquals(task.getStatus(),Status.FAILED); Assert.assertTrue(diff >= 110l); }
@Test public void testHTTPGETReadTimeOut() throws Exception{ Task task = new Task(); Input input = new Input(); input.setReadTimeOut(-1); input.setMethod("GET"); input.setUri("http: task.getInputData().put(HttpTask.REQUEST_PARAMETER_NAME, input); task.setStatus(Status.SCHEDULED); task.setScheduledTime(0); httpTask.start(workflow,task,workflowExecutor); Assert.assertEquals(task.getStatus(),Status.FAILED); }
@Test public void testOptional() { Task task = new Task(); Input input = new Input(); input.setUri("http: input.setMethod("GET"); task.getInputData().put(HttpTask.REQUEST_PARAMETER_NAME, input); httpTask.start(workflow, task, workflowExecutor); assertEquals("Task output: " + task.getOutputData(), Task.Status.FAILED, task.getStatus()); assertEquals(ERROR_RESPONSE, task.getReasonForIncompletion()); assertTrue(!task.getStatus().isSuccessful()); task.setStatus(Status.SCHEDULED); task.getInputData().remove(HttpTask.REQUEST_PARAMETER_NAME); task.setReferenceTaskName("t1"); httpTask.start(workflow, task, workflowExecutor); assertEquals(Task.Status.FAILED, task.getStatus()); assertEquals(HttpTask.MISSING_REQUEST, task.getReasonForIncompletion()); assertTrue(!task.getStatus().isSuccessful()); WorkflowTask workflowTask = new WorkflowTask(); workflowTask.setOptional(true); workflowTask.setName("HTTP"); workflowTask.setWorkflowTaskType(TaskType.USER_DEFINED); workflowTask.setTaskReferenceName("t1"); WorkflowDef def = new WorkflowDef(); def.getTasks().add(workflowTask); Workflow workflow = new Workflow(); workflow.setWorkflowDefinition(def); workflow.getTasks().add(task); MetadataDAO metadataDAO = mock(MetadataDAO.class); ExternalPayloadStorageUtils externalPayloadStorageUtils = mock(ExternalPayloadStorageUtils.class); ParametersUtils parametersUtils = mock(ParametersUtils.class); Configuration configuration = mock(Configuration.class); Map<String, TaskMapper> taskMappers = new HashMap<>(); new DeciderService(parametersUtils, metadataDAO, externalPayloadStorageUtils, taskMappers, configuration).decide(workflow); System.out.println(workflow.getTasks()); System.out.println(workflow.getStatus()); }
@Test public void testOAuth() { Task task = new Task(); Input input = new Input(); input.setUri("http: input.setMethod("POST"); input.setOauthConsumerKey("someKey"); input.setOauthConsumerSecret("someSecret"); task.getInputData().put(HttpTask.REQUEST_PARAMETER_NAME, input); httpTask.start(workflow, task, workflowExecutor); Map<String, Object> response = (Map<String, Object>) task.getOutputData().get("response"); Map<String, String> body = (Map<String, String>) response.get("body"); assertEquals("someKey", body.get("oauth_consumer_key")); assertTrue("Should have OAuth nonce", body.containsKey("oauth_nonce")); assertTrue("Should have OAuth signature", body.containsKey("oauth_signature")); assertTrue("Should have OAuth signature method", body.containsKey("oauth_signature_method")); assertTrue("Should have OAuth oauth_timestamp", body.containsKey("oauth_timestamp")); assertTrue("Should have OAuth oauth_version", body.containsKey("oauth_version")); assertEquals("Task output: " + task.getOutputData(), Status.COMPLETED, task.getStatus()); }