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

			private AtomicInteger counter = new AtomicInteger(0);

			@Override
			public void test(Message<?> message) throws MessagingException {
				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) {
					super.messageHandled = true;
				}
				return super.messageHandled;
			}

		};
		StreamTestSupport.getSinkInputChannel("textLine").subscribe(test);
		dropFile("foo1.txt", 10);
		test.waitForCompletion(3000);
		undeployStream("textLine");
		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