(
type: string | Eip1193EventName,
listener:
| Web3Eip1193ProviderEventCallback
| Web3ProviderMessageEventCallback
| Web3ProviderEventCallback,
): void {
this._eventEmitter.on(type, listener);
}
/**
* Registers a listener for the specified event type that will be invoked at most once.
* @param type - The event type to listen for
* @param listener - The callback to be invoked when the event is emitted
*/
public once(
type: 'disconnect',
listener: Web3Eip1193ProviderEventCallback,
): void;
public once(
type: 'connect',
listener: Web3Eip1193ProviderEventCallback,
): void;
public once(type: 'chainChanged', listener: Web3Eip1193ProviderEventCallback): void;
public once(
type: 'accountsChanged',
listener: Web3Eip1193ProviderEventCallback,
): void;
public once(
type: 'message',
listener:
| Web3Eip1193ProviderEventCallback
| Web3ProviderMessageEventCallback,
): void;
public once(
type: string,
listener: Web3Eip1193ProviderEventCallback | Web3ProviderEventCallback,
): void;
public once(
type: string | Eip1193EventName,
listener:
| Web3Eip1193ProviderEventCallback
| Web3ProviderMessageEventCallback
| Web3ProviderEventCallback,
): void {
this._eventEmitter.once(type, listener);
}
/**
* Removes a listener for the specified event type.
* @param type - The event type to remove the listener for
* @param listener - The callback to be executed
*/
public removeListener(
type: 'disconnect',
listener: Web3Eip1193ProviderEventCallback,
): void;
public removeListener(
type: 'connect',
listener: Web3Eip1193ProviderEventCallback,
): void;
public removeListener(
type: 'chainChanged',
listener: Web3Eip1193ProviderEventCallback,
): void;
public removeListener(
type: 'accountsChanged',
listener: Web3Eip1193ProviderEventCallback,
): void;
public removeListener(
type: 'message',
listener:
| Web3Eip1193ProviderEventCallback
| Web3ProviderMessageEventCallback,
): void;
public removeListener(
type: string,
listener: Web3Eip1193ProviderEventCallback | Web3ProviderEventCallback,
): void;
public removeListener(
type: string | Eip1193EventName,
listener:
| Web3Eip1193ProviderEventCallback
| Web3ProviderMessageEventCallback
| Web3ProviderEventCallback,
): void {
this._eventEmitter.removeListener(type, listener);
}
protected _onDisconnect(code: number, data?: string) {
this._connectionStatus = 'disconnected';
super._onDisconnect(code, data);
}
/**
* Disconnects the socket
* @param code - The code to be sent to the server
* @param data - The data to be sent to the server
*/
public disconnect(code?: number, data?: string): void {
const disconnectCode = code ?? NORMAL_CLOSE_CODE;
this._removeSocketListeners();
if (this.getStatus() !== 'disconnected') {
this._closeSocketConnection(disconnectCode, data);
}
this._onDisconnect(disconnectCode, data);
}
/**
* Safely disconnects the socket, async and waits for request size to be 0 before disconnecting
* @param forceDisconnect - If true, will clear queue after 5 attempts of waiting for both pending and sent queue to be 0
* @param ms - Determines the ms of setInterval
* @param code - The code to be sent to the server
* @param data - The data to be sent to the server
*/
public async safeDisconnect(code?: number, data?: string, forceDisconnect = false,ms = 1000) {
let retryAttempt = 0;
const checkQueue = async () =>
new Promise(resolve => {
const interval = setInterval(() => {
if (forceDisconnect && retryAttempt === 5) {
this.clearQueues();
}
if (this.getPendingRequestQueueSize() === 0 && this.getSentRequestsQueueSize() === 0) {
clearInterval(interval);
resolve(true);
}
retryAttempt+=1;
}, ms)
})
await checkQueue();
this.disconnect(code, data);
}
/**
* Removes all listeners for the specified event type.
* @param type - The event type to remove the listeners for
*/
public removeAllListeners(type: string): void {
this._eventEmitter.removeAllListeners(type);
}
protected _onError(event: ErrorEvent): void {
// do not emit error while trying to reconnect
if (this.isReconnecting) {
this._reconnect();
} else {
this._eventEmitter.emit('error', event);
}
}
/**
* Resets the socket, removing all listeners and pending requests
*/
public reset(): void {
this._sentRequestsQueue.clear();
this._pendingRequestsQueue.clear();
this._init();
this._removeSocketListeners();
this._addSocketListeners();
}
protected _reconnect(): void {
if (this.isReconnecting) {
return;
}
this.isReconnecting = true;
if (this._sentRequestsQueue.size > 0) {
this._sentRequestsQueue.forEach(
(request: SocketRequestItem, key: JsonRpcId) => {
request.deferredPromise.reject(new PendingRequestsOnReconnectingError());
this._sentRequestsQueue.delete(key);
},
);
}
if (this._reconnectAttempts < this._reconnectOptions.maxAttempts) {
this._reconnectAttempts += 1;
setTimeout(() => {
this._removeSocketListeners();
this.connect();
this.isReconnecting = false;
}, this._reconnectOptions.delay);
} else {
this.isReconnecting = false;
this._clearQueues();
this._removeSocketListeners();
this._eventEmitter.emit(
'error',
new MaxAttemptsReachedOnReconnectingError(this._reconnectOptions.maxAttempts),
);
}
}
/**
* Creates a request object to be sent to the server
*/
public async request<
Method extends Web3APIMethod,
ResultType = Web3APIReturnType,
>(request: Web3APIPayload): Promise> {
if (isNullish(this._socketConnection)) {
throw new Error('Connection is undefined');
}
// if socket disconnected - open connection
if (this.getStatus() === 'disconnected') {
this.connect();
}
const requestId = jsonRpc.isBatchRequest(request)
? (request as unknown as JsonRpcBatchRequest)[0].id
: (request as unknown as JsonRpcRequest).id;
if (!requestId) {
throw new Web3WSProviderError('Request Id not defined');
}
if (this._sentRequestsQueue.has(requestId)) {
throw new RequestAlreadySentError(requestId);
}
const deferredPromise = new Web3DeferredPromise>();
deferredPromise.catch(error => {
this._eventEmitter.emit('error', error);
});
const reqItem: SocketRequestItem> = {
payload: request,
deferredPromise,
};
if (this.getStatus() === 'connecting') {
this._pendingRequestsQueue.set(requestId, reqItem);
return reqItem.deferredPromise;
}
this._sentRequestsQueue.set(requestId, reqItem);
try {
this._sendToSocket(reqItem.payload);
} catch (error) {
this._sentRequestsQueue.delete(requestId);
this._eventEmitter.emit('error', error);
}
return deferredPromise;
}
protected _onConnect() {
this._connectionStatus = 'connected';
this._reconnectAttempts = 0;
super._onConnect();
this._sendPendingRequests();
}
private _sendPendingRequests() {
for (const [id, value] of this._pendingRequestsQueue.entries()) {
this._sendToSocket(value.payload as Web3APIPayload);
this._pendingRequestsQueue.delete(id);
this._sentRequestsQueue.set(id, value);
}
}
protected _onMessage(event: MessageEvent): void {
const responses = this._parseResponses(event);
if (isNullish(responses) || responses.length === 0) {
return;
}
for (const response of responses) {
if (
jsonRpc.isResponseWithNotification(response as JsonRpcNotification) &&
(response as JsonRpcNotification).method.endsWith('_subscription')
) {
this._eventEmitter.emit('message', response);
return;
}
const requestId = jsonRpc.isBatchResponse(response)
? (response as unknown as JsonRpcBatchResponse)[0].id
: (response as unknown as JsonRpcResponseWithResult).id;
const requestItem = this._sentRequestsQueue.get(requestId);
if (!requestItem) {
return;
}
if (
jsonRpc.isBatchResponse(response) ||
jsonRpc.isResponseWithResult(response) ||
jsonRpc.isResponseWithError(response)
) {
this._eventEmitter.emit('message', response);
requestItem.deferredPromise.resolve(response);
}
this._sentRequestsQueue.delete(requestId);
}
}
public clearQueues(event?: ConnectionEvent) {
this._clearQueues(event);
}
protected _clearQueues(event?: ConnectionEvent) {
if (this._pendingRequestsQueue.size > 0) {
this._pendingRequestsQueue.forEach(
(request: SocketRequestItem, key: JsonRpcId) => {
request.deferredPromise.reject(new ConnectionNotOpenError(event));
this._pendingRequestsQueue.delete(key);
},
);
}
if (this._sentRequestsQueue.size > 0) {
this._sentRequestsQueue.forEach(
(request: SocketRequestItem, key: JsonRpcId) => {
request.deferredPromise.reject(new ConnectionNotOpenError(event));
this._sentRequestsQueue.delete(key);
},
);
}
this._removeSocketListeners();
}
}