摘要：建立一個實時地震我們將為地震儀表板應用程序構建服務器和客戶端部件，實時記錄地震的位置并可視化顯示。添加地震列表新儀表板的第一個功能是顯示地震的實時列表，包括有關其位置，大小和日期的信息。

Rxjs 響應式編程-第一章：響應式
Rxjs 響應式編程-第二章：序列的深入研究
Rxjs 響應式編程-第三章: 構建并發程序
Rxjs 響應式編程-第四章 構建完整的Web應用程序
Rxjs 響應式編程-第五章 使用Schedulers管理時間
Rxjs 響應式編程-第六章 使用Cycle.js的響應式Web應用程序

在本章中，我們將構建一個典型的Web應用程序，在前端和后端使用RxJS。我們將轉換文檔對象模型（DOM）并使用Node.js服務器中的WebSockets進行客戶端 - 服務器通信。

對于用戶界面位，我們將使用RxJS-DOM庫，這是由RxJS制作的同一團隊的庫，它提供了方便的Operator來處理DOM和瀏覽器相關的東西，這將使我們的編程更簡潔。對于服務器部分，我們將使用兩個完善的節點庫，并將一些API與Observables包裝在一起，以便在我們的應用程序中使用它們。

在本章之后，您將能夠使用RxJS以聲明方式構建用戶界面，使用我們目前為止看到的技術并將它們應用于DOM。 您還可以在任何Node.js項目中使用RxJS，并且能夠在任何項目中使用反應式編程和RxJS。

我們將為地震儀表板應用程序構建服務器和客戶端部件，實時記錄地震的位置并可視化顯示。我們將在Node.js中構建服務器，并且改進我們的應用程序，使其更具互動性和更充足的信息量。

一開始的代碼如下：

examples_earthquake/code1_3.js

var quakes = Rx.Observable
.interval(5000)
.flatMap(function() {
    return Rx.DOM.jsonpRequest({
        url: QUAKE_URL,
        jsonpCallback: "eqfeed_callback"
    }).retry(3);
})
.flatMap(function(result) {
    return Rx.Observable.from(result.response.features);
})
.distinct(function(quake) { return quake.properties.code; });

quakes.subscribe(function(quake) {
    var coords = quake.geometry.coordinates;
    var size = quake.properties.mag * 10000;
    L.circle([coords[1], coords[0]], size).addTo(map);
});

這段代碼已經有一個潛在的錯誤：它可以在DOM準備好之前執行，每當我們嘗試在代碼中使用DOM元素時就會拋出錯誤。我們想要的是在觸發DOMContentLoaded事件之后加載我們的代碼，這表示瀏覽器已經準備好dom了。

RxJS-DOM提供Rx.DOM.readyObservable，當觸發DOMContentLoaded時，它會發出一次。 因此，讓我們將代碼包裝在initialize函數中，并在訂閱Rx.DOM.ready時執行它：

examples_earthquake_ui/code1.js

function initialize() {
    var quakes = Rx.Observable
    .interval(5000)
    .flatMap(function() {
    return Rx.DOM.jsonpRequest({
        url: QUAKE_URL,
        jsonpCallback: "eqfeed_callback"
    });
    })
    .flatMap(function(result) {
        return Rx.Observable.from(result.response.features);
    })
    .distinct(function(quake) { return quake.properties.code; });
    
    quakes.subscribe(function(quake) {
        var coords = quake.geometry.coordinates;
        var size = quake.properties.mag * 10000;
        L.circle([coords[1], coords[0]], size).addTo(map);
    });
}
Rx.DOM.ready().subscribe(initialize);

接下來，我們將在HTML中添加一個空表，我們將在下一部分填充地震數據：

    

        

            
Location
            
Magnitude
            
Time
        
    
    

    

有了這個，我們準備開始為我們的儀表板編寫新代碼。

新儀表板的第一個功能是顯示地震的實時列表，包括有關其位置，大小和日期的信息。此列表的數據與來自USGS網站的地圖相同。我們首先創建一個函數，在給定props對象參數的情況下返回一個row元素：

examples_earthquake_ui/code2.js

function makeRow(props) {
    var row = document.createElement("tr");
    row.id = props.net + props.code;
    var date = new Date(props.time);
    var time = date.toString();
    [props.place, props.mag, time].forEach(function(text) {
        var cell = document.createElement("td");
        cell.textContent = text;
        row.appendChild(cell);
    });
    return row;
}

props參數與我們從USGS站點檢索的JSON中的properties屬性相同。

為了生成行，我們將再次訂閱地震Observable。此訂閱會在表格中為每次收到的新地震創建一行。 我們在initialize函數的末尾添加代碼：

examples_earthquake_ui/code2.js

var table = document.getElementById("quakes_info");
quakes
.pluck("properties")
.map(makeRow)
.subscribe(function(row) { table.appendChild(row); });

pluck運算符從每個地震對象中提取屬性值，因為它包含makeRow所需的所有信息。 然后我們將每個地震對象映射到makeRow，將其轉換為填充的HTML tr元素。 最后，在訂閱中，我們將每個發出的行追加到我們的table中。

每當我們收到地震數據時，這應該得到一個數據稠密的表格。

看起來不錯，而且很容易！不過，我們可以做一些改進。首先，我們需要探索RxJS中的一個重要概念：冷熱Observable。

無論Observers是否訂閱它們，“熱”Observable都會發出值。另一方面，“冷”Observables從Observer開始訂閱就發出整個值序列。

訂閱熱Observable的Observer將接收從訂閱它的確切時刻發出的值。在那一刻訂閱的每個其他Observer將收到完全相同的值。 這類似于JavaScript事件的工作方式。

鼠標事件和股票交易代碼是熱的Observables的例子。在這兩種情況下，Observable都會發出值，無論它是否有訂閱者，并且在任何訂閱者收聽之前可能已經生成了值。這是一個例子：

hot_cold.js

var onMove = Rx.Observable.fromEvent(document, "mousemove");
var subscriber1 = onMove.subscribe(function(e) {
    console.log("Subscriber1:", e.clientX, e.clientY);
});
var subscriber2 = onMove.subscribe(function(e) {
    console.log("Subscriber2:", e.clientX, e.clientY);
});

// Result:
// Subscriber1: 23 24
// Subscriber2: 23 24
// Subscriber1: 34 37
// Subscriber2: 34 37
// Subscriber1: 46 49
// Subscriber2: 46 49
// ...

在該示例中，兩個訂閱者在發出Observable時都會收到相同的值。 對于JavaScript程序員來說，這種行為感覺很自然，因為它類似于JavaScript事件的工作方式。

現在讓我們看看冷Observables是如何工作的。

只有當Observers訂閱它時，冷Observable才會發出值。

例如，Rx.Observable.range返回一個冷Observable。訂閱它的每個新觀察者都將收到整個范圍：

hot_cold.js

function printValue(value) {
    console.log(value);
}
var rangeToFive = Rx.Observable.range(1, 5);
var obs1 = rangeToFive.subscribe(printValue); // 1, 2, 3, 4, 5
var obs2 = Rx.Observable
.delay(2000)
.flatMap(function() {
    return rangeToFive.subscribe(printValue); // 1, 2, 3, 4, 5
});

了解我們何時處理熱或冷的Observable對于避免細微和隱藏的錯誤至關重要。例如，Rx.Observable.interval返回一個Observable，它以固定的時間間隔生成一個遞增的整數值。 想象一下，我們想用它來將相同的值推送給幾個觀察者。 我們可以像這樣實現它：

hot_cold.js

var source = Rx.Observable.interval(2000);

var observer1 = source.subscribe(function (x) {
    console.log("Observer 1, next value: " + x);
});

var observer2 = source.subscribe(function (x) {
    console.log("Observer 2: next value: " + x);
});

輸出

Observer 1, next value: 0
Observer 2: next value: 0
Observer 1, next value: 1
Observer 2: next value: 1
...

這似乎沒什么問題。 但現在想象我們需要第二個用戶在第一個用戶加入后三秒鐘加入：

hot_cold.js

var source = Rx.Observable.interval(1000);
var observer1 = source.subscribe(function (x) {
    console.log("Observer 1: " + x);
});
setTimeout(function() {
    var observer2 = source.subscribe(function (x) {
        console.log("Observer 2: " + x);
    });
}, 3000);

輸出

Observer 1: 0
Observer 1: 1
Observer 1: 2
Observer 1: 3
Observer 2: 0
Observer 1: 4
Observer 2: 1
...

現在我們看到有些東西真的沒了。三秒后訂閱時，observer2接收源已經推送過的所有值，而不是從當前值開始并從那里繼續，因為Rx.Observable.interval是一個冷Observable。 如果熱和冷Observables之間的的區別不是很清楚的話，那么這樣的場景可能會令人驚訝。

如果我們有幾個Observers訂閱冷的Observable，他們將收到相同序列值的副本。嚴格來說，盡管觀察者共享相同的Observable，但它們并沒有共享相同的值序列。如果我們希望Observers共享相同的序列，我們需要一個熱的Observable。

我們可以使用publish將冷的Observable變成熱的。調用publish會創建一個新的Observable，它充當原始Observable的代理。它通過訂閱原始版本并將其收到的值推送給訂閱者來實現。

已發布的Observable實際上是一個ConnectableObservable，它有一個名為connect的額外方法，我們調用它來開始接收值。 這允許我們在開始運行之前訂閱它：

hot_cold.js

// Create an Observable that yields a value every second
var source = Rx.Observable.interval(1000);
var publisher = source.publish();
// Even if we are subscribing, no values are pushed yet.
var observer1 = publisher.subscribe(function (x) {
    console.log("Observer 1: " + x);
});
// publisher connects and starts publishing values
publisher.connect();
setTimeout(function() {
    // 5 seconds later, observer2 subscribes to it and starts receiving
    // current values, not the whole sequence.
    var observer2 = publisher.subscribe(function (x) {
        console.log("Observer 2: " + x);
    });
}, 5000);

讓我們回到我們的地震示例。到目前為止，我們的代碼看起來很合理;我們有一個帶有兩個訂閱的Observable地震：一個在地圖上繪制地震，另一個在表格中列出地震。

但我們可以使代碼更有效率。 通過讓兩個地震用戶，我們實際上要求兩次數據。 您可以通過在quakes的flatMap操作符中放入一個console.log來檢查。

發生這種情況是因為quakes是一個冷Observable，并且它會將所有值重新發送給每個新訂閱者，因此新訂閱意味著新的JSONP請求。這會通過網絡請求兩次相同的資源來影響我們的應用程序性能。

對于下一個示例，我們將使用`share·運算符，當Observers的數量從0變為1時，它自動創建對Observable的預訂。 這使我們免于重新連接：

examples_earthquake_ui/code2.js

var quakes = Rx.Observable
.interval(5000)
.flatMap(function() {
    return Rx.DOM.jsonpRequest({
        url: QUAKE_URL,
        jsonpCallback: "eqfeed_callback"
    });
})
.flatMap(function(result) {
    return Rx.Observable.from(result.response.features);
})
.distinct(function(quake) { return quake.properties.code; })
.share()

現在地震的行為就像一個熱的Observable，我們不必擔心我們連接多少觀察者，因為他們都會收到完全相同的數據。

我們之前的代碼運行良好，但請注意，每次我們收到有關地震的信息時都會插入一個tr節點。 這是低效的，因為每次插入我們都會修改DOM并導致重新繪制頁面，使瀏覽器不必要地計算新布局。 這可能會導致性能下降。

理想情況下，我們會批處理幾個傳入的地震對象，并每隔幾秒插入一批地震對象。手動實現會很棘手，因為我們必須保留計數器和元素緩沖區，我們必須記住每次批量重置它們。 但是使用RxJS，我們可以使用一個基于緩沖區的RxJS運算符，比如bufferWithTime。

使用bufferWithTime，我們可以緩沖傳入的值，并在每x個時間段將它們作為數組釋放：

examples_earthquake_ui/code3.bufferWithTime.js

var table = document.getElementById("quakes_info");
quakes
.pluck("properties")
.map(makeRow)
? .bufferWithTime(500)
? .filter(function(rows) { return rows.length > 0; }
.map(function(rows) {
var fragment = document.createDocumentFragment();
rows.forEach(function(row) {
? fragment.appendChild(row);
});
return fragment;
})
.subscribe(function(fragment) {
? table.appendChild(fragment);
});

這是新代碼中正在發生的事情：

B緩存每個傳入值并每500毫秒釋放一批值。

無論如何，bufferWithTime每500ms執行一次，如果沒有傳入值，它將產生一個空數組。 我們會過濾掉這些空數組。

我們將每一行插入一個文檔片段，這是一個沒有父文檔的文檔。這意味著它不在DOM中，并且修改其內容非常快速和有效。

最后，我們將片段附加到DOM。附加片段的一個優點是它被視為單個操作，只會導致一次重繪。 它還將片段的子元素附加到我們附加片段本身的同一元素。

使用緩沖區和片段，我們設法保持行插入性能，同時保持應用程序的實時性（最大延遲為半秒）。 現在我們已準備好為我們的儀表板添加下一個功能：交互性！

我們現在在地圖上和列表中發生地震，但兩個表示之間沒有相互作用。例如，每當我們點擊列表上的地圖時，就可以在地圖上居中地震，并在我們將鼠標移動到其行上時突出顯示地圖上帶圓圈的地震。 我們開始吧。

在Leaflet中，您可以在地圖上繪制并將繪圖放在各自的圖層中，以便您可以多帶帶操作它們。 讓我們創建一組名為quakeLayer的圖層，我們將存儲所有地震圈。每個圓圈都是該組內的一個圖層。 我們還將創建一個對象codeLayers，我們將存儲地震代碼和內部圖層ID之間的相關性，以便我們可以通過地震ID來查找圓圈：

examples_earthquake_ui/code3.js

var codeLayers = {};
var quakeLayer = L.layerGroup([]).addTo(map);

現在，在初始化內部的地震Observable訂閱中，我們將每個圓圈添加到圖層組并將其ID存儲在codeLayers中。 如果這看起來有點錯綜復雜，那是因為這是Leaflet允許我們在地圖中引用圖層的唯一方式。

examples_earthquake_ui/code3.js

quakes.subscribe(function(quake) {
    var coords = quake.geometry.coordinates;
    var size = quake.properties.mag * 10000;
    var circle = L.circle([coords[1], coords[0]], size).addTo(map);
    quakeLayer.addLayer(circle);
    codeLayers[quake.id] = quakeLayer.getLayerId(circle);
});

我們現在創建懸停效果。我們將編寫一個新函數isHovering，它返回一個Observable，它發出一個布爾值，表示在任何給定時刻鼠標是否在特定地震圈上：

examples_earthquake_ui/code3.js

? var identity = Rx.helpers.identity;
function isHovering(element) {
? var over = Rx.DOM.mouseover(element).map(identity(true));
? var out = Rx.DOM.mouseout(element).map(identity(false));
? return over.merge(out);
}

Rx.helpers.identity是定義函數。 給定參數x，它返回x。 這樣我們就不必編寫返回它們收到的值的函數。

over是一個Observable，當用戶將鼠標懸停在元素上時會發出true。

out是一個Observable，當用戶將鼠標移動到元素之外時，它會發出false。

isHovering將over和out合并，返回一個Observable，當鼠標懸停在元素上時發出true，當它離開時返回false。

使用isHovering，我們可以修改創建rows的訂閱，這樣我們就可以在創建時訂閱每行中的事件：

examples_earthquake_ui/code3.js

var table = document.getElementById("quakes_info");
quakes
.pluck("properties")
.map(makeRow)
.bufferWithTime(500)
.filter(function(rows) { return rows.length > 0; })
.map(function(rows) {
    var fragment = document.createDocumentFragment();
    rows.forEach(function(row) {
        fragment.appendChild(row);
    });
    return fragment;
})
.subscribe(function(fragment) {
    var row = fragment.firstChild; // Get row from inside the fragment
    ? var circle = quakeLayer.getLayer(codeLayers[row.id]);
    ? isHovering(row).subscribe(function(hovering) {
        circle.setStyle({ color: hovering ? "#ff0000" : "#0000ff" });
    });
    ? Rx.DOM.click(row).subscribe(function() {
        map.panTo(circle.getLatLng());
    });
    table.appendChild(fragment);
})

我們使用從行元素獲得的ID在地圖上獲取地震的圓元素。 有了它，codeLayers為我們提供了相應的內部ID，它使用quakeLayer.getLayer獲取了circle元素。

我們用當前行調用isHovering，然后我們訂閱生成的Observable。 如果懸停參數為真，我們會將圓圈畫成紅色; 不然，它會是藍色的。

我們訂閱了從當前行中的click事件創建的Observable。 單擊列表中的行時，地圖將以地圖中相應圓圈為中心。

經驗豐富的前端開發人員知道在頁面上創建許多事件是導致性能不佳的一個因素。 在前面的示例中，我們為每一行創建了三個事件。 如果我們在列表中獲得100次地震，我們將在頁面周圍浮動300個事件，只是為了做一些亮點突出工作！ 這對于表現來說太糟糕了，我們可以做得更好。

因為DOM中的事件總是冒泡（從子元素到父元素），前端開發人員中一個眾所周知的技術是避免將鼠標事件多帶帶附加到多個元素，而是將它們附加到父元素。 一旦在父項上觸發了事件，我們就可以使用事件的target屬性來查找作為事件目標的子元素。

因為我們需要為事件click和mouseover提供類似的功能，所以我們將創建一個函數getRowFromEvent：

examples_earthquake_ui/code3.pairwise.js

function getRowFromEvent(event) {
return Rx.Observable
.fromEvent(table, event)
? .filter(function(event) {
    var el = event.target;
    return el.tagName === "TD" && el.parentNode.id.length;
})
? .pluck("target", "parentNode")
? .distinctUntilChanged();
}

getRowFromEvent為我們提供了事件發生的表行。 以下是詳細信息：

我們確保在表格單元格中發生事件，并檢查該單元格的父級是否是具有ID屬性的行。 這些行是我們用地震ID標記的行。

pluck運算符在element的target屬性中提取嵌套屬性parentNode。

這可以防止多次獲得相同的元素。 例如，使用mouseover事件會發生很多事情。

examples_earthquake_ui/code3.pairwise.js

在上一節中，我們在每行上附加事件mouseover和mouseout，以便在每次鼠標輸入或退出行時更改地震圈顏色。 現在，我們將僅使用桌面上的mouseover事件，并結合方便的pairwise運算符：

examples_earthquake_ui/code3.pairwise.js

getRowFromEvent("mouseover")
.pairwise()
.subscribe(function(rows) {
    var prevCircle = quakeLayer.getLayer(codeLayers[rows[0].id]);
    var currCircle = quakeLayer.getLayer(codeLayers[rows[1].id]);
    prevCircle.setStyle({ color: "#0000ff" });
    currCircle.setStyle({ color: "#ff0000" });
});

pairwise將每個發射值與先前在陣列中發射的值進行分組。 因為我們總是獲得不同的行，所以成對將始終產生鼠標剛剛離開的行和鼠標現在懸停的行。 有了這些信息，就可以相應地為每個地震圈著色。

處理click事件更簡單：

examples_earthquake_ui/code3.pairwise.js

getRowFromEvent("click")
.subscribe(function(row) {
    var circle = quakeLayer.getLayer(codeLayers[row.id]);
    map.panTo(circle.getLatLng());
});

我們可以回到訂閱quakes來生成行：

examples_earthquake_ui/code3.pairwise.js

quakes
.pluck("properties")
.map(makeRow)
.subscribe(function(row) { table.appendChild(row); });

我們的代碼現在更加干凈，并且它不依賴于別處的row。 如果沒有row，getRowFromEvent將不會嘗試產生任何item。

更重要的是，我們的代碼現在非常高效。 無論我們檢索的地震信息量如何，我們總是只有一個鼠標懸停事件和單擊事件，而不是數百個事件。

我們為地震制作實時儀表板的計劃的第二部分是從Twitter添加與地球上發生的不同地震有關的報告和信息。 為此，我們將創建一個小型Node.js程序，該程序將獲取與地震相關的文章流。

讓我們開始配置我們的Node.js應用程序吧。除了RxJS，我們將使用兩個第三方模塊：ws和twit。這種類似的模塊都是讓我們保持最少的代碼。

首先，讓我們為我們的應用程序創建一個文件夾，并安裝我們將使用的模塊。 （請注意，npm命令的輸出可能會因軟件包的當前版本而異。）

現在我們準備開始構建我們的應用程序了。讓我們在tweet_stream文件夾中創建一個名為index.js的新文件來加載我們將使用的模塊：

examples_earthquake_ui/tweet_stream/index.js

var WebSocketServer = require("ws").Server; var Twit = require("twit");
var Rx = require("rx");

要使用Twitter API，您需要在Twitter網站中請求使用者密鑰和訪問令牌。 完成后，使用配置對象創建一個新的Twit對象，如下所示：

examples_earthquake_ui/tweet_stream/index.js

var T = new Twit({
    consumer_key: "rFhfB5hFlth0BHC7iqQkEtTyw",
    consumer_secret: "zcrXEM1jiOdKyiFFlGYFAOo43Hsz383i0cdHYYWqBXTBoVAr1x", 
    access_token: "14343133-nlxZbtLuTEwgAlaLsmfrr3D4QAoiV2fa6xXUVEwW9", 
    access_token_secret: "57Dr99wECljyyQ9tViJWz0H3obNG3V4cr5Lix9sQBXju1"
});

現在我們可以創建一個函數onConnect，它將完成搜索推文和將來與客戶端通信的所有工作，并且我們可以啟動一個WebSocket服務器，一旦WebSocket連接并準備好就會調用onConnect：

examples_earthquake_ui/tweet_stream/index.js

function onConnect(ws) {
    console.log("Client connected on localhost:8080");
}
var Server = new WebSocketServer({ port: 8080 });
Rx.Observable.fromEvent(Server, "connection").subscribe(onConnect);

我們現在可以啟動我們的應用程序，它應該在端口8080上啟動WebSocket連接：

~/tweet_stream$ node index.js

由于我們尚未將任何瀏覽器連接到此服務器，因此尚未打印有關客戶端連接的消息?，F在讓我們切換到dashboard的代碼并執行此操作。我們將在RxJS-DOM中使用fromWebSocket運算符：

examples_earthquake_ui/code4.js

function initialize() {
    var socket = Rx.DOM.fromWebSocket("ws://127.0.0.1:8080"); ...

在前面的代碼中，fromWebSocket創建一個Subject，作為WebSocket服務器的消息的發送者和接收者。 通過調用socket.onNext，我們將能夠向服務器發送消息，通過訂閱套接字，我們將收到服務器發送給我們的任何消息。

我們現在可以發送包含我們收到的地震數據的服務器消息：

examples_earthquake_ui/code4.js

quakes.bufferWithCount(100)
.subscribe(function(quakes) {
    console.log(quakes);
    var quakesData = quakes.map(function(quake) {
        return {
            id: quake.properties.net + quake.properties.code,
            lat: quake.geometry.coordinates[1],
            lng: quake.geometry.coordinates[0],
            mag: quake.properties.mag
        };
    });
    socket.onNext(JSON.stringify({quakes: quakesData }));
});

我們可以為來自服務器的消息設置訂閱者：

examples_earthquake_ui/code4.js

socket.subscribe(function(message) {
    console.log(JSON.parse(message.data));
});

現在，當我們重新加載瀏覽器時，客戶端消息應出現在服務器終端中：

~/tweet_stream$ node index.js
Client connected on localhost:8080

太棒了！ 一旦開始從遠程JSONP資源接收地震，瀏覽器就應該向服務器發送命令。 但是現在，服務器完全忽略了這些消息。 是時候回到我們的推文流代碼并用它們做點什么了。

首先，我們將連接到從瀏覽器客戶端到達服務器的消息事件。 每當客戶端發送消息時，WebSocket服務器都會發出包含消息內容的消息事件。 在我們的例子中，內容是一個JSON字符串。

我們可以在onConnect函數中編寫以下代碼：

examples_earthquake_ui/tweet_stream/index.js

var onMessage = Rx.Observable.fromEvent(ws, "message")
.subscribe(function(quake) {
    quake = JSON.parse(quake);
    console.log(quake);
});

如果我們重新啟動服務器（終端中的Ctrl-C）并重新加載瀏覽器，我們應該會看到終端上的地震細節打印出來。這是完美的。 現在我們已經準備好開始尋找與我們的地震有關的推文了。

我們正在使用Node.js twit的流式Twitter客戶端連接到Twitter和搜索推文。 從現在開始，服務器中的所有代碼都將在onConnect函數內部發生，因為它假定已經建立了與WebSocket的連接。 讓我們初始化推文流：

examples_earthquake_ui/tweet_stream/index.js

var stream = T.stream("statuses/filter", {
    track: "earthquake",
    locations: []
});

這告訴我們的Twit實例T開始流式傳輸Twitter狀態，按關鍵字地震過濾。 當然，這是非常通用的，而不是與現在發生的地震直接相關。 但請注意空位置數組。 這是一個緯度和經度邊界的數組，我們可以用它們按地理位置過濾推文，以及地震一詞。 那更加具體！ 好的，讓我們訂閱這個流并開始向瀏覽器發送推文：

examples_earthquake_ui/tweet_stream/index.js

Rx.Observable.fromEvent(stream, "tweet").subscribe(function(tweetObject) {
    ws.send(JSON.stringify(tweetObject), function(err) {
        if (err) {
            console.log("There was an error sending the message");
        }
    });
});

如果我們重新啟動服務器并重新加載瀏覽器，我們應該在瀏覽器中收到推文，開發面板中的控制臺應該打印推文。

這些推文尚未按地震位置進行過濾。 為此，我們需要對收到的每一條地震信息做以下事情：

取每個地震的經度和緯度對的震中坐標，創建一個邊界框，界定我們認為與地震相關的推文的地理區域。

累積所有邊界坐標，以便發送給客戶端的推文與地圖上的地震保持相關。

每次收到新地震的消息時，都會使用新坐標更新twit流。

這是一種方法：

examples_earthquake_ui/tweet_stream/index.js

Rx.Observable
.fromEvent(ws, "message")
.flatMap(function(quakesObj){
    quakesObj = JSON.parse(quakesObj);
    return Rx.Observable.from(quakesObj.quakes);
    })
? .scan([], function(boundsArray, quake) {
? var bounds = [
    quake.lng - 0.3, quake.lat - 0.15,
    quake.lng + 0.3, quake.lat + 0.15
].map(function(coordinate) {
    coordinate = coordinate.toString();
    return coordinate.match(/-?d+(.-?d{2})?/)[0];
});
boundsArray.concat(bounds);
?   return boundsArray.slice(Math.max(boundsArray.length - 50, 0));
})
? .subscribe(function(boundsArray) {
    stream.stop();
    stream.params.locations = boundsArray.toString();
    stream.start();
});

以下是前面代碼中發生的事情的一步一步：

我們再次見到我們的老朋友scan。 任何時候我們需要累積結果并產生每個中間結果，scan是我們的朋友。 在這種情況下，我們將繼續在boundsArray數組中累積地震坐標。

從地震震中的單緯度/經度坐標對，我們創建一個陣列，其中包含由西北坐標和東南坐標確定的區域。 用于近似邊界的數字創建了一個大城市大小的矩形。之后，我們使用正則表達式將每個坐標的小數精度限制為兩位小數，以符合Twitter API要求。

我們將生成的邊界連接到boundsArray，它包含以前每個地震的邊界。 然后我們采用最后25對邊界（數組中的50個項目），因為這是Twitter API的限制。

最后，我們訂閱了Observable，在onNext函數中，我們重新啟動當前的twit流來重新加載更新的位置，以便通過我們新的累積位置數組進行過濾，轉換為字符串。

重新啟動服務器并重新加載瀏覽器后，我們應該在瀏覽器應用程序中收到相關的推文。 但是現在，我們只能看到開發人員控制臺中顯示的原始對象。 在下一節中，我們將生成HTML以在儀表板中顯示推文。

既然我們正在接收來自服務器的推文，那么剩下要做的就是在屏幕上很好地展示它們。 為此，我們將創建一個新的HTML元素，我們附加傳入的推文：

examples_earthquake_ui/index_final.html

我們還將更新socket Observable訂閱以處理傳入的tweet對象并將它們附加到我們剛剛創建的tweet_container元素：

examples_earthquake_ui/code5.js

socket
.map(function(message) { return JSON.parse(message.data); })
.subscribe(function(data) {
    var container = document.getElementById("tweet_container");
    container.insertBefore(makeTweetElement(data), container.firstChild);
});

任何新的推文都會出現在列表的頂部，它們將由makeTweetElement創建，這是一個創建推文元素的簡單函數，并使用我們作為參數傳遞的數據填充它：

examples_earthquake_ui/code5.js

function makeTweetElement(tweetObj) {
    var tweetEl = document.createElement("div");
    tweetEl.className = "tweet";
    var content = "" +
    "
$text
" +
    "
$time
";
    
    var time = new Date(tweetObj.created_at);
    var timeText = time.toLocaleDateString() + " " + time.toLocaleTimeString();
    content = content.replace("$tweetImg", tweetObj.user.profile_image_url);
    content = content.replace("$text", tweetObj.text);
    content = content.replace("$time", timeText);
    tweetEl.innerHTML = content;
    return tweetEl;
}

有了這個，我們終于有了一個帶有相關的地理定位推文的側邊欄，可以讓我們更深入地了解受地震影響的區域。

此儀表板已經正常運行，但可以進行許多改進。 一些想法，使它更好：

添加更多地震數據庫。 USGS是一個很棒的資源，但它主要提供在美國發生的地震。 合并來自世界各地的地震報告，而不僅僅是美國，并在地圖中將它們全部展示在一起將會很有趣。 為此，您可以使用merge和mergeAll的幫助，并使用distinct與選擇器函數來避免重復。

每當用戶點擊推文時，將地圖置于相關地震中心。 這將涉及通過地震在服務器上對推文進行分組，并且您可能希望使用groupBy運算符將推文分組到特定地理區域。

在本章中，我們使用RxJS創建了一個響應式用戶界面，使我們能夠實時查看地球上發生的地震的各種數據。我們在瀏覽器客戶端和Node.js服務器中都使用了RxJS，顯示了使用Observable管理應用程序的不同區域是多么容易。

更重要的是，我們已經看到我們可以在客戶端和服務器上以相同的方式使用RxJS，在我們的應用程序中隨處可見Observable序列抽象。 不僅如此。我們實際上可以在其他編程語言中使用RxJS概念和運算符，因為許多編程語言都支持RxJS。

接下來我們將介紹Scheduler，它是RxJS中更高級的對象類型，它允許我們更精確地控制時間和并發性，并為測試代碼提供了很大的幫助。

關注我的微信公眾號，更多優質文章定時推送