refactor: rename OhMyClient to PowerJobClient

powerjob-client/src/main/java/tech/powerjob/client/PowerJobClient.java

@@ -27,13 +27,13 @@ import java.util.Objects;
 import static tech.powerjob.client.TypeStore.*;
 
 /**
- * OhMyClient, the client for OpenAPI.
+ * PowerJobClient, the client for OpenAPI.
  *
  * @author tjq
  * @since 2020/4/15
  */
 @Slf4j
-public class OhMyClient {
+public class PowerJobClient {
 
     private Long appId;
     private String currentAddress;
@@ -42,25 +42,25 @@ public class OhMyClient {
     private static final String URL_PATTERN = "http://%s%s%s";
 
     /**
-     * Init OhMyClient with domain, appName and password.
+     * Init PowerJobClient with domain, appName and password.
      *
      * @param domain   like powerjob-server.apple-inc.com (Intranet Domain)
      * @param appName  name of the application
      * @param password password of the application
      */
-    public OhMyClient(String domain, String appName, String password) {
+    public PowerJobClient(String domain, String appName, String password) {
         this(Lists.newArrayList(domain), appName, password);
     }
 
 
     /**
-     * Init OhMyClient with server address, appName and password.
+     * Init PowerJobClient with server address, appName and password.
      *
      * @param addressList IP:Port address list, like 192.168.1.1:7700
      * @param appName     name of the application
      * @param password    password of the application
      */
-    public OhMyClient(List<String> addressList, String appName, String password) {
+    public PowerJobClient(List<String> addressList, String appName, String password) {
 
         CommonUtils.requireNonNull(addressList, "addressList can't be null!");
         CommonUtils.requireNonNull(appName, "appName can't be null");
@@ -86,9 +86,9 @@ public class OhMyClient {
         }
 
         if (StringUtils.isEmpty(currentAddress)) {
-            throw new PowerJobException("no server available for OhMyClient");
+            throw new PowerJobException("no server available for PowerJobClient");
         }
-        log.info("[OhMyClient] {}'s OhMyClient bootstrap successfully, using server: {}", appName, currentAddress);
+        log.info("[PowerJobClient] {}'s PowerJobClient bootstrap successfully, using server: {}", appName, currentAddress);
     }
 
     private static String assertApp(String appName, String password, String url) throws IOException {
@@ -539,7 +539,7 @@ public class OhMyClient {
                 return res;
             }
         } catch (IOException e) {
-            log.warn("[OhMyClient] request url:{} failed, reason is {}.", url, e.toString());
+            log.warn("[PowerJobClient] request url:{} failed, reason is {}.", url, e.toString());
         }
 
         // 失败，开始重试
@@ -551,16 +551,16 @@ public class OhMyClient {
             try {
                 String res = HttpUtils.post(url, requestBody);
                 if (StringUtils.isNotEmpty(res)) {
-                    log.warn("[OhMyClient] server change: from({}) -> to({}).", currentAddress, addr);
+                    log.warn("[PowerJobClient] server change: from({}) -> to({}).", currentAddress, addr);
                     currentAddress = addr;
                     return res;
                 }
             } catch (IOException e) {
-                log.warn("[OhMyClient] request url:{} failed, reason is {}.", url, e.toString());
+                log.warn("[PowerJobClient] request url:{} failed, reason is {}.", url, e.toString());
             }
         }
 
-        log.error("[OhMyClient] do post for path: {} failed because of no server available in {}.", path, allAddress);
+        log.error("[PowerJobClient] do post for path: {} failed because of no server available in {}.", path, allAddress);
         throw new PowerJobException("no server available when send post request");
     }
 }

powerjob-client/src/test/java/tech/powerjob/client/test/ClientInitializer.java

@@ -1,6 +1,6 @@
 package tech.powerjob.client.test;
 
-import tech.powerjob.client.OhMyClient;
+import tech.powerjob.client.PowerJobClient;
 import org.junit.jupiter.api.BeforeAll;
 
 /**
@@ -11,10 +11,10 @@ import org.junit.jupiter.api.BeforeAll;
  */
 public class ClientInitializer {
 
-    protected static OhMyClient ohMyClient;
+    protected static PowerJobClient powerJobClient;
 
     @BeforeAll
     public static void initClient() throws Exception {
-        ohMyClient = new OhMyClient("127.0.0.1:7700", "powerjob-agent-test", "123");
+        powerJobClient = new PowerJobClient("127.0.0.1:7700", "powerjob-agent-test", "123");
     }
 }

powerjob-client/src/test/java/tech/powerjob/client/test/TestClient.java

@@ -1,7 +1,7 @@
 package tech.powerjob.client.test;
 
 import com.alibaba.fastjson.JSONObject;
-import tech.powerjob.client.OhMyClient;
+import tech.powerjob.client.PowerJobClient;
 import tech.powerjob.common.enums.ExecuteType;
 import tech.powerjob.common.enums.ProcessorType;
 import tech.powerjob.common.enums.TimeExpressionType;
@@ -16,7 +16,7 @@ import org.junit.jupiter.api.Test;
 import java.util.concurrent.TimeUnit;
 
 /**
- * Test cases for {@link OhMyClient}
+ * Test cases for {@link PowerJobClient}
  *
  * @author tjq
  * @author Echo009
@@ -45,86 +45,86 @@ class TestClient extends ClientInitializer {
         newJobInfo.setMinMemorySpace(1.2);
         newJobInfo.setMinDiskSpace(1.3);
 
-        ResultDTO<Long> resultDTO = ohMyClient.saveJob(newJobInfo);
+        ResultDTO<Long> resultDTO = powerJobClient.saveJob(newJobInfo);
         System.out.println(JSONObject.toJSONString(resultDTO));
         Assertions.assertNotNull(resultDTO);
     }
 
     @Test
     void testCopyJob() {
-        ResultDTO<Long> copyJobRes = ohMyClient.copyJob(JOB_ID);
+        ResultDTO<Long> copyJobRes = powerJobClient.copyJob(JOB_ID);
         System.out.println(JSONObject.toJSONString(copyJobRes));
         Assertions.assertNotNull(copyJobRes);
     }
 
     @Test
     void testFetchJob() {
-        ResultDTO<JobInfoDTO> fetchJob = ohMyClient.fetchJob(JOB_ID);
+        ResultDTO<JobInfoDTO> fetchJob = powerJobClient.fetchJob(JOB_ID);
         System.out.println(JSONObject.toJSONString(fetchJob));
         Assertions.assertNotNull(fetchJob);
     }
 
     @Test
     void testDisableJob() {
-        ResultDTO<Void> res = ohMyClient.disableJob(JOB_ID);
+        ResultDTO<Void> res = powerJobClient.disableJob(JOB_ID);
         System.out.println(res);
         Assertions.assertNotNull(res);
     }
 
     @Test
     void testEnableJob() {
-        ResultDTO<Void> res = ohMyClient.enableJob(JOB_ID);
+        ResultDTO<Void> res = powerJobClient.enableJob(JOB_ID);
         System.out.println(res);
         Assertions.assertNotNull(res);
     }
 
     @Test
     void testDeleteJob() {
-        ResultDTO<Void> res = ohMyClient.deleteJob(JOB_ID);
+        ResultDTO<Void> res = powerJobClient.deleteJob(JOB_ID);
         System.out.println(res);
         Assertions.assertNotNull(res);
     }
 
     @Test
     void testRun() {
-        ResultDTO<Long> res = ohMyClient.runJob(JOB_ID);
+        ResultDTO<Long> res = powerJobClient.runJob(JOB_ID);
         System.out.println(res);
         Assertions.assertNotNull(res);
     }
 
     @Test
     void testRunJobDelay() {
-        ResultDTO<Long> res = ohMyClient.runJob(JOB_ID, "this is instanceParams", 60000);
+        ResultDTO<Long> res = powerJobClient.runJob(JOB_ID, "this is instanceParams", 60000);
         System.out.println(res);
         Assertions.assertNotNull(res);
     }
 
     @Test
     void testFetchInstanceInfo() {
-        ResultDTO<InstanceInfoDTO> res = ohMyClient.fetchInstanceInfo(205436386851946560L);
+        ResultDTO<InstanceInfoDTO> res = powerJobClient.fetchInstanceInfo(205436386851946560L);
         System.out.println(res);
         Assertions.assertNotNull(res);
     }
 
     @Test
     void testStopInstance() {
-        ResultDTO<Void> res = ohMyClient.stopInstance(205436995885858880L);
+        ResultDTO<Void> res = powerJobClient.stopInstance(205436995885858880L);
         System.out.println(res);
         Assertions.assertNotNull(res);
     }
 
     @Test
     void testFetchInstanceStatus() {
-        ResultDTO<Integer> res = ohMyClient.fetchInstanceStatus(205436995885858880L);
+        ResultDTO<Integer> res = powerJobClient.fetchInstanceStatus(205436995885858880L);
         System.out.println(res);
         Assertions.assertNotNull(res);
     }
 
     @Test
     void testCancelInstanceInTimeWheel() {
-        ResultDTO<Long> startRes = ohMyClient.runJob(JOB_ID, "start by OhMyClient", 20000);
+        ResultDTO<Long> startRes = powerJobClient.runJob(JOB_ID, "start by OhMyClient", 20000);
         System.out.println("runJob result: " + JSONObject.toJSONString(startRes));
-        ResultDTO<Void> cancelRes = ohMyClient.cancelInstance(startRes.getData());
+        ResultDTO<Void> cancelRes = powerJobClient.cancelInstance(startRes.getData());
         System.out.println("cancelJob result: " + JSONObject.toJSONString(cancelRes));
         Assertions.assertTrue(cancelRes.isSuccess());
     }
@@ -132,20 +132,20 @@ class TestClient extends ClientInitializer {
     @Test
     @SneakyThrows
     void testCancelInstanceInDatabase() {
-        ResultDTO<Long> startRes = ohMyClient.runJob(15L, "start by OhMyClient", 2000000);
+        ResultDTO<Long> startRes = powerJobClient.runJob(15L, "start by OhMyClient", 2000000);
         System.out.println("runJob result: " + JSONObject.toJSONString(startRes));
 
         // Restart server manually and clear all the data in time wheeler.
         TimeUnit.MINUTES.sleep(1);
 
-        ResultDTO<Void> cancelRes = ohMyClient.cancelInstance(startRes.getData());
+        ResultDTO<Void> cancelRes = powerJobClient.cancelInstance(startRes.getData());
         System.out.println("cancelJob result: " + JSONObject.toJSONString(cancelRes));
         Assertions.assertTrue(cancelRes.isSuccess());
     }
 
     @Test
     void testRetryInstance() {
-        ResultDTO<Void> res = ohMyClient.retryInstance(169557545206153344L);
+        ResultDTO<Void> res = powerJobClient.retryInstance(169557545206153344L);
         System.out.println(res);
         Assertions.assertNotNull(res);
     }

powerjob-client/src/test/java/tech/powerjob/client/test/TestConcurrencyControl.java

@@ -28,7 +28,7 @@ class TestConcurrencyControl extends ClientInitializer {
         saveJobInfoRequest.setTimeExpressionType(TimeExpressionType.API);
         saveJobInfoRequest.setMaxInstanceNum(1);
 
-        Long jobId = ohMyClient.saveJob(saveJobInfoRequest).getData();
+        Long jobId = powerJobClient.saveJob(saveJobInfoRequest).getData();
 
         System.out.println("jobId: " + jobId);
 
@@ -37,7 +37,7 @@ class TestConcurrencyControl extends ClientInitializer {
         for (int i = 0; i < 100; i++) {
             String params = "index-" + i;
             pool.execute(() -> {
-                ResultDTO<Long> res = ohMyClient.runJob(jobId, params, 0);
+                ResultDTO<Long> res = powerJobClient.runJob(jobId, params, 0);
                 System.out.println(params + ": " + res);
             });
         }

powerjob-client/src/test/java/tech/powerjob/client/test/TestQuery.java

@@ -25,7 +25,7 @@ class TestQuery extends ClientInitializer {
 
     @Test
     void testFetchAllJob() {
-        ResultDTO<List<JobInfoDTO>> allJobRes = ohMyClient.fetchAllJob();
+        ResultDTO<List<JobInfoDTO>> allJobRes = powerJobClient.fetchAllJob();
         System.out.println(JSON.toJSONString(allJobRes));
     }
 
@@ -41,7 +41,7 @@ class TestQuery extends ClientInitializer {
                 .setProcessorTypeIn(Lists.newArrayList(ProcessorType.BUILT_IN.getV(), ProcessorType.SHELL.getV(), ProcessorType.EXTERNAL.getV()))
                 .setProcessorInfoLike("com.github.kfcfans");
 
-        ResultDTO<List<JobInfoDTO>> jobQueryResult = ohMyClient.queryJob(jobInfoQuery);
+        ResultDTO<List<JobInfoDTO>> jobQueryResult = powerJobClient.queryJob(jobInfoQuery);
         System.out.println(JSON.toJSONString(jobQueryResult));
         System.out.println(jobQueryResult.getData().size());
     }

powerjob-client/src/test/java/tech/powerjob/client/test/TestWorkflow.java

@@ -1,7 +1,7 @@
 package tech.powerjob.client.test;
 
 import com.alibaba.fastjson.JSONObject;
-import tech.powerjob.client.OhMyClient;
+import tech.powerjob.client.PowerJobClient;
 import tech.powerjob.common.enums.ExecuteType;
 import tech.powerjob.common.enums.ProcessorType;
 import tech.powerjob.common.enums.TimeExpressionType;
@@ -21,7 +21,7 @@ import org.junit.jupiter.api.Test;
 import java.util.List;
 
 /**
- * Test cases for {@link OhMyClient} workflow.
+ * Test cases for {@link PowerJobClient} workflow.
  *
  * @author tjq
  * @author Echo009
@@ -43,7 +43,7 @@ class TestWorkflow extends ClientInitializer {
         for (int i = 0; i < 5; i++) {
             SaveJobInfoRequest request = JSONObject.parseObject(JSONObject.toJSONBytes(base), SaveJobInfoRequest.class);
             request.setJobName(request.getJobName() + i);
-            ResultDTO<Long> res = ohMyClient.saveJob(request);
+            ResultDTO<Long> res = powerJobClient.saveJob(request);
             System.out.println(res);
             Assertions.assertNotNull(res);
 
@@ -61,7 +61,7 @@ class TestWorkflow extends ClientInitializer {
         req.setTimeExpressionType(TimeExpressionType.API);
 
         System.out.println("req ->" + JSONObject.toJSON(req));
-        ResultDTO<Long> res = ohMyClient.saveWorkflow(req);
+        ResultDTO<Long> res = powerJobClient.saveWorkflow(req);
         System.out.println(res);
         Assertions.assertNotNull(res);
 
@@ -85,7 +85,7 @@ class TestWorkflow extends ClientInitializer {
         saveWorkflowNodeRequest3.setType(WorkflowNodeType.JOB);
 
 
-        List<WorkflowNodeInfoDTO> nodeList = ohMyClient.saveWorkflowNode(Lists.newArrayList(saveWorkflowNodeRequest1,saveWorkflowNodeRequest2,saveWorkflowNodeRequest3)).getData();
+        List<WorkflowNodeInfoDTO> nodeList = powerJobClient.saveWorkflowNode(Lists.newArrayList(saveWorkflowNodeRequest1,saveWorkflowNodeRequest2,saveWorkflowNodeRequest3)).getData();
         System.out.println(nodeList);
         Assertions.assertNotNull(nodeList);
 
@@ -104,7 +104,7 @@ class TestWorkflow extends ClientInitializer {
 
         // 保存完整信息
         req.setDag(peWorkflowDAG);
-        res = ohMyClient.saveWorkflow(req);
+        res = powerJobClient.saveWorkflow(req);
 
         System.out.println(res);
         Assertions.assertNotNull(res);
@@ -113,7 +113,7 @@ class TestWorkflow extends ClientInitializer {
 
     @Test
     void testCopyWorkflow() {
-        ResultDTO<Long> res = ohMyClient.copyWorkflow(WF_ID);
+        ResultDTO<Long> res = powerJobClient.copyWorkflow(WF_ID);
         System.out.println(res);
         Assertions.assertNotNull(res);
     }
@@ -121,70 +121,70 @@ class TestWorkflow extends ClientInitializer {
 
     @Test
     void testDisableWorkflow() {
-        ResultDTO<Void> res = ohMyClient.disableWorkflow(WF_ID);
+        ResultDTO<Void> res = powerJobClient.disableWorkflow(WF_ID);
         System.out.println(res);
         Assertions.assertNotNull(res);
     }
 
     @Test
     void testDeleteWorkflow() {
-        ResultDTO<Void> res = ohMyClient.deleteWorkflow(WF_ID);
+        ResultDTO<Void> res = powerJobClient.deleteWorkflow(WF_ID);
         System.out.println(res);
         Assertions.assertNotNull(res);
     }
 
     @Test
     void testEnableWorkflow() {
-        ResultDTO<Void> res = ohMyClient.enableWorkflow(WF_ID);
+        ResultDTO<Void> res = powerJobClient.enableWorkflow(WF_ID);
         System.out.println(res);
         Assertions.assertNotNull(res);
     }
 
     @Test
     void testFetchWorkflowInfo() {
-        ResultDTO<WorkflowInfoDTO> res = ohMyClient.fetchWorkflow(WF_ID);
+        ResultDTO<WorkflowInfoDTO> res = powerJobClient.fetchWorkflow(WF_ID);
         System.out.println(res);
         Assertions.assertNotNull(res);
     }
 
     @Test
     void testRunWorkflow() {
-        ResultDTO<Long> res = ohMyClient.runWorkflow(WF_ID);
+        ResultDTO<Long> res = powerJobClient.runWorkflow(WF_ID);
         System.out.println(res);
         Assertions.assertNotNull(res);
     }
 
     @Test
     void testStopWorkflowInstance() {
-        ResultDTO<Void> res = ohMyClient.stopWorkflowInstance(149962433421639744L);
+        ResultDTO<Void> res = powerJobClient.stopWorkflowInstance(149962433421639744L);
         System.out.println(res);
         Assertions.assertNotNull(res);
     }
 
     @Test
     void testRetryWorkflowInstance() {
-        ResultDTO<Void> res = ohMyClient.retryWorkflowInstance(149962433421639744L);
+        ResultDTO<Void> res = powerJobClient.retryWorkflowInstance(149962433421639744L);
         System.out.println(res);
         Assertions.assertNotNull(res);
     }
 
     @Test
     void testMarkWorkflowNodeAsSuccess() {
-        ResultDTO<Void> res = ohMyClient.markWorkflowNodeAsSuccess(149962433421639744L, 1L);
+        ResultDTO<Void> res = powerJobClient.markWorkflowNodeAsSuccess(149962433421639744L, 1L);
         System.out.println(res);
         Assertions.assertNotNull(res);
     }
 
     @Test
     void testFetchWfInstanceInfo() {
-        ResultDTO<WorkflowInstanceInfoDTO> res = ohMyClient.fetchWorkflowInstanceInfo(149962433421639744L);
+        ResultDTO<WorkflowInstanceInfoDTO> res = powerJobClient.fetchWorkflowInstanceInfo(149962433421639744L);
         System.out.println(res);
         Assertions.assertNotNull(res);
     }
 
     @Test
     void testRunWorkflowPlus() {
-        ResultDTO<Long> res = ohMyClient.runWorkflow(WF_ID, "this is init Params 2", 90000);
+        ResultDTO<Long> res = powerJobClient.runWorkflow(WF_ID, "this is init Params 2", 90000);
         System.out.println(res);
         Assertions.assertNotNull(res);
     }

powerjob-server/powerjob-server-core/src/main/java/tech/powerjob/server/core/DispatchService.java

@@ -51,8 +51,6 @@ public class DispatchService {
     private InstanceMetadataService instanceMetadataService;
     @Resource
     private InstanceInfoRepository instanceInfoRepository;
-    @Resource
-    private DispatchService self;
 
     /**
      * 重新派发任务实例（不考虑实例当前的状态）
@@ -60,14 +58,14 @@ public class DispatchService {
      * @param jobInfo    任务信息（注意，这里传入的任务信息有可能为“空”）
      * @param instanceId 实例ID
      */
+    @UseSegmentLock(type = "redispatch", key = "#jobInfo.getId() ?: 0", concurrencyLevel = 16)
     public void redispatch(JobInfoDO jobInfo, long instanceId) {
         InstanceInfoDO instance = instanceInfoRepository.findByInstanceId(instanceId);
         // 将状态重置为等待派发
         instance.setStatus(InstanceStatus.WAITING_DISPATCH.getV());
         instance.setGmtModified(new Date());
         instanceInfoRepository.saveAndFlush(instance);
-        // support for aop
+        dispatch(jobInfo, instanceId);
-        self.dispatch(jobInfo, instanceId);
     }
 
     /**

powerjob-server/powerjob-server-core/src/main/java/tech/powerjob/server/core/workflow/WorkflowService.java

@@ -10,6 +10,7 @@ import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
 import tech.powerjob.common.PowerJobException;
 import tech.powerjob.common.enums.TimeExpressionType;
+import tech.powerjob.common.enums.WorkflowNodeType;
 import tech.powerjob.common.model.PEWorkflowDAG;
 import tech.powerjob.common.request.http.SaveWorkflowNodeRequest;
 import tech.powerjob.common.request.http.SaveWorkflowRequest;
@@ -134,7 +135,7 @@ public class WorkflowService {
         }
         dag.setNodes(newNodes);
         int deleteCount = workflowNodeInfoRepository.deleteByWorkflowIdAndIdNotIn(wfId, nodeIdList);
-        log.warn("[WorkflowService-{}]delete {} dissociative nodes of workflow", wfId, deleteCount);
+        log.warn("[WorkflowService-{}] delete {} dissociative nodes of workflow", wfId, deleteCount);
         return JSON.toJSONString(dag);
     }
 
@@ -282,7 +283,6 @@ public class WorkflowService {
 
     /**
      * 保存工作流节点（新增 或者 保存）
-     * 允许工作流 ID 为空
      *
      * @param workflowNodeRequestList 工作流节点
      * @return 更新 或者 创建后的工作流节点信息
@@ -307,16 +307,20 @@ public class WorkflowService {
                 workflowNodeInfo = new WorkflowNodeInfoDO();
                 workflowNodeInfo.setGmtCreate(new Date());
             }
-            JobInfoDO jobInfoDO = jobInfoRepository.findById(req.getJobId()).orElseThrow(() -> new IllegalArgumentException("can't find job by id: " + req.getJobId()));
+
-            if (!jobInfoDO.getAppId().equals(appId)) {
+            // valid job info
-                throw new PowerJobException("Permission Denied! can't use other app's job!");
+            if (req.getType() == WorkflowNodeType.JOB) {
+                JobInfoDO jobInfoDO = jobInfoRepository.findById(req.getJobId()).orElseThrow(() -> new IllegalArgumentException("can't find job by id: " + req.getJobId()));
+                if (!jobInfoDO.getAppId().equals(appId)) {
+                    throw new PowerJobException("Permission Denied! can't use other app's job!");
+                }
+                if (StringUtils.isEmpty(workflowNodeInfo.getNodeName())) {
+                    workflowNodeInfo.setNodeName(jobInfoDO.getJobName());
+                }
             }
+
             BeanUtils.copyProperties(req, workflowNodeInfo);
             workflowNodeInfo.setType(req.getType().getCode());
-            // 如果名称为空则默认取任务名称
-            if (StringUtils.isEmpty(workflowNodeInfo.getNodeName())) {
-                workflowNodeInfo.setNodeName(jobInfoDO.getJobName());
-            }
             workflowNodeInfo.setGmtModified(new Date());
             workflowNodeInfo = workflowNodeInfoRepository.saveAndFlush(workflowNodeInfo);
             res.add(workflowNodeInfo);

powerjob-server/powerjob-server-migrate/src/main/java/tech/powerjob/server/migrate/V3ToV4MigrateService.java

@@ -37,7 +37,7 @@ import java.util.concurrent.TimeUnit;
 @Slf4j
 public class V3ToV4MigrateService {
 
-    private static final String MIGRATE_LOCK_TEMPLATE = "migrateLock-%s-%s";
+    private static final String MIGRATE_LOCK_TEMPLATE = "v3to4MigrateLock-%s-%s";
 
     @Resource
     private LockService lockService;
@@ -252,6 +252,4 @@ public class V3ToV4MigrateService {
         return true;
 
     }
-
-
 }