摘要：本文是響應式編程第二章序列的深入研究這篇文章的學習筆記。函數科里化的基本應用，也是函數式編程中運算管道構建的基本方法。四資料參考函數式編程指南

本文是Rxjs 響應式編程-第二章：序列的深入研究這篇文章的學習筆記。
示例代碼托管在：http://www.github.com/dashnowords/blogs
更多博文：《大史住在大前端》目錄

文中使用到的一些基本運算符：

map-映射

filter-過濾

reduce-有限列聚合

scan-無限列聚合

flatMap-拉平操作（重點）

catch-捕獲錯誤

retry-序列重試

from-生成可觀測序列

range-生成有限的可觀測序列

interval-每隔指定時間發出一次順序整數

distinct-去除出現過的重復值

建議自己動手嘗試一下，記住就可以了，有過lodash使用經驗的開發者來說并不難。原文中使用flatMap轉換序列時有一處應該是手誤：

原文中在http請求拿到獲取到數據后，最初使用了forEach實現了手動流程管理，于是原文提出了優化設想，試圖探究如何依賴響應式編程的特性將手動的數據加工轉換改造為對流的轉換，好讓最終的消費者能夠拿到直接可用的數據，而不是得到一個響應后手動進行很多后處理。在代碼層面需要解決的問題就是，如何在不使用手動遍歷的前提下將一個有限序列中的數據逐個發給訂閱者，而不是一次性將整個數據集發過去。

假設我們現在并不知道有flatMap這樣一個可以使用的方法，那么先來做一些嘗試：

var quakes = Rx.Observable.create(function(observer) {
    //模擬得到的響應流
    var response = { 
       features:[{
        earth:1
       },{
        earth:2
       }],
       test:1
    }
/*  最初的手動遍歷代碼  
    var quakes = response.features;
        quakes.forEach(function(quake) {
            observer.onNext(quake);
        });*/

    observer.onNext(response);
})
//為了能將features數組中的元素逐個發送給訂閱者，需要構建新的流
.map(dataset){
    return Rx.Observable.from(dataset.features)
}

當我們訂閱quakes這個事件流的時候，每次都會得到另一個Observable，它是因為數據源經過了映射變換，從數據變成了可觀測對象。那么為了得到最終的序列值，就需要再次訂閱這個Observable，這里需要注意的是可觀測對象被訂閱前是不啟動的，所以不用擔心它的時序問題。

quakes.subscribe(function(data){
    data.subscribe(function(quake){
        console.log(quake);
    })
})；

如果將Observable看成一個盒子，那么每一層盒子只是實現了流程控制功能性的封裝，為了取得真正需要使用的數據，最終的訂閱者不得不像剝洋蔥似的通過subscribe一層層打開盒子拿到最里面的數據，這樣的封裝性對于數據在流中的傳遞具有很好的隔離性，但是對最終的數據消費者而言，卻是一件很麻煩的事情。

這時flatMap運算符就派上用場了，它可以將冗余的包裹除掉，從而在主流被訂閱時直接拿到要使用的數據，從大理石圖來直觀感受一下flatMap:

乍看之下會覺得它和merge好像是一樣的，其實還是有一些區別的。merge的作用是將多個不同的流合并成為一個流，而上圖中A1，A2，A3這三個流都是當主流A返回數據時新生成的，可以將他們想象為A的支流，如果你想在支流里撈魚，就需要在每個支流里布網，而flatMap相當于提供了一張大網，將所有A的支流里的魚都給撈上來。

所以在使用了flatMap后，就可以直接在一級訂閱中拿到需要的數據了：

var quakes = Rx.Observable.create(function(observer) {
    var response = { 
       features:[{
        earth:1
       },{
        earth:2
       }],
       test:1
    }
    observer.onNext(response);
}).flatMap((data)=>{
    return Rx.Observable.from(data.features);
});

quakes.subscribe(function(quake) {
   console.log(quake)
});

如果本節的基本知識你尚不熟悉，可以通過javascript基礎修煉（8）——指向FP世界的箭頭函數這篇文章來簡單回顧一下函數式編程的基本知識，然后再繼續后續的部分。

/*map運算符的作用
*對所有容器類而言，它相當于打開容器，進行操作，然后把容器再蓋上。
*Container在這里只是一個抽象定義，為了看清楚它對于容器中包含的值意味著什么。
*你會發現它其實就是Observable的抽象原型。
*/
Container.prototype.map = function(f){
  return Container.of(f(this.__value))
}

//基本的科里化函數
var curry = function(fn){
  args = [].slice.call(arguments, 1);
  return function(){
     [].push.apply(args, arguments);
     return fn.apply(this, args);
  }
}

//map pointfree風格的map運算符
var map = curry(function(f, any_functor_at_all) {
  return any_functor_at_all.map(f);
});

/*compose函數組合方法
*運行后返回一個新函數，這個函數接受一個參數。
*函數科里化的基本應用，也是函數式編程中運算管道構建的基本方法。
*/
var compose = function (f, g) {
    return function (x) {
        return f(g(x));
    }
};
/*IO容器
*一個簡單的Container實現，用來做流程管理
*這里需要注意，IO實現的作用是函數的緩存，且總是返回新的IO實例
*可以看做一個簡化的Promise,重點是直觀感受一下它作為函數的
*容器是如何被使用的，對于理解Observable有很大幫助
*/
var IO = function(f) {
  this.__value = f;
}

IO.of = function(x) {
  return new IO(function() {
    return x;
  });
}

IO.prototype.map = function(f) {
  return new IO(compose(f, this.__value));
}

如果上面的基本知識沒有問題，那么就繼續。

現在來實現這樣一個功能，讀入一個文件的內容，將其中的a字符全部換成b字符，接著存入另一個文件，完成后在控制臺輸出一個消息，為了更明顯地看到數據容器的作用，我們使用同步方法并將其包裹在IO容器中，然后利用函數式編程:

var fs = require("fs");

//讀取文件
var readFile = (filename)=>IO.of(fs.readFileSync(filename,"utf-8"));

//轉換字符
var transContent = (content)=>IO.of((content)=>content.replace("a","b"));

//寫入字符串
var writeFile = (content)=>IO.of(fs.writeFileSync("dest.txt",content));

當具體的函數被IO容器包裹起來而實現延遲執行的效果時，就無法按原來的方式使用compose( )運算符直接對功能進行組合，因為readFile函數運行時的輸出結果（一個io容器實例）和transContent函數需要的參數類型（字符串）不再匹配，在不修改原有函數定義的前提下，函數式編程中采用的做法是使用map操作符來預置一個參數：

/*
*map(transContent)是一個高階函數，它的返回函數就可以接收一個容器實例，
*并對容器中的內容執行map操作。
*/
var taskStep12 = compose(map(transContent), readFile);

這里比較晦澀，涉及到很多功能性函數的嵌套，建議手動推導一下taskStep12這個變量的值，它的結構是這樣一種形式：

io{
    __value:io{
        __value:someComposedFnExpression
    }
}

如果試圖一次性將所有的步驟組合在一起，就需要采用下面的形式：

var task = compose(map(map(writeFile)),map(transContent),readFile);
//組合后的task形式就是
//io{io{io{__value:someComposedFnExpression}}}

問題已經浮出水面了，每多加一個針對容器操作的步驟，書寫時就需要多包裹一層map，而運行時就需要多進入一層才能觸及組合好的可以實現真正功能的函數表達式，真的是很麻煩。

 提示一：
現在來回想一下原示例中的Observable對象，將其看做是一個容器（含有map類方法），那么如果map方法調用時傳入的參數是一個運行時會生成新的Observable對象的方法時，就會產生Observable嵌套，得到observable{observable{.....}}這樣的結構，那么在最終的數據消費者通過subscribe方法訂閱數據時，就不得不用很多個subscribe才能拿到實際需要的數據。
提示二:
沒有相關經驗的讀者在使用pointfree風格的map操作符時可能會感到非常不適應，如果你覺得它很難理解，也可以嘗試直接使用IO.prototype.map這種鏈式調用風格的寫法將上例中的三個步驟組合在一起來查看最后的結果，畢竟在Rxjs中常使用的也就是Observable這一個容器類。

當我們看到問題所在后就不難發現，其實這個問題的解決方法并不復雜，我們要做的不過就是在必要的時候合并內容的容器，為此來定義兩個合并運算的方法：

//鏈式調用風格
IO.prototype.join =  function(){
    return this.isNothing() ? IO.of(null):this.__value;
}
//pointfree風格運算符
var join = (m)=>m.join();

這里引入一個新的概念Monad，它的定義是可以被展平的容器，也就是說擁有join和of方法并遵循一定規則的容器，都是Monad，在這種設定下，3.1中的示例就可以被改寫為下面的形式：

var task = compose(join,map(writeFile),join,map(transContent),readFile);

不難發現map和join總是需要成對出現的，那么再利用函數科里化的技巧將map和join連起來：

var chain = curry(function(f,m){
    return m.map(f).join();
})

那么組合后的函數就變成了下面的形式：

var task = compose(chain(writeFile),chain(transContent),readFile);

這里的chain，就是FlatMap。

最后將上面幾種形式放在一起再來回顧一下：

//原有形式
var task = compose(map(map(writeFile)),map(transContent),readFile);

//map-join形式
var task = compose(join,map(writeFile),join,map(transContent),readFile);

//chain形式(flatMap)
var task = compose(chain(writeFile),chain(transContent),readFile);

如果理解了這幾種形式，就不難理解flatMap的拉平效應了，所謂flatMap，說白了其實就是將容器展開的一種操作。

flatMap所解決問題，是在函數式編程引入了Functor的概念將邏輯函數包裹在容器中后才產生的，那么這種容器概念的引入對函數式編程到底有什么意義，筆者尚未搞清楚，相關內容留作以后補充。

《javascript函數式編程指南》https://llh911001.gitbooks.io/mostly-adequate-guide-chinese/content/