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

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

		httpSource.ensureReady().postData("Hello").postData("World").postData("!");
		httpSource.ensureReady().postData("I").postData("am").postData("1");
		httpSource.ensureReady().postData("tu").postData("es").postData("fifth");

		String expected = "! I 1";
		expected += "am tu es";
		expected += "Hello World fifth";
		assertThat(fileSink, eventually(hasContentsThat(equalTo(expected))));

	}
Shell integration tests for various simple processor modules.

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