com.netflix.conductor.common.metadata.tasks.Task.getInputData()方法的使用及代码示例

x33g5p2x  于2022-01-30 转载在 其他  
字(15.5k)|赞(0)|评价(0)|浏览(141)

本文整理了Java中com.netflix.conductor.common.metadata.tasks.Task.getInputData()方法的一些代码示例，展示了Task.getInputData()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。Task.getInputData()方法的具体详情如下：
包路径：com.netflix.conductor.common.metadata.tasks.Task
类名称：Task
方法名：getInputData

Task.getInputData介绍

暂无

代码示例

代码示例来源：origin: Netflix/conductor

@Override
public void cancel(Workflow workflow, Task task, WorkflowExecutor provider) {
  String workflowId = (String) task.getOutputData().get(SUB_WORKFLOW_ID);
  if(workflowId == null) {
    workflowId = (String) task.getInputData().get(SUB_WORKFLOW_ID);	//Backward compatibility
  }
  
  if(StringUtils.isEmpty(workflowId)) {
    return;
  }
  Workflow subWorkflow = provider.getWorkflow(workflowId, false);
  subWorkflow.setStatus(WorkflowStatus.TERMINATED);
  provider.terminateWorkflow(subWorkflow, "Parent workflow has been terminated with status " + workflow.getStatus(), null);
}

代码示例来源：origin: Netflix/conductor

@SuppressWarnings("unchecked")
@Override
public void start(Workflow workflow, Task task, WorkflowExecutor provider) {
  Map<String, Object> input = task.getInputData();
  String name = input.get("subWorkflowName").toString();
  int version = (int) input.get("subWorkflowVersion");
  Map<String, Object> wfInput = (Map<String, Object>) input.get("workflowInput");
  if (wfInput == null || wfInput.isEmpty()) {
    wfInput = input;
  }
  String correlationId = workflow.getCorrelationId();
  
  try {
    String subWorkflowId = provider.startWorkflow(name, version, wfInput, null, correlationId, workflow.getWorkflowId(), task.getTaskId(), null, workflow.getTaskToDomain());
    task.getOutputData().put(SUB_WORKFLOW_ID, subWorkflowId);
    task.getInputData().put(SUB_WORKFLOW_ID, subWorkflowId);
    task.setStatus(Status.IN_PROGRESS);
  } catch (Exception e) {
    task.setStatus(Status.FAILED);
    task.setReasonForIncompletion(e.getMessage());
    logger.error(e.getMessage(), e);
  }
}

代码示例来源：origin: Netflix/conductor

this.reasonForIncompletion = task.getReasonForIncompletion();
this.queueWaitTime = task.getQueueWaitTime();
if (task.getInputData() != null) {
  this.input = task.getInputData().toString();

代码示例来源：origin: Netflix/conductor

/**
 * This method creates a FORK task and adds the list of dynamic fork tasks keyed by "forkedTaskDefs" and
 * their names keyed by "forkedTasks" into {@link Task#getInputData()}
 *
 * @param taskToSchedule    A {@link WorkflowTask} representing {@link TaskType#FORK_JOIN_DYNAMIC}
 * @param workflowInstance: A instance of the {@link Workflow} which represents the workflow being executed.
 * @param taskId:           The string representation of {@link java.util.UUID} which will be set as the taskId.
 * @param dynForkTasks:     The list of dynamic forked tasks, the reference names of these tasks will be added to the forkDynamicTask
 * @return A new instance of {@link Task} representing a {@link SystemTaskType#FORK}
 */
@VisibleForTesting
Task createDynamicForkTask(WorkflowTask taskToSchedule, Workflow workflowInstance, String taskId, List<WorkflowTask> dynForkTasks) {
  Task forkDynamicTask = new Task();
  forkDynamicTask.setTaskType(SystemTaskType.FORK.name());
  forkDynamicTask.setTaskDefName(SystemTaskType.FORK.name());
  forkDynamicTask.setReferenceTaskName(taskToSchedule.getTaskReferenceName());
  forkDynamicTask.setWorkflowInstanceId(workflowInstance.getWorkflowId());
  forkDynamicTask.setCorrelationId(workflowInstance.getCorrelationId());
  forkDynamicTask.setScheduledTime(System.currentTimeMillis());
  forkDynamicTask.setEndTime(System.currentTimeMillis());
  List<String> forkedTaskNames = dynForkTasks.stream()
      .map(WorkflowTask::getTaskReferenceName)
      .collect(Collectors.toList());
  forkDynamicTask.getInputData().put("forkedTasks", forkedTaskNames);
  forkDynamicTask.getInputData().put("forkedTaskDefs", dynForkTasks);    //TODO: Remove this parameter in the later releases
  forkDynamicTask.setTaskId(taskId);
  forkDynamicTask.setStatus(Task.Status.COMPLETED);
  forkDynamicTask.setWorkflowTask(taskToSchedule);
  return forkDynamicTask;
}

代码示例来源：origin: Netflix/conductor

@VisibleForTesting
List<Task> getNextTask(Workflow workflow, Task task) {
  final WorkflowDef workflowDef = workflow.getWorkflowDefinition();
  // Get the following task after the last completed task
  if (SystemTaskType.is(task.getTaskType()) && SystemTaskType.DECISION.name().equals(task.getTaskType())) {
    if (task.getInputData().get("hasChildren") != null) {
      return Collections.emptyList();
    }
  }
  String taskReferenceName = task.getReferenceTaskName();
  WorkflowTask taskToSchedule = workflowDef.getNextTask(taskReferenceName);
  while (isTaskSkipped(taskToSchedule, workflow)) {
    taskToSchedule = workflowDef.getNextTask(taskToSchedule.getTaskReferenceName());
  }
  if (taskToSchedule != null) {
    return getTasksToBeScheduled(workflow, taskToSchedule, 0);
  }
  return Collections.emptyList();
}

代码示例来源：origin: Netflix/conductor

@Override
public void start(Workflow workflow, Task task, WorkflowExecutor provider) {
  Map<String, Object> payload = new HashMap<>(task.getInputData());
  payload.put("workflowInstanceId", workflow.getWorkflowId());
  payload.put("workflowType", workflow.getWorkflowName());
  payload.put("workflowVersion", workflow.getWorkflowVersion());
  payload.put("correlationId", workflow.getCorrelationId());
  String payloadJson;
  try {
    payloadJson = objectMapper.writeValueAsString(payload);
  } catch (JsonProcessingException e) {
    String msg = String.format("Error serializing JSON payload for task: %s, workflow: %s", task.getTaskId(), workflow.getWorkflowId());
    throw new ApplicationException(INTERNAL_ERROR, msg);
  }
  Message message = new Message(task.getTaskId(), payloadJson, task.getTaskId());
  ObservableQueue queue = getQueue(workflow, task);
  if(queue != null) {
    queue.publish(Collections.singletonList(message));
    task.getOutputData().putAll(payload);
    task.setStatus(Status.COMPLETED);
  } else {
    task.setReasonForIncompletion("No queue found to publish.");
    task.setStatus(Status.FAILED);
  }
}

代码示例来源：origin: Netflix/conductor

@Override
public boolean execute(Workflow workflow, Task task, WorkflowExecutor provider) {
  String workflowId = (String) task.getOutputData().get(SUB_WORKFLOW_ID);
  if (workflowId == null) {
    workflowId = (String) task.getInputData().get(SUB_WORKFLOW_ID);	//Backward compatibility
  }
  
  if(StringUtils.isEmpty(workflowId)) {
    return false;
  }
  
  Workflow subWorkflow = provider.getWorkflow(workflowId, false);
  WorkflowStatus subWorkflowStatus = subWorkflow.getStatus();
  if(!subWorkflowStatus.isTerminal()){
    return false;
  }
  task.getOutputData().putAll(subWorkflow.getOutput());
  if (subWorkflowStatus.isSuccessful()) {
    task.setStatus(Status.COMPLETED);
  } else {
    task.setReasonForIncompletion(subWorkflow.getReasonForIncompletion());
    task.setStatus(Status.FAILED);
  }
  return true;
}

代码示例来源：origin: Netflix/conductor

@SuppressWarnings("unchecked")
@Test
public void testUploadTaskPayload() throws IOException {
  AtomicInteger uploadCount = new AtomicInteger(0);
  InputStream stream = ExternalPayloadStorageUtilsTest.class.getResourceAsStream("/payload.json");
  Map<String, Object> payload = objectMapper.readValue(stream, Map.class);
  when(externalPayloadStorage.getLocation(ExternalPayloadStorage.Operation.WRITE, ExternalPayloadStorage.PayloadType.TASK_INPUT, "")).thenReturn(location);
  doAnswer(invocation -> {
    uploadCount.incrementAndGet();
    return null;
  }).when(externalPayloadStorage).upload(anyString(), any(), anyLong());
  Task task = new Task();
  task.setInputData(payload);
  externalPayloadStorageUtils.verifyAndUpload(task, ExternalPayloadStorage.PayloadType.TASK_INPUT);
  assertNull(task.getInputData());
  assertEquals(1, uploadCount.get());
  assertNotNull(task.getExternalInputPayloadStoragePath());
}

代码示例来源：origin: Netflix/conductor

eventTask.setScheduledTime(System.currentTimeMillis());
eventTask.setInputData(eventTaskInput);
eventTask.getInputData().put("sink", sink);
eventTask.setTaskId(taskId);
eventTask.setStatus(Task.Status.SCHEDULED);

代码示例来源：origin: Netflix/conductor

@Test
public void testPost() {
  Task task = new Task();
  Input input = new Input();
  input.setUri("http://localhost:7009/post");
  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());
  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);
}

代码示例来源：origin: Netflix/conductor

@Test
public void testTextGET() {
  Task task = new Task();
  Input input = new Input();
  input.setUri("http://localhost:7009/text");
  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);
}

代码示例来源：origin: Netflix/conductor

@Test
public void testJsonGET() throws JsonProcessingException {
  Task task = new Task();
  Input input = new Input();
  input.setUri("http://localhost:7009/json");
  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));
}

代码示例来源：origin: Netflix/conductor

@Test
public void testFailTaskWithInputPayload() {
  Task task = new Task();
  task.setInputData(new HashMap<>());
  expectedException.expect(TerminateWorkflowException.class);
  externalPayloadStorageUtils.failTask(task, ExternalPayloadStorage.PayloadType.TASK_INPUT, "error");
  assertNotNull(task);
  assertNull(task.getInputData());
}

代码示例来源：origin: Netflix/conductor

@Test
public void testNumberGET() {
  Task task = new Task();
  Input input = new Input();
  input.setUri("http://localhost:7009/numeric");
  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);
}

代码示例来源：origin: Netflix/conductor

@Test
public void testPostNoContent() {
  Task task = new Task();
  Input input = new Input();
  input.setUri("http://localhost:7009/post2");
  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);
}

代码示例来源：origin: Netflix/conductor

@Test
public void testOAuth() {
  Task task = new Task();
  Input input = new Input();
  input.setUri("http://localhost:7009/oauth");
  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());
}

代码示例来源：origin: Netflix/conductor

@Test
public void testExecute() {
  Task task = new Task();
  Input input = new Input();
  input.setUri("http://localhost:7009/json");
  input.setMethod("GET");
  task.getInputData().put(HttpTask.REQUEST_PARAMETER_NAME, input);
  task.setStatus(Status.SCHEDULED);
  task.setScheduledTime(0);
  boolean executed = httpTask.execute(workflow, task, workflowExecutor);
  assertFalse(executed);
}

代码示例来源：origin: Netflix/conductor

@Test
public void testFailure() {
  Task task = new Task();
  Input input = new Input();
  input.setUri("http://localhost:7009/failure");
  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());
}

代码示例来源：origin: Netflix/conductor

private void verify(String inputParam1, String wfid, String task1Op, boolean fail) {
  Task task = workflowExecutionService.poll("junit_task_2", "task2.junit.worker");
  assertNotNull(task);
  assertTrue(workflowExecutionService.ackTaskReceived(task.getTaskId()));
  String task2Input = (String) task.getInputData().get("tp2");
  assertNotNull(task2Input);
  assertEquals(task1Op, task2Input);
  task2Input = (String) task.getInputData().get("tp1");
  assertNotNull(task2Input);
  assertEquals(inputParam1, task2Input);
  if (fail) {
    task.setStatus(FAILED);
    task.setReasonForIncompletion("failure...0");
  } else {
    task.setStatus(COMPLETED);
  }
  workflowExecutionService.updateTask(task);
  Workflow es = workflowExecutionService.getExecutionStatus(wfid, false);
  assertNotNull(es);
  if (fail) {
    assertEquals(RUNNING, es.getStatus());
  } else {
    assertEquals(WorkflowStatus.COMPLETED, es.getStatus());
  }
}

代码示例来源：origin: Netflix/conductor

@Override
  public int hashCode() {
    return Objects.hash(getTaskType(), getStatus(), getInputData(), getReferenceTaskName(), getRetryCount(), getSeq(), getCorrelationId(), getPollCount(), getTaskDefName(), getScheduledTime(), getStartTime(), getEndTime(), getUpdateTime(), getStartDelayInSeconds(), getRetriedTaskId(), isRetried(), isExecuted(), isCallbackFromWorker(), getResponseTimeoutSeconds(), getWorkflowInstanceId(), getWorkflowType(), getTaskId(), getReasonForIncompletion(), getCallbackAfterSeconds(), getWorkerId(), getOutputData(), getWorkflowTask(), getDomain(), getInputMessage(), getOutputMessage(), getRateLimitPerFrequency(), getRateLimitFrequencyInSeconds(), getExternalInputPayloadStoragePath(), getExternalOutputPayloadStoragePath());
  }
}

内容来源于网络，如有侵权，请联系作者删除！

task

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多

    Task类方法

    • 蜀ICP备2022004421号-2 NULL123 https://www.null123.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com