Understanding Observable in Rx Programming with Higher‑Order Functions

Background: Learning Rx programming requires a solid grasp of the Observable concept. The article likens the learning process to learning to ride a bicycle – repeated failures lead to insight, but a clear explanation can shorten the path.

Observable: An Observable ("可观察者") is a data or event source that can be observed, similar to turning on a faucet after subscribing. The article uses the metaphor of a "parcel box" to illustrate how subscribing to an Observable yields a stream of data.

Higher‑order functions: A higher‑order function takes a function as an argument or returns one. Example:

function foo(arg){
    return function(){
        console.log(arg);
    };
}
const bar = foo("hello world");
bar(); // hello world

The returned function caches the value and executes it later, embodying the idea of delayed execution.

More examples of higher‑order functions:

function foo1(arg){
    return function(){
        console.log(arg + "?");
    };
}
function foo2(arg){
    return function(){
        console.log(arg + "!");
    };
}
const bar1 = foo1("hello world");
const bar2 = foo2("yes");
bar1();
bar2(); // hello world? yes!

"Parcel box" model for Observable implementations:

1. fromEvent – wraps addEventListener :

btn.addEventListener("click", callback);
rx.fromEvent(btn, "click").subscribe(callback);

Implementation:

function fromEvent(target, evtName){
    return function(callback){
        target.addEventListener(evtName, callback);
    };
}
const ob = fromEvent(btn, "click");
ob(console.log); // equivalent to subscribe

2. interval – wraps setInterval :

function interval(period){
    let i = 0;
    return function(callback){
        setInterval(period, () => callback(i++));
    };
}
const ob = interval(1000);
ob(console.log); // equivalent to subscribe

Both are higher‑order functions that produce "parcel boxes" (Observables) which can be opened (subscribed).

Combining parcel boxes – merge :

function merge(...obs){
    return function(callback){
        obs.forEach(ob => ob(callback)); // open all boxes
    };
}
const ob1 = fromEvent(btn, 'click'); // box 1
const ob2 = interval(1000);          // box 2
const ob = merge(ob1, ob2);           // big box
ob(console.log); // open the big box

The merged Observable opens all constituent Observables when subscribed.

Destroying (unsubscribing) a parcel box:

function fromEvent(target, evtName){
    return function(callback){
        target.addEventListener(evtName, callback);
        return function(){
            target.removeEventListener(evtName, callback);
        };
    };
}
const ob = fromEvent(btn, 'click');
const sub = ob(console.log); // open and get destroy function
sub(); // destroy the box

Similarly for interval :

function interval(period){
    let i = 0;
    return function(callback){
        const id = setInterval(period, () => callback(i++));
        return function(){
            clearInterval(id);
        };
    };
}
const ob = interval(1000);
const sub = ob(console.log);
sub(); // destroy

Destroying a merged Observable:

function merge(...obs){
    return function(callback){
        const subs = obs.map(ob => ob(callback)); // collect destroy functions
        return function(){
            subs.forEach(sub => sub());
        };
    };
}
const ob1 = fromEvent(btn, 'click');
const ob2 = interval(1000);
const ob = merge(ob1, ob2);
const sub = ob(console.log);
sub(); // destroy all boxes

The article concludes that Observable’s core actions are subscribe and unsubscribe, with additional actions (emit, complete/error) handled internally. It emphasizes that wrapping asynchronous functions into Observables provides a unified way to manage them.