Socket.io disconnect and redis

Question

I'm trying to make my socket.io app scale by switching from a regular data structure that holds all my data, to using redis along with cluster. However, I'm encountering some problems because at some point in the current implementation, I store the socket object along with other properties in this data structure data[socket.id].socket = socket because in my application sometimes I need to do data[someId].socket.disconnect() to manually disconnect the socket.

I understand I cannot store objects directly into redis, so I tried using JSON.stringify(socket) with no success, since socket is circular. Is there another way to disconnect a socket using only the id? That way I can store the id like this data[socket.id].id = socket.id and maybe call it like data[someId].id.disconnect() or something. So basically I'm looking for a way to disconnect a socket without having access to the actual socket object (I do have access to the io object).

Thank you all for your help.

Answer 1

From the documentation of socket.io-redis you need to use remoteDisconnect :

io.of('/').adapter.remoteDisconnect(socketId, true, (err) => {
            if (err) {
              console.log('remoteDisconnect err:', err);
            }
        });

Answer 2

Use nginx in front of nodejs, pm2 and socket.io-redis.

NGINX.conf

    server {
        server_name                         www.yoursite.io;
        listen                              443 ssl http2;
        listen                              [::]:443 ssl http2;

        location / {
            proxy_set_header                X-Real-IP $remote_addr;
            proxy_set_header                X-Forwarded-For $proxy_add_x_forwarded_for;
            proxy_set_header                Host $http_host;
            proxy_set_header                X-NginX-Proxy false;
            proxy_redirect                  off;
            proxy_http_version              1.1;
            proxy_set_header                Upgrade $http_upgrade;
            proxy_set_header                Connection "upgrade";
            proxy_pass                      http://[::1]:4000;
        }
   }

PM2 Run cluster mode, four instances...

pm2 start app.js -i 4

app.js

console.clear()
require('dotenv').config()
const express = require('express'),
  app = express(),
  Redis = require('ioredis')
if(process.env.debug === 'true')
  app.use(require('morgan')(':method :url :status :res[content-length] - :response-time ms'))
app.locals = Object.assign(app.locals, {
  sock_transports: ['websocket', 'xhr-polling'],
  sock_timeout: process.env.sock_timeout,
  title: process.env.title,
  meta_desc: process.env.meta_desc,
  app_url: ['https://', process.env.app_subdomain, '.', process.env.app_domain].join('')
})
app.set('functions', require('./lib/functions')(app))
app.set('view engine', 'hbs')
app.set('view cache', false)
app.engine('hbs', require('./lib/hbs')(require('express-handlebars')).engine)
app.use(express.json({
  type: [
    'json'
  ]
}), express.urlencoded({
  extended: true
}))
const redis = new Redis({
  path: process.env.redis_socket,
  db: 1,
  enableReadyCheck: true
})
console.time('Redis')
redis.on('ready', () => {
  console.timeEnd('Redis')
  app.set('redis', redis)
})
redis.on('error', err => {
  console.log('Redis: ' + app.get('colors').redBright(err))
  exitHandler()
})
function loadSessionMiddleware() {
  const session = require('express-session'),
    RedisSession = require('connect-redis')(session),
    client = new Redis({
      path: process.env.redis_socket,
      db: 5
    }),
    ua = require('useragent')
  ua(true)
  app.set('useragent', ua)
  app.set('session_vars', {
    secret: process.env.session_secret,
    name: process.env.session_name,
    store: new RedisSession({
      client
    }),
    rolling: true,
    saveUninitialized: true,
    unset: 'destroy',
    resave: true,
    proxy: true,
    logErrors: process.env.debug === 'true',
    cookie: {
      path: '/',
      domain: '.' + process.env.app_domain,
      secure: true,
      sameSite: true,
      httpOnly: true,
      expires: false,
      maxAge: 60000 * process.env.session_exp_mins,
    }
  })
  app.set('session', session(app.get('session_vars')))
  app.use(
    app.get('session'),
    require('./middleware')(app)
  )
  loadControllers()
}
function loadControllers() {
  require('fs').readdirSync('./controllers').filter(file => {
    return file.slice(-3) === '.js'
  }).forEach(file => {
    require('./controllers/' + file)(app)
  })
  app.get('*', (req, res) => {
    app.get('functions').show404(req, res)
  })
  initServer()
}
function initServer() {
  console.time('Server')
  const server = require('http').createServer(app)
  server.on('error', err => {
    console.err('express err: ' + err)
    app.get('functions').stringify(err)
  })
  server.listen(process.env.app_port)
  app.set('server', server)
  require('./websocket').listen(app, websocket => {
    console.timeEnd('Server')
    app.set('websocket', websocket)
    // www-data
    process.setuid(process.env.app_uid)
  })
}
console.time('Database')
require('./model').load(app, db => {
  console.timeEnd('Database')
  app.set('model', db)
  loadSessionMiddleware()
})
function exitHandler() {
  if(app.get('server'))
    app.get('server').close()
  if(app.get('redis'))
    app.get('redis').quit()
  if(app.get('mail'))
    app.get('mail').close()
  process.exit(0)
}
process.on('SIGINT SIGUSR1 SIGUSR2', () => {
  exitHandler()
})
process.stdin.resume()

websocket.js

var exports = {}
exports.listen = (app, cb) => {
  const websocket = require('socket.io')(app.get('server'), {
    transports: process.env.transports
  }),
  req = {}
  websocket.setMaxListeners(0)
  websocket.adapter(require('socket.io-redis')({
    path: process.env.redis_socket,
    key: 'socket_io',
    db: 2,
    enableOfflineQueue: true
  }))
  websocket.use((socket, next) => {
    app.get('session')(socket.request, socket.request.res || {}, next)
  })
  websocket.isAccountLocked = cb => {
    if(!req.session.user_id) {
      cb(false)
      return
    }
    if(isNaN(req.session.user_id)) {
      cb(false)
      return
    }
    app.get('model').users.get(req.session.user_id, user_rec => {
      if(!user_rec) {
        cb(false)
        return
      }
      if(user_rec.account_locked === 'yes') {
        websocket.showClient(client => {
          app.get('model').users.logout(req.session, () => {
            console.log(client + ' ' + app.get('colors').redBright('Account Locked'))
            cb(true)
          })
        })
        return
      }
      cb(false)
    })
  }
  websocket.showClient = cb => {
    var outp = []
    if(!req.session.user_id && !req.session.full_name)
      outp.push(req.session.ip)
    if(req.session.user_id) {
      outp.push('# ' + req.session.user_id)
      if(req.session.full_name)
        outp.push(' - ' + req.session.full_name)
    }
    cb(app.get('colors').black.bgWhite(outp.join('')))
  }
  websocket.on('connection', socket => {
    if(!socket.request.session)
      return
    req.session = socket.request.session
    socket.use((packet, next) => {
        websocket.isAccountLocked(locked => {
            if(locked)
                return
            var save_sess = false
            if(typeof(socket.handshake.headers['x-real-ip']) !== 'undefined') {
          if(socket.handshake.headers['x-real-ip'] !== req.session.ip) {
            req.session.ip = socket.handshake.headers['x-real-ip']
            save_sess = true
          }
        }
        var ua = app.get('useragent').parse(socket.handshake.headers['user-agent']).toString()
        if(ua !== req.session.useragent) {
          req.session.useragent = ua
          save_sess = true
        }
            websocket.of('/').adapter.remoteJoin(socket.id, req.session.id, err => {
                delete socket.rooms[socket.id]
                if(!save_sess) {
            next()
            return
          }
                req.session.save(() => {
                  next()
                })
            })
      })
    })
    socket.on('disconnecting', () => {
      websocket.of('/').adapter.remoteDisconnect(req.session.id, true, err => {
      })
    })
    socket.on('auth', sess_vars => {
      function setSess() {
        if(sess_vars.path)
          req.session.path = sess_vars.path
        if(sess_vars.search_query)
          req.session.search_query = sess_vars.search_query
        if(sess_vars.search_query_long)
          req.session.search_query_long = sess_vars.search_query_long
        if(sess_vars.dispensary_id)
          req.session.dispensary_id = sess_vars.dispensary_id
        if(sess_vars.city)
          req.session.city = sess_vars.city
        if(sess_vars.state)
          req.session.state = sess_vars.state
        if(sess_vars.zip)
          req.session.zip = sess_vars.zip
        if(sess_vars.country)
          req.session.country = sess_vars.country
        if(sess_vars.hash)
          req.session.hash = sess_vars.hash
        req.session.save(() => {
          websocket.to(req.session.id).emit('auth', sess)
          app.get('functions').showVisitor({
            session: sess
          }, {
            statusCode: 200
          })
        })
      }
      setSess()
    })
    socket.on('logout', () => {
      var sess_ip = req.session.ip,
        sess_id = req.session.id,
        sess_email = req.session.email
      app.get('model').users.logout(req.session, () => {
        websocket.showClient(client => {
          console.log(client + ' - Logged Out')
        })
      })
    })
  })
  cb(websocket)
}
module.exports = exports

Answer 3

It seems this is done already but not documented anywhere... io.sockets.clients(someId) gets the socket object regardless of on what instance it is called, so the only thing that is needed is to use io.sockets.clients(someId).disconnect() and will actually disconnect the client, regardless of the instance it is connected to. I was storing them on my own array without actually needing to.

Answer 4

I do the similar thing like this:

var connTable = {};
function onConnection(socket) {
  connTable[socket.id] = socket;

  socket.on("close", function(data, callback) {
    socket.disconnect();
    onDisconnect(socket.id);
  });


  socket.on('disconnect', function(){
    onDisconnect(socket.id);
  });
}

function manuallyDisconnect(socket_id) {
  connTable[socket_id].disconnect();
}

function onDisconnect() {
  //closing process, or something....
}