nelanka · February 13, 2015 15:18
diff --git a/rx.scala b/rx.scala
 Add rx things together and log stuff:
 (consumer.stream1 ++ consumer.stream2 ++ consumer.stream3).subscribe(event => logger.info("Received event " + event))

 Filter out messages:
 consumer.stream1.filter(_.isEnabled != true).subscribe(event => map.remove(event.id))

 It’s kinda handy that rx streams are basically infinite lists that you can apply all scala magical collection incantations to.
	Add rx things together and log stuff:
	(consumer.stream1 ++ consumer.stream2 ++ consumer.stream3).subscribe(event => logger.info("Received event " + event))

	Filter out messages:
	consumer.stream1.filter(_.isEnabled != true).subscribe(event => map.remove(event.id))

	It’s kinda handy that rx streams are basically infinite lists that you can apply all scala magical collection incantations to.