quantus/api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove send queue, always require connection (#2881)

Browse Source 
* Remove send queue, always require connection

* Adjust tests for non-queued
This commit is contained in:
Jaco Greeff
2020-11-28 12:57:24 +01:00
committed by GitHub
parent 25e29d9565
commit 09aa046ab4
5 changed files with 95 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/rpc-provider/src/ws/index.ts
+28 -35
View File
@@ -71,7 +71,7 @@ export class WsProvider implements ProviderInterface {
readonly #handlers: Record<string, WsStateAwaiting> = {};
readonly #queued: Record<string, string> = {};
readonly #isReadyPromise: Promise<WsProvider>;
readonly #waitingForId: Record<string, JsonRpcResponse> = {};
@@ -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<WsProvider> {
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<any> {
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<string>): 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<void> => {
Promise.all(Object.keys(subscriptions).map(async (id): Promise<void> => {
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);
}
}
packages/rpc-provider/src/ws/onOpen.spec.ts
-43
View File
@@ -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<void> => {
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');
});
});
});
packages/rpc-provider/src/ws/send.spec.ts
+33 -25
View File
@@ -17,10 +17,10 @@ function createMock (requests: any[]): void {
mock = mockWs(requests);
}
function createWs (autoConnect = 1000): WsProvider {
function createWs (autoConnect = 1000): Promise<WsProvider> {
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<void> => {
@@ -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<any> => {
@@ -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<void> => {
@@ -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);
})
);
});
});
packages/rpc-provider/src/ws/subscribe.spec.ts
+11 -9
View File
@@ -17,10 +17,10 @@ function createMock (requests: any[]): void {
mock = mockWs(requests);
}
function createWs (autoConnect = 1000): WsProvider {
function createWs (autoConnect = 1000): Promise<WsProvider> {
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);
})
);
});
});
packages/rpc-provider/src/ws/unsubscribe.spec.ts
+23 -23
View File
@@ -17,10 +17,10 @@ function createMock (requests: any): void {
mock = mockWs(requests);
}
function createWs (autoConnect = 1000): WsProvider {
function createWs (autoConnect = 1000): Promise<WsProvider> {
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<boolean> => {
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<boolean> => {
return ws.unsubscribe('test', 'subscribe_test', id);
})
);
});
it('fails when sub not found', (): Promise<void> => {
@@ -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<boolean> => {
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<boolean> => {
return ws.unsubscribe('test', 'subscribe_test', 111);
})
.then((result): void => {
expect(result).toBe(false);
})
);
});
});