trade/trade-drift-dlob
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'master' into staging

Browse Source 
mem management
This commit is contained in:
Nour Alharithi
2023-12-12 09:11:20 -08:00
parent 759f0616c6 436e63dbd5
commit 0985b7d3e5
3 changed files with 29 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
package.json
View File

@@ -6,7 +6,7 @@
"license": "Apache-2.0",
"dependencies": {
"@coral-xyz/anchor": "^0.29.0",
"@drift-labs/sdk": "2.52.0-beta.0",
"@drift-labs/sdk": "2.52.0-beta.1",
"@opentelemetry/api": "^1.1.0",
"@opentelemetry/auto-instrumentations-node": "^0.31.1",
"@opentelemetry/exporter-prometheus": "^0.31.0",
@@ -56,6 +56,7 @@
"dlob-publish": "ts-node src/publishers/dlobPublisher.ts",
"trades-publish": "ts-node src/publishers/tradesPublisher.ts",
"ws-manager": "ts-node src/wsConnectionManager.ts",
"ws-manager:inspect": "yarn build && node --inspect ./lib/wsConnectionManager.js",
"dev:inspect": "yarn build && node --inspect ./lib/index.js",
"dev:debug": "yarn build && node --inspect-brk --inspect=2230 ./lib/index.js",
"example": "ts-node example/client.ts",

29
src/wsConnectionManager.ts
View File

@@ -23,7 +23,11 @@ const wsConnectionsGauge = new Gauge({
});
const server = http.createServer(app);
const wss = new WebSocketServer({ server, path: '/ws' });
const wss = new WebSocketServer({
server,
path: '/ws',
perMessageDeflate: false,
});
const REDIS_HOST = process.env.REDIS_HOST || 'localhost';
const REDIS_PORT = process.env.REDIS_PORT || '6379';
@@ -99,7 +103,10 @@ async function main() {
const subscribers = channelSubscribers.get(subscribedChannel);
if (subscribers) {
subscribers.forEach((ws) => {
ws.send(JSON.stringify({ channel: subscribedChannel, data: message }));
if (ws.readyState === WebSocket.OPEN && ws.bufferedAmount < 20)
ws.send(
JSON.stringify({ channel: subscribedChannel, data: message })
);
});
}
});
@@ -235,6 +242,7 @@ async function main() {
// Handle disconnection
ws.on('close', () => {
console.log('Client disconnected');
// Clear any existing intervals and timeouts
channelSubscribers.forEach((subscribers, channel) => {
if (subscribers.delete(ws) && subscribers.size === 0) {
@@ -263,11 +271,20 @@ async function main() {
server.on('error', (error) => {
console.error('Server error:', error);
});
}
process.on('unhandledRejection', (reason, promise) => {
console.error('Unhandled Rejection at:', promise, 'reason:', reason);
});
// Periodic check for bad clients and disconnect them to alleviate backpressure
setInterval(() => {
let set = new Set<WebSocket>();
for (const [_channel, wsSet] of channelSubscribers) {
set = new Set([...set, ...wsSet]);
}
for (const ws of set) {
if (ws.bufferedAmount > 500) {
ws.close();
}
}
}, 5000);
}
async function recursiveTryCatch(f: () => void) {
try {

8
yarn.lock
View File

@@ -115,10 +115,10 @@
enabled "2.0.x"
kuler "^2.0.0"
"@drift-labs/sdk@2.52.0-beta.0":
version "2.52.0-beta.0"
resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.52.0-beta.0.tgz#03da2e192e3f2180238e5e464336e0fc1697c0e0"
integrity sha512-SnuXh06oJFzAvTQZ/dOjcWduhKC8rRMlyTE5Xqb3LfCcEkQ/B+XvxTItCSzAPh33qq0MMhZKjxGdqKezUxnV1w==
"@drift-labs/sdk@2.52.0-beta.1":
version "2.52.0-beta.1"
resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.52.0-beta.1.tgz#6b26c434edf9e796010b0be358c732cacc251a24"
integrity sha512-qlfMvN9Fc7TOtRXtkQbTgok7nN0pObosfL1oh7uiEvgQrMeQY2lJoUzSHi4U8djxR68MkVnLnwVGABEgKEBdRQ==
dependencies:
"@coral-xyz/anchor" "0.28.1-beta.2"
"@ellipsis-labs/phoenix-sdk" "^1.4.2"