AnyEvent で並列ダウンローダを作ってみた。

Youtube からのダウンロードは以下のような 2 ステップに分かれます。

HTML から flv ファイルの URL を抽出
実際に flv をダウンロード

こういう形でいくつかのステップに分かれる場合に AnyEvent ではどう書くのかを検証したかったこともあり、チャレンジしてみました。

結論から言うと、

イベント発生
コールバック(1) の呼び出し
1. コールバック(1) の中でコールバック(2) の設定
2. コールバック(2) の呼び出し
  1. ....

というようにコールバックを連鎖させることで、上記のようなステップが分割されるようなケースに対応することができそうです(本当はもっとスマートな方法があるのかもしれませんが)。AnyEvent の POD 中にある "REAL-WORLD EXAMPLE" で Net::FCP の説明がなされていますが、ここでの AnyEvent の使い方が非常に参考になりました。
# AnyEvent はドキュメントが非常に充実しており、目を通す度に新しい発見があります。

とりあえず動くという程度ではありますが、プログラム骨子は以下の通り、flv ダウンロード + ffmpeg による変換の 3 ステップにチャレンジしてみました。

http://www.youtube.com/watch?v=xxxxxxxxxxx という URL から HTML をダウンロード
1. のダウンロード終了後、コールバックで flv の URL を抽出し、flv をダウンロード
2. のダウンロード後、コールバックで ffmpeg を実行

HTTP アクセス

HTTP によるアクセスについては、AnyEvent::HTTP を利用しました。AnyEvent::HTTP はローレベルという話は聞いてたんですが、単純利用だけを考えればコールバックの設定のみで十分利用できそうでした*1。

外部コマンド実行

AnyEvent::Util に外部コマンド実行用の run_cmd が用意されていたので、それをそのまま利用しています。

Coro

最初は Coro を使ってということを考えてたんですが、Coro マジよくわからない。

my $cv = AnyEvent->condvar;

$cv->begin;
for my $url (@urls) {

    $cv->begin;
    $watchers{$url} = http_get( $url => sub {
         my ($body, $headers) = @_;
         $body or do {
             warn "error, $headers->{Status} $headers->{Reason}\n";
             $cv->end; return;
         };
         my $map   = construct_map( $body );   # FLV の URL 抽出
         my $title = title( $body );           # タイトル抽出
         my $flv = $map->{18} || $map->{35} || $map->{34} || $map->{22} || $map->{6}; # flv ファイルの URL

         my $tmp = "$title.flv";
         open my $fh, '>', $tmp or do { $cv->end; return; };
         $watchers{$url} = http_get( $flv,
             on_body => sub {
                 my ($body, $headers) = @_;
                 print {$fh} $body;
             },
             sub {
                 my @cmd = split /\s+/ => "/opt/local/bin/ffmpeg -i $tmp -f mp4 -y -ac 2 -ab 128k ./$tmp.m4a";
                 my $cv2 = run_cmd( \@cmd, '>' => '/dev/null', '2>' => '/devnull' );
                 $cv2->cb( sub { shift->recv; $cv->end; } );
             }
         );
     });
}

$cv->end;
$cv->recv;