summaryrefslogtreecommitdiff
path: root/src/api/streaming.ts
blob: 38068d1e3d345f5d0e76acba211553dc33ddc052 (plain)
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
import * as http from 'http';
import * as websocket from 'websocket';
import * as redis from 'redis';
import User from './models/user';

import homeStream from './stream/home';
import messagingStream from './stream/messaging';

module.exports = (server: http.Server) => {
	/**
	 * Init websocket server
	 */
	const ws = new websocket.server({
		httpServer: server
	});

	ws.on('request', async (request) => {
		const connection = request.accept();

		const user = await authenticate(connection);

		// Connect to Redis
		const subscriber = redis.createClient(
			config.redis.port, config.redis.host);

		connection.on('close', () => {
			subscriber.unsubscribe();
			subscriber.quit();
		});

		const channel =
			request.resourceURL.pathname === '/' ? homeStream :
			request.resourceURL.pathname === '/messaging' ? messagingStream :
			null;

		if (channel !== null) {
			channel(request, connection, subscriber, user);
		} else {
			connection.close();
		}
	});
};

function authenticate(connection: websocket.connection): Promise<any> {
	return new Promise((resolve, reject) => {
		// Listen first message
		connection.once('message', async (data) => {
			const msg = JSON.parse(data.utf8Data);

			// Fetch user
			// SELECT _id
			const user = await User
				.findOne({
					token: msg.i
				}, {
					_id: true
				});

			if (user === null) {
				connection.close();
				return;
			}

			connection.send('authenticated');

			resolve(user);
		});
	});
}