@Test
	public void testAggregatorEarlyRelease() throws IOException {
		HttpSource httpSource = newHttpSource();
		FileSink fileSink = newFileSink().binary(true);

		int timeout = 1000;

		stream().create(
				generateStreamName(),
				"%s | aggregator --count=100 --timeout=%d --aggregation=T(org.springframework.util.StringUtils).collectionToDelimitedString(#this.![payload],' ') | %s",
				httpSource, timeout, fileSink);

		httpSource.ensureReady().postData("Hello").postData("World").postData("!");

		// Although the reaper and the task scheduler are both configured with 'timeout',
		// this does not take into account the amount of time it may take to deploy the
		// stream in a distributed environment; therefore this assert will use the default
		// timeouts configured in the eventually/hasContentsThat methods.
		assertThat(fileSink, eventually(hasContentsThat(equalTo("Hello World !"))));

	}
Shell integration tests for various simple processor modules.

@author Eric Bottard
spring-xd-shell/src/test/java/org/springframework/xd/shell/command/AbstractProcessorsTests.java