overleaf/services/real-time/app/coffee/WebsocketLoadBalancer.coffee
2019-07-18 12:55:23 +01:00

72 lines
3.1 KiB
CoffeeScript

Settings = require 'settings-sharelatex'
logger = require 'logger-sharelatex'
RedisClientManager = require "./RedisClientManager"
SafeJsonParse = require "./SafeJsonParse"
EventLogger = require "./EventLogger"
HealthCheckManager = require "./HealthCheckManager"
RoomManager = require "./RoomManager"
ChannelManager = require "./ChannelManager"
module.exports = WebsocketLoadBalancer =
rclientPubList: RedisClientManager.createClientList(Settings.redis.pubsub)
rclientSubList: RedisClientManager.createClientList(Settings.redis.pubsub)
emitToRoom: (room_id, message, payload...) ->
if !room_id?
logger.warn {message, payload}, "no room_id provided, ignoring emitToRoom"
return
data = JSON.stringify
room_id: room_id
message: message
payload: payload
logger.log {room_id, message, payload, length: data.length}, "emitting to room"
for rclientPub in @rclientPubList
ChannelManager.publish rclientPub, "editor-events", room_id, data
emitToAll: (message, payload...) ->
@emitToRoom "all", message, payload...
listenForEditorEvents: (io) ->
logger.log {rclients: @rclientPubList.length}, "publishing editor events"
logger.log {rclients: @rclientSubList.length}, "listening for editor events"
for rclientSub in @rclientSubList
rclientSub.subscribe "editor-events"
rclientSub.on "message", (channel, message) ->
EventLogger.debugEvent(channel, message) if Settings.debugEvents > 0
WebsocketLoadBalancer._processEditorEvent io, channel, message
for rclientSub in @rclientSubList
@handleRoomUpdates(rclientSub)
handleRoomUpdates: (rclientSub) ->
roomEvents = RoomManager.eventSource()
roomEvents.on 'project-active', (project_id) ->
ChannelManager.subscribe rclientSub, "editor-events", project_id
roomEvents.on 'project-empty', (project_id) ->
ChannelManager.unsubscribe rclientSub, "editor-events", project_id
_processEditorEvent: (io, channel, message) ->
SafeJsonParse.parse message, (error, message) ->
if error?
logger.error {err: error, channel}, "error parsing JSON"
return
if message.room_id == "all"
io.sockets.emit(message.message, message.payload...)
else if message.room_id?
if message._id? && Settings.checkEventOrder
status = EventLogger.checkEventOrder("editor-events", message._id, message)
if status is "duplicate"
return # skip duplicate events
# send messages only to unique clients (due to duplicate entries in io.sockets.clients)
clientList = io.sockets.clients(message.room_id)
# avoid unnecessary work if no clients are connected
return if clientList.length is 0
logger.log {channel:channel, message: message.message, room_id: message.room_id, message_id: message._id, socketIoClients: (client.id for client in clientList)}, "distributing event to clients"
seen = {}
for client in clientList when not seen[client.id]
seen[client.id] = true
client.emit(message.message, message.payload...)
else if message.health_check?
logger.debug {message}, "got health check message in editor events channel"
HealthCheckManager.check channel, message.key