title: rxjs check if stream is empty before handling data
body: <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>