[RxJS] Sharing Streams with Share

Posted Answer1215

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了[RxJS] Sharing Streams with Share相关的知识,希望对你有一定的参考价值。

A stream will run with each new subscription added to it. This lesson shows the benefits of using share so that the same stream can be shared across multiple subscriptions.

 

const timer$ = starters$
    .switchMap(intervalActions)
    .startWith(data)
    .scan((acc, curr)=> curr(acc))


timer$
    .do((x)=> console.log(x))
    .takeWhile((data)=> data.count <= 3)
    .withLatestFrom(
        input$.do((x)=> console.log(x)),
        (timer, input)=> ({count: timer.count, text: input})
    )
    .filter((data)=> data.count === parseInt(data.text))
    .reduce((acc, curr)=> acc + 1, 0)
    .repeat()
    .subscribe(
        (x)=> document.querySelector(‘#score‘).innerhtml = `
            ${x}
        `,
        err=> console.log(err),
        ()=> console.log(‘complete‘)
    );

Current code has a problem every time we type in the input, we need to click delete keyborad to delete the input. This is not conveninet. 

 

So what we want is every time when the code check whether the value match, then we clean the input:

const runningGame$ = timer$
    .do((x)=> console.log(x))
    .takeWhile((data)=> data.count <= 3)
    .withLatestFrom(
        input$.do((x)=> console.log(x)),
        (timer, input)=> ({count: timer.count, text: input})
    );

runningGame$
  .subscribe( (x) => input.value = "");

 runningGame$
    .filter((data)=> data.count === parseInt(data.text))
    .reduce((acc, curr)=> acc + 1, 0)
    .repeat()
    .subscribe(
        (x)=> document.querySelector(‘#score‘).innerHTML = `
            ${x}
        `,
        err=> console.log(err),
        ()=> console.log(‘complete‘)
    );

So we split the code and add another subscribe to clean the input, but now the problem is everytime it console log twice. This is because we have two subscribe on the runningGame$. 

 

TO solve this problem, we need to share() the stream:

const runnintGame$ = timer$
    .do((x)=> console.log(x))
    .takeWhile((data)=> data.count <= 3)
    .withLatestFrom(
        input$.do((x)=> console.log(x)),
        (timer, input)=> ({count: timer.count, text: input})
    )
.share();

Now it only log out once, but it doesn‘t clean the input when we click start again. THis is because we only repeat() this part of code:

 runningGame$
    .filter((data)=> data.count === parseInt(data.text))
    .reduce((acc, curr)=> acc + 1, 0)
    .repeat()

What we need to do is also make runningGame$ (clean the input) stream repeat itself:

// To clean the input
runningGame$
  .repeat()
  .subscribe( (x) => input.value = "");

 

----------------------------

Code:

const timer$ = starters$
    .switchMap(intervalActions)
    .startWith(data)
    .scan((acc, curr)=> curr(acc))


const runningGame$ = timer$
    .do((x)=> console.log(x))
    .takeWhile((data)=> data.count <= 3)
    .withLatestFrom(
        input$.do((x)=> console.log(x)),
        (timer, input)=> ({count: timer.count, text: input})
    )
.share();

// To clean the input
runningGame$
  .repeat()
  .subscribe( (x) => input.value = "");

 runningGame$
    .filter((data)=> data.count === parseInt(data.text))
    .reduce((acc, curr)=> acc + 1, 0)
    .repeat()
    .subscribe(
        (x)=> document.querySelector(‘#score‘).innerHTML = `
            ${x}
        `,
        err=> console.log(err),
        ()=> console.log(‘complete‘)
    );

 

以上是关于[RxJS] Sharing Streams with Share的主要内容,如果未能解决你的问题,请参考以下文章

[RxJS] Reactive Programming - Sharing network requests with shareReplay()

[RxJS] Handling Multiple Streams with Merge

[Angular2 Form] Use RxJS Streams with Angular 2 Forms

[Recompose] Merge RxJS Button Event Streams to Build a React Counter Component

[Vue-rx] Access Events from Vue.js Templates as RxJS Streams with domStreams

Node.js Streams 与 Observables