rxjs check if stream is empty before handling data
<p>We have the following stream.</p>

<pre><code>const recorders = imongo.listCollections('recorders')
    .flatMapConcat(names =&gt; {
        const recorders = names
            .map(entry =&gt; entry.name)
            .filter(entry =&gt; !_.contains(
                ['recorders.starts',
                 'recorders.sources',
                 'system.indexes',
                 'system.users'],
                entry));
        console.log(recorders);
        return Rx.Observable.fromArray(recorders);
    });

recorders.isEmpty()
    .subscribe(
        empty =&gt; {
            if(empty) {
                logger.warn('No recorders found.');
            }
        },
        () =&gt; {}
    );

recorders.flatMapConcat(createRecorderIntervals)
    .finally(() =&gt; process.exit(0))
    .subscribe(
        () =&gt; {},
        e =&gt; logger.error('Error while updating: %s', e, {}),
        () =&gt; logger.info('Finished syncing all recorders')
    );
</code></pre>

<p>If the stream is empty then we don't want to <code>createRecorderIntervals</code>. The above piece of code is working. However, checking if the stream is empty, is causing the <code>console.log</code> to be executed twice. Why is this happening? Can I fix it somehow?</p>

<p>EDIT: So, I went the following way, after rethinking it thanks to @Martin's answer</p>

<pre><code>const recorders = imongo.listCollections('recorders')
    .flatMapConcat(names =&gt; {
        const recorders = names
            .map(entry =&gt; entry.name)
            .filter(entry =&gt; !_.contains(
                ['recorders.starts',
                 'recorders.sources',
                 'system.indexes',
                 'system.users'],
                entry));

        if(!recorders.length) {
            logger.warn('No recorders found.');
            return Rx.Observable.empty();
        }

        return Rx.Observable.fromArray(recorders);
    })
    .flatMapConcat(createRecorderIntervals)
    .finally(() =&gt; scheduleNextRun())
    .subscribe(
        () =&gt; {},
        e =&gt; logger.error('Error while updating: %s', e, {}),
        () =&gt; logger.info('Finished syncing all recorders')
    );
</code></pre>