title: Trying to make my own RxJs observable
body: <p>I'm trying to convert an existing API to work with RxJS... fairly new to node, and very new to RxJs, so please bear with me.</p>
  
  <p>I have an existing API (getNextMessage), that either blocks (asynchronously), or returns a new item or error via a node-style (err, val) callback, when the something becomes available.</p>
  
  <p>so it looks something like:</p>
  
  <p>getNextMessage(nodeStyleCompletionCallback);</p>
  
  <p>You could think of getNextMessage like an http request, that completes in the future, when the server responds, but you do need to call getNextMessage again, once a message is received, to keep getting new items from the server.</p>
  
  <p>So, in order to make it into an observable collection, I have to get RxJs to keep calling my getNextMessage function until the subscriber is disposed();</p>
  
  <p>Basically, I'm trying to create my own RxJs observable collection.</p>
  
  <p>The problems are:</p>
  
  <ol>
  <li>I don't know how to make subscriber.dispose() kill the async.forever</li>
  <li>I probably shouldn't be using async.forever in the first place</li>
  <li>I'm not sure I should be even getting 'completed' for each message - shouldn't that be at the end of a sequence</li>
  <li>I'd like to eventually remove the need for using fromNodeCallback, to have a first class RxJS observable</li>
  <li>Clearly I'm a little confused.</li>
  </ol>
  
  <p>Would love a bit of help, thanks!</p>
  
  <p>Here is my existing code:</p>
  
  <pre><code>var Rx = require('rx');
  var port = require('../lib/port');
  var async = require('async');
  
  function observableReceive(portName)
  {
      var observerCallback;
      var listenPort = new port(portName);
      var disposed = false;
  
      var asyncReceive = function(asyncCallback)
      {
          listenPort.getNextMessage(
              function(error, json)
              {
                  observerCallback(error, json);
  
                  if (!disposed)
                      setImmediate(asyncCallback);
              }
          );
      }
  
      return function(outerCallback)
      {
          observerCallback = outerCallback;
          async.forever(asyncReceive);
      }
  }
  
  var receive = Rx.Observable.fromNodeCallback(observableReceive('rxtest'));
  var source = receive();
  
  var subscription = source.forEach(
      function (json)
      {
          console.log('receive completed: ' + JSON.stringify(json));
      },
      function (error) {
          console.log("receive failed: " + error.toString());
      },
      function () {
          console.log('Completed');
          subscription.dispose();
      }
  );
  </code></pre>