[クラス設計] 時間のかかる処理を逐次的に行いたい

Question

前提・実現したいこと

入力データA[]をデータB[]に、さらにデータB[]をデータC[]に変換して、最終的にC[]を出力するという処理を行いたいです。ここで条件は以下の通りです。

• 開発はTypeScriptで行う
• データのサイズは大きく変換処理にも時間がかかるため、非同期で行う (実際はWebWorkerを用いる)
• ユーザーは変換処理中に関わらず任意のタイミングでデータAを何個でも入力することができる
• 処理順を保証する必要がある (入力順と出力順が同じ)
• データはいくつかまとめて処理をする (ex. 1000個のAを入力すると、100個ずつに分割されて順番に処理される)
• 変換A→ Bが全て終わってから変換B→ Cを行うのではなく、例えば変換A→ Bが100個終わったら変換B→ Cを100個...と逐次的に行う (データAの最初の入力時からデータCの最初の出力までのレスポンスをできるだけ短くしたいため)
• 変換B→ Cが処理されている間にも変換A→ Bを処理したい

このような場合、クラス設計(アーキテクチャ？)はどのようにしたら良いのでしょうか。将来的に機能の追加・修正が行われるため、できるだけ保守性の高いコードにしたいと考えています。

次の方法はどうか

初心者ながら次のような方法を考えたのですが、問題点などありますでしょうか。
継承の必要性についても知りたいです。

こちらにもコードを記載しています。Playground

// アウトプットのあるノード
interface ISourceNode<Out> {
  connect(nextNode: IDestinationNode<Out>): void
}

// インプットのあるノード
interface IDestinationNode<In> {
  post(data: In[]): void
}

interface IConverterNode<In, Out> extends ISourceNode<Out>, IDestinationNode<In> {
}

// データInを受け取ってOutに変換するような基底クラス
class ConverterNode<In, Out> implements IConverterNode<In, Out> {
  private queue: In[] = []
  private isConverting = false // 変換処理中か
  private batchSize: number // まとめて送るデータの数
  private nextNode!: IDestinationNode<Out>

  constructor(batchSize: number) {
    this.batchSize = batchSize
  }

  // ノードを繋げる
  connect(nextNode: IDestinationNode<Out>) {
    this.nextNode = nextNode
  }

  // データTを受け取る
  post(data: In[]): void {
    this.queue.push(...data); // 受け取ったデータをキューに保持

    // 変換中でなければ変換を開始する (順序を保証するため)
    if (!this.isConverting && this.queue.length >= this.batchSize) {
      const batch = this.queue.splice(0, this.batchSize)
      this.isConverting = true
      this.convert(batch)
    }
  }

  // 変換処理を行う
  protected convert(data: In[]): void {
  }

  // 変換されたデータを次のノード(IConverterNodeまたはIDestinationNode)に送る
  protected converted(data: Out[]) {
    this.nextNode.post(data)

    // キューにまだ変換できるデータが存在するならconvertする
    if (this.queue.length >= this.batchSize) {
      const batch = this.queue.splice(0, this.batchSize)
      this.convert(batch)
    } else {
      this.isConverting = false
    }
  }
}

type A = number
type B = number

// データAをデータBに変換するクラス
class ABConverter extends ConverterNode<A, B> {
  constructor(batchSize: number) {
    super(batchSize)
  }

  // 変換処理を行う
  protected convert(data: A[]): void {
    new Promise<B[]>(resolve => {
      // 何かしらの重い処理
      setTimeout(() => {
        resolve(data.map(a => a * 2))
      }, Math.random()*100)
    }).then(dataB => {
      // 変換されたらconvertedを呼ぶ
      this.converted(dataB)
    })
  }
}

// 入力ノード
class SourceNode implements ISourceNode<A> {
  private nextNode!: IDestinationNode<B>

  connect(nextNode: IDestinationNode<A>) {
    this.nextNode = nextNode
  }

  // データをConverterに送信
  send(data: A[]) {
    this.nextNode.post(data)
  }
}

// 出力ノード (変換されたデータCがここにたどり着く)
class DestinationNode implements IDestinationNode<B> {
  post(data: B[]) {
    console.log(data)
  }
}

const source = new SourceNode()
const ab = new ABConverter(3)
const ab2 = new ABConverter(3)
const dest = new DestinationNode()

// source -> ab -> ab2 -> dest とデータが流れる
source.connect(ab)
ab.connect(ab2)
ab2.connect(dest)

source.send([1, 2, 3, 4, 5, 6])

このコードの場合、入力[1, 2, 3, 4, 5, 6]を[1, 2, 3] [4, 5, 6]に分割し、それぞれに対して各要素を2倍するという変換を2回施しているため、コンソールには

[4, 8, 12]
[16, 20, 24]

と表示されます。

Comments

[maisumakun]

当該のコードを実行するのはブラウザ内ですか？サーバサイドですか？それとも、どちらでもない環境でしょうか？

[退会済みユーザー]

ブラウザ内です

[dameo]

ブラウザ内でやるなら、好きにやればいいのではないかと思いますが、c○in-hiveみたいな事件とかにはならないようにお願いしますね。何をやるにせよ、入出力と目的くらいは具体的に説明しないと設計し始めることも無理ではないかと思いますよ。今分かってるのはWebWorkerで並列実行してもいいってことだけなので。

Answer 1

A→B→Cの作業は直列の処理なので非同期化はこんなんです

A(A1,A2,A3)→B(B1,B2,B3)→Cのような各処理分割できるなら可能性はあります

Comments

[退会済みユーザー]

各処理はそのように分割出来るという前提です。

[yambejp]

流れ的にはこんな感じ

(async()=>{
await (async()=>{
return Promise.all([
new Promise(resolve=>setTimeout(()=>{console.log("A1");resolve();},300)),
new Promise(resolve=>setTimeout(()=>{console.log("A2");resolve();},500)),
new Promise(resolve=>setTimeout(()=>{console.log("A3");resolve();},200)),
]);
})();
await (async()=>{
return Promise.all([
new Promise(resolve=>setTimeout(()=>{console.log("B1");resolve();},500)),
new Promise(resolve=>setTimeout(()=>{console.log("B2");resolve();},200)),
new Promise(resolve=>setTimeout(()=>{console.log("B3");resolve();},300)),
]);
})();
console.log("C");
})();

[退会済みユーザー]

すみません、前提と質問内容を見返してから回答いただけるとありがたいです。

[yambejp]

逆に上記を理解ください
promose.allで非同期の並行処理を実行し、
async/awaitで直列につなげるということです

[退会済みユーザー]

理解しています。これだと変換処理中にデータAを追加することができませんし、そもそも逐次的な処理では無いと思いますがどうなのでしょうか

[yambejp]

ではなにが聞きたいのでしょうか？
非同期ではない処理を非同期に分割する方法ですか？

[退会済みユーザー]

クラス設計についてです

[yambejp]

ごめんなさい、なんか要領を得ないので私は撤退します
お役に立てず申し訳ありません
がんばってください

[退会済みユーザー]

こちらこそ不明瞭な文章で申し訳ありません