java8streamprocessing.java

public Mono<Void> gitterSlackRelay() {
		ObjectMapper mapper = new ObjectMapper();
		return create()
				.get(gitterStreamUrl, gitterStreamHandler())
				.flatMap(replies -> replies
						.receiveByteArray()
						.filter(b -> b.length > 2) // ignore gitter keep-alives (\r)
						.map(b -> {
							try {
								return mapper.readValue(b, Map.class);
							}
							catch (IOException e) {
								throw Exceptions.propagate(e);
							}
						}) // ObjectMapper.readValue(Map
						// .class)
						.window(10, 1_000) // microbatch 10 items or 1s worth into individual streams (for reduce ops)
						.flatMap(w -> postToSlack(
										w.map(m -> formatLink(m) + ": " + formatText(m))
										.reduce("", GitterSlackRelayApplication::appendLines))
								)
				)
				.then(); // only complete when all windows have completed AND gitter GET connection has closed
}