source: class EsperPatternTest  implements EsperStatementFactory {
  
      private static List<ComplexEvent> resultingEvents;
  
      @Test
      public void testEsperPattern() throws Exception {
  
target:         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));
      }
  
  }