collectBuffer.ts

import { Component, OnInit } from '@angular/core';
import { Subject, Observable } from 'rxjs';
@Component({
  selector: 'app-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.css']
})
export class AppComponent implements OnInit {
  title = 'rxtest';

  collectBuffer<T>(items: T[]) {
    return (source: Observable<T>) => {
      const buffer = new Array<T>();
      const result = new Subject<T[]>();
      source.subscribe({
        next: (value: T) => {
          buffer.push(value);
          let check = true;
          for (const item of items) {
            check = check && buffer.indexOf(item) > -1;
          }
          if (check) {
            const output = [];
            for (const item of items) {
              const findIndex = buffer.indexOf(item);
              if (findIndex > -1) {
                output.push({ index: findIndex, value: item });
                buffer.splice(findIndex, 1);
              }
            }

            result.next(output.sort((a, b) => a.index - b.index).map(x => x.value));
          }
        },
        error: (error: any) => {
          result.error(error);
        },
        complete: () => {
          result.complete();
        }
      });
      return result;
    };
  }
  ngOnInit(): void {
    const ss = new Subject<number>();

    ss.pipe(
      this.collectBuffer([1, 2, 3])
    ).subscribe(x => {
      console.log(x);
    });

    ss.next(2);
    ss.next(1);
    ss.next(1);
    ss.next(3);
    ss.next(2);
    ss.next(3);
  }
}