Combine initial stream with modify streams with RxJS

Question

I'm trying to think functional reactive and set up a list of elements on a page as a stream. This is using Angular2 but the problem should be similar to any stream-based architecture. So, I have two streams currently, the initial stream (http call to get a list of users from github) and a remove user stream (occurs when the remove user button is clicked). I believe the marble diagram would look like the following:

|[user1,user2,user3]|                       <--- http initial stream
|---------------------x----------x-----...  <--- x denotes user removed

How do I combine these streams to get this to work? I'm also thinking later of having more streams for sorting and ordering. Am I going about this the right way? Here's the code (note this code is incomplete, currently the removeUser$ is not interactive with the user$ which it should):

export class UserGridComponent implements OnInit { 
    public users$: Observable<any>;
    public removeUser$: Subject;

    constructor(private _githubUserService: GithubUserService) { }

    ngOnInit() { 
       this.removeUser$ = new Subject()
           .subscribe((user) => { console.log('next ' + JSON.stringify(user)});
       this.users$ = this._githubUserService.getUsers()
           .map((res) => res.json());
    }
}

Here is the Plunker

Currently I am only logging to console that the remove button is clicked and passes the user.

Here is the html template which shows that I subscribe to the user$ by using the async pipe (Angular2):

<md-list>
  <h1 md-header>GitHub Users</h1>
  <md-list-item *ngFor="let user of users$ | async">
    <a href="https://github.com/{{user.login}}" target="_new">
      <img md-list-avatar [src]="user.avatar_url">
    </a>
    <h4>{{user.login}}</h4>
    <button md-icon-button (click)="removeUser$.next(user)">
      <md-icon>cancel</md-icon>
    </button>
  </md-list-item>
</md-list>

I'm not 100% clear on what you're trying to achieve, but in your code you never subscribed to the users$ observable so it is likely still cold. — hartz89
– hartz89, Commented Jun 29, 2016 at 13:04
I do subscribe to the user$ observable in the template using angular2 async pipe. I'm trying to remove the user from the user$ whenever a removeUser$ event occurs. I'll post the template too, it's in the plunker, thanks. — Ryan Langton
– Ryan Langton, Commented Jun 29, 2016 at 13:13

Can Nguyen · Accepted Answer · 2016-07-09 15:59:14Z

1

ngOnInit() { 
   this.users$ = new BehaviorSubject([]); // init with empty user array
   this._githubUserService.getUsers()
       .map((res) => res.json())
       .subscribe(this.users$);

   this.removeUser$ = new Subject();
   this.removeUser$.subscribe((user) => { 
       this.users$.take(1)  // get current users, as users$ is a BehaviorSubject
           .map(users => users.filter(u => u != user))   // remove `user`
           .subscribe(this.users$); // update users$ stream
   });
}

Hope my comments can serve as explanations. BehaviorSubject cache last emitted value, so this.users$.take(1) is synchronous. I find myself using a BehaviorSubject as source for angular2 ' | async' pipe very often.

EDIT: Same idea but a little shorter for the removeUser$ part:

.....
   this.removeUser$.withLatestFrom(user$, 
       (toRemove, users) => users.filter(u => u != toRemove)   // remove `toRemove`
   ).subscribe(this.users$); // update users$ stream

edited Jul 9, 2016 at 15:59

answered Jun 29, 2016 at 15:26

Can Nguyen

1,47010 silver badges11 bronze badges

Sign up to request clarification or add additional context in comments.

2 Comments

Ryan Langton Over a year ago

though this may work, it seems to be really confusing and not easy to wrap my head around... I wonder if in this use case it would be best to keep state (using redux) so that we can remove from a stream.. where my other use cases - filter, sort - combining the streams would be fine.. or if there's a better way to write this without the embedded user stream

Davy Over a year ago

A state store is indeed the missing piece here. I see what you are trying to do, i was at the same point not to long ago. You can indeed do this with pure rxjs (using the scan operator), but using a state store like ngrx of ngxs will definitely help you programming reactively. I'll see if i can find the same to whip up an example using the scan operator: it is a nice operator to know. But you shoul definitely give ngrx or ngxs a try.

Davy · Accepted Answer · 2018-12-19 13:05:45Z

Take a look at this stackblitz: it shows how you can use scan to mutate a single value (which can be anything ...) by events emitted from one or multiple sources. In the example, all incoming actions come from on modify$ stream, but you might as well merge together an add$ and a remove$ stream.

https://stackblitz.com/edit/angular-evc48h

But ... i would not recommend doing this because this is -exactly- what a state store does for you.

So be sure to check out the well-known ngrx: https://github.com/ngrx/platform

Or the lesser known but cleaner ngxs: https://github.com/ngxs/store

I often hear people claim that using a state store introduces unnecessary complexity, and this might be true for small and simple screens. The more complex your screen gets however, the more a state store will help you keep it elegant and clean.

Collectives™ on Stack Overflow

Combine initial stream with modify streams with RxJS

2 Answers 2

2 Comments

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

2 Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Related