diff --git a/packages/rpc-provider/src/ws/index.ts b/packages/rpc-provider/src/ws/index.ts index e9adcb0dab..cd72917bf1 100644 --- a/packages/rpc-provider/src/ws/index.ts +++ b/packages/rpc-provider/src/ws/index.ts @@ -71,7 +71,7 @@ export class WsProvider implements ProviderInterface { readonly #handlers: Record = {}; - readonly #queued: Record = {}; + readonly #isReadyPromise: Promise; readonly #waitingForId: Record = {}; @@ -113,6 +113,12 @@ export class WsProvider implements ProviderInterface { // does not throw }); } + + this.#isReadyPromise = new Promise((resolve): void => { + this.#eventemitter.once('connected', (): void => { + resolve(this); + }); + }); } /** @@ -130,6 +136,13 @@ export class WsProvider implements ProviderInterface { return this.#isConnected; } + /** + * @description Promise that resolves the first time we are connected and loaded + */ + public get isReady (): Promise { + return this.#isReadyPromise; + } + /** * @description Returns a clone of the object */ @@ -225,12 +238,14 @@ export class WsProvider implements ProviderInterface { /** * @summary Send JSON data using WebSockets to configured HTTP Endpoint or queue. * @param method The RPC methods to execute - * @param params Encoded paramaters as appliucable for the method + * @param params Encoded parameters as applicable for the method * @param subscription Subscription details (internally used) */ public send (method: string, params: any[], subscription?: SubscriptionHandler): Promise { return new Promise((resolve, reject): void => { try { + assert(this.isConnected && !isNull(this.#websocket), 'WebSocket is not connected'); + const json = this.#coder.encodeJson(method, params); const id = this.#coder.getId(); @@ -240,7 +255,7 @@ export class WsProvider implements ProviderInterface { : resolve(result); }; - l.debug((): string[] => ['calling', method, json]); + l.debug(() => ['calling', method, json]); this.#handlers[id] = { callback, @@ -249,11 +264,7 @@ export class WsProvider implements ProviderInterface { subscription }; - if (this.isConnected && !isNull(this.#websocket)) { - this.#websocket.send(json); - } else { - this.#queued[id] = json; - } + this.#websocket.send(json); } catch (error) { reject(error); } @@ -300,7 +311,7 @@ export class WsProvider implements ProviderInterface { // a slight complication in solving - since we cannot rely on the send id, but rather // need to find the actual subscription id to map it if (isUndefined(this.#subscriptions[subscription])) { - l.debug((): string => `Unable to find active subscription=${subscription}`); + l.debug(() => `Unable to find active subscription=${subscription}`); return false; } @@ -334,15 +345,14 @@ export class WsProvider implements ProviderInterface { } #onSocketError = (error: Event): void => { - l.debug((): any => ['socket error', error]); + l.debug(() => ['socket error', error]); this.#emit('error', error); } - #onSocketMessage = (message: MessageEvent): void => { - // eslint-disable-next-line @typescript-eslint/no-unsafe-return + #onSocketMessage = (message: MessageEvent): void => { l.debug(() => ['received', message.data]); - const response = JSON.parse(message.data as string) as JsonRpcResponse; + const response = JSON.parse(message.data) as JsonRpcResponse; return isUndefined(response.method) ? this.#onSocketMessageResult(response) @@ -353,7 +363,7 @@ export class WsProvider implements ProviderInterface { const handler = this.#handlers[response.id]; if (!handler) { - l.debug((): string => `Unable to find handler for id=${response.id}`); + l.debug(() => `Unable to find handler for id=${response.id}`); return; } @@ -396,7 +406,7 @@ export class WsProvider implements ProviderInterface { // store the JSON, we could have out-of-order subid coming in this.#waitingForId[subId] = response; - l.debug((): string => `Unable to find handler for subscription=${subId}`); + l.debug(() => `Unable to find handler for subscription=${subId}`); return; } @@ -416,12 +426,11 @@ export class WsProvider implements ProviderInterface { #onSocketOpen = (): boolean => { assert(!isNull(this.#websocket), 'WebSocket cannot be null in onOpen'); - l.debug((): any[] => ['connected to', this.#endpoints[this.#endpointIndex]]); + l.debug(() => ['connected to', this.#endpoints[this.#endpointIndex]]); this.#isConnected = true; this.#emit('connected'); - this.#sendQueue(); this.#resubscribe(); return true; @@ -432,8 +441,7 @@ export class WsProvider implements ProviderInterface { this.#subscriptions = {}; - // eslint-disable-next-line @typescript-eslint/no-misused-promises - Object.keys(subscriptions).forEach(async (id): Promise => { + Promise.all(Object.keys(subscriptions).map(async (id): Promise => { const { callback, method, params, type } = subscriptions[id]; // only re-create subscriptions which are not in author (only area where @@ -448,21 +456,6 @@ export class WsProvider implements ProviderInterface { } catch (error) { l.error(error); } - }); - } - - #sendQueue = (): void => { - Object.keys(this.#queued).forEach((id): void => { - try { - // we have done the websocket check in onSocketOpen, if an issue, will catch it - (this.#websocket as WebSocket).send( - this.#queued[id] - ); - - delete this.#queued[id]; - } catch (error) { - l.error(error); - } - }); + })).catch(l.error); } } diff --git a/packages/rpc-provider/src/ws/onOpen.spec.ts b/packages/rpc-provider/src/ws/onOpen.spec.ts deleted file mode 100644 index f8e78cc716..0000000000 --- a/packages/rpc-provider/src/ws/onOpen.spec.ts +++ /dev/null @@ -1,43 +0,0 @@ -// Copyright 2017-2020 @polkadot/rpc-provider authors & contributors -// SPDX-License-Identifier: Apache-2.0 - -import { Mock } from './../mock/types'; -import { mockWs, TEST_WS_URL } from '../../test/mockWs'; - -import { WsProvider } from './'; - -let ws: WsProvider; -let mock: Mock; - -function createWs (requests: any[], autoConnect: number | undefined = 1000): WsProvider { - mock = mockWs(requests); - ws = new WsProvider(TEST_WS_URL, autoConnect); - - return ws; -} - -describe('onOpen', (): void => { - afterEach((): void => { - if (mock) { - mock.done(); - } - }); - - it('sends messages when connected', (): Promise => { - const ws = createWs([{ - id: 1, - method: 'test_queue', - reply: { - result: 'ok' - } - }], 0); - const sendPromise = ws.send('test_queue', []); - - // eslint-disable-next-line @typescript-eslint/no-floating-promises - ws.connect(); - - return sendPromise.then((result): void => { - expect(result).toEqual('ok'); - }); - }); -}); diff --git a/packages/rpc-provider/src/ws/send.spec.ts b/packages/rpc-provider/src/ws/send.spec.ts index 2eb8331d69..694d44b428 100644 --- a/packages/rpc-provider/src/ws/send.spec.ts +++ b/packages/rpc-provider/src/ws/send.spec.ts @@ -17,10 +17,10 @@ function createMock (requests: any[]): void { mock = mockWs(requests); } -function createWs (autoConnect = 1000): WsProvider { +function createWs (autoConnect = 1000): Promise { provider = new WsProvider(TEST_WS_URL, autoConnect); - return provider; + return provider.isReady; } describe('send', (): void => { @@ -47,11 +47,13 @@ describe('send', (): void => { } }]); - return createWs() - .send('test_encoding', [{ error: 'send error' }]) - .catch((error): void => { - expect((error as Error).message).toEqual('send error'); - }); + return createWs().then((ws) => + ws + .send('test_encoding', [{ error: 'send error' }]) + .catch((error): void => { + expect((error as Error).message).toEqual('send error'); + }) + ); }); it('passes the body through correctly', (): Promise => { @@ -63,14 +65,16 @@ describe('send', (): void => { } }]); - return createWs() - .send('test_body', ['param']) - .then((): void => { - expect( - // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access - (mock.body as any).test_body - ).toEqual('{"id":1,"jsonrpc":"2.0","method":"test_body","params":["param"]}'); - }); + return createWs().then((ws) => + ws + .send('test_body', ['param']) + .then((): void => { + expect( + // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access + (mock.body as any).test_body + ).toEqual('{"id":1,"jsonrpc":"2.0","method":"test_body","params":["param"]}'); + }) + ); }); it('throws error when !response.ok', (): Promise => { @@ -82,11 +86,13 @@ describe('send', (): void => { id: 1 }]); - return createWs() - .send('test_error', []) - .catch((error): void => { - expect((error as Error).message).toMatch(/666: error/); - }); + return createWs().then((ws) => + ws + .send('test_error', []) + .catch((error): void => { + expect((error as Error).message).toMatch(/666: error/); + }) + ); }); it('adds subscriptions', (): Promise => { @@ -98,10 +104,12 @@ describe('send', (): void => { } }]); - return createWs() - .send('test_sub', []) - .then((id): void => { - expect(id).toEqual(1); - }); + return createWs().then((ws) => + ws + .send('test_sub', []) + .then((id): void => { + expect(id).toEqual(1); + }) + ); }); }); diff --git a/packages/rpc-provider/src/ws/subscribe.spec.ts b/packages/rpc-provider/src/ws/subscribe.spec.ts index fcb416dd05..0c6ea28b1f 100644 --- a/packages/rpc-provider/src/ws/subscribe.spec.ts +++ b/packages/rpc-provider/src/ws/subscribe.spec.ts @@ -17,10 +17,10 @@ function createMock (requests: any[]): void { mock = mockWs(requests); } -function createWs (autoConnect = 1000): WsProvider { +function createWs (autoConnect = 1000): Promise { ws = new WsProvider(TEST_WS_URL, autoConnect); - return ws; + return ws.isReady; } describe('subscribe', (): void => { @@ -47,12 +47,14 @@ describe('subscribe', (): void => { } }]); - return createWs() - .subscribe('type', 'test_sub', [], (cb): void => { - expect(cb).toEqual(expect.anything()); - }) - .then((id): void => { - expect(id).toEqual(1); - }); + return createWs().then((ws) => + ws + .subscribe('type', 'test_sub', [], (cb): void => { + expect(cb).toEqual(expect.anything()); + }) + .then((id): void => { + expect(id).toEqual(1); + }) + ); }); }); diff --git a/packages/rpc-provider/src/ws/unsubscribe.spec.ts b/packages/rpc-provider/src/ws/unsubscribe.spec.ts index 36d9d6374a..bbae0cbba0 100644 --- a/packages/rpc-provider/src/ws/unsubscribe.spec.ts +++ b/packages/rpc-provider/src/ws/unsubscribe.spec.ts @@ -17,10 +17,10 @@ function createMock (requests: any): void { mock = mockWs(requests); } -function createWs (autoConnect = 1000): WsProvider { +function createWs (autoConnect = 1000): Promise { ws = new WsProvider(TEST_WS_URL, autoConnect); - return ws; + return ws.isReady; } describe('subscribe', (): void => { @@ -56,15 +56,15 @@ describe('subscribe', (): void => { } ]); - const ws = createWs(); - - return ws - .subscribe('test', 'subscribe_test', [], (cb): void => { - expect(cb).toEqual(expect.anything()); - }) - .then((id): Promise => { - return ws.unsubscribe('test', 'subscribe_test', id); - }); + return createWs().then((ws) => + ws + .subscribe('test', 'subscribe_test', [], (cb): void => { + expect(cb).toEqual(expect.anything()); + }) + .then((id): Promise => { + return ws.unsubscribe('test', 'subscribe_test', id); + }) + ); }); it('fails when sub not found', (): Promise => { @@ -76,17 +76,17 @@ describe('subscribe', (): void => { } }]); - const ws = createWs(); - - return ws - .subscribe('test', 'subscribe_test', [], (cb): void => { - expect(cb).toEqual(expect.anything()); - }) - .then((): Promise => { - return ws.unsubscribe('test', 'subscribe_test', 111); - }) - .then((result): void => { - expect(result).toBe(false); - }); + return createWs().then((ws) => + ws + .subscribe('test', 'subscribe_test', [], (cb): void => { + expect(cb).toEqual(expect.anything()); + }) + .then((): Promise => { + return ws.unsubscribe('test', 'subscribe_test', 111); + }) + .then((result): void => { + expect(result).toBe(false); + }) + ); }); });