working on #138

phasenraum2010 · phasenraum2010 · commit 4512f8b8a0f1 · 2017-07-23T20:58:36.000+02:00
diff --git a/src/main/java/org/woehlke/twitterwall/conf/TwitterwallSchedulerProperties.java b/src/main/java/org/woehlke/twitterwall/conf/TwitterwallSchedulerProperties.java
@@ -126,6 +126,8 @@ public static class Facade {
 
         private List<Long> idTwitterToFetchForUserControllerTest = new ArrayList<Long>();
 
+        private List<String> screenNamesToFetchForUserControllerTest = new ArrayList<String>();
+
         public List<Long> getIdTwitterToFetchForTweetTest() {
             return idTwitterToFetchForTweetTest;
         }
@@ -141,6 +143,14 @@ public List<Long> getIdTwitterToFetchForUserControllerTest() {
         public void setIdTwitterToFetchForUserControllerTest(List<Long> idTwitterToFetchForUserControllerTest) {
             this.idTwitterToFetchForUserControllerTest = idTwitterToFetchForUserControllerTest;
         }
+
+        public List<String> getScreenNamesToFetchForUserControllerTest() {
+            return screenNamesToFetchForUserControllerTest;
+        }
+
+        public void setScreenNamesToFetchForUserControllerTest(List<String> screenNamesToFetchForUserControllerTest) {
+            this.screenNamesToFetchForUserControllerTest = screenNamesToFetchForUserControllerTest;
+        }
     }
 
     public Facade getFacade() {
diff --git a/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/CreateImprintUser.java b/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/CreateImprintUser.java
@@ -0,0 +1,10 @@
+package org.woehlke.twitterwall.scheduled.mq.endoint;
+
+import org.springframework.messaging.Message;
+import org.woehlke.twitterwall.scheduled.mq.msg.TaskMessage;
+import org.woehlke.twitterwall.scheduled.mq.msg.TwitterProfileMessage;
+
+public interface CreateImprintUser {
+
+    Message<TwitterProfileMessage> createImprintUser(Message<TaskMessage> mqMessage);
+}
diff --git a/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/CreateTestDataForUsers.java b/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/CreateTestDataForUsers.java
@@ -0,0 +1,12 @@
+package org.woehlke.twitterwall.scheduled.mq.endoint;
+
+import org.springframework.messaging.Message;
+import org.woehlke.twitterwall.scheduled.mq.msg.TaskMessage;
+import org.woehlke.twitterwall.scheduled.mq.msg.TwitterProfileMessage;
+
+import java.util.List;
+
+public interface CreateTestDataForUsers {
+
+    List<TwitterProfileMessage> splitMessage(Message<TaskMessage> message);
+}
diff --git a/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/FetchDataFromRemoteTwitterApi.java b/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/FetchDataFromRemoteTwitterApi.java
@@ -1,7 +1,6 @@
 package org.woehlke.twitterwall.scheduled.mq.endoint;
 
 import org.springframework.messaging.Message;
-import org.woehlke.twitterwall.scheduled.mq.msg.TwitterProfileMessage;
 import org.woehlke.twitterwall.scheduled.mq.msg.TaskMessage;
 
 public interface FetchDataFromRemoteTwitterApi {
@@ -11,6 +10,4 @@ public interface FetchDataFromRemoteTwitterApi {
     void fetchUsersFromDefinedUserList(Message<TaskMessage> mqMessage);
 
     void createTestDataForUser(Message<TaskMessage> mqMessage);
-
-    Message<TwitterProfileMessage> createImprintUser(Message<TaskMessage> mqMessage);
 }
diff --git a/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/UserFinisher.java b/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/UserFinisher.java
@@ -0,0 +1,12 @@
+package org.woehlke.twitterwall.scheduled.mq.endoint;
+
+import org.springframework.messaging.Message;
+import org.woehlke.twitterwall.oodm.entities.User;
+import org.woehlke.twitterwall.scheduled.mq.msg.UserResultList;
+
+import java.util.List;
+
+public interface UserFinisher {
+
+    UserResultList finish(Message<List<User>> incomingMessageList);
+}
diff --git a/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/UserPersistor.java b/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/UserPersistor.java
@@ -6,5 +6,5 @@
 
 public interface UserPersistor {
 
-    User persistUser(Message<UserMessage> incomingUserMessage);
+    UserMessage persistUser(Message<UserMessage> incomingUserMessage);
 }
diff --git a/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/impl/CreateImprintUserImpl.java b/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/impl/CreateImprintUserImpl.java
@@ -0,0 +1,41 @@
+package org.woehlke.twitterwall.scheduled.mq.endoint.impl;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.social.twitter.api.TwitterProfile;
+import org.springframework.stereotype.Component;
+import org.woehlke.twitterwall.conf.TwitterwallFrontendProperties;
+import org.woehlke.twitterwall.scheduled.mq.endoint.CreateImprintUser;
+import org.woehlke.twitterwall.scheduled.mq.msg.TaskMessage;
+import org.woehlke.twitterwall.scheduled.mq.msg.TwitterProfileMessage;
+import org.woehlke.twitterwall.scheduled.service.backend.TwitterApiService;
+
+@Component("mqCreateImprintUser")
+public class CreateImprintUserImpl implements CreateImprintUser {
+
+    private final TwitterApiService twitterApiService;
+
+    private final TwitterwallFrontendProperties twitterwallFrontendProperties;
+
+    @Autowired
+    public CreateImprintUserImpl(TwitterApiService twitterApiService, TwitterwallFrontendProperties twitterwallFrontendProperties) {
+        this.twitterApiService = twitterApiService;
+        this.twitterwallFrontendProperties = twitterwallFrontendProperties;
+    }
+
+    @Override
+    public Message<TwitterProfileMessage> createImprintUser(Message<TaskMessage> mqMessageIn) {
+        String logMsg = "createImprintUser: ";
+        TaskMessage receivedMessage = mqMessageIn.getPayload();
+        String screenName = twitterwallFrontendProperties.getImprintScreenName();
+        TwitterProfile twitterProfile = twitterApiService.getUserProfileForScreenName(screenName);
+        TwitterProfileMessage outMsg = new TwitterProfileMessage(receivedMessage,screenName,twitterProfile);
+        Message<TwitterProfileMessage> mqMessageOut = MessageBuilder.withPayload(outMsg).copyHeaders(mqMessageIn.getHeaders())
+            .setHeader("twitter_profile_id", twitterProfile.getId())
+            .build();
+        return mqMessageOut;
+    }
+}
+
+
diff --git a/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/impl/CreateTestDataForTweetsImpl.java b/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/impl/CreateTestDataForTweetsImpl.java
@@ -1,7 +1,6 @@
 package org.woehlke.twitterwall.scheduled.mq.endoint.impl;
 
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.integration.splitter.AbstractMessageSplitter;
 import org.springframework.messaging.Message;
 import org.springframework.social.twitter.api.Tweet;
 import org.springframework.stereotype.Component;
diff --git a/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/impl/CreateTestDataForUsersImpl.java b/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/impl/CreateTestDataForUsersImpl.java
@@ -0,0 +1,47 @@
+package org.woehlke.twitterwall.scheduled.mq.endoint.impl;
+
+import org.springframework.messaging.Message;
+import org.springframework.social.twitter.api.TwitterProfile;
+import org.springframework.stereotype.Component;
+import org.woehlke.twitterwall.conf.TwitterwallSchedulerProperties;
+import org.woehlke.twitterwall.oodm.entities.Task;
+import org.woehlke.twitterwall.oodm.service.TaskService;
+import org.woehlke.twitterwall.scheduled.mq.endoint.CreateTestDataForUsers;
+import org.woehlke.twitterwall.scheduled.mq.msg.TaskMessage;
+import org.woehlke.twitterwall.scheduled.mq.msg.TwitterProfileMessage;
+import org.woehlke.twitterwall.scheduled.service.backend.TwitterApiService;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@Component("mqCreateTestDataForUsers")
+public class CreateTestDataForUsersImpl implements CreateTestDataForUsers {
+
+    private final TwitterwallSchedulerProperties twitterwallSchedulerProperties;
+
+    private final TwitterApiService twitterApiService;
+
+    private final TaskService taskService;
+
+    public CreateTestDataForUsersImpl(TwitterwallSchedulerProperties twitterwallSchedulerProperties, TwitterApiService twitterApiService, TaskService taskService) {
+        this.twitterwallSchedulerProperties = twitterwallSchedulerProperties;
+        this.twitterApiService = twitterApiService;
+        this.taskService = taskService;
+    }
+
+    @Override
+    public List<TwitterProfileMessage> splitMessage(Message<TaskMessage> message) {
+        List<TwitterProfileMessage> userProfileList = new ArrayList<>();
+        TaskMessage msgIn = message.getPayload();
+        long id = msgIn.getTaskId();
+        Task task = taskService.findById(id);
+        task =  taskService.start(task);
+        List<String> userIdList = twitterwallSchedulerProperties.getFacade().getScreenNamesToFetchForUserControllerTest();
+        for (String screenName : userIdList) {
+            TwitterProfile userProfile = twitterApiService.getUserProfileForScreenName(screenName);
+            TwitterProfileMessage userMsg = new TwitterProfileMessage(msgIn,userProfile);
+            userProfileList.add(userMsg);
+        }
+        return userProfileList;
+    }
+}
diff --git a/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/impl/FetchDataFromRemoteTwitterApiImpl.java b/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/impl/FetchDataFromRemoteTwitterApiImpl.java
@@ -3,16 +3,13 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
-import org.springframework.social.twitter.api.TwitterProfile;
 import org.springframework.stereotype.Component;
 import org.woehlke.twitterwall.conf.TwitterwallFrontendProperties;
 import org.woehlke.twitterwall.oodm.entities.Task;
 import org.woehlke.twitterwall.oodm.service.TaskService;
 import org.woehlke.twitterwall.scheduled.service.backend.TwitterApiService;
 import org.woehlke.twitterwall.scheduled.mq.endoint.FetchDataFromRemoteTwitterApi;
-import org.woehlke.twitterwall.scheduled.mq.msg.TwitterProfileMessage;
 import org.woehlke.twitterwall.scheduled.mq.msg.TaskMessage;
 
 @Component("mqFetchDataFromRemoteTwitterApi")
@@ -64,17 +61,4 @@ public void createTestDataForUser(Message<TaskMessage> mqMessage) {
         this.react(logMsg, mqMessage);
     }
 
-    @Override
-    public Message<TwitterProfileMessage> createImprintUser(Message<TaskMessage> mqMessageIn) {
-        String logMsg = "createImprintUser: ";
-        TaskMessage receivedMessage = mqMessageIn.getPayload();
-        String screenName = twitterwallFrontendProperties.getImprintScreenName();
-        TwitterProfile twitterProfile = twitterApiService.getUserProfileForScreenName(screenName);
-        TwitterProfileMessage outMsg = new TwitterProfileMessage(receivedMessage,screenName,twitterProfile);
-        Message<TwitterProfileMessage> mqMessageOut = MessageBuilder.withPayload(outMsg).copyHeaders(mqMessageIn.getHeaders())
-            .setHeader("twitter_profile_id", twitterProfile.getId())
-            .build();
-        return mqMessageOut;
-    }
-
 }
diff --git a/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/impl/StartTaskImpl.java b/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/impl/StartTaskImpl.java
@@ -15,6 +15,8 @@
 import org.woehlke.twitterwall.scheduled.mq.endoint.StartTask;
 import org.woehlke.twitterwall.scheduled.mq.msg.TaskMessage;
 import org.woehlke.twitterwall.scheduled.mq.msg.TweetResultList;
+import org.woehlke.twitterwall.scheduled.mq.msg.UserMessage;
+import org.woehlke.twitterwall.scheduled.mq.msg.UserResultList;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -121,18 +123,41 @@ public User createImprintUser() {
         MessagingTemplate mqTemplate = new MessagingTemplate();
         Message<?> returnedMessage = mqTemplate.sendAndReceive(startTaskChannel, mqMessage);
         Object o = returnedMessage.getPayload();
-        if( o instanceof User){
-            return (User) o;
-        }    else {
+        if( o instanceof UserMessage){
+            UserMessage msg = (UserMessage) o;
+            long taskId = msg.getTaskId();
+            task = taskService.findById(taskId);
+            taskService.done(task);
+            return msg.getUser();
+        } else {
+            taskService.error(task,"Wrong type of returnedMessage");
             return null;
         }
     }
 
     @Override
     public List<User> createTestDataForUser() {
         TaskType taskType = TaskType.CONTROLLER_GET_TESTDATA_USER;
-        messageSender(taskType);
-        return new ArrayList<>();
+        Task task = taskService.create("Start via MQ", taskType);
+        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();
+        if( o instanceof UserResultList){
+            UserResultList result = (UserResultList) o;
+            long taskId =result.getTaskId();
+            task = taskService.findById(taskId);
+            taskService.done(task);
+            return result.getUserList();
+        } else {
+            taskService.error(task,"Wrong type of returnedMessage");
+            return new ArrayList<>();
+        }
     }
 
     @Override
diff --git a/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/impl/TweetPersistorImpl.java b/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/impl/TweetPersistorImpl.java
@@ -9,7 +9,6 @@
 import org.woehlke.twitterwall.oodm.service.TaskService;
 import org.woehlke.twitterwall.scheduled.mq.endoint.TweetPersistor;
 import org.woehlke.twitterwall.scheduled.mq.msg.TweetFromTwitter;
-import org.woehlke.twitterwall.scheduled.mq.msg.TweetMsg;
 import org.woehlke.twitterwall.scheduled.service.persist.StoreOneTweetPerform;
 
 @Component("mqTweetPersistor")
diff --git a/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/impl/UserFinisherImpl.java b/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/impl/UserFinisherImpl.java
@@ -0,0 +1,22 @@
+package org.woehlke.twitterwall.scheduled.mq.endoint.impl;
+
+import org.springframework.messaging.Message;
+import org.springframework.stereotype.Component;
+import org.woehlke.twitterwall.oodm.entities.User;
+import org.woehlke.twitterwall.scheduled.mq.endoint.UserFinisher;
+import org.woehlke.twitterwall.scheduled.mq.msg.UserResultList;
+
+import java.util.List;
+
+@Component("mqUserFinisher")
+public class UserFinisherImpl implements UserFinisher {
+
+
+    @Override
+    public UserResultList finish(Message<List<User>> incomingMessageList) {
+        long taskId = 0L;
+        List<User> userList = incomingMessageList.getPayload();
+        UserResultList userResultList = new UserResultList(taskId,userList);
+        return userResultList;
+    }
+}
diff --git a/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/impl/UserPersistorImpl.java b/src/main/java/org/woehlke/twitterwall/scheduled/mq/endoint/impl/UserPersistorImpl.java
@@ -24,12 +24,12 @@ public UserPersistorImpl(TaskService taskService, StoreUserProcess storeUserProc
     }
 
     @Override
-    public User persistUser(Message<UserMessage> incomingUserMessage) {
+    public UserMessage persistUser(Message<UserMessage> incomingUserMessage) {
         UserMessage receivedMessage = incomingUserMessage.getPayload();
         long id = receivedMessage.getTwitterProfileMessage().getTaskMessage().getTaskId();
         Task task = taskService.findById(id);
         User user = storeUserProcess.storeUserProcess(receivedMessage.getUser(),task);
-        taskService.done(task);
-        return user;
+        UserMessage userMessage = new UserMessage(user,id);
+        return userMessage;
     }
 }
diff --git a/src/main/java/org/woehlke/twitterwall/scheduled/mq/helper/CreateTestDataForUsersReleaseStrategy.java b/src/main/java/org/woehlke/twitterwall/scheduled/mq/helper/CreateTestDataForUsersReleaseStrategy.java
@@ -0,0 +1,23 @@
+package org.woehlke.twitterwall.scheduled.mq.helper;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.integration.aggregator.ReleaseStrategy;
+import org.springframework.integration.store.MessageGroup;
+import org.springframework.stereotype.Component;
+import org.woehlke.twitterwall.conf.TwitterwallSchedulerProperties;
+
+@Component("mqCreateTestDataForUsersReleaseStrategy")
+public class CreateTestDataForUsersReleaseStrategy implements ReleaseStrategy {
+
+    private final TwitterwallSchedulerProperties twitterwallSchedulerProperties;
+
+    @Autowired
+    public CreateTestDataForUsersReleaseStrategy(TwitterwallSchedulerProperties twitterwallSchedulerProperties) {
+        this.twitterwallSchedulerProperties = twitterwallSchedulerProperties;
+    }
+
+    @Override
+    public boolean canRelease(MessageGroup group) {
+        return ! (group.size() < twitterwallSchedulerProperties.getFacade().getIdTwitterToFetchForUserControllerTest().size());
+    }
+}
diff --git a/src/main/java/org/woehlke/twitterwall/scheduled/mq/msg/TweetResultList.java b/src/main/java/org/woehlke/twitterwall/scheduled/mq/msg/TweetResultList.java
@@ -2,9 +2,10 @@
 
 import org.woehlke.twitterwall.oodm.entities.Tweet;
 
+import java.io.Serializable;
 import java.util.List;
 
-public class TweetResultList {
+public class TweetResultList implements Serializable {
 
     private final long taskId;
     private final List<Tweet> tweetList;
diff --git a/src/main/java/org/woehlke/twitterwall/scheduled/mq/msg/TwitterProfileMessage.java b/src/main/java/org/woehlke/twitterwall/scheduled/mq/msg/TwitterProfileMessage.java
@@ -10,6 +10,12 @@ public class TwitterProfileMessage implements Serializable {
     private final String screenName;
     private final TwitterProfile twitterProfile;
 
+    public TwitterProfileMessage(TaskMessage taskMessage, TwitterProfile twitterProfile){
+        this.taskMessage = taskMessage;
+        this.screenName = twitterProfile.getScreenName();
+        this.twitterProfile = twitterProfile;
+    }
+
     public TwitterProfileMessage(TaskMessage taskMessage, String screenName, TwitterProfile twitterProfile) {
         this.taskMessage = taskMessage;
         this.screenName = screenName;
diff --git a/src/main/java/org/woehlke/twitterwall/scheduled/mq/msg/UserMessage.java b/src/main/java/org/woehlke/twitterwall/scheduled/mq/msg/UserMessage.java
diff --git a/src/main/java/org/woehlke/twitterwall/scheduled/mq/msg/UserResultList.java b/src/main/java/org/woehlke/twitterwall/scheduled/mq/msg/UserResultList.java
diff --git a/src/main/java/org/woehlke/twitterwall/scheduled/service/backend/TwitterApiService.java b/src/main/java/org/woehlke/twitterwall/scheduled/service/backend/TwitterApiService.java
diff --git a/src/main/java/org/woehlke/twitterwall/scheduled/service/backend/impl/TwitterApiServiceImpl.java b/src/main/java/org/woehlke/twitterwall/scheduled/service/backend/impl/TwitterApiServiceImpl.java
diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml
diff --git a/src/main/resources/integration.xml b/src/main/resources/integration.xml

Original file line number	Diff line number	Diff line change
`@@ -6,5 +6,5 @@`
`6`	`6`
`7`	`7`	`public interface UserPersistor {`
`8`	`8`
`9`		`- User persistUser(Message<UserMessage> incomingUserMessage);`
	`9`	`+ UserMessage persistUser(Message<UserMessage> incomingUserMessage);`
`10`	`10`	`}`