title: forcing completion of an rxjs observer
body: <p>I've got an rxjs observer (really a Subject) that tails a file forever, just like tail -f.  It's awesome for monitoring logfiles, for example.</p>
  
  <p>This "forever" behavior is great for my application, but terrible for testing.  Currently my application works but my tests hang forever.</p>
  
  <p>I'd like to force an observer change to complete early, because my test code knows how many lines should be in the file.  How do I do this?</p>
  
  <p>I tried calling onCompleted on the Subject handle I returned but at that point it's basically cast as an observer and you can't force it to close, the error is:</p>
  
  <blockquote>
    <p>Object # has no method 'onCompleted'</p>
  </blockquote>
  
  <p>Here's the source code:</p>
  
  <pre><code>function ObserveTail(filename) {
  
  source = new Rx.Subject();
  
  if (fs.existsSync(filename) == false) {
      console.error("file doesn't exist: " + filename);
  }
  
  var lineSep = /[\r]{0,1}\n/;
  tail = new Tail(filename, lineSep, {}, true);
  
  tail.on("line", function(line) {
          source.onNext(line);
  });
  tail.on('close', function(data) {
      console.log("tail closed");
      source.onCompleted();
  });     
  tail.on('error', function(error) {
      console.error(error);
  });     
  
  this.source = source;
  }           
  </code></pre>
  
  <p>And here's the test code that can't figure out how to force forever to end (tape style test). Note the "ILLEGAL" line:</p>
  
  <pre><code>test('tailing a file works correctly', function(tid) {
  
  var lines = 8;
  var i = 0;
  var filename = 'tape/tail.json';
  var handle = new ObserveTail(filename);
  touch(filename);
  
  handle.source
  .filter(function (x) {
      try {
          JSON.parse(x);
          return true;
      } catch (error) {
          tid.pass("correctly caught illegal JSON");
          return false;
      }
  })
  .map(function(x) { return JSON.parse(x) })
  .map(function(j) { return j.name })
  .timeout(10000, "observer timed out")
  .subscribe (
      function(name) {
          tid.equal(name, "AssetMgr", "verified name field is AssetMgr");
          i++;
          if (i &gt;= lines) {
              handle.onCompleted();   // XXX ILLEGAL
          }
      },
      function(err) {  
          console.error(err)
          tid.fail("err leaked through to subscriber");
      },
      function() {
          tid.end();
          console.log("Completed");
      }
  );
  
  })
  </code></pre>