Observable.scala

package org.photon.common

import scala.collection.mutable
import java.util.concurrent.locks.ReentrantReadWriteLock
import java.util.concurrent.{Executors, Executor}

object Observable {
  type Listener = Any => Any
}

trait Observable {
  import JavaConversion.RichReentrantReadWriteLock
  import Observable.Listener

  implicit val executor: Executor

  private val subs = mutable.Map.empty[Symbol, mutable.ArrayBuffer[Listener]]
  private val l = new ReentrantReadWriteLock

  def subscribe(s: Symbol, fn: Listener): Unit = l.write {
    subs.getOrElseUpdate(s, mutable.ArrayBuffer.empty) += fn
  }

  def subscribe(s: Symbol)(fn: Listener): Unit = subscribe(s, fn)
  
  def unsubscribe(s: Symbol, fn: Listener): Unit = l.write {
    subs.get(s) foreach { _ -= fn }
  }

  def emit(s: Symbol, args: Any = ()): Unit = Async {
    l.read {
      subs.get(s) foreach { _.foreach { fn => fn(args) } }
    }
  }
}

Test.scala

object Test extends Observable {
  implicit val executor = Executors.newSingleThreadExecutor()

  subscribe('lol) {
    case i: Int => println(i * 2)
    case s: String => println(s"lol $s")
    case Unit => println("lol!")
  }

  emit('lol, 21)
  emit('lol)
  emit('lol, "Jon Snow")
}