crowetic/protocol
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Consolidate back to one channel and expose only the factory

Browse Source
This commit is contained in:
Brandon Millman
2018-05-25 16:08:15 -07:00
parent 47debf0134
commit a4b6112a31
10 changed files with 65 additions and 284 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
packages/connect/src/index.ts
View File

@@ -1,12 +1,9 @@
export { HttpClient } from './http_client';
export { BrowserWebSocketOrderbookChannel } from './browser_ws_orderbook_channel';
export { NodeWebSocketOrderbookChannel } from './node_ws_orderbook_channel';
export { orderbookChannelFactory } from './orderbook_channel_factory';
export {
Client,
FeesRequest,
FeesResponse,
NodeWebSocketOrderbookChannelConfig,
OrderbookChannel,
OrderbookChannelHandler,
OrderbookChannelSubscriptionOpts,

158
packages/connect/src/node_ws_orderbook_channel.ts
View File

@@ -1,158 +0,0 @@
import * as _ from 'lodash';
import * as WebSocket from 'websocket';
import { schemas as clientSchemas } from './schemas/schemas';
import {
NodeWebSocketOrderbookChannelConfig,
OrderbookChannel,
OrderbookChannelHandler,
OrderbookChannelMessageTypes,
OrderbookChannelSubscriptionOpts,
WebsocketClientEventType,
WebsocketConnectionEventType,
} from './types';
import { assert } from './utils/assert';
import { orderbookChannelMessageParser } from './utils/orderbook_channel_message_parser';
const DEFAULT_HEARTBEAT_INTERVAL_MS = 15000;
const MINIMUM_HEARTBEAT_INTERVAL_MS = 10;
/**
* This class includes all the functionality related to interacting with a websocket endpoint
* that implements the standard relayer API v0 in a node environment
*/
export class NodeWebSocketOrderbookChannel implements OrderbookChannel {
private _apiEndpointUrl: string;
private _client: WebSocket.client;
private _connectionIfExists?: WebSocket.connection;
private _heartbeatTimerIfExists?: NodeJS.Timer;
private _subscriptionCounter = 0;
private _heartbeatIntervalMs: number;
/**
* Instantiates a new NodeWebSocketOrderbookChannelConfig instance
* @param url The relayer API base WS url you would like to interact with
* @param config The configuration object. Look up the type for the description.
* @return An instance of NodeWebSocketOrderbookChannelConfig
*/
constructor(url: string, config?: NodeWebSocketOrderbookChannelConfig) {
assert.isUri('url', url);
if (!_.isUndefined(config)) {
assert.doesConformToSchema('config', config, clientSchemas.nodeWebSocketOrderbookChannelConfigSchema);
}
this._apiEndpointUrl = url;
this._heartbeatIntervalMs =
_.isUndefined(config) || _.isUndefined(config.heartbeatIntervalMs)
? DEFAULT_HEARTBEAT_INTERVAL_MS
: config.heartbeatIntervalMs;
this._client = new WebSocket.client();
}
/**
* Subscribe to orderbook snapshots and updates from the websocket
* @param subscriptionOpts An OrderbookChannelSubscriptionOpts instance describing which
* token pair to subscribe to
* @param handler An OrderbookChannelHandler instance that responds to various
* channel updates
*/
public subscribe(subscriptionOpts: OrderbookChannelSubscriptionOpts, handler: OrderbookChannelHandler): void {
assert.isOrderbookChannelSubscriptionOpts('subscriptionOpts', subscriptionOpts);
assert.isOrderbookChannelHandler('handler', handler);
this._subscriptionCounter += 1;
const subscribeMessage = {
type: 'subscribe',
channel: 'orderbook',
requestId: this._subscriptionCounter,
payload: subscriptionOpts,
};
this._getConnection((error, connection) => {
if (!_.isUndefined(error)) {
handler.onError(this, subscriptionOpts, error);
} else if (!_.isUndefined(connection) && connection.connected) {
connection.on(WebsocketConnectionEventType.Error, wsError => {
handler.onError(this, subscriptionOpts, wsError);
});
connection.on(WebsocketConnectionEventType.Close, (_code: number, _desc: string) => {
handler.onClose(this, subscriptionOpts);
});
connection.on(WebsocketConnectionEventType.Message, message => {
this._handleWebSocketMessage(subscribeMessage.requestId, subscriptionOpts, message, handler);
});
connection.sendUTF(JSON.stringify(subscribeMessage));
}
});
}
/**
* Close the websocket and stop receiving updates
*/
public close(): void {
if (!_.isUndefined(this._connectionIfExists)) {
this._connectionIfExists.close();
}
if (!_.isUndefined(this._heartbeatTimerIfExists)) {
clearInterval(this._heartbeatTimerIfExists);
}
}
private _getConnection(callback: (error?: Error, connection?: WebSocket.connection) => void): void {
if (!_.isUndefined(this._connectionIfExists) && this._connectionIfExists.connected) {
callback(undefined, this._connectionIfExists);
} else {
this._client.on(WebsocketClientEventType.Connect, connection => {
this._connectionIfExists = connection;
if (this._heartbeatIntervalMs >= MINIMUM_HEARTBEAT_INTERVAL_MS) {
this._heartbeatTimerIfExists = setInterval(() => {
connection.ping('');
}, this._heartbeatIntervalMs);
} else {
callback(
new Error(
`Heartbeat interval is ${
this._heartbeatIntervalMs
}ms which is less than the required minimum of ${MINIMUM_HEARTBEAT_INTERVAL_MS}ms`,
),
undefined,
);
}
callback(undefined, this._connectionIfExists);
});
this._client.on(WebsocketClientEventType.ConnectFailed, error => {
callback(error, undefined);
});
this._client.connect(this._apiEndpointUrl);
}
}
private _handleWebSocketMessage(
requestId: number,
subscriptionOpts: OrderbookChannelSubscriptionOpts,
message: WebSocket.IMessage,
handler: OrderbookChannelHandler,
): void {
if (!_.isUndefined(message.utf8Data)) {
try {
const utf8Data = message.utf8Data;
const parserResult = orderbookChannelMessageParser.parse(utf8Data);
if (parserResult.requestId === requestId) {
switch (parserResult.type) {
case OrderbookChannelMessageTypes.Snapshot: {
handler.onSnapshot(this, subscriptionOpts, parserResult.payload);
break;
}
case OrderbookChannelMessageTypes.Update: {
handler.onUpdate(this, subscriptionOpts, parserResult.payload);
break;
}
default: {
handler.onError(
this,
subscriptionOpts,
new Error(`Message has missing a type parameter: ${utf8Data}`),
);
}
}
}
} catch (error) {
handler.onError(this, subscriptionOpts, error);
}
} else {
handler.onError(this, subscriptionOpts, new Error(`Message does not contain utf8Data`));
}
}
}

35
packages/connect/src/orderbook_channel_factory.ts
View File

@@ -1,16 +1,21 @@
// import * as WebSocket from 'websocket';
import * as WebSocket from 'websocket';
import { BrowserWebSocketOrderbookChannel } from './browser_ws_orderbook_channel';
import { NodeWebSocketOrderbookChannel } from './node_ws_orderbook_channel';
import { OrderbookChannel, WebsocketClientEventType } from './types';
import { assert } from './utils/assert';
import { WebSocketOrderbookChannel } from './ws_orderbook_channel';
export const orderbookChannelFactory = {
async createBrowserOrderbookChannelAsync(url: string): Promise<BrowserWebSocketOrderbookChannel> {
return new Promise<BrowserWebSocketOrderbookChannel>((resolve, reject) => {
const client = new WebSocket(url);
console.log(client);
/**
* Instantiates a new WebSocketOrderbookChannel instance
* @param url The relayer API base WS url you would like to interact with
* @return An OrderbookChannel Promise
*/
async createWebSocketOrderbookChannelAsync(url: string): Promise<OrderbookChannel> {
assert.isUri('url', url);
return new Promise<OrderbookChannel>((resolve, reject) => {
const client = new WebSocket.w3cwebsocket(url);
client.onopen = () => {
const orderbookChannel = new BrowserWebSocketOrderbookChannel(client);
console.log(orderbookChannel);
const orderbookChannel = new WebSocketOrderbookChannel(client);
resolve(orderbookChannel);
};
client.onerror = err => {
@@ -18,16 +23,4 @@ export const orderbookChannelFactory = {
};
});
},
// async createNodeOrderbookChannelAsync(url: string): Promise<NodeWebSocketOrderbookChannel> {
// return new Promise<BrowserWebSocketOrderbookChannel>((resolve, reject) => {
// const client = new WebSocket.w3cwebsocket(url);
// client.onopen = () => {
// const orderbookChannel = new BrowserWebSocketOrderbookChannel(client);
// resolve(orderbookChannel);
// };
// client.onerror = err => {
// reject(err);
// };
// });
// },
};

10
packages/connect/src/schemas/node_websocket_orderbook_channel_config_schema.ts
View File

@@ -1,10 +0,0 @@
export const nodeWebSocketOrderbookChannelConfigSchema = {
id: '/NodeWebSocketOrderbookChannelConfig',
type: 'object',
properties: {
heartbeatIntervalMs: {
type: 'number',
minimum: 10,
},
},
};

2
packages/connect/src/schemas/schemas.ts
View File

@@ -1,5 +1,4 @@
import { feesRequestSchema } from './fees_request_schema';
import { nodeWebSocketOrderbookChannelConfigSchema } from './node_websocket_orderbook_channel_config_schema';
import { orderBookRequestSchema } from './orderbook_request_schema';
import { ordersRequestOptsSchema } from './orders_request_opts_schema';
import { pagedRequestOptsSchema } from './paged_request_opts_schema';
@@ -7,7 +6,6 @@ import { tokenPairsRequestOptsSchema } from './token_pairs_request_opts_schema';
export const schemas = {
feesRequestSchema,
nodeWebSocketOrderbookChannelConfigSchema,
orderBookRequestSchema,
ordersRequestOptsSchema,
pagedRequestOptsSchema,

7
packages/connect/src/types.ts
View File

@@ -15,13 +15,6 @@ export interface OrderbookChannel {
close: () => void;
}
/**
* heartbeatInterval: Interval in milliseconds that the orderbook channel should ping the underlying websocket. Default: 15000
*/
export interface NodeWebSocketOrderbookChannelConfig {
heartbeatIntervalMs?: number;
}
/**
* baseTokenAddress: The address of token designated as the baseToken in the currency pair calculation of price
* quoteTokenAddress: The address of token designated as the quoteToken in the currency pair calculation of price

37
packages/connect/src/browser_ws_orderbook_channel.ts → packages/connect/src/ws_orderbook_channel.ts
View File

@@ -1,4 +1,5 @@
import * as _ from 'lodash';
import * as WebSocket from 'websocket';
import {
OrderbookChannel,
@@ -18,19 +19,27 @@ interface Subscription {
/**
* This class includes all the functionality related to interacting with a websocket endpoint
* that implements the standard relayer API v0 in a browser environment
* that implements the standard relayer API v0
*/
export class BrowserWebSocketOrderbookChannel implements OrderbookChannel {
private _client: WebSocket;
export class WebSocketOrderbookChannel implements OrderbookChannel {
private _client: WebSocket.w3cwebsocket;
private _subscriptions: Subscription[] = [];
/**
* Instantiates a new WebSocketOrderbookChannel instance
* @param url The relayer API base WS url you would like to interact with
* @return An instance of WebSocketOrderbookChannel
*/
constructor(client: WebSocket) {
// assert.isUri('url', url);
constructor(client: WebSocket.w3cwebsocket) {
this._client = client;
this._client.onerror = err => {
this._alertAllHandlersToError(err);
};
this._client.onclose = () => {
this._alertAllHandlersToClose();
};
this._client.onmessage = message => {
this._handleWebSocketMessage(message);
};
}
/**
* Subscribe to orderbook snapshots and updates from the websocket
@@ -53,17 +62,6 @@ export class BrowserWebSocketOrderbookChannel implements OrderbookChannel {
requestId: this._subscriptions.length - 1,
payload: subscriptionOpts,
};
this._client.onerror = () => {
this._alertAllHandlersToError(new Error('hello'));
};
this._client.onclose = () => {
_.forEach(this._subscriptions, subscription => {
subscription.handler.onClose(this, subscription.subscriptionOpts);
});
};
this._client.onmessage = message => {
this._handleWebSocketMessage(message);
};
this._sendMessage(subscribeMessage);
}
/**
@@ -76,7 +74,7 @@ export class BrowserWebSocketOrderbookChannel implements OrderbookChannel {
* Send a message to the client if it has been instantiated and it is open
*/
private _sendMessage(message: any): void {
if (this._client.readyState === WebSocket.OPEN) {
if (this._client.readyState === WebSocket.w3cwebsocket.OPEN) {
this._client.send(JSON.stringify(message));
}
}
@@ -88,6 +86,11 @@ export class BrowserWebSocketOrderbookChannel implements OrderbookChannel {
subscription.handler.onError(this, subscription.subscriptionOpts, error);
});
}
private _alertAllHandlersToClose(): void {
_.forEach(this._subscriptions, subscription => {
subscription.handler.onClose(this, subscription.subscriptionOpts);
});
}
private _handleWebSocketMessage(message: any): void {
// if we get a message with no data, alert all handlers and return
if (_.isUndefined(message.data)) {

63
packages/connect/test/browser_ws_orderbook_channel_test.ts
View File

@@ -1,63 +0,0 @@
// import * as chai from 'chai';
// import * as dirtyChai from 'dirty-chai';
// import * as _ from 'lodash';
// import 'mocha';
// import * as WebSocket from 'websocket';
// import { BrowserWebSocketOrderbookChannel } from '../src/browser_ws_orderbook_channel';
// chai.config.includeStack = true;
// chai.use(dirtyChai);
// const expect = chai.expect;
// describe('BrowserWebSocketOrderbookChannel', () => {
// const websocketUrl = 'ws://localhost:8080';
// const client = new WebSocket.w3cwebsocket(websocketUrl);
// const orderbookChannel = new BrowserWebSocketOrderbookChannel(client);
// const subscriptionOpts = {
// baseTokenAddress: '0x323b5d4c32345ced77393b3530b1eed0f346429d',
// quoteTokenAddress: '0xef7fff64389b814a946f3e92105513705ca6b990',
// snapshot: true,
// limit: 100,
// };
// const emptyOrderbookChannelHandler = {
// onSnapshot: () => {
// _.noop();
// },
// onUpdate: () => {
// _.noop();
// },
// onError: () => {
// _.noop();
// },
// onClose: () => {
// _.noop();
// },
// };
// describe('#subscribe', () => {
// it('throws when subscriptionOpts does not conform to schema', () => {
// const badSubscribeCall = orderbookChannel.subscribe.bind(
// orderbookChannel,
// {},
// emptyOrderbookChannelHandler,
// );
// expect(badSubscribeCall).throws(
// 'Expected subscriptionOpts to conform to schema /RelayerApiOrderbookChannelSubscribePayload\nEncountered: {}\nValidation errors: instance requires property "baseTokenAddress", instance requires property "quoteTokenAddress"',
// );
// });
// it('throws when handler has the incorrect members', () => {
// const badSubscribeCall = orderbookChannel.subscribe.bind(orderbookChannel, subscriptionOpts, {});
// expect(badSubscribeCall).throws(
// 'Expected handler.onSnapshot to be of type function, encountered: undefined',
// );
// });
// it('does not throw when inputs are of correct types', () => {
// const goodSubscribeCall = orderbookChannel.subscribe.bind(
// orderbookChannel,
// subscriptionOpts,
// emptyOrderbookChannelHandler,
// );
// expect(goodSubscribeCall).to.not.throw();
// });
// });
// });

26
packages/connect/test/orderbook_channel_factory_test.ts Normal file
View File

@@ -0,0 +1,26 @@
import * as chai from 'chai';
import * as dirtyChai from 'dirty-chai';
import * as _ from 'lodash';
import 'mocha';
import * as WebSocket from 'websocket';
import { orderbookChannelFactory } from '../src/orderbook_channel_factory';
chai.config.includeStack = true;
chai.use(dirtyChai);
const expect = chai.expect;
describe('orderbookChannelFactory', () => {
const websocketUrl = 'ws://localhost:8080';
describe('#createWebSocketOrderbookChannelAsync', () => {
it('throws when input is not a url', () => {
const badInput = 54;
const badSubscribeCall = orderbookChannelFactory.createWebSocketOrderbookChannelAsync.bind(
orderbookChannelFactory,
badInput,
);
expect(orderbookChannelFactory.createWebSocketOrderbookChannelAsync(badInput as any)).to.be.rejected();
});
});
});

8
packages/connect/test/node_ws_orderbook_channel_test.ts → packages/connect/test/ws_orderbook_channel_test.ts
View File

@@ -2,16 +2,18 @@ import * as chai from 'chai';
import * as dirtyChai from 'dirty-chai';
import * as _ from 'lodash';
import 'mocha';
import * as WebSocket from 'websocket';
import { NodeWebSocketOrderbookChannel } from '../src/node_ws_orderbook_channel';
import { WebSocketOrderbookChannel } from '../src/ws_orderbook_channel';
chai.config.includeStack = true;
chai.use(dirtyChai);
const expect = chai.expect;
describe('NodeWebSocketOrderbookChannel', () => {
describe('WebSocketOrderbookChannel', () => {
const websocketUrl = 'ws://localhost:8080';
const orderbookChannel = new NodeWebSocketOrderbookChannel(websocketUrl);
const client = new WebSocket.w3cwebsocket(websocketUrl);
const orderbookChannel = new WebSocketOrderbookChannel(client);
const subscriptionOpts = {
baseTokenAddress: '0x323b5d4c32345ced77393b3530b1eed0f346429d',
quoteTokenAddress: '0xef7fff64389b814a946f3e92105513705ca6b990',