One of this app’s features allows a user to choose a song and add tags or comments to that song at different points throughout its duration. These tags can then be displayed in sync with the song whenever it is played.

Essentially, the challenge here is that any time the song begins to play, all of the tags for that song need to be synced up based on the point where playback is started so they show at the correct time. Likewise, any time playback is stopped, tags should be hidden.

Creating Observables From User Events

There are a number of user events that can trigger this syncing of tags. There is the play/pause button for playback control. A user can also scrub to a certain part of the song to begin playback at that point.

At the most basic level, we have a function that accepts an Observable of user events and returns an Observable of UI actions. The logic behind this view needs to be able to handle the different kinds of events and determine what kind of actions should take place.

User Events → UI Actions

The first step in this chain is tracking the actual user events. Similarly to what we have discussed in Part I, we can create Observables in a number of ways:

1. Observable.of()
2. Observable.from()
3. Observable.fromEvent()
4. Creating a Subject and using subject.next()

In this case we first want to create an Observable from a click event on the play/pause button. Therefore, we’ll use the Observable.fromEvent() operator. As we can see in the following code, whenever the button is clicked, we simply emit a string identifying it as a ‘togglePlaybackEvent’.

let togglePlaybackEvents = Observable.fromEvent(this.playButton.nativeElement, ‘click’)
  .map(_ => ‘togglePlaybackEvent’);

A user can also scrub to a point on the Wavesurfer, which is an event handled by the Wavesurfer plugin. However, we can be notified that the user has taken this action. In order to incorporate these events into our Rx chain, we setup seekEvents as a Subject. Similarly to the events emitted by the play/pause button, we’ll emit a ‘seekEvent’ through the seekEvents subject whenever our Wavesurfer.on(‘seek’) callback is called.

seekEvents: Subject<any> = new Subject();

this.waveSurfer.on(‘seek’, _ => this.seekEvents.next(‘seekEvent’));

At this point we have two separate user event streams that we would like to combine into one. To do so, we can use the Observable.merge() operator.

The merge operator takes the emissions of multiple streams and combines them into one stream in order of emission. Using it here, these two separate Observables are merged together into one Observable, so as the user interacts with the interface, user events are emitted by a single stream.

let userEvents = Observable.merge(
                            togglePlaybackEvents, this.seekEvents);

Once we have the user events organized into a single stream we will use these events to trigger the syncing of tags. Before doing this however, there is one side effect we have to deal with. If a user interacts with the play/pause button, signified by the ‘togglePlaybackEvent’ emission, it means they want to toggle the song playback, so we must do that. As we discussed in Part I of this blog post, we can use the Observable.do() operator here which is generally used for synchronous side-effects you might need to do mid-chain.

*The Wavesurfer plugin handles scrubbing actions by starting playback at the desired position so we don’t need to worry about that particular side-effect.

userEvents
  .do(userEvent => {
    if (userEvent == "togglePlaybackEvent") {
      togglePlayback();
    }   
  })

We next need to figure out whether we should start or stop showing tags. If the Wavesurfer is playing, we need to start showing tags and sync them based on the current position of the Wavesurfer. Otherwise we need to stop showing tags. We have a helper function called playbackStats() that returns the playback status of the Wavesurfer.

userEvents
  .do(...)
  .map(_ => {
    let stats = playbackStats();
    return {
      currentTime: stats.currentTime,
      command: stats.playbackState 
             ? 'startTagsPlayback' : 'stopTagsPlayback'
    }
  });

Syncing Tags

At this point we have a mapping from user events to objects that describe what we should do with the tags. For the sake of this post, let’s call these objects tag commands. From the above function we can see that we have two types of commands:

1. ‘startTagsPlayback’
2. ‘stopTagsPlayback’

When we receive a ‘startTagsPlayback’ command as an emission, we know that we want to start showing the tags as the song is playing. But in order to do this, we first need to calculate the subset of all the tags that come at or after the playback start point.

let createShowTagsActions = (currentTime, tags) => {
  tags.filter(tag => tag.time > currentTime)
}

Here, we take the tags array and return an array that includes only tags where their show time is more than or equal to the current playback time. We won’t be displaying tags that have already passed, so we have no reason to deal with them.

Great, so we have the tags we want to show, but how do we actually display these tags? Well, displaying a tag is actually made up of two actions — a show action and a hide action. Just as we want to display the tags at the correct time, we also want them to disappear a certain period of time after showing in order to create a pop up effect. So how can we do that?

Concatenating Observables

To achieve the desired effect, we need to create an observable that will emit the show action followed by the hide action. Each of the actions will have a reference to the tag that is being shown/hidden. To do so, we can use the concat operator which allows us to concatenate two Observables together. The following function builds the observable that emits a show action followed by a hide one.

let createTagActionsObservable = (tag, currentTime) => {
  let showAction = {'tag': tag, 'action': 'show'};
  let showTagObservable = Observable.of(showAction);

  let hideAction = {'tag': tag, 'action': 'hide'};
  let hideTagObservable = Observable.of(hideAction);
   
  return Observable.concat(showTagObservable, hideTagObservable);
}

Unlike the merge operator which combines emissions in the order that they are emitted, the concat operator combines all of the emissions from the first Observable followed by all of the emissions of the second Observable. So currently, we can subscribe to one of these concatenated Observables for a single tag and we’ll get two emissions back to back, one to show the tag followed by one to hide it.

Emitting Observables With Delays

So now that we have the show and hide emission for a tag being emitted in the correct order, we have to handle the timing aspect of these emissions — we don’t want to show the tag and immediately hide it. We want to give users a chance to read the comment, so we want to wait two seconds before hiding the comment pop up. Returning back to our function that creates this Observable for a single tag, we see that we can do this by adding on the delay() operator when we create the hideTagObservable.

let createTagActionsObservable = (tag, currentTime) => {
  let showAction = {'tag': tag, 'action': 'show'};
  let showTagObservable = Observable.of(showAction);

  let hideAction = {'tag': tag, 'action': 'hide'};
  let hideTagObservable = Observable.of(hideAction).delay(2000);
   
  return Observable.concat(showTagObservable, hideTagObservable);
}

The delay operator takes a measure of time, and returns an Observable that delays all the emissions of its upstream. In our case, we’re applying a two second delay to the hideAction observable. So now we can subscribe to this concatenated observable and we will get the emitted action to show this tag and two seconds later we will get the emitted action to hide it.

Almost there, but we have one additional layer of complexity here. We also don’t want to show the tag right away. Instead, we know that we need to show the tag at a specific time in the song’s playback, indicated by the tag object’s time value. To do this, we can use the delay operator again when creating the showTagObservable.

let createTagActionsObservable = (tag, currentTime) => {
  let showAction = {'tag': tag, 'action': 'show'};
  let timeToShowTag = (currentTime - tag.time) * 1000;
  let showTagObservable = Observable.of(showAction)
      .delay(timeToShowTag);

  let hideAction = {'tag': tag, 'action': 'hide'};
  let hideTagObservable = Observable.of(hideAction).delay(2000);
   
  return Observable.concat(showTagObservable, hideTagObservable);
}

We calculate the delay based on the current time in playback that the playback action was taken. If we play from the beginning of the song (time 0) and the tag is to be shown at second ten, then we emit that tag to be shown with a delay of ten seconds.

So now, for each individual tag, we will receive the show action at the correct time in the song’s playback to show the tag and two seconds later we will receive the hide action to hide it again.

We can apply the function to every tag that we intend to show. This will result in a tag actions Observable for each tag. But again, our goal here is to flatten this all into one stream so we can go ahead and merge these Observables together.

let createShowTagsActions = (currentTime, tags) => {
  let allActions = Observable.empty();
    tags
     .filter(tag => tag.time > currentTime)
     .forEach(tag => {
        let actions = createTagActionsObservable(tag, currentTime);
        allActions = Observable.merge(actions, allActions);
    })
  return allActions;
}

Once we create an actions observable for a tag, using the merge operator again, we can merge each of those Observables into a single Observable, initialized by using the Observable.empty() operator. Now we can subscribe to this Observable and we will receive the emission for the first tag to show at the calculated time, followed two seconds later by its associated hide action, followed then by the show tag emission of the next tag at the calculated time, followed two seconds later by its hide action and so on.

Now that we have handled the more complex logic of what to do when we receive the ‘startTagsPlayback’ command, we can easily handle the logic for the ‘stopTagsPlayback’ command. When this happens we simply want to emit a hide all tags action because we don’t want to show any tags.

let hideAll = Observable.of({tag: null, action: 'hideAll'});

Now that we have created Observables for each command we need to tell Rx which one to use.

From User Events to Actions

At this point, we have translated user events into commands which we can further map into actions that the UI can handle. We are mapping a value (‘startTagsPlayback’ or ‘stopTagsPlayback’) to an actions stream which will be flattened into the UIActions stream. For this we have chosen to use the switchMap() operator as the mapping operator in order to cancel any outstanding actions that haven’t been emitted yet. To learn more about how switchMap() works, please refer to Part I of this blog post.

userEvents
  .do(...)
  .map(_ => {
    let stats = playbackStats();
    return {
      currentTime: stats.currentTime,
      command: stats.playbackState 
             ? 'startTagsPlayback' : 'stopTagsPlayback'
    }
  })
  .switchMap(cmd => {
    let hideAll = Observable.of({tag: null, action: 'hideAll'});
    if(cmd.command === 'startTagsPlayback') {
      return Observable.concat(hideAll,
        createShowTagsActions(cmd.currentTime, savedTags));
    } else {
        return hideAll;
    }
  });

If we get ‘startTagsPlayback’ command we return the chain with show/hide pairs. In this case, we need to take one more step and prepend a ‘hideAll’ action to this Observable using the concat() operator again. This is just some basic clean up that ensures any tags which might still be showing from a previous event are hidden and we start with a blank slate.

If we receive a ‘stopTagsPlayback’ command, we simply hide all tags by emitting a ‘hideAll’ action.

Handling Tag Actions

Finally, the TaggedSongComponent subscribes to this Observable and handles all incoming actions.

uiActions.subscribe(tagAction => {  
  if(tagAction['action'] === 'hideAll') {
    this.hideTags();
  } else if(tagAction['action'] === 'show'){
    this.showTag(tagAction['tag'])
  } else {
    this.hideTag(tagAction['tag']);
  }
}, error => console.log(error));

As we know, the tag action is an object that has a few key pieces of information on it: the individual tag’s key information and the action that the UI should take.

const TagAction = { 
  "tag": { 
    "position": 12.5,
     "text": "This song is awesome!",
     "time": 1.06
   },
  "action": "show"
}

The tag Observable is emitted according to its delay, at which point, because the action is ‘show’, we find the tag in the UI based on it’s tag.position value and change it’s display property to block. Two seconds later, the same tag’s hide Observable is emitted with the ‘hide’ action, at which point we find the tag in the UI and switch the display property back to ‘none’. Likewise, if the action is ‘hideAll’ then we hide all of the tags by switching their display property to ‘none’. So now the UI can react in different ways based on the actions.

So there you have it, an example of using RxJS to drive asynchronous, time-based content. This of course is only scratching the surface of what RxJS is capable of, but it’s been a great way for me to become acquainted with the library as a beginner. I hope it’s been useful for you as a reader. In the future, I hope to follow up this post with another that discusses how I was able to test this chain — so check back soon!

A Beginner’s Adventure with RxJS: Part II was originally published in Def Method on Medium, where people are continuing the conversation by highlighting and responding to this story.

What is RxJS?

There are many tools and libraries that exist that enable developers to easily implement FRP patterns, perhaps one of the most common being Reactive Extensions. Originally developed by Microsoft as Rx.NET, Rx has been adapted to many languages, including Javascript.

RxJS is a library that makes use of an Observable, which is a stream or a collection of items over time. It follows the Observer Pattern where an Observable is subscribed to by an Observer, to which it actively pushes its emissions. Most importantly with any FRP implementation, in order for an Observable to begin emitting items, a subscription needs to be established by an Observer. The source backing the Observable will not run until this is done — this concept is known as lazy loading.

Once an Observable emits an item, it can undergo a series of transformations before it reaches the Observer. The RxJS library contains a number of operators such as Observable.map(), Observable.delay(), Observable.concat(), and Observable.zip() that can be applied to the Observable in a chain-like fashion, making these transformations easy to do.

To help explain RxJS and its operators, I have included a tool commonly used to illustrate the operators called Marble diagrams. The line represents time, whereas the circles represent emissions that the Observable would emit to its subscribers. The end state of the Observable can either be a successful completion (demonstrated by a vertical line) or an error (demonstrated by an X).

Implementing RxJS

The project I chose to work on is one that I knew would introduce interesting challenges I could solve with RxJS while also being something I personally would use. The feature is something I would like to see integrated into Spotify which allows a user to search for a song and annotate it with comments, which are then saved.

Users of the feature can view a stream of tagged songs and can playback the songs with all users’ tags for that song made visible. Because I couldn’t actually make this feature a part of the Spotify application, I made an Angular2/Node.js app which makes use of the Spotify Web API and the 30 second song previews to illustrate my idea.

I made use of RxJS in varying degrees throughout the application, with the most complex implementation being the component which syncs displaying tags with music playback. The scope of this blogpost will cover some of the basic ways I utilized some operators to achieve things like data loading and transformation, and capturing user input. There will be another part to this post where I will explore some of the more complex chains that I built in this project.

Observable and Observer

The first component I would like to discuss is the public feed which loads a feed of all of the publicly tagged songs. This is a simple and common use case of RxJS in Angular2. When the component initializes, it uses a service that prepares an Observable that will make a backend call to fetch the songs from the database when it is subscribed to by an observer.

Here is what the service code looks like:

@Injectable()

export class TaggedSongsService {
  constructor(private http: Http) {}

  getAllTaggedSongs() {
    return this.http.get(‘/api/tagged_songs’);
  }
}

The http.get function is an Observable factory which will create our Observable. That Observable will make the GET request once an Observer is present and will emit the response returned from the backend. There are two important things to note here:

1. Being an observable, this GET request won’t run, until something subscribes to it.
2. If we subscribe to the same observable multiple times, it’ll make a backend call for every subscription, unlike a Promise that resolves only once in its lifetime.

If we look at the tagged songs list component which makes use of this service, upon initialization of the component is where I have subscribed to the observable.

ngOnInit(): void {
  this.taggedSongsService.getAllTaggedSongs().subscribe(songs => {
    this.songs = songs.tracks;
  });
}

When a user navigates to the public feed view, the component will get initialized and in turn will subscribe to the Observable created by the TaggedSongsService.

Simple Data Transformation

Ok, so the component initializes and subscribes to the Observable which will make a backend call and emit the response. However, I really only want the JSON body from the response, so how would I do this transformation? Luckily, there is another commonly used RxJS operator called map() that takes a mapping function. It transforms emissions from the upstream Observable by applying the mapping function and forwarding the result downstream. If we examine the below marble diagram we’ll see that the upstream Observable is emitting values 1, 2 and 3. The mapping function we pass is multiplying each emission by 10. As a result our Observer (downstream from the map) receives 10, 20 and 30.

In that sense we can use the map() operator to map a response to a JSON object. Here, I am taking the emission and extracting just the JSON to be passed along downstream. (i.e. I’m mapping a Response to a JSON object).

@Injectable()

export class TaggedSongsService {
  constructor(private http: Http) { }

  getAllTaggedSongs() {
    return this.http.get(‘/api/tagged_songs’)
      .map(res => res.json());
  }
}

Side-effects & Syncing Observables’ Emissions

So, we’ve discussed the idea of subscribing to an Observable and synchronously transforming an Observable‘s emissions using Observable.map(). Moving onto something a little more complicated, we can look at the component that allows a user to tag a song with comments. In order for a tag to be successfully added, the action is two-fold:

1. When the song has reached the place where a user wants to place a tag she will first click the Add Tag button.
2. A text box then pops up where she can enter the text followed by clicking a Submit Text button.

These two actions will be the subject of our next discussion.

Breaking it down to focus on the first step (clicking the Add Tag button), I have the following code.

this.position = Observable.fromEvent(this.button.nativeElement, ‘click’)
  .do(_ => {
    this.render.TextBox();
  })
  .map(_ => this.toCurrentPosition());

I start by creating an Observable that emits click events for my Add Tag button. Now every time the user clicks on that button, my Observable will emit an event. I don’t care so much about the actual emission, but I do want to do a couple of things any time this happens. First I use the Observable.do() operator to pop up a text box on the UI. The Observable.do() operator is generally used for synchronous side-effects you might need to do mid-chain. I then use the familiar Observable.map() to obtain two important pieces of information: the current time in the song and the position the tag needs to be on the Wavesurfer view. I return the two key pieces of information provided by other entities within the component, as we can see below:

toCurrentPosition() {
  return {
    time: waveSurfer.getCurrentTime(),
    currentPosition: getTagPosition()
  };
}

getTagPosition() {
  let currentPostion = getPosition();
  let waveWidth = getWaveWidth();
  return (currentPostion / waveWidth) * 100;
}

The second part of the tagging process is the actual tag text that is submitted via the textbox. I could have taken a similar approach and made onClick events on the Submit Text button an Observable. However, because what I am really interested in is the text in the textbox, I would have had to take an extra step to map the click event to the input text. Therefore, I’ve taken another approach where I instead use a regular onClick event which passes the text as an argument in the onClick function defined in the component. Once received in the component, the text is emitted through a Subject.

Note: We should usually avoid subjects since they can be tricky to deal with. Especially when building bigger chains. For this particular case though, I chose to use a Subject to demonstrate its uses.

A Subject is an Observable and an Observer, meaning you can subscribe to a Subject. For example, if you call subject.next(“hello”), then your Observer will receive “hello” as an emission. If you call subject.completed(), your Observer will receive the completed event, and the same goes when calling with an error.

Using a Subject like this is a way to make an Observable out of something that is not one. Note that in the SongComponent I have defined a global variable tagTextSubject and set its type to be a Subject.

export class SongComponent implements OnInit {
  clicks: Observable<any>;
  tagTextSubject:Subject<any> = new Subject();
}

When the submitText onClick function is executed with the input text, I call tagTextSubject.next(text), allowing the text value to be emitted from this Observable.

submitText(text: string): void {
  this.tagTextSubject.next(text);
}

Great, so now the text is the second Observable associated with a single tag. In order to create the tag object from the first Observable, emitting time and position, to the second Observable, emitting the user’s input, I can apply the Observable.zip() operator to sync the two observable emissions.

As illustrated below, the Observable.zip() operator takes the two Observables as the first two arguments and combine function as the third argument.

Observable.zip(
  this.position,
  this.tagTextSubject,
  (position, tagText) => {
    return {
      position: position,
      text: tagText
    }
  }
).subscribe(val => {
  this.renderAndStoreTag(val)
});

Observable.zip() joins the last two emissions from each Observable together. So the most recently emitted position item (with the current time and position of the tag) is zipped together with the most recently emitted tagTextSubject item (the text of the tag).

The third argument in Observable.zip() determines exactly how I want the emissions from the Observables joined. Here, I combine the first object containing the position and time of the tag with the text of the tag into a single tag object. Observable.zip() emits objects containing the tag info which are passed along to the renderAndStoreTag function. This function adds each tag object to an array of tags (which will be passed to the backend to save all tags upon clicking the save button) and also creates and shows the UI for the tag.

Asynchronous Data Transformation

Great, so we’ve just mastered a small RxJS chain — we’re nearly experts! At this point I would like to wrap up part one of this blog post by taking a step back to look at one more operator which is a bit more complicated. To do that we must backtrack a bit to the initialization of the SongComponent we were just in. Whenever the component is initialized, we create the following chain and subscribe to it:

this.route.params
  .switchMap(params => {
    return this.songService.getSong(params[‘track_id’]);
  })
  .subscribe(song => {
    this.song = song.name;
    this.image = song.album.images[1].url;
    this.artist = song.artists[0].name;
    this.waveSurfer.load(song.preview_url);
  }, error => console.log(error));

A user navigates to this view by clicking on the song title in the list of tagged songs in the Feed view. When this is done, the user is redirected to the song view and the song’s Spotify ID is passed along as a parameter of this url. However, we have no other information about the song at this point. Therefore, we need to map the id to a Spotify song object — but that means making an async call to the Spotify API to retrieve the song object by ID.

In the past we’ve used Observable.map() to synchronously map an emission to some other type. Here, since we don’t have the song object yet and have to retrieve it asynchronously, we can’t use Observable.map(). But fear not, there are a few operators that allow for asynchronous mapping in RxJS. They all behave in the same way but have semantic differences (a potential topic for another blog post). The general pattern is, we provide a mapping function that maps upstream emissions to an Observable. Whatever Observable we return, RxJS will subscribe to and forward its emissions downstream. For our purposes here, I have used Observable.switchMap().

Now, whenever the params Observable emits a song ID, I map that to an Observable created by the service that fetches the song details from backend. As a result, my Observer receives full song objects instead of just IDs thus allowing me to perform the proper initialization of the component with the pertinent song information.

That concludes my introduction to some basic implementations of RxJS. I hope you’ve learned a little bit about how easy it can be to implement FRP with a helpful library such as this. Please check back soon to read Part II of my journey in applying RxJS where I will focus on the complex chain that is the driving force behind the TaggedSongComponent view.

A Beginner’s Adventure with RxJS: Part I was originally published in Def Method on Medium, where people are continuing the conversation by highlighting and responding to this story.