// Generated by CoffeeScript 2.6.1 (function() { var fs, log, redis, unread_counts; fs = require('fs'); redis = require('redis'); log = require('./log.js'); unread_counts = (server) => { var ENV_DEV, ENV_DOCKER, ENV_PROD, REDIS_PORT, REDIS_SERVER, SECURE, active_connections, io, pub_client, redis_opts, setup_redis_client, sub_client; ENV_DEV = process.env.NODE_ENV === 'development' || process.env.NODE_ENV === 'debug'; ENV_PROD = process.env.NODE_ENV === 'production'; ENV_DOCKER = process.env.NODE_ENV === 'docker'; REDIS_SERVER = "db_redis"; if (ENV_DEV) { REDIS_SERVER = 'localhost'; } else if (ENV_PROD) { REDIS_SERVER = 'db-redis-pubsub.service.nyc1.consul'; } SECURE = !!process.env.NODE_SSL; REDIS_PORT = ENV_DOCKER ? 6579 : 6383; // client = redis.createClient 6379, REDIS_SERVER // RedisStore = require 'socket.io/lib/stores/redis' // rpub = redis.createClient 6379, REDIS_SERVER // rsub = redis.createClient 6379, REDIS_SERVER // rclient = redis.createClient 6379, REDIS_SERVER log.debug("Starting NewsBlur unread count server..."); if (!ENV_DEV && !process.env.NODE_ENV) { log.debug("Specify NODE_ENV="); return; } else if (ENV_DEV) { log.debug("Running as development server"); } else if (ENV_DOCKER) { log.debug("Running as docker server"); } else { log.debug("Running as production server"); } // Create Redis clients for Socket.IO adapter with improved configuration redis_opts = { host: REDIS_SERVER, port: REDIS_PORT, retry_strategy: function(options) { // Exponential backoff with a cap return Math.min(options.attempt * 100, 3000); }, connect_timeout: 10000 }; pub_client = redis.createClient(redis_opts); sub_client = redis.createClient(redis_opts); // Handle Redis adapter client errors pub_client.on("error", function(err) { return log.debug(`Redis Pub Error: ${err}`); }); sub_client.on("error", function(err) { return log.debug(`Redis Sub Error: ${err}`); }); pub_client.on("reconnecting", function(attempt) { return log.debug(`Redis Pub reconnecting... Attempt ${attempt}`); }); sub_client.on("reconnecting", function(attempt) { return log.debug(`Redis Sub reconnecting... Attempt ${attempt}`); }); io = require('socket.io')(server, { path: "/v3/socket.io", pingTimeout: 60000, // Increase ping timeout to 60 seconds pingInterval: 25000, // Send ping every 25 seconds connectTimeout: 45000, // Connection timeout transports: ['websocket'], // Prefer websocket transport adapter: require('@socket.io/redis-adapter').createAdapter(pub_client, sub_client) }); // Setup Redis error handling and reconnection setup_redis_client = function(socket, username) { var client; client = redis.createClient({ host: REDIS_SERVER, port: REDIS_PORT, retry_strategy: function(options) { return Math.min(options.attempt * 100, 3000); }, connect_timeout: 10000 }); client.on("error", (err) => { return log.info(username, `Redis Error: ${err}`); }); // Don't quit on error, let retry strategy handle it client.on("reconnecting", (attempt) => { return log.info(username, `Redis reconnecting... Attempt ${attempt}`); }); return client; }; // Track active connections by username for debugging active_connections = {}; // Log engine events for debugging io.engine.on('connection', function(socket) { return log.debug(`Engine connection established: ${socket.id}`); }); io.engine.on('close', function(socket) { return log.debug(`Engine connection closed: ${socket.id}`); }); io.on('connection', function(socket) { var ip, socket_id; ip = socket.handshake.headers['X-Forwarded-For'] || socket.handshake.address; socket_id = socket.id; log.debug(`Socket connected: ${socket_id} from ${ip}`); socket.conn.on('error', function(err) { return log.debug(`Socket ${socket_id} - connection error: ${err}`); }); socket.conn.on('close', function(reason) { return log.debug(`Socket ${socket_id} - connection closed: ${reason}`); }); socket.on('subscribe:feeds', (feeds, username1) => { var ref; this.feeds = feeds; this.username = username1; log.info(this.username, `Connecting (${this.feeds.length} feeds, ${ip}), (${io.engine.clientsCount} connected) ${SECURE ? "(SSL)" : ""}`); // Track connections by username for debugging active_connections[this.username] = active_connections[this.username] || []; active_connections[this.username].push(socket_id); log.debug(`${this.username} now has ${active_connections[this.username].length} active connections`); if (!this.username) { return; } socket.on("error", function(err) { return log.debug(`Error (socket): ${err}`); }); if ((ref = socket.subscribe) != null) { ref.quit(); } socket.subscribe = setup_redis_client(socket, this.username); socket.subscribe.on("connect", () => { var feeds_story; log.info(this.username, `Connected (${this.feeds.length} feeds, ${ip}), (${io.engine.clientsCount} connected) ${SECURE ? "(SSL)" : "(non-SSL)"}`); socket.subscribe.subscribe(this.feeds); feeds_story = this.feeds.map(function(f) { return `${f}:story`; }); socket.subscribe.subscribe(feeds_story); return socket.subscribe.subscribe(this.username); }); return socket.subscribe.on('message', (channel, message) => { var event_name; event_name = 'feed:update'; if (channel === this.username) { event_name = 'user:update'; } else if (channel.indexOf(':story') >= 0) { event_name = 'feed:story:new'; } log.info(this.username, `Update on ${channel}: ${event_name} - ${message}`); return socket.emit(event_name, channel, message); }); }); return socket.on('disconnect', (reason) => { var idx, ref, ref1; log.debug(`Socket ${socket_id} disconnected: ${reason}`); // Update connection tracking if (this.username && active_connections[this.username]) { idx = active_connections[this.username].indexOf(socket_id); if (idx > -1) { active_connections[this.username].splice(idx, 1); log.debug(`${this.username} now has ${active_connections[this.username].length} active connections`); } if (active_connections[this.username].length === 0) { delete active_connections[this.username]; } } if ((ref = socket.subscribe) != null) { ref.quit(); } return log.info(this.username, `Disconnect (${(ref1 = this.feeds) != null ? ref1.length : void 0} feeds, ${ip}), there are now ${io.engine.clientsCount} users. ${SECURE ? "(SSL)" : "(non-SSL)"}`); }); }); io.engine.on('connection_error', function(err) { return log.debug(`Connection Error: ${err.code} - ${err.message}`); }); io.sockets.on('error', function(err) { return log.debug(`Error (sockets): ${err}`); }); // Periodically log connection stats setInterval(function() { var total_connections, total_users; total_users = Object.keys(active_connections).length; total_connections = io.engine.clientsCount; return log.debug(`Connection stats: ${total_users} users with ${total_connections} total connections`); }, 60000); return io; }; exports.unread_counts = unread_counts; }).call(this);