Potential solution for diamond problem/glitches - close but need help!

[anthonyjoeseph]

I'm trying to write an operator batchSync that behaves like so:

declare const batchSync: <A>() => OperatorFunction<A, A[]>

const syncThenAsync = r.merge(
  r.of(0, 1, 2), 
  r.of(null).pipe(r.delay(0), r.mergeMap(() => r.of(3, 4, 5)))
);
syncThenAsync.pipe(batchSync()).subscribe(console.log);
// [0, 1, 2]
// [3, 4, 5]

Batching "glitches" together (as defined here)

It turns out to be trivial to write a version of that merges cold, synchronous emissions - code:

const batchColdSync: <A>() => OperatorFunction<A, A[]> = ...

syncThenAsync.pipe(batchColdSync()).subscribe(console.log);
// [0, 1, 2]
// [3]
// [4]
// [5]

The underlying idea is that cold subscriptions will block until all of their synchronous emissions have completed.

But I can't seem to make the leap into a full-on batchSync() implementation

I've tried using share() to turn cold observables hot, ReplaySubject to turn hot observables cold, and the queueScheduler to turn recursive subscriptions iterative

I feel like I'm so close. If I can batch synchronous cold emissions, and I can turn synchronous hot emissions into synchronous cold emissions, why can't I batch synchronous hot emissions?

Am I running into a hard boundary of rx/rxjs/javascript that I just can't see?

P.S. I don't want to use the asapScheduler or anything like that - I'd like to preserve the synchronous nature of the emissions if possible