Bitfinex Public Market Data sharding on multiple sockets

bfx_pub_shard.js

/*
  npm install ws request crc-32
*/

const WebSocket = require('ws')
const request = require('request')
const CRC = require('crc-32')

const LIMIT_CHAN = 250

const sockets = []
let symbols = []
let requiredSocketNum = 0

async function getSymbols() {
  return new Promise((resolve, reject) => {
    request('https://api.bitfinex.com/v2/tickers?symbols=ALL', (err, rep, body) => {
      if (err) {
        return reject(err)
      }

      const tickers = JSON.parse(body)

      return resolve(tickers.map(t => t[0]))
    })
  })
}

async function run() {
  symbols = await getSymbols()
  requiredSocketNum = Math.ceil(symbols.length / LIMIT_CHAN)

  for (let i = 0; i < requiredSocketNum; i++) {
    reconnect(i)
  }

  setTimeout(() => {
    test()
  }, 5000)
}

function getSymbolSocketIx(s) {
  return Math.abs(CRC.str(s)) % requiredSocketNum 
}

function reconnect(ix) {
  if (sockets[ix]) {
    return
  }

  console.log(`connecting socket ${ix}`)

  const wss = new WebSocket('wss://api.bitfinex.com/ws/2')
  sockets[ix] = wss

  wss.onmessage = (msg) => console.log(msg.data)

  wss.onopen = () => {
    symbols.forEach(s => {
      const six = getSymbolSocketIx(s)
      if (six !== ix) {
        return
      }

      console.log(`subscribing to ${s} on socket ${ix}`)
      wss.send(JSON.stringify({ event: 'subscribe', channel: 'ticker', symbol: s }))
    })
  }

  wss.onclose = () => {
    sockets[ix] = null
    setTimeout(() => {
      reconnect(ix)
    }, 1000)
  }

  wss._ix = ix
}

function test () {
  for (let i = 0; i < symbols.length; i++) {
    const s = symbols[i]

    console.log(`${s} is on socket ${getSymbolSocketIx(s)}`)
  }
}

run()