add necessary api for console end-point

dataadapter-console/src/main/java/com/tbyf/dataadapter/console/controlle/WorkerController.java

@@ -92,4 +92,44 @@ public class WorkerController {
         }
         return Result.ok(false);
     }
+
+    @PostMapping("/startTask")
+    public Result<?> startTask(String group, String worker, String taskId) {
+        WorkerNode workerNode = workerRegistry.getWorker(group, worker);
+        if (workerNode != null) {
+            WorkerBizClient client = workerNode.getClient();
+            return client.startTask(taskId);
+        }
+        return Result.ok(false);
+    }
+
+    @PostMapping("/stopTask")
+    public Result<?> stopTask(String group, String worker, String taskId) {
+        WorkerNode workerNode = workerRegistry.getWorker(group, worker);
+        if (workerNode != null) {
+            WorkerBizClient client = workerNode.getClient();
+            return client.stopTask(taskId);
+        }
+        return Result.ok(false);
+    }
+
+
+    @PostMapping("/pauseTask")
+    public Result<?> pauseTask(String group, String worker, String taskId) {
+        WorkerNode workerNode = workerRegistry.getWorker(group, worker);
+        if (workerNode != null) {
+            WorkerBizClient client = workerNode.getClient();
+            return client.pauseTask(taskId);
+        }
+        return Result.ok(false);
+    }
+    @PostMapping("/resumeTask")
+    public Result<?> resumeTask(String group, String worker, String taskId) {
+        WorkerNode workerNode = workerRegistry.getWorker(group, worker);
+        if (workerNode != null) {
+            WorkerBizClient client = workerNode.getClient();
+            return client.resumeTask(taskId);
+        }
+        return Result.ok(false);
+    }
 }

dataadapter-core/src/main/java/com/tbyf/dataadapter/WorkerApiRequestHandler.java

@@ -38,6 +38,12 @@ public class WorkerApiRequestHandler implements RequestHandler {
             case "/stopTask":
                 result = workerBizApi.stopTask(request.getContent());
                 break;
+            case "/pauseTask":
+                result = workerBizApi.pauseTask(request.getContent());
+                break;
+            case "/resumeTask":
+                result = workerBizApi.resumeTask(request.getContent());
+                break;
             case "/getProcessors":
                 result = workerBizApi.getProcessors();
                 break;

dataadapter-core/src/main/java/com/tbyf/dataadapter/api/WorkerBizApi.java

@@ -20,6 +20,10 @@ public interface WorkerBizApi {
 
     Result<Boolean> stopTask(String taskId);
 
+    Result<Boolean> pauseTask(String taskId);
+
+    Result<Boolean> resumeTask(String taskId);
+
     Result<List<Task>> getTasks();
 
     Result<List<ProcessorVo>> getProcessors();

dataadapter-core/src/main/java/com/tbyf/dataadapter/api/impl/WorkerBizClient.java

@@ -68,6 +68,28 @@ public class WorkerBizClient implements WorkerBizApi {
     }
 
     @Override
+    public Result<Boolean> pauseTask(String taskId) {
+        try {
+            String result = HttpClient.post(workerUrl + "/pauseTask", taskId);
+            return JsonUtils.fromJson(result, new TypeToken<Result<Boolean>>() {
+            });
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Override
+    public Result<Boolean> resumeTask(String taskId) {
+        try {
+            String result = HttpClient.post(workerUrl + "/resumeTask", taskId);
+            return JsonUtils.fromJson(result, new TypeToken<Result<Boolean>>() {
+            });
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Override
     public Result<List<Task>> getTasks() {
         try {
             String result = HttpClient.get(workerUrl + "/getTasks");

dataadapter-core/src/main/java/com/tbyf/dataadapter/api/impl/WorkerBizImpl.java

@@ -91,6 +91,26 @@ public class WorkerBizImpl implements WorkerBizApi {
     }
 
     @Override
+    public Result<Boolean> pauseTask(String taskId) {
+        try {
+            taskManager.pauseTask(taskId);
+            return Result.ok(true);
+        } catch (Exception e) {
+            return Result.ok(false);
+        }
+    }
+
+    @Override
+    public Result<Boolean> resumeTask(String taskId) {
+        try {
+            taskManager.resumeTask(taskId);
+            return Result.ok(true);
+        } catch (Exception e) {
+            return Result.ok(false);
+        }
+    }
+
+    @Override
     public Result<List<Task>> getTasks() {
         return Result.ok(taskManager.getTasks());
     }

dataadapter-core/src/main/java/com/tbyf/dataadapter/task/TaskManager.java

@@ -21,6 +21,7 @@ public class TaskManager {
     private final Map<String, TaskState> taskStateMap = new HashMap<>();
     private final Map<String, List<TaskThread>> taskExecutors = new HashMap<>();
     private final Map<String, Task> tasks = new ConcurrentHashMap<>();
+    private volatile List<String> taskIds = new ArrayList<>();
 
 
     public TaskManager() {
@@ -52,6 +53,10 @@ public class TaskManager {
         }
         task.setParallelism(parallelism);
         tasks.put(taskId, task);
+        List<String> old = this.taskIds;
+        List<String> updatedTaskIds = new ArrayList<>(old);
+        updatedTaskIds.add(taskId);
+        taskIds = updatedTaskIds;
     }
 
     private ProcessorPipeline buildPipeline(PipelineDef def) {
@@ -89,14 +94,14 @@ public class TaskManager {
                 ts.add(t);
             }
             taskExecutors.put(taskId, ts);
-            taskStateMap.put(taskId, TaskState.STARTED);
-            updateState(taskId, TaskState.STARTED);
+            taskStateMap.put(taskId, TaskState.RUNNING);
+            updateState(taskId, TaskState.RUNNING);
         }
     }
 
     public synchronized void pauseTask(String taskId) {
         TaskState taskState = taskStateMap.get(taskId);
-        if (taskState == TaskState.STARTED) {
+        if (taskState == TaskState.RUNNING) {
             taskStateMap.put(taskId, TaskState.PAUSED);
             taskExecutors.get(taskId).forEach(TaskThread::pause);
             updateState(taskId, TaskState.PAUSED);
@@ -106,15 +111,15 @@ public class TaskManager {
     public synchronized void resumeTask(String taskId) {
         TaskState taskState = taskStateMap.get(taskId);
         if (taskState == TaskState.PAUSED) {
-            taskStateMap.put(taskId, TaskState.STARTED);
+            taskStateMap.put(taskId, TaskState.RUNNING);
             taskExecutors.get(taskId).forEach(TaskThread::proceed);
-            updateState(taskId, TaskState.STARTED);
+            updateState(taskId, TaskState.RUNNING);
         }
     }
 
     public synchronized void terminateTask(String taskId) {
         TaskState taskState = taskStateMap.get(taskId);
-        if (taskState == TaskState.STARTED || taskState == TaskState.PAUSED) {
+        if (taskState == TaskState.RUNNING || taskState == TaskState.PAUSED) {
             taskStateMap.put(taskId, TaskState.TERMINATED);
             taskExecutors.get(taskId).forEach(TaskThread::terminate);
             updateState(taskId, TaskState.TERMINATED);
@@ -126,7 +131,15 @@ public class TaskManager {
     }
 
     public List<Task> getTasks() {
-        return new ArrayList<>(tasks.values());
+        ArrayList<Task> result = new ArrayList<>();
+        List<String> taskIds = this.taskIds;
+        for (String taskId : taskIds) {
+            Task task = tasks.get(taskId);
+            if (task != null) {
+                result.add(tasks.get(taskId));
+            }
+        }
+        return result;
     }
 
 

dataadapter-core/src/main/java/com/tbyf/dataadapter/task/TaskState.java

@@ -3,7 +3,7 @@ package com.tbyf.dataadapter.task;
 public enum TaskState {
 
     NEW,
-    STARTED,
+    RUNNING,
     PAUSED,
     TERMINATED
 }