serum-vial - npm Package Compare versions

Comparing version 0.6.2 to 0.7.0

dist/boot_server.d.ts

 import { SerumMarket } from './types';
 export declare function bootServer({ port, nodeEndpoint, validateL3Diffs, minionsCount, markets, commitment }: BootOptions): Promise<void>;
+export declare function stopServer(): Promise<void>;
 declare type BootOptions = {
@@ -4,0 +5,0 @@ port: number;

dist/boot_server.js

@@ -6,3 +6,3 @@ "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-exports.bootServer = void 0;
+exports.stopServer = exports.bootServer = void 0;
 const os_1 = __importDefault(require("os"));
@@ -70,2 +70,7 @@ const path_1 = __importDefault(require("path"));
 exports.bootServer = bootServer;
+async function stopServer() {
+    helpers_1.cleanupChannel.postMessage('cleanup');
+    await helpers_1.wait(10 * 1000);
+}
+exports.stopServer = stopServer;
 //# sourceMappingURL=boot_server.js.map

dist/consts.d.ts

 export declare const OPS: readonly ["subscribe", "unsubscribe"];
 export declare const CHANNELS: readonly ["level3", "level2", "level1", "trades"];
-declare const TRADES_MESSAGE_TYPES: readonly ["trade"];
-declare const LEVEL1_MESSAGE_TYPES: readonly ["quote", "trade"];
-declare const LEVEL2_MESSAGE_TYPES: readonly ["l2snapshot", "l2update", "trade"];
+declare const TRADES_MESSAGE_TYPES: readonly ["recent_trades", "trade"];
+declare const LEVEL1_MESSAGE_TYPES: readonly ["recent_trades", "trade", "quote"];
+declare const LEVEL2_MESSAGE_TYPES: readonly ["l2snapshot", "l2update", "recent_trades", "trade"];
 declare const LEVEL3_MESSAGE_TYPES: readonly ["l3snapshot", "open", "fill", "change", "done"];
@@ -7,0 +7,0 @@ export declare const MESSAGE_TYPES_PER_CHANNEL: {

dist/consts.js

@@ -6,5 +6,5 @@ "use strict";
 exports.CHANNELS = ['level3', 'level2', 'level1', 'trades'];
-const TRADES_MESSAGE_TYPES = ['trade'];
-const LEVEL1_MESSAGE_TYPES = ['quote', 'trade'];
-const LEVEL2_MESSAGE_TYPES = ['l2snapshot', 'l2update', 'trade'];
+const TRADES_MESSAGE_TYPES = ['recent_trades', 'trade'];
+const LEVEL1_MESSAGE_TYPES = ['recent_trades', 'trade', 'quote'];
+const LEVEL2_MESSAGE_TYPES = ['l2snapshot', 'l2update', 'recent_trades', 'trade'];
 const LEVEL3_MESSAGE_TYPES = ['l3snapshot', 'open', 'fill', 'change', 'done'];
@@ -11,0 +11,0 @@ exports.MESSAGE_TYPES_PER_CHANNEL = {

dist/data_mapper.d.ts

@@ -18,2 +18,3 @@ import { Market } from '@project-serum/serum';
     private _zeroWithPrecision;
+    private readonly _recentTrades;
     constructor(_options: {
@@ -20,0 +21,0 @@ readonly symbol: string;

dist/data_mapper.js

@@ -5,2 +5,3 @@ "use strict";
 const serum_1 = require("@project-serum/serum");
+const helpers_1 = require("./helpers");
 const logger_1 = require("./logger");
@@ -22,2 +23,3 @@ // DataMapper maps bids, asks and evenQueue accounts data to normalized messages
         this._currentQuote = undefined;
+        this._recentTrades = new helpers_1.CircularBuffer(100);
         this._mapToL2Level = (level) => {
@@ -191,2 +193,10 @@ const price = this._options.market.priceLotsToNumber(level[0]).toFixed(this._options.priceDecimalPlaces);
                     yield this._putInEnvelope(tradeMessage, true);
+                    this._recentTrades.append(tradeMessage);
+                    const recentTradesMessage = {
+                        type: 'recent_trades',
+                        symbol: this._options.symbol,
+                        timestamp,
+                        trades: [...this._recentTrades.items()]
+                    };
+                    yield this._putInEnvelope(recentTradesMessage, false);
                 }
@@ -193,0 +203,0 @@ }

dist/helpers.d.ts

@@ -13,3 +13,3 @@ import { SerumMarket } from './types';
     append(value: T): T | undefined;
-    items(): Generator<T | undefined, void, unknown>;
+    items(): Generator<NonNullable<T>, void, unknown>;
     get count(): number;
@@ -22,2 +22,3 @@ clear(): void;
 export declare const serumMarketsChannel: BroadcastChannel;
+export declare const cleanupChannel: BroadcastChannel;
 export declare function executeAndRetry<T>(operation: (attempt: number) => Promise<T>, { maxRetries }: {
@@ -24,0 +25,0 @@ maxRetries: number;

dist/helpers.js

@@ -6,3 +6,3 @@ "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-exports.getDefaultMarkets = exports.executeAndRetry = exports.serumMarketsChannel = exports.serumDataChannel = exports.serumProducerReadyChannel = exports.minionReadyChannel = exports.CircularBuffer = exports.decimalPlaces = exports.batch = exports.getAllowedValuesText = exports.getDidYouMean = exports.wait = void 0;
+exports.getDefaultMarkets = exports.executeAndRetry = exports.cleanupChannel = exports.serumMarketsChannel = exports.serumDataChannel = exports.serumProducerReadyChannel = exports.minionReadyChannel = exports.CircularBuffer = exports.decimalPlaces = exports.batch = exports.getAllowedValuesText = exports.getDidYouMean = exports.wait = void 0;
 const serum_1 = require("@project-serum/serum");
@@ -89,2 +89,3 @@ const didyoumean2_1 = __importDefault(require("didyoumean2"));
 exports.serumMarketsChannel = new BroadcastChannel('SerumMarkets');
+exports.cleanupChannel = new BroadcastChannel('Cleanup');
 async function executeAndRetry(operation, { maxRetries }) {
@@ -91,0 +92,0 @@ let attempts = 0;

dist/index.d.ts

@@ -1,2 +0,2 @@
-export { bootServer } from './boot_server';
+export { bootServer, stopServer } from './boot_server';
 export { logger } from './logger';
@@ -3,0 +3,0 @@ export { getDefaultMarkets } from './helpers';

dist/index.js

@@ -13,5 +13,6 @@ "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-exports.getDefaultMarkets = exports.logger = exports.bootServer = void 0;
+exports.getDefaultMarkets = exports.logger = exports.stopServer = exports.bootServer = void 0;
 var boot_server_1 = require("./boot_server");
 Object.defineProperty(exports, "bootServer", { enumerable: true, get: function () { return boot_server_1.bootServer; } });
+Object.defineProperty(exports, "stopServer", { enumerable: true, get: function () { return boot_server_1.stopServer; } });
 var logger_1 = require("./logger");
@@ -18,0 +19,0 @@ Object.defineProperty(exports, "logger", { enumerable: true, get: function () { return logger_1.logger; } });

dist/minion.js

@@ -52,28 +52,5 @@ "use strict";
         this._l3SnapshotsSerialized = {};
-        this._recentTrades = {};
         this._recentTradesSerialized = {};
         this._quotesSerialized = {};
-        this._listRecentTrades = async (res, req) => {
-            res.onAborted(() => {
-                res.aborted = true;
-            });
-            const marketName = decodeURIComponent(req.getParameter(0));
-            const { isValid, error } = this._validateMarketName(marketName);
-            if (isValid === false) {
-                res.writeHeader('content-type', 'application/json');
-                res.writeStatus('400');
-                res.end(JSON.stringify({ error }));
-                return;
-            }
-            let serializedRecentTrades = this._recentTradesSerialized[marketName];
-            if (serializedRecentTrades === undefined) {
-                const recentTrades = this._recentTrades[marketName] !== undefined ? [...this._recentTrades[marketName].items()] : [];
-                recentTrades.reverse();
-                serializedRecentTrades = `[${recentTrades.join(',')}]`;
-            }
-            if (!res.aborted) {
-                res.writeHeader('content-type', 'application/json');
-                res.end(serializedRecentTrades);
-            }
-        };
+        this._listenSocket = undefined;
         this._cachedListMarketsResponse = undefined;
@@ -136,4 +113,3 @@ //async based on https://github.com/uNetworking/uWebSockets.js/blob/master/examples/AsyncFunction.js
         })
-            .get(`${apiPrefix}/markets`, this._listMarkets)
-            .get(`${apiPrefix}/recent-trades/:market`, this._listRecentTrades);
+            .get(`${apiPrefix}/markets`, this._listMarkets);
     }
@@ -144,2 +120,3 @@ async start(port) {
                 if (socket) {
+                    this._listenSocket = socket;
                     logger_1.logger.log('info', `Listening on port ${port}`, meta);
@@ -156,2 +133,7 @@ resolve();
     }
+    async stop() {
+        if (this._listenSocket !== undefined) {
+            uWebSockets_js_1.us_listen_socket_close(this._listenSocket);
+        }
+    }
     initMarketsCache(cachedResponse) {
@@ -176,9 +158,2 @@ this._cachedListMarketsResponse = cachedResponse;
         }
-        if (message.type === 'trade') {
-            if (this._recentTrades[message.symbol] === undefined) {
-                this._recentTrades[message.symbol] = new helpers_1.CircularBuffer(100);
-            }
-            this._recentTrades[message.symbol].append(message.payload);
-            this._recentTradesSerialized[message.symbol] = undefined;
-        }
         if (message.publish) {
@@ -231,2 +206,17 @@ this._server.publish(topic, message.payload);
                         ws.subscribe(topic);
+                        if (type === 'recent_trades') {
+                            const recentTrades = this._recentTradesSerialized[market];
+                            if (recentTrades !== undefined) {
+                                ws.send(recentTrades);
+                            }
+                            else {
+                                const emptyRecentTradesMessage = {
+                                    type: 'recent_trades',
+                                    symbol: market,
+                                    timestamp: new Date().toISOString(),
+                                    trades: []
+                                };
+                                ws.send(JSON.stringify(emptyRecentTradesMessage));
+                            }
+                        }
                         if (type === 'quote') {
@@ -270,14 +260,2 @@ const quote = this._quotesSerialized[market];
     }
-    _validateMarketName(marketName) {
-        if (this._marketNames.includes(marketName) === false) {
-            const error = `Invalid market name provided: '${marketName}'.${helpers_1.getDidYouMean(marketName, this._marketNames)} ${helpers_1.getAllowedValuesText(this._marketNames)}`;
-            return {
-                isValid: false,
-                error
-            };
-        }
-        return {
-            isValid: true
-        };
-    }
     _validateRequestPayload(message) {
@@ -344,2 +322,5 @@ let payload;
 });
+helpers_1.cleanupChannel.onmessage = async () => {
+    await minion.stop();
+};
 //# sourceMappingURL=minion.js.map

dist/types.d.ts

@@ -16,2 +16,7 @@ import { Op, Channel, MessageType } from './consts';
 }
+export interface RecentTrades extends Message {
+    readonly type: 'recent_trades';
+    readonly symbol: string;
+    readonly trades: Trade[];
+}
 export interface DataMessage extends Message {
@@ -18,0 +23,0 @@ readonly symbol: string;

package.json

 {
   "name": "serum-vial",
-  "version": "0.6.2",
+  "version": "0.7.0",
   "engines": {
@@ -16,3 +16,3 @@ "node": ">=15"
     "precommit": "lint-staged",
-    "test": "jest --runInBand --forceExit",
+    "test": "npm run build && jest --forceExit",
     "prepare": "npm run build",
@@ -19,0 +19,0 @@ "start:debug": "npm run build && node ./bin/serum-vial.js --log-level=debug",

src/boot_server.ts

 import os from 'os'
 import path from 'path'
 import { Worker } from 'worker_threads'
-import { minionReadyChannel, serumProducerReadyChannel, wait } from './helpers'
+import { cleanupChannel, minionReadyChannel, serumProducerReadyChannel, wait } from './helpers'
 import { logger } from './logger'
@@ -93,2 +93,8 @@ import { SerumMarket } from './types'
 
+export async function stopServer() {
+  cleanupChannel.postMessage('cleanup')
+
+  await wait(10 * 1000)
+}
+
 type BootOptions = {
@@ -95,0 +101,0 @@ port: number

src/consts.ts

 export const OPS = ['subscribe', 'unsubscribe'] as const
 export const CHANNELS = ['level3', 'level2', 'level1', 'trades'] as const
 
-const TRADES_MESSAGE_TYPES = ['trade'] as const
-const LEVEL1_MESSAGE_TYPES = ['quote', 'trade'] as const
-const LEVEL2_MESSAGE_TYPES = ['l2snapshot', 'l2update', 'trade'] as const
+const TRADES_MESSAGE_TYPES = ['recent_trades', 'trade'] as const
+const LEVEL1_MESSAGE_TYPES = ['recent_trades', 'trade', 'quote'] as const
+const LEVEL2_MESSAGE_TYPES = ['l2snapshot', 'l2update', 'recent_trades', 'trade'] as const
 const LEVEL3_MESSAGE_TYPES = ['l3snapshot', 'open', 'fill', 'change', 'done'] as const
@@ -8,0 +8,0 @@

src/data_mapper.ts

@@ -5,2 +5,3 @@ import { EVENT_QUEUE_LAYOUT, Market, Orderbook, getLayoutVersion } from '@project-serum/serum'
 import BN from 'bn.js'
+import { CircularBuffer } from './helpers'
 import { logger } from './logger'
@@ -21,2 +22,3 @@ import { AccountsNotificationPayload } from './rpc_client'
   Quote,
+  RecentTrades,
   Trade
@@ -57,2 +59,4 @@ } from './types'
 
+  private readonly _recentTrades: CircularBuffer<Trade> = new CircularBuffer(100)
+
   constructor(
@@ -250,2 +254,13 @@ private readonly _options: {
           yield this._putInEnvelope(tradeMessage, true)
+
+          this._recentTrades.append(tradeMessage)
+
+          const recentTradesMessage: RecentTrades = {
+            type: 'recent_trades',
+            symbol: this._options.symbol,
+            timestamp,
+            trades: [...this._recentTrades.items()]
+          }
+
+          yield this._putInEnvelope(recentTradesMessage, false)
         }
@@ -532,3 +547,3 @@ }
 
-  private _putInEnvelope(message: DataMessage, publish: boolean) {
+  private _putInEnvelope(message: DataMessage | RecentTrades, publish: boolean) {
     const envelope: MessageEnvelope = {
@@ -535,0 +550,0 @@ type: message.type,

src/helpers.ts

@@ -73,3 +73,3 @@ import { MARKETS } from '@project-serum/serum'
       const index = (this._index + i) % this._buffer.length
-      yield this._buffer[index]
+      yield this._buffer[index]!
     }
@@ -94,2 +94,3 @@ }
 export const serumMarketsChannel = new BroadcastChannel('SerumMarkets') as BroadcastChannel
+export const cleanupChannel = new BroadcastChannel('Cleanup') as BroadcastChannel
 
@@ -96,0 +97,0 @@ export async function executeAndRetry<T>(

src/index.ts

@@ -1,4 +0,4 @@
-export { bootServer } from './boot_server'
+export { bootServer, stopServer } from './boot_server'
 export { logger } from './logger'
 export { getDefaultMarkets } from './helpers'
 export * from './types'

src/minion.ts

 import { Market, getLayoutVersion } from '@project-serum/serum'
 import { Connection, PublicKey } from '@solana/web3.js'
-import { App, SSLApp, HttpRequest, HttpResponse, SHARED_COMPRESSOR, TemplatedApp, WebSocket } from 'uWebSockets.js'
+import {
+  App,
+  SSLApp,
+  HttpResponse,
+  SHARED_COMPRESSOR,
+  TemplatedApp,
+  WebSocket,
+  us_listen_socket_close
+} from 'uWebSockets.js'
 import { isMainThread, threadId, workerData } from 'worker_threads'
 import { CHANNELS, MESSAGE_TYPES_PER_CHANNEL, OPS } from './consts'
 import {
-  CircularBuffer,
+  cleanupChannel,
   getAllowedValuesText,
@@ -16,3 +24,3 @@ getDidYouMean,
 import { MessageEnvelope } from './serum_producer'
-import { ErrorResponse, SerumListMarketItem, SerumMarket, SubRequest, SuccessResponse } from './types'
+import { ErrorResponse, RecentTrades, SerumListMarketItem, SerumMarket, SubRequest, SuccessResponse } from './types'
 
@@ -68,6 +76,6 @@ const meta = {
   private readonly _l3SnapshotsSerialized: { [symbol: string]: string } = {}
-  private readonly _recentTrades: { [symbol: string]: CircularBuffer<string> } = {}
-  private readonly _recentTradesSerialized: { [symbol: string]: string | undefined } = {}
+  private readonly _recentTradesSerialized: { [symbol: string]: string } = {}
   private readonly _quotesSerialized: { [symbol: string]: string } = {}
   private readonly _marketNames: string[]
+  private _listenSocket: any | undefined = undefined
 
@@ -101,3 +109,2 @@ constructor(private readonly _nodeEndpoint: string, private readonly _markets: SerumMarket[]) {
       .get(`${apiPrefix}/markets`, this._listMarkets)
-      .get(`${apiPrefix}/recent-trades/:market`, this._listRecentTrades)
   }
@@ -109,2 +116,3 @@
         if (socket) {
+          this._listenSocket = socket
           logger.log('info', `Listening on port ${port}`, meta)
@@ -121,31 +129,6 @@ resolve()
 
-  private _listRecentTrades = async (res: HttpResponse, req: HttpRequest) => {
-    res.onAborted(() => {
-      res.aborted = true
-    })
-
-    const marketName = decodeURIComponent(req.getParameter(0))
-
-    const { isValid, error } = this._validateMarketName(marketName)
-    if (isValid === false) {
-      res.writeHeader('content-type', 'application/json')
-      res.writeStatus('400')
-      res.end(JSON.stringify({ error }))
-      return
+  public async stop() {
+    if (this._listenSocket !== undefined) {
+      us_listen_socket_close(this._listenSocket)
     }
-
-    let serializedRecentTrades = this._recentTradesSerialized[marketName]
-    if (serializedRecentTrades === undefined) {
-      const recentTrades =
-        this._recentTrades[marketName] !== undefined ? [...this._recentTrades[marketName]!.items()] : []
-
-      recentTrades.reverse()
-
-      serializedRecentTrades = `[${recentTrades.join(',')}]`
-    }
-
-    if (!res.aborted) {
-      res.writeHeader('content-type', 'application/json')
-      res.end(serializedRecentTrades)
-    }
   }
@@ -223,11 +206,2 @@
 
-    if (message.type === 'trade') {
-      if (this._recentTrades[message.symbol] === undefined) {
-        this._recentTrades[message.symbol] = new CircularBuffer(100)
-      }
-
-      this._recentTrades[message.symbol]!.append(message.payload)
-      this._recentTradesSerialized[message.symbol] = undefined
-    }
-
     if (message.publish) {
@@ -294,2 +268,18 @@ this._server.publish(topic, message.payload)
 
+            if (type === 'recent_trades') {
+              const recentTrades = this._recentTradesSerialized[market]
+              if (recentTrades !== undefined) {
+                ws.send(recentTrades)
+              } else {
+                const emptyRecentTradesMessage: RecentTrades = {
+                  type: 'recent_trades',
+                  symbol: market,
+                  timestamp: new Date().toISOString(),
+                  trades: []
+                }
+
+                ws.send(JSON.stringify(emptyRecentTradesMessage))
+              }
+            }
+
             if (type === 'quote') {
@@ -337,20 +327,2 @@ const quote = this._quotesSerialized[market]
 
-  private _validateMarketName(marketName: string) {
-    if (this._marketNames.includes(marketName) === false) {
-      const error = `Invalid market name provided: '${marketName}'.${getDidYouMean(
-        marketName,
-        this._marketNames
-      )} ${getAllowedValuesText(this._marketNames)}`
-
-      return {
-        isValid: false,
-        error
-      }
-    }
-
-    return {
-      isValid: true
-    }
-  }
-
   private _validateRequestPayload(message: Buffer) {
@@ -433,1 +405,5 @@ let payload
 })
+
+cleanupChannel.onmessage = async () => {
+  await minion.stop()
+}

src/types.ts

@@ -20,2 +20,8 @@ import { Op, Channel, MessageType } from './consts'
 
+export interface RecentTrades extends Message {
+  readonly type: 'recent_trades'
+  readonly symbol: string
+  readonly trades: Trade[]
+}
+
 export interface DataMessage extends Message {
@@ -22,0 +28,0 @@ readonly symbol: string

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

240720

increased by0.05%

Lines of code

3982

increased by0.28%

No dependency changes