Weblogs.com Ping Gateway to はてなアンテナ (POE版)

前回紹介した、はてなアンテナ Ping ゲートウェイには、1点問題がありました。Ping を受け取った後、HTTPリクエストを発行してレスポンスを待っているため、Ping 受信に非常に時間がかかることがあるという点です。ここに時間がかかると、Blog ツールのエントリビルド処理が待たされることになり、好ましくありません。Ping を受けとったら即座にレスポンスを返し、非同期にアンテナへのリクエストを投げる方がよいでしょう。

今回のレシピでは、POE と呼ばれるフレームワークを使用して、この処理の非同期化を行ってみます。

Perl の非同期処理

• fork() する
• スレッド (threads)を使用する
• POE (Persistent Object Environment) を使用する

fork() は UNIX では一般によく利用される手法で、プロセスの複製をつくって、親プロセスと子プロセスで処理を分岐させます。Perl でも関数 fork を利用すれば実装は容易です。ただし、今回の場合は Web サーバ (httpd) から CGI 経由で XML-RPC サーバを動作させている都合上、あまり fork を利用するのは好ましくない(*2)気もします。

perl のスレッドは非常に実験的ですが、5.8.1 からはより安定して使用できるようになりました。5.8 でのスレッド実装は、スレッド1つ毎に perl インタプリタを作成する実装(*3)となっており、5.6 以前のネイティブなスレッド実装とはまったく異なります。ネイティブなスレッドに比べると速度やメモリ消費の点で劣りますが、より安全に利用/実装できる点が特長です。ただ、5.8.1 でより安定したとはいえ、インタプリタ間で変数がデフォルトでは共有されない(*4)など、注意すべき点も多いため、今回は threads の利用は避けておきます。

POE は Persistent Object Environment (など様々なモノ)の略で(*5)、Perl でイベントドリブンなプログラミングを行うためのフレームワークです。POE は内部で select や alarm を用いてノンブロッキングな処理を実行する仕組みを持っており、ユーザはイベントを非同期にキューイングして実行することができ、ユーザレベルでのスレッドプログラミングを簡単に行うことができます。CPAN には様々な POE Component(*6) が登録されていて、マルチタスクなアプリケーションを組むには最適といえます。コーディングのルールには多少クセがありますが、ハマるとなかなか抜け出せない魅力も同時に持っています。

サンプルコード

use POE qw(Component::Server::XMLRPC
           Component::Client::HTTP
           Component::TSTP);

use POE;
use POE::Component::Server::XMLRPC;
...

use POE::Sugar::Args;

use HTTP::Request;
use Sys::Hostname;
use URI;

our $VERSION    = "1.00";
our $HatenaURL  = "http://a.hatena.ne.jp/check";

my $port = shift || 10080;

# session to trap Ctrl-Z
POE::Component::TSTP->create();

# session for XML-RPC server
POE::Component::Server::XMLRPC->new(
    alias => 'xmlrpc',
    port  => $port,
);

# session to send pings in HTTP
POE::Component::Client::HTTP->spawn(
    Alias => 'ua',
    Agent => "ping2hatena/$VERSION",
);

# main session
POE::Session->create(
    inline_states => {
        _start => \&setup_service,
        _stop  => \&shutdown_service,
        "weblogUpdates.ping" => \&ping_handler,
        got_response => \&response_handler,
    },
);

$poe_kernel->run();

sub setup_service {
    my $poe = sweet_args;
    $poe->kernel->alias_set("ping2hatena");
    $poe->kernel->post(xmlrpc  => 'publish',
                       ping2hatena => "weblogUpdates.ping");
    my $hostname = Sys::Hostname::hostname || 'localhost';
    warn "URI: http://$hostname:$port/?session=ping2hatena\n";
}

ここでは main のセッションに ping2hatena という別名をつけ、次に xmlrpc の publish() メソッドを呼び出す処理を POE カーネルにキューイングしています。publish メソッドの引数はそれぞれ、公開した XML-RPC API の処理を行う Session 名 (ping2hatena) と、公開するメソッド名 (= イベント名: weblogUpdates.ping) を表しています。

Sys::Hostname の hostname でホスト名を得た後、XML-RPC サーバの URL を warn で表示しています。URL には ?session={Session名} のクエリが必要なことに注意してください(*8)。

sub shutdown_service {
    my $poe = sweet_args;
    $poe->kernel->post(xmlrpc  => 'rescind',
                       ping2hatena => "weblogUpdates.ping");
}

sub ping_handler {
    my $poe = sweet_args;
    my $transaction = $poe->args->[0];
    my($blog_name, $blog_url) = @{$transaction->params()};
    my $url = URI->new($HatenaURL);
       $url->query_form(url => $blog_url);
    my $req = HTTP::Request->new(GET => $url);
    $poe->kernel->post(ua => request => got_response => $req);

    my %data = (flerror => XMLRPC::Data->type(boolean => 0));
    $transaction->return(\%data);
}

sub response_handler {
    my $poe = sweet_args;
    my $request  = $poe->args->[0]->[0];
    my $response = $poe->args->[1]->[0];
    if (defined($response->content)) {
        warn "successful request to ", $request->uri();
    } else {
        warn "error while requesting ", $request->uri();
    }
}

% ./ping_to_hatena_poe.pl
URI: http://localhost:10080/?session=ping2hatena

% ./ping_client.pl "http://localhost:10080/?session=ping2hatena" \
  "Blog Developer's Cookbook" "http://blog.bulknews.net/cookbook/"

flerror=0

successful request to http://a.hatena.ne.jp/check?url=http%3A%2F%2F
blog.bulknews.net%2Fcookbook%2F at ./ping_to_hatena_poe.pl line 79.

#!/usr/local/bin/perl -w
# ping_to_hatena_poe
# - Weblogs.Com Ping Gateway to Hatena Antenna (POE)

use strict;
use POE qw(Component::Server::XMLRPC
           Component::Client::HTTP
           Component::TSTP);
use POE::Sugar::Args;
use HTTP::Request;
use Sys::Hostname;
use URI;

our $VERSION    = "1.00";
our $HatenaURL  = "http://a.hatena.ne.jp/check";

my $port = shift || 10080;

# session to trap Ctrl-Z
POE::Component::TSTP->create();

# session for XML-RPC server
POE::Component::Server::XMLRPC->new(
    alias => 'xmlrpc',
    port  => $port,
);

# session to send pings in HTTP
POE::Component::Client::HTTP->spawn(
    Alias => 'ua',
    Agent => "ping2hatena/$VERSION",
);

# main session
POE::Session->create(
    inline_states => {
        _start => \&setup_service,
        _stop  => \&shutdown_service,
        "weblogUpdates.ping" => \&ping_handler,
        got_response => \&response_handler,
    },
);

$poe_kernel->run();

sub setup_service {
    my $poe = sweet_args;
    $poe->kernel->alias_set("ping2hatena");
    $poe->kernel->post(xmlrpc  => 'publish',
                       ping2hatena => "weblogUpdates.ping");
    my $hostname = Sys::Hostname::hostname || 'localhost';
    warn "URI: http://$hostname:$port/?session=ping2hatena\n";
}

sub shutdown_service {
    my $poe = sweet_args;
    $poe->kernel->post(xmlrpc  => 'rescind',
                       ping2hatena => "weblogUpdates.ping");
}

sub ping_handler {
    my $poe = sweet_args;
    my $transaction = $poe->args->[0];
    my($blog_name, $blog_url) = @{$transaction->params()};
    my $url = URI->new($HatenaURL);
       $url->query_form(url => $blog_url);
    my $req = HTTP::Request->new(GET => $url);
    $poe->kernel->post(ua => request => got_response => $req);

    my %data = (flerror => XMLRPC::Data->type(boolean => 0));
    $transaction->return(\%data);
}

sub response_handler {
    my $poe = sweet_args;
    my $request  = $poe->args->[0]->[0];
    my $response = $poe->args->[1]->[0];
    if (defined($response->content)) {
        warn "successful request to ", $request->uri();
    } else {
        warn "error while requesting ", $request->uri();
    }
}

#!/usr/local/bin/perl -w
# ping_client - weblogUpdates.ping client

use strict;
use XMLRPC::Lite;
my($proxy, @args) = @ARGV;

my $result = XMLRPC::Lite
    ->proxy($proxy)
    ->call("weblogUpdates.ping", @args)
    ->result;
print "flerror=$result->{flerror}\n";