Added ShrimpyWsClient to connect to the Shrimpy Websocket API

Author: nishant-shrimpy

README.md

@@ -439,4 +439,40 @@ const assetDominance = await client.getAssetDominance();
 * [`getAssetPopularity`](https://developers.shrimpy.io/docs/#get-asset-popularity)
 ```js
 const assetPopularity = await client.getAssetPopularity();
-```
+```
+
+## Websocket
+
+Users can access the Shrimpy websocket feed using the [`ShrimpyWsClient`](https://github.com/shrimpy-dev/shrimpy-node/blob/master/lib/client/shrimpy-ws-client.ts) class. A handler must be passed in on subscription
+that is responsible for processing incoming messages from the websocket stream. It is recommended that you simply send the message
+to another processing thread from your custom handler to prevent blocking the incoming message stream.
+
+The client handles pings to the Shrimpy server based on the [`API Documentation`](https://developers.shrimpy.io/docs/#websocket)
+
+```js
+import { ShrimpyWsClient, ISubscriptionRequest, IWebsocketMessage, IErrorMessage } from 'shrimpy-node';
+
+let errorHandler = (error: IErrorMessage) => { console.log(error) };
+let client = new ShrimpyWsClient(errorHandler);
+
+const subscribeData: ISubscriptionRequest = {
+	"type": "subscribe",
+	"pair": "btc-usd",
+	"exchange": "coinbasepro",
+	"channel": "trade"
+};
+
+const unsubscribeData : ISubscriptionRequest = {
+	"type": "unsubscribe",
+	"pair": "btc-usd",
+	"exchange": "coinbasepro",
+	"channel": "trade"
+};
+
+let handler = (msg: IWebsocketMessage) => { console.log(msg); };
+
+client.connect();
+client.subscribe(subscribeData, handler);
+client.unsubscribe(unsubscribeData);
+client.forceDisconnect();
+```

lib/client/index.ts

@@ -1,3 +1,3 @@
-
-export * from './authentication-provider';
-export * from './shrimpy-api-client';
+export * from './authentication-provider';
+export * from './shrimpy-api-client';
+export * from './shrimpy-ws-client';

lib/client/shrimpy-ws-client.ts

@@ -0,0 +1,119 @@
+import * as WebSocket from 'ws';
+import { ISubscriptionRequest, IWebsocketMessage, IPingMessage, IErrorMessage } from '../models';
+
+
+export class ShrimpyWsClient {
+    private _websocket: WebSocket;
+    private _subscriptionCallbacks: { 
+        [subscription: string]: (data: IWebsocketMessage) => void
+    } = {};
+
+    private _websocketErrorCallback: (error: IErrorMessage) => void;
+
+    constructor (errorCallback: (error: IErrorMessage) => void) {
+        this._websocket = new WebSocket('wss://ws-feed.shrimpy.io/');
+        this._websocketErrorCallback = errorCallback;
+    }
+
+    public connect() {
+        
+        this._websocket.on('open', function open() {
+            // Open the connection
+        });
+
+        this._websocket.on('error', (error: Error) => {
+            const wsError : IErrorMessage = {
+                'type' : 'WebsocketClientError',
+                'code' : 2404,
+                'message' : error.message
+            };
+            this._websocketErrorCallback(wsError);
+        });
+
+        this._websocket.on('message', (message) => {
+            const parsedMessage: IWebsocketMessage = JSON.parse(message.toString());
+            const topic = this._getTopic(parsedMessage);
+
+            if (topic === 'ping') {
+                // Handle Ping
+                this._pong(parsedMessage as IPingMessage);
+                return;
+            }
+
+            if (topic === 'error') {
+                this._websocketErrorCallback(parsedMessage as IErrorMessage);
+                return;
+            }
+
+            const successCallback = this._subscriptionCallbacks[topic];
+            if (successCallback !== undefined) {
+                successCallback(parsedMessage);
+            }
+        });
+
+        this._websocket.on('close', () => {
+            // Connection has been closed, delete all callbacks
+            this._subscriptionCallbacks = {};
+        });
+    }
+
+    public disconnect() {
+        if (this._websocket !== undefined) {
+            this._websocket.close();
+        }
+    }
+
+    public forceDisconnect() {
+        if (this._websocket !== undefined) {
+            this._websocket.terminate();
+        }
+    }
+
+    public subscribe(
+        subscriptionRequest: ISubscriptionRequest,
+        successCallback: (data: IWebsocketMessage) => void,
+    ) {
+        if (this._websocket.OPEN) {
+            const topic = this._getTopic(subscriptionRequest);
+            this._subscriptionCallbacks[topic] = successCallback;
+            this._websocket.send(JSON.stringify(subscriptionRequest));
+        }
+    }
+
+    public unsubscribe(
+        unsubscriptionRequest: ISubscriptionRequest
+    ) {
+        const topic = this._getTopic(unsubscriptionRequest);
+        delete this._subscriptionCallbacks[topic];
+        this._websocket.send(JSON.stringify(unsubscriptionRequest));
+    }
+
+    private _getTopic(message: any): string {
+
+        if (message.hasOwnProperty('type')) {
+            const messageType = message['type'];
+            if (messageType.indexOf('subscribe') === -1) {
+                return messageType.toLowerCase();
+            }
+        }
+
+        const exchange = message['exchange'];
+        const pair = message['pair'];
+        const channel = message['channel'];
+
+        const rawKeys = [exchange.toLowerCase(), pair.toLowerCase(), channel.toLowerCase()];
+        const nonNullKeys = rawKeys.filter((k) => {
+            return k !== undefined;
+        });
+
+        return nonNullKeys.join('-');
+    }
+
+    private _pong(parsedData: IPingMessage) {
+        const pong = {
+            'type': 'pong',
+            'data': parsedData.data
+        };
+        this._websocket.send(JSON.stringify(pong));
+    }
+}

lib/enums/subscription-type-enum.ts

@@ -0,0 +1,3 @@
+export enum SubscriptionTypeEnum {
+    
+}

lib/models/index.ts

@@ -27,5 +27,8 @@ export * from './itotal-balance-history-item';
 export * from './itrade';
 export * from './itrade-changes';
 export * from './itrade-fill';
+export * from './itrade-item';
 export * from './itrading-pair';
+export * from './isubscription-request';
 export * from './iuser';
+export * from './iwebsocket-message';

lib/models/isubscription-request.ts

@@ -0,0 +1,12 @@
+export type ISubscriptionRequest = IExchangeSubscription | IOrderSubscription;
+
+export interface IExchangeSubscription {
+    type: 'subscribe' | 'unsubscribe';
+    exchange: string;
+    pair: string;
+    channel: string;
+}
+
+export interface IOrderSubscription {
+    channel: string;
+}

lib/models/itrade-item.ts

@@ -0,0 +1,10 @@
+import Decimal from "decimal.js";
+
+export interface ITradeItem {
+    id: number;
+    price: Decimal;
+    quantity: Decimal;
+    time: Date;
+    btcValue: Decimal;
+    usdValue: Decimal;
+}

lib/models/iwebsocket-message.ts

@@ -0,0 +1,41 @@
+import { IOrderBookItem } from "./iorder-book-item";
+import { ITradeItem } from "./itrade-item";
+
+export type IWebsocketMessage = IExchangePairMessage | IOrderMessage | IErrorMessage | IPingMessage;
+
+export interface IExchangePairMessage {
+    exchange: string;
+    pair: string;
+    channel: 'bbo' | 'orderbook' | 'trades'
+    content: WebsocketContent
+}
+
+export interface IOrderMessage {
+    channel: 'orders';
+    content: WebsocketContent;
+}
+
+export interface IErrorMessage {
+    type: string;
+    code: number;
+    message: string;
+}
+
+export interface IPingMessage {
+    type: string;
+    data: number;
+}
+
+export type WebsocketContent = OrderBookContent | TradeContent | OrdersContent;
+
+export interface OrderBookContent {
+    sequence: number;
+    asks: IOrderBookItem[];
+    bids: IOrderBookItem[];
+}
+
+export interface TradeContent {
+    trades: ITradeItem[];
+}
+
+export type OrdersContent = string[];

lib/public-api.ts

@@ -1,3 +1,3 @@
 
-export { ShrimpyApiClient } from './client';
+export { ShrimpyApiClient, ShrimpyWsClient } from './client';
 export * from './models';