EsperPattern implements EsperStatementFactory { EsperPattern(String pattern) { this.pattern = pattern; } EsperPattern(String pattern); @Override EPStatement createStatement(EPAdministrator administrator);  }
@Test public void testEsperPattern() throws Exception { StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment(); executionEnvironment.setParallelism(1); List<ComplexEvent> expectedValues = Lists.newArrayList(); ComplexEvent complexEvent = new ComplexEvent(Event.start(), Event.end()); expectedValues.add(complexEvent); List<Event> events = Arrays.asList(complexEvent.getStartEvent(), complexEvent.getEndEvent()); DataStream<Event> dataStream = executionEnvironment.fromCollection(events); EsperStream<Event> esperStream = Esper.pattern(dataStream, "every (A=Event(type='start') -> B=Event(type='end'))"); DataStream<ComplexEvent> complexEventDataStream = esperStream.select(new EsperSelectFunction<ComplexEvent>() { @Override public ComplexEvent select(EventBean eventBean) throws Exception { return new ComplexEvent((Event) eventBean.get("A"), (Event) eventBean.get("B")); } }); complexEventDataStream.addSink(new SinkFunction<ComplexEvent>() { @Override public void invoke(ComplexEvent value) throws Exception { System.err.println(value); resultingEvents.add(value); } }); executionEnvironment.execute("test-2"); assertThat(resultingEvents, is(expectedValues)); }