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

must use order subscriber

Browse Source
This commit is contained in:
wphan
2023-12-18 15:40:07 -08:00
parent b318bc7a54
commit 746ca73c5c
1 changed files with 43 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
src/index.ts
View File

@@ -18,7 +18,6 @@ import {
DriftClient, DriftClient,
DriftClientSubscriptionConfig, DriftClientSubscriptionConfig,
DriftEnv, DriftEnv,
SlotSource,
SlotSubscriber, SlotSubscriber,
Wallet, Wallet,
getUserStatsAccountPublicKey, getUserStatsAccountPublicKey,
@@ -52,10 +51,7 @@ import {
validateDlobQuery, validateDlobQuery,
} from './utils/utils'; } from './utils/utils';
import FEATURE_FLAGS from './utils/featureFlags'; import FEATURE_FLAGS from './utils/featureFlags';
import { import { getDLOBProviderFromOrderSubscriber } from './dlobProvider';
DLOBProvider,
getDLOBProviderFromOrderSubscriber,
} from './dlobProvider';
import { RedisClient } from './utils/redisClient'; import { RedisClient } from './utils/redisClient';
require('dotenv').config(); require('dotenv').config();
@@ -74,8 +70,6 @@ const serverPort = process.env.PORT || 6969;
export const ORDERBOOK_UPDATE_INTERVAL = 1000; export const ORDERBOOK_UPDATE_INTERVAL = 1000;
const WS_FALLBACK_FETCH_INTERVAL = ORDERBOOK_UPDATE_INTERVAL * 10; const WS_FALLBACK_FETCH_INTERVAL = ORDERBOOK_UPDATE_INTERVAL * 10;
const useWebsocket = process.env.USE_WEBSOCKET?.toLowerCase() === 'true'; const useWebsocket = process.env.USE_WEBSOCKET?.toLowerCase() === 'true';
const useOrderSubscriber =
process.env.USE_ORDER_SUBSCRIBER?.toLowerCase() === 'true';
const rateLimitCallsPerSecond = process.env.RATE_LIMIT_CALLS_PER_SECOND const rateLimitCallsPerSecond = process.env.RATE_LIMIT_CALLS_PER_SECOND
? parseInt(process.env.RATE_LIMIT_CALLS_PER_SECOND) ? parseInt(process.env.RATE_LIMIT_CALLS_PER_SECOND)
: 1; : 1;
@@ -152,7 +146,6 @@ const wsEndpoint = process.env.WS_ENDPOINT;
logger.info(`RPC endpoint: ${endpoint}`); logger.info(`RPC endpoint: ${endpoint}`);
logger.info(`WS endpoint: ${wsEndpoint}`); logger.info(`WS endpoint: ${wsEndpoint}`);
logger.info(`useWebsocket: ${useWebsocket}`); logger.info(`useWebsocket: ${useWebsocket}`);
logger.info(`useOrderSubscriber: ${useOrderSubscriber}`);
logger.info(`DriftEnv: ${driftEnv}`); logger.info(`DriftEnv: ${driftEnv}`);
logger.info(`Commit: ${commitHash}`); logger.info(`Commit: ${commitHash}`);
@@ -233,7 +226,6 @@ const main = async () => {
let slotSubscriber: SlotSubscriber | undefined; let slotSubscriber: SlotSubscriber | undefined;
let accountSubscription: DriftClientSubscriptionConfig; let accountSubscription: DriftClientSubscriptionConfig;
let slotSource: SlotSource;
if (!useWebsocket) { if (!useWebsocket) {
bulkAccountLoader = new BulkAccountLoader( bulkAccountLoader = new BulkAccountLoader(
@@ -246,7 +238,6 @@ const main = async () => {
type: 'polling', type: 'polling',
accountLoader: bulkAccountLoader, accountLoader: bulkAccountLoader,
}; };
slotSource = bulkAccountLoader;
} else { } else {
accountSubscription = { accountSubscription = {
type: 'websocket', type: 'websocket',
@@ -264,42 +255,38 @@ const main = async () => {
env: driftEnv, env: driftEnv,
}); });
let dlobProvider: DLOBProvider; let subscriptionConfig;
if (useOrderSubscriber) { if (useWebsocket) {
let subscriptionConfig; subscriptionConfig = {
if (useWebsocket) { type: 'websocket',
subscriptionConfig = { commitment: stateCommitment,
type: 'websocket', resyncIntervalMs: WS_FALLBACK_FETCH_INTERVAL,
commitment: stateCommitment, };
resyncIntervalMs: WS_FALLBACK_FETCH_INTERVAL, } else {
}; subscriptionConfig = {
} else { type: 'polling',
subscriptionConfig = { frequency: ORDERBOOK_UPDATE_INTERVAL,
type: 'polling', commitment: stateCommitment,
frequency: ORDERBOOK_UPDATE_INTERVAL, };
commitment: stateCommitment,
};
}
let updatesReceivedTotal = 0;
const orderSubscriber = new OrderSubscriber({
driftClient,
subscriptionConfig,
});
orderSubscriber.eventEmitter.on(
'updateReceived',
(_pubkey: PublicKey, _slot: number, _dataType: 'raw' | 'decoded') => {
setLastReceivedWsMsgTs(Date.now());
// eslint-disable-next-line @typescript-eslint/no-unused-vars
updatesReceivedTotal++;
accountUpdatesCounter.add(1);
}
);
dlobProvider = getDLOBProviderFromOrderSubscriber(orderSubscriber);
slotSource = dlobProvider;
} }
let updatesReceivedTotal = 0;
const orderSubscriber = new OrderSubscriber({
driftClient,
subscriptionConfig,
});
orderSubscriber.eventEmitter.on(
'updateReceived',
(_pubkey: PublicKey, _slot: number, _dataType: 'raw' | 'decoded') => {
setLastReceivedWsMsgTs(Date.now());
// eslint-disable-next-line @typescript-eslint/no-unused-vars
updatesReceivedTotal++;
accountUpdatesCounter.add(1);
}
);
const dlobProvider = getDLOBProviderFromOrderSubscriber(orderSubscriber);
const dlobCoder = DLOBOrdersCoder.create(); const dlobCoder = DLOBOrdersCoder.create();
await driftClient.subscribe(); await driftClient.subscribe();
@@ -325,7 +312,7 @@ const main = async () => {
const dlobSubscriber = new DLOBSubscriber({ const dlobSubscriber = new DLOBSubscriber({
driftClient, driftClient,
dlobSource: dlobProvider, dlobSource: dlobProvider,
slotSource, slotSource: dlobProvider,
updateFrequency: ORDERBOOK_UPDATE_INTERVAL, updateFrequency: ORDERBOOK_UPDATE_INTERVAL,
}); });
await dlobSubscriber.subscribe(); await dlobSubscriber.subscribe();
@@ -361,10 +348,10 @@ const main = async () => {
app.get( app.get(
'/health', '/health',
handleHealthCheck(2 * WS_FALLBACK_FETCH_INTERVAL, slotSource) handleHealthCheck(2 * WS_FALLBACK_FETCH_INTERVAL, dlobProvider)
); );
app.get('/startup', handleStartup); app.get('/startup', handleStartup);
app.get('/', handleHealthCheck(2 * WS_FALLBACK_FETCH_INTERVAL, slotSource)); app.get('/', handleHealthCheck(2 * WS_FALLBACK_FETCH_INTERVAL, dlobProvider));
if (FEATURE_FLAGS.ENABLE_ORDERS_ENDPOINTS) { if (FEATURE_FLAGS.ENABLE_ORDERS_ENDPOINTS) {
app.get('/orders/json/raw', async (_req, res, next) => { app.get('/orders/json/raw', async (_req, res, next) => {
@@ -372,7 +359,7 @@ const main = async () => {
// object with userAccount key and orders object serialized // object with userAccount key and orders object serialized
const orders: Array<any> = []; const orders: Array<any> = [];
const oracles: Array<any> = []; const oracles: Array<any> = [];
const slot = slotSource.getSlot(); const slot = dlobProvider.getSlot();
for (const market of driftClient.getPerpMarketAccounts()) { for (const market of driftClient.getPerpMarketAccounts()) {
const oracle = driftClient.getOracleDataForPerpMarket( const oracle = driftClient.getOracleDataForPerpMarket(
@@ -417,7 +404,7 @@ const main = async () => {
app.get('/orders/json', async (_req, res, next) => { app.get('/orders/json', async (_req, res, next) => {
try { try {
// object with userAccount key and orders object serialized // object with userAccount key and orders object serialized
const slot = slotSource.getSlot(); const slot = dlobProvider.getSlot();
const orders: Array<any> = []; const orders: Array<any> = [];
const oracles: Array<any> = []; const oracles: Array<any> = [];
for (const market of driftClient.getPerpMarketAccounts()) { for (const market of driftClient.getPerpMarketAccounts()) {
@@ -587,7 +574,7 @@ const main = async () => {
res.end( res.end(
JSON.stringify({ JSON.stringify({
slot: slotSource.getSlot(), slot: dlobProvider.getSlot(),
data: dlobCoder.encode(dlobOrders).toString('base64'), data: dlobCoder.encode(dlobOrders).toString('base64'),
}) })
); );
@@ -678,7 +665,7 @@ const main = async () => {
.getDLOB() .getDLOB()
.getRestingLimitBids( .getRestingLimitBids(
normedMarketIndex, normedMarketIndex,
slotSource.getSlot(), dlobProvider.getSlot(),
normedMarketType, normedMarketType,
oracle oracle
) )
@@ -689,7 +676,7 @@ const main = async () => {
.getDLOB() .getDLOB()
.getRestingLimitAsks( .getRestingLimitAsks(
normedMarketIndex, normedMarketIndex,
slotSource.getSlot(), dlobProvider.getSlot(),
normedMarketType, normedMarketType,
oracle oracle
) )
@@ -763,7 +750,7 @@ const main = async () => {
} }
if ( if (
redisL2 && redisL2 &&
slotSource.getSlot() - parseInt(JSON.parse(redisL2).slot) < 10 dlobProvider.getSlot() - parseInt(JSON.parse(redisL2).slot) < 10
) )
l2Formatted = redisL2; l2Formatted = redisL2;
} else if ( } else if (
@@ -789,7 +776,7 @@ const main = async () => {
} }
if ( if (
redisL2 && redisL2 &&
slotSource.getSlot() - parseInt(JSON.parse(redisL2).slot) < 10 dlobProvider.getSlot() - parseInt(JSON.parse(redisL2).slot) < 10
) )
l2Formatted = redisL2; l2Formatted = redisL2;
} }
@@ -945,7 +932,7 @@ const main = async () => {
} }
if (redisL2) { if (redisL2) {
const parsedRedisL2 = JSON.parse(redisL2); const parsedRedisL2 = JSON.parse(redisL2);
if (slotSource.getSlot() - parseInt(parsedRedisL2.slot) < 10) if (dlobProvider.getSlot() - parseInt(parsedRedisL2.slot) < 10)
l2Formatted = parsedRedisL2; l2Formatted = parsedRedisL2;
} }
} else if ( } else if (
@@ -970,7 +957,7 @@ const main = async () => {
} }
if (redisL2) { if (redisL2) {
const parsedRedisL2 = JSON.parse(redisL2); const parsedRedisL2 = JSON.parse(redisL2);
if (slotSource.getSlot() - parseInt(parsedRedisL2.slot) < 10) if (dlobProvider.getSlot() - parseInt(parsedRedisL2.slot) < 10)
l2Formatted = parsedRedisL2; l2Formatted = parsedRedisL2;
} }
} }