WebSocket и несколько Dyno в приложении Heroku Node.js Использование Redis

Question

Я создаю приложение, развернутое на Heroku, которое использует WebSocket и Redis.

Соединение WebSocket работает правильно, когда я использую только 1 dyno, но когда я масштабируюсь до 2, я посылаю событие, мое приложение делает это дважды.

const ws = require('ws')
const jwt = require('jsonwebtoken')
const redis = require('redis')

const User = require('../models/user')

function verifyClient (info, callback) {
  let token = info.req.headers['sec-websocket-protocol']
  if (!token) { callback(false, 401, 'Unauthorized') } else {
    jwt.verify(token, Config.APP_SECRET, (err, decoded) => {
      if (err) { callback(false, 401, 'Unauthorized') } else {
        if (info.req.headers.gameId) { info.req.gameId = info.req.headers.gameId }
        info.req.userId = decoded.aud
        callback(true)
      }
    })
  }
};

let websocketServer, pub, sub

let clients = {}
let namespaces = {}

exports.initialize = function (httpServer) {
  websocketServer = new ws.Server({
    server: httpServer,
    verifyClient: verifyClient
  })
  pub = redis.createClient(Config.REDIS_URL, { no_ready_check: true, detect_buffers: true })
  pub.auth(Config.REDIS_PASSWORD, function (err) {
    if (err) throw err
  })
  sub = redis.createClient(Config.REDIS_URL, { no_ready_check: true, detect_buffers: true })
  sub.auth(Config.REDIS_PASSWORD, function (err) {
    if (err) throw err
  })

  function handleConnection (socket) {
    // socket.send(socket.upgradeReq.userId);
    socket.userId = socket.upgradeReq.userId // get the user id parsed from the decoded JWT in the middleware
    socket.isAlive = true
    socket.scope = socket.upgradeReq.url.split('/')[1] // url = "/scope/whatever" => ["", "scope", "whatever"]
    console.log('New connection: ' + socket.userId + ', scope: ' + socket.scope)

    socket.on('message', (data, flags) => { handleIncomingMessage(socket, data, flags) })
    socket.once('close', (code, reason) => { handleClosedConnection(socket, code, reason) })
    socket.on('pong', heartbeat)
    if (socket.scope === 'gameplay') {
      try {
        User.findByIdAndUpdate(socket.userId, { $set: { isOnLine: 2, lastSeen: Date.now() } }).select('id').lean()
        let key = [socket.userId, socket.scope].join(':')
        clients[key] = socket
        sub.psubscribe(['dispatch', '*', socket.userId, socket.scope].join(':'))
      } catch (e) { console.log(e) }
    } else {
      console.log('Scope : ' + socket.scope)
    }
    console.log('Connected Users : ' + Object.keys(clients))
  }
  function handleIncomingMessage (socket, message, flags) {
    let scope = socket.scope
    let userId = socket.userId
    let channel = ['dispatch', 'in', userId, scope].join(':')
    pub.publish(channel, message)
  }
  function handleClosedConnection (socket, code, reason) {
    console.log('Connection with ' + socket.userId + ' closed. Code: ' + code)

    if (socket.scope === 'gameplay') {
      try {
        User.findByIdAndUpdate(socket.userId, { $set: { isOnLine: 1 } }).select('id').lean()
        let key = [socket.userId, socket.scope].join(':')
        delete clients[key]
      } catch (e) {
        console.log(e)
      }
    } else {
      console.log('Scope : ' + socket.scope)
    }
  }
  function heartbeat (socket) {
    socket.isAlive = true
  }
  sub.on('pmessage', (pattern, channel, message) => {
    let channelComponents = channel.split(':')
    let dir = channelComponents[1]
    let userId = channelComponents[2]
    let scope = channelComponents[3]
    if (dir === 'in') {
      try {
        let handlers = namespaces[scope] || []
        if (handlers.length) {
          handlers.forEach(h => {
            h(userId, message)
          })
        }
      } catch (e) {
        console.log(e)
      }
    } else if (dir === 'out') {
      try {
        let key = [userId, scope].join(':')
        if (clients[key]) { clients[key].send(message) }
      } catch (e) {
        console.log(e)
      }
    }
    // otherwise ignore
  })
  websocketServer.on('connection', handleConnection)
}

exports.on = function (scope, callback) {
  if (!namespaces[scope]) { namespaces[scope] = [callback] } else { namespaces[scope].push(callback) }
}
exports.send = function (userId, scope, data) {
  let channel = ['dispatch', 'out', userId, scope].join(':')
  if (typeof (data) === 'object') { data = JSON.stringify(data) } else if (typeof (data) !== 'string') { throw new Error('DispatcherError: Cannot send this type of message ' + typeof (data)) }
  pub.publish(channel, data)
}
exports.clients = clients

Это работает на localhost.

Пожалуйста, дайте мне знать, если мне нужно предоставить дополнительную информацию или код. Любая помощь по этому вопросу будет принята с благодарностью, спасибо заранее!

Answer 1

В коде, который вы разместили, есть много посторонней информации, поэтому трудно понять, что именно вы имеете в виду.

Однако, если я правильно понимаю, в настоящее время у вас есть несколько рабочих экземпляров dyno, подписывающихся на одни и те же каналы в некоторой сети пабов / подсетей. Если вы не хотите, чтобы все dyno подписывались на одни и те же каналы, вам нужно добавить некоторую логику, чтобы убедиться, что ваши каналы распределены по dynos.

Один простой способ сделать это - использовать что-то вроде логики, описанной в этого ответа .

В вашем случае вы можете использовать socket.userId в качестве ключа для распределения ваших каналов по dynos.