@Test
	public void testRelaxedFollowedByStrictContiguity() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		DataStream<Event> input = env.fromElements(
				new Event(1, "pre-condition", 1.0),
				new Event(2, "start", 2.0),
				new Event(3, "middle", 3.0),
				new Event(4, "start", 4.0),
				new Event(5, "middle", 5.0),
				new Event(6, "end", 6.0),
				new Event(7, "end", 3.0)
		);

		Pattern<Event, ?> pattern = Pattern.<Event>begin("pre-condition")
				.where(new SimpleCondition<Event>() {
					@Override
					public boolean filter(Event value) throws Exception {
						return value.getName().equalsIgnoreCase("pre-condition");
					}
				}).followedBy("start")
				.where(new SimpleCondition<Event>() {
					@Override
					public boolean filter(Event value) throws Exception {
						return value.getName().equalsIgnoreCase("start");
					}
				}).next("middle")
				.where(new SimpleCondition<Event>() {
					@Override
					public boolean filter(Event value) throws Exception {
						return value.getName().equalsIgnoreCase("middle");
					}
				}).next("end")
				.where(new SimpleCondition<Event>() {
					@Override
					public boolean filter(Event value) throws Exception {
						return value.getName().equalsIgnoreCase("end");
					}
				});

		DataStream<String> result = CEP.pattern(input, pattern).select(new PatternSelectFunction<Event, String>() {

			@Override
			public String select(Map<String, List<Event>> pattern) {
				StringBuilder builder = new StringBuilder();

				builder.append(pattern.get("pre-condition").get(0).getId()).append(",")
						.append(pattern.get("start").get(0).getId()).append(",")
						.append(pattern.get("middle").get(0).getId()).append(",")
						.append(pattern.get("end").get(0).getId());

				return builder.toString();
			}
		});

		result.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);

		// expected sequence of matching event ids
		expected = "1,4,5,6";

		env.execute();
	}


	@Test
	public void testStrictFollowedByRelaxedContiguity() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		DataStream<Event> input = env.fromElements(
				new Event(1, "pre-condition", 1.0),
				new Event(2, "start", 2.0),
				new Event(3, "middle", 3.0),
				new Event(4, "start", 4.0),
				new Event(5, "middle", 5.0),
				new Event(6, "end", 6.0),
				new Event(7, "start", 3.0),
				new Event(8, "post-condition", 1.0)
		);

		Pattern<Event, ?> pattern = Pattern.<Event>begin("start")
				.where(new SimpleCondition<Event>() {
					@Override
					public boolean filter(Event value) throws Exception {
						return value.getName().equalsIgnoreCase("start");
					}
				}).next("middle")
				.where(new SimpleCondition<Event>() {
					@Override
					public boolean filter(Event value) throws Exception {
						return value.getName().equalsIgnoreCase("middle");
					}
				}).next("end")
				.where(new SimpleCondition<Event>() {
					@Override
					public boolean filter(Event value) throws Exception {
						return value.getName().equalsIgnoreCase("end");
					}
				}).followedBy("post-condition")
				.where(new SimpleCondition<Event>() {
					@Override
					public boolean filter(Event value) throws Exception {
						return value.getName().equalsIgnoreCase("post-condition");
					}
				});

		DataStream<String> result = CEP.pattern(input, pattern).select(new PatternSelectFunction<Event, String>() {

			@Override
			public String select(Map<String, List<Event>> pattern) {
				StringBuilder builder = new StringBuilder();

				builder	.append(pattern.get("start").get(0).getId()).append(",")
						.append(pattern.get("middle").get(0).getId()).append(",")
						.append(pattern.get("end").get(0).getId()).append(",")
						.append(pattern.get("post-condition").get(0).getId());

				return builder.toString();
			}
		});

		result.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);

		// expected sequence of matching event ids
		expected = "4,5,6,8";

		env.execute();
	}