PubSubHubBubを経由してBuzzのFirehoseを #appengine で使ってみる

ちょっと前に BuzzでFirehoseが使えるようになりました。

• Activity firehose - Developer's Guide (v1): Using REST - Google Buzz API - Google Code
• Working with real-time updates - Developer's Guide (v1): Using REST - Google Buzz API - Google Code

今の空うさぎでは、FriendFeedはリアルタイム（FriendsがPostした瞬間にクライアントに通知される）になっていますし、TwitterもUserStreamを使うことでFirendFeed同様にリアルタイムになっています（Twitter側の制限のために、リリース版ではTwitterのリアルタイムはオフになってますが）。さらに開発版ではGoogle Buzzも利用可能になっています。であればBuzzもリアルタイムに…！と思うんですが、これがTwitterやFriendFeedとはちょっと仕組みが違う。Webアプリでしか使えないようです。なので、AppEngineで受けてXMPP経由で空うさぎに送ってしまえないか？という事でPubSubHubBub を使ってAppEngine/JavaでBuzzのFirehoseを使ってみようかと、そんな流れです。
ここではまずはリアルタイムに取得されるBuzzのActivityをデータストアにどんどん保存するアプリを作ってみます。

まずAppEngine/Javaのプロジェクトを作る

slim3のarchetype-pluginを使って簡単にプロジェクトを構築します。

1. $ mvn archetype:generate -DarchetypeCatalog=http://slim3.googlecode.com/svn/trunk/repository
2. Projectの生成が対話形式で進むので、質問に従ってプロジェクト名などを入力
3. $ cd ${プロジェクト名}
4. $ mvn eclipse:eclipse
5. Eclipseでインポートします。

BuzzのActivityを保存するモデルクラスを作成

今回はお試し程度なので、Atomの解析はせずにそのまま格納するようにします。

	package com.shin1ogawa.model;
	import java.io.Serializable;
	import java.util.Date;
	import org.slim3.datastore.Attribute;
	import org.slim3.datastore.Model;
	import com.google.appengine.api.datastore.Key;
	@Model(kind = "buzz")
	public class BuzzActivity implements Serializable {
	private static final long serialVersionUID = -7133080152735096368L;
	@Attribute(primaryKey = true)
	private Key key;
	@Attribute(lob = true, name = "b")
	private String body;
	@Attribute(name = "at")
	private Date createdAt = new Date();
	public Key getKey() {
	return key;
	}
	public void setKey(Key key) {
	this.key = key;
	}
	public String getBody() {
	return body;
	}
	public void setBody(String body) {
	this.body = body;
	}
	public void setCreatedAt(Date createdAt) {
	this.createdAt = createdAt;
	}
	public Date getCreatedAt() {
	return createdAt;
	}
	}

view raw BuzzActivity.java hosted with ❤ by GitHub

PubSubHubBub用のコントローラを作成

http://myapp.appspot.com/subscriberというURLでPubSubHubBubをSubscribeすると想定して、それをハンドルするようにします。ここではSlim3のControllerを使います。

	package com.shin1ogawa.controller;
	import java.io.IOException;
	import java.util.logging.Logger;
	import org.slim3.controller.Controller;
	import org.slim3.controller.Navigation;
	import org.slim3.datastore.Datastore;
	import org.slim3.util.StringUtil;
	import com.shin1ogawa.model.BuzzActivity;
	public class SubscriberController extends Controller {
	static final Logger logger = Logger.getLogger(SubscriberController.class.getName());
	@Override
	protected Navigation run() throws Exception {
	if (isPost()) {
	return doPost();
	}
	return doGet();
	}
	private Navigation doGet() throws IOException {
	String hubMode = asString("hub.mode");
	String hubChallenge = asString("hub.challenge");
	if (StringUtil.isEmpty(hubMode) || StringUtil.isEmpty(hubChallenge)) {
	response.setStatus(400);
	return null;
	}
	String hubTopic = asString("hub.topic");
	if (hubMode.equals("subscribe")) {
	logger.info("subscribe: hub.challenge=" + hubChallenge
	+ ", hub.topic=" + hubTopic);
	} else if (hubMode.equals("unsubscribe")) {
	logger.info("unsubscribe: hub.challenge=" + hubChallenge
	+ ", hub.topic=" + hubTopic);
	} else {
	response.setStatus(400);
	return null;
	}
	response.setStatus(200);
	response.setContentType("text/plain");
	response.getWriter().print(hubChallenge);
	response.flushBuffer();
	return null;
	}
	private Navigation doPost() throws IOException {
	response.setStatus(204);
	byte[] buffer = new byte[10 * 1024];
	request.getInputStream().read(buffer);
	BuzzActivity entity = new BuzzActivity();
	entity.setBody(new String(buffer, "utf-8"));
	Datastore.put(entity);
	return null;
	}
	}

view raw SubscriberController.java hosted with ❤ by GitHub

PubSubHubBubでSubscriberの登録をする

1. 先の手順で作成したアプリケーションをデプロイします。
2. http://pubsubhubbub.appspot.com/subscribeを開きます。
3. Callback: (the subscriber URL)に、先に作成したSubscriberのURLを入力します。例えばhttp://myapp.appspot.com/subscriberのようなURLになるはずです。
4. Topic: (the feed URL)に、https://www.googleapis.com/buzz/v1/activities/@all/@publicを入力します。
5. Do itをクリックします。

登録処理が問題なくAppEngineの管理コンソールのINFOレベルのログを見ると、subscribe: hub.challenge=...のようなログが出力されるはずです。

どんどん流れてきます

AppEngineの管理コンソールでall requestログを見るとどんどんリクエストが飛んできている事が確認できます。DatastoreViewerの"buzz"カインドにはどんどんエンティティが増えていきます。こんなカンジでfirehoseが使えます。が、恐ろしい勢いでデータがたまっていくのでちょっと怖いですね。

PubSubHubBubからSubscriberの登録を解除する

登録した時の手順と同様に、Callback: (the subscriber URL)とTopic: (the feed URL)を入力して、Mode:にUnsubscribeを設定して、Do itをクリックします。問題なくUnsubscribeができればAppEngineの管理コンソールのINFOレベルのログを見ると、unsubscribe: hub.challenge=...のようなログが出力されるはずです。