package de.hdm.wim;

import de.hdm.wim.source.EventSource;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;

/**
 * Created by Ben on 16.01.2017.
 */
public class CEPSpeechTokens {

        public static void main(String[] args) throws Exception {


            final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();

            final LocalStreamEnvironment env2 = LocalStreamEnvironment.createLocalEnvironment();




            EventSource evtSrc = new EventSource();


            DataStream<List<List<String>>> messageStream2 = env2
                    .addSource(evtSrc);


            DataStream<List<List<String>>> messageStream = env
                    .addSource(evtSrc);


            System.out.print("blubb");




            // print to stdout
            messageStream.print();
            messageStream2.print();

            env.execute("CEP monitoring job");

            env2.execute("CEP monitoring job2");

        }
}
@@ -36,87 +36,154 @@
 wim.
-source.EventSource;%0Aimport org.apache.flink.streaming.api.datastream.DataStream
+events.MessageEvent;%0Aimport de.hdm.wim.patterns.SenderPattern;%0Aimport de.hdm.wim.patterns.TestPattern;%0Aimport de.hdm.wim.source.MessageEventSource
 ;%0Aim
@@ -222,42 +222,29 @@
 api.
-environment.LocalStreamEnvironment
+datastream.DataStream
 ;%0Aim
@@ -324,32 +324,8 @@
 t;%0A%0A
-import java.util.List;%0A%0A
 /**%0A
@@ -460,17 +460,16 @@
 tion %7B%0A%0A
-%0A
         
@@ -560,25 +560,24 @@
 ironment();%0A
-%0A
             
@@ -580,95 +580,84 @@
     
-final LocalStreamEnvironment env2 = LocalStreamEnvironment.createLocalEnvironment();%0A%0A%0A
+env.setParallelism(1); // set Parallelism to 1 Task, to get chat order right
 %0A%0A  
@@ -666,16 +666,23 @@
         
+Message
 EventSou
@@ -698,16 +698,23 @@
 c = new 
+Message
 EventSou
@@ -717,25 +717,24 @@
 tSource();%0A%0A
-%0A
             
@@ -737,37 +737,98 @@
     
-DataStream%3CList%3CList%3CString%3E%3E
+// create a DataStream of MessageEvent from our source%0A            DataStream%3CMessageEvent
 %3E me
@@ -842,16 +842,14 @@
 ream
-2
  = env
-2
 %0A   
@@ -903,293 +903,1682 @@
     
-DataStream%3CList%3CList%3CString%3E%3E%3E messageStream = env%0A                    .addSourc
+TestPattern testPattern = new TestPattern();%0A            testPattern.run(env,messageStream);%0A%0A            SenderPattern senderPattern = new SenderPattern();%0A            senderPattern.run(env,messageStream);%0A%0A%0A            /*%0A            // Warning pattern: Two consecutive temperature events whose temperature is higher than the given threshold%0A            // appearing within a time interval of 10 seconds%0A            Pattern%3CMessageEvent, ?%3E projectPattern = Pattern.%3CMessageEvent%3Ebegin(%22first%22)%0A                    .subtype( ProjectEvent.class )%0A                    .wher
 e(
+ 
 evt
-Src);%0A%0A%0A            System.out.print(%22blubb%22);%0A%0A%0A%0A%0A            // print to stdout%0A            messageStream.print();%0A            messageStream2.print();%0A%0A            env.execute(%22CEP monitoring job%22);
+ -%3E evt.getTokens().contains(%22concerning%22) );%0A                            //.stream()%0A                            //.anyMatch( token -%3E token.equals(%22concerning%22) )%0A                    //);%0A//                            .anyMatch( token -%3E (token.equals(%22concerning%22) %7C%7C token.equals(%22regarding%22) ) && token.equals(%22project%22) ));%0A%0A            // Create a pattern stream from our project pattern%0A            PatternStream%3CMessageEvent%3E projectPatternStream = CEP.pattern(%0A                    messageStream.keyBy(%22_messageId%22),%0A                    projectPattern);%0A%0A            // Generate ProjectEvents for each matched project pattern%0A            DataStream%3CProjectEvent%3E projectStream = projectPatternStream.select(%0A                    (Map%3CString, MessageEvent%3E pattern) -%3E %7B%0A                        ProjectEvent projectEvent = (ProjectEvent) pattern.get(%22first%22);%0A%0A                        return projectEvent;%0A                    %7D%0A            );%0A%0A            // print to stdout%0A            //projectStream.print();%0A%0A%0A*/%0A            // print message stream%0A            messageStream.print();%0A
 %0A%0A  
@@ -2590,17 +2590,16 @@
      env
-2
 .execute
@@ -2608,28 +2608,27 @@
 CEP 
-monitoring
+chat stream
  job
-2
 %22);%0A
-%0A
     

Clean + moved patterns to own folder