feat: enhance WebSocket proxy with user connection management and status reporting
- Updated StudioOpts type to include infoList for user connection status. - Added rendering of connection info in createStudioAppListHtml. - Modified self-restart logic to use a specific app path. - Improved WebSocket connection handling in wsProxyManager, including user registration and ID management. - Implemented connection status checks and responses in UserV1Proxy. - Introduced renderServerHtml function to inject server data into HTML responses. - Refactored page-proxy request handling for better URL management.
This commit is contained in:
@@ -1,7 +1,9 @@
|
||||
import { nanoid } from 'nanoid';
|
||||
import { customAlphabet } from 'nanoid';
|
||||
import { WebSocket } from 'ws';
|
||||
import { logger } from '../logger.ts';
|
||||
import { EventEmitter } from 'eventemitter3';
|
||||
const letters = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ';
|
||||
const nanoid = customAlphabet(letters, 10);
|
||||
|
||||
class WsMessage {
|
||||
ws: WebSocket;
|
||||
@@ -9,11 +11,16 @@ class WsMessage {
|
||||
emitter: EventEmitter;
|
||||
private pingTimer?: NodeJS.Timeout;
|
||||
private readonly PING_INTERVAL = 30000; // 30 秒发送一次 ping
|
||||
|
||||
constructor({ ws, user }: WssMessageOptions) {
|
||||
id?: string;
|
||||
status?: 'waiting' | 'connected' | 'closed';
|
||||
manager: WsProxyManager;
|
||||
constructor({ ws, user, id, isLogin, manager }: WssMessageOptions) {
|
||||
this.ws = ws;
|
||||
this.user = user;
|
||||
this.id = id;
|
||||
this.emitter = new EventEmitter();
|
||||
this.manager = manager;
|
||||
this.status = isLogin ? 'connected' : 'waiting';
|
||||
this.startPing();
|
||||
}
|
||||
|
||||
@@ -39,12 +46,44 @@ class WsMessage {
|
||||
this.stopPing();
|
||||
this.emitter.removeAllListeners();
|
||||
}
|
||||
isClosed() {
|
||||
return this.ws.readyState === WebSocket.CLOSED;
|
||||
}
|
||||
|
||||
async sendResponse(data: any) {
|
||||
if (data.id) {
|
||||
this.emitter.emit(data.id, data?.data);
|
||||
}
|
||||
}
|
||||
async sendConnected() {
|
||||
const id = this.id;
|
||||
const user = this.user;
|
||||
const data = { type: 'verified', user, id };
|
||||
if (this.ws.readyState === WebSocket.OPEN) {
|
||||
this.ws.send(JSON.stringify(data));
|
||||
this.status = 'connected';
|
||||
}
|
||||
if (id.includes('-registry-')) {
|
||||
const newId = id.split('-registry-')[0];
|
||||
this.manager.changeId(id, newId);
|
||||
const ws = this.ws;
|
||||
// @ts-ignore
|
||||
if (this.ws?.data) {
|
||||
// @ts-ignore
|
||||
this.ws.data.userApp = newId;
|
||||
}
|
||||
}
|
||||
}
|
||||
getInfo() {
|
||||
const shortAppId = this.id ? this.id.split('--')[1] : '';
|
||||
return {
|
||||
user: this.user,
|
||||
id: this.id,
|
||||
status: this.status,
|
||||
shortAppId,
|
||||
pathname: this.id ? `/${this.user}/v1/${shortAppId}` : '',
|
||||
};
|
||||
}
|
||||
async sendData(data: any, context?: any, opts?: { timeout?: number }) {
|
||||
if (this.ws.readyState !== WebSocket.OPEN) {
|
||||
return { code: 500, message: 'WebSocket is not open' };
|
||||
@@ -64,6 +103,7 @@ class WsMessage {
|
||||
const msg = { path: data?.path, key: data?.key, id: data?.id };
|
||||
return new Promise((resolve) => {
|
||||
const timer = setTimeout(() => {
|
||||
console.log('ws-proxy sendData timeout', msg);
|
||||
resolve({
|
||||
code: 500,
|
||||
message: `运行超时，执行的id: ${id}，参数是${JSON.stringify(msg)}`,
|
||||
@@ -79,7 +119,12 @@ class WsMessage {
|
||||
type WssMessageOptions = {
|
||||
ws: WebSocket;
|
||||
user?: string;
|
||||
id?: string;
|
||||
realId?: string;
|
||||
isLogin?: boolean;
|
||||
manager: WsProxyManager;
|
||||
};
|
||||
|
||||
export class WsProxyManager {
|
||||
wssMap: Map<string, WsMessage> = new Map();
|
||||
PING_INTERVAL = 30000; // 30 秒检查一次连接状态
|
||||
@@ -89,7 +134,7 @@ export class WsProxyManager {
|
||||
}
|
||||
this.checkConnceted();
|
||||
}
|
||||
register(id: string, opts?: { ws: WebSocket; user: string }) {
|
||||
register(id: string, opts?: { ws: WebSocket; user: string, id?: string }) {
|
||||
if (this.wssMap.has(id)) {
|
||||
const value = this.wssMap.get(id);
|
||||
if (value) {
|
||||
@@ -100,8 +145,24 @@ export class WsProxyManager {
|
||||
const [username, appId] = id.split('--');
|
||||
const url = new URL(`/${username}/v1/${appId}`, 'https://kevisual.cn/');
|
||||
console.log('WsProxyManager register', id, '访问地址', url.toString());
|
||||
const value = new WsMessage({ ws: opts?.ws, user: opts?.user });
|
||||
const value = new WsMessage({ ...opts, manager: this } as WssMessageOptions);
|
||||
this.wssMap.set(id, value);
|
||||
return value;
|
||||
}
|
||||
changeId(oldId: string, newId: string) {
|
||||
const value = this.wssMap.get(oldId);
|
||||
const originalValue = this.wssMap.get(newId);
|
||||
if (originalValue) {
|
||||
logger.debug(`WsProxyManager changeId: ${newId} already exists, close old connection`);
|
||||
originalValue.ws.close();
|
||||
originalValue.destroy();
|
||||
}
|
||||
if (value) {
|
||||
this.wssMap.delete(oldId);
|
||||
this.wssMap.set(newId, value);
|
||||
value.id = newId;
|
||||
logger.debug(`WsProxyManager changeId: ${oldId} -> ${newId}`);
|
||||
}
|
||||
}
|
||||
unregister(id: string) {
|
||||
const value = this.wssMap.get(id);
|
||||
@@ -117,9 +178,33 @@ export class WsProxyManager {
|
||||
}
|
||||
return Array.from(this.wssMap.keys());
|
||||
}
|
||||
getIdsInfo(beginWith?: string) {
|
||||
const ids = this.getIds(beginWith);
|
||||
const infoList = this.getInfoList(ids);
|
||||
return {
|
||||
ids,
|
||||
infoList,
|
||||
};
|
||||
}
|
||||
getInfoList(ids: string[]) {
|
||||
return ids.map(id => {
|
||||
const value = this.wssMap.get(id);
|
||||
if (value) {
|
||||
return value.getInfo();
|
||||
}
|
||||
return null;
|
||||
}).filter(Boolean);
|
||||
}
|
||||
get(id: string) {
|
||||
return this.wssMap.get(id);
|
||||
}
|
||||
createId(id: string) {
|
||||
if (!this.wssMap.has(id)) {
|
||||
return id;
|
||||
}
|
||||
const newId = id + '-' + nanoid(6);
|
||||
return newId;
|
||||
}
|
||||
checkConnceted() {
|
||||
const that = this;
|
||||
setTimeout(() => {
|
||||
@@ -132,4 +217,40 @@ export class WsProxyManager {
|
||||
that.checkConnceted();
|
||||
}, this.PING_INTERVAL);
|
||||
}
|
||||
async createNewConnection(opts: { ws: any; user: string, userApp: string, isLogin?: boolean }) {
|
||||
const id = opts.userApp;
|
||||
let realId: string = id;
|
||||
const isLogin = opts.isLogin || false;
|
||||
const has = this.wssMap.has(id);
|
||||
const registryId = '-registry-' + generateRegistryId(); // 生成一个随机六位字符串作为注册 ID
|
||||
let isNeedVerify = !isLogin;
|
||||
if (has) {
|
||||
const value = this.wssMap.get(id);
|
||||
if (value) {
|
||||
if (value.isClosed()) {
|
||||
// 短时间内还在, 等于简单重启了一下应用,不需要重新注册.
|
||||
logger.debug('之前的连接已关闭，复用注册 ID 连接 ws', id);
|
||||
this.unregister(id);
|
||||
await new Promise(resolve => setTimeout(resolve, 100));
|
||||
const wsMessage = this.register(id, { ws: opts.ws, user: opts.user, id });
|
||||
wsMessage.sendConnected();
|
||||
return { wsMessage, isNew: false, id: id };
|
||||
} else {
|
||||
// 没有关闭,需要重新注册鉴权一下, 生成新的 id 连接.
|
||||
isNeedVerify = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
// 没有连接, 直接注册新的连接.
|
||||
if (isNeedVerify) {
|
||||
realId = id + registryId;
|
||||
logger.debug('未登录用户，使用临时注册 ID 连接 ws', realId);
|
||||
}
|
||||
const wsMessage = this.register(realId, { ws: opts.ws, user: opts.user, id: realId });
|
||||
return { wsMessage, isNew: true, id: realId };
|
||||
}
|
||||
}
|
||||
// 生成一个随机六位字符串作为注册 ID
|
||||
const generateRegistryId = () => {
|
||||
return Math.random().toString(36).substring(2, 8);
|
||||
}
|
||||
Reference in New Issue
Block a user