root package

This is the API documentation for the Monifu library.

Overview

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 Future for 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

Example Usage

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.