RxJS handle multiple database connections

Question

this is the scenario:

I have multiple connection to different databases and i want to be sure that code runs whenever all connections are active.

I am using Rxjs to handle this (another solution is welcome ) but im facing that if i combine the connection events AFTER one of them is active i never get the subscription run, since combineLatest wants all observables to be emitted, but they were !

const a = new Rx.Subject();
const b = new Rx.Subject();

var bool = false;

setInterval(()=>{
    bool = !bool
    a.next(bool ? ' i am connected' : 'im NOT connected');
},1000)

setTimeout(()=>{
    b.next('i am always connected!')
},400)

// this variable will be exported to all js that run queries
var obs = new Rx.Observable.combineLatest(a,b);

setTimeout(()=>{
    obs.subscribe((andb)=>{
        console.log( andb )
        // i can check all connections at once and run the code
    })
},399)

// problem is here, i want to subscribe later than the connections 
//emit, if you edit 399 to 401 you will see that nothing is going on

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.6/Rx.js"></script>

bygrace · Answer 1 · 2018-01-23T19:37:51.367

2

With a timeout of 399 you subscribe before b emits so you see its value. With a timeout of 401 you subscribe after b emits so you don't see its value or that of a since combineLatest requires both. combineLatest wont keep track of the latest value for a and b till there is a subscription.

So you could use a different kind of subject that tracks the last value (BehaviorSubject or ReplaySubject) or use the repeat operator.

Here is an example with ReplaySubject(1) (basically same as BehaviorSubject but doesn't require an initial value) and subscribe at 401:

const a = new Rx.Subject();
const b = new Rx.ReplaySubject(1);

var bool = false;

setInterval(()=>{
    bool = !bool
    a.next(bool ? ' i am connected' : 'im NOT connected');
},1000)

setTimeout(()=>{
    b.next('i am always connected!')
},400)

// this variable will be exported to all js that run queries
var obs = new Rx.Observable.combineLatest(a,b);

setTimeout(()=>{
    obs.subscribe((andb)=>{
        console.log( andb )
        // i can check all connections at once and run the code
    })
},401)

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.6/Rx.js"></script>

edited Jan 23 '18 at 19:37

answered Jan 23 '18 at 18:14

bygrace

5,868
1
31
58

thanks for the reply , i can't find those type of subjects in rxjs 5 , what is the best approach with the latest version? – Ændri Domi Jan 23 '18 at 18:22
1

@ÆndriDomi Those subjects exist in the latest version of rxjs. I'll add your example with rxjs 5 to my answer. – bygrace Jan 23 '18 at 18:49
You are right ! can you please explain me differences between subjects ? – Ændri Domi Jan 23 '18 at 19:10
@ÆndriDomi It's better to ask a separate question if your original question has been answered. That way the question can be useful to others. Do some research yourself, and if you can't figure it out, pose a new question showing what code you have tried. – Ruan Mendes Jan 23 '18 at 19:14
@ÆndriDomi Yeah, you probably don't want a ReplaySubject without a buffer size or it will replay all messages. I updated the example to have a buffer size of 1. `ReplaySubject(1)` is effectively the same as the `BehaviorSubject` except that the BehaviorSubject requires an initial value. – bygrace Jan 23 '18 at 19:37

RxJS handle multiple database connections

1 Answers1