Resolve "Blockchain Adapter: Add Chaingraph Adapter"
This commit is contained in:
devperate
@@ -0,0 +1,465 @@
|
||||
import { WebsocketClient } from './ws-client.js';
|
||||
import { EventEmitter } from './event-emitter.js';
|
||||
|
||||
import { jsonToGraphQLQuery } from 'json-to-graphql-query';
|
||||
import { z } from 'zod';
|
||||
|
||||
// The format of the querys accepted by 'request' and 'subscribe' functions.
|
||||
export type GraphqlFetchQuery =
|
||||
| Record<'query', unknown>
|
||||
| Record<'mutation', unknown>;
|
||||
export type GraphqlSubscriptionQuery = Record<'subscription', unknown>;
|
||||
|
||||
// Format for query variables.
|
||||
export type GraphqlQueryVariables = Record<string, string | number>;
|
||||
|
||||
// Shape of the Connection Initialization Message chaingraph accepts.
|
||||
type ConnectionInitMessage = {
|
||||
type: 'connection_init';
|
||||
// 'Record<never, never>' stand for empty payload.
|
||||
payload: Record<never, never> | Record<'token', string>;
|
||||
};
|
||||
|
||||
// Shape of the data query used for Requests and Subscriptions.
|
||||
type DataRequestMessage = {
|
||||
id: string;
|
||||
type: 'start';
|
||||
payload: {
|
||||
queryId: string;
|
||||
query: string;
|
||||
variables?: GraphqlQueryVariables;
|
||||
};
|
||||
};
|
||||
|
||||
// Shape of the unsubscription query.
|
||||
type UnsubscriptionMessage = {
|
||||
id: string;
|
||||
type: 'stop';
|
||||
};
|
||||
|
||||
// Shape of a chaingraph response.
|
||||
export type GraphqlRequestResponse = {
|
||||
data: Record<string, unknown>;
|
||||
};
|
||||
|
||||
// Shape of the subscription objects stored in our Map of active subscriptions.
|
||||
type Subscriptions = {
|
||||
queryId: string;
|
||||
|
||||
body: {
|
||||
type: DataRequestMessage['type'];
|
||||
|
||||
payload: {
|
||||
query: GraphqlSubscriptionQuery;
|
||||
variables?: GraphqlQueryVariables;
|
||||
};
|
||||
};
|
||||
};
|
||||
|
||||
export type GraphqlClientEventsPayload =
|
||||
| Record<string, unknown>
|
||||
| Error
|
||||
| boolean;
|
||||
|
||||
export type GraphQLClientEvents<T extends string = string> = {
|
||||
/**
|
||||
* Emitted when data has been received from a subscription via websocket.
|
||||
* Emitted when the socket has failed in some way.
|
||||
*
|
||||
* List of events emitted:
|
||||
* 'isConnectionInitialized' : boolean emitted when graphql protocol handshake was accepted and websocket messaging started.
|
||||
* 'error' : Error emitted when the websocket or chaingraph notify about an error.
|
||||
* '<SubscriptionId>' : Record<string, unknown> emitted when server send a message that carries updated data of a subscription made.
|
||||
*
|
||||
* @eventProperty
|
||||
*/
|
||||
[K in T]: GraphqlClientEventsPayload;
|
||||
};
|
||||
|
||||
export class GraphQLClient extends EventEmitter<GraphQLClientEvents> {
|
||||
// Our websocket client implementation that will handle subscriptions.
|
||||
private websocket: WebsocketClient;
|
||||
|
||||
// Track if the websocket connection has been initialized and handshake made.
|
||||
private isWebSocketConnectionInitialized: boolean;
|
||||
|
||||
// Map that contain relative data to currently active subscriptions.
|
||||
// This will allow us to re-subscribe in case of websocket disconnetion.
|
||||
private activeSubscriptions: Map<
|
||||
Subscriptions['queryId'],
|
||||
Subscriptions['body']
|
||||
> = new Map();
|
||||
|
||||
constructor(
|
||||
public endpoint: string,
|
||||
private token?: string,
|
||||
) {
|
||||
super();
|
||||
|
||||
this.isWebSocketConnectionInitialized = false;
|
||||
|
||||
// Defines the subprotocol to comunicate with a graphql server.
|
||||
const options = {
|
||||
protocols: 'graphql-ws',
|
||||
};
|
||||
|
||||
// Adapt the given endpoint to setup a websocket comunication protocol.
|
||||
const websocketHost = this.endpoint.includes('http')
|
||||
? this.endpoint.replace('http', 'ws')
|
||||
: this.endpoint.includes('https')
|
||||
? this.endpoint.replace('https', 'wss')
|
||||
: this.endpoint;
|
||||
|
||||
// Instantiate our WebSocket Client class.
|
||||
this.websocket = new WebsocketClient(websocketHost, options);
|
||||
|
||||
// Subscribe to events emited by our websocket client.
|
||||
this.websocket.on('isOpen', (isOpen: boolean) => {
|
||||
// If WebSocket is open...
|
||||
if (isOpen) {
|
||||
this.handleOpenSocket.bind(this);
|
||||
|
||||
// return to prevent further code execution.
|
||||
return;
|
||||
}
|
||||
|
||||
// Otherwise update internal status to disconnected and emit.
|
||||
this.isWebSocketConnectionInitialized = false;
|
||||
|
||||
this.emit(
|
||||
'isConnectionInitialized',
|
||||
this.isWebSocketConnectionInitialized,
|
||||
);
|
||||
});
|
||||
this.websocket.on('message', this.handleMessageResponse.bind(this));
|
||||
this.websocket.on('error', (error) => {
|
||||
this.emit('error', new Error(`WebSocket error: ${error}`));
|
||||
});
|
||||
}
|
||||
|
||||
private async handleOpenSocket(): Promise<void> {
|
||||
// Re-subscribe to the currently active subscriptions.
|
||||
if (this.activeSubscriptions.size > 0) {
|
||||
// Create a list that will carry the subscriptions to be made.
|
||||
const subscriptionPromises = [];
|
||||
|
||||
for (const key of this.activeSubscriptions.keys()) {
|
||||
const subscriptionBody = this.activeSubscriptions.get(key);
|
||||
|
||||
if (subscriptionBody)
|
||||
subscriptionPromises.push(
|
||||
async () =>
|
||||
await this.subscribe(
|
||||
key,
|
||||
subscriptionBody.payload.query,
|
||||
false,
|
||||
subscriptionBody.payload.variables,
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
// Request the server to run all the subscriptions in parallel.
|
||||
await Promise.all(subscriptionPromises);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse and handle websocket subscription messages done by the graphl server.
|
||||
*/
|
||||
private handleMessageResponse(event: MessageEvent): void {
|
||||
// Validate the payload body.
|
||||
const eventData = GraphQLClient.subscriptionResponseSchema.parse(
|
||||
JSON.parse(event.data),
|
||||
);
|
||||
|
||||
// If Graphql server notifies of error trying to establish connection.
|
||||
if (eventData.type === 'connection_error') {
|
||||
this.isWebSocketConnectionInitialized = false;
|
||||
|
||||
this.emit(
|
||||
'error',
|
||||
new Error(
|
||||
`Error of connection. Code: ${eventData.payload.code}, Message: ${eventData.payload.message}`,
|
||||
),
|
||||
);
|
||||
|
||||
// Return to avoid further code execution.
|
||||
return;
|
||||
}
|
||||
|
||||
// If Graphql server notifies of successfully connection initialized.
|
||||
if (eventData.type === 'connection_ack') {
|
||||
this.isWebSocketConnectionInitialized = true;
|
||||
|
||||
this.emit(
|
||||
'isConnectionInitialized',
|
||||
this.isWebSocketConnectionInitialized,
|
||||
);
|
||||
|
||||
// Return to avoid further code execution.
|
||||
return;
|
||||
}
|
||||
|
||||
// If Graphql server notifies with a data message for a given subscription.
|
||||
if (eventData.type === 'data') {
|
||||
// Check if the payload contain error messages and propagate properly.
|
||||
if (eventData.payload?.errors) {
|
||||
const errorsMessage = (
|
||||
eventData.payload.errors as Array<{ message: string }>
|
||||
)
|
||||
.map((item) => item.message)
|
||||
.join();
|
||||
this.emit('error', new Error(errorsMessage));
|
||||
|
||||
// Return to avoid further code execution.
|
||||
return;
|
||||
}
|
||||
|
||||
// Otherwise just emit the data with the subscription id.
|
||||
this.emit(eventData.id, eventData.payload as GraphqlRequestResponse);
|
||||
|
||||
// Try to split the 'queryId' by `-`.
|
||||
// This indicates that user wants the subscription to emit twice.
|
||||
// Once for the full queryId and once for the part before the `-` separator.
|
||||
const queryId = eventData.id.split('-');
|
||||
// If 'queryId' contained a separator `-` symbol...
|
||||
if (queryId.length > 0)
|
||||
this.emit(queryId[0], eventData.payload as GraphqlRequestResponse);
|
||||
}
|
||||
// If payload type is 'error' propagate it properly.
|
||||
else if (eventData.type === 'error') {
|
||||
this.emit('error', new Error(eventData.payload.message as string));
|
||||
}
|
||||
}
|
||||
|
||||
public stopAllSubscriptions(): void {
|
||||
this.websocket.stop();
|
||||
|
||||
this.activeSubscriptions.clear();
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute a GraphQL query.
|
||||
*
|
||||
* @param {string} queryId The query id or name.
|
||||
* @param {string} query The query object.
|
||||
* @param {string} variables An object containing the query variables.
|
||||
*
|
||||
* @returns {Promise<RequestResponseWithData>} A promise that resolves with data for the given query if successfull.
|
||||
* @throws {Error} If the server encounter an error trying to run the query.
|
||||
* @throws {Error} If the fetch request didn't resolves successfully.
|
||||
*/
|
||||
async request(
|
||||
queryId: string,
|
||||
query: GraphqlFetchQuery,
|
||||
variables?: GraphqlQueryVariables,
|
||||
): Promise<GraphqlRequestResponse> {
|
||||
try {
|
||||
// Build the graphql request.
|
||||
const body: DataRequestMessage['payload'] = {
|
||||
queryId,
|
||||
query: jsonToGraphQLQuery(query),
|
||||
variables: variables ? variables : {},
|
||||
};
|
||||
|
||||
console.log(body);
|
||||
|
||||
const response = await fetch(this.endpoint, {
|
||||
method: 'POST',
|
||||
headers: { 'content-type': 'application/json' },
|
||||
body: JSON.stringify(body),
|
||||
});
|
||||
|
||||
if (!response.ok) {
|
||||
throw new Error(`HTTP fetch error with status: ${response.status}`);
|
||||
}
|
||||
|
||||
// Validate the response payload.
|
||||
const result = GraphQLClient.requestResponseSchema.parse(
|
||||
await response.json(),
|
||||
);
|
||||
|
||||
if ('errors' in result) {
|
||||
const errorsString = result.errors
|
||||
.map((error) => error.message)
|
||||
.join(', ');
|
||||
|
||||
throw new Error(
|
||||
`GraphQL request failed. Detailed errors: ${errorsString}`,
|
||||
);
|
||||
}
|
||||
|
||||
return result;
|
||||
} catch (error: unknown) {
|
||||
throw new Error(`GraphQL request failed. Detailed error: ${error}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute a GraphQL subscription.
|
||||
*
|
||||
* NOTE: If you use a queryId that contains a minus(`-`) char in its name (e.g. MonitorForPayments-0xaqwe),
|
||||
* This will emit twice, once for the full queryId and once for the first part before the `-` separator.
|
||||
* This is usefull if you want to setup notification handlers that are standard for several subscriptions.
|
||||
*
|
||||
* @param {string} queryId The query id or name. Must be unique to avoid conflicts with other active subscriptions.
|
||||
* @param {string} query The query object.
|
||||
* @param {string} waitForResponse Indicates if the subscription must await for response data to return.
|
||||
* @param {string} variables An object containing the query variables.
|
||||
*
|
||||
* @returns {Promise<GraphqlRequestResponse>} A promise that resolves with data for the given query if successfull.
|
||||
* @returns {Promise<void>} Returns void if 'waitForResponse' is false.
|
||||
* @throws {Error} If the server encounter an error trying to run the subscription.
|
||||
* @throws {Error} If the websocket connection is unable to resolve the data exchange successfully.
|
||||
* @throws {Error} If a subscription with the same queryId has been already send.
|
||||
*/
|
||||
async subscribe(
|
||||
queryId: string,
|
||||
query: GraphqlSubscriptionQuery,
|
||||
waitForResponse: boolean,
|
||||
variables?: GraphqlQueryVariables,
|
||||
): Promise<GraphqlRequestResponse | void> {
|
||||
// Check if there is an active subscription with the same id.
|
||||
if (this.activeSubscriptions.has(queryId))
|
||||
throw new Error(`Subscription with id (${queryId} is already in use.)`);
|
||||
|
||||
// A first shakehand with a chaingraph server its necessary before any subscription.
|
||||
if (!this.isWebSocketConnectionInitialized)
|
||||
await this.initializeWebsocketConnection();
|
||||
|
||||
// Build the graphql subscription.
|
||||
const subscription: DataRequestMessage = {
|
||||
id: queryId,
|
||||
type: 'start',
|
||||
payload: {
|
||||
queryId: queryId,
|
||||
query: jsonToGraphQLQuery(query),
|
||||
variables: variables ? variables : {},
|
||||
},
|
||||
};
|
||||
|
||||
await this.websocket.send(JSON.stringify(subscription));
|
||||
|
||||
// If 'waitForResponse' was set to 'true' we proceed to await for the server to return data as if were a normal fetch.
|
||||
// This function will not return until that point.
|
||||
if (waitForResponse) {
|
||||
let dataOrError: GraphqlRequestResponse | Error;
|
||||
|
||||
// After subscribe await for the server to send us back a message with the id of the subscription we just execute.
|
||||
await this.waitFor(queryId, (event) => {
|
||||
// If response is instance of boolean, keep waiting.
|
||||
if (event instanceof Boolean) return false;
|
||||
|
||||
// Since a parsed server message can only be any of these {<GraphQLClientEventsPayload>} 3 types,
|
||||
// We assert that if not boolean the response is a 'Record' of data or an 'Error' and proceed to finallize the await.
|
||||
dataOrError = event as GraphqlRequestResponse | Error;
|
||||
|
||||
return true;
|
||||
});
|
||||
|
||||
// If server response is instance of error...
|
||||
if (dataOrError! instanceof Error)
|
||||
throw new Error(
|
||||
`Error trying to subscribe to ${queryId}. Detailed info: ${dataOrError.message}`,
|
||||
);
|
||||
|
||||
// Otherwise if server response is a 'Record' of valuable data we proceed to store the query as an active subscription.
|
||||
this.activeSubscriptions.set(queryId, {
|
||||
type: 'start',
|
||||
payload: { query, variables: variables || {} },
|
||||
});
|
||||
|
||||
// Since we are waiting for a server response we grant here that 'dataOrError' variable is not undefined.
|
||||
return dataOrError!;
|
||||
}
|
||||
|
||||
// Stay waiting asynchronously for a valid server response before add this query to our 'activeSubscriptions' list.
|
||||
this.waitFor(queryId, (event) => {
|
||||
if (!(event instanceof Error))
|
||||
this.activeSubscriptions.set(queryId, {
|
||||
type: 'start',
|
||||
payload: { query, variables: variables || {} },
|
||||
});
|
||||
|
||||
return true;
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop a given list of subscriptions from server.
|
||||
*
|
||||
* @param {Array<string>} queryIds A list of queryId.
|
||||
*/
|
||||
async unsubscribe(queryIds: Array<string>): Promise<void> {
|
||||
const unsubscribePromises = [];
|
||||
|
||||
for (const id of queryIds) {
|
||||
// Check if the query 'id' is on our list of 'activeSubscriptions'.
|
||||
// If is not, it means that is not an active subscription and there is nothing to unsubscribe.
|
||||
if (!this.activeSubscriptions.has(id)) continue;
|
||||
|
||||
// Build the graphql unsubscription query.
|
||||
const unsubscriptionSignal: UnsubscriptionMessage = {
|
||||
id,
|
||||
type: 'stop',
|
||||
};
|
||||
|
||||
unsubscribePromises.push(async (): Promise<void> => {
|
||||
await this.websocket.send(JSON.stringify(unsubscriptionSignal));
|
||||
|
||||
this.activeSubscriptions.delete(id);
|
||||
});
|
||||
}
|
||||
|
||||
// Run all the unsubscriptions in parallel and wait all of them resolve or reject.
|
||||
await Promise.allSettled(unsubscribePromises);
|
||||
|
||||
// If there is no more 'activeSubscriptions' close the WebSocket connection to save bandwith.
|
||||
if (this.activeSubscriptions.size === 0) this.websocket.stop();
|
||||
}
|
||||
|
||||
private async initializeWebsocketConnection(): Promise<void> {
|
||||
// Build the graphql connection_init query.
|
||||
const connectionInitSignal: ConnectionInitMessage = {
|
||||
type: 'connection_init',
|
||||
payload: this.token ? { token: this.token } : {},
|
||||
};
|
||||
|
||||
await this.websocket.send(JSON.stringify(connectionInitSignal));
|
||||
|
||||
// Wait for the handshake to be accepted by the server.
|
||||
await this.waitFor(
|
||||
'isConnectionInitialized',
|
||||
(isConnectionInitialized) => isConnectionInitialized === true,
|
||||
);
|
||||
}
|
||||
|
||||
// Validations for our subscription response body.
|
||||
public static subscriptionResponseSchema = z.union([
|
||||
z.object({
|
||||
id: z.string(),
|
||||
type: z.enum(['connection_error', 'data', 'error']),
|
||||
payload: z.record(
|
||||
z.string(),
|
||||
z.union([
|
||||
z.record(z.string(), z.unknown()),
|
||||
z.string(),
|
||||
z.array(z.unknown()),
|
||||
]),
|
||||
),
|
||||
}),
|
||||
z.object({
|
||||
type: z.enum(['connection_ack', 'ka']),
|
||||
}),
|
||||
]);
|
||||
|
||||
// Validations for our fetch response body.
|
||||
public static requestResponseSchema = z.union([
|
||||
z.object({
|
||||
data: z.record(z.string(), z.unknown()),
|
||||
}),
|
||||
z.object({
|
||||
errors: z.array(z.object({ message: z.string() })),
|
||||
}),
|
||||
]);
|
||||
}
|
||||
@@ -1,6 +1,7 @@
|
||||
export * from './event-emitter.js';
|
||||
export * from './exponential-backoff.js';
|
||||
export * from './ext-error.js';
|
||||
export * from './graphql-client.js';
|
||||
export * from './inverse-bloom-filter.js';
|
||||
export * from './memoizer.js';
|
||||
export * from './misc.js';
|
||||
@@ -8,3 +9,4 @@ export * from './sse-session.js';
|
||||
export * from './sse-session-msgpackr.js';
|
||||
export * from './type-utils.js';
|
||||
export * from './uuid.js';
|
||||
export * from './ws-client.js';
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
export type MemoizerItem<T> = {
|
||||
value: T;
|
||||
timestamp: number;
|
||||
}
|
||||
};
|
||||
|
||||
export class Memoizer {
|
||||
private cache = new Map<string, MemoizerItem<any>>();
|
||||
@@ -32,7 +32,7 @@ export class Memoizer {
|
||||
if (ttl) {
|
||||
this.removeIfExpired(key, ttl);
|
||||
}
|
||||
|
||||
|
||||
if (!this.cache.has(key)) {
|
||||
this.set(key, compute());
|
||||
}
|
||||
|
||||
@@ -0,0 +1,250 @@
|
||||
import { EventEmitter } from './event-emitter.js';
|
||||
import { ExponentialBackoff } from './exponential-backoff.js';
|
||||
|
||||
/**
|
||||
* List of events emitted by the WebSocket.
|
||||
* @event
|
||||
* @ignore
|
||||
*/
|
||||
export type WebSocketEvents = {
|
||||
/**
|
||||
* Emitted when data has been received.
|
||||
* @eventProperty
|
||||
*/
|
||||
message: MessageEvent;
|
||||
|
||||
/**
|
||||
* Emitted when socket connects.
|
||||
* @eventProperty
|
||||
*/
|
||||
isOpen: boolean;
|
||||
|
||||
/**
|
||||
* Emitted when the socket has failed in some way.
|
||||
* @eventProperty
|
||||
*/
|
||||
error: Error;
|
||||
};
|
||||
|
||||
/**
|
||||
* Defines the accepted options by the WebSocket Client.
|
||||
*/
|
||||
type Options = {
|
||||
/**
|
||||
* Specifies the server port upon the WebSocket Client is going to be connected.
|
||||
*/
|
||||
port?: number;
|
||||
|
||||
/**
|
||||
* Specifies the subprotocols(messaging structures) necessary to comunicate with the server side.
|
||||
*/
|
||||
protocols?: Array<string> | string;
|
||||
|
||||
/**
|
||||
* Defines the amount of reconnection attempts before throwing failure.
|
||||
*/
|
||||
reconnectionAttempts?: number;
|
||||
};
|
||||
|
||||
/**
|
||||
* A custom Websocket Client implementation.
|
||||
*/
|
||||
export class WebsocketClient extends EventEmitter<WebSocketEvents> {
|
||||
// The websocket connection.
|
||||
private socket: WebSocket | undefined;
|
||||
|
||||
// The exponential backoff handler to re-attempt in case of connection failures.
|
||||
private exponentialBackoff: ExponentialBackoff;
|
||||
|
||||
// Track if service is started or not by the user.
|
||||
private isServiceStarted: boolean;
|
||||
|
||||
// Track if there is an open and working connection to the server.
|
||||
public isConnected: boolean;
|
||||
|
||||
constructor(
|
||||
public readonly host: string,
|
||||
public readonly options?: Options,
|
||||
) {
|
||||
super();
|
||||
|
||||
this.exponentialBackoff = ExponentialBackoff.from({
|
||||
// NOTE: Here the max delay between attempts is not an arbitray number,
|
||||
// is an average waiting time that will work for even commonly appalling internet connections.
|
||||
maxDelay: 15000,
|
||||
// NOTE: We dont want a lot of forced reconnection attemps by default unless is the desired behavior the user wants,
|
||||
// 3 attempts by default is an small enough window time frame to not cause any long waiting time.
|
||||
maxAttempts: this.options?.reconnectionAttempts || 3,
|
||||
growthRate: 1.3,
|
||||
});
|
||||
|
||||
this.isConnected = false;
|
||||
|
||||
this.isServiceStarted = false;
|
||||
}
|
||||
|
||||
static async from(host: string, options?: Options) {
|
||||
const webSocket = new this(host, options);
|
||||
|
||||
webSocket.start();
|
||||
|
||||
return webSocket;
|
||||
}
|
||||
|
||||
/**
|
||||
* Open the WebSocket connection.
|
||||
*/
|
||||
public start() {
|
||||
// If socket is open, not necessary to continue.
|
||||
if (this.isConnected) return;
|
||||
|
||||
// Defines and open the socket connection with the subprotocol if any.
|
||||
this.socket = new WebSocket(this.getUrl(), this.options?.protocols);
|
||||
|
||||
// Subscribe to events emited by the socket.
|
||||
this.socket.onopen = this.onOpen.bind(this);
|
||||
this.socket.onmessage = (event: MessageEvent) =>
|
||||
this.emit('message', event);
|
||||
this.socket.onclose = this.onClose.bind(this);
|
||||
this.socket.onerror = this.onError.bind(this);
|
||||
|
||||
this.isServiceStarted = true;
|
||||
|
||||
// Run the polling function that tracks the 'isConnected' socket status.
|
||||
this.updateWebsocketClientIsConnectedStatus();
|
||||
}
|
||||
|
||||
public stop(): void {
|
||||
this.isServiceStarted = false;
|
||||
|
||||
// If there is no socket defined is not necessary to continue.
|
||||
if (!this.socket) return;
|
||||
|
||||
this.socket.close();
|
||||
|
||||
this.socket = undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* Transmits data using the WebSocket connection.
|
||||
*
|
||||
* @throws If was not able to connect with the server to send the message.
|
||||
*/
|
||||
public async send(
|
||||
data: string | ArrayBuffer | Blob | ArrayBufferView,
|
||||
): Promise<void> {
|
||||
// If socket is not connected try to reconnet it before send the message.
|
||||
if (!this.isConnected) await this.reconnect();
|
||||
|
||||
this.socket!.send(data);
|
||||
}
|
||||
|
||||
private onOpen(): void {
|
||||
this.isConnected = true;
|
||||
|
||||
this.emit('isOpen', true);
|
||||
}
|
||||
|
||||
private async onClose(): Promise<void> {
|
||||
this.isConnected = false;
|
||||
|
||||
this.emit('isOpen', false);
|
||||
|
||||
// If connection closes due to some error, but the service is started, try to reconnect.
|
||||
if (this.isServiceStarted) await this.reconnect();
|
||||
}
|
||||
|
||||
private onError(event: Event): void {
|
||||
const error = new Error('WebSocket error');
|
||||
if ('error' in event) {
|
||||
Object.assign(error, event);
|
||||
}
|
||||
|
||||
this.emit('error', error);
|
||||
}
|
||||
|
||||
/**
|
||||
* Try to reconnects with the server using an exponential backoff approach.
|
||||
*/
|
||||
private async reconnect(): Promise<void> {
|
||||
await this.exponentialBackoff.run(async () => {
|
||||
// If socket is 'undefined' or is 'CLOSED' try to start.
|
||||
if (!this.socket || this.socket.readyState === WebSocket.CLOSED) {
|
||||
this.start();
|
||||
|
||||
throw new Error(
|
||||
`WebSocket was not initialized, trying to reconnect to ${this.getUrl()}`,
|
||||
);
|
||||
}
|
||||
|
||||
// If socket is already open just return;
|
||||
if (this.socket!.readyState === WebSocket.OPEN) return;
|
||||
|
||||
if (this.socket!.readyState === WebSocket.CONNECTING) {
|
||||
try {
|
||||
// Socket should jump from 'CONNECTING' to 'OPEN' state quite quickly.
|
||||
// Here we just waits that happens for 5 seconds wich is good timing even for very bad connections.
|
||||
await this.waitFor('isOpen', (isOpen) => isOpen, 5000);
|
||||
} catch (error: unknown) {
|
||||
console.debug(error);
|
||||
|
||||
this.socket!.close();
|
||||
|
||||
throw new Error(
|
||||
`Error trying to establish WebSocket connection to ${this.getUrl()}, stuck at 'CONNECTING'`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
if (this.socket!.readyState === WebSocket.CLOSING) {
|
||||
try {
|
||||
// Socket should jump from 'CLOSING' to 'CLOSED' state quite quickly.
|
||||
// Here we just waits that happens for 2 seconds or force socket destruction.
|
||||
await this.waitFor('isOpen', (isOpen) => !isOpen, 2000);
|
||||
} catch (error: unknown) {
|
||||
console.debug(error);
|
||||
|
||||
// Detaching socket event emitter functions to avoid memory leaks.
|
||||
this.socket.onopen = null;
|
||||
this.socket.onclose = null;
|
||||
this.socket.onmessage = null;
|
||||
this.socket.onerror = null;
|
||||
this.socket = undefined;
|
||||
|
||||
throw new Error(
|
||||
`Error trying to close WebSocket connection from ${this.getUrl()}, stuck at 'CLOSING'`,
|
||||
);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Update the WebSocket Client 'isConnected' status every 1 second as long as the service has been started.
|
||||
*/
|
||||
private async updateWebsocketClientIsConnectedStatus(): Promise<void> {
|
||||
while (this.isServiceStarted) {
|
||||
await new Promise((resolve) => {
|
||||
this.isConnected = this.socket?.readyState === WebSocket.OPEN;
|
||||
|
||||
// Resolve the promise after 1 second.
|
||||
setTimeout(resolve, 1000);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Util function to transfrom 'https:' urls to 'wss:' urls
|
||||
*/
|
||||
private getUrl(): string {
|
||||
let url = '';
|
||||
|
||||
if (!this.host.includes('wss://') && !this.host.includes('ws://'))
|
||||
// Use encrypted connection by default
|
||||
url = 'wss://';
|
||||
|
||||
url += this.options?.port ? `${this.host}:${this.options.port}` : this.host;
|
||||
|
||||
return url;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user