package de.hdm.wim.cep.source;

import de.hdm.wim.classes.Chat;
import de.hdm.wim.classes.Participant;
import de.hdm.wim.events.MessageEvent;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.time.LocalDateTime;
import java.util.Arrays;
import java.util.List;

/**
 * Created by Ben on 15.01.2017.
 */
public class TokenSource implements SourceFunction<String> {

    private volatile boolean isRunning = true;

    @Override
    public void run(SourceContext<String> sourceContext) throws Exception {

        // set up 2 participants
        final Participant participant1 = new Participant("Mike", "Turbo");
        final Participant participant2 = new Participant("Sheila", "Pure");

        // define tokens
        List<String> tokens0 = Arrays.asList("Hey");
        List<String> tokens1 = Arrays.asList("Hello");
        List<String> tokens2 = Arrays.asList("lets", "talk", "about", "current",  "activities", "concerning", "HighNet", "project");
        List<String> tokens3 = Arrays.asList("ok", "shall", "we", "look", "at", "tasks", "leading","to", "milestone", "ahead" );
        List<String> tokens4 = Arrays.asList("sure", "we", "working", "on", "network","issues", "for", "diagnosis", "module", "item 3", "task", "list", "blabla", "we", "will", "come up", "with", "something", "viable", "shortly");
        List<String> tokens5 = Arrays.asList("that", "sounds", "great", "what", "about", "expenses", "do", "you", "think", "you", "will", "be", "able", "to", "stay", "whithin", "limits", "we", "aggreed", "upon", "last week");
        List<String> tokens6 = Arrays.asList("that", "should", "be", "no", "problem", "i", "will", "leave", "detailed", "report", "on", "google", "drive");
        List<String> tokens7 = Arrays.asList("ok", "thanks", "lets", "make", "appointment", "for", "our", "next", "meeting");
        List<String> tokens8 = Arrays.asList("let", "me", "check", "my", "calendar", "how", "about", "next", "thursday", "at", "16", "hours", "your", "time");
        List<String> tokens9 = Arrays.asList("perfect", "see", "you", "then", "bye");
        List<String> tokens10 = Arrays.asList("have", "a", "nice", "day", "bye");
        List<String> tokens11 = Arrays.asList("expenses", "previous", "monday");
        List<String> tokens12 = Arrays.asList("Let", "us", "make", "an", "appointment", "for", "next", "Monday", "at", "09", "am");
        List<String> tokens13 = Arrays.asList("Let's", "plan", "a", "meeting", "at", "16:00", "on", "next", "monday");
        List<String> tokens14 = Arrays.asList("expenses", "blaaa", "at", "1pm");

        // set up chat
        Chat chat = new Chat();
        chat.addParticipant(participant1);
        chat.addParticipant(participant2);

        chat.addMessage(new MessageEvent(tokens0, participant1, LocalDateTime.now(),0));
        chat.addMessage(new MessageEvent(tokens1, participant2, LocalDateTime.now(),1));
        chat.addMessage(new MessageEvent(tokens2, participant1, LocalDateTime.now(),2));
        chat.addMessage(new MessageEvent(tokens3, participant2, LocalDateTime.now(),3));
        chat.addMessage(new MessageEvent(tokens4, participant1, LocalDateTime.now(),4));
        chat.addMessage(new MessageEvent(tokens5, participant2, LocalDateTime.now(),5));
        chat.addMessage(new MessageEvent(tokens6, participant1, LocalDateTime.now(),6));
        chat.addMessage(new MessageEvent(tokens7, participant2, LocalDateTime.now(),7));
        chat.addMessage(new MessageEvent(tokens8, participant1, LocalDateTime.now(),8));
        chat.addMessage(new MessageEvent(tokens9, participant2, LocalDateTime.now(),9));
        chat.addMessage(new MessageEvent(tokens10, participant1, LocalDateTime.now(),10));
        chat.addMessage(new MessageEvent(tokens11, participant2, LocalDateTime.now(),11));
        chat.addMessage(new MessageEvent(tokens12, participant1, LocalDateTime.now(),12));
        chat.addMessage(new MessageEvent(tokens13, participant2, LocalDateTime.now(),13));
        chat.addMessage(new MessageEvent(tokens14, participant1, LocalDateTime.now(),15));

        int messageIndex = 0;

        // go through all the messages in chat
        while (isRunning && messageIndex < chat.messages.size()) {

            MessageEvent chatMessage = chat.messages.get(messageIndex);
            int tokenCount = chatMessage.getTokens().size();

            System.out.print("tokenCount: " + tokenCount);

            // add each token from message to sourceContext
            for(int tokenIndex = 1; tokenIndex < tokenCount; tokenIndex++){
                sourceContext.collect(chatMessage.getTokens().get(tokenIndex));
            }

            messageIndex++;
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}

@@ -114,16 +114,20 @@
 hdm.wim.
+cep.
 events.M
@@ -135,24 +135,65 @@
 ssageEvent;%0A
+import de.hdm.wim.cep.events.TokenEvent;%0A
 import org.a
@@ -425,22 +425,26 @@
 unction%3C
-String
+TokenEvent
 %3E %7B%0A%0A   
@@ -536,22 +536,26 @@
 Context%3C
-String
+TokenEvent
 %3E source
@@ -4124,17 +4124,17 @@
 .now(),1
-5
+4
 ));%0A%0A   
@@ -4414,68 +4414,8 @@
 );%0A%0A
-            System.out.print(%22tokenCount: %22 + tokenCount);%0A%0A
     
@@ -4503,17 +4503,17 @@
 Index = 
-1
+0
 ; tokenI
@@ -4546,16 +4546,17 @@
 dex++)%7B%0A
+%0A
         
@@ -4567,69 +4567,329 @@
     
-sourceContext.collect(chatMessage.getTokens().get(tokenIndex)
+TokenEvent tknEvt =%0A                        new TokenEvent(%0A                                chatMessage.getMessageId(),%0A                                chatMessage.getTokens().get(tokenIndex),%0A                                chatMessage.getSender()%0A                        );%0A%0A                sourceContext.collect(tknEvt
 );%0A 
@@ -4897,25 +4897,24 @@
           %7D%0A
-%0A
             

update class to changes in TokenEvent constructor