[rxjs] Creating An Observable with RxJS

时间：2015-08-19 19:13:54 阅读：220 评论：0 收藏：0 [点我收藏+]

Create an observable

var Observable = Rx.Observable;

var source = Observable.create(function(observe){
    var person = {
      name: "Zhentian",
      message: "Hello World!"
    };
  
    observe.onNext(person);
    observe.onCompleted();
});

var sub = source.subscribe(function onNext(person){
  console.log(person.name + ‘ say ‘ + person.message);
}, function onError(err){
  console.log(err);
}, function onCompleted(){
  console.log("done");
});

//Zhentian say Hello World!
//done

Async

var Observable = Rx.Observable;

var source = Observable.create(function(observe){
  
  setTimeout(function(){
    
    var person = {
      name: "Zhentian",
      message: "Hello World!"
    };  
    
    observe.onNext(person);
    observe.onCompleted();
  }, 1000);

  console.log("ansyc finished!");

});

var sub = source.subscribe(function onNext(person){
  console.log(person.name + ‘ say ‘ + person.message);
}, function onError(err){
  console.log(err);
}, function onCompleted(){
  console.log("done");
});

//"ansyc finished!"
//"Zhentian say Hello World!"
//"done"

Dispose the async

When you dispose the operation, we can see it log out "start timeout", which is not good, because, the onNext() would never be called, what we want is it even don‘t get inside setTimeout function.

var Observable = Rx.Observable;

var source = Observable.create(function(observe){
  
  setTimeout(function(){
    console.log("Starat timeout");
    var person = {
      name: "Zhentian",
      message: "Hello World!"
    };  
    
    observe.onNext(person);
    observe.onCompleted();
  }, 1000);

  console.log("ansyc finished!");

});

var sub = source.subscribe(function onNext(person){
  console.log(person.name + ‘ say ‘ + person.message);
}, function onError(err){
  console.log(err);
}, function onCompleted(){
  console.log("done");
});

setTimeout(function(){
  
  sub.dispose();
}, 500);

/*
"ansyc finished!"
"Starat timeout"
*/

Define the dispose

We can give setTimeout and id, and in the return function, we clear this timeout.

var Observable = Rx.Observable;

var source = Observable.create(function(observe){
  
  var id = setTimeout(function(){
    console.log("Starat timeout");
    var person = {
      name: "Zhentian",
      message: "Hello World!"
    };  
    
    observe.onNext(person);
    observe.onCompleted();
  }, 1000);

  console.log("ansyc finished!");
  
  // Note that this is optional, you do not have to return this if you require no cleanup
  return function(){
    clearTimeout(id);
  }

});

var sub = source.subscribe(function onNext(person){
  console.log(person.name + ‘ say ‘ + person.message);
}, function onError(err){
  console.log(err);
}, function onCompleted(){
  console.log("done");
});

setTimeout(function(){
  
  sub.dispose();
}, 500);

/*
"ansyc finished!"
*/

Catch error

If we throw an error in the code, but we found it actually not catched by the onError handler.

var Observable = Rx.Observable;

var source = Observable.create(function(observe){
  
  var id = setTimeout(function(){
    throw "there is an error"; //Throw an error here
    var person = {
      name: "Zhentian",
      message: "Hello World!"
    };  
    
    observe.onNext(person);
    observe.onCompleted();
  }, 1000);
  
  // Note that this is optional, you do not have to return this if you require no cleanup
  return function(){
    clearTimeout(id);
  }

});

var sub = source.subscribe(function onNext(person){
  console.log(person.name + ‘ say ‘ + person.message);
}, function onError(err){
  console.log("Error: " + err);
}, function onCompleted(){
  console.log("done");
});

/*
"error"
"Uncaught there is an error (line 6)"
*/

What we can do is to add try catch in the block.

var Observable = Rx.Observable;

var source = Observable.create(function(observe){
  
  var id = setTimeout(function(){
    try{
       throw "there is an error"; //Throw an error here
       var person = {
         name: "Zhentian",
         message: "Hello World!"
       };  
    
       observe.onNext(person);
       observe.onCompleted();
    }catch(error){
       observe.onError(error);
    }

  }, 1000);
  
  // Note that this is optional, you do not have to return this if you require no cleanup
  return function(){
    clearTimeout(id);
  }

});

var sub = source.subscribe(function onNext(person){
  console.log(person.name + ‘ say ‘ + person.message);
}, function onError(err){
  console.log("Error: " + err);
}, function onCompleted(){
  console.log("done");
});

/*
"Error: there is an error"
*/

[rxjs] Creating An Observable with RxJS

原文：http://www.cnblogs.com/Answer1215/p/4742631.html

踩

(0)

评论一句话评论（0）

分享档案

更多>

2021年09月23日 (328)
2021年09月24日 (313)
2021年09月17日 (191)
2021年09月15日 (369)
2021年09月16日 (411)
2021年09月13日 (439)
2021年09月11日 (398)
2021年09月12日 (393)
2021年09月10日 (160)
2021年09月08日 (222)