天天看點
傳回

RxJS入門（7）----建立一個完整的web application

在本章中，使用Rxjs，我們将建立一個典型的web application。我們将轉化Dom Object Model（DOM）并且使用node.js中的websocket做c/s的互動。

  • 使用者界面部分，我們将使用RxJs-Domlibrary，這同樣是Rxjs團隊做的庫，提供了友善的操作符來處理DOM和浏覽器相關的使我們的編碼更容易。伺服器端：我們将是使用兩個建立很好的node庫，并用Observable封裝他們的一些API到我們的application中。
  • 在這一章之後，你将能使用RxJs建立使用者界面在一個公開的方式，使用這種技術到我們目前看到為止，并應用到DOM上。你将任何的nodejs項目中使用RxJs并使用響應式程式設計和Rxjs在任何項目中。

Building a Real-Time Earthquake Dashboard

  • 我們将建立一個伺服器和用戶端部分為地震儀表程式，接着我們遺留在第二章“真實地震的可視化“，我們将建立伺服器端在Node.js,并提高我們的程式更具資訊和互動性。
  • screenshot展示給我們當完成時，儀表的樣子。
  • 我們的代碼接着“真實地震的可視化”開始，如下是我們遺留的： 
var quakes = Rx.Observable
.interval()
.flatMap(function() {
return Rx.DOM.jsonpRequest({
url: QUAKE_URL,
jsonpCallback: 'eqfeed_callback'
}).retry();
})
.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 * ;
L.circle([coords[], coords[]], size).addTo(map);
});
  • 這些代碼已經有一個潛在的bug：它可以在DOM準備好之前執行。當我們試着操作DOM元素的時候就會抛出異常。我們想要執行的是DOMContentLoaded事件被觸發後加載我們的代碼，這标志着浏覽器對這個頁面的所有元素都是知道的。
  • Rxjs—Dom 提供了Rx.DOM.ready() Observable，它發射一次，當DOMContentLoaded觸發的時候。是以我們用一個initialize函數封裝下我們的代碼，并執行它當我們訂閱Rx.DOM.ready()； 
function initialize() {
var quakes = Rx.Observable
.interval()
.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 * ;
L.circle([coords[], coords[]], size).addTo(map);
});
}
Rx.DOM.ready().subscribe(initialize);
  • 接着，我們将增加一個空的table到我們的HTML中，這裡我們将顯示下一個選擇的地震資訊： 
<table>
<thead>
<tr>
<th>Location</th>
<th>Magnitude</th>
<th>Time</th>
</tr>
</thead>
<tbody id="quakes_info">
</tbody>
</table>
RxJS入門（7）----建立一個完整的web application
  • 這樣我們就可以開始我們儀表的新代碼了。

Adding a List of Earthquakes

  • 新表盤的第一個特點是展示真實地震清單，包含他們位置、量級和時間這些資訊。這個清單資料和地圖是一樣的，也來自USGS網站。我們首先建立一個函數，它傳回一個給定props對象參數的row元素： 
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屬性是一樣的。
  • 為了生成rows，我們将會建立另外一個訂閱到quakes Observable上。這個訂閱為沒一個新收到的地震在table中建立了個一個row（行）。
  • 我們在initialize函數的最後添加如下代碼： 
var table = document.getElementById('quakes_info');
quakes
.pluck('properties')
.map(makeRow)
.subscribe(function(row) { table.appendChild(row); });
  • pluck操作符提取了每個地震對象的properties值，由于它包含我們需要給makeRow的所有資訊。之後，我們map每個地震對象去makeRow，把它落戶到HTML的tr元素上。最後，在這個訂閱中，我們把每一個發射的row追加到我們的table上。

  • 無論何時我們收到這個地震資料，這都會給我們一個漂亮的表格。

    看起來很好，這也很簡單。我們任然可以做些改善，我們需要研究RxJS中一個比較重要的概念：熱和冷Observable。

  • Hot and Cold Observables
  • “hot”Observable 發射值無論Observable有沒有被訂閱。換句話說，“cold”Observable 隻有每個Observable啟動了才會發射序列全部的值。
  • Hot and Cold Observables
  • 一個observer訂閱到一個hot Observable上，當訂閱那刻開始會受到Observable發射的所有值。其他的Observer也會那時刻受到同樣的值。這和JavaScript的事件工作機制很相似。
  • 滑鼠事件和證券交易所的股票就是hot Observable的例子。在這些狀态下，無論Observable是否有訂閱，它都會發射值。在任何訂閱将要監聽之前它已經在産生值了。下面就是例子： 
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的事件工作很相似。
  • 現在讓我們看看cold Observable如何工作。
  • Cold Observables
  • 一個冷Observable僅當Observer訂閱的時候才發射值。
  • 例如，Rx.Observable.range傳回一個cold Observable，每一個新訂閱的observer都将受到一個全部的range： 
function printValue(value) {
console.log(value);
}
var rangeToFive = Rx.Observable.range(, );
var obs1 = rangeToFive.subscribe(printValue); // 1, 2, 3, 4, 5
var obs2 = Rx.Observable
.delay()
.flatMap(function() {
return rangeToFive.subscribe(printValue); // 1, 2, 3, 4, 5
});
  • 明白啥時候通過hot或者是cold Observable來處理去避免那些微妙和猥瑣的bug是很有必要的。例如，Rx.Observable.interval 傳回一個在規律的intervals機關時間段自增的整數。設想，我們想使用它，并推送同樣的值到其他的Observer。我們可以這樣實作： 
var source = Rx.Observable.interval();
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

…

  • 看起來好像起作用了。但是想想，我們需啊喲第二個訂閱者在一個訂閱這三秒後加入： 
var source = Rx.Observable.interval();
var observer1 = source.subscribe(function (x) {
console.log('Observer 1: ' + x);
});
setTimeout(function() {
var observer2 = source.subscribe(function (x) {
console.log('Observer 2: ' + x);
});
}, );

》

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是一個cold Observable。如果hot和cold Observable不是很清楚，這些情況就會很出人意料之外的。
  • 如果我們有若幹observer監聽cold Observable，他們将會受到這個序列值同樣的副本。是以嚴格地将，盡管這些observer是在共用同樣的Observable，它們沒有真正共享嚴格上的序列值。如果我們需要observer共享同樣的序列，我們需要hot Observable。
  • From Cold to Hot Using publish
  • 我們使用publish把cold Observable變為hot。調用publish建立了一個新的Observable，它扮演了源Observable的代理。訂閱源Observable自己，并推送它收到的值到它的訂閱者。
  • 一個published Observable是一個真正的ConnectableObservable，它有一個connect方法，我們調用它開始接受值。這允許我們在它開始運作前訂閱它： 
// Create an Observable that yields a value every second
var source = Rx.Observable.interval();
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);
});
}, );

》

index.html:29 Observer 1: 0

index.html:29 Observer 1: 1

index.html:29 Observer 1: 2

index.html:29 Observer 1: 3

index.html:29 Observer 1: 4

index.html:29 Observer 1: 5

index.html:37 Observer 2: 5

…

  • Sharing a Cold Observable
  • 讓我們回到我們的地震例子。到目前為止我們的代碼看起來合理；我們有一個Observable的quakes和兩個訂閱。一個是在地圖上繪制地震，另外一個是在table上把它們列出來。
  • 但是我們可以使我們的代碼更佳高效。由于有兩個quakes的訂閱，實際上請求這些資料兩次。你可以在控制台列印quakes的flatMap操作符内部來檢查它們。
  • 這些發生應為quakes是cold Observable，并且它重複發射它所有的值給每個新的訂閱者，是以，一個新的訂閱意味着新的jsonp請求。這就導緻我們的程式請求同樣的網絡資源兩次。
  • 在下一個例子中，我們将使用share操作符，它将自動建立一個訂閱到那個Observable當Observer的數目重從0到1。這樣我們就不用調用connect： 
var quakes = Rx.Observable
.interval()
.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()

現在，quakes表現的像個hot Observable，我們米有必要擔心有多少observer我們連接配接了，因為他們将會接收到嚴格的同樣資料。

  • Buffering Values
  • 我們之前的代碼工作的很好，但是注意到，我們每次接受到一個關于地震的資訊就要插入一個tr節點。這不是很高效，因為每次插入我們調整DOM并引起頁面的重繪，這使浏覽器做不必要的工作來計算新的布局。這導緻顯而易見的性能下降。
  • 我們不得不維護計數器和元素緩存，并且，我們必須沒一次的重置他們。但是在RxJS中，我們僅僅隻要使用基于buffer的操作符，像bufferWithTime。
  • 使用bufferWithTime我們可以緩存進來的值并基于每個x時間段的像數組一樣地釋放它們： 
var table = document.getElementById('quakes_info');
quakes
.pluck('properties')
.map(makeRow)
❶ .bufferWithTime()
❷ .filter(function(rows) { return rows.length > ; }
.map(function(rows) {
var fragment = document.createDocumentFragment();
rows.forEach(function(row) {
❸ fragment.appendChild(row);
});
return fragment;
})
.subscribe(function(fragment) {
❹ table.appendChild(fragment);
});
  • 這是這些新的代碼在執行：
  • 1：緩存每個進來的值，并每500毫秒釋放一批。
  • 2：bufferWithTime每500毫秒執行，如果沒有進來的值，它都會産生一個空數組。我們将會過濾它。
  • 3：我們插入每個row到一個document fragment，它是一個沒有parent的document，這意味着它沒有在DOM裡，并且調整它的内容是非常快和有效率的。
  • 4：我們把fragment添加到DOM上。追加fragment的優勢是它是個單個的操作。僅僅引起一次重繪。它也添加fragment的孩子到同樣的元素（我們追加fragment本身的）。
  • 使用buffers和fragments，我們成功地保證了row的插入操作當，同時儲存了我們程式的實時性（最大半秒的延時）。現在我們将準備添加我們儀表的下一個特性：互動性。
  • Adding Interaction
  • 現在再地圖和清單上有地震，但是沒有互動在兩者間。它将會更好，例如，當我們點選清單的地震是，它給我們在地圖上圈出來，并高亮顯示當我們的滑鼠移到它的row上。讓我們來實作它。
  • 在Leaflet上，你可以在地圖上畫，并畫它們自己的層以便單獨的操作它們。讓我們建立一組叫做quakeLayer的layer，這裡我們将素有地震的圈。每個圈将是一個layer在組裡面。我們也将建立一個codeLayers對象，這裡我們存儲地震代碼和内部layer ID的相關性，是以，我們可以通過地震ID指向圓圈： 
var codeLayers = {};
var quakeLayer = L.layerGroup([]).addTo(map);
  • 在quakes Observable的initialize内部訂閱裡，我們添加每一個圈到layer group并在codeLayers裡儲存它的ID。如果這看起來難了解，這是因為這是Leaflet允許我們地圖上有針對性畫地唯一途徑。 
quakes.subscribe(function(quake) {
var coords = quake.geometry.coordinates;
var size = quake.properties.mag * ;
var circle = L.circle([coords[], coords[]], size).addTo(map);
➤ quakeLayer.addLayer(circle);
➤ codeLayers[quake.id] = quakeLayer.getLayerId(circle);
});
  • 現在我們建立了懸浮的效果。我們将寫一個新函數，isHovering，它傳回一個發射任意給定時刻滑鼠是否在特定的地震上的Boolean值的Observable： 
❶ 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);
}
  • 1：Rx.helpers.identity是一個identity函數。給一個x參數，它傳回一個x。這裡我們沒有必要寫一個函數傳回它們接受到的值。
  • 2：over是一個Observable發射的true，當使用者的滑鼠懸浮在元素上。
  • 3：out是一個Observable發射false，當使用者的滑鼠離開元素。
  • 4：isHovering混合了over和out，傳回一個Observable（發射true當滑鼠懸浮一個元素上，發射false當滑鼠離開）。
  • 使用isHovering我們可以修改建立rows的訂閱，是以我們訂閱到每個row建立的事件。 
var table = document.getElementById('quakes_info');
quakes
.pluck('properties')
.map(makeRow)
.bufferWithTime()
.filter(function(rows) { return rows.length > ; })
.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);
})
  • 1：使用從row element擷取的ID我們在地圖上定位地震的圓圈。codeLayers給我們内部相關性的ID，使用quakeLayer.getLayer我們擷取圓圈元素。
  • 2：我們用現在的row調用isHovering同時我們訂閱到結果的Observable上。若幹懸浮的參數是true，我們将會花紅色圓圈，否則将會是藍色的。
  • 3：我們訂閱到目前row click事件建立的Observable上。當清單的row被點選，地圖上對應位置将會有圓圈。
  • Making It Efficient
  • 在一個頁面上建立好多事件是一個糟糕的性能的處方對前端開發者來說。在我們之前的例子，對每個row我們建立了三個事件。如果清單中有一百個地震，那麼我們就有300個事件在這個頁面上流動僅僅是為了高亮的動作。這是個相當糟糕的性能，我們可以做的更好。
  • 由于DOM裡面的事件是冒泡的（從位元組點到父節點），一個衆所周知的技術對前端開發者來說避免滑鼠事件涉及到太多單個元素的方法是，把滑鼠事件綁定到這些元素的父元素上。一旦父元素的事件被捕獲，我們可以使用事件的target屬性來尋找事件目标的元素。
  • 由于我們需要事件click和mouseover相似的功能，是以我們将建立一個getRowFromEvent函數： 
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給我們事件發生的table的row，如下是細節：
  • 1：我們确認我們擷取表中沒個機關正在發生的事件，并且，我們檢查這些機關的parent是否是一個ID屬相的row。這些row是我們添加地震ID的那些。
  • 2：pluck操作符精确提取嵌套parentNode裡面的元素的target屬性。
  • 3：這阻止多次擷取同樣的元素。例如當mouseover事件多次發生時。
  • 在上面的選擇中，我們綁定每個mouseover和mouseout事件去改變地震圈的顔色，每次當滑鼠進入或者是退出那個row時。現在，我們在table上僅僅使用mouseover事件，使用pairwise操作符來組合。 
getRowFromEvent('mouseover')
.pairwise()
.subscribe(function(rows) {
var prevCircle = quakeLayer.getLayer(codeLayers[rows[].id]);
var currCircle = quakeLayer.getLayer(codeLayers[rows[].id]);
prevCircle.setStyle({ color: '#0000ff' });
currCircle.setStyle({ color: '#ff0000' });
});
  • pairwise每個發射的值和之前發射的值組到一個數組中。由于我們一直在擷取去不同的row，pairwise将會一直産生這樣的row，滑鼠剛離開和滑鼠正在懸停的。于是使用這些資訊，很容易地給每個地震圈上顔色。
  • 處理click事件也是相當簡單的： 
getRowFromEvent('click')
.subscribe(function(row) {
var circle = quakeLayer.getLayer(codeLayers[row.id]);
map.panTo(circle.getLatLng());
});
  • 我們傳回到剛訂閱到quakes來産生row： 
quakes
.pluck('properties')
.map(makeRow)
.subscribe(function(row) { table.appendChild(row); });
  • 我們的代碼顯示更清新和符合語言規範了，它不依賴與以後的rows，如果沒有rows，getRowFromEvent不會産生任何結果。
  • 更重要的是，我們的代碼很高效，我們我們擷取的地震的規模如何，我們一直都是一個的那個的mouseover滑鼠事件和單個的click事件，而不是上百個。
RxJs
上一篇: rxjs基礎知識
下一篇: how to run JavaWordCount in SparkCreated by Jerry Wang, last modified on Aug 17, 2015

繼續閱讀