Trying to make my own RxJs observable

Question

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. 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. so it looks something like: getNextMessage(nodeStyleCompletionCallback); 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. 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(); Basically, I'm trying to create my own RxJs observable collection. The problems are: - I don't know how to make subscriber.dispose() kill the async.forever - I probably shouldn't be using async.forever in the first place - I'm not sur…

Accepted Answer

So here's probably what I would do.

```
var Rx = require('Rx');

// This is just for kicks. You have your own getNextMessage to use. ;)
var getNextMessage = (function(){

var i = 1;

return function (callback) {
    setTimeout(function () {
      if (i > 10) {
        callback("lawdy lawd it's ova' ten, ya'll.");
      } else {
        callback(undefined, i++);
      }
    }, 5);
  };

}());

// This just makes an observable version of getNextMessage.
var nextMessageAsObservable = Rx.Observable.create(function (o) {
  getNextMessage(function (err, val) {
    if (err) {
      o.onError(err);
    } else {
      o.onNext(val);
      o.onCompleted(); 
    }
  });
});

// This repeats the call to getNextMessage as many times (11) as you want.
// "take" will cancel the subscription after receiving 11 items.
nextMessageAsObservable
  .repeat()
  .take(11)
  .subscribe(
    function (x)   { console.log('next', x);    },
    function (err) { console.log('error', err); },
    function ()    { console.log('done');       }
  );
```

Trying to make my own RxJs observable

Solution

Problem