How To Delay Event Emission With Rxpy/rxjs?

I've got two event streams. One is from an inductance loop, the other is an IP camera. Cars will drive over the loop and then hit the camera. I want to combine them if the events a

Solution 1:

You should be able to solve the problem using auditTime and buffer. Like this:

function matchWithinTime(a$, b$, N) {
  const merged$ = Rx.Observable.merge(a$, b$);
  // Use auditTime to compose a closing notifier for the buffer.
  const audited$ = merged$.auditTime(N);
  // Buffer emissions within an audit and filter out empty buffers.
  return merged$
    .buffer(audited$)
    .filter(x => x.length > 0);
}

const a$ = new Rx.Subject();
const b$ = new Rx.Subject();
matchWithinTime(a$, b$, 50).subscribe(x => console.log(JSON.stringify(x)));

setTimeout(() => a$.next("a"), 0);
setTimeout(() => b$.next("b"), 0);
setTimeout(() => a$.next("a"), 100);
setTimeout(() => b$.next("b"), 125);
setTimeout(() => a$.next("a"), 200);
setTimeout(() => b$.next("b"), 275);
setTimeout(() => a$.next("a"), 400);
setTimeout(() => b$.next("b"), 425);
setTimeout(() => a$.next("a"), 500);
setTimeout(() => b$.next("b"), 575);
setTimeout(() => b$.next("b"), 700);
setTimeout(() => b$.next("a"), 800);

.as-console-wrapper { max-height: 100% !important; top: 0; }

<script src="https://unpkg.com/rxjs@5/bundles/Rx.min.js"></script>

If it's possible for b values to be closely followed by a values and you do not want them to be matched, you could use a more specific audit, like this:

const audited$ = merged$.audit(x => x === "a" ?
  // If an `a` was received, audit upcoming values for `N` milliseconds.
  Rx.Observable.timer(N) :
  // If a `b` was received, don't audit the upcoming values.
  Rx.Observable.of(0, Rx.Scheduler.asap)
);

Solution 2:

I have developed a different strategy than Cartant, and clearly much less elegant, which may give you somehow a different result. I apologize if I have not understood the question and if my answer turns out to be useless.

My strategy is based on using switchMap on a$ and then bufferTime on b$.

This code emits at every timeInterval and it emits an object which contains the last a received and an array of bs representing the bs received during the time interval.

a$.pipe(
    switchMap(a => {
        return b$.pipe(
            bufferTime(timeInterval),
            mergeMap(arrayOfB => of({a, arrayOfB})),
        )
    })
)

If arrayOfB is empty, than it means that the last a in unmatched.

If arrayOfB has just one element, than it means that the last a has been matched by the b of the array.

If arrayOfB has more than one element, than it means that the last a has been matched by the first b of the array while all other bs are unmatched.

Now it is a matter of avoiding the emission of the same a more than once and this is where the code gets a bit messy.

In summary, the code could look like the following

const a$ = new Subject();
const b$ = new Subject();

setTimeout(() => a$.next("a1"), 0);
setTimeout(() => b$.next("b1"), 0);
setTimeout(() => a$.next("a2"), 100);
setTimeout(() => b$.next("b2"), 125);
setTimeout(() => a$.next("a3"), 200);
setTimeout(() => b$.next("b3"), 275);
setTimeout(() => a$.next("a4"), 400);
setTimeout(() => b$.next("b4"), 425);
setTimeout(() => b$.next("b4.1"), 435);
setTimeout(() => a$.next("a5"), 500);
setTimeout(() => b$.next("b5"), 575);
setTimeout(() => b$.next("b6"), 700);
setTimeout(() => b$.next("b6.1"), 701);
setTimeout(() => b$.next("b6.2"), 702);
setTimeout(() => a$.next("a6"), 800);


setTimeout(() => a$.complete(), 1000);
setTimeout(() => b$.complete(), 1000);


let currentA;

a$.pipe(
    switchMap(a => {
        currentA = a;
        return b$.pipe(
            bufferTime(50),
            mergeMap(arrayOfB => {
                let aVal = currentA ? currentA : null;
                if (arrayOfB.length === 0) {
                    const ret = of({a: aVal, b: null})
                    currentA = null;
                    return ret;
                }
                if (arrayOfB.length === 1) {
                    const ret = of({a: aVal, b: arrayOfB[0]})
                    currentA = null;
                    return ret;
                }
                const ret = from(arrayOfB)
                            .pipe(
                                map((b, _indexB) => {
                                    aVal = _indexB > 0 ? null : aVal;
                                    return {a: aVal, b}
                                })
                            )
                currentA = null;
                return ret;
            }),
            filter(data => data.a !== null || data.b !== null)
        )
    })
)
.subscribe(console.log);