working on #141 #142 #143

Author: phasenraum2010

src/main/java/org/woehlke/twitterwall/frontend/controller/TaskController.java

@@ -11,17 +11,14 @@
 import org.springframework.web.bind.annotation.PathVariable;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestParam;
-import org.woehlke.twitterwall.conf.TwitterProperties;
 import org.woehlke.twitterwall.conf.TwitterwallFrontendProperties;
 import org.woehlke.twitterwall.frontend.controller.common.Symbols;
 import org.woehlke.twitterwall.frontend.controller.common.ControllerHelper;
 import org.woehlke.twitterwall.oodm.entities.Task;
 import org.woehlke.twitterwall.oodm.entities.TaskHistory;
 import org.woehlke.twitterwall.oodm.service.TaskHistoryService;
 import org.woehlke.twitterwall.oodm.service.TaskService;
-import org.woehlke.twitterwall.scheduled.service.facade.FetchTweetsFromTwitterSearch;
-import org.woehlke.twitterwall.scheduled.service.facade.FetchUsersFromDefinedUserList;
-import org.woehlke.twitterwall.scheduled.service.facade.UpdateTweets;
+import org.woehlke.twitterwall.scheduled.mq.endoint.StartTask;
 
 /**
  * Created by tw on 11.07.17.
@@ -125,54 +122,45 @@ public String updateUserProfilesRequest(Model model) {
 
     @Async
     protected void fetchTweetsFromTwitterSearch() {
-        fetchTweetsFromTwitterSearch.fetchTweetsFromTwitterSearch();;
+        startTask.fetchTweetsFromTwitterSearch();
     }
 
     @Async
     protected void updateTweets() {
-        updateTweets.updateTweets();
+        startTask.updateTweets();
     }
 
     @Async
     protected void fetchUsersFromDefinedUserList(){
-        fetchUsersFromDefinedUserList.fetchUsersFromDefinedUserList();
+        startTask.fetchUsersFromDefinedUserList();
     }
 
     @Async
     protected void updateUserProfilesFromMentions(){
-
+        startTask.updateUserProfilesFromMentions();
     }
 
     @Async
     protected void updateUserProfiles() {
-
+        startTask.updateUserProfiles();
     }
 
     private final TaskService taskService;
 
     private final TaskHistoryService taskHistoryService;
 
-    private final FetchTweetsFromTwitterSearch fetchTweetsFromTwitterSearch;
-
-    private final FetchUsersFromDefinedUserList fetchUsersFromDefinedUserList;
-
-    private final UpdateTweets updateTweets;
-
     private final TwitterwallFrontendProperties twitterwallFrontendProperties;
 
-    private final TwitterProperties twitterProperties;
-
     private final ControllerHelper controllerHelper;
 
+    private final StartTask startTask;
+
     @Autowired
-    public TaskController(TaskService taskService, TaskHistoryService taskHistoryService, FetchTweetsFromTwitterSearch fetchTweetsFromTwitterSearch, FetchUsersFromDefinedUserList fetchUsersFromDefinedUserList, UpdateTweets updateTweets, TwitterwallFrontendProperties twitterwallFrontendProperties, TwitterProperties twitterProperties, ControllerHelper controllerHelper) {
+    public TaskController(TaskService taskService, TaskHistoryService taskHistoryService, TwitterwallFrontendProperties twitterwallFrontendProperties, ControllerHelper controllerHelper, StartTask startTask) {
         this.taskService = taskService;
         this.taskHistoryService = taskHistoryService;
-        this.fetchTweetsFromTwitterSearch = fetchTweetsFromTwitterSearch;
-        this.fetchUsersFromDefinedUserList = fetchUsersFromDefinedUserList;
-        this.updateTweets = updateTweets;
         this.twitterwallFrontendProperties = twitterwallFrontendProperties;
-        this.twitterProperties = twitterProperties;
         this.controllerHelper = controllerHelper;
+        this.startTask = startTask;
     }
 }

src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/impl/StartTaskImpl.java

@@ -85,31 +85,54 @@ private void messageSender(final TaskType taskType) {
     @Scheduled(fixedRate = FIXED_RATE_FOR_SCHEDULAR_FETCH_TWEETS)
     public void fetchTweetsFromTwitterSearch() {
         TaskType taskType = TaskType.FETCH_TWEETS_FROM_TWITTER_SEARCH;
-        sendAndReceive(taskType);
+        sendAndReceiveTweet(taskType);
     }
 
     @Override
     @Scheduled(fixedRate = FIXED_RATE_FOR_SCHEDULAR_UPDATE_TWEETS)
     public void updateTweets() {
         TaskType taskType = TaskType.UPDATE_TWEETS;
-        sendAndReceive(taskType);
+        sendAndReceiveTweet(taskType);
     }
 
     @Override
     @Scheduled(fixedRate = FIXED_RATE_FOR_SCHEDULAR_UPDATE_USER)
     public void updateUserProfiles() {
         TaskType taskType = TaskType.UPDATE_USER_PROFILES;
-        sendAndReceive(taskType);
+        sendAndReceiveUser(taskType);
     }
 
     @Override
     @Scheduled(fixedRate = FIXED_RATE_FOR_SCHEDULAR_UPDATE_USER_BY_MENTION)
     public void updateUserProfilesFromMentions() {
         TaskType taskType = TaskType.UPDATE_USER_PROFILES_FROM_MENTIONS;
-        sendAndReceive(taskType);
+        sendAndReceiveUser(taskType);
     }
 
-    private void sendAndReceive(TaskType taskType){
+    private void sendAndReceiveTweet(TaskType taskType){
+        CountedEntities countedEntities = countedEntitiesService.countAll();
+        Task task = taskService.create("Start via MQ", taskType,countedEntities);
+        TaskMessage taskMessage = new TaskMessage(task.getId(), taskType, task.getTimeStarted());
+        Message<TaskMessage> mqMessage = MessageBuilder.withPayload(taskMessage)
+                .setHeader("task_id", task.getId())
+                .setHeader("task_uid", task.getUniqueId())
+                .setHeader("task_type", task.getTaskType())
+                .build();
+        MessagingTemplate mqTemplate = new MessagingTemplate();
+        Message<?> returnedMessage = mqTemplate.sendAndReceive(startTaskChannel, mqMessage);
+        Object o = returnedMessage.getPayload();
+        countedEntities = countedEntitiesService.countAll();
+        if( o instanceof TweetResultList){
+            TweetResultList msg = (TweetResultList) o;
+            long taskId = msg.getTaskId();
+            task = taskService.findById(taskId);
+            taskService.done(task,countedEntities);
+        } else {
+            taskService.error(task,"Wrong type of returnedMessage",countedEntities);
+        }
+    }
+
+    private void sendAndReceiveUser(TaskType taskType){
         CountedEntities countedEntities = countedEntitiesService.countAll();
         Task task = taskService.create("Start via MQ", taskType,countedEntities);
         TaskMessage taskMessage = new TaskMessage(task.getId(), taskType, task.getTimeStarted());
@@ -136,7 +159,7 @@ private void sendAndReceive(TaskType taskType){
     @Scheduled(fixedRate = FIXED_RATE_FOR_SCHEDULAR_FETCH_USER_LIST)
     public void fetchUsersFromDefinedUserList() {
         TaskType taskType = TaskType.FETCH_USERS_FROM_DEFINED_USER_LIST;
-        sendAndReceive(taskType);
+        sendAndReceiveUser(taskType);
     }
 
     @Override