Idiomatic Reactive Extensions for Scala and Scala.js.
Monifu is a high-performance Scala / Scala.js library for composing asynchronous and event-based programs using observable sequences that are exposed as asynchronous streams, expanding on the observer pattern, strongly inspired by Reactive Extensions (Rx), but designed from the ground up for back-pressure and made to cleanly interact with Scala's standard library and compatible out-of-the-box with the Reactive Streams protocol.
Highlights:
- zero dependencies
- clean and user-friendly API, with the observer interface using
Futurefor back-pressure purposes - Observable operators exposed in a way that's idiomatic to Scala
- compatible with for-comprehensions
- compatible with Scalaz
- designed to be completely asynchronous - Rx operators that are
blocking or that are not compatible with back-pressure semantics
are not going to be supported - does not depend on any particular mechanism for asynchronous execution and can be made to work with threads, actors, event loops, or whatnot, running perfectly both on top of the JVM or in Node.js or the browser
- really good test coverage as a project policy
See monifu-sample for a project exemplifying Monifu used both on the server and on the client.
Currently compiled for Scala 2.10.x, 2.11.x and Scala.js 0.6.x. Monifu's compatibility extends to the latest 2 major Scala versions and to the latest major Scala.js version. In other words support for Scala 2.10.x will be kept for as long as possible, but you should not rely on continued support long after Scala 2.12 is out.
The packages are published on Maven Central.
- Current stable release is:
1.0-RC3
libraryDependencies += "org.monifu" %% "monifu" % "1.0-RC3"libraryDependencies += "org.monifu" %%% "monifu" % "1.0-RC3"In order for subscriptions to work, we need an implicit
Scheduler imported in our
context. A Scheduler inherits from Scala's own ExecutionContext
and any ExecutionContext can be quickly converted into a Scheduler.
And then you're off ...
// scala.concurrent.ExecutionContext.Implicits.global
// is being used under the hood
import monifu.concurrent.Implicits.globalScheduler
// or we can simply convert our own execution context
// import play.api.libs.concurrent.Execution.Implicits.defaultContext
// implicit val scheduler = Scheduler(defaultContext)
import concurrent.duration._
import monifu.reactive._
val subscription = Observable.intervalAtFixedRate(1.second)
.take(10)
.subscription(x => println(x))We can then try out more complex things:
import monifu.concurrent.Implicits.globalScheduler
import play.api.libs.ws._
import monifu.reactive._
// emits an auto-incremented number, every second
Observable.interval(1.second)
// drops the items emitted over the first 5 secs
.dropByTimespan(5.seconds)
// takes the first 100 emitted events
.take(100)
// per second, makes requests and concatenates the results
.flatMap(x => WS.request(s"http://some.endpoint.com/request?tick=$x").get())
// filters only valid responses
.filter(response => response.status == 200)
// samples by 3 seconds, repeating previous results in case of nothing new
.sampleRepeated(3.seconds)
// processes response, selecting the body
.map(response => response.body)
// creates subscription, foreach response print it
.foreach(x => println(x))There's actually a lot more to Monifu.
The documentation is a work in progress.
All code in this repository is licensed under the Apache License, Version 2.0. See LICENCE.txt.
Development of Monifu has been initiated by Eloquentix engineers, with Monifu being introduced at E.ON Connecting Energies, powering the next generation energy grid solutions.