Gitter to Slack relay

gistfile1.java

HttpClient<Buffer, Buffer> client = NetStreams.httpClient();

final String postUri = "https://hooks.slack.com/services/xxxxx";
final NioEventLoopGroup group = new NioEventLoopGroup(12,
new NamedDaemonThreadFactory("reactor-tcp-io"));



client.get("https://stream.gitter.im/v1/rooms/xxxxxx/chatMessages", ch -> {
	ch
			.header("Authorization", "Bearer xxxxxx")
			.header("Accept", "application/json");
	return Streams.never();
})
.flatMap(p -> p
		.filter(m -> m.remaining() > 2)
		.decode(new JsonCodec<>(Map.class))
		.log("after-consume")
		.window(10, 1, TimeUnit.SECONDS)
		.flatMap(msgs ->
			NetStreams.httpClient(spec ->
				spec.options(new NettyClientSocketOptions().eventLoopGroup(group))
			)
			.post(postUri, ch ->
				ch
				.header("content-type", "application/x-www-form-urlencoded")
				.writeWith(
					msgs.map(msg -> "*"+((Map)msg.get("fromUser")).get("displayName")+"*"+" _at "+msg.get("sent")+"_ : "+msg.get("text").toString().replaceAll("\n","\\\\n"))
								.reduce("", (prev, next) -> prev.isEmpty() ? next : (prev + "\\\\n" + next))
								.log("before-form")
								.map(text -> {
									try {
										return "payload=" + URLEncoder
												.encode("{ \"text\" : \"" + text + "\"" + " }", "UTF-8");
									} catch (UnsupportedEncodingException uee) {
										uee.printStackTrace();
										return null;
									}
								})
								.log("new-message")
								.map(new StringCodec())
					)
			)
			.stream()
			.flatMap(Streams::after) // complete after all replies consumed and closed
		)
		.log("after-windows")
		.after())
.awaitSuccess();