YammerArtieClient.scala

import org.eclipse.jetty.client.HttpClient
import org.cometd.client.BayeuxClient
import org.cometd.client.transport.LongPollingTransport
import java.util.HashMap
import org.cometd.bayeux.client.{ClientSession, ClientSessionChannel}
import org.cometd.bayeux.{Message, Channel}
import org.cometd.bayeux.Message.Mutable

/**
 * Simple cometd client for talking to the Yammer Artie (RT...realtime, get it?) service. You will 
 * need to procure the channel name and auth token from a web service call to the feed you want to 
 * monitor in realtime (e.g. https://www.yammer.com/api/v1/messages/following.json).
 *
 * You will need the following dependencies for your project. Version numbers may not be fully up 
 * to date as this sample came from a couple of months ago.
 * 
 *   val bayeux = "org.cometd.java" % "bayeux-api" % "2.1.0.beta2" withSources()
 *   val cometdCommon = "org.cometd.java" % "cometd-java-common" % "2.1.0.beta2" withSources()
 *   val bayeuxClient = "org.cometd.java" % "cometd-java-client" % "2.1.0.beta2" withSources()
 *
 * The client will call the supplied "listener" method for each incoming message. You can get the 
 * JSON payload in your listener by calling Message.getJSON, which Jerkson is more than happy to
 * turn into case classes for you.
 * 
 */
class YammerArtieClient(url: String, channelName: String, authToken: String, listener: (ClientSessionChannel, Message) => Unit) {
  // Listener for the underlying cometd client, passes messages on to the listener.
  val cometdListener = new ClientSessionChannel.MessageListener() {
    def onMessage(channel: ClientSessionChannel, message: Message) {
      listener(channel, message)
    }
  }
  var client:ClientSession = null

  def connect() {
    val httpClient = new HttpClient
    httpClient.start()

    val options = new HashMap[String, Object]()
    val transport = LongPollingTransport.create(options, httpClient)

    client = new BayeuxClient(url, transport)
    client.addExtension(new ClientSession.Extension() {
      def sendMeta(client: ClientSession, message: Mutable) = {
        if (Channel.META_HANDSHAKE == message.getChannel) {
          message.getExt(true).put("token", authToken)
        }
        true
      }

      def send(p1: ClientSession, p2: Mutable) = true

      def rcvMeta(p1: ClientSession, p2: Mutable) = true

      def rcv(p1: ClientSession, p2: Mutable) = true
    })

    client.getChannel("/meta/*").addListener(new ClientSessionChannel.MessageListener() {
      def onMessage(channel: ClientSessionChannel, message: Message) {
        if (message.getChannel == Channel.META_HANDSHAKE && message.isSuccessful) {
          client.getChannel(channelName).subscribe(cometdListener)
        }
      }
    })

    client.handshake()
  }

  def disconnect() {
    client.getChannel(channelName).unsubscribe(cometdListener)
    client.disconnect()
  }
}