3719

pending completion

Build # 3719

Build Type

push

travis-ci-com

Committed by

web-flow

Commit Message

Merge pull request #1116 from stanford-oval/wip/update-deps

Small fixes

Run Details

2007 of 3773 branches covered (53.19%)

Branch coverage included in aggregate %.

6 of 7 new or added lines in 1 file covered. (85.71%)

6 existing lines in 3 files now uncovered.

7707 of 11503 relevant lines covered (67.0%)

505.62 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

65.32

/src/routes/my_conversation.ts

// -*- mode: js; indent-tabs-mode: nil; js-basic-offset: 4 -*-
//
// This file is part of Almond
//
// Copyright 2019-2020 The Board of Trustees of the Leland Stanford Junior University
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//    http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Author: Giovanni Campagna <gcampagn@cs.stanford.edu>

import type WebSocket from 'ws';
import type express from 'express';
import * as rpc from 'transparent-rpc';

import type { WebSocketConnnectionWrapper, NotificationWrapper } from '../almond/engine';
import * as userModel from '../model/user';
import * as user from '../util/user';
import * as EngineManager from '../almond/enginemanagerclient';
import { makeRandom } from '../util/random';

import * as Config from '../config';

export function anonymous(ws : WebSocket, req : express.Request) {
    if (req.user) {
        ws.close();
        return;
    }

    user.getAnonymousUser(req.locale).then((user) => {
        return doConversation(user, true, ws, req.query);
    });
}

class WebsocketDelegate implements rpc.Stubbable {
    $rpcMethods = ['send'] as const;
    private _ws : WebSocket;

    constructor(ws : WebSocket) {
        this._ws = ws;
    }

    async send(data : any) {
        if (data.result && data.result.icon)
            data.result.icon = Config.CDN_HOST + '/icons/' + data.result.icon + '.png';
        else if (data.error && data.error.icon)
            data.error.icon = Config.CDN_HOST + '/icons/' + data.error.icon + '.png';

        try {
            this._ws.send(JSON.stringify(data));
        } catch(e) {
            // ignore if the socket is closed
            if (e.message !== 'not opened')
                throw e;
        }
    }
}

export function results(ws : WebSocket, req : express.Request, next : express.NextFunction) {
    const user = req.user!;

    Promise.resolve().then(async () => {
        const engine = await EngineManager.get().getEngine(user.id);
        const onclosed = (userId : number) => {
            if (userId === user.id)
                ws.close();
            EngineManager.get().removeListener('socket-closed', onclosed);
        };
        EngineManager.get().on('socket-closed', onclosed);

        const delegate = new WebsocketDelegate(ws);
        let wrapper : rpc.Proxy<NotificationWrapper>|undefined = undefined;
        ws.on('error', (err) => {
            ws.close();
        });
        ws.on('close', async () => {
            if (!wrapper)
                return;
            try {
                await wrapper.destroy();
            } catch(e) {
                // ignore errors if engine died
            }
            wrapper.$free();
            wrapper = undefined;
        });
        ws.on('ping', (data) => ws.pong(data));

        wrapper = await engine.addNotificationOutput(delegate);
    }).catch((error) => {
        console.error('Error in API websocket: ' + error.message);

        // ignore "Not Opened" error in closing
        try {
            ws.close();
        } catch(e) { /**/ }
    });
}

interface ConversationQueryParams {
    hide_welcome ?: '1'|''|undefined;
    id ?: string;
    flags ?: Record<string, unknown>;
    skip_history ?: '1'|''|undefined;
    sync_devices ?: '1'|''|undefined;
}

async function doConversation(user : userModel.RowWithOrg, anonymous : boolean, ws : WebSocket, query : ConversationQueryParams) {
    try {
        let wrapper : rpc.Proxy<WebSocketConnnectionWrapper>|undefined;
        let opened = false;

        ws.on('error', (err) => {
            ws.close();
        });
        ws.on('close', async () => {
            if (!wrapper)
                return;
            try {
                await wrapper.destroy();
            } catch(e) {
                // ignore errors if engine died
            }
            wrapper.$free();
            wrapper = undefined;
        });
        let initQueue : any[] = [];
        ws.on('message', (data) => {
            Promise.resolve().then(async () => {
                const parsed = JSON.parse(String(data));
                if (opened) {
                    if (!wrapper) // race condition, connection closed
                        return;
                    await wrapper.handle(parsed);
                } else {
                    initQueue.push(parsed);
                }
            }).catch((e) => {
                // either the message didn't parse as json, or we had an error sending the error
                // (ie the websocket is closed)
                // eat the error and close the socket
                ws.terminate();
            });
        });

        const engine = await EngineManager.get().getEngine(user.id);
        const onclosed = (userId : number) => {
            if (userId === user.id)
                ws.close();
            EngineManager.get().removeListener('socket-closed', onclosed);
        };
        EngineManager.get().on('socket-closed', onclosed);

        const flags : Record<string, boolean> = {};
        if (query.flags) {
            for (const key in query.flags) {
                if (query.flags[key])
                    flags[key] = true;
            }
        }

        const options = {
            replayHistory: !query.skip_history,
            syncDevices: !!query.sync_devices,

            showWelcome: !query.hide_welcome,
            anonymous,
            dialogueFlags: flags,

            // in anonymous mode, set a very large timeout so we don't get recycled until the socket is closed
            // in user mode, we always share the same conversation so we set no inactivity timeout at all
            inactivityTimeout: anonymous ? (3600 * 1000) : -1
        };

        const delegate = new WebsocketDelegate(ws);

        const id = query.id || (anonymous ? 'web-' + makeRandom(4) : 'main');
        wrapper = await engine.getOrOpenConversationWebSocket(id, delegate, options);
        for (const msg of initQueue)
            await wrapper.handle(msg);
        opened = true;
        initQueue = [];
    } catch(error) {
        console.error('Error in conversation websocket: ' + error.message);

        // ignore "Not Opened" error in closing
        try {
            ws.close();
        } catch(e) { /**/ }
    }
}

export function conversation(ws : WebSocket, req : express.Request<any, any, any, ConversationQueryParams>, next : express.NextFunction) {
    doConversation(req.user!, false, ws, req.query);
}

1	// -- mode: js; indent-tabs-mode: nil; js-basic-offset: 4 --
2	//
3	// This file is part of Almond
4	//
5	// Copyright 2019-2020 The Board of Trustees of the Leland Stanford Junior University
6	//
7	// Licensed under the Apache License, Version 2.0 (the "License");
8	// you may not use this file except in compliance with the License.
9	// You may obtain a copy of the License at
10	//
11	// http://www.apache.org/licenses/LICENSE-2.0
12	//
13	// Unless required by applicable law or agreed to in writing, software
14	// distributed under the License is distributed on an "AS IS" BASIS,
15	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16	// See the License for the specific language governing permissions and
17	// limitations under the License.
18	//
19	// Author: Giovanni Campagna <gcampagn@cs.stanford.edu>
20
21	import type WebSocket from 'ws';
22	import type express from 'express';
23	import * as rpc from 'transparent-rpc';
24
25	import type { WebSocketConnnectionWrapper, NotificationWrapper } from '../almond/engine';
26	import * as userModel from '../model/user';
27	import * as user from '../util/user';	2✔
28	import * as EngineManager from '../almond/enginemanagerclient';	2✔
29	import { makeRandom } from '../util/random';	2✔
30
31	import * as Config from '../config';	2✔
32
33	export function anonymous(ws : WebSocket, req : express.Request) {	2✔
34	if (req.user) {	×
35	ws.close();	×
36	return;	×
37	}
38
39	user.getAnonymousUser(req.locale).then((user) => {	×
40	return doConversation(user, true, ws, req.query);	×
41	});
42	}
43
44	class WebsocketDelegate implements rpc.Stubbable {
45	$rpcMethods = ['send'] as const;	13✔
46	private _ws : WebSocket;
47
48	constructor(ws : WebSocket) {
49	this._ws = ws;	13✔
50	}
51
52	async send(data : any) {
53	if (data.result && data.result.icon)	52✔
54	data.result.icon = Config.CDN_HOST + '/icons/' + data.result.icon + '.png';	4✔
55	else if (data.error && data.error.icon)	48!
56	data.error.icon = Config.CDN_HOST + '/icons/' + data.error.icon + '.png';	×
57
58	try {	52✔
59	this._ws.send(JSON.stringify(data));	52✔
60	} catch(e) {
61	// ignore if the socket is closed
62	if (e.message !== 'not opened')	×
63	throw e;	×
64	}
65	}
66	}
67
68	export function results(ws : WebSocket, req : express.Request, next : express.NextFunction) {	2✔
69	const user = req.user!;	2✔
70
71	Promise.resolve().then(async () => {	2✔
72	const engine = await EngineManager.get().getEngine(user.id);	2✔
73	const onclosed = (userId : number) => {	2✔
74	if (userId === user.id)	2✔
75	ws.close();	1✔
76	EngineManager.get().removeListener('socket-closed', onclosed);	2✔
77	};
78	EngineManager.get().on('socket-closed', onclosed);	2✔
79
80	const delegate = new WebsocketDelegate(ws);	2✔
81	let wrapper : rpc.Proxy<NotificationWrapper>\|undefined = undefined;	2✔
82	ws.on('error', (err) => {	2✔
83	ws.close();	×
84	});
85	ws.on('close', async () => {	2✔
86	if (!wrapper)	2!
87	return;	×
88	try {	2✔
89	await wrapper.destroy();	2✔
90	} catch(e) {
91	// ignore errors if engine died
92	}
93	wrapper.$free();	2✔
94	wrapper = undefined;	2✔
95	});
96	ws.on('ping', (data) => ws.pong(data));	2✔
97
98	wrapper = await engine.addNotificationOutput(delegate);	2✔
99	}).catch((error) => {
100	console.error('Error in API websocket: ' + error.message);	×
101
102	// ignore "Not Opened" error in closing
103	try {	×
104	ws.close();	×
105	} catch(e) { /**/ }
106	});
107	}
108
109	interface ConversationQueryParams {
110	hide_welcome ?: '1'\|''\|undefined;
111	id ?: string;
112	flags ?: Record<string, unknown>;
113	skip_history ?: '1'\|''\|undefined;
114	sync_devices ?: '1'\|''\|undefined;
115	}
116
117	async function doConversation(user : userModel.RowWithOrg, anonymous : boolean, ws : WebSocket, query : ConversationQueryParams) {
118	try {	11✔
119	let wrapper : rpc.Proxy<WebSocketConnnectionWrapper>\|undefined;
120	let opened = false;	11✔
121
122	ws.on('error', (err) => {	11✔
123	ws.close();	×
124	});
125	ws.on('close', async () => {	11✔
126	if (!wrapper)	10✔
127	return;	4✔
128	try {	6✔
129	await wrapper.destroy();	6✔
130	} catch(e) {
131	// ignore errors if engine died
132	}
133	wrapper.$free();	6✔
134	wrapper = undefined;	6✔
135	});
136	let initQueue : any[] = [];	11✔
137	ws.on('message', (data) => {	11✔
138	Promise.resolve().then(async () => {	2✔
139	const parsed = JSON.parse(String(data));	2✔
140	if (opened) {	2!
141	if (!wrapper) // race condition, connection closed	2!
142	return;	×
143	await wrapper.handle(parsed);	2✔
144	} else {
145	initQueue.push(parsed);	×
146	}
147	}).catch((e) => {
148	// either the message didn't parse as json, or we had an error sending the error
149	// (ie the websocket is closed)
150	// eat the error and close the socket
UNCOV 151	ws.terminate();	×
152	});
153	});
154
155	const engine = await EngineManager.get().getEngine(user.id);	11✔
156	const onclosed = (userId : number) => {	11✔
UNCOV 157	if (userId === user.id)	×
UNCOV 158	ws.close();	×
UNCOV 159	EngineManager.get().removeListener('socket-closed', onclosed);	×
160	};
161	EngineManager.get().on('socket-closed', onclosed);	11✔
162
163	const flags : Record<string, boolean> = {};	11✔
164	if (query.flags) {	11!
165	for (const key in query.flags) {	×
166	if (query.flags[key])	×
167	flags[key] = true;	×
168	}
169	}
170
171	const options = {	11✔
172	replayHistory: !query.skip_history,
173	syncDevices: !!query.sync_devices,
174
175	showWelcome: !query.hide_welcome,
176	anonymous,
177	dialogueFlags: flags,
178
179	// in anonymous mode, set a very large timeout so we don't get recycled until the socket is closed
180	// in user mode, we always share the same conversation so we set no inactivity timeout at all
181	inactivityTimeout: anonymous ? (3600 * 1000) : -1	11!
182	};
183
184	const delegate = new WebsocketDelegate(ws);	11✔
185
186	const id = query.id \|\| (anonymous ? 'web-' + makeRandom(4) : 'main');	11!
187	wrapper = await engine.getOrOpenConversationWebSocket(id, delegate, options);	11✔
188	for (const msg of initQueue)	10✔
189	await wrapper.handle(msg);	×
190	opened = true;	10✔
191	initQueue = [];	10✔
192	} catch(error) {
193	console.error('Error in conversation websocket: ' + error.message);	×
194
195	// ignore "Not Opened" error in closing
196	try {	×
197	ws.close();	×
198	} catch(e) { /**/ }
199	}
200	}
201
202	export function conversation(ws : WebSocket, req : express.Request<any, any, any, ConversationQueryParams>, next : express.NextFunction) {	2✔
203	doConversation(req.user!, false, ws, req.query);	11✔
204	}

stanford-oval / almond-cloud / 3719

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous