RXJS start buffer when a new element is added and it is the first

user_9314755 0
Viewed 157

There is an Update Below

I am trying to create a buffer of id strings. I want this buffer to close on two predicates:

  1. times out after a set amount of time that I decide
  2. it reaches the max buffer size that I decide

Once the buffer closes on either of those two predicates, I want it to pass the array of id strings it collected over to another function. I have code below is the exact setup of how I want to pass data but I am new to 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);

UPDATE:

I have updated Pseudo Code describing a problem I am facing and how to solve it.

Problem: Essentially I call fun1 multiple times to add things into a buffer, and I want to call fun2 once I have reached a time limit I specify or max buffer size! I then want fun1 to return the corresponding data object that fun2 returns. So I am batching my calls instead of calling fun2 every time I call fun1, I want to wait for the buffer or timeout and return all of the data at once!

I need to find a way to only subscribe the FIRST time fun1 is called and then wait for it's completion to return all the elements, and then once the buffer is empty again subscribe the next FIRST element to be added into the buffer

/*
   * 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 javascript
1 Answers

I think I found a working solution! I am posting the example code I have here if anyone wants to take a look!

import {
  Observable,
  Subject,
  firstValueFrom,
  Subscription,
} from 'rxjs';
import {
  bufferCount,
  filter,
  take,
  bufferTime,
  raceWith,
} from 'rxjs/operators';

class TestClass {
  private timeoutTime = 7000; // ms
  private maxBatchSize = 5; // ids

  private ids: Array<string>;

  private id$: Subject<string> = new Subject();
  private bufferedTime$: Observable<Array<string>>;
  private bufferedCount$: Observable<Array<string>>;
  private bufferedIdAccumulator$: Observable<Array<string>>;
  private idBufferSubscription$: Subscription;
  private firstBufferFromAccumulator: Promise<any>;

  constructor() {
    this.ids = [];

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

    this.bufferedTime$ = this.id$.pipe(bufferTime(this.timeoutTime));
    this.bufferedCount$ = this.id$.pipe(
      bufferCount(this.maxBatchSize),
      filter((arr) => !!arr.length) // don't emit empty array
    );

    // returns the emission of the Observable that emitted firstChild
    this.bufferedIdAccumulator$ = this.bufferedCount$.pipe(
      raceWith(this.bufferedTime$),
      take(1) // takes 1 of the buffer
    );

    this.idBufferSubscription$ = null;
  }

  fun1(id: string, x: number): Promise<DataClass> {
    // console.log('In Function 1!');

    // Check if this.idBufferSubscription$ is null or a closed subscription
    if (this.idBufferSubscription$ === null || this.idBufferSubscription$.closed) {
      console.log("idBufferSubscription$ null or closed!");
      this.idBufferSubscription$ = this.bufferedIdAccumulator$.subscribe();
      this.firstBufferFromAccumulator = firstValueFrom(this.bufferedIdAccumulator$);
    }

    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>((resolve) => {
      // I want to call fun2 here with
      //  1. array from buffer Observable
      //  2. the x value from this function
      this.firstBufferFromAccumulator.then((idBuffer) => {
        console.log(`<fun1>(2) idBuffer: ${idBuffer}`);
        let dataClasses = this.fun2(idBuffer, x);
        let dataClass = dataClasses[0];
        resolve(dataClass);
      });
    });
  }

  fun2(ids: Array<string>, x: number): Array<DataClass> {
    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);

This will subscribe to the Observer only when it is the first element being added into the buffer, and I have another variable firstBufferFromAccumulator which will return a Promise of the string[] returned from the bufferedIdAccumulator$. I store this as a variable so I can return it later in the Promise within fun1 so that it returns the same String array for every call to fun1 in the buffer! Hope this helps!

user_93147550
Related