twitter4j-stream with play framework.

Question

I'm trying to get started with the play framework by building a live twitter stream app. I'm super confused though, and have spent a few days trying to figure this out but I can't seem to get it.. hoping someone here wouldn't mind pointing me in the right direction.

A few questions.

1) The actual TwitterStreamFactory instantiation, where would this go? Is this part of a model, or a controller?

2)Websockets, comet, SSE? What's the most efficient way - for every status that comes through the stream - to handle and send to the browser.

3)And the StatusListener. Every time I try to create this in eclipse it tells me to add unimplemented methods.. but I have implemented them. It just won't play nicely.

3 is a big problem for me, because I can't get it to run anywhere. The code works on a tomcat server, but it's obviously quite different using play. Any advice on where I should be writing the twitter stream part of the code?

Thanks guys.. these questions are probably considered silly, but I am truly, truly in a rut here.

Answer 1

I made something similar a while back, i used websockets to push the updates to the browser and Akka actors to handle the status listener. The controller can then request a stream (Enumerator) from an actor and return it as a websocket.

2) Websocket is the most efficient but not compatible with older browsers, if you need better browser compatibility then use comet

I've modified the code to show a very simple example

Controller:

object Application extends Controller {

  implicit val timeout = Timeout(1 second)

  val cb = new ConfigurationBuilder()
  cb.setDebugEnabled(true)
    .setOAuthConsumerKey("")
    .setOAuthConsumerSecret("")
    .setOAuthAccessToken("")
    .setOAuthAccessTokenSecret("")

  val twitterListener = Akka.system.actorOf(TwitterListener.props(cb.build()))

  def join = WebSocket.async[JsValue] { request =>
    (twitterListener ? RequestStream()).mapTo[Connected].map {
      case Connected(stream) => (Iteratee.ignore, stream)
    }
  }
}

Actor:

object TwitterListener {
  case class RequestStream()
  case class Connected(numerator: Enumerator[JsValue])
  def props(conf: Configuration) = Props(new TwitterListener(conf))
}

/**
 * Twitter Stream Listener
 *
 * @param config Twitter4j Configuration
 */
class TwitterListener(config: Configuration) extends Actor {

  import TwitterListener._

  val listener = new StatusListener() {

    val (enum, channel) = Concurrent.broadcast[JsValue]

    def onStatus(status: Status) {
      channel.push(Json.obj(
        "msg" -> status.getText,
        "user" -> status.getUser.getName,
        "timestamp" -> DateTime.now.toString("yyyy-MM-dd HH:mm:ss")
      ))
    }

    def onDeletionNotice(statusDeletionNotice: StatusDeletionNotice) {

    }

    def onTrackLimitationNotice(numberOfLimitedStatuses: Int) {

    }

    def onException(ex: Exception) {
      ex.printStackTrace()
    }

    def onScrubGeo(userId: Long, upToStatusId: Long) = {

    }

    def onStallWarning(warning: StallWarning) = {

    }
  }

  override def preStart() = {
    val query = new FilterQuery(0, Array(), Array("birthday"))
    val twitterStream = new TwitterStreamFactory(config).getInstance
    twitterStream.addListener(listener)
    twitterStream.filter(query)
  }


  def receive = {
    case RequestStream() => sender ! Connected(listener.enum)
  }

}