Merge pull request #125 from FlowCI/develop

Develop

Author: yang.guo

platform-api/src/main/java/com/flow/platform/api/config/AppConfig.java

@@ -38,6 +38,8 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
+import org.springframework.context.event.ApplicationEventMulticaster;
+import org.springframework.context.event.SimpleApplicationEventMulticaster;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
 /**
@@ -49,7 +51,7 @@ public class AppConfig extends AppConfigBase {
 
     public final static String NAME = "API";
 
-    public final static String VERSION = "alpha-0.1";
+    public final static String VERSION = "0.1.0";
 
     public final static String DEFAULT_YML_FILE = ".flow.yml";
 
@@ -61,13 +63,22 @@ public class AppConfig extends AppConfigBase {
 
     private final static String THREAD_NAME_PREFIX = "async-task-";
 
+    private final static String MULTICASTER_THREAD_NAME_PREFIX = "multi_async-task-";
+
+    private final static int MULTICASTER_ASYNC_POOL_SIZE = 1;
+
     public final static String DEFAULT_USER_EMAIL = "[email protected]";
     public final static String DEFAULT_USER_NAME = "admin";
+
     public final static String DEFAULT_USER_PASSWORD = "123456";
 
     private final static ThreadPoolTaskExecutor executor =
         ThreadUtil.createTaskExecutor(ASYNC_POOL_SIZE, ASYNC_POOL_SIZE / 10, 100, THREAD_NAME_PREFIX);
 
+    private final static ThreadPoolTaskExecutor multicasterExecutor =
+        ThreadUtil.createTaskExecutor(MULTICASTER_ASYNC_POOL_SIZE, MULTICASTER_ASYNC_POOL_SIZE, 1000,
+            MULTICASTER_THREAD_NAME_PREFIX);
+
     @Value("${api.workspace}")
     private String workspace;
 
@@ -85,6 +96,14 @@ public Path workspace() {
         }
     }
 
+    @Bean(name = "applicationEventMulticaster")
+    public ApplicationEventMulticaster simpleApplicationEventMulticaster() {
+        multicasterExecutor.initialize();
+        SimpleApplicationEventMulticaster eventMulticaster = new SimpleApplicationEventMulticaster();
+        eventMulticaster.setTaskExecutor(multicasterExecutor);
+        return eventMulticaster;
+    }
+
     @Bean
     public ThreadLocal<User> currentUser() {
         return new ThreadLocal<>();

platform-api/src/main/java/com/flow/platform/api/config/WebConfig.java

@@ -83,7 +83,8 @@ public void addCorsMappings(CorsRegistry registry) {
             .allowedOrigins("*")
             .allowedMethods("*")
             .allowCredentials(true)
-            .allowedHeaders("origin", "content-type", "accept", "x-requested-with", "X-Authorization", "authenticate", "library");
+            .allowedHeaders("origin", "content-type", "accept", "x-requested-with", "X-Authorization", "authenticate",
+                "library");
     }
 
     @Bean

platform-api/src/main/java/com/flow/platform/api/consumer/CmdLoggingConsumer.java

@@ -84,7 +84,7 @@ protected void handleTextMessage(WebSocketSession session, TextMessage message)
      */
     private void sendCmdLog(String cmdId, String content, String number) {
         String event = String.format("/topic/cmd/%s", cmdId);
-        template.convertAndSend(event, "{\"number\": \"" + number + "\", \"content\": \"" + content + "\"}");
+        template.convertAndSend(event, "{\"number\": " + number + ", \"content\": \"" + content + "\"}");
     }
 
     /**

platform-api/src/main/java/com/flow/platform/api/consumer/JobEventPushHandler.java

@@ -20,6 +20,7 @@
 import com.flow.platform.api.domain.job.Job;
 import com.flow.platform.api.push.PushHandler;
 import com.flow.platform.api.service.job.JobService;
+import com.flow.platform.core.exception.NotFoundException;
 import java.math.BigInteger;
 import org.springframework.beans.factory.annotation.Autowired;
 
@@ -37,4 +38,15 @@ protected void push(BigInteger jobId) {
         super.push(jobTopic, job);
     }
 
+    //TODO: because transaction not found job, add READ_UNCOMMITTED also no use
+    protected void push(Job job) {
+        Job jobConsist;
+        try {
+            jobConsist = jobService.find(job.getId());
+        } catch (NotFoundException e) {
+            jobConsist = job;
+        }
+        String jobTopic = String.format("%s/%s", WebSocketConfig.TOPIC_FOR_JOB, jobConsist.getNodePath());
+        super.push(jobTopic, jobConsist);
+    }
 }

platform-api/src/main/java/com/flow/platform/api/consumer/JobStatusEventConsumer.java

@@ -20,9 +20,7 @@
 import com.flow.platform.api.domain.job.JobStatus;
 import com.flow.platform.api.events.JobStatusChangeEvent;
 import com.flow.platform.api.service.MessageService;
-import com.flow.platform.util.ExceptionUtil;
 import com.flow.platform.util.Logger;
-import java.math.BigInteger;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationListener;
 import org.springframework.core.task.TaskExecutor;
@@ -44,9 +42,10 @@ public class JobStatusEventConsumer extends JobEventPushHandler implements Appli
 
     @Override
     public void onApplicationEvent(JobStatusChangeEvent event) {
-        LOGGER.debug("Job %s status change event from %s to %s", event.getJob().getId(), event.getFrom(), event.getTo());
+        LOGGER
+            .debug("Job %s status change event from %s to %s", event.getJob().getId(), event.getFrom(), event.getTo());
 
-        push(event.getJob().getId());
+        push(event.getJob());
 
         // async send message TODO:// only send failure message
         if (Job.FAILURE_STATUS.contains(event.getTo())) {

platform-api/src/main/java/com/flow/platform/api/consumer/NodeStatusEventConsumer.java

@@ -33,7 +33,6 @@ public void onApplicationEvent(NodeStatusChangeEvent event) {
         NodeResultKey resultKey = event.getResultKey();
         LOGGER.debug("Node result %s status change event from %s to %s",
             resultKey.getPath(), event.getFrom(), event.getTo());
-
         push(resultKey.getJobId());
     }
 }

platform-api/src/main/java/com/flow/platform/api/service/AgentServiceImpl.java

@@ -26,6 +26,7 @@
 import com.flow.platform.api.service.job.JobService;
 import com.flow.platform.api.util.PlatformURL;
 import com.flow.platform.core.exception.HttpException;
+import com.flow.platform.core.exception.IllegalParameterException;
 import com.flow.platform.core.exception.IllegalStatusException;
 import com.flow.platform.core.service.ApplicationEventService;
 import com.flow.platform.domain.Agent;
@@ -38,6 +39,7 @@
 import com.flow.platform.domain.Jsonable;
 import com.flow.platform.util.CollectionUtil;
 import com.flow.platform.util.Logger;
+import com.flow.platform.util.StringUtil;
 import com.flow.platform.util.http.HttpClient;
 import com.flow.platform.util.http.HttpResponse;
 import com.flow.platform.util.http.HttpURL;
@@ -144,6 +146,10 @@ public Boolean shutdown(String zone, String name, String password) {
 
     @Override
     public AgentWithFlow create(AgentPath agentPath) {
+        if (StringUtil.hasSpace(agentPath.getZone()) || StringUtil.hasSpace(agentPath.getName())) {
+            throw new IllegalParameterException("Zone name or agent name cannot contain empty space");
+        }
+
         try {
             AgentPathWithWebhook pathWithWebhook = new AgentPathWithWebhook(agentPath, buildAgentWebhook());
 
@@ -158,9 +164,7 @@ public AgentWithFlow create(AgentPath agentPath) {
             }
 
             Agent agent = Agent.parse(response.getBody(), Agent.class);
-
-            AgentWithFlow agentWithFlow = new AgentWithFlow(agent, null);
-            return agentWithFlow;
+            return new AgentWithFlow(agent, null);
 
         } catch (UnsupportedEncodingException | JsonSyntaxException e) {
             throw new IllegalStatusException("Unable to create agent", e);

platform-api/src/main/java/com/flow/platform/api/service/job/JobNodeServiceImpl.java

@@ -25,14 +25,12 @@
 import org.springframework.cache.Cache;
 import org.springframework.cache.CacheManager;
 import org.springframework.stereotype.Service;
-import org.springframework.transaction.annotation.Transactional;
 
 /**
  * @author lhl
  */
 
 @Service
-@Transactional
 public class JobNodeServiceImpl implements JobNodeService {
 
     private final Logger LOGGER = new Logger(JobYml.class);

platform-api/src/main/java/com/flow/platform/api/service/job/JobSearchServiceImpl.java

@@ -82,7 +82,7 @@ public List<Job> match(SearchCondition searchCondition, List<Job> jobs) {
             for (Job job : jobs) {
                 if (job.getNumber().toString().equals(words)) { // compare job number
                     copyJobs.add(job);
-                } else if (words.equals(
+                } else if (job.getRootResult() != null && words.equals(
                     job.getRootResult().getOutputs().get(GitEnvs.FLOW_GIT_BRANCH.toString()))) { //compare branch
                     copyJobs.add(job);
                 }
@@ -105,7 +105,8 @@ public List<Job> match(SearchCondition searchCondition, List<Job> jobs) {
 
             List<Job> copyJobs = new LinkedList<>();
             for (Job job : jobs) {
-                if (branch.equals(job.getRootResult().getOutputs().get(GitEnvs.FLOW_GIT_BRANCH.toString()))) {
+                if (job.getRootResult() != null && branch
+                    .equals(job.getRootResult().getOutputs().get(GitEnvs.FLOW_GIT_BRANCH.toString()))) {
                     copyJobs.add(job);
                 }
             }

platform-api/src/main/java/com/flow/platform/api/service/job/JobServiceImpl.java

@@ -74,21 +74,16 @@
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
-import org.springframework.transaction.annotation.Isolation;
-import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
 
 /**
  * @author yh@firim
  */
-@Service(value = "jobService")
-@Transactional(isolation = Isolation.REPEATABLE_READ)
+@Service
 public class JobServiceImpl extends ApplicationEventService implements JobService {
 
     private static Logger LOGGER = new Logger(JobService.class);
 
-    private Integer RETRY_TIMEs = 5;
-
     private final Integer createSessionRetryTimes = 5;
 
     @Value("${task.job.toggle.execution_timeout}")
@@ -268,6 +263,7 @@ public void callback(CmdCallbackQueueItem cmdQueueItem) {
     }
 
     @Override
+    @Transactional
     public void deleteJob(String path) {
         List<BigInteger> jobIds = jobDao.findJobIdsByPath(path);
         // TODO :  Late optimization and paging jobIds
@@ -392,7 +388,6 @@ private void onRunShellCallback(String path, Cmd cmd, Job job) {
     }
 
     @Override
-    @Transactional(propagation = Propagation.NOT_SUPPORTED)
     public void enterQueue(CmdCallbackQueueItem cmdQueueItem) {
         cmdCallbackQueue.enqueue(cmdQueueItem);
     }