非同期処理をシーケンシャルに扱うために - snippets from shinichitomita’s journal

http://hail2u.net/blog/coding/synchronous-requests-to-jsonp.html

forループなら並列リクエストでも工夫次第でできると思うけど、前後的な依存関係が各リクエストにある場合は、やっぱり直列処理をしなければいけない。

2、3段くらいの直列処理ならコールバック関数を連鎖させて書いたり、インラインで無名関数指定したりしても困らないと思うけど、もっと多段階の直列処理をコールバック関数名の指定だけで記述しようとすると、コードを書く人でも頭の中にちゃんとフローのイメージが出来上がってないと厳しいし、コードを読む側はもっとこんがらがることになる。

たとえば

del.icio.usからnetwork情報をJSONPで取得(http://del.icio.us/feeds/json/network/stomita)
del.icio.usからfan情報をJSONPで取得(http://del.icio.us/feeds/json/fans/stomita)
上記結果から mutual connection（networkとfanの双方に登録されているIDリスト）を抽出
del.icio.usから自分自身のtagリストをJSONPで取得(http://del.icio.us/feeds/json/tags/stomita)
自分自身がもっとも頻繁にtagしているタグで絞り込んで、mutual connectionのpostsをすべてJSONPで取得
取得したpostsの中からランダムに一件選択
選択されたURLのはてブコメントをJSONPで取得
はてブコメントをMECAPI API(JSONP)で形態素解析
さらに形態素解析されたキーワードを元にYahoo! Web Search JSONPで検索実行

みたいな一連の処理*1を書くのは、適切に関数で分けたりしても結構つらいし、そもそも全部に目を通さないと流れが理解できない。

つまり、非同期処理であっても、できるだけシーケンシャルな記述でちゃんと流れを理解できるようにコーディングしたいなあ、と思っていた。

ということで、最近こんなのを書いてみた。

SequentialAction

機能としては、処理を記述した関数の配列をあらかじめ与えておくと、その処理を配列に指定された順に順次実行していくというだけのもの。

ただし、非同期処理のために（擬似的に）順次実行プロセスを停止するsuspend()、および停止されたプロセスを再び開始するresume()を定義している。

各関数の間でのデータ受け渡しは、SequentialActionオブジェクトにcontextというプロパティを定義しており、これを介して行う。

使い方はこんな感じで。ここではdel.icio.usネットワークに登録されている全IDのポストをフェッチしてくる処理を、シーケンシャルに実装している。

http://stomita.web.fc2.com/seqtest.html

var actions = [

  // del.icio.usのnetwork情報をフェッチ
  function() {
    var url = 'http://del.icio.us/feeds/json/network/'+this.context.username;
    JsonWebServicesStub.invoke(url, this.resume.bind(this)); // JSONP呼び出し、非同期処理
    this.suspend(); // プロセスを一旦停止、非同期処理待ち
  }
  ,

  // 取得したnetwork情報を変数にストア
  function() {
    this.context.network = this.context.returned; // 直前のJSONPリクエストでの戻り値
  }
  ,

  // del.icio.us postsのフェッチのためのループのスタート
  function fetchPostsLoopStart() {
    var user = this.context.network.pop();
    if (user) {
      this.context.postUser = user;
    } else {
      this.goto('renderPosts'); // ループ終了、レンダリングへ
    }
  }
  ,

  // 指定されたユーザのdel.icio.us postsをフェッチ
  function() {
    var url = 'http://del.icio.us/feeds/json/'+this.context.postUser;
    JsonWebServicesStub.invoke(url, this.resume.bind(this)); // JSONP呼び出し、非同期処理
    this.suspend(); // プロセスを一旦停止、非同期処理待ち
  }
  ,

  // 取得したpostsを全体にマージ
  function() {
    var posts = this.context.returned; // 直前のJSONPリクエストでの戻り値
    this.context.allPosts = this.context.allPosts ? this.context.allPosts.concat(posts) : posts;
    this.goto('fetchPostsLoopStart'); // ループ開始へ戻る
  }
  ,

  // postsをレンダリング
  function renderPosts() {
    var template = TrimPath.parseTemplate($('postsViewTemplate').value)
    $('postsView').innerHTML = template.process({ posts : this.context.allPosts });

    // for benchmark;
    alert(new Date().getTime() - this.context.startTime + 'msec');
  }

]; // end of action definitions

var seq = new SequentialAction(actions);
seq.start({ username : 'stomita', startTime : new Date().getTime() });

シーケンシャルといいながら、gotoもサポートしてループ処理させてたり。

たぶんこれをもっと発展させると、分岐や並列実行待ちも含めて、ステートマシーンみたいな実装になっちゃうかも。ただ、そこまでくると今度はコード記述がわかりにくくなるな、と思ってるので、たぶんやらない。

（追記）
やっぱり並列実行待ちの機能も作ってみた。

var actions = [

  function() {
    var url = 'http://del.icio.us/feeds/json/network/'+this.context.username;
    JsonWebServicesStub.invoke(url, this.resume.bind(this));
    this.suspend(); 
  }
  ,

  function() {
    this.context.network = this.context.returned;
    for (var i=0; i<this.context.network.length; i++) {
      var postUser = this.context.network[i];
      var url = 'http://del.icio.us/feeds/json/'+postUser;
      var subproc = this.fork('sp'+i); // （疑似）サブプロセスを新規作成
      JsonWebServicesStub.invoke(url, subproc.join); // 非同期リクエスト終了時に作成したサブプロセスを回収
    }
    this.wait(); // サブプロセスの実行終了待ち
  }
  ,

  function() {
    var posts = [];
    for (var i=0; i<this.context.network.length; i++) {
      posts = posts.concat(this.context.returned['sp'+i]); // サブプロセスごとのJSONP実行結果をマージ
    }

    var template = TrimPath.parseTemplate($('postsViewTemplate').value)
    $('postsView').innerHTML = template.process({ posts : posts });

    // for benchmark;
    alert(new Date().getTime() - this.context.startTime + 'msec');
  }

]; // end of action definitions

var seq = new SequentialAction(actions);
seq.start({ username : 'stomita', startTime : new Date().getTime() });

*1:この処理自体に意味があるかは知らないけど、こんなmashupプロセスも今や全部ブラウザ側でできちゃうのだから、面白い