<commit_before>package org.woehlke.twitterwall.scheduled.mq.endoint.impl;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.stereotype.Component;
import org.woehlke.twitterwall.oodm.entities.Task;
import org.woehlke.twitterwall.oodm.entities.Tweet;
import org.woehlke.twitterwall.oodm.entities.User;
import org.woehlke.twitterwall.oodm.entities.parts.CountedEntities;
import org.woehlke.twitterwall.oodm.entities.parts.TaskType;
import org.woehlke.twitterwall.oodm.service.TaskService;
import org.woehlke.twitterwall.scheduled.mq.channel.SenderType;
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 org.woehlke.twitterwall.scheduled.service.persist.CountedEntitiesService;

import java.util.ArrayList;
import java.util.List;

@Component("mqStartTask")
public class StartTaskImpl implements StartTask {

    @Override
    public void fetchTweetsFromTwitterSearch() {
        TaskType taskType = TaskType.FETCH_TWEETS_FROM_TWITTER_SEARCH;
        sendAndReceiveTweet(taskType);
    }

    @Override
    public void updateTweets() {
        TaskType taskType = TaskType.UPDATE_TWEETS;
        sendAndReceiveTweet(taskType);
    }

    @Override
    public void updateUserProfiles() {
        TaskType taskType = TaskType.UPDATE_USER_PROFILES;
        sendAndReceiveUser(taskType);
    }

    @Override
    public void updateUserProfilesFromMentions() {
        TaskType taskType = TaskType.UPDATE_USER_PROFILES_FROM_MENTIONS;
        sendAndReceiveUser(taskType);
    }

    @Override
    public void fetchUsersFromDefinedUserList() {
        TaskType taskType = TaskType.FETCH_USERS_FROM_DEFINED_USER_LIST;
        sendAndReceiveUser(taskType);
    }

    @Override
    public User createImprintUser() {
        CountedEntities countedEntities = countedEntitiesService.countAll();
        TaskType taskType = TaskType.CONTROLLER_CREATE_IMPRINT_USER;
        String logMsg = "Start task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER;
        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();
        log.info(logMsg);
        Message<?> returnedMessage = mqTemplate.sendAndReceive(startTaskChannel, mqMessage);
        Object o = returnedMessage.getPayload();
        countedEntities = countedEntitiesService.countAll();
        if( o instanceof UserMessage){
            UserMessage msg = (UserMessage) o;
            long taskId = msg.getTaskId();
            task = taskService.findById(taskId);
            logMsg = "Sucessfully finished task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER;
            taskService.done(logMsg, task, countedEntities);
            log.info(logMsg);
            return msg.getUser();
        } else {
            logMsg = "Finished with Error: task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER+": Wrong type of returnedMessage";
            taskService.finalError(task,logMsg,countedEntities);
            log.error(logMsg);
            return null;
        }
    }

    @Override
    public List<User> createTestDataForUser() {
        CountedEntities countedEntities = countedEntitiesService.countAll();
        TaskType taskType = TaskType.CONTROLLER_GET_TESTDATA_USER;
        String logMsg = "Start task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER;
        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();
        log.info(logMsg);
        Message<?> returnedMessage = mqTemplate.sendAndReceive(startTaskChannel, mqMessage);
        Object o = returnedMessage.getPayload();
        countedEntities = countedEntitiesService.countAll();
        if (o instanceof UserResultList) {
            UserResultList result = (UserResultList) o;
            long taskId = result.getTaskId();
            task = taskService.findById(taskId);
            logMsg = "Sucessfully finished task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER;
            taskService.done(logMsg, task, countedEntities);
            log.info(logMsg);
            return result.getUserList();
        } else {
            logMsg = "Finished with Error: task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER+": Wrong type of returnedMessage";
            taskService.finalError(task,logMsg,countedEntities);
            log.error(logMsg);
            return new ArrayList<>();
        }
    }

    @Override
    public List<Tweet> createTestDataForTweets() {
        CountedEntities countedEntities = countedEntitiesService.countAll();
        TaskType taskType = TaskType.CONTROLLER_GET_TESTDATA_TWEETS;
        String logMsg = "Start task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER;
        Task task = taskService.create("Start via MQ by Scheduler ", 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();
        log.info(logMsg);
        Message<?> returnedMessage = mqTemplate.sendAndReceive(startTaskChannel, mqMessage);
        Object o = returnedMessage.getPayload();
        countedEntities = countedEntitiesService.countAll();
        if (o instanceof TweetResultList) {
            TweetResultList result = (TweetResultList) o;
            long taskId = result.getTaskId();
            task = taskService.findById(taskId);
            logMsg = "Sucessfully finished task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER;
            taskService.done(logMsg, task, countedEntities);
            log.info(logMsg);
            return result.getTweetList();
        } else {
            logMsg = "Finished with Error: task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER+": Wrong type of returnedMessage";
            taskService.finalError(task,logMsg,countedEntities);
            log.error(logMsg);
            return new ArrayList<>();
        }
    }

    private void sendAndReceiveTweet(TaskType taskType){
        String logMsg = "Start task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER;
        CountedEntities countedEntities = countedEntitiesService.countAll();
        Task task = taskService.create("Start via MQ by Scheduler ", 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();
        log.info(logMsg);
        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);
            logMsg = "Sucessfully finished task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER;
            taskService.done(logMsg,task,countedEntities);
        } else {
            logMsg = "Finished with Error: task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER+": Wrong type of returnedMessage";
            taskService.finalError(task,logMsg,countedEntities);
            log.error(logMsg);
        }
    }

    private void sendAndReceiveUser(TaskType taskType){
        String logMsg = "Start task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER;
        CountedEntities countedEntities = countedEntitiesService.countAll();
        Task task = taskService.create(logMsg, 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();
        log.info(logMsg);
        Message<?> returnedMessage = mqTemplate.sendAndReceive(startTaskChannel, mqMessage);
        Object o = returnedMessage.getPayload();
        countedEntities = countedEntitiesService.countAll();
        if( o instanceof UserMessage){
            UserMessage msg = (UserMessage) o;
            long taskId = msg.getTaskId();
            task = taskService.findById(taskId);
            logMsg = "Sucessfully finished task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER;
            taskService.done(logMsg,task,countedEntities);
            log.info(logMsg);
        } else {
            logMsg = "Finished with Error: task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER+": Wrong type of returnedMessage";
            taskService.finalError(task,logMsg,countedEntities);
            log.error(logMsg);
        }
    }

    private final MessageChannel startTaskChannel;

    private final TaskService taskService;

    private final CountedEntitiesService countedEntitiesService;

    private static final Logger log = LoggerFactory.getLogger(StartTaskImpl.class);

    @Autowired
    public StartTaskImpl(MessageChannel startTaskChannel,TaskService taskService, CountedEntitiesService countedEntitiesService) {
        this.startTaskChannel = startTaskChannel;<commit_msg>working on #141 #142 #143<commit_after>package org.woehlke.twitterwall.scheduled.mq.endoint.impl;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.stereotype.Component;
import org.woehlke.twitterwall.oodm.entities.Task;
import org.woehlke.twitterwall.oodm.entities.Tweet;
import org.woehlke.twitterwall.oodm.entities.User;
import org.woehlke.twitterwall.oodm.entities.parts.CountedEntities;
import org.woehlke.twitterwall.oodm.entities.parts.TaskType;
import org.woehlke.twitterwall.oodm.service.TaskService;
import org.woehlke.twitterwall.scheduled.mq.channel.SenderType;
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 org.woehlke.twitterwall.scheduled.service.persist.CountedEntitiesService;

import java.util.ArrayList;
import java.util.List;

@Component("mqStartTask")
public class StartTaskImpl implements StartTask {

    @Override
    public void fetchTweetsFromTwitterSearch() {
        TaskType taskType = TaskType.FETCH_TWEETS_FROM_TWITTER_SEARCH;
        sendAndReceiveTweet(taskType);
    }

    @Override
    public void updateTweets() {
        TaskType taskType = TaskType.UPDATE_TWEETS;
        sendAndReceiveTweet(taskType);
    }

    @Override
    public void updateUserProfiles() {
        TaskType taskType = TaskType.UPDATE_USER_PROFILES;
        sendAndReceiveUser(taskType);
    }

    @Override
    public void updateUserProfilesFromMentions() {
        TaskType taskType = TaskType.UPDATE_USER_PROFILES_FROM_MENTIONS;
        sendAndReceiveUser(taskType);
    }

    @Override
    public void fetchUsersFromDefinedUserList() {
        TaskType taskType = TaskType.FETCH_USERS_FROM_DEFINED_USER_LIST;
        sendAndReceiveUser(taskType);
    }

    @Override
    public User createImprintUser() {
        TaskType taskType = TaskType.CONTROLLER_CREATE_IMPRINT_USER;
        String logMsg = "Start task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER;
        log.info(logMsg);
        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 UserMessage){
            UserMessage msg = (UserMessage) o;
            long taskId = msg.getTaskId();
            task = taskService.findById(taskId);
            logMsg = "Sucessfully finished task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER;
            taskService.done(logMsg, task, countedEntities);
            log.info(logMsg);
            return msg.getUser();
        } else {
            logMsg = "Finished with Error: task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER+": Wrong type of returnedMessage";
            taskService.finalError(task,logMsg,countedEntities);
            log.error(logMsg);
            return null;
        }
    }

    @Override
    public List<User> createTestDataForUser() {
        TaskType taskType = TaskType.CONTROLLER_GET_TESTDATA_USER;
        String logMsg = "Start task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER;
        log.info(logMsg);
        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 UserResultList) {
            UserResultList result = (UserResultList) o;
            long taskId = result.getTaskId();
            task = taskService.findById(taskId);
            logMsg = "Sucessfully finished task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER;
            taskService.done(logMsg, task, countedEntities);
            log.info(logMsg);
            return result.getUserList();
        } else {
            logMsg = "Finished with Error: task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER+": Wrong type of returnedMessage";
            taskService.finalError(task,logMsg,countedEntities);
            log.error(logMsg);
            return new ArrayList<>();
        }
    }

    @Override
    public List<Tweet> createTestDataForTweets() {
        TaskType taskType = TaskType.CONTROLLER_GET_TESTDATA_TWEETS;
        String logMsg = "Start task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER;
        log.info(logMsg);
        CountedEntities countedEntities = countedEntitiesService.countAll();
        Task task = taskService.create("Start via MQ by Scheduler ", 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 result = (TweetResultList) o;
            long taskId = result.getTaskId();
            task = taskService.findById(taskId);
            logMsg = "Sucessfully finished task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER;
            taskService.done(logMsg, task, countedEntities);
            log.info(logMsg);
            return result.getTweetList();
        } else {
            logMsg = "Finished with Error: task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER+": Wrong type of returnedMessage";
            taskService.finalError(task,logMsg,countedEntities);
            log.error(logMsg);
            return new ArrayList<>();
        }
    }

    private void sendAndReceiveTweet(TaskType taskType){
        String logMsg = "Start task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER;
        log.info(logMsg);
        CountedEntities countedEntities = countedEntitiesService.countAll();
        Task task = taskService.create("Start via MQ by Scheduler ", 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);
            logMsg = "Sucessfully finished task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER;
            taskService.done(logMsg,task,countedEntities);
        } else {
            logMsg = "Finished with Error: task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER+": Wrong type of returnedMessage";
            taskService.finalError(task,logMsg,countedEntities);
            log.error(logMsg);
        }
    }

    private void sendAndReceiveUser(TaskType taskType){
        String logMsg = "Start task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER;
        log.info(logMsg);
        CountedEntities countedEntities = countedEntitiesService.countAll();
        Task task = taskService.create(logMsg, 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 UserMessage){
            UserMessage msg = (UserMessage) o;
            long taskId = msg.getTaskId();
            task = taskService.findById(taskId);
            logMsg = "Sucessfully finished task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER;
            taskService.done(logMsg,task,countedEntities);
            log.info(logMsg);
        } else {
            logMsg = "Finished with Error: task "+taskType+"via MQ by "+ SenderType.SEND_AND_WAIT_FOR_RESULT_SENDER+": Wrong type of returnedMessage";
            taskService.finalError(task,logMsg,countedEntities);
            log.error(logMsg);
        }
    }

    private final MessageChannel startTaskChannel;

    private final TaskService taskService;

    private final CountedEntitiesService countedEntitiesService;

    private static final Logger log = LoggerFactory.getLogger(StartTaskImpl.class);

    @Autowired
    public StartTaskImpl(MessageChannel startTaskChannel,TaskService taskService, CountedEntitiesService countedEntitiesService) {
        this.startTaskChannel = startTaskChannel;
{'fraction_non_alphanumeric': 0.0957688609144156, 'fraction_numerical': 0.0005682314887665006, 'mean_word_length': 3.2740519334952363, 'pattern_counts': {'":': 10, '<': 31, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5329117', 'n_tokens_mistral': 6468, 'n_tokens_neox': 6488, 'n_words': 1209}