Twitter is preparing to launch several impressive new features, including a new streaming API that will give desktop client applications real-time access to the user’s message timeline. The new streaming API was announced last week at Twitter’s Chirp conference, where it was made available to conference attendees on-site for some preliminary experimentation. Twitter opened it up to the broader third-party developer community on Monday so that programmers can begin testing it to offer informed feedback.

This tutorial will show you how to consume and process data from Twitter’s new streaming API. The code examples, which are written in the Python programming language, demonstrate how to establish a long-lived HTTP connection with PyCurl, buffer the incoming data, and process it to perform the basic message display functions of a Twitter client application. We will also take a close look at how the new streaming API differs from the existing polling-based REST API.

Understanding the polling model

Twitter client software generally uses a polling method to communicate with the Twitter service. Applications post and retrieve Twitter messages by sending HTTP requests with certain parameters to specific Twitter URLs. Twitter’s servers send back XML or JSON responses, which are then parsed and processed by the client application. This mechanism is based loosely on the concept of Representational State Transfer (REST), though it does not strictly conform with the REST paradigm. Twitter provides REST endpoints that make virtually all of its functionality accessible to third-party software.

Most Twitter applications that display a message timeline are programmed to request message updates on a configurable interval, typically ranging from two to five minutes. This polling model fundamentally distinguishes Twitter from instant messaging because it introduces unavoidable latency between when messages are posted and when they are received. The only way to decrease the latency with Twitter’s current standard API is to reduce the polling interval and request updates more frequently. That obviously puts a high load on Twitter’s servers and creates scalability challenges. In order to prevent abuse, Twitter has a rate-limiting mechanism that sets a cap on the number of API requests that a single user can make per hour.

Twitter’s heavy reliance on the polling model poses a bit of a paradox. One of Twitter’s greatest strengths is timeliness—the site’s front page touts it as a source of “instant information” and a window into what is happening “right now” in the world. Although Twitter largely delivers on that promise by enabling news and ideas to propagate faster than is generally possible in other mediums, its reliance on the polling model prevents it from facilitating truly real-time messaging.

Overcoming the limitations of polling

Twitter’s new streaming API will make it possible for third-party software to overcome the limitations of polling, giving Twitter a big boost as its emphasis continues to shift toward real-time messaging. The streaming API allows client applications to establish a persistent connection with Twitter’s servers and constantly receive messages right after they are posted, obviating the need for polling on an interval. This communication model could be described as “push” messaging.

Handling millions of simultaneous persistent connections comes with its own set of scalability challenges, so Twitter is testing the waters before it proceeds with a full-scale production roll-out. At this time, the streaming API is still experimental and is not ready to be adopted in applications that are released for general use. The streaming capacity is limited and the API is still subject to change. It is being made available with pre-beta status so that developers can begin testing the functionality.

Twitter’s documentation warns that it could change the streaming endpoint arbitrarily and without notice if it thinks it’s being abused. As the testing progresses, Twitter will begin hammering out a schedule for a beta launch and a full production release. Although it’s not yet ready for widespread use in client applications, this is a great time to start playing around to see how it works.

A look at the streaming API

Unlike Twitter’s conventional REST API, which supports both JSON and XML, the streaming API offers only JSON output. JSON, which stands for JavaScript Object Notation, is a simple and elegant format for describing structured data. The syntax, which is based on JavaScript, is human-readable and very easy to parse. It’s so simple that the entire grammar can fit on the back of a business card.

To use the streaming API, an application makes a long-lived HTTP request to the streaming endpoint. Unlike a conventional REST API request, where the connection to the server is terminated after data is received, the streaming API leaves the connection open for as long as possible and will perpetually push new data as it is available. The data is sent as blobs of JSON that describe messages and events, such as retweets and message deletion. The structure of the message data that is emitted by the streaming API matches that of the REST API, which means that application developers who are already using the JSON output format can reuse their existing message parsing code.

There are several different techniques that are commonly used in Web programming to achieve push messaging with long-lived HTTP connections. These techniques are collectively referred to as Comet communication. It’s important to understand that the Twitter streaming API is designed for true HTTP streaming and isn’t intended to be used with other common Comet methods such as long-polling. In a long-polling scenario, a connection is established and held open until data is received, at which point the connection is terminated and then reestablished in anticipation of more data. When HTTP streaming is used, the connection just stays open and keeps getting data.

Consuming the streaming API with PyCurl

The easiest way to handle HTTP streaming in Python is to use PyCurl, the Python bindings for the well-known Curl network library. PyCurl allows you to provide a callback function that will be executed every time a new block of data is available. The following code is a simple demonstration of HTTP streaming with PyCurl:

import pycurl, json

STREAM_URL = "http://chirpstream.twitter.com/2b/user.json"

USER = "segphault"
PASS = "XXXXXXXXX"

def on_receive(data):
  print data

conn = pycurl.Curl()
conn.setopt(pycurl.USERPWD, "%s:%s" % (USER, PASS))
conn.setopt(pycurl.URL, STREAM_URL)
conn.setopt(pycurl.WRITEFUNCTION, on_receive)
conn.perform()

The code example above shows how to instantiate a Curl object, set the URL, provide login credentials, and send the data to a callback function. The callback function in the example simply echoes the received data to the terminal. If you put in your own Twitter credentials and run that code in a Python script at the command line, you will see the stream of JSON data transmitted by the Twitter service.

When the connection is idle and there is no other data to send, the streaming API will emit an empty line every 30 seconds. The empty line is a keep-alive signal that is intended to prevent client applications from timing out and dropping the connection. PyCurl doesn’t require any special configuration, but other network libraries might require the user to set a custom timeout duration for idle connections. You should make sure that it is set to something that is higher than the streaming API’s 30-second keep-alive interval so that the connection isn’t dropped.

Establishing the connection and receiving the data is clearly easy, but processing it is a bit more complex. The blocks that are sent to the callback function are each less than 1500 bytes. If a JSON object from Twitter exceeds the size of the block of data that is sent to the callback, then we have to do some buffering in order to ensure that we get the complete object. The JSON parser will choke if you try to pass it a partial structure.

At the end of every complete JSON object, the Twitter streaming API outputs a carriage return, which we can use to determine if we have a full object. Every time we receive a block, we append it to a buffer string and then check to see if it has the carriage return at the end. If it does, then we can assume that the buffer string contains a complete JSON object, which can be handed off to the JSON parser and then processed. After we parse the JSON in the buffer string, we clear the buffer so that it can start collecting new data.

According to Twitter’s documentation, it’s possible for a linebreak character (“n”) to appear in the strings within the JSON object. The full carriage return (“nr”), however, will only appear at the end of an object. It’s important to keep that in mind when you are implementing a buffering system. It’s also important to watch out for the keep-alive signal because it consists solely of a carriage return. When we detect a carriage return at the end of a line, we have to make sure that the buffer actually has something in it before we pass it to the JSON parser. This prevents us from accidentally trying to parse the keep-alive line.

In the code example below, we detect the end of a JSON object by using the “data.endswith("rn") and self.buffer.strip()” conditional expression. In Python conditional expressions, empty strings evaluate to boolean False, which means that “and self.buffer.strip()” is equivalent to “and len(self.buffer.strip()) > 0”.

import pycurl, json

STREAM_URL = "http://chirpstream.twitter.com/2b/user.json"

USER = "segphault"
PASS = "XXXXXXXXX"

class Client:
  def __init__(self):
    self.buffer = ""
    self.conn = pycurl.Curl()
    self.conn.setopt(pycurl.USERPWD, "%s:%s" % (USER, PASS))
    self.conn.setopt(pycurl.URL, STREAM_URL)
    self.conn.setopt(pycurl.WRITEFUNCTION, self.on_receive)
    self.conn.perform()

  def on_receive(self, data):
    self.buffer += data
    if data.endswith("rn") and self.buffer.strip():
      content = json.loads(self.buffer)
      self.buffer = ""
      print content

client = Client()

Now that we have a simple mechanism for obtaining and parsing JSON data from the streaming API, we can start looking at how to process it so that we can build useful tools. As we process the data that is returned by the API, we will use a series of conditional expressions to look for specific keys that are unique to the individual kinds of objects that are sent by the API. To detect a message, for example, we look for the “text” key. The following code example shows how to display the sender name and text of each incoming message.

  def on_receive(self, data):
    self.buffer += data
    if data.endswith("rn") and self.buffer.strip():
      content = json.loads(self.buffer)
      self.buffer = ""

      if "text" in content:
        print u"{0[user][name]}: {0[text]}".format(content)

One of the major differences between the output of the REST API and the streaming API is that the streaming API includes replies that the user’s friends are sending to people who the user doesn’t follow. This can clutter up the stream quite a bit and lead to some confusion. To filter out those messages, we can look at the “in_reply_to_user_id” key of the message and determine if the recipient is one of the user’s friends.

Fortunately, when a connection is first established, the streaming API will always start by passing down a list of the Twitter IDs of the people who the user is following. We can store that list in a variable and use it to determine if a message is directed at someone who the user isn’t following. You can distinguish the friend list from other kinds of data by checking to see if the “friends” key is in the JSON object. The friend list object looks like this:

{"friends":[29710117,23359726,14211758,22332381,686793,14542767,14866691,32447686,5352732,17009124,33406310,19122978,26114930,17549127,990951,23335036,15304159,12032,14625444,11375732,22289742,52604183,17158868,14575846,14671896]}

When we are processing a message object, we can check to see if the value of the “in_reply_to_user_id” key is in the friends list. If it is not, then we could simply not display the message.

class Client:
  def __init__(self):
    self.buffer = ""
    self.friends = []
    self.conn = pycurl.Curl()
    self.conn.setopt(pycurl.USERPWD, "%s:%s" % (USER, PASS))
    self.conn.setopt(pycurl.URL, STREAM_URL)
    self.conn.setopt(pycurl.WRITEFUNCTION, self.on_receive)
    self.conn.perform()

  def on_receive(self, data):
    self.buffer += data
    if data.endswith("rn") and self.buffer.strip():
      content = json.loads(self.buffer)
      self.buffer = ""

      if "friends" in content:
        self.friends = content["friends"]

      if "text" in content:
        to = content["in_reply_to_user_id"]
        if to and to not in self.friends: return
        print u"{0[user][name]}: {0[text]}".format(content)

There is a problem with that approach, however. It also has the undesired affect of filtering out all of the replies that are directed to the actual user. The user’s own ID is obviously not going to be in the friend list, which means that we need to get it separately so that we can compare “in_reply_to_user_id” against it when deciding whether to drop a message. We can also use it to determine when a message is a reply to the user.

The streaming API doesn’t provide that information, so we have to use a separate REST API call before we connect to the streaming API. We are going to use the account/verify_credentials.json REST method, which will also helpfully allow us to validate the user’s login information. That API method returns data about the user, including the user’s ID, which we can store in an instance variable for later use. It will return an error if the user’s credentials are wrong.

import pycurl, json, StringIO

STREAM_URL = "http://chirpstream.twitter.com/2b/user.json"
REST_URL = "http://api.twitter.com/1/"

class Client:
  def __init__(self):
    self.friends = []
    self.buffer = ""
    self.userid = None
    self.conn = pycurl.Curl()

  def authenticate(self, username, password):
    output = StringIO.StringIO()
    self.conn.setopt(pycurl.USERPWD, "%s:%s" % (username, password))
    self.conn.setopt(pycurl.URL, REST_URL + "account/verify_credentials.json")
    self.conn.setopt(pycurl.WRITEFUNCTION, output.write)
    self.conn.perform()

    data = json.loads(output.getvalue())
    if "error" in data: return False
    self.userid = data["id"]
    return True

  def connect(self):
    self.conn.setopt(pycurl.URL, STREAM_URL)
    self.conn.setopt(pycurl.WRITEFUNCTION, self.on_receive)
    self.conn.perform()

  def on_receive(self, data):
    self.buffer += data
    if data.endswith("rn") and self.buffer.strip():
      content = json.loads(self.buffer)
      self.buffer = ""

      if "friends" in content:
        self.friends = content["friends"]

      elif "text" in content:
        to = content["in_reply_to_user_id"]
        if to and to != self.userid and to not in self.friends: return
        if to == self.userid: print "(REPLY)",
        print u"{0[user][name]}: {0[text]}".format(content)

client = Client()
if client.authenticate("segphault", "XXXXXXXXX"):
  client.connect()
else:
  print "Login credentials aren't valid!"

Now that we have basic support for messaging, we are going to add one more feature. The streaming API makes it easy to detect when the user gets a new follower. To detect follow event objects, look for the “event” key and check if it has the string “follow” as the value. The follow event object will contain “source” and “target” objects with the user ID of the follower and the user ID of the person being followed.

{"target":{"id":17591742},"source":{"id":15878405},"event":"follow"}

If the target ID matches the ID of the user, then we know that the user got a new follower. Similarly, if the source ID matches the ID of the user, then we know that the target ID is a new friend of the user and should be added to the friends list. Unfortunately, the streaming API only gives you the ID, so you have to use a REST API call if you want to get more details about a new follower.

def on_receive(self, data):
  self.buffer += data
  if data.endswith("rn") and self.buffer.strip():
    content = json.loads(self.buffer)
    self.buffer = ""

    if "friends" in content:
      self.friends = content["friends"]

    elif "event" in content and content["event"] == "follow":
      if content["target"]["id"] == self.userid:
        print "New follower:", content["source"]["id"]
      elif content["source"]["id"] == self.userid:
        self.friends.append(content["target"]["id"])

    elif "text" in content:
      to = content["in_reply_to_user_id"]
      if to and to != self.userid and to not in self.friends: return
      if to == self.userid: print "(REPLY)",
      print u"{0[user][name]}: {0[text]}".format(content)

The future role of the REST API

Although the streaming API gives developers powerful real-time access to the user’s message timeline, it doesn’t support the full scope of Twitter’s functionality. The REST API will still be used for many things, including posting messages. The REST API will also be useful for retrieving a backlog of messages when a client application starts up. The user’s initial stream can be populated from the REST API before the client establishes its connection to the streaming API to get new incoming messages.

14 Comments