実践DDD ch.4.7の イベント駆動アーキテクチャをRxJSで実装してみたやつ

event-driven-arch.ts

/**
 * 実践DDD cp.4.7 のイベント駆動アーキテクチャの実装を、RxJSをつかってやってみたやつ.
 */

import { Subject, Observable, Observer } from 'rxjs';
import { count, filter } from 'rxjs/operators';

/**
 * ダミーデータストリームの作成.
 */
class PhoneNumberPublisher extends Observable<string> {

  subscriber: Observer<string>;

  constructor() {
    super((subscriber) => {
      this.subscriber = subscriber;
    });
  }

  publish() {
    [
      '123-4567-8901',
      '303-4567-8901',
      '123-4303-8901',
      '123-4563-0301',
    ].forEach(v => this.subscriber.next(v))

    this.subscriber.complete();
  }
}

function _main() :void {
  /**
   * 電話番号の読み出し->部分一致する文字列を抽出->マッチ数をカウント
   * をおこなうパイプラインを構築.
   * 
   * いちいちSubjectに書きださなくてもRxJSではpipeだけで実現できそうではあるのだが、
   * パイプラインらしい、分岐や組み替えに対応するために結果ごとにSubjectにしている.
   * 
   */
  const phoneNumberPublisher = new PhoneNumberPublisher();
  const allPhoneNumberListed = new Subject<string>();
  const phoneNumberFinder = filter<string>(v => v.match(/303/) != null);
  const phoneNumberMatched = new Subject<string>();
  const matchedPhoneNumberCounter = count();
  const matchedPhoneNumberCounted = new Subject<number>();
  const phoneNumberExecutive = (v: any) => console.log(v);

  phoneNumberPublisher.subscribe(allPhoneNumberListed);
  allPhoneNumberListed.pipe(phoneNumberFinder).subscribe(phoneNumberMatched);
  phoneNumberMatched.pipe(matchedPhoneNumberCounter).subscribe(matchedPhoneNumberCounted);
  matchedPhoneNumberCounted.subscribe(phoneNumberExecutive);

  phoneNumberPublisher.publish();
}

_main();

long-time-saga.ts

/**
 * 実践DDD cp.4.7 のイベント駆動アーキテクチャの実装を、RxJSをつかってやってみたやつ.
 */

import { Subject, Observable, Observer, zip } from 'rxjs';
import { count, filter } from 'rxjs/operators';

/**
 * ダミーデータストリームの作成.
 */
class PhoneNumberPublisher extends Observable<string> {

  subscriber: Observer<string>;

  constructor() {
    super((subscriber) => {
      this.subscriber = subscriber;
    });
  }

  publish() {
    [
      '123-4567-8901',
      '303-4567-8901',
      '123-4303-8901',
      '123-4563-0301',
    ].forEach(v => this.subscriber.next(v))

    this.subscriber.complete();
  }
}

function _main() :void {
  /**
   * 電話番号の読み出し->部分一致する文字列を抽出->マッチ数をカウント
   * をおこなうパイプラインを構築.
   * また、電話番号の読み出し後、全数カウントするパイプラインと分岐し、
   * 最後に結果を一緒に表示する.
   * 
   */
  const phoneNumberPublisher = new PhoneNumberPublisher();
  const allPhoneNumberListed = new Subject<string>();
  const totalPhoneNumberCounter = count();
  const allPhoneNumberCounted = new Subject<number>();
  const phoneNumberFinder = filter<string>(v => v.match(/303/) != null);
  const phoneNumberMatched = new Subject<string>();
  const matchedPhoneNumberCounter = count();
  const matchedPhoneNumberCounted = new Subject<number>();
  const phoneNumberExecutive = (a: any) => console.log(a[1] + ' items are matched in ' + a[0] + ' items.');

  phoneNumberPublisher.subscribe(allPhoneNumberListed);
  allPhoneNumberListed.pipe(totalPhoneNumberCounter).subscribe(allPhoneNumberCounted);
  allPhoneNumberListed.pipe(phoneNumberFinder).subscribe(phoneNumberMatched);
  phoneNumberMatched.pipe(matchedPhoneNumberCounter).subscribe(matchedPhoneNumberCounted);

  zip(allPhoneNumberCounted, matchedPhoneNumberCounted).subscribe(phoneNumberExecutive);
  /**
   * zipで、全数カウントのpipelineの完了とマッチ数カウントのpipelineの完了を待ち合わせているが、
   * 本の説明の趣旨としてはちょっと違うか？
   * 
   * マルチスレッドでほぼ同時にデータ読み出しが発生した場合に、取り違えないしくみにしなければいけないような気がするが、、
   */

  phoneNumberPublisher.publish();
}

_main();