zoukankan      html  css  js  c++  java
  • [RxJS] Combining streams in RxJS

    Source: Link

    We will looking some opreators for combining stream in RxJS:

    • merge
    • combineLatest
    • withLatestFrom
    • concat
    • forkJoin
    • flatMap / switchMap

     Merge: 

    Observable.merge behaves like a "logical OR" to have your stream handle one interaction OR another.

    let btn1 = document.querySelector("#btn1");
    let btn2 = document.querySelector("#btn2");
    
    let btn1Click$ = Rx.Observable.fromEvent(btn1, "click");
    let btn2Click$ = Rx.Observable.fromEvent(btn2, "click");
    
    let btn1Log$ = btn1Click$.map( (ev) => {
      console.log("Button 1 clicked");
    });
    let btn2Log$ = btn2Click$.map( (ev) => {
      console.log("Button 2 clicked");
    });
    let clicks$ = Rx.Observable.merge(btn1Log$, btn2Log$);
    
    clicks$.subscribe();

    combineLatest:

    Ofter used when one of streams value changed, then produce a side effect:

    var source1 = Rx.Observable.interval(1000)
      .map(function (i) { return 'First: ' + i; });
    
    var source2 = Rx.Observable.interval(2000)
      .map(function (i) { return 'Second: ' + i; });
    
    // Combine latest of source1 and source2 whenever either gives a value
    var source = Rx.Observable.combineLatest(
        source1,
        source2
      ).take(4);
    
    var subscription = source.subscribe(
      function (x) {
        console.log(x);
      },
      function (err) {
        console.log('Error: %s', err);
      },
      function () {
        console.log('Completed');
      });
    
    /*
    ["First: 0", "Second: 0"]
    ["First: 1", "Second: 0"]
    ["First: 2", "Second: 0"]
    ["First: 2", "Second: 1"]
    "Completed"
    */

    withLatestFrom: 

    var source1 = Rx.Observable.interval(1000)
      .map(function (i) { return i; });
    
    var btn = document.querySelector("#btn");
    var source2 = Rx.Observable.fromEvent(btn, "click");
    
    var source =source1
    .withLatestFrom(
      source2,
      (source1, click) => ({timer: source1, clicks: click.x})
    ).take(4);
    
    var subscription = source.subscribe(
      function (x) {
        console.log(x);
      },
      function (err) {
        console.log('Error: %s', err);
      },
      function () {
        console.log('Completed');
      });

    Read the difference between combineLatest and withLatestFrom: Link.

    concat:

    Concat will combine two observables into a combined sequence, but the second observable will not start emitting until the first one has completed.

    let first = Rx.Observable.interval(1000).take(3).do( (i) => { console.log("First: ", i);});
    
    let second = Rx.Observable.interval(500).take(3).do( (i) => { console.log("Second: ", i);});
    
    first.concat(second).subscribe();
    
    /*
    "First: "
    0
    "First: "
    1
    "First: "
    2
    "Second: "
    0
    "Second: "
    1
    "Second: "
    2
    */

    forkJoin:
    We use forkJoin to execute observables in parallel. One common use case of this is making multiple http requests in parallel. In my sample I am forkJoining two very simple observables, but the key point is that the subscriber won't receive any values until both observables have completed.

    let first = Rx.Observable.interval(1000).take(6);
    
    let second = Rx.Observable.interval(500).take(3);
    
    Rx.Observable.forkJoin(first, second).subscribe(
      (res) =>{
        console.log(res); // [5, 2]
      },
      (err) => {
        console.log(err);
      },
      () => {
        console.log("Completed");  // Completed
      }
    );

    flatMap / switchMap

    flatMap and switchMap basic are the same.

    Just switchMap only care about the latest value, will ignore the previous value. So good to use with http reuqest.

    The reason to use flatMap is because inside Observable you migth return another Observable, such as:

    var btn = document.querySelector("#btn");
    
    var click$ = Rx.Observable.fromEvent(btn, "click");
    
    var promise$ = Rx.Observable.fromPromise( jquery.http('xxx'));
    var xhrCall$ = click$.flatMap( () => {
        return promise$;
    });
    
    xhrCall$.subscribe( (res) => {
      console.log(res);    
    })

    Inside Observalbe return another Observable, will create a 2-d Observable, just like inside map ruturn another array, will create 2-d array.

    So we need to flatten it.

  • 相关阅读:
    windows pm2 开机启动
    微信小程序自定义组件封装及父子间组件传值
    cloc 统计代码行数工具
    小程序获取当前页面路径url
    小程序navigator点击有时候会闪一下
    Centos7安装python3与Python2共存
    Docker配置国内官方镜像
    Centos7下Docker的安装与使用
    Centos7yum源修改为国内阿里源
    Vim下一键运行python代码
  • 原文地址:https://www.cnblogs.com/Answer1215/p/5410864.html
Copyright © 2011-2022 走看看