CODEWITHSUNDEEP
promiserxjs
mettjus
mettjus

Reputation: 2988

Rx.js: check sequence of items asynchronously with promise and take until true is returned

I'm very new to Rx. I have an array of usernames and I'm testing them against a db until I get to one which doesn't exist (a free one). The db call gives me a promise.

How can I implement it with rx(.js)?

The code I implemented until now is:

the usernames stream

var usernamesStream = Rx.Observable.from(['user1','user2','user3'])

the db query method

var checkUsernameIsFree = function (username) {
  return db.users.find({username:username}).toArray().then(function(users) {
    return users.length == 0
  })
}

I guess I should create a stream by using .fromPromise on the promise. How can I implement that and join the 2 streams so that subscribe gets called only with the first free username?

Upvotes: 0

Views: 718

Answers (1)

paulpdaniels
paulpdaniels

Reputation: 18663

You can use concatMap to preserve the order of your requests, also it will accept promises implicitly:

var usernames = Rx.Observable.from(['user1','user2','user3']);

//Preserve the order of the requests
var subscription = usernames.concatMap(function(name) {
           return Rx.Observable.fromPromise(db.users.find({username:name}).toArray())
                               .filter(function(x) { return x.length == 0; })
                               .map(function() {return name; });
          })
         .first()
         .subscribe(
           function(availableUserName){
             /*Only gets called once*/
           },
           function(e){ 
             /*Will get raised if no names were found or
               there was a problem accessing the database*/
          });

A couple things that will change depending on your requirements:

  1. If no result is not an error, you should use .firstOrDefault() or .filter().take(1) instead.

  2. In the above solution, concatMap will probably kick off database requests on all the names, and simply preserve the order of the responses. If you want to delay the execution of the the db request until the previous finished you should wrap db.users.find with startAsync:

return Rx.Observable.startAsync(function() { 
  return db.users.find({username : name}).toArray();
});

Edit 1

Fixed a bug in the previous code that didn't pass the name value to the final subscriber.

Upvotes: 1

Related Questions