Event dispatcher

Question

In the last 2 years or so, I've learned a great deal about how to write better Scala code, but I know that I've barely scratched the surface. This is part of a library that I've been using. It has evolved a great deal since the first version and I'm curious to see how many mistakes I'm still making. I am particularly curious about what I can do to make better use of the Scala language.

This is already fairly long, so I'm leaving out my test class, though I would be happy to see test case suggestions.

EventDispatch.scala

package edu.stsci.efl.services

import edu.stsci.efl.events.EFLEvent
import net.liftweb.actor.LiftActor

/**
 * This is the interface definition for classes that provide application event services.
 */
trait EventDispatchService {
  def registerForEvent(name: String, callback: (EFLEvent) => Unit): EventRegistration
  def registerConditionallyForEvent(name: String, callback: (EFLEvent) => Unit, condition: (String, Any)): EventRegistration
  def registerActorForEvent(name: String, actor: LiftActor): EventRegistration
  def postEvent(event: EFLEvent)
  def postDelayedEvent(event: EFLEvent, delay: Int)
}

trait EventRegistration {

  private var status: EventRegistrationStatus = ERSPending
  def deregister()
  def is (s: EventRegistrationStatus) = s == status
  def update(s: EventRegistrationStatus) {status = s}
  def isAcceptingEvents = status == ERSActive
}

case class EventRegistrationStatus(name: String)

object ERSPending extends EventRegistrationStatus("pending")
object ERSActive extends EventRegistrationStatus("active")
object ERSPaused extends EventRegistrationStatus("paused")
object ERSTerminating extends EventRegistrationStatus("terminating")
object ERSTerminal extends EventRegistrationStatus("terminal")

Event.scala

package edu.stsci.efl.events

import edu.stsci.efl.ml.EFLContext

import org.slf4j.Logger

import scala.collection.mutable
import scala.xml.Node

/**
  * A generic event class with a name and named properties.
  */
class EFLEvent(val name: String, properties: Map[String, Any]) {
  var context: EFLContext = null
  var state: EventState = ESConstructed

  /**
   * Query an event to see if it is in a particular state
   * @param s - the state of interest
   * @return true if the event is in that state
   */
  def is(s: EventState): Boolean = state == s

  def update(s: EventState) { if (state != ESComplete) state = s }

  /**
    * Retrieves the value of a named property
    *
    * @param name - the key under which the property is stored
    * @return Object - the value attached to the given name
    */
  def getProperty(name: String): Option[Any] = { properties.get(name) }

  /**
    * Check to see if an event contains a property
    * @param name - the name of the property of interest
    * @return Boolean - true if the property is defined for this event
    */
  def hasProperty(name: String): Boolean = properties.contains(name)

  /**
    * Retrieves the value of a named property, but only if it is a String
    *
    * @param name - the key under which the property is stored
    * @return String - the String value attached to the given name
    */
  def getStringProperty(name: String): String = {
    properties.get(name) match {
      case None => null
      case Some(s: String) => s
      case _ => null
    }
  }

  /** Copies the properties of this event into another map.
    *
    * @param data - the destination map for the contents of properties.
    */
  def copyProperties(data: mutable.HashMap[String, Any]) { data ++= properties }

  def dump(logger: Logger) {
    logger.debug("[dump] event properties for '" + name + "' in context: " + context)
    properties.foreach {
      case (k: String, null) => logger.debug("[EFLEvent.dump] " + k + " is null.")
      case (k: String, v: Any) => logger.debug("[EFLEvent.dump] " + k + ": " + v)
    }
  }
}

case class EventState(name: String)

object ESConstructed extends EventState("constructed")
object ESPending extends EventState("pending")
object ESComplete extends EventState("complete")

class EventBuilder(name: String) {
  val properties = new mutable.HashMap[String, Any]()

  def build(): EFLEvent = { new EFLEvent(name, properties.toMap) }

    /**
      * Sets the value of a property
      *
      * @param name - the key under which the property is stored.
      * @param value - the value object attached to a property
      */
    def setProperty(name: String, value: Any) { properties.put(name, value) }

    /** Adds a collection of properties to this event.
      *
      * @param data - a map containing properties to be added
      */
    def addProperties[A <: Any](data: mutable.HashMap[String, A]) { properties ++= data }
}

object XMLEventConstructor {
  def createOneEvent(data: Node): EFLEvent = {
    val eventName = data.attribute("name").head.toString()
    val result = new EventBuilder(eventName)

    data.child.foreach((n: Node) => {
      if (n.label == "property") {
        val name = n.attribute("name").head.toString()
        val value = n.attribute("value").head.toString()
        result.setProperty(name, value)
      }
    })

    result.build()
  }
}

EventModule.scala

package edu.stsci.efl.events

import edu.stsci.efl.ml.{EFLContext, EFLModule}
import edu.stsci.efl.services._
import edu.stsci.util.ThreadTransactionMarker

import org.slf4j.{LoggerFactory, Logger}

import net.liftweb.actor.LiftActor
import net.liftweb.util.{Schedule, Helpers}

import scala.collection.mutable
import scala.xml.Node

import java.io.{StringWriter, PrintWriter}
import scala.Some

/**
 * The event module creates and registers an EventService that provides application event services.
 */
class EventModule extends EFLModule {

  private var logger: Logger = null
  private var eventDispatchService: EventService = null

  /**
   * Called by the ModuleLoader at program startup so that the module can initialize its functional components
   * and register its services.
   *
   * @param context - the specific EFLContext within which this module will be operating.
   * @param data - XML giving module specific initialization and configuration details.
   */
  def start(context: EFLContext, data: Node) {
    context.findService(classOf[LoggerFactoryService]) match {
      case None => // use whatever the default is, or was setup more conventionally
        logger = LoggerFactory.getLogger(getClass.getName)
      case Some(loggerFactory) => logger = loggerFactory.getLogger(getClass.getName)
    }

    logger.debug("[start] enter.")

    eventDispatchService = new EventService(context, logger)
    context.addService(eventDispatchService)
  }

  /**
   * Called by the ModuleLoader when the context is unloaded (usually at or just before program termination)
   * so that modules will be able to write out unsaved data, free up resources, etc.
   *
   * @param context - the EFLContext this module has been running in.
   */
  def stop(context: EFLContext) {
    logger.debug("[stop] enter.")
    waitForShutdown()

    eventDispatchService.shutDown()
    context.removeService(eventDispatchService)
    eventDispatchService = null
  }

  /**
   * Called by a context that wants to import this module from a parent module.
   *
   * @param context - the context that wishes to use this module's services
   */
  def addContext(context: EFLContext) {
    context.addService(eventDispatchService)
  }

  /**
   * Called by an importing context at shutdown
   *
   * @param context - the context that no longer needs this module's services
   */
  def removeContext(context: EFLContext) {
    context.removeService(eventDispatchService)
  }

  private def waitForShutdown() {
    var iterationCount = 0

    while (eventDispatchService.registerCount > 0 && iterationCount < 25 ) {
      Thread.sleep(50)
      iterationCount += 1
    }

    logger.trace("[waitForShutdown] waited {} iterations for shutdown.", iterationCount)
  }
}

case object EVENT_SHUTDOWN

trait CallableRegistration extends EventRegistration {
  def call(event: EFLEvent)
}

private class EventService(context: EFLContext, logger: Logger) extends EventDispatchService {

  val actor = new EventActor(context, this)
  var registerCount = 0

  def shutDown() {actor ! EVENT_SHUTDOWN}

  def postEvent(event: EFLEvent) {
    logger.trace("[EventService.postEvent] ({}) posting: {}", context.name, event.name, "")
    actor ! event
  }

  def postDelayedEvent(event: EFLEvent, delay: Int) {
    Schedule.schedule(actor, event, Helpers.TimeSpan(delay))
  }

  def registerForEvent(name: String, callback: (EFLEvent) => Unit): EventRegistration = {
    logger.trace("[EventService.registerForEvent] ({}) registering for: {}", context.name, name, "")
    val registration = new SimpleRegistration(name, callback)
    val message = new SimpleRegistrationMessage(registration)
    actor ! message
    registration
  }

  def registerConditionallyForEvent(name: String, callback: (EFLEvent) => Unit, condition: (String, Any)): EventRegistration = {
    val result = new ConditionalRegistration(name, callback, condition)
    val message = new ConditionalRegisterMessage(result)
    actor ! message
    result
  }

  def registerActorForEvent(name: String, actor: LiftActor): EventRegistration = {
    val message = new ActorRegistration(name, actor)
    this.actor ! new RegisterActorMessage(message)
    message
  }

  class SimpleRegistration(val name: String, val callback: (EFLEvent) => Unit) extends CallableRegistration {
    def call(event: EFLEvent) {
      if (isAcceptingEvents) callback(event)
    }

    def deregister() {
      actor ! this
    }

    override def equals(obj: scala.Any): Boolean = callback.equals(obj)

    override def hashCode(): Int = callback.hashCode()
  }

  class ConditionalRegistration(val name: String, val callback: (EFLEvent) => Unit, val condition: (String, Any)) extends CallableRegistration {
    def call(event: EFLEvent) {
      callback(event)
    }

    def deregister() {
      actor ! this
    }
  }

  class ActorRegistration(val name: String, val callback: LiftActor) extends CallableRegistration {
    def call(event: EFLEvent) {
      if (isAcceptingEvents) callback ! event
    }

    def deregister() {
      actor ! this
    }
  }

  class EventActor(context: EFLContext, parent: EventService) extends LiftActor with ThreadTransactionMarker {
    val logger = context.findService(classOf[LoggerFactoryService]).get.getLogger(getClass.getName)
    val byEventName = new mutable.HashMap[String, mutable.HashSet[SimpleRegistration]]
    val actorByEventName = new mutable.HashMap[String, mutable.HashSet[ActorRegistration]]
    val conditionals = new mutable.HashMap[String, ConditionalDispatcher]

    private def shutDown() {
      logger.debug("[shutDown] Registrations not unregistered: {}", new Integer(byEventName.size))
      byEventName.keys.foreach((k) => logger.debug("[shutDown] " + k))
      byEventName.clear()

      logger.debug("[shutDown] Actor registrations not unregistered: {}", actorByEventName.size)
      actorByEventName.keys.foreach((k) => logger.debug("[shutDown] " + k))
      actorByEventName.clear()

      logger.debug("[shutDown] Conditional registrations not unregistered: {}" + conditionals.size)
      conditionals.keys.foreach((k) => logger.debug("[shutDown] " + k))
      conditionals.clear()
    }

    protected def messageHandler = {
      case m: SimpleRegistrationMessage => registerForEvent(m.r)
      case s: SimpleRegistration => deregisterForEvent(s)
      case cr: ConditionalRegisterMessage => registerConditionallyForEvent(cr.c)
      case cRe: ConditionalRegistration => deregisterConditionallyForEvent(cRe)
      case RegisterActorMessage(r) => registerActorForEvent(r.name, r)
      case ar: ActorRegistration => deregisterActorForEvent(ar)
      case event: EFLEvent => doDispatch(event)
      case EVENT_SHUTDOWN => shutDown()
      case a: Any =>
        val transKey = beginTrans(logger)
        logger.warn("[EventService$EventActor.messageHandler] got unknown message: {}", a)
        endTrans(logger, transKey)
    }

    def registerForEvent(r: SimpleRegistration) {
      val transKey = beginTrans(logger, "registerForEvent." + context.name + "." + r.name)
      try {
        val forEvent = {
          byEventName.get(r.name) match {
            case None =>
              val result = new mutable.HashSet[SimpleRegistration]
              byEventName.put(r.name, result)
              result
            case Some(s) => s
          }
        }
        logger.trace("[registerForEvent] {}", byEventName)

        logger.trace("[registerForEvent] adding: {}", r)
        forEvent.add(r)
        r.update(ERSActive)
      }
      catch {
        case t: Throwable => logger.error("[registerForEvent]", t)
      }

      parent.registerCount = byEventName.size + actorByEventName.size
      endTrans(logger, transKey)
    }

    def deregisterForEvent(s: SimpleRegistration) {
      val transKey = beginTrans(logger, "deregisterForEvent." + context.name + "." + s.name)
      try {
        byEventName.get(s.name) match {
          case None =>
            logger.warn("[deregisterForEvent] not found.")
          case Some(hashSet) =>
            hashSet.remove(s)
            logger.trace("[deregisterForEvent] remaining callbacks: {}", hashSet.size)
            if (hashSet.isEmpty) {
              logger.trace("[deregisterForEvent] removing empty hash set.")
              byEventName.remove(s.name)
            }
        }

        s.update(ERSTerminal)
      }
      catch {
        case t: Throwable => logger.error("[EventService$EventActor.deregisterForEvent]", t)
      }

      parent.registerCount = byEventName.size + actorByEventName.size
      endTrans(logger, transKey)
    }

    def registerConditionallyForEvent(c: ConditionalRegistration) {
      val transKey = beginTrans(logger, "registerConditionallyForEvent." + context.name + "." + c.name)

      try {
        val conditional = getConditional(c.name)

        conditional.add(c)

        c.update(ERSActive)
      }
      catch {
        case t: Throwable => logger.error("[registerConditionallyForEvent]", t)
      }

      parent.registerCount = byEventName.size + actorByEventName.size
      endTrans(logger, transKey)
    }

    def deregisterConditionallyForEvent(c: ConditionalRegistration) {
      val transKey = beginTrans(logger, "deregisterConditionallyForEvent." + context.name + "." + c.name)

      try {
        val conditional = getConditional(c.name)

        conditional.remove(c)

        if (conditional.myConditions.isEmpty) {
          conditionals.remove(c.name)
        }

        c.update(ERSTerminal)
      }
      catch {
        case t: Throwable => logger.error("[deregisterConditionallyForEvent]", t)
      }

      parent.registerCount = byEventName.size + actorByEventName.size
      endTrans(logger, transKey)
    }

    def registerActorForEvent(name: String, reg: ActorRegistration) {
      val transKey = beginTrans(logger, "EventService.registerActorForEvent." + context.name + "." + name)

      try {
        val forEvent = {
          actorByEventName.get(name) match {
            case None =>
              val result = new mutable.HashSet[ActorRegistration]
              actorByEventName.put(name, result)
              result
            case Some(m) => m
          }
        }

        forEvent.add(reg)
        reg.update(ERSActive)
      }
      catch {
        case t: Throwable => logger.error("[registerActorForEvent]", t)
      }

      parent.registerCount = byEventName.size + actorByEventName.size
      endTrans(logger, transKey)
    }

    def deregisterActorForEvent(ar: ActorRegistration) {
      val transKey = beginTrans(logger, "deregisterActorForEvent." + context.name + "." + ar.name)

      try {
        actorByEventName.get(ar.name) match {
          case None => logger.warn("[deregisterActorForEvent] did not find registration.")
          case Some(hashSet) =>
            hashSet.remove(ar)
            if (hashSet.isEmpty) actorByEventName.remove(ar.name)
        }

        ar.update(ERSTerminal)
      }
      catch {
        case t: Throwable => logger.error("[deregisterActorForEvent]", t)
      }

      parent.registerCount = byEventName.size + actorByEventName.size
      endTrans(logger, transKey)
    }

    def getConditional(name: String): ConditionalDispatcher = {
      conditionals.get(name) match {
        case None =>
          val conditional = new ConditionalDispatcher(name)
          conditionals.put(name, conditional)
          conditional
        case Some(conditional) => conditional
      }
    }

    def doDispatch(event: EFLEvent) {
      val transKey = beginTrans(logger, "messageHandler.dispatching." + event.name)
      try {
        event.context = context
        event.update(ESPending)
        logger.trace("[doDispatch] {}", byEventName)
        logger.trace("[doDispatch] {}", actorByEventName)

        Thread.sleep(5)

        byEventName.get(event.name) match {
          case None => logger.trace("[doDispatch] no regular callbacks.")
          case Some(set) =>
            logger.trace("[doDispatch] {} callbacks.", set.size)
            set.foreach(_.call(event))
        }

        actorByEventName.get(event.name) match {
          case None => logger.trace("[doDispatch] no actor callbacks.")
          case Some(set) =>
            logger.trace("[doDispatch] actors: {}", set.size)
            set.foreach(_.call(event))
        }
      }
      catch {
        case t: Throwable => logger.error("[doDispatch] ", t)
      }

      event.update(ESComplete)
      endTrans(logger, transKey)
    }
  }

  case class SimpleRegistrationMessage(r: SimpleRegistration)
  case class ConditionalRegisterMessage(c: ConditionalRegistration)
  case class DeregisterMessage(name: String, callback: (EFLEvent) => Unit)
  case class ConditionalDeregisterMessage(name: String, callback: (EFLEvent) => Unit, condition: (String, Any))
  case class RegisterActorMessage(r: ActorRegistration)
  case class DeregisterActorMessage(name: String, actor: LiftActor)

  class ConditionalDispatcher(val eventName: String) {
    var registration: EventRegistration = registerForEvent(eventName, execute)

    val myConditions = new mutable.HashMap[(String, Any), mutable.HashSet[ConditionalRegistration]]

    def add(c: ConditionalRegistration) {
      logger.trace("[add] enter with condition {}", c.condition)

      myConditions.get(c.condition) match {
        case None =>
          val callbacks = mutable.HashSet[ConditionalRegistration](c)
          myConditions.put(c.condition, callbacks)

        case Some(callbacks) =>
          callbacks.add(c)
      }
      logger.trace("[EventService$ConditionalDispatcher.add] exit.")
    }

    def remove(c: ConditionalRegistration) {
      logger.trace("[remove] enter with condition {}.", c.condition)
      val so = myConditions.get(c.condition)
      if (so == None) {
        logger.warn("[remove] tried to remove conditional that did not exist.")
        return
      }

      val callbacks = so.get
      val result = callbacks.remove(c)
      if (! result) {
        logger.debug("[remove] remove of {} failed.", c.callback)
      }

      if (callbacks.isEmpty) {
        logger.trace("[remove] removing set for condition {}.", c.condition)
        myConditions.remove(c.condition)
      }
      if (myConditions.isEmpty) {
        logger.trace("[remove] removing callback for event {}.", eventName)
        registration.deregister()
      }

      logger.trace("[remove] exit.")
    }

    def execute(event: EFLEvent) {
      logger.trace("[execute] enter.")
      myConditions.foreach((a) => evaluate(event, a._1, a._2))
    }

    def evaluate(event: EFLEvent, condition: (String, Any), callbacks: mutable.HashSet[ConditionalRegistration]) {
      val vOption = event.getProperty(condition._1)
      logger.trace("[evaluate] property: {}, looking for: {}, found: {}", condition._1, condition._2, vOption)
      vOption match {
        case None => // automatic false, do nothing
        case Some(value) => if (condition._2 == value) callbacks.foreach((callback: ConditionalRegistration) => callback.call(event))

      }
    }
  }

  /**
   * Gets the exception stack trace as a string.
   * @return String representation of exception.
   */
  def getStackTraceAsString(exception: Exception): String = {
    val sw = new StringWriter
    val pw: PrintWriter = new PrintWriter (sw)
    pw.print (" [ ")
    pw.print (exception.getClass.getName )
    pw.print (" ] ")
    pw.print (exception.getMessage )
    exception.printStackTrace (pw)
    sw.toString
  }
}

Answer 1

Your whole design looks very mutable, but because I don't know your domain or the libraries you use I can't say to what extend it makes sense to functionalize it. Some thoughts while looking through the code:

Always specify return types for public members (even consider Unit), especially in an interface definition: def postEvent(event: EFLEvent) → def postEvent(event: EFLEvent): Unit. This makes it easier to understand someone else code (sometimes the inferred types are nontrivial).

---

The return types in subclasses need to be specified most of the time too (what you did). Scala allows covariant return types in override definitions which can be a problem because type inference makes use of this feature and therefore can result in compilation errors in some special cases where one only works with concrete subclasses and not the interfaces.

---

Never initialize an instance field with a default value, always use the underscore: var context: EFLContext = null → var context: EFLContext = _. They have a different semantic meaning! The latter one leaves the initialization to the JVM, which does the thing one expects. The former one will initialize the field with null in the constructor, the JVM initialization is done before the constructor is even called. For subclass relationships this can make a huge difference:

scala> trait A { var x: String; init(); def init(): Unit = x = "initialized" }
defined trait A

scala> class B extends A { override var x: String = null }
defined class B

scala> class C extends A { override var x: String = _ }
defined class C

scala> (new B).x
res0: String = null

scala> (new C).x
res1: String = initialized

---

xs.foreach(n => {}) can be written as xs.foreach { n => }

---

I'm not familiar with Lift and therefore don't know their preferred coding style but when I see def registerForEvent(name: String, callback: (EFLEvent) => Unit): EventRegistration and an actor I can only think about callback hell. When I have actors then I prefer having only actors that send each other messages. Maybe a callback free and actor only solution makes sense for you too, think about it.

---

Your try-catch blocks look cumbersome. They can be abstracted away with something like

def withErrorLogging(message: String)(f: => Unit) =
  try f catch { case t: Throwable => logger.error(message, t) }

withErrorLogging("[registerActorForEvent]") { ... }

In fact, all of your register*/deregister* functions look the same. Try to move out the part which differ (which seems to be only the code in the try-block)

---

Sometimes you use a Thread.sleep(n), which shouldn't be used in an actor environment, because they block.

---

I would never pass a mutable HashSet: def evaluate(..., callbacks: mutable.HashSet[ConditionalRegistration]). Do you know that it is not changed in this function, or is it intended to be changed by this function? In any case always prefer immutability to the outside, it makes it a lot easier to reason about control flow.