Modify Streams with User Input

Before we get into the next addition to our stream, let’s briefly recap what our getGifs stream does so far. This is the general flow:

1. Create a gifsForCurrentPage$ stream from the pagination$ stream
2. Every time the pagination$ stream emits, use concatMap to emit the results from fetchFromReddit using the after value passed in by the pagination
3. Since gifsForCurrentPage$ will emit the results of fetchFromReddit from each page individually, we create an allGifs$ stream that collects all of the emissions together using scan. Any time a new set of results come in, we add it to all of the previous results, and re-emit it.

The key thing we are going to address this lesson is that, at the moment, the subreddit we are using to pull in GIFs is hardcoded to the gifs subreddit. We want our user to be able to supply whatever subreddit they like to pull in GIFs from.

This means that we want to achieve the following:

1. The user should be able to type a subreddit into a form control
2. The getGifs stream will now emit GIFs from that particular subreddit (and it will get rid of any GIFs from the previous subreddit)

The question is, how do we modify getGifs to support this behavior? This is where things get a little tricky to think about. Let’s first look at the solution for some context and then we will talk through it.

Using switchMap to reset the stream

Modify the getGifs method to reflect the following:

  getGifs(subredditFormControl: FormControl) {
    // Start with a default emission of 'gifs', then only emit when
    // subreddit changes
    const subreddit$ = subredditFormControl.valueChanges.pipe(
      debounceTime(300),
      distinctUntilChanged(),
      startWith(subredditFormControl.value)
    );

    return subreddit$.pipe(
      switchMap((subreddit) => {
        // Fetch Gifs
        const gifsForCurrentPage$ = this.pagination$.pipe(
          concatMap((pagination) =>
            this.fetchFromReddit(subreddit, 'hot', pagination.after)
          )
        );

        // Every time we get a new batch of gifs, add it to the cached gifs
        const allGifs$ = gifsForCurrentPage$.pipe(
          scan((previousGifs, currentGifs) => [...previousGifs, ...currentGifs])
        );

        return allGifs$;
      })
    );
  }

The first notable change here is that we are supplying the getGifs method with a FormControl. In a moment, we will create a FormControl for our HomeComponent that will be used to enter in the subreddit value. We will pass that directly to the getGifs method which will allow it to use the valueChanges observable for that control to update the stream as necessary:

    const subreddit$ = subredditFormControl.valueChanges.pipe(
      debounceTime(300),
      distinctUntilChanged(),
      startWith(subredditFormControl.value)
    );

We use the usual tricks we talked about before - we use debounceTime so this stream won’t emit for every single keystroke, we use distinctUntilChanged so it won’t emit the value if it hasn’t actually changed, and we start the stream with a default value of whatever the form control value is.

Then we get into the part that actually makes this stream behave how we want:

    return subreddit$.pipe(
      switchMap((subreddit) => {
        // Fetch Gifs
        const gifsForCurrentPage$ = this.pagination$.pipe(
          concatMap((pagination) =>
            this.fetchFromReddit(subreddit, 'hot', pagination.after)
          )
        );

        // Every time we get a new batch of gifs, add it to the cached gifs
        const allGifs$ = gifsForCurrentPage$.pipe(
          scan((previousGifs, currentGifs) => [...previousGifs, ...currentGifs])
        );

        return allGifs$;
      })
    );

We take the current value from the subreddit$ stream using switchMap, and then we switch to the same allGifs$ stream that we created before (but now we make use of the subreddit value when calling fetchFromReddit).

The use of the switchMap here is important, but it might be hard to see why. When we need multiple values to create the result in our stream, in this example we need both subreddit$ and the values from pagination$ to get the result we want, we usually would use a combine latest, e.g:

// Fetch Gifs
const gifsForCurrentPage$ = combineLatest([this.pagination$, subreddit$].pipe(
    concatMap(([pagination, subreddit]) =>
        this.fetchFromReddit(subreddit, 'hot', pagination.after)
    )
);

// Every time we get a new batch of gifs, add it to the cached gifs
const allGifs$ = gifsForCurrentPage$.pipe(
    scan((previousGifs, currentGifs) => [...previousGifs, ...currentGifs])
);

return allGifs$;