RxJSのbindNodeCallbackの使い道をようやく発見した

(2018年11月6日 update: RxJS 6 の記法に合わせた)

RxJSのbindCallbackとかbindNodeCallbackとかいうObservableのcreate関数があります。ほとんどの人はその存在さえ知らないと思いますが。
どういう使い道があるんだろうと思うだけで全く使うことを考えてはいなかったのですが、理解してみたらそれなりに使い道があるということがわかったのでメモを残します。

今回はMySQLを叩くライブラリ( https://www.npmjs.com/package/mysql )と組み合わせてbindNodeCallbackを使ったサンプルコードを書きます。

bindNodeCallback

• bindNodeCallback(fs.readFile)のように使うとfs.readFileをObservableに変換しつつコールバックの第一引数を自動的にObserverのerrorにバインドする。
• bindNodeCallbackにより生成された関数の戻り値は[arg1, arg2, ...]のように配列で返される。
• 第一引数がエラー情報ではない普通の(?)コールバックをラップするときはbindCallbackを使う。
• DB操作系のライブラリと組み合わせるとすごく捗りそう。

以下イメージが伝わるつもりで書いたサンプルコードです。

index.ts

import * as mysql from 'mysql';
import { bindNodeCallback, Observable, of } from 'rxjs';
import { map, tap, concatMap } from 'rxjs/operators';

type QueryCallback = (query: string) => Observable<[Record<string, any>[], any]>;

const connection = mysql.createConnection({
  host: '127.0.0.1',
  user: 'user', // <= set your username
  password: 'password' // <= set your password
});
const connect$ = bindNodeCallback(connection.connect.bind(connection)); // <= !?
const query$ = bindNodeCallback(connection.query.bind(connection)) as QueryCallback; // <= !?
const end$ = bindNodeCallback(connection.end.bind(connection)); // <= !?

async function main() {
  console.log('start');
  await of(true) // falsyな値だとストリームが流れないのでtrueを入れただけ。
    .pipe(
      concatMap(() => connect$()),
      concatMap(() => query$('create database if not exists hoge;')),
      concatMap(() => query$('show databases;')),
      map(([results, fields]) => results),
      tap(results => console.log(results)),
      concatMap(() => end$())
    )
    .toPromise();
  console.log('finished');
}

main().catch(console.error);

出力はこのようになります。

OUTPUT

start
[ RowDataPacket { Database: 'information_schema' },
  RowDataPacket { Database: 'hoge' },
  RowDataPacket { Database: 'mysql' },
  RowDataPacket { Database: 'performance_schema' },
  RowDataPacket { Database: 'sys' } ]
finished

{ Database: 'hoge' }が作成されたことがわかります。

まとめ

Observableに変換するためにconnection.connect.bind(connection))のように関数をbindして与える必要がありますが、それでも一行で書けてエラーハンドリングまで済んでいるというのが感動的です。

使ってみたら超絶便利だったので特にDBアクセス周りでどんどん使ってみようと思いました。