プロフィール

江島健太郎

ブログ管理

最近のエントリー

• WWDCとLightBike 2とiPhone 4
  2010/07/01
• iPad初体験レビュー
  2010/04/05
• SinatraでTwitter Streaming APIにアクセスする超簡単なWebアプリのつくりかた
  2010/03/20
• スケールアウトからスケールアップへの回帰
  2010/01/12
• ネオフィリアの生きる道
  2010/01/02

エントリー一覧

Twitterには、ストリーミングAPIという、プッシュでリアルタイムに情報を受け取ることのできるAPIが用意されています。

これを使えば、定期的にポーリングをしたりすることなく、誰かが発言した時点で即座にメッセージを受け取れます。

しかし、これを使うためには、TCP接続をTwitterに対して張りっぱなしにしておく必要があり、Webアプリなどで使うのに敷居が高いと感じている人もいるのではないでしょうか。

今回は、「そんなことないよ、超簡単だよ」ということを、Rubyベースのサンプルで示したいと思います。

なぜそんなことが簡単にできるのか、その秘密を先に種明かしすると、Ruby用のWebサーバとして急速に人気を獲得し、デファクトの座をとりつつある「Thin」というWebサーバが、内部的にEventMachineという非同期サーバを使用しているので、その機能を使うことで、今までのWebアプリでは考えられなかったような自由度で様々なことができるようになる、ということなのです。まぁ、非同期とか難しいことがわからなければ、とりあえず聞き流してください。以下で実際に動かすのはとても簡単ですから。

まず、TwitterのStreaming APIそのものに関しては、「しばそんノート」のこの記事が参考になると思いますので、ここでは割愛。

今回つくるのは、「全世界の全てのパブリック・タイムライン（から5%をサンプリングしたもの）を常時とってきて、最新10件をキープしておき、それをHTMLで表示する」というものです。開発環境はMacを想定していますが、Linuxでも大丈夫です。

最終的には http://lingr.heroku.com/tweets ←これと同じものが動くようになるはずです。

実際に作ってみる

では、まずセットアップ方法からです。

gem install sinatra
gem install thin
gem install em-http-request
gem install json

ターミナルを開いてsinatra, thin, em-http-request, json をインストールします。どれもRubyベースでWebアプリを開発するうえでポピュラーなものばかりなので、遠慮なく盛大にインストールしちゃってください。

特にSinatraは、Railsで作るには大袈裟すぎる、ちょっとしたウェブアプリを作るのに最適な、軽量フレームワークです。最近はすごく勢いがあるので、今までRailsしか知らなかった人とか、Railsが面倒くさそうで始められなかった人は、この機会に触れてみるのがいいと思います。

では次に、ソースコードです。以下のファイルを、「tweets.rb」という名前で、適当なところに保存してください。

このとき、

1. USERNAMEとPASSWORDを、自分のTwitterアカウントのものに置き換える
2. CNET Blogの制約で「buffer.slice!(/.+＼r?＼n/)」という部分で「＼（バックスラッシュ）」が全角で入っているので、これを半角に置き換える

の２点をお忘れなく。

require 'rubygems'
require 'sinatra'
require 'em-http'
require 'json'

get '/tweets' do
  content_type 'text/html', :charset => 'utf-8'
  TWEETS.map {|tweet| "<p><b>#{tweet['user']['screen_name']}</b>: #{tweet['text']}</p>" }.join
end

class RingBuffer < Array
  def initialize(size)
    @max = size
    super(0)
  end

  def push(object)
    shift if size == @max
    super
  end
end

TWEETS = RingBuffer.new(10)
STREAMING_URL = 'http://stream.twitter.com/1/statuses/sample.json'

def handle_tweet(tweet)
  return unless tweet['text']
  TWEETS.push(tweet)
end

EM.schedule do
  http = EM::HttpRequest.new(STREAMING_URL).get :head => { 'Authorization' => [ 'USERNAME', 'PASSWORD' ] }
  buffer = ""
  http.stream do |chunk|
    buffer += chunk
    while line = buffer.slice!(/.+＼r?＼n/)
      handle_tweet JSON.parse(line)
    end
  end
end

はい、これで出来上がりです。では、ターミナルから「ruby tweets.rb」と入力して、サーバを起動してみてください。Sinatraは、デフォルトでまずthinを起動しようと試みるので、これだけでthinが起動します。

$ ruby tweets.rb
== Sinatra/0.9.6 has taken the stage on 4567 for development with backup from Thin
>> Thin web server (v1.2.7 codename No Hup)
>> Maximum connections set to 1024
>> Listening on 0.0.0.0:4567, CTRL+C to stop

こんな感じのプロンプトが出ていれば、サーバが起動して、さっそくTwitterからガシガシStreamingでデータをとってきています。（もしエラーになったら、たぶんユーザ名とパスワードだと思うので、もう一度よく確認してください）

アクティビティ・モニタでネットワークを見てみると、

こんな感じで毎秒30-40KBぐらいのペースでデータがダウンロードされてきていることがわかります。CPUの使用率は、ぼくの環境では2%とか。ほぼ誤差ですね。メモリの使用量は、しばらく走らせて放置しておいてみた感じ、64bitのRuby 1.9.1ベースで35MBぐらいで安定しています。つまり、どこにも大きな負荷はかかっていない、ということです。

では、次に http://localhost:4567/tweets にアクセスしてみてください。

こんな感じの画面が表示されたら成功です。リロードを素早く連打してみると、どんどん新しいメッセージが入ってきているのがわかります。

解説編

さて、このコードは具体的に何をやってるのでしょうか。

逐次解説すると、まず、

get '/tweets' do
  content_type 'text/html', :charset => 'utf-8'
  TWEETS.map {|tweet| "<p><b>#{tweet['user']['screen_name']}</b>: #{tweet['text']}</p>" }.join
end

この部分は、「/tweets」というURLにアクセスされると、TWEETSの中身をHTMLにして表示する、というSinatra流の書き方です。

次に、

class RingBuffer < Array
  def initialize(size)
    @max = size
    super(0)
  end

  def push(object)
    shift if size == @max
    super
  end
end

この部分では、標準のArrayを少し拡張して、サイズに上限のあるRingBufferというクラスを定義しています。ものすごい勢いで次々にやってくるメッセージを全部メモリ上に保管していたら大変なことになるので、最新N件だけを保管するためです。

そして最後に、

TWEETS = RingBuffer.new(10)
STREAMING_URL = 'http://stream.twitter.com/1/statuses/sample.json'

def handle_tweet(tweet)
  return unless tweet['text']
  TWEETS.push(tweet)
end

EM.schedule do
  http = EM::HttpRequest.new(STREAMING_URL).get :head => { 'Authorization' => [ 'USERNAME', 'PASSWORD' ] }
  buffer = ""
  http.stream do |chunk|
    buffer += chunk
    while line = buffer.slice!(/.+＼r?＼n/)
      handle_tweet JSON.parse(line)
    end
  end
end

この部分が今回のミソです。

見ての通り、TWEETSは、最新10件を保管する入れ物を用意してやって、STREAMING_URLは実際に接続するURLを定義しています。次のhandle_tweetでは、1件分のtweetが取れるごとに呼び出され、ちゃんとしたtextの入ったメッセージであればリングバッファに入れます。ここまではまぁ、普通のRubyのプログラムです。

次には、見慣れない「EM」というモジュールが登場します。これこそが、「Thin」が「EventMachine（略称EM）」で動いている、ということの意味です。「EM.schedule」は、このファイルが読み込まれた時点ではまだEMのイベントループが開始されてないので、開始された直後にブロックの中身を（1度だけ）実行するようにスケジュールする、というメソッドです。

そして、「EM::HttpRequest.new(STREAMING_URL).get」で指定のURLに接続し、ダウンロードを開始します。すると、以後に不定期な間隔（秒あたり数十回ぐらい？）で、「http.stream」で登録されたコールバックが呼び出されます。このとき、その時点までにダウンロードされている内容が丸ごとchunkで渡されるので、それをバッファリングして、改行を検出したらそこまでの内容をひとつの「行（line）」とし、その行がJSON文字列になっているので、行単位で取り出してRubyオブジェクト（Hash）に変換します。そのHashオブジェクト（＝1件のtweet）を、都度、handle_tweetに渡してリングバッファに登録しているのです。だから、「/tweets」にアクセスされたときには、単にこの常時更新されているリングバッファの中身を表示するだけでよい、というわけです。

以上、簡単だったでしょう？

このテクニックを応用すれば、EM.add_periodic_timerをつかって、5秒に一度とか、cronよりも細かい周期でバックグラウンドで何か処理させたり、タイムラインから特定のキーワードが見つかったら（ストリーミングAPIの「filter」を使います）、iPhoneのプッシュ通知機能をつかって通知したり（こちらも同じテクニックでTCPでAppleのサーバへ常時接続することになります）、などが、このソースコードの延長線上で可能になるのです。しかも、スレッドを使わずに、です。どうです、ちょっと興味わいてきませんか？

これで興味がわいてきたら、EventMachineというライブラリを深掘りすることをオススメします。ぼくは、最近何をやるにもこれがないと始まらない、、、というぐらいに依存しまくり、活用しまくりのライブラリです。もちろんLingrでもこれを最大限に活用しています。

おまけ

さて、以下はオマケ。今回のコードを実際に本番環境へデプロイしちゃう方法です。（しかも無料で！）

「heroku.com」というRubyアプリ専用のクラウドがあります。このHeroku、実はThinで動いているので、上記のコードがそのまま動くのです。Thinのインスタンス1個分だけならタダで使えるので、ちょっとしたコードをデプロイして動かすのには最適の環境です。

では、先程つくった「tweets.rb」と同じディレクトリに、「config.ru」というファイルを作成し、以下の2行を入力してください。

require 'tweets'
run Sinatra::Application

これで、Herokuのチュートリアルにしたがって、gitリポジトリの初期化やらアプリケーションの登録やらをやって、最後に「git push」とやれば、アプリケーションがデプロイされます。

そうやってデプロイされたものが http://lingr.heroku.com/tweets ←こちらです。しばらく連続稼動させてると接続を切られてしまうようなので、それを再接続させるには「unbind」をフックして「reconnect」を行う、などの処理が必要なのですが、ここから先はみなさんの宿題ということで。すでに接続されてる状態で、同じユーザ名で別の環境から接続を試みると、前の接続が強制切断されるので、開発環境とデプロイ環境で別々のアカウントを使うようにしましょう。

あと、PerlならAnyEventとかPSGI/PlackとかTwiggyというのを組み合わせると同様なことが実現できるようです。って、この話をブログに書こうかなーとチャットで話していたら同じネタでみやーんに先をこされてしまった。

♪ Marie Digby / Avalanche