Emitter Monad

Emitter.scala

import Emitter._

import scala.annotation.tailrec
import scala.concurrent.Future._
import scala.concurrent.{ExecutionContext, Future}

object Emitter {
  type Event[A] = Future[Option[(A, Emitter[A])]]

  case object Zero extends Emitter[Nothing] {
    def flatMap[B](f: (Nothing) => Emitter[B])(implicit context: ExecutionContext) = Zero
    def map[B](f: (Nothing) => B)(implicit context: ExecutionContext): Emitter[Nothing] = Zero
    def emit(implicit context: ExecutionContext): Event[Nothing] = successful(None)
  }

  def pure[A](a: A): Emitter[A] = new Emitter[A] {
    def emit(implicit context: ExecutionContext): Event[A] = successful(Some(a, Zero))
    def map[B](f: A => B)(implicit context: ExecutionContext): Emitter[B] = pure[B](f(a))
    def flatMap[B](f: A => Emitter[B])(implicit context: ExecutionContext): Emitter[B] = f(a)

  }
  
  def promise[A](e: Event[A]): Emitter[A] = new Emitter[A] {
    def emit(implicit context: ExecutionContext): Event[A] = e
    def map[B](f: A => B)(implicit context: ExecutionContext): Emitter[B] = promise(e map( o => o map{ case (a, em) => f(a) -> (em map f) }))
    def flatMap[B](f: A => Emitter[B])(implicit context: ExecutionContext): Emitter[B] = promise(e flatMap {
        case Some((a, es)) => f(a).emit
        case None => successful(None)
      })
  }

  def apply[A](as: List[A])(implicit context: ExecutionContext): Emitter[A] = {
    @tailrec
    def go[B](l: List[B], em: Emitter[B])(implicit context: ExecutionContext): Emitter[B] = l match {
      case Nil => em
      case x :: xs => go(xs, em :+ x)
    }
    go(as, Zero)
  }
  def apply[A](as: A*)(implicit context: ExecutionContext): Emitter[A] = Emitter(as.toList)
}

trait Emitter[+A] { self =>
  def map[B](f: A => B)(implicit context: ExecutionContext): Emitter[B]
  def flatMap[B](f: A => Emitter[B])(implicit context: ExecutionContext): Emitter[B]
  def emit(implicit context: ExecutionContext): Event[A]
  def or[B >: A](em: Emitter[B])(implicit context: ExecutionContext): Event[B] = self.emit flatMap {
    case None => em.emit
    case o => successful(o map {
      case (a, nem) => a -> promise(nem or em)
    })
  }
  def :+:[B >: A](em: Emitter[B])(implicit context: ExecutionContext): Emitter[B] = promise(self or em)
  def :+[B >: A](b: B)(implicit context: ExecutionContext): Emitter[B] = self :+: pure(b)
}