diff --git a/dump.rdb b/dump.rdb deleted file mode 100644 index 9c1c315..0000000 Binary files a/dump.rdb and /dev/null differ diff --git a/package.json b/package.json index 08c8698..b548095 100644 --- a/package.json +++ b/package.json @@ -49,6 +49,7 @@ "clean": "rm -rf lib", "start": "node lib/index.js", "dev": "ts-node src/index.ts", + "ws-publish": "ts-node src/wsPublish.ts", "manager": "ts-node src/wsConnectionManager.ts", "dev:inspect": "yarn build && node --inspect ./lib/index.js", "dev:debug": "yarn build && node --inspect-brk --inspect=2230 ./lib/index.js", diff --git a/src/dlob-subscriber/DLOBSubscriberIO.ts b/src/dlob-subscriber/DLOBSubscriberIO.ts index f0d4682..188dd62 100644 --- a/src/dlob-subscriber/DLOBSubscriberIO.ts +++ b/src/dlob-subscriber/DLOBSubscriberIO.ts @@ -9,8 +9,6 @@ import { groupL2, } from '@drift-labs/sdk'; import { getOracleForMarket, l2WithBNToStrings } from '../utils/utils'; -import { Server } from 'socket.io'; -import { getIOServer } from '..'; import { RedisClient } from '../utils/redisClient'; type wsMarketL2Args = { @@ -28,7 +26,6 @@ type wsMarketL2Args = { export class DLOBSubscriberIO extends DLOBSubscriber { public marketL2Args: wsMarketL2Args[] = []; public lastSeenL2Formatted: Map>; - io: Server; redisClient: RedisClient; constructor(config: DLOBSubscriptionConfig & { redisClient: RedisClient }) { @@ -67,7 +64,6 @@ export class DLOBSubscriberIO extends DLOBSubscriber { } override async updateDLOB(): Promise { - this.io = getIOServer(); await super.updateDLOB(); for (const l2Args of this.marketL2Args) { this.getL2AndSendMsg(l2Args); @@ -85,12 +81,13 @@ export class DLOBSubscriberIO extends DLOBSubscriber { l2Formatted = l2WithBNToStrings(l2); } - if ( - l2Args.updateOnChange - ) { - if (this.lastSeenL2Formatted - .get(l2Args.marketType) - ?.get(l2Args.marketIndex) === JSON.stringify(l2Formatted)) return; + if (l2Args.updateOnChange) { + if ( + this.lastSeenL2Formatted + .get(l2Args.marketType) + ?.get(l2Args.marketIndex) === JSON.stringify(l2Formatted) + ) + return; } this.lastSeenL2Formatted .get(l2Args.marketType) diff --git a/src/index.ts b/src/index.ts index f0f132c..891e3ac 100644 --- a/src/index.ts +++ b/src/index.ts @@ -23,12 +23,12 @@ import { groupL2, Wallet, UserStatsMap, + DLOBSubscriber, } from '@drift-labs/sdk'; import { logger, setLogLevel } from './utils/logger'; import * as http from 'http'; -import { Server } from 'socket.io'; import { l2WithBNToStrings, sleep, @@ -40,10 +40,8 @@ import { getSerumSubscriber, validateDlobQuery, } from './utils/utils'; -import { DLOBSubscriberIO } from './dlob-subscriber/DLOBSubscriberIO'; import { handleResponseTime } from './core/middleware'; import { handleHealthCheck } from './core/metrics'; -import { RedisClient } from './utils/redisClient'; require('dotenv').config(); const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; @@ -103,11 +101,6 @@ app.use((req, _res, next) => { app.use(errorHandler); const server = http.createServer(app); -const io = new Server(server); - -export function getIOServer(): Server { - return io; -} const opts = program.opts(); setLogLevel(opts.debug ? 'debug' : 'info'); @@ -204,15 +197,11 @@ const main = async () => { ); await userStatsMap.subscribe(); - const redisClient = new RedisClient('localhost', '6379'); - await redisClient.connect(); - - const dlobSubscriber = new DLOBSubscriberIO({ + const dlobSubscriber = new DLOBSubscriber({ driftClient, dlobSource: userMap, slotSource: slotSubscriber, updateFrequency: ORDERBOOK_UPDATE_INTERVAL, - redisClient, }); await dlobSubscriber.subscribe(); @@ -826,14 +815,6 @@ const main = async () => { server.listen(serverPort, () => { logger.info(`DLOB server listening on port http://localhost:${serverPort}`); }); - - io.on('connection', (socket) => { - console.log('a user connected'); - - socket.on('disconnect', () => { - console.log('user disconnected'); - }); - }); }; async function recursiveTryCatch(f: () => void) { diff --git a/src/wsConnectionManager.ts b/src/wsConnectionManager.ts index e5120ac..1faf185 100644 --- a/src/wsConnectionManager.ts +++ b/src/wsConnectionManager.ts @@ -14,8 +14,11 @@ app.set('trust proxy', 1); const server = http.createServer(app); const io = new Server(server); +const REDIS_HOST = process.env.REDIS_HOST || 'localhost'; +const REDIS_PORT = process.env.REDIS_PORT || '6379'; + async function main() { - const redisClient = new RedisClient('localhost', '6379'); + const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT); await redisClient.connect(); io.on('connection', (socket) => { diff --git a/src/wsPublish.ts b/src/wsPublish.ts new file mode 100644 index 0000000..c814e33 --- /dev/null +++ b/src/wsPublish.ts @@ -0,0 +1,121 @@ +import { program } from 'commander'; + +import { Connection, Commitment, PublicKey, Keypair } from '@solana/web3.js'; + +import { + DriftClient, + initialize, + DriftEnv, + SlotSubscriber, + UserMap, + Wallet, + UserStatsMap, +} from '@drift-labs/sdk'; + +import { logger, setLogLevel } from './utils/logger'; + +import { sleep } from './utils/utils'; +import { DLOBSubscriberIO } from './dlob-subscriber/DLOBSubscriberIO'; +import { RedisClient } from './utils/redisClient'; + +require('dotenv').config(); +const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; +const commitHash = process.env.COMMIT; +const REDIS_HOST = process.env.REDIS_HOST || 'localhost'; +const REDIS_PORT = process.env.REDIS_PORT || '6379'; + +//@ts-ignore +const sdkConfig = initialize({ env: process.env.ENV }); + +const stateCommitment: Commitment = 'processed'; +const ORDERBOOK_UPDATE_INTERVAL = 1000; + +let driftClient: DriftClient; + +const opts = program.opts(); +setLogLevel(opts.debug ? 'debug' : 'info'); + +const endpoint = process.env.ENDPOINT; +const wsEndpoint = process.env.WS_ENDPOINT; +logger.info(`RPC endpoint: ${endpoint}`); +logger.info(`WS endpoint: ${wsEndpoint}`); +logger.info(`DriftEnv: ${driftEnv}`); +logger.info(`Commit: ${commitHash}`); + +const main = async () => { + const wallet = new Wallet(new Keypair()); + const clearingHousePublicKey = new PublicKey(sdkConfig.DRIFT_PROGRAM_ID); + + const connection = new Connection(endpoint, { + wsEndpoint: wsEndpoint, + commitment: stateCommitment, + }); + + driftClient = new DriftClient({ + connection, + wallet, + programID: clearingHousePublicKey, + accountSubscription: { + type: 'websocket', + }, + env: driftEnv, + userStats: true, + }); + + const slotSubscriber = new SlotSubscriber(connection, {}); + + const lamportsBalance = await connection.getBalance(wallet.publicKey); + logger.info( + `DriftClient ProgramId: ${driftClient.program.programId.toBase58()}` + ); + logger.info(`Wallet pubkey: ${wallet.publicKey.toBase58()}`); + logger.info(` . SOL balance: ${lamportsBalance / 10 ** 9}`); + + await driftClient.subscribe(); + driftClient.eventEmitter.on('error', (e) => { + logger.info('clearing house error'); + logger.error(e); + }); + + await slotSubscriber.subscribe(); + + const userMap = new UserMap( + driftClient, + driftClient.userAccountSubscriptionConfig, + false + ); + await userMap.subscribe(); + const userStatsMap = new UserStatsMap( + driftClient, + driftClient.userStatsAccountSubscriptionConfig + ); + await userStatsMap.subscribe(); + + const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT); + await redisClient.connect(); + + const dlobSubscriber = new DLOBSubscriberIO({ + driftClient, + dlobSource: userMap, + slotSource: slotSubscriber, + updateFrequency: ORDERBOOK_UPDATE_INTERVAL, + redisClient, + }); + await dlobSubscriber.subscribe(); + + console.log('DLOBSubscriber Publishing Messages'); +}; + +async function recursiveTryCatch(f: () => void) { + try { + await f(); + } catch (e) { + console.error(e); + await sleep(15000); + await recursiveTryCatch(f); + } +} + +recursiveTryCatch(() => main()); + +export { sdkConfig, endpoint, wsEndpoint, driftEnv, commitHash, driftClient };