RXJS实现首个元素触发缓存、超时/达最大容量关闭的批处理调用问题

需求描述 #

我想要创建一个id字符串的缓存，该缓存满足两个关闭条件：

• 达到我设定的时长后超时关闭
• 达到我设定的最大缓存容量后关闭

只要任一条件触发缓存关闭，我就需要将收集到的id字符串数组传递给另一个函数。我是RXJS新手，下方是我想要实现的数据传递逻辑的基础代码：

class TestClass {
  private ids: Array<string>;

  private id$: Subject<string> = new Subject();
  private buffered$: Observable<Array<string>>;
  private timer$: Observable<number>;

  constructor() {
    this.ids = [];

    this.id$ = new Subject<string>();
    this.id$.subscribe(); 

    this.timer$ = interval(2000); // timeoutTimer

    this.buffered$ = this.id$.pipe(buffer(this.timer$));
    this.buffered$.subscribe((idBuffer) => console.log("Id buffer: " + idBuffer));
  }

  fun1(id: string, x: number) {
    this.ids.push(id);
    this.id$.next(id); // Source data to be Observed and buffered 

    return new Promise((reslove) => {
      // I want to call fun2 here with 
      //  1. array from buffer Observable
      //  2. the x value from this function
      resolve(fun2(buffered, x))
    }); 
  }

  fun2(ids: Array<string>, x: number) {
    console.log('In function 2!');
  }
}

let test = new TestClass();

setInterval(() => {
  test.fun1('hey' + Math.floor(Math.random() * 10), Math.random());
}, 300);

更新说明 #

我更新了伪代码来描述我遇到的问题以及解决方案的预期效果：

问题描述 #

我会多次调用fun1往缓存中添加内容，期望达到我指定的时间限制或最大缓存容量时才调用fun2，同时每次调用fun1都需要返回fun2返回的对应数据对象。我需要实现调用批处理，而非每次调用fun1都触发fun2执行，等待缓存关闭或超时后一次性返回所有数据。

我需要找到一种实现方式：仅在第一次调用fun1时启动订阅，等待缓存执行完成后返回所有收集的元素，待缓存清空后，下一次往缓存中新增第一个元素时再重新启动订阅。

/*
   * PROBLEM: I want this to run only for the first addition of an ID string to a new buffer
   * but currently it is returning the new Promise<>... after every call to fun1! This results in
   * repeating elements being returned from the Promise
   *
   * Example:
   *
   * Buffer = ["id1", "id2", "id3", "id4", "id5"] // We called fun1 five times with these respective IDs
   *
   * // We are assuming the buffers max out at 3 elements
   * (1) idBuffer = ["id1", "id2", "id3"] // After first emission of bufferedIdAccumulator$
   * (2) idBuffer = ["id2", "id3", "id4"] // After second emission of bufferedIdAccumulator$
   * (3) idBuffer = ["id3", "id4", "id5"] // After third emission of bufferedIdAccumulator$
   *
   * What I Want:
   * (1) idBuffer = ["id1", "id2", "id3"] // After first emission of bufferedIdAccumulator$
   * (2) idBuffer = ["id1", "id2", "id3"] // After second emission of bufferedIdAccumulator$
   * (3) idBuffer = ["id1", "id2", "id3"] // After third emission of bufferedIdAccumulator$
   *
   * This is becasuse the first three calls to fun1 contain the first three objects I need to return so I
   * only want to return once for all three of these objects
   *
   */
  fun1(id: string, x: number): Promise<DataClass> {
    // console.log("In Function 1!");

    this.ids.push(id);
    this.id$.next(id); // Source data to be Observed and buffered

    console.log(`Adding id: ${id} to buffer!`);

    return new Promise<DataClass>(async (resolve) => {
      // I want to call fun2 here with
      //  1. array from buffer Observable
      //  2. the x value from this function
      let idBuffer = await firstValueFrom(this.bufferedIdAccumulator$);
      // let idBuffer = await new Promise<Array<string>>((resolve) => {
      //   this.bufferedIdAccumulator$.subscribe((idBuffer) => {
      //     console.log(idBuffer);
      //     resolve(idBuffer);
      //   });
      // });
      console.log(`<fun1> idBuffer: ${idBuffer}`);
      let dataClasses = this.fun2(idBuffer, x);
      let dataClass = dataClasses[0];
      resolve(dataClass);
    });
  }

  fun2(ids: Array<string>, x: number): Array<DataClass> {
    // console.log('In function 2!');
    // console.log('\nFunction 2:');
    // console.log(ids);

    let dataClasses = [];
    ids.forEach((id) => {
      dataClasses.push(new DataClass(id, x));
    });
    return dataClasses;
  }
}

class DataClass {
  id: string;
  x: number;

  constructor(id: string, x: number) {
    this.id = id;
    this.x = x;
  }
}

let test = new TestClass();

let x = 0;
setInterval(() => {
  test.fun1('hey' + x++, Math.random()); // Math.floor(Math.random() * 10), Math.random()
}, 2500);

实现方案 #

你可以用RxJS的race操作符实现双触发条件的竞争，同时维护待resolve的Promise队列对应每次fun1的调用请求，具体实现如下：

import { Subject, buffer, race, interval, take, skip } from 'rxjs';

class TestClass {
  // 可配置参数
  private MAX_BUFFER_SIZE = 3;
  private BUFFER_TIMEOUT = 2000;

  private id$ = new Subject<{id: string, x: number}>();
  // 存储当前批次所有待resolve的回调
  private pendingResolvers: Array<(data: DataClass) => void> = [];

  constructor() {
    this.initBufferHandler();
  }

  private initBufferHandler() {
    this.id$.pipe(
      buffer(
        // 两个触发条件竞赛，先触发的就关闭当前缓存
        race(
          // 条件1：超时触发
          interval(this.BUFFER_TIMEOUT),
          // 条件2：达到最大容量触发，跳过第一个元素计数，满MAX_BUFFER_SIZE-1个就触发
          this.id$.pipe(skip(1), take(this.MAX_BUFFER_SIZE - 1))
        )
      )
    ).subscribe(bufferList => {
      // 批次收集完成，调用fun2
      const ids = bufferList.map(item => item.id);
      // 这里用第一个请求的x参数，你可以根据需求调整x的取值逻辑
      const resultList = this.fun2(ids, bufferList[0].x);
      // 依次返回对应每个请求的结果
      this.pendingResolvers.forEach((resolve, index) => resolve(resultList[index]));
      // 重置状态等待下一批次
      this.pendingResolvers = [];
      this.initBufferHandler();
    });
  }

  fun1(id: string, x: number): Promise<DataClass> {
    return new Promise(resolve => {
      this.pendingResolvers.push(resolve);
      this.id$.next({id, x});
    })
  }

  fun2(ids: Array<string>, x: number): Array<DataClass> {
    console.log('触发fun2，当前批次id列表：', ids);
    return ids.map(id => new DataClass(id, x));
  }
}

class DataClass {
  id: string;
  x: number;
  constructor(id: string, x: number) {
    this.id = id;
    this.x = x;
  }
}

// 测试代码
const test = new TestClass();
let count = 0;
setInterval(() => {
  const curId = 'hey' + count++;
  test.fun1(curId, Math.random()).then(data => {
    console.log(`fun1返回结果，当前id=${curId}，对应数据：`, data);
  });
}, 300);

上述实现完全匹配需求：

1. 任意批次满足「超时2秒」或「满3个元素」任一条件就自动关闭，触发fun2执行
2. 同一批次的所有fun1调用都会拿到对应自身id的返回结果，无重复无遗漏
3. 批次处理完成后自动重置状态，下一次fun1调用会自动开启新的批次
4. 无需手动管理订阅的启动和销毁，完全由流自动控制生命周期

内容的提问来源于stack exchange，提问作者Alex Thomas