From b3f42e62af698a67c2250533c437569559f1fdf9 Mon Sep 17 00:00:00 2001 From: syuilo Date: Thu, 29 Dec 2016 07:49:51 +0900 Subject: Initial commit :four_leaf_clover: --- src/api/streaming.ts | 69 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 69 insertions(+) create mode 100644 src/api/streaming.ts (limited to 'src/api/streaming.ts') diff --git a/src/api/streaming.ts b/src/api/streaming.ts new file mode 100644 index 0000000000..38068d1e3d --- /dev/null +++ b/src/api/streaming.ts @@ -0,0 +1,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 { + 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); + }); + }); +} -- cgit v1.2.3-freya