@casual-simulation/causal-tree-server-socketio/CausalTreeServerSocketIO.js

var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
    return new (P || (P = Promise))(function (resolve, reject) {
        function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
        function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
        function step(result) { result.done ? resolve(result.value) : new P(function (resolve) { resolve(result.value); }).then(fulfilled, rejected); }
        step((generator = generator.apply(thisArg, _arguments || [])).next());
    });
};
import { device as deviceEvent, DEVICE_ID_CLAIM, SESSION_ID_CLAIM, USERNAME_CLAIM, } from '@casual-simulation/causal-trees';
import { Observable, Subscription, empty, } from 'rxjs';
import { loadChannel, connectDeviceChannel, } from '@casual-simulation/causal-tree-server';
import { switchMap, mergeMap, tap, concatMap, } from 'rxjs/operators';
import { socketEvent } from './Utils';
import { devicesForEvent, } from '@casual-simulation/causal-tree-server/DeviceManagerHelpers';
/**
 * Defines a class that is able to serve a set causal trees over Socket.io.
 *
 */
export class CausalTreeServerSocketIO {
    /**
     * Creates a new causal tree factory that uses the given socket server, and channel manager.
     * @param socketServer The Socket.IO server that should be used.
     * @param channelManager The channel manager that should be used.
     */
    constructor(serverDevice, socketServer, deviceManager, channelManager, authenticator, authorizer) {
        this._serverDevice = serverDevice;
        this._server = socketServer;
        this._subs = [];
        this._deviceManager = deviceManager;
        this._authenticator = authenticator;
        this._authorizer = authorizer;
        this._channelManager = channelManager;
        this._channelSiteMap = new Map();
        this._sessionIdMap = new Map();
        this._deviceIdMap = new Map();
        this._usernameMap = new Map();
        this._init();
    }
    _listenForAtoms(info, socket, channel, siteMap) {
        const eventName = `event_${info.id}`;
        const listener = (refs) => __awaiter(this, void 0, void 0, function* () {
            if (refs.length > 0) {
                let site = refs[0].id.site;
                siteMap.set(site, socket);
            }
            yield this._channelManager.addAtoms(channel, refs);
        });
        socket.on(eventName, listener);
        return new Subscription(() => {
            socket.off(eventName, listener);
        });
    }
    _listenForInfoEvents(info, socket, channel) {
        const listener = (event, callback) => __awaiter(this, void 0, void 0, function* () {
            const currentInfo = yield this._channelManager.updateVersionInfo(channel, event);
            callback(null, currentInfo);
        });
        const eventName = `info_${info.id}`;
        socket.on(eventName, listener);
        return new Subscription(() => {
            socket.off(eventName, listener);
        });
    }
    _listenForSiteIdEvents(info, socket, channel, siteMap) {
        const eventName = `siteId_${info.id}`;
        let grantedSiteId = null;
        let listener = (site, callback) => __awaiter(this, void 0, void 0, function* () {
            const allowed = yield this._channelManager.requestSiteId(channel, site);
            if (allowed) {
                grantedSiteId = site.id;
                siteMap.set(site.id, socket);
                socket.to(info.id).emit(eventName, site);
            }
            callback(null, allowed);
        });
        socket.on(eventName, listener);
        return new Subscription(() => {
            socket.off(eventName, listener);
            if (grantedSiteId) {
                siteMap.delete(grantedSiteId);
            }
        });
    }
    _listenForWeaveEvents(info, socket, channel) {
        const eventName = `weave_${info.id}`;
        const listener = (event, callback) => __awaiter(this, void 0, void 0, function* () {
            const exported = yield this._channelManager.exchangeWeaves(channel, event);
            callback(null, exported);
        });
        socket.on(eventName, listener);
        return new Subscription(() => {
            socket.off(eventName, listener);
        });
    }
    _listenForLeaveEvents(device, info, socket, loaded) {
        const eventName = `leave_${info.id}`;
        const listener = () => {
            socket.leave(info.id);
            this._deviceManager.leaveChannel(device, info);
        };
        socket.on(eventName, listener);
        return new Subscription(() => {
            socket.off(eventName, listener);
        });
    }
    _listenForRemoteEvents(device, info, socket, channel) {
        const eventName = `remote_event_${info.id}`;
        const listener = (events) => __awaiter(this, void 0, void 0, function* () {
            yield this._routeRemoteEvents(info, events, device, channel);
        });
        socket.on(eventName, listener);
        return new Subscription(() => {
            socket.off(eventName, listener);
        });
    }
    _getSocketsForDevice(deviceId) {
        let sockets = this._deviceIdMap.get(deviceId);
        if (!sockets) {
            sockets = new Set();
            this._deviceIdMap.set(deviceId, sockets);
        }
        return sockets;
    }
    _getSocketsForUsername(username) {
        let sockets = this._usernameMap.get(username);
        if (!sockets) {
            sockets = new Set();
            this._usernameMap.set(username, sockets);
        }
        return sockets;
    }
    _listenForServerEvents(device, info, socket, channel) {
        const handle = {
            socket,
            info,
        };
        const sessionId = device.claims[SESSION_ID_CLAIM];
        const deviceId = device.claims[DEVICE_ID_CLAIM];
        const username = device.claims[USERNAME_CLAIM];
        this._sessionIdMap.set(sessionId, handle);
        const deviceSockets = this._getSocketsForDevice(deviceId);
        const usernameSockets = this._getSocketsForUsername(username);
        deviceSockets.add(handle);
        usernameSockets.add(handle);
        return new Subscription(() => {
            this._sessionIdMap.delete(device.claims[SESSION_ID_CLAIM]);
            deviceSockets.delete(handle);
            usernameSockets.delete(handle);
        });
    }
    _routeRemoteEvents(info, events, device, channel) {
        return __awaiter(this, void 0, void 0, function* () {
            const connectedDevices = this._deviceManager.getConnectedDevices(info);
            const devices = connectedDevices.map(d => [d, d.extra.info]);
            let server = [];
            let deviceEvents = new Map();
            for (let event of events) {
                let dEvent = deviceEvent(device, event.event);
                if (event.sessionId || event.deviceId || event.username) {
                    let result = devicesForEvent(event, devices);
                    for (let device of result) {
                        let list = deviceEvents.get(device);
                        if (!list) {
                            list = [];
                            deviceEvents.set(device, list);
                        }
                        list.push(dEvent);
                    }
                }
                else {
                    server.push(dEvent);
                }
            }
            if (server.length > 0) {
                yield this._channelManager.sendEvents(channel, server);
            }
            deviceEvents.forEach((events, device) => {
                let socket = device.extra.socket;
                socket.emit(`remote_event_${info.id}`, events);
            });
        });
    }
    _setupListeners(socket, device, extra, channel, loaded) {
        let siteMap = this._channelSiteMap.get(channel.info.id);
        if (!siteMap) {
            siteMap = new Map();
            this._channelSiteMap.set(channel.info.id, siteMap);
        }
        return [
            this._listenForAtoms(channel.info, socket, loaded, siteMap),
            this._listenForInfoEvents(channel.info, socket, loaded),
            this._listenForSiteIdEvents(channel.info, socket, loaded, siteMap),
            this._listenForWeaveEvents(channel.info, socket, loaded),
            this._listenForLeaveEvents(device, channel.info, socket, loaded),
            this._listenForRemoteEvents(extra.info, channel.info, socket, loaded),
            this._listenForServerEvents(extra.info, channel.info, socket, loaded),
        ];
    }
    _init() {
        this._subs.push(this._channelManager.whileCausalTreeLoaded((tree, info, events) => {
            let siteMap = this._channelSiteMap.get(info.id);
            if (!siteMap) {
                siteMap = new Map();
                this._channelSiteMap.set(info.id, siteMap);
            }
            return [
                events.subscribe(events => {
                    let socketEvents = new Map();
                    for (let event of events) {
                        if (event.sessionId) {
                            const session = this._sessionIdMap.get(event.sessionId);
                            if (session) {
                                addEvent(session, event);
                            }
                        }
                        if (event.deviceId) {
                            const sockets = this._deviceIdMap.get(event.deviceId);
                            if (sockets) {
                                for (let socket of sockets) {
                                    addEvent(socket, event);
                                }
                            }
                        }
                        if (event.username) {
                            const sockets = this._usernameMap.get(event.username);
                            if (sockets) {
                                for (let socket of sockets) {
                                    addEvent(socket, event);
                                }
                            }
                        }
                    }
                    for (let [handle, events] of socketEvents) {
                        const mapped = [...events].map(e => deviceEvent(this._serverDevice, e.event));
                        handle.socket.emit(`remote_event_${handle.info.id}`, mapped);
                    }
                    function addEvent(socket, event) {
                        let events = socketEvents.get(socket);
                        if (!events) {
                            events = new Set([event]);
                            socketEvents.set(socket, events);
                        }
                        else {
                            events.add(event);
                        }
                    }
                }),
                tree.atomAdded.subscribe(atoms => {
                    if (atoms.length > 0) {
                        let site = atoms[0].id.site;
                        let siteSocket = siteMap.get(site);
                        if (siteSocket) {
                            siteSocket
                                .to(info.id)
                                .emit(`event_${info.id}`, atoms);
                        }
                        else {
                            this._server
                                .to(info.id)
                                .emit(`event_${info.id}`, atoms);
                        }
                    }
                }),
            ];
        }), this._deviceManager.whenConnectedToChannel((device, channel) => __awaiter(this, void 0, void 0, function* () {
            let subs = [];
            const extra = device.extra;
            const socket = extra.socket;
            const loaded = yield this._channelManager.loadChannel(channel.info);
            subs.push(yield this._channelManager.connect(loaded, extra.info), loaded.subscription, ...this._setupListeners(socket, device, extra, channel, loaded));
            return subs;
        })));
        this._server.on('connection', (socket) => __awaiter(this, void 0, void 0, function* () {
            const loginEvents = socketEvent(socket, 'login', (token) => ({ token }));
            const loginFlow = loginEvents.pipe(tap(({ token }) => {
                console.log(`[CasualTreeServerSocketIO] Logging ${token.username} in...`);
            }), switchMap(({ token }) => this._authenticator.authenticate(token), (data, result) => (Object.assign({}, data, { result }))), tap(({ token, result }) => {
                if (!result.success) {
                    console.log(`[CasualTreeServerSocketIO] ${token.username} not authenticated.`);
                    socket.emit('login_result', {
                        error: result.error,
                        message: 'Unable to authenticate',
                    }, null);
                }
                else {
                    console.log(`[CasualTreeServerSocketIO] ${token.username} logged in!`);
                }
            }), switchMap(({ result }) => !result.success
                ? empty()
                : connectDevice(this._deviceManager, socket.id, {
                    info: result.info,
                    socket: socket,
                }).pipe(tap(device => {
                    socket.emit('login_result', null, result.info);
                }), mergeMap(device => socketEvent(socket, 'join_channel', (info) => ({ info })), (device, info) => (Object.assign({ device }, info))))), mergeMap(({ info, device }) => this._authorizer.isAllowedToLoad(device.extra.info, info), (data, canLoad) => (Object.assign({}, data, { canLoad }))), tap(({ info, canLoad }) => {
                if (!canLoad) {
                    console.log('[CausalTreeServerSocketIO] Not allowed to load channel: ' +
                        info.id);
                    socket.emit(`join_channel_result_${info.id}`, 'channel_doesnt_exist');
                }
                else {
                    console.log('[CausalTreeServerSocketIO] Allowed to load channel: ' +
                        info.id);
                }
            }), mergeMap(({ canLoad, info }) => !canLoad
                ? empty()
                : loadChannel(this._channelManager, info), (data, loaded) => (Object.assign({}, data, { loaded }))), switchMap(({ device, loaded }) => this._authorizer.isAllowedAccess(device.extra.info, loaded), (data, authorized) => (Object.assign({}, data, { authorized }))), tap(({ info, authorized, loaded }) => {
                if (!authorized) {
                    console.log('[CausalTreeServerSocketIO] Not authorized:' +
                        info.id);
                    loaded.subscription.unsubscribe();
                    socket.emit(`join_channel_result_${info.id}`, 'unauthorized');
                }
                else {
                    console.log('[CausalTreeServerSocketIO] Authorized:' + info.id);
                }
            }), switchMap(({ authorized, info, device, loaded }) => !authorized
                ? empty()
                : join(socket, info.id).pipe(concatMap(() => connectDeviceChannel(this._deviceManager, device, info)), tap(() => {
                    console.log('[CausalTreeServerSocketIO] Finish join channel:' +
                        info.id);
                    loaded.subscription.unsubscribe();
                    socket.emit(`join_channel_result_${info.id}`, null);
                }))));
            const sub = loginFlow.subscribe(null, err => console.error(err));
            socket.on('disconnect', () => {
                if (sub) {
                    sub.unsubscribe();
                }
            });
        }));
    }
}
function connectDevice(manager, id, extra) {
    return Observable.create((observer) => {
        let device;
        setup();
        return new Subscription(() => {
            if (device) {
                manager.disconnectDevice(device);
            }
        });
        function setup() {
            return __awaiter(this, void 0, void 0, function* () {
                try {
                    device = yield manager.connectDevice(id, extra);
                    observer.next(device);
                }
                catch (err) {
                    observer.error(err);
                }
            });
        }
    });
}
function join(socket, id) {
    return Observable.create((observer) => {
        socket.join(id, err => {
            if (err) {
                observer.error(err);
                return;
            }
            observer.next();
        });
        return () => {
            socket.leave(id);
        };
    });
}
//# sourceMappingURL=CausalTreeServerSocketIO.js.map