@Test
	public void testLinesModeWithMarkersTrue() throws IOException {
		deployStream(
				"textLineWithEmitMarkersTrue",
				"file --mode=lines --withMarkers=true --dir=" + sourceDirName + " --fixedDelay=0 | sink");
		MessageTest test = new MessageTest() {

			private AtomicInteger counter = new AtomicInteger(0);

			private FileMarker startFileMarker = null;

			private FileMarker endFileMarker = null;

			@Override
			public void test(Message<?> message) throws MessagingException {
				if (message.getPayload() instanceof FileMarker) {
					final FileMarker fileMarker = (FileMarker) message.getPayload();
					if (FileMarker.Mark.START.equals(fileMarker.getMark())) {
						startFileMarker = fileMarker;
					}
					else if (FileMarker.Mark.END.equals(fileMarker.getMark())) {
						endFileMarker = fileMarker;
					}
				}
				else {
					assertTrue("Extected a String", message.getPayload() instanceof String);
					assertEquals("foo", message.getPayload());
					assertEquals("foo1.txt", message.getHeaders().get(FileHeaders.FILENAME, String.class));
					counter.incrementAndGet();
				}
			}

			@Override
			public boolean getMessageHandled() {
				if (counter.get() == 10 && startFileMarker != null && endFileMarker != null) {
					super.messageHandled = true;
				}
				return super.messageHandled;
			}

		};
		StreamTestSupport.getSinkInputChannel("textLineWithEmitMarkersTrue").subscribe(test);
		dropFile("foo1.txt", 10);
		test.waitForCompletion(3000);
		undeployStream("textLineWithEmitMarkersTrue");
		assertTrue(test.getMessageHandled());

	}
@author David Turanski
@author Gunnar Hillert
@author Gary Russell
spring-xd-dirt/src/test/java/org/springframework/xd/dirt/stream/FileSourceModuleTests.java