forked from xdevplatform/gcloud-toolkit-filtered-stream
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathstream.js
More file actions
113 lines (97 loc) · 3.88 KB
/
Copy pathstream.js
File metadata and controls
113 lines (97 loc) · 3.88 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
const express = require("express");
const gcp_infra_svcs = require('.././services/gcp-infra.js');
const config = require('../config.js');
const needle = require('needle');
const router = express.Router();
router.get("/", function (req, res) {
gcp_infra_svcs.provisionDB().then(function (status) {
if (status != null && status.includes('Successfully provisioned')) {
gcp_infra_svcs.setupMsgInfra().then(function (statusMsg) {
if (statusMsg != null && statusMsg.includes(config.gcp_infra.topicName)) {
streamTweets();
}
})
res.send("Now streaming tweets with new GCP infra ..");
}
}).catch(error => {
streamTweets();
res.send("Now streaming tweets with existing GCP infra ..");
})
});
router.get("/clean", function (req, res) {
gcp_infra_svcs.cleanUp();
res.send('GCP resources deleted');
});
router.get("/connect", function (req, res) {
streamTweets();
res.send('Connecting to stream');
});
router.get("/alive", function (req, res) {
//console.log('staying alive ..');
res.send('Alive');
});
router.get("/poll/:frequency/:delay", function (req, res) {
console.log('polling Tweets from PubSub ', req.params.frequency);
for (var i = 0; i < req.params.frequency; i++) {
setTimeout(() => {
gcp_infra_svcs.synchronousPull(config.gcp_infra.projectId, config.gcp_infra.subscriptionName, config.gcp_infra.messageCount).then((messenger) => {
if (messenger === 'disconnect') {
console.log('Stream reconnecting => ', messenger);
streamTweets();
}
})
}, req.params.delay);
}
res.send('polling Tweets from PubSub');
});
async function streamTweets() {
console.log('Streaming Tweets ..')
//Listen to the stream
const options = {
timeout: 20000
}
const streamURL = config.filtered_stream.host + config.filtered_stream.path + config.filtered_stream.tweet_fields +
config.filtered_stream.user_fields + config.filtered_stream.expansions + config.filtered_stream.media_fields + config.filtered_stream.place_fields +
config.filtered_stream.poll_fields;
const stream = needle.get(streamURL, {
headers: {
Authorization: config.twitter_bearer_token
}
}, options);
stream.on('data', data => {
var splited_payload = '';
try {
const json_payload = data.toString();
console.log('Received Tweet ');
if (json_payload) {
try {
JSON.parse(json_payload);
gcp_infra_svcs.publishMessage(config.gcp_infra.topicName, JSON.stringify(json_payload));
} catch (e) {
if (json_payload[0] === undefined || json_payload[0] === '\r' || json_payload[0] === '' || json_payload[0] === '\n') {
console.log('~~~ Heartbeat payload ~~~ ');
} else {
if (splited_payload.length > 0) {
splited_payload.append(json_payload);
gcp_infra_svcs.publishMessage(config.gcp_infra.topicName, JSON.stringify(splited_payload));
console.log('splited_payload ', JSON.parse(splited_payload));
splited_payload = '';
}
else
splited_payload = json_payload;
}
}
}
} catch (err) {
console.log('Error ', err);
// Keep alive signal received. Do nothing.
}
}).on('error', error => {
if (error.code === 'ETIMEDOUT') {
stream.emit('timeout');
}
});
return stream;
}
module.exports = router
module.exports.streamTweets = streamTweets;