Lift has a great implementation of comet model in which long-held HTTP requests lets the server to notify clients almost immediately that something happened. Under the hood, Lift comets are actors that lives outside of the HTTP request/response cycle. Even though comet-browser communication is trivial, the comet-to-comet (or comet-to-actor) communication can be a bit problematic, especially when two comets live in separate sessions. In this post, I would like to describe comet-to-comet and comet-to-backend communication patterns available in Lift and present a new approach for which we have just released a Lift module.

LiftSession

LiftSession lets to retrieve reference to comet or send a message to comet with one of these methods:

1
2
3

def findComet(theType: String)
def findComet(theType: String, name: Box[String])
def sendCometActorMessage(theType: String, name: Box[String], msg: Any)

There is one important limitation here – the comet you want to refer to must live in the same LiftSession that your calling code lives in.

If all you need to do is to send a message between two or more user’s comets, these methods may be sufficient. However, the most frequent scenario is that you want to send a message to all active comets interested in some activity no matter in which LiftSession they live in. As an example, let’s consider a web chat application. When user posts a message in a chat room, we want all other users in this room to see this message immediately. Polling database each second from all comets is less than optimal. What we need to have is a way to ping all active comets about the new message or, even better, send them a notification containing this message.

CometListener and ListenerManager

The Lift’s ListenerManager contains the sendListenersMessage(msg: Any) method that allows to send a message to all comets extending CometListener trait that registered for the given ListenerManager instance using registerWith method:

1
2
3
4
5
6
7
8
9
10
11

class CometListenerExample extends NamedCometActorTrait with CometListener {

  override protected def registerWith = ListenerManagerExample

  override def mediumPriority = {
    case NewChatMessage(msg) =>
      partialUpdate(...)
  }

  override def render: RenderOut = ...
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14

object ListenerManagerExample extends ListenerManager with LiftActor {
  case object Tick

  // just a dummy placeholder - it's required by ListenerManager
  val createUpdate: Any = "nothing"

  LAPinger.schedule(this, Tick, 5 seconds)

  override def mediumPriority = {
    case Tick =>
      sendListenersMessage(NewChatMessage(...))
      LAPinger.schedule(this, Tick, 5 seconds)
  }
}

ListenerManagerExample will send a NewChatMessage to all CometListenerExample instances each 5 seconds. The createUpdate method is required by ListenerManager trait – it’s used to create the first message that a newly subscribed CometListener will receive. In the example above it’s just a dummy, unhandled message. It’s not important in which session, the comet instance lives in. All comets registered in ListenerManagerExample will receive a message.

NamedCometListener

In case of NamedCometListener, the comet receiving messages does not have to extend any trait. NamedCometListener lets to retrieve a message dispatcher for the given comet name via getDispatcherFor method:

1
2
3
4
5
6
7

class NamedCometListenerExample extends CometActor {

  override def mediumPriority = {
    case NewChatMessage(msg) =>
      partialUpdate(...)
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

object NamedCometListenerExample extends LiftActor {
  case object Tick

  LAPinger.schedule(this, Tick, 2 seconds)

  override def mediumPriority = {
    case Tick =>
      val message = NewChatMessage(...)
      NamedCometListener.getDispatchersFor(Full("chat-comet")).map { dispatcher =>
        dispatcher.map(_ ! message)
      }

      LAPinger.schedule(this, Tick, 2 seconds)
  }
}

When you insert comet to a page, you may provide it’s name. For example:

1

<lift:comet type="ChatComet" name="chat-comet">

or you use NamedCometActorSnippet for that:

1
2
3
4
5

class Chat extends NamedCometActorSnippet {
  val cometClass = "ChatComet"

  val name = "chat-comet"
}

NamedCometListener retrieves message dispatchers for all comets (from all users’ sessions) with the given name. This approach may be a bit more handy than CometListener+ListenerManager when you need to send a message directly between two comets.

MessageBus

MessageBus is a new approach that has been released as a Lift module. You can include it in your project by adding it as a dependency. For example, in SBT:

1

"net.liftmodules" %% "messagebus_3.0" % "1.0" % "compile"

MessageBus facilitates communication between any two or more LiftActors (under the hood, LiftCometActors are LiftActors).

Actors subscribe to Topics which are abstractions allowing to specify in which type of messages the given actor is interested in. Topic is a trait with just one method def name: String. Example Topic implementation looks as follows:

1

case class ChatRoomTopic(val name: String) extends Topic

In order to subscribe actor to the given topic, you need to send a Subscribe message. On a flip side, actor unsubscribes from topic sending an Unsubscribe message. In case of comets, this will be ususally done in localStartup and localShutdown method.

Actors should listen for messages from MessageBus in the same way as they listen for any other type of message. That is, for example, messageHandler method for LiftActors and high/low/mediumPriority methods for CometActors:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

class MessageBusExample extends CometActor {

  override def localSetup = {
    super.localSetup
    MessageBus ! Subscribe(this, ChatRoomTopic(...))
  }

  override def localShutdown = {
    super.localShutdown
    MessageBus ! Unsubscribe(this, ChatRoomTopic(...))
  }

  override def mediumPriority = {
    case NewChatMessage(msg) =>
      partialUpdate(...)
  }
}

There are two ways to send a message using MessageBus: For and ForAll. The payload of For message will be delivered by MessageBus to all LiftActors that subscribed to the given Topic (we look for their equality). The payload of ForAll message will be delivered by MessageBus to all LiftActors that subscribed to the given Topic type.

For example:

1
2
3
4
5
6
7
8
9
10
11
12
13
14

object MessageBusExample extends LiftActor {
  case object Tick

  LAPinger.schedule(this, Tick, 2 seconds)

  def handleUserLeftRoom: PartialFunction[Any, Unit] = {
    case Tick =>
      MessageBus ! For(ChatRoomTopic("chat-comet-europe"), NewChatMessage(...)) // first message
      MessageBus ! For(ChatRoomTopic("chat-comet-north-america"), NewChatMessage(...)) // second message
      MessageBus ! ForAll[ChatRoomTopic](NewChatMessage(...)) // third message

      LAPinger.schedule(this, Tick, 2 seconds)
  }
}

• The first message will be delivered to all LiftActors that subscribed to ChatRoomTopic("chat-comet-europe"),
• The second message will be delivered to all LiftActors that subscribed to ChatRoomTopic("chat-comet-north-america"),
• The third message will be delivered to all LiftActors that subscribed to any instance of ChatRoomTopic type. That is, actors that subscribed to ChatRoomTopic("chat-comet-europe") and actors that subscribed to ChatRoomTopic("chat-comet-north-america") will receive this message.

As always, an example application using all inter-session communication mechanisms described here is available on GitHub: https://github.com/pdyraga/lift-samples/tree/master/comet-messages

You can find more information about MessageBus module in its repository: https://github.com/pdyraga/lift-message-bus

Special thanks goes to Antonio Salazar Cardozo who significantly contributed to the idea and code.