Facebook Flux implementation for scalajs-react

Dispatcher.scala

package flux

import utils.Loggable

import scala.concurrent.Future

class Dispatcher[Payload <: AnyRef] extends Loggable {

  type Callback = (Payload) => Unit

  private var lastId = 0
  private val prefix = "ID_"

  private var _callbacks = Map[String, Callback]()
  private var _isPending = Map[String, Boolean]()
  private var _isHandled = Map[String, Boolean]()
  private var _isDispatching = false
  private var _pendingPayload: Option[Payload] = None
  
  def isDispatching = _isDispatching
  def isPending(id:String) = _isPending.getOrElse(id, false)
  def isHandled(id:String) = _isHandled.getOrElse(id, false)
  
  def assert(condition: => Boolean, text: String, args: String*) {
    if (!condition) {
      throw new RuntimeException(text.format(args: _*))
    }
  }

  def register(callback: Callback) = {
    lastId += 1
    val id = prefix + lastId
    _callbacks += (id -> callback)
    
    logger.debug(s"register(): Registered a callback with id $id")
    
    id
  }

  def unregister(id: String) {
    assert(_callbacks.contains(id), "Dispatcher.unregister(...): `%s` does not map to a registered callback.", id)
    _callbacks -= id
  }

  def waitFor(ids: List[String]) {
    assert(_isDispatching, "Dispatcher.waitFor(...): Must be invoked while dispatching.")
    ids.foreach { id =>
      if (isPending(id)) {
        assert(isHandled(id), "Dispatcher.waitFor(...): Circular dependency detected while waiting for `%s`.", id)
      } else {
        assert(_callbacks.contains(id), "Dispatcher.waitFor(...): `%s` does not map to a registered callback.", id)
        invokeCallback(id)
      }
    }
  }

  def dispatch(payload: Payload): Future[Int] = {
    logger.debug(s"dispatch(): Dispatch requested for payload $payload")
    
    assert(!_isDispatching, "Dispatch.dispatch(...): Cannot dispatch in the middle of a dispatch.")
    startDispatching(payload)
    try {
      _callbacks.foreach {
        case (id, _) =>
          if (!isPending(id)) {
            invokeCallback(id)
          }
      }

      Future.successful(_callbacks.size)
    } finally {
      stopDispatching()
    }
  }

  private def invokeCallback(id: String) {
    logger.debug(s"invokeCallback(): Invoking callback with id '$id' with payload ${_pendingPayload.get}")
    
	  _isPending += (id -> true)
	  _callbacks(id)(_pendingPayload.get)
	  _isHandled += (id -> true)
  }

  private def startDispatching(payload: Payload) {
	  _isPending = Map[String, Boolean]()
	  _isHandled = Map[String, Boolean]()
	  _pendingPayload = Some(payload)
	  _isDispatching = true
  }

  private def stopDispatching() {
	  _pendingPayload = None
	  _isDispatching = false
  }

}

SomeAction.scala

sealed trait SomeAction {
    def dispatch(): Future[Int] = SomeDispatcher.dispatch(this)
  }
  
sealed trait ViewAction extends SomeAction

case class AddSomething(someThing: Thing) extends ViewAction

SomeDispatcher.scala

package dispatchers

import actions.ProjectEditorAction
import flux.Dispatcher

object SomeDispatcher extends Dispatcher[SomeAction]

Store.scala

package flux

import japgolly.scalajs.react.extra.{Broadcaster, Listenable}
import utils.Loggable

import scala.collection.mutable
import scala.concurrent.Future

abstract class StoreEvent[Item]()
case class StoreAdded[Item](item: Item)   extends StoreEvent[Item]
case class StoreRemoved[Item](item: Item) extends StoreEvent[Item]
case class StoreChanged[Item](item: Item) extends StoreEvent[Item]

trait Store[Id, Item] extends Loggable {
  val ALL_ITEMS: String = "ALL_ITEMS"
  
  private [Store] class StoreBroadcaster(key: String) extends Broadcaster[StoreEvent[Item]] {
    def broadcastAdd(stored: Item)    = broadcast(StoreAdded(stored))
    def broadcastRemove(stored: Item) = broadcast(StoreRemoved(stored))
    def broadcastChange(stored: Item) = broadcast(StoreChanged(stored))
    
    override def register(f: StoreEvent[Item] => Unit) = {
      logger.debug(s"register: Registered a listener")
      super.register(f)
    }
    
    protected def broadcast[Event <: StoreEvent[Item]](a: Event): Unit = {
      logger.debug(s"""broadcast(): Broadcasting $a to channel "$key" with ${listeners.size} listeners""")
      super.broadcast(a)
    }
  }
  
  protected val store = mutable.Map[Id, Item]()
  protected val broadcasters = mutable.Map[String, StoreBroadcaster]()

  protected def emitAdd(id: Id) = {
    broadCasterFor(id).broadcastAdd(store(id))
    broadCasterForAll.broadcastAdd(store(id))
  }
  
  protected def emitChange(id: Id) = {
    broadCasterFor(id).broadcastChange(store(id))
    broadCasterForAll.broadcastChange(store(id))
  }
  
  protected def emitRemove(id: Id, item: Item) = {
    broadCasterFor(id).broadcastRemove(item)
    broadCasterForAll.broadcastRemove(item)
  }
  
  def listenableFor(id: Id): Listenable[StoreEvent[Item]] =
    broadCasterFor(id)
  
  def listenableForAll: Listenable[StoreEvent[Item]] =
    broadCasterForAll
  
  protected def broadCasterFor(id: Id): StoreBroadcaster =
    broadcasters.getOrElseUpdate(id.toString, new StoreBroadcaster(id.toString))

  protected def broadCasterForAll: StoreBroadcaster =
    broadcasters.getOrElseUpdate(ALL_ITEMS, new StoreBroadcaster(ALL_ITEMS))
  
  def all: Iterable[Item] = store.values

  def by(id: Id): Option[Item] = store.get(id)
  
  def hydrateWith(items: Seq[Item], f: Item => Id): Future[Int] = {
    store.clear()
    for(item <- items) store(f(item)) = item
    Future.successful{store.size}
  }
  
  def hydrateWith(item: Item, f: Item => Id):                    Future[Int] = hydrateWith(Seq(item), f)
  def hydrateWith(maybeItems: Option[Seq[Item]], f: Item => Id): Future[Int] = hydrateWith(maybeItems getOrElse Seq(), f)
}

trait StoredItem[Id] {
  def id: Id
}

trait StoreFactory[Store] {
  var instance: Option[Store] = None
  
  def apply(): Store
}

ThingStore.scala

package stores

import actions._
import dispatchers.ProjectEditorDispatcher
import models.{Thing, ThingId}
import flux.{Store, StoreFactory}
import utils.Loggable

class ThingStore extends Store[ThingId, Thing] with Loggable {

  val dispatchToken = SomeDispatcher.register({
    case AddSomething(thing) =>
      store(thing.id) = thing
      emitAdd(thing.id)

    case _ =>
  })
}

object ThingStore extends StoreFactory[ThingStore] {
  def apply(): ThingStore = {
    instance.getOrElse {
      instance = Some(new ThingStore)
      instance.get
    }
  }
}