Iterators and Observers in asynchronous javascript - part 1

Functions map, filter, forEach, takeUntil and concatAll

Suppose we have an array [1, 2, 3] and we want to add 1 to all its elements, we can use the function map to do so.

[1, 2, 3].map( x => x + 1 );

This gives a new array [2, 3, 4]. Remember function map does not change the original array. It transforms the original array and makes a new array with new elements.

Now if we want to filter an array we can use the function filter(criteria for filtering the array).

[1, 2, 3, 4].filter(x => x > 2);

This gives a new array as filter does not change the original array. It makes a new array and copies only those elements in that array that passes the test to make it in to the new array. After applying the filter above we get [3, 4].

Here is another example.

var getOnlyTopRatedMovies = userName =>
userName.playLists.
    map(playList =>
        playList.videos.
            filter(video => video.rating === 5.0)
           ).concatAll();
         
getOnlyTopRatedMovies(userName).
forEach(movie =>
    console.log(movie));


concatAll will flatten the two-dimensional array in to a one-dimensional array. concatAll only works on Observable of Observables and not on a single Observable. We don't want to use concatAll on an Observable that has infinite Observables as we will not be able to flatten them as the data inside the infinite Observable never ends. An analogy of such a situation would be traffic situation where there are three lanes and one lane is under repair so we are left with two lanes. We can't use a strategy that involves allowing traffic from one lane first until all the vehicles are finished and then allowing vehicles from the other lane because arrival of the vehicles on the lanes never finish and will create a blocking situation for one of the lanes. We have a liste of generes and then we have a list of all the films in that genre.

For each userName (user) we have an array representing all the generes allowed to that user. Then wihin each genre we have list of movies belonging
to that genre. So it makes a two-dimensional array. In the script above we loop on those genres and one by one and pass a list of movies in a genre
to the operator map to make a new array and copy those film names in to the new array that pass the criteria described in the function inside the
operator filter. So we get a two-dimensional array of all films that have been rated 5.0 and next will the function concatAll make a one-dimensional
array out of it. Next the forEach function will print them to the console one by one.

None of the functions used above alter the original array. map, filter, concatAll, forEach does not change the array they work on. They just
make new copies.

In the code below we are using the function takeUntil instead of filter but its function is the same as that of the filter.

Mouse Drags Collection

This code creates  a stream of all the mouseDrags events that occur on an elment in the DOM. We pass that DOM element to the function getElmentDrags.

A mouse drag is a series of events that happen between a mouse down and a mouse up. We will show how we can compose simple events to make new and more complex events using methods like filter, concatAll, map, takeUntil etc.

//elmt could be any elment e.g., a button , a picture etc.
var getElmentDrags = elmt => 
elmt.mouseDowns.
/* map functin is all about replacing. We want to replace mouseDown events
with all the mouseMoves event that occur between the mouseDown event and mouseup event.
    */
map(mouseDown =>
/* For each mouseDown we are detecting the mouseMoves on the document level
  untill the event mouseup occurs. For each mouseDown we are going to put
  in the stream a collection of all the mouseMoves events untill the event
  mouseUps occurs. So we are taking each item (mouseDown) in a collection (mouseDowns) and replacing it with another collection (mouseMoves) in the stream.
*/
document.mouseMoves. 
takeUntil(document.mouseUps)). 
/* So now we have a two-dimensional array as we an array of arrays that have
mouseMoves events between mouseDown and mouseUps events. We can flattern       this two--dimensional array in to a single-dimensional array using concatAll.
*/
concatAll();

getElmentDrags(image).forEach(pos => image.position = pos);

/* Here in forEach we are now consuming the data that we created in the collection above and
doing something wit that data. Here we are moving the position of the image so that
the image actually drags around.
       */

Here is the complete code for Mouse Drags Collection.

var getElementDrags = elmt => {
 elmt.mouseDowns = Observable.fromEvent(elmt, 'mousedown');
 elmt.mouseUps = Observable.fromEvent(elmt, 'mouseup');
 elmt.mouseMoves = Observable.fromEvent(elmt, 'mousemove');
 return elmt.mouseDowns.
  map(mouseDown =>
   document.mouseMoves.
    takeUntil(document.mouseUps)).
  concatAll();
};
//using forEach we now consume the data generated above and stored in the flattened Observable created by concatAll.
getElementDrags(image).forEach(pos => image.position = pos);

//Here is the method fromEvent

/* Converting Events to Observables. */
//fromEvent is a static method in the Observable.
Observable.fromEvent = function(dom, eventName){
    //returning Observable object
    return{
        forEach: function(observer){
            /* observer is an object with three things 1. onNext to push more data 2. onError to push error message 3. onCompleted to push "done" to consumers.*/
            var handler = (e) => observer.onNext(e);
            dom.addEventListener(eventName, handler);
            //returning Subscription object
            return{
                dispose: function(){
                    dom.removeEventListener(eventName, handler);
                }
            };
        }
    };
}

---

The takeUntil works like a filter as we are reducing the number of mouseMoves events by
recording only those events that are between mouseDown and mouseUps events. So takeUntil
recudces the number of items in a collection just like filter does.

We can diagrammatically show the takeUntill as below. We are using the notation {...1..2.....3}
to denote an Observable which is a stream in which the data arrives over time. In the diagram below {......1......2...........................3} is the source collection from which we want to consume data and {........................4} is the stop collection which means that we do not want to consume more data from Observable {......1......2...........................3} as soon as the Observable {........................4} has some data to be consumed by us.

-------------------------time-------------------->
{......1......2...........................3}.takeUntil(
{........................4})

gives the Observable
 {......1......2........}

The Observable {......1......2...........................3} pushed data and we have a filter takeUntil which consumes data from this Observable until the second Observable {........................4} arrives at which point takeUnitl stops consuming more data from the first Observable. takeUntil() consumed 1 and 2 when 4 arrives and at that point it stops consuming more so 3 does not make it in the newly created Observable created by function takeUntil(). Therefore, the resultant Observable has data 1 and 2 only.

This approach makes the unsubscribing from consuming data manually by calling unsubscribe redundant. Normally we consume data until some event is fired (e.g., mouse is moved etc.) and then we unhook our handler once the event is fired. In the takeUntil approach we do not need to call unsubscribe to unsubscribe from events. Create streams of data that complete when you want them to.  Create new events from existing events and make those new events end when you want them to end. Use takeUntil() to do this.

We do not wait for the onComplete to happen in the stop collection rather we stop consuming data from the source collection as soon as onNext() happen in the stop collection i.e., as soon as the stop collection has some data to be consumed for us and we call onNext() on it to consume that data. At this point the first Observable will call onCompletion to indicate that it is not going to push more data. Since we are not interested int he data from the stop collection, therefore, at this point method dispose() will be called on the stop collection.

If the stop collection or the source collection hits an error then the outer Observable (i.e., the Observable that we are creating) will end with that error. We always forward along the errors.

Note: concatAll does not exist on an array in javascript but you can write the function yourself.