Catch circular dependency between observables

问题:

I have a user-programming scenario where user can end up creating two observables that depend on each other. RxJS does not allow circular dependencies, as far as I can see, the memory or stack reaches its limits and the onError callback is triggered with the value true.

How to detect the circular dependency explicitly and throw a more descriptive error message?

This codes illustrates how to create a circular dependency in RxJS:

var obsA,
    obsB;

obsA = Rx.Observable
    .returnValue(42)
    .combineLatest(obsB, function (a, b) {
        return a + b;
    });

obsB = Rx.Observable
    .returnValue(42)
    .combineLatest(obsA, function (b, a) {
        return b + a;
    });


obsA
    .subscribe(function (val) {
        console.log('onNext:' + val);
    },
    function (err) {
        console.error('onError: ' + err);
    },
    function () {
        console.log('onCompleted');
    });

The error message is simply true.

回答1:

The code in the original question does not create a circular dependency. At the time you define ObsA, ObsB is undefined and so what you've really done is call combineLatest(undefined, function ...). So the error you are seeing is because you are passing undefined to combinedLatest().

It actually takes some effort to create a real circular dependency. If you use defer, then you would have a true circular dependency:

var obsA,
    obsB,
    aRef,
    bRef;

aRef = Rx.Observable.defer(function () {
    return obsA;
});

bRef = Rx.Observable.defer(function () {
    return obsB;
});

obsA = Rx.Observable
    .returnValue(42)
    .combineLatest(bRef, function (a, b) {
            return a + b;
    });

obsB = Rx.Observable
    .returnValue(42)
    .combineLatest(aRef, function (b, a) {
        return b + a;
    });

obsA.subscribe();

<script src='https://rawgit.com/Reactive-Extensions/RxJS/v.2.5.3/dist/rx.all.js'></script>

Now that is a real circular dependency. Unfortunately you still get the same error, though with a much deeper stack trace:

RangeError: Maximum call stack size exceeded.
/* ... stack ... */

There is no fool-proof way to detect cycles. You could wrap the observables in a new observable and detect recursive calls to your subscribe method. But such an algorithm would be defeated if the underlying observables are using subscribeOn or publish or concat anything else that delays the actual circular subscriptions.

The best suggestion I have is to append a catch clause that checks for a range error and replaces it with a better error:

var obsA,
    obsB,
    aRef,
    bRef;

aRef = Rx.Observable.defer(function () {
    return obsA;
});

bRef = Rx.Observable.defer(function () {
    return obsB;
});

obsA = Rx.Observable
    .returnValue(42)
    .combineLatest(bRef, function (a, b) {
            return a + b;
    })
    .catch(function (e) {
        var isStackError = e instanceof RangeError && e.message === 'Maximum call stack size exceeded';
    
        return Rx.Observable.throw(isStackError ? new Error('Invalid, possibly circular observables.') : e);
    });

obsB = Rx.Observable
    .returnValue(42)
    .combineLatest(aRef, function (b, a) {
        return b + a;
    })
    .catch(function (e) {
        var isStackError = e instanceof RangeError && e.message === 'Maximum call stack size exceeded';
    
        return Rx.Observable.throw(isStackError ? new Error('Invalid, possibly circular observables.') : e);
    });

obsA.subscribe();

<script src='https://rawgit.com/Reactive-Extensions/RxJS/v.2.5.3/dist/rx.all.js'></script>