Q4M って DBI 使うとブロックしちゃうけど、非同期にするにはどうしたらいいんだろう。AnyEvent::Worker|AnyEvent::DBI (=fork) 使うのがいいのかな。スタンダードな方法がわからない

LDR Streaming API では自力で MQ 作ったというのを見たけど、これ使っとけ的なのはないのかなあ……

Memo

• http://wiki.secondlife.com/wiki/Message_Queue_Evaluation_Notes

かなりざっくり書くと以下のような感じになる気がするけど、こういう HTTP ベースの内部用途の pub/sub サーバって既に良いのはあるのかな……

package Buspub;

use strict;
use warnings;
sub route ($$);
use Router::Simple;
use Tie::IxHash;
use AnyEvent::HTTP;
use HTTP::Request::Common;
use JSON::XS;
use POSIX ();

use MongoDB;
my $connection = MongoDB::Connection->new(host => 'localhost', port => 27017);
my $subscriptions = $connection->get_database('pubsub')->subscriptions;
$subscriptions->ensure_index(Tie::IxHash->new( key => 1, created => -1 ), { background => 1 });

route '/sub' => {
	action => sub {
		my ($r) = @_;
		my $key = $r->req->param('key') or return $r->json({ error => "key required"});;
		my $callback = $r->req->param('callback') or return $r->json({ error => "callback required"});;
		my $id = $subscriptions->insert({
			key      => $key,
			messages => [],
			callback => $callback,
			max      => 25,
			created  => scalar time(),
		});
		LOG("SUB:: %s with callback: %s", $key, $callback);
		$r->json({ id => "$id" });
	}
};

route '/pub' => {
	action => sub {
		my ($r) = @_;
		my $key = $r->req->param('key') or return $r->json({ error => "key required"});;
		my $message = decode_json $r->req->param('message');

		my $cursor = $subscriptions->query({ key => $key });
		my %count;
		while (my $obj = $cursor->next) {
			$count{all}++;
			my $id  = $obj->{_id} . "";
			my $uri = $obj->{callback};
			my $messages = [ @{$obj->{messages}}, $message ];
			my $req = POST $uri, [ id => $id, messages => encode_json($messages) ];

			LOG("PUB:: => %s => %s with %d messages", $key, $req->uri, scalar @$messages);
			http_request $req->method => $req->uri,
				body => $req->content,
				headers => {
					map { $_ => $req->header($_), } $req->headers->header_field_names
				},
				timeout => 20,
				sub {
					my ($body, $headers) = @_;
					LOG("PUB:: <= %s <= %s with status:%d", $key, $req->uri, $headers->{Status});
					if ($headers->{Status} =~ /^2/) {
						$subscriptions->update({ _id => $obj->{_id} }, { '$pullAll' => { messages => $obj->{messages} } });
					} elsif ($headers->{Status} =~ /^4/) {
						$subscriptions->remove({ _id => $obj->{_id} });
					} elsif ($headers->{Status} =~ /^5/) {
						if (@$messages > $obj->{max}) {
							$subscriptions->remove({ _id => $obj->{_id} });
						} else {
							$subscriptions->update({ _id => $obj->{_id} }, { '$push' => { messages => $message } });
						}
					}
				}
			;
		}
		$r->json({ key => $key, delivered => \%count });
	}
};

route '/test/callback' =>  {
	action => sub {
		my ($r) = @_;
		use Data::Dumper;
		warn Dumper $r->req->param('id') ;
		warn Dumper decode_json $r->req->param('messages') ;
		$r->res->status($r->req->param('code') || 404);
		$r->res->content_type('application/json; charset=utf8');
		$r->res->content('{}');
		$r->res->finalize;
	}
};


BEGIN {
	my $router = Router::Simple->new;
	sub route ($$) { $router->connect(@_) };

	sub run {
		my ($env) = @_;
		if ( my $handler = $router->match($env) ) {
			my $c = Buspub::Context->new($env);
			$handler->{action}->($c);
		} else {
			[ 404, [ 'Content-Type' => 'text/html' ], ['Not Found'] ];
		}
	}

	sub LOG {
		my ($message, @args) = @_;
		print sprintf("[%s] $message", POSIX::strftime("%Y-%m-%d %H:%M:%S", localtime), @args), "\n";
	}
};

package Buspub::Request;
use parent qw(Plack::Request);

package Buspub::Response;
use parent qw(Plack::Response);

package Buspub::Context;
use JSON::XS;

sub new {
	my ($class, $env) = @_;
	bless {
		req => Buspub::Request->new($env),
		res => Buspub::Response->new(200),
	}, $class;
}

sub req { $_[0]->{req} }
sub res { $_[0]->{res} }

sub json {
	my ($self, $vars) = @_;
	my $body = JSON::XS->new->ascii(1)->encode($vars);
	$self->res->content_type('application/json; charset=utf8');
	$self->res->content($body);
	$self->res->finalize;
}

\&Buspub::run;

binding を書いた (まだブランチに入れてある)

• https://github.com/cho45/jsdeferred/blob/jetpack/jsdeferred.jetpack.js#L368

• setTimeout/clearTimeout がないので nsITimer で実装 (これでいいのか?)
• Deferred.postie という utility を追加

Deferred.postie(constructor, options) を使って widget/panel をつくるとメソッドが増えます。

• post(args..., function) //=> Deferred
  • function をレシーバの content context 内で args と共に呼び、結果を取得
• bind(selector, event, function) //=> Deferred
  • content 内の selector でマッチする要素全てで、 event が発火時 function (chrome context) を呼びだす

以下のように書けます。post に渡す関数は文字列化されて content に渡されるので、スコープを外部と共有していない。args も JSON で表現できるものしか渡せない。この点わかりにくいのでイマイチかもしれない。

完全に message のやりとりを置き換えられるわけではないけど、ちょっとしたイベント処理とかは直列で簡単に書けるようになるので、便利な場合は便利かもしれない。

const Deferred = require("jsdeferred").Deferred;
Deferred.define(this); //define 使う場合必ず引数に this を渡す必用あり

widget = Deferred.postie(widgets.Widget, {
	label      : "Foo",
	contentURL : 'http://example.com/',
	width      : 32,
	onClick    : function () {
		var self = this;

		next(function () {
			return self.
				post(1, 2, function (a, b) { // content context 
					return a + b;
				}).
				next(function (res) { // chrome context
					console.log(res);
				});
		}).
		next(function () {
			return wait(1);
		}).
		next(function () {
			return self.
				post(function (a, b) {
					throw "foo";
				}).
				next(function (res) {
					console.log(res);
				}).
				error(function (e) {
					console.log(e);
				});
		});
	},
	onMessage : function (message) {
		console.log(message);
	},
	contentScript : 'setTimeout(function () { postMessage("custom message") }, 1000)',
});

widget.post(function () {
	var d = new Deferred();
	setTimeout(function () {
		d.call(1);
	}, 1000);
	return d;
}).
next(function (a) {
	console.log("Hello! " + a);
}).
error(function (e) {
	console.log(e);
});

widget.bind("body", "click", function (e) {
	console.log("body clicked" + e);
}).
error(function (e) {
	console.log(e);
});

Twitter でマジメな話ができないので、変な議論を呼びそうだなーとか、フローに流しても仕方ないこととか、ここに書いていこうという気持ちだけど、それに対して Twitter で返信されたらどうしたらいいのだろうなあ。追記するだけだと相手が気付かないし……