`createIteratorSubject` allows someone to create an observable out of a generator and tell RxJS when to pull the next value instead of relying on the official implementation which creates backpressure.

createIteratorSubject.js

const { BehaviorSubject, Subject } = require('rxjs')
const { map, tap, withLatestFrom } = require('rxjs/operators')

const createIteratorSubject = (
  iterator,
) => {
  const iterator$ = (
    new BehaviorSubject()
  )

  const pushNextValue = ({
    done,
    value,
  }) => {
    done
    && value === undefined
    ? (
      iterator$
      .complete()
    )
    : (
      iterator$
      .next(value)
    )
  }

  iterator$
  .push = (
    value,
  ) => (
    pushNextValue(
      iterator
      .next(value)
    )
  )

  iterator$
  .push()

  return iterator$
}

module.exports = createIteratorSubject

Hi, I've gone through your blog post that led me here.
Your approach to backpressure is probably what I need in my use case:
I have the following typescript code where I would like to add backpressure control:

private emitter!: Observable<any>;
private rxSubscription!: RxSubscription;

// Connect the bluetooth event to an RxJS emitter
this.emitter = fromEventPattern(this.addHandler, this.removeHandler, (err, char) => [err, char]);

// save all incoming data to disk (unfiltered)
this.rxSubscription = this.emitter.subscribe(this.handleUpdateValueForDiskStorage);

What would be the proper way to use your approach in my code?
How to turn my observable to an iterator to call createIteratorSubject on it?

The second challenge is that I have a child subscription.
Should I call push on the iterator in the child subscription, on in the parent subscription?

Thanks in advance.

I found the above snippet difficult to read. Consider using a different name for subject, as well as a more compact format overall. For example:

const { BehaviorSubject } = require('rxjs')

function createIteratorSubject(iterator) {
  const subject$ = new BehaviorSubject();

  subject$.push = function(value) {
    const {done, value} = iterator.next(value);.

    if (done && value === undefined) {
        subject$.complete();
    } else {
        subject$.next(value);
    } 
  }

  subject$.push()

  return subject$
}

module.exports = createIteratorSubject

What an interesting code format, do you have an auto formatter for that?

@anilanar I'd love to have an auto-formatter and will be looking into using ESLint rules to achieve it (if that's even possible).

Unlike Prettier, there are a set of rules I follow so it's easy for someone else to replicate and get the same results. At my last job, you'd be unable to tell my code apart from my coworkers'. In fact, there was one bug we ended up fixing in separate PRs with the exact same solution line-by-line.

@anilanar I now have an eslint extension available.