Node.js で依存関係をうまく制御するために、Pub/Sub パターンを使いたかった。調べて見つけた Rxjs の Subject を利用することで上手く実装できたと思う。Pub/Sub を使うと何が嬉しいのかと、Subject を使った実装をまとめてみる。

例には以下の FizzBuzz クラスを使う。300ms ごとに 1~15までカウントアップして、Fizz Buzz が表示されるだけ。

そして、このクラスに対して Fizz が表示される回数を数える ことを考えてみる。

class FizzBuzz {
    static async run(): Promise<void>{
        const sleep = (ms: number) => new Promise((r) => setTimeout(() => r(null), ms));
        for(let i=1; i<=15; i++){
            await sleep(300);
            if(i % 15 === 0){
                console.log("FizzBuzz");
                continue;
            }
            if(i % 3 === 0){
                console.log("Fizz");
                continue;
            }
            if(i % 5 === 0){
                console.log("Buzz");
                continue;
            }
            console.log(i);
        }
    }
}

FizzBuzz.run();

/*
1
2
Fizz
4
Buzz
Fizz
7
8
Fizz
Buzz
11
Fizz
13
14
FizzBuzz
*/

素直に実装してみる

FizzCounter を追加して、"Fizz" を出力するタイミングでインクリメントするという感じにしてみた。問題なく動いてそうですね。

const fizzCounter = new class FizzCounter {
    private _count = 0;

    get count(): number {
        return this._count
    }

    increaseCount(): void {
        this._count++;
    }
}();

class FizzBuzz {
    static async run(): Promise<void>{
        const sleep = (ms: number) => new Promise((r) => setTimeout(() => r(null), ms));
        for(let i=1; i<=15; i++){
            await sleep(300);
            if(i % 15 === 0){
                console.log("FizzBuzz");
                continue;
            }
            if(i % 3 === 0){
                console.log("Fizz");
                fizzCounter.increaseCount();  // 👈 追加
                continue;
            }
            if(i % 5 === 0){
                console.log("Buzz");
                continue;
            }
            console.log(i);
        }
    }
}

(async () => {
    await FizzBuzz.run();
    console.log(`> Fizz count is ${fizzCounter.count}`);
})();

/*
1
2
Fizz
4
Buzz
Fizz
7
8
Fizz
Buzz
11
Fizz
13
14
FizzBuzz
> Fizz count is 4
*/

正直、シンプルなプログラムであれば問題にはならないと思うが、以下のような点が気になる。

fizzCounter.increaseCount() を公開しているため他の箇所からでもアクセスでき count の不整合になる可能性ある。
FizzBuzz クラスが自分の処理には関係ない fizzCounter.increaseCount() に依存してしまっている（無駄な結合）。
fizzCounter.increaseCount() のインターフェースが変わったりすると一番重要な FizzBuzz クラスが壊れる可能性がある。

これを、以下のように直したいなと思った。

依存方向を FizzCounter --> FizzBuzz の方向に逆転させたい（安定方向への依存）。
fizzCounter が変化したとしても、FizzBuzz クラスに影響を与えないようにしたい（疎結合）。

Subject を使って書いてみる

Subject を使うことで Pub/Sub 的な形が作れる。Observable と似ているが、subscribe のタイミングや同時に publish できる数の制限はなくて融通が利く感じがする。

rxjs.dev

以下の様にすることで、依存方向が fizzCounter --> FizzBuzz になり、結合も疎になった（MQ を使った Pub/Sub には及ばないが、前のコードよりは疎結合だと思う）。

さらに、fizzCounter.increaseCount() は公開する必要がなくなったため非公開にすることができるようになった。やったー。
（テストのために公開したいということはあるかもしれない）

import { Subject } from "rxjs"

class FizzBuzz {
    static readonly fizzed = new Subject<void>();
    static async run(): Promise<void>{
        const sleep = (ms: number) => new Promise((r) => setTimeout(() => r(null), ms));
        for(let i=1; i<=15; i++){
            await sleep(300);
            if(i % 15 === 0){
                console.log("FizzBuzz");
                continue;
            }
            if(i % 3 === 0){
                console.log("Fizz");
                this.fizzed.next();  // 👈 Subject の next を呼び出す（Publish）
                continue;
            }
            if(i % 5 === 0){
                console.log("Buzz");
                continue;
            }
            console.log(i);
        }
    }
}

const fizzCounter = new class FizzCounter {
    private _count = 0;
    constructor() {
        // 👇 FizzBuzz.fizzed に Subscribe する。
        FizzBuzz.fizzed.subscribe(() => this.increaseCount());
    }

    get count(): number {
        return this._count
    }

    // 👇 private に変更した
    private increaseCount(): void {
        this._count++;
    }
}();

(async () => {
    await FizzBuzz.run();
    console.log(`> Fizz count is ${fizzCounter.count}`);
})();


/*
1
2
Fizz
4
Buzz
Fizz
7
8
Fizz
Buzz
11
Fizz
13
14
FizzBuzz
> Fizz count is 4
*/