上接(7)
Getting Real-Time Updates from Twitter
- 我们计划的的第二部分是做一个实时的仪表给地震,添加从Twitter相关的地球上正在发生的不同地震报告和信息。为了实现这个,我们需要创建一个小的Node.js程序,它获取tweets相关的地震的流。
- Setting Up Our Node.js Environment
- 配置我们的Node.js程序。包括RxJS,我们将会使用两个比较重要的第三方modules使我们的编程会更容易:ws和twit。其他任何相似的modules应该对代码的改动最小。
- 首先,为我们的程序创建一个目录,并install那些modules(我们将要使用的)。(注意npm命令的输出,依赖当前版本和包)
- Client–Server Communication
- 现在我们已经准备开搭建我们的程序了。让我们创建一个名叫index.js的新文件,在它里面, tweet_stream这个文件夹就是用来加载我们将用使用的modules:
var WebSocketServer = require(‘ws‘).Server;
var Twit = require(‘twit‘);
var Rx = require(‘rx‘);
var T = new Twit({
consumer_key: ‘rFhfB5hFlth0BHC7iqQkEtTyw‘,
consumer_secret: ‘zcrXEM1jiOdKyiFFlGYFAOo43Hsz383i0cdHYYWqBXTBoVAr1x‘,
access_token: ‘14343133-nlxZbtLuTEwgAlaLsmfrr3D4QAoiV2fa6xXUVEwW9‘,
access_token_secret: ‘57Dr99wECljyyQ9tViJWz0H3obNG3V4cr5Lix9sQBXju1‘
});
- 现在我们可以创建一个函数,onConnect,它将会做所有的搜素tweets和与client交互的工作。我们可以调用onConnect初始化一个WebSocket server一旦这个WebSocket连接和准备好了:
function onConnect(ws) {
console.log(‘Client connected on localhost:8080‘);
}
var Server = new WebSocketServer({ port: 8080 });
Rx.Observable.fromEvent(Server, ‘connection‘).subscribe(onConnect);
- 我们可以登录我们的应用程序了,以WebSocket的8080端口开始:
- 这个关于client连接的消息不会被打印,是因为我们让任何的浏览器连接到这个服务器上。让我们调整我们的仪表代码,我们将会使用RxJS-DOM的fromWebSocket操作符:
function initialize() {
var socket = Rx.DOM.fromWebSocket(‘ws://127.0.0.1:8080‘);
...
- 上面的处理代码中,fromWebSocket创建了一个Subject,它作为一个接受和发送消息到WebSocket server的服务提供者,通过订阅socket,我们将受到服务器发给我们的任何消息。
- 现在可以发送我们接收的服务器发送的地震消息了:
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 }));
});
- 我们也创建一个来自服务器消息的订阅者:
socket.subscribe(function(message) {
console.log(JSON.parse(message.data));
});
- 现在我们重新加载浏览器,这个client消息将会在terminal上出现的:
- 古怪的!这个浏览器可以发送命令到服务器当它开始接收远程的JSONP资源的地震。到目前为止,这个server完全忽略了这些消息。回到我们的tweet流的代码并做些什么。
- 首先,我们将从browser client上连到那个message事件,它到达服务器。无论何时,client发送了一条消息,这个WebSocket服务器将会发射一个message事件连接到这个消息。这种情况下,这个内容是分层的对象。
- 在onConnect函数里我么写如下代码:
var onMessage = Rx.Observable.fromEvent(ws, ‘message‘)
.subscribe(function(quake) {
quake = JSON.parse(quake);
console.log(quake);
});
- 如果我们重启服务器(在terminal中ctrl-c)并重载浏览器,我们可以看到地震的详请,这些它们进来的将会子啊terminal上打印。这样很好,我们现在可以搜索跟tweets相关的地震了。
- Retrieving and Sending Tweets
- 我们正在使用基于Node.js twit流的Twitter客户端去连接Twitter和搜索tweets。从现在开始所有的代码将会在onConnect函数的内部起作用,是由于它假设一个到WebSocket的连接已经建立了。现在让我们初始化tweets的流:
var stream = T.stream(‘statuses/filter‘, {
track: ‘earthquake‘,
locations: []
});
- 这告诉我们Twit实例T开会Twitter的statues的流,以地震的keyword来过滤。当然,这是普通和不直接和现在正发生的地震相关的。注意到空的locations数组。它是经纬度的数组,通过earthquake这个词,利用它们的地理位置过滤tweets。
这很特别!让我们订阅到这个stream并开始发送tweets到浏览器:
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‘);
}
});
});
- 如果我们重启服务器并重新加载浏览器,我们在浏览器上收到tweets,并在在开发者的布局的控制上打印这些推特。
- 这些推特没有通过地震位置俩过滤。为了实现这项,我们需要对收到的每条地震消息作如下处理:
- 取得每个地震中心的经纬度并创建一个限制盒子,它限定了我们认为和地震相关的推特的地理位置。
- 累积所有的坐标盒子,以便推特发送到client上并在地图上描述素有相关的地震。
- 每当我们收新地震的消息,使用新做包更新twit流。
- 如下便是:
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();
});
- 这里上面代码一步步的发生的:
- 1:我们又要使用scan了。每次我们需要累加结果并立即产生新的值,scan就是我们的朋友。这种情况下,我们将会在boundsArray数组中一直保存地震的坐标。
- 2:从单个的地震中心的坐标的经纬度中,我们创建一个包含西北、东南坐标的的区域的数组。这些大致的界限创建了一个大型城市的矩形框。之后,我们使用了一个规律的表达式来限制每个坐标两位小数的精度,来遵循Twitter API的规范。
- 3:我们连接产生的边界到boundsArray,它包含之前的每个地震边界。之后,我们取最近的25对边界(这个数组里面有50项),这是TwitterAPI的限制。
- 4:自后,我们订阅到Observable,在onNext函数中,我们重启当前的twit流来重载更新位置,并通过我们新累存的位置数组来过滤,转化为字符串。
- 在重启服务器和重新加载浏览器后,我们可以在我们的浏览器程序中收到相关的推特。到目前为值,我们可仅能在开发者控制台看到原始的数据对象。下一部分,我们将通过HTML来在我们的仪表上展示tweets。
- Showing Tweets on the Dashboard
- 现在我们冲服务器收到tweets,仅仅遗留需要完成的是把它们在屏幕上显示。我们将创建一个新的HTML元素在我们追加新来的tweets的地方。
<div id="tweet_container"></div>
我们将会跟新我们的socket Observable订阅俩处理新的tweet对象并追加它们到刚创建的tweet_container元素上。
socket
.map(function(message) { return JSON.parse(message.data); })
.subscribe(function(data) {
var container = document.getElementById(‘tweet_container‘);
container.insertBefore(makeTweetElement(data), container.firstChild);
});
- 任何新的tweets将会出现在列表的最顶上,它们将会被makeTweetElement创建,一个创建tweet元素的简单函数,并把我们传递的数据地位到上面:
function makeTweetElement(tweetObj) {
var tweetEl = document.createElement(‘div‘);
tweetEl.className = ‘tweet‘;
var content = ‘<img src="$tweetImg" class="avatar" />‘ +
‘<div class="content">$text</div>‘ +
‘<div class="time">$time</div>‘;
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;
}
- 最后,我们使用一个有关的工具栏,定位能给我们更多关于地震的区域结果的信息的tweets。
Ideas for Improvements
- 这个仪表已经起作用了,但是还有许多提高可以做。如下是一些更好的建议:
- 添加更多的地震数据,USGS是一个很好的资源,但是它主要提供发生在美国的。汇聚全世界的地震将会更加有趣,而不仅仅是美国,并把它们在地图上显示出来。这样,你需要使用merge和mergeAll来帮忙了,并使用distinct这个选择函数来去重。
- 无论何时用户点击了tweet,在地图上圈出相关的地震。这将会包含对服务器地震tweet的归类,你将有可能使用groupBy操作符来归类tweets到一个特定的地理区域。
时间: 2024-10-08 20:14:36