<commit_before>import { promisifyAll } from 'bluebird'
import { createClient as createRedisClient } from 'redis'
import _ from 'lodash'
import IoServer from 'socket.io'
import redis_adapter from 'socket.io-redis'
import jwt from 'jsonwebtoken'

import { dbAdapter, LikeSerializer, PostSerializer, PubsubCommentSerializer } from './models'
import { load as configLoader } from '../config/config'


promisifyAll(jwt)

export default class PubsubListener {
  constructor(server, app) {
    this.app = app

    const config = configLoader()

    var redisPub = createRedisClient(config.redis.port, config.redis.host, config.redis.options)
      , redisSub = createRedisClient(config.redis.port, config.redis.host, _.extend(config.redis.options, { detect_buffers: true }))

    redisPub.on('error', function(err) { app.logger.error('redisPub error', err) })
    redisSub.on('error', function(err) { app.logger.error('redisSub error', err) })

    this.io = IoServer(server)
    this.io.adapter(redis_adapter({
      pubClient: redisPub,
      subClient: redisSub
    }))

    this.io.sockets.on('error', function(err) { app.logger.error('socket.io error', err) })
    this.io.sockets.on('connection', this.onConnect)

    var redisClient = createRedisClient(config.redis.port, config.redis.host, {})
    redisClient.on('error', function(err) { app.logger.error('redis error', err) })
    redisClient.subscribe('post:new', 'post:destroy', 'post:update',
      'comment:new', 'comment:destroy', 'comment:update',
      'like:new', 'like:remove', 'post:hide', 'post:unhide')

    redisClient.on('message', this.onRedisMessage)
  }

  onConnect = async (socket) => {
    let authToken = socket.handshake.query.token
    const config = configLoader()
    let secret = config.secret
    let logger = this.app.logger

    try {
      let decoded = await jwt.verifyAsync(authToken, secret)
      socket.user = await dbAdapter.getUserById(decoded.userId)
    } catch(e) {
      socket.user = { id: null }
    }

    socket.on('error', (e) => {
      this.app.logger.error('socket.io socket error', e);
    });

    socket.on('subscribe', (data) => {
      if (!data) {
        this.app.logger.warn('socket.io got "subscribe" request without data');
        return;
      }

      for (let channel of Object.keys(data)) {
        if (data[channel]) {
          data[channel].forEach(function(id) {
            if (id) {
              logger.info('User has subscribed to ' + id + ' ' + channel)

              socket.join(channel + ':' + id)
            }
          })
        }
      }
    })

    socket.on('unsubscribe', (data) => {
      if (!data) {
        this.app.logger.warn('socket.io got "unsubscribe" request without data');
        return;
      }

      for (let channel of Object.keys(data)) {
        if (data[channel]) {
          data[channel].forEach(function(id) {
            if (id) {
              logger.info('User has unsubscribed from ' + id + ' ' + channel)

              socket.leave(channel + ':' + id)
            }
          })
        }
      }
    })
  }

  onRedisMessage = async (channel, msg) => {
    const messageRoutes = {
      'post:new':         this.onPostNew,
      'post:update':      this.onPostUpdate,
      'post:destroy':     this.onPostDestroy,
      'post:hide':        this.onPostHide,
      'post:unhide':      this.onPostUnhide,

      'comment:new':      this.onCommentNew,
      'comment:update':   this.onCommentUpdate,
      'comment:destroy':  this.onCommentDestroy,

      'like:new':         this.onLikeNew,
      'like:remove':      this.onLikeRemove
    }
<commit_msg>Handle more cases of bogus subscribe-requests<commit_after>import { promisifyAll } from 'bluebird'
import { createClient as createRedisClient } from 'redis'
import { extend, isArray, isPlainObject } from 'lodash'
import IoServer from 'socket.io'
import redis_adapter from 'socket.io-redis'
import jwt from 'jsonwebtoken'

import { dbAdapter, LikeSerializer, PostSerializer, PubsubCommentSerializer } from './models'
import { load as configLoader } from '../config/config'


promisifyAll(jwt)

export default class PubsubListener {
  constructor(server, app) {
    this.app = app

    const config = configLoader()

    var redisPub = createRedisClient(config.redis.port, config.redis.host, config.redis.options)
      , redisSub = createRedisClient(config.redis.port, config.redis.host, extend(config.redis.options, { detect_buffers: true }))

    redisPub.on('error', function(err) { app.logger.error('redisPub error', err) })
    redisSub.on('error', function(err) { app.logger.error('redisSub error', err) })

    this.io = IoServer(server)
    this.io.adapter(redis_adapter({
      pubClient: redisPub,
      subClient: redisSub
    }))

    this.io.sockets.on('error', function(err) { app.logger.error('socket.io error', err) })
    this.io.sockets.on('connection', this.onConnect)

    var redisClient = createRedisClient(config.redis.port, config.redis.host, {})
    redisClient.on('error', function(err) { app.logger.error('redis error', err) })
    redisClient.subscribe('post:new', 'post:destroy', 'post:update',
      'comment:new', 'comment:destroy', 'comment:update',
      'like:new', 'like:remove', 'post:hide', 'post:unhide')

    redisClient.on('message', this.onRedisMessage)
  }

  onConnect = async (socket) => {
    let authToken = socket.handshake.query.token
    const config = configLoader()
    let secret = config.secret
    let logger = this.app.logger

    try {
      let decoded = await jwt.verifyAsync(authToken, secret)
      socket.user = await dbAdapter.getUserById(decoded.userId)
    } catch(e) {
      socket.user = { id: null }
    }

    socket.on('error', (e) => {
      logger.error('socket.io socket error', e);
    });

    socket.on('subscribe', (data) => {
      if (!isPlainObject(data)) {
        logger.warn('socket.io got "subscribe" request without data');
        return;
      }


      for (let channel of Object.keys(data)) {
        if (!isArray(data[channel])) {
          logger.warn('socket.io got "unsubscribe" request with bogus list of channels');
          continue;
        }

        data[channel].filter(Boolean).forEach((id) => {
          socket.join(channel + ':' + id)
          logger.info(`User has subscribed to ${id} ${channel}`)
        })
      }
    })

    socket.on('unsubscribe', (data) => {
      if (!isPlainObject(data)) {
        logger.warn('socket.io got "unsubscribe" request without data');
        return;
      }

      for (let channel of Object.keys(data)) {
        if (!isArray(data[channel])) {
          logger.warn('socket.io got "unsubscribe" request with bogus list of channels');
          continue;
        }

        data[channel].filter(Boolean).forEach((id) => {
          socket.leave(channel + ':' + id)
          logger.info(`User has unsubscribed from ${id} ${channel}`)
        })
      }
    })
  }

  onRedisMessage = async (channel, msg) => {
    const messageRoutes = {
      'post:new':         this.onPostNew,
      'post:update':      this.onPostUpdate,
      'post:destroy':     this.onPostDestroy,
      'post:hide':        this.onPostHide,
      'post:unhide':      this.onPostUnhide,

      'comment:new':      this.onCommentNew,
      'comment:update':   this.onCommentUpdate,
      'comment:destroy':  this.onCommentDestroy,

      'like:new':         this.onLikeNew,
      'like:remove':      this.onLikeRemove
    }

{'fraction_non_alphanumeric': 0.14272300469483568, 'fraction_numerical': 0.0, 'mean_word_length': 2.8021417644059152, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1839205', 'n_tokens_mistral': 2352, 'n_tokens_neox': 2309, 'n_words': 472}