9058363707

Committed 13 May 2024 06:36AM UTC coverage: 62.97% (-0.04%) from 63.007%

Build # 9058363707

Build Type

Pull #84

github

Committed by

web-flow

Commit Message

Merge 538c4dc2e into d7d59e4ce

Pull Request Pull Request #84: optimize the time of socket send data

Run Details

1339 of 1517 branches covered (88.27%)

Branch coverage included in aggregate %.

56 of 70 new or added lines in 2 files covered. (80.0%)

1 existing line in 1 file now uncovered.

5463 of 9285 relevant lines covered (58.84%)

32.72 hits per line

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

91.96

/packages/base/src/socket-base.ts

/**
 * socket base class
 * 不同平台 socket 的 api 不同但功能相同，这里抽象一层
 */

import { SpyMessage, SpySocket, SpyBase } from '@huolala-tech/page-spy-types';
import { PackedEvent } from '@huolala-tech/page-spy-types/lib/socket-event';
import { getRandomId, psLog, stringifyData } from './index';
import {
  makeMessage,
  makeBroadcastMessage,
  makeUnicastMessage,
} from './message';
import * as SERVER_MESSAGE_TYPE from './message/server-type';
import atom from './atom';

type InteractiveType = SpyMessage.InteractiveType;
type InternalMsgType = SpyMessage.InternalMsgType;
interface GetterMember {
  key: string; // 属性名
  id: string; // 当前键的 id
  parentId: string; // 当前属性所在容器的 id
  instanceId: string; // 当前实例的 id
}

export type WebSocketEvents = keyof WebSocketEventMap;
type CallbackType = (data?: any) => void;

// fork WebSocket state
export enum SocketState {
  CONNECTING = 0,
  OPEN = 1,
  CLOSING = 2,
  CLOSED = 3,
}

const HEARTBEAT_INTERVAL = 5000;

// The reconnect interval has an initial time of 2000 ms,
// for each failed reconnection attempt, the time will be increased by 1.5x,
// until the attempt number reaches 4, the time will be fixed, which is Math.pow(1.5, 4) * 2000.
// retry interval
const INIT_RETRY_INTERVAL = 2000;
// retry interval time will increase by 1.5x each time.
const RETRY_TIME_INCR = 1.5;
// the time increase pow limit.
const MAX_RETRY_INTERVAL = Math.pow(RETRY_TIME_INCR, 4) * INIT_RETRY_INTERVAL;

// 封装不同平台的 socket
export abstract class SocketWrapper {
  abstract init(url: string): void;
  abstract send(data: string): void;
  abstract close(data?: {}): void;
  abstract getState(): SocketState;
  protected events: Record<WebSocketEvents, CallbackType[]> = {
    open: [],
    close: [],
    error: [],
    message: [],
  };

  protected emit(event: WebSocketEvents, data: any) {
    this.events[event].forEach((fun) => {
      fun(data);
    });
    // for close and error, clear all listeners or they will be called on next socket instance.
    if (event === 'close' || event === 'error') {
      this.clearListeners();
    }
  }

  onOpen(fun: (res: { header?: Record<string, string> }) => void) {
    this.events.open.push(fun);
  }

  onClose(fun: (res: { code: number; reason: string }) => void) {
    this.events.close.push(fun);
  }

  onError(fun: (msg: string) => void) {
    this.events.error.push(fun);
  }

  onMessage(fun: (data: string | ArrayBuffer) => void) {
    this.events.message.push(fun);
  }

  clearListeners() {
    // clear listeners
    Object.entries(this.events).forEach(([, funs]) => {
      funs.splice(0);
    });
  }
}

export abstract class SocketStoreBase {
  protected abstract socketWrapper: SocketWrapper;

  protected abstract updateRoomInfo(): void;

  private socketUrl: string = '';

  private socketConnection: SpySocket.Connection | null = null;

  private debuggerConnection: SpySocket.Connection | null = null;

  // ping timer used for send next ping.
  // a ping is sent after last msg (normal msg or pong) received.
  private pingTimer: ReturnType<typeof setTimeout> | null = null;

  // pong timer used for waiting for pong, if pong not received, close the connection
  private pongTimer: ReturnType<typeof setTimeout> | null = null;

  private retryTimer: ReturnType<typeof setTimeout> | null = null;

  // Cache messages only in online mode
  public isOffline = false;

  // Maximum message length,
  // the 0 meant no limitation.
  public messageCapacity: number = 0;

  // messages store
  private messages: SpySocket.BroadcastEvent[] = [];

  // events center
  private events: Record<
    InteractiveType | InternalMsgType,
    SpyBase.EventCallback[]
  > = {
    debug: [],
    refresh: [],
    'atom-detail': [],
    'atom-getter': [],
    'debugger-online': [],
    'database-pagination': [],
    'public-data': [],
  };

  // initial retry interval.
  private retryInterval = INIT_RETRY_INTERVAL;

  // indicated connected  whether or not
  public connectionStatus: boolean = false;

  // response message filters, to handle some wired messages
  public static messageFilters: ((data: any) => any)[] = [];

  constructor() {
    this.addListener('atom-detail', SocketStoreBase.handleResolveAtom);
    this.addListener('atom-getter', SocketStoreBase.handleAtomPropertyGetter);
    this.addListener('debugger-online', this.handleFlushBuffer);
  }

  // Simple offline listener
  abstract onOffline(): void;

  public init(url: string) {
    try {
      if (!url) {
        throw Error('WebSocket url cannot be empty');
      }
      this.socketWrapper.clearListeners();
      // close existing connection
      if (this.socketWrapper.getState() === SocketState.OPEN) {
        this.socketWrapper.close();
      }
      this.socketWrapper?.onOpen(() => {
        this.connectOnline();
      });
      // Strictly, the onMessage should be called after onOpen. But for some platform(alipay,)
      // this may cause some message losing.
      this.socketWrapper?.onMessage((evt) => {
        this.handleMessage(evt);
      });
      this.socketWrapper?.onClose(() => {
        this.connectOffline();
      });
      this.socketWrapper?.onError(() => {
        // we treat on error the same with on close.
        this.connectOffline();
      });
      this.socketUrl = url;
      this.socketWrapper?.init(url);
    } catch (e: any) {
      psLog.error(e.message);
    }
  }

  public addListener(
    type: SpyMessage.InteractiveType,
    fn: SpyBase.InteractiveEventCallback,
  ): void;
  public addListener(
    type: InternalMsgType,
    fn: SpyBase.InternalEventCallback,
  ): void;
  public addListener(type: any, fn: any) {
    /* c8 ignore next 3 */
    if (!this.events[type]) {
      this.events[type] = [];
    }
    this.events[type].push(fn);
  }

  public broadcastMessage(
    msg: SpyMessage.MessageItem<SpyMessage.DataType>,
    noCache: boolean = false,
  ) {
    const message = makeBroadcastMessage(msg);
    this.send(message, noCache);
  }

  public close() {
    this.clearPing();
    this.socketWrapper?.close();
    this.messages = [];
    Object.entries(this.events).forEach(([, fns]) => {
      fns.splice(0);
    });
  }

  private connectOnline() {
    this.connectionStatus = true;
    this.retryInterval = INIT_RETRY_INTERVAL;
    this.updateRoomInfo();
    this.ping();
  }

  private connectOffline() {
    this.connectionStatus = false;
    this.socketConnection = null;
    this.debuggerConnection = null;
    this.clearPing();
    if (this.retryTimer) {
      clearTimeout(this.retryTimer);
    }

    this.retryTimer = setTimeout(() => {
      if (this.retryInterval < MAX_RETRY_INTERVAL) {
        this.retryInterval *= RETRY_TIME_INCR;
      }
      this.retryTimer = null;
      this.tryReconnect();
    }, this.retryInterval);
  }

  tryReconnect() {
    this.init(this.socketUrl);
  }

  private ping() {
    if (this.pingTimer) {
      clearTimeout(this.pingTimer);
    }
    if (this.pongTimer) {
      clearTimeout(this.pongTimer);
    }
    /* c8 ignore start */
    this.pingTimer = setTimeout(() => {
      this.send({
        type: 'ping',
        content: null,
      });
      this.pingTimer = null;
      this.pongTimer = setTimeout(() => {
        // lost connection
        this.connectOffline();
        this.pongTimer = null;
      }, HEARTBEAT_INTERVAL);
    }, HEARTBEAT_INTERVAL);
    /* c8 ignore stop */
  }

  private clearPing() {
    if (this.pingTimer) {
      clearTimeout(this.pingTimer);
      this.pingTimer = null;
    }
    if (this.pongTimer) {
      clearTimeout(this.pongTimer);
      this.pongTimer = null;
    }
  }

  private handlePong() {
    clearTimeout(this.pongTimer!);
    this.pongTimer = null;
    this.ping();
  }

  // get the data which we expected from nested structure of the message
  protected handleMessage(evt: any) {
    if (SocketStoreBase.messageFilters.length) {
      SocketStoreBase.messageFilters.forEach((filter) => {
        evt = filter(evt);
      });
    }
    const {
      CONNECT,
      MESSAGE,
      ERROR,
      JOIN,
      PING,
      PONG,
      LEAVE,
      CLOSE,
      BROADCAST,
    } = SERVER_MESSAGE_TYPE;
    const result = JSON.parse(evt.data) as SpySocket.Event;
    const { type } = result;
    switch (type) {
      case CONNECT:
        const { selfConnection, roomConnections } = result.content;
        this.socketConnection = selfConnection;
        this.debuggerConnection =
          roomConnections.find((i) => i.userId === 'Debugger') || null;
        break;
      case JOIN:
      case LEAVE:
        const { connection } = result.content;
        if (connection.userId === 'Debugger') {
          if (type === JOIN) {
            this.debuggerConnection = connection;
          } else {
            this.debuggerConnection = null;
          }
        }
        break;
      case MESSAGE:
        const { data, from, to } = result.content;
        if (to.address === this.socketConnection?.address) {
          this.dispatchEvent(data.type, {
            source: data,
            from,
            to,
          });
        }
        break;
      case CLOSE:
      case ERROR:
        // TODO: we should handle this error
        // if (result.content.code === 'RoomNotFoundError') {
        //   // Room not exist, might because used an old connection.
        // }
        this.connectOffline();
        break;
      /* c8 ignore start */
      case PONG:
      case PING:
      case BROADCAST:
      default:
        // noting
        break;
      /* c8 ignore stop */
    }
    // whatever the type is, we should handle pong
    this.handlePong();
  }

  public dispatchEvent(
    type: SpyMessage.InteractiveType,
    data: SpyBase.InteractiveEvent,
  ): void;
  public dispatchEvent(type: InternalMsgType, data: any): void;
  public dispatchEvent(type: any, data: any) {
    if (['public-data'].includes(type)) {
      this.events['public-data'].forEach((fn) => {
        (fn as SpyBase.InternalEventCallback)(data);
      });
      return;
    }
    this.events[type]?.forEach((fn) => {
      (fn as SpyBase.InteractiveEventCallback).call(
        this,
        data,
        (d: SpyMessage.MessageItem<SpyMessage.InteractiveType>) => {
          this.unicastMessage(d, data.from);
        },
      );
    });
  }

  private unicastMessage(
    msg: SpyMessage.MessageItem<SpyMessage.InteractiveType>,
    to: SpySocket.Connection,
  ) {
    const message = makeUnicastMessage(msg, this.socketConnection!, to);
    this.send(message);
  }

  private handleFlushBuffer(
    message: SpyBase.InteractiveEvent<{ latestId: string }>,
  ) {
    const { latestId } = message.source.data;

    const msgIndex = this.messages.findIndex(
      (i) => i.content.data.data.id === latestId,
    );

    /* c8 ignore start */
    this.messages.slice(msgIndex + 1).forEach((msg) => {
      const data: SpySocket.UnicastEvent = {
        type: SERVER_MESSAGE_TYPE.MESSAGE,
        content: {
          data: msg.content.data as any,
          from: this.socketConnection!,
          to: message.from,
        },
      };
      this.send(data, true);
    });
    /* c8 ignore stop */
  }

  private static handleResolveAtom(
    { source }: SpyBase.InteractiveEvent<string>,
    reply: (data: any) => void,
  ) {
    const { type, data } = source;
    if (type === 'atom-detail') {
      const atomData = atom.get(data) || {};
      const msg = makeMessage(`atom-detail-${data}`, atomData, false);
      reply(msg);
    }
  }

  private static handleAtomPropertyGetter(
    { source }: SpyBase.InteractiveEvent<GetterMember>,
    reply: (data: any) => void,
  ) {
    const { type, data } = source;
    if (type === 'atom-getter') {
      const { id, parentId, key, instanceId } = data;
      const instance = atom.getOrigin(instanceId);
      const current = atom.getOrigin(parentId);
      let value = {};
      /* c8 ignore start */
      if (instance && current) {
        value = Object.getOwnPropertyDescriptor(current, key)?.get?.call(
          instance,
        );
      } else {
        value = new Error('Getter computed failed');
      }
      /* c8 ignore stop */
      const msg = makeMessage(`atom-getter-${id}`, atom.transformToAtom(value));
      reply(msg);
    }
  }

  protected send(msg: SpySocket.ClientEvent, noCache: boolean = false) {
    const sendable = this.checkIfSend(msg);
    if (sendable) {
      /* c8 ignore start */
      try {
        const pkMsg = msg as PackedEvent;
        pkMsg.createdAt = Date.now();
        pkMsg.requestId = getRandomId();
        const dataString = stringifyData(pkMsg);
        this.socketWrapper?.send(dataString);
      } catch (e) {
        psLog.error(`Incompatible: ${(e as Error).message}`);
        if (this.connectionStatus) {
          this.connectOffline();
        }
      }
      /* c8 ignore stop */
    }
    const cacheable = this.checkIfCache(msg, noCache);
    if (cacheable) {
      if (
        this.messageCapacity !== 0 &&
        this.messages.length >= this.messageCapacity
      ) {
        this.messages.shift();
      }
      this.messages.push(msg as SpySocket.BroadcastEvent);
    }
  }

  private checkIfSend(msg: SpySocket.ClientEvent) {
    if (this.socketWrapper.getState() !== SocketState.OPEN) return false;
    if (
      [SERVER_MESSAGE_TYPE.UPDATE_ROOM_INFO, SERVER_MESSAGE_TYPE.PING].includes(
        msg.type,
      )
    ) {
      return true;
    }

    if (!this.debuggerConnection) return false;
    return true;
  }

  private checkIfCache(msg: SpySocket.ClientEvent, noCache: boolean = false) {
    if (this.isOffline || noCache) return false;
    if (
      [SERVER_MESSAGE_TYPE.MESSAGE, SERVER_MESSAGE_TYPE.PING].includes(msg.type)
    ) {
      return false;
    }
    return true;
  }
}

1	/**	2✔
2	* socket base class	2✔
3	* 不同平台 socket 的 api 不同但功能相同，这里抽象一层	2✔
4	*/	2✔
5		2✔
6	import { SpyMessage, SpySocket, SpyBase } from '@huolala-tech/page-spy-types';	2✔
7	import { PackedEvent } from '@huolala-tech/page-spy-types/lib/socket-event';	2✔
8	import { getRandomId, psLog, stringifyData } from './index';	2✔
9	import {	2✔
10	makeMessage,	2✔
11	makeBroadcastMessage,	2✔
12	makeUnicastMessage,	2✔
13	} from './message';	2✔
14	import * as SERVER_MESSAGE_TYPE from './message/server-type';	2✔
15	import atom from './atom';	2✔
16		2✔
17	type InteractiveType = SpyMessage.InteractiveType;	2✔
18	type InternalMsgType = SpyMessage.InternalMsgType;	2✔
19	interface GetterMember {	2✔
20	key: string; // 属性名	2✔
21	id: string; // 当前键的 id	2✔
22	parentId: string; // 当前属性所在容器的 id	2✔
23	instanceId: string; // 当前实例的 id	2✔
24	}	2✔
25		2✔
26	export type WebSocketEvents = keyof WebSocketEventMap;	2✔
27	type CallbackType = (data?: any) => void;	2✔
28		2✔
29	// fork WebSocket state	2✔
30	export enum SocketState {	34✔
31	CONNECTING = 0,	34✔
32	OPEN = 1,	34✔
33	CLOSING = 2,	34✔
34	CLOSED = 3,	34✔
35	}	34✔
36		2✔
37	const HEARTBEAT_INTERVAL = 5000;	2✔
38		2✔
39	// The reconnect interval has an initial time of 2000 ms,	2✔
40	// for each failed reconnection attempt, the time will be increased by 1.5x,	2✔
41	// until the attempt number reaches 4, the time will be fixed, which is Math.pow(1.5, 4) * 2000.	2✔
42	// retry interval	2✔
43	const INIT_RETRY_INTERVAL = 2000;	2✔
44	// retry interval time will increase by 1.5x each time.	2✔
45	const RETRY_TIME_INCR = 1.5;	2✔
46	// the time increase pow limit.	2✔
47	const MAX_RETRY_INTERVAL = Math.pow(RETRY_TIME_INCR, 4) * INIT_RETRY_INTERVAL;	2✔
48		2✔
49	// 封装不同平台的 socket	2✔
50	export abstract class SocketWrapper {	2✔
51	abstract init(url: string): void;	2✔
52	abstract send(data: string): void;	2✔
53	abstract close(data?: {}): void;	2✔
54	abstract getState(): SocketState;	2✔
55	protected events: Record<WebSocketEvents, CallbackType[]> = {	2✔
56	open: [],	40✔
57	close: [],	40✔
58	error: [],	40✔
59	message: [],	40✔
60	};	40✔
61		2✔
62	protected emit(event: WebSocketEvents, data: any) {	2✔
63	this.events[event].forEach((fun) => {	×
64	fun(data);	×
65	});	×
66	// for close and error, clear all listeners or they will be called on next socket instance.	×
67	if (event === 'close' \|\| event === 'error') {	×
68	this.clearListeners();	×
69	}	×
70	}	×
71		2✔
72	onOpen(fun: (res: { header?: Record<string, string> }) => void) {	2✔
73	this.events.open.push(fun);	34✔
74	}	34✔
75		2✔
76	onClose(fun: (res: { code: number; reason: string }) => void) {	2✔
77	this.events.close.push(fun);	34✔
78	}	34✔
79		2✔
80	onError(fun: (msg: string) => void) {	2✔
81	this.events.error.push(fun);	34✔
82	}	34✔
83		2✔
84	onMessage(fun: (data: string \| ArrayBuffer) => void) {	2✔
85	this.events.message.push(fun);	34✔
86	}	34✔
87		2✔
88	clearListeners() {	2✔
89	// clear listeners	34✔
90	Object.entries(this.events).forEach(([, funs]) => {	34✔
91	funs.splice(0);	136✔
92	});	34✔
93	}	34✔
94	}	2✔
95		2✔
96	export abstract class SocketStoreBase {	2✔
97	protected abstract socketWrapper: SocketWrapper;	2✔
98		2✔
99	protected abstract updateRoomInfo(): void;	2✔
100		2✔
101	private socketUrl: string = '';	2✔
102		40✔
103	private socketConnection: SpySocket.Connection \| null = null;	40✔
104		40✔
105	private debuggerConnection: SpySocket.Connection \| null = null;	40✔
106		40✔
107	// ping timer used for send next ping.	40✔
108	// a ping is sent after last msg (normal msg or pong) received.	40✔
109	private pingTimer: ReturnType<typeof setTimeout> \| null = null;	40✔
110		40✔
111	// pong timer used for waiting for pong, if pong not received, close the connection	40✔
112	private pongTimer: ReturnType<typeof setTimeout> \| null = null;	40✔
113		40✔
114	private retryTimer: ReturnType<typeof setTimeout> \| null = null;	40✔
115		40✔
116	// Cache messages only in online mode	40✔
117	public isOffline = false;	40✔
118		40✔
119	// Maximum message length,	40✔
120	// the 0 meant no limitation.	40✔
121	public messageCapacity: number = 0;	40✔
122		40✔
123	// messages store	40✔
124	private messages: SpySocket.BroadcastEvent[] = [];	40✔
125		40✔
126	// events center	40✔
127	private events: Record<	40✔
128	InteractiveType \| InternalMsgType,	40✔
129	SpyBase.EventCallback[]	40✔
130	> = {	40✔
131	debug: [],	40✔
132	refresh: [],	40✔
133	'atom-detail': [],	40✔
134	'atom-getter': [],	40✔
135	'debugger-online': [],	40✔
136	'database-pagination': [],	40✔
137	'public-data': [],	40✔
138	};	40✔
139		40✔
140	// initial retry interval.	40✔
141	private retryInterval = INIT_RETRY_INTERVAL;	40✔
142		40✔
143	// indicated connected whether or not	40✔
144	public connectionStatus: boolean = false;	2✔
145		2✔
146	// response message filters, to handle some wired messages	2✔
147	public static messageFilters: ((data: any) => any)[] = [];	2✔
148		2✔
149	constructor() {	2✔
150	this.addListener('atom-detail', SocketStoreBase.handleResolveAtom);	40✔
151	this.addListener('atom-getter', SocketStoreBase.handleAtomPropertyGetter);	40✔
152	this.addListener('debugger-online', this.handleFlushBuffer);	40✔
153	}	40✔
154		2✔
155	// Simple offline listener	2✔
156	abstract onOffline(): void;	2✔
157		2✔
158	public init(url: string) {	2✔
159	try {	36✔
160	if (!url) {	36✔
161	throw Error('WebSocket url cannot be empty');	2✔
162	}	2✔
163	this.socketWrapper.clearListeners();	34✔
164	// close existing connection	34✔
165	if (this.socketWrapper.getState() === SocketState.OPEN) {	36!
166	this.socketWrapper.close();	×
167	}	×
168	this.socketWrapper?.onOpen(() => {	36✔
169	this.connectOnline();	10✔
170	});	36✔
171	// Strictly, the onMessage should be called after onOpen. But for some platform(alipay,)	36✔
172	// this may cause some message losing.	36✔
173	this.socketWrapper?.onMessage((evt) => {	36✔
174	this.handleMessage(evt);	12✔
175	});	36✔
176	this.socketWrapper?.onClose(() => {	36✔
177	this.connectOffline();	24✔
178	});	36✔
179	this.socketWrapper?.onError(() => {	36✔
180	// we treat on error the same with on close.	14✔
181	this.connectOffline();	14✔
182	});	36✔
183	this.socketUrl = url;	36✔
184	this.socketWrapper?.init(url);	36✔
185	} catch (e: any) {	36✔
186	psLog.error(e.message);	2✔
187	}	2✔
188	}	36✔
189		2✔
190	public addListener(	2✔
191	type: SpyMessage.InteractiveType,	2✔
192	fn: SpyBase.InteractiveEventCallback,	2✔
193	): void;	2✔
194	public addListener(	2✔
195	type: InternalMsgType,	2✔
196	fn: SpyBase.InternalEventCallback,	2✔
197	): void;	2✔
198	public addListener(type: any, fn: any) {	2✔
199	/* c8 ignore next 3 */	2✔
200	if (!this.events[type]) {	2✔
201	this.events[type] = [];	2✔
202	}	2✔
203	this.events[type].push(fn);	244✔
204	}	244✔
205		2✔
206	public broadcastMessage(	2✔
207	msg: SpyMessage.MessageItem<SpyMessage.DataType>,	1,962✔
208	noCache: boolean = false,	1,962✔
209	) {	1,962✔
210	const message = makeBroadcastMessage(msg);	1,962✔
211	this.send(message, noCache);	1,962✔
212	}	1,962✔
213		2✔
214	public close() {	2✔
215	this.clearPing();	20✔
216	this.socketWrapper?.close();	20✔
217	this.messages = [];	20✔
218	Object.entries(this.events).forEach(([, fns]) => {	20✔
219	fns.splice(0);	140✔
220	});	20✔
221	}	20✔
222		2✔
223	private connectOnline() {	2✔
224	this.connectionStatus = true;	10✔
225	this.retryInterval = INIT_RETRY_INTERVAL;	10✔
226	this.updateRoomInfo();	10✔
227	this.ping();	10✔
228	}	10✔
229		2✔
230	private connectOffline() {	2✔
231	this.connectionStatus = false;	40✔
232	this.socketConnection = null;	40✔
233	this.debuggerConnection = null;	40✔
234	this.clearPing();	40✔
235	if (this.retryTimer) {	40✔
236	clearTimeout(this.retryTimer);	16✔
237	}	16✔
238		40✔
239	this.retryTimer = setTimeout(() => {	40✔
240	if (this.retryInterval < MAX_RETRY_INTERVAL) {	14✔
241	this.retryInterval *= RETRY_TIME_INCR;	10✔
242	}	10✔
243	this.retryTimer = null;	14✔
244	this.tryReconnect();	14✔
245	}, this.retryInterval);	40✔
246	}	40✔
247		2✔
248	tryReconnect() {	2✔
249	this.init(this.socketUrl);	14✔
250	}	14✔
251		2✔
252	private ping() {	2✔
253	if (this.pingTimer) {	22✔
254	clearTimeout(this.pingTimer);	10✔
255	}	10✔
256	if (this.pongTimer) {	22!
257	clearTimeout(this.pongTimer);	×
258	}	×
259	/* c8 ignore start */	2✔
260	this.pingTimer = setTimeout(() => {	2✔
261	this.send({	2✔
262	type: 'ping',	2✔
263	content: null,	2✔
264	});	2✔
265	this.pingTimer = null;	2✔
266	this.pongTimer = setTimeout(() => {	2✔
267	// lost connection	2✔
268	this.connectOffline();	2✔
269	this.pongTimer = null;	2✔
270	}, HEARTBEAT_INTERVAL);	2✔
271	}, HEARTBEAT_INTERVAL);	2✔
272	/* c8 ignore stop */	2✔
273	}	22✔
274		2✔
275	private clearPing() {	2✔
276	if (this.pingTimer) {	60✔
277	clearTimeout(this.pingTimer);	12✔
278	this.pingTimer = null;	12✔
279	}	12✔
280	if (this.pongTimer) {	60!
281	clearTimeout(this.pongTimer);	×
282	this.pongTimer = null;	×
283	}	×
284	}	60✔
285		2✔
286	private handlePong() {	2✔
287	clearTimeout(this.pongTimer!);	12✔
288	this.pongTimer = null;	12✔
289	this.ping();	12✔
290	}	12✔
291		2✔
292	// get the data which we expected from nested structure of the message	2✔
293	protected handleMessage(evt: any) {	2✔
294	if (SocketStoreBase.messageFilters.length) {	12!
295	SocketStoreBase.messageFilters.forEach((filter) => {	×
296	evt = filter(evt);	×
297	});	×
298	}	×
299	const {	12✔
300	CONNECT,	12✔
301	MESSAGE,	12✔
302	ERROR,	12✔
303	JOIN,	12✔
304	PING,	12✔
305	PONG,	12✔
306	LEAVE,	12✔
307	CLOSE,	12✔
308	BROADCAST,	12✔
309	} = SERVER_MESSAGE_TYPE;	12✔
310	const result = JSON.parse(evt.data) as SpySocket.Event;	12✔
311	const { type } = result;	12✔
312	switch (type) {	12✔
313	case CONNECT:	12✔
314	const { selfConnection, roomConnections } = result.content;	2✔
315	this.socketConnection = selfConnection;	2✔
316	this.debuggerConnection =	2✔
317	roomConnections.find((i) => i.userId === 'Debugger') \|\| null;	2!
318	break;	2✔
319	case JOIN:	12!
320	case LEAVE:	12!
NEW 321	const { connection } = result.content;	×
NEW 322	if (connection.userId === 'Debugger') {	×
NEW 323	if (type === JOIN) {	×
NEW 324	this.debuggerConnection = connection;	×
NEW 325	} else {	×
NEW 326	this.debuggerConnection = null;	×
NEW 327	}	×
NEW 328	}	×
UNCOV 329	break;	×
330	case MESSAGE:	12✔
331	const { data, from, to } = result.content;	8✔
332	if (to.address === this.socketConnection?.address) {	8✔
333	this.dispatchEvent(data.type, {	8✔
334	source: data,	8✔
335	from,	8✔
336	to,	8✔
337	});	8✔
338	}	8✔
339	break;	8✔
340	case CLOSE:	12!
341	case ERROR:	12✔
342	// TODO: we should handle this error	2✔
343	// if (result.content.code === 'RoomNotFoundError') {	2✔
344	// // Room not exist, might because used an old connection.	2✔
345	// }	2✔
346	this.connectOffline();	2✔
347	break;	2✔
348	/* c8 ignore start */	2✔
349	case PONG:	2✔
350	case PING:	2✔
351	case BROADCAST:	2✔
352	default:	2✔
353	// noting	2✔
354	break;	2✔
355	/* c8 ignore stop */	2✔
356	}	12✔
357	// whatever the type is, we should handle pong	12✔
358	this.handlePong();	12✔
359	}	12✔
360		2✔
361	public dispatchEvent(	2✔
362	type: SpyMessage.InteractiveType,	2✔
363	data: SpyBase.InteractiveEvent,	2✔
364	): void;	2✔
365	public dispatchEvent(type: InternalMsgType, data: any): void;	2✔
366	public dispatchEvent(type: any, data: any) {	2✔
367	if (['public-data'].includes(type)) {	384✔
368	this.events['public-data'].forEach((fn) => {	370✔
369	(fn as SpyBase.InternalEventCallback)(data);	×
370	});	370✔
371	return;	370✔
372	}	370✔
373	this.events[type]?.forEach((fn) => {	384✔
374	(fn as SpyBase.InteractiveEventCallback).call(	14✔
375	this,	14✔
376	data,	14✔
377	(d: SpyMessage.MessageItem<SpyMessage.InteractiveType>) => {	14✔
378	this.unicastMessage(d, data.from);	10✔
379	},	14✔
380	);	14✔
381	});	384✔
382	}	384✔
383		2✔
384	private unicastMessage(	2✔
385	msg: SpyMessage.MessageItem<SpyMessage.InteractiveType>,	10✔
386	to: SpySocket.Connection,	10✔
387	) {	10✔
388	const message = makeUnicastMessage(msg, this.socketConnection!, to);	10✔
389	this.send(message);	10✔
390	}	10✔
391		2✔
392	private handleFlushBuffer(	2✔
393	message: SpyBase.InteractiveEvent<{ latestId: string }>,	2✔
394	) {	2✔
395	const { latestId } = message.source.data;	2✔
396		2✔
397	const msgIndex = this.messages.findIndex(	2✔
398	(i) => i.content.data.data.id === latestId,	2✔
399	);	2✔
400		2✔
401	/* c8 ignore start */	2✔
402	this.messages.slice(msgIndex + 1).forEach((msg) => {	2✔
403	const data: SpySocket.UnicastEvent = {	2✔
404	type: SERVER_MESSAGE_TYPE.MESSAGE,	2✔
405	content: {	2✔
406	data: msg.content.data as any,	2✔
407	from: this.socketConnection!,	2✔
408	to: message.from,	2✔
409	},	2✔
410	};	2✔
411	this.send(data, true);	2✔
412	});	2✔
413	/* c8 ignore stop */	2✔
414	}	2✔
415		2✔
416	private static handleResolveAtom(	2✔
417	{ source }: SpyBase.InteractiveEvent<string>,	2✔
418	reply: (data: any) => void,	2✔
419	) {	2✔
420	const { type, data } = source;	2✔
421	if (type === 'atom-detail') {	2✔
422	const atomData = atom.get(data) \|\| {};	2✔
423	const msg = makeMessage(`atom-detail-${data}`, atomData, false);	2✔
424	reply(msg);	2✔
425	}	2✔
426	}	2✔
427		2✔
428	private static handleAtomPropertyGetter(	2✔
429	{ source }: SpyBase.InteractiveEvent<GetterMember>,	2✔
430	reply: (data: any) => void,	2✔
431	) {	2✔
432	const { type, data } = source;	2✔
433	if (type === 'atom-getter') {	2✔
434	const { id, parentId, key, instanceId } = data;	2✔
435	const instance = atom.getOrigin(instanceId);	2✔
436	const current = atom.getOrigin(parentId);	2✔
437	let value = {};	2✔
438	/* c8 ignore start */	2✔
439	if (instance && current) {	2✔
440	value = Object.getOwnPropertyDescriptor(current, key)?.get?.call(	2✔
441	instance,	2✔
442	);	2✔
443	} else {	2✔
444	value = new Error('Getter computed failed');	2✔
445	}	2✔
446	/* c8 ignore stop */	2✔
447	const msg = makeMessage(`atom-getter-${id}`, atom.transformToAtom(value));	2✔
448	reply(msg);	2✔
449	}	2✔
450	}	2✔
451		2✔
452	protected send(msg: SpySocket.ClientEvent, noCache: boolean = false) {	2✔
453	const sendable = this.checkIfSend(msg);	1,972✔
454	if (sendable) {	1,972✔
455	/* c8 ignore start */	2✔
456	try {	2✔
457	const pkMsg = msg as PackedEvent;	2✔
458	pkMsg.createdAt = Date.now();	2✔
459	pkMsg.requestId = getRandomId();	2✔
460	const dataString = stringifyData(pkMsg);	2✔
461	this.socketWrapper?.send(dataString);	2✔
462	} catch (e) {	2✔
463	psLog.error(`Incompatible: ${(e as Error).message}`);	2✔
464	if (this.connectionStatus) {	2✔
465	this.connectOffline();	2✔
466	}	2✔
467	}	2✔
468	/* c8 ignore stop */	2✔
469	}	8✔
470	const cacheable = this.checkIfCache(msg, noCache);	1,972✔
471	if (cacheable) {	1,972✔
472	if (	284✔
473	this.messageCapacity !== 0 &&	284✔
474	this.messages.length >= this.messageCapacity	46✔
475	) {	284!
476	this.messages.shift();	×
477	}	×
478	this.messages.push(msg as SpySocket.BroadcastEvent);	284✔
479	}	284✔
480	}	1,972✔
481		2✔
482	private checkIfSend(msg: SpySocket.ClientEvent) {	2✔
483	if (this.socketWrapper.getState() !== SocketState.OPEN) return false;	1,972✔
484	if (	8✔
485	[SERVER_MESSAGE_TYPE.UPDATE_ROOM_INFO, SERVER_MESSAGE_TYPE.PING].includes(	8✔
486	msg.type,	8✔
487	)	8✔
488	) {	8!
NEW 489	return true;	×
NEW 490	}	×
491		8✔
492	if (!this.debuggerConnection) return false;	8!
493	return true;	8✔
494	}	8✔
495		2✔
496	private checkIfCache(msg: SpySocket.ClientEvent, noCache: boolean = false) {	2✔
497	if (this.isOffline \|\| noCache) return false;	1,972✔
498	if (	294✔
499	[SERVER_MESSAGE_TYPE.MESSAGE, SERVER_MESSAGE_TYPE.PING].includes(msg.type)	294✔
500	) {	1,948✔
501	return false;	10✔
502	}	10✔
503	return true;	284✔
504	}	284✔
505	}	2✔

HuolalaTech / page-spy / 9058363707

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