DocumentUpdaterController.coffee 3.1 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465
  1. logger = require "logger-sharelatex"
  2. settings = require 'settings-sharelatex'
  3. redis = require("redis-sharelatex")
  4. rclient = redis.createClient(settings.redis.documentupdater)
  5. SafeJsonParse = require "./SafeJsonParse"
  6. EventLogger = require "./EventLogger"
  7. metrics = require "metrics-sharelatex"
  8. MESSAGE_SIZE_LOG_LIMIT = 1024 * 1024 # 1Mb
  9. module.exports = DocumentUpdaterController =
  10. # DocumentUpdaterController is responsible for updates that come via Redis
  11. # Pub/Sub from the document updater.
  12. listenForUpdatesFromDocumentUpdater: (io) ->
  13. rclient.subscribe "applied-ops"
  14. rclient.on "message", (channel, message) ->
  15. metrics.inc "rclient", 0.001 # global event rate metric
  16. EventLogger.debugEvent(channel, message) if settings.debugEvents > 0
  17. DocumentUpdaterController._processMessageFromDocumentUpdater(io, channel, message)
  18. _processMessageFromDocumentUpdater: (io, channel, message) ->
  19. SafeJsonParse.parse message, (error, message) ->
  20. if error?
  21. logger.error {err: error, channel}, "error parsing JSON"
  22. return
  23. if message.op?
  24. if message._id?
  25. status = EventLogger.checkEventOrder("applied-ops", message._id, message)
  26. if status is 'duplicate'
  27. return # skip duplicate events
  28. DocumentUpdaterController._applyUpdateFromDocumentUpdater(io, message.doc_id, message.op)
  29. else if message.error?
  30. DocumentUpdaterController._processErrorFromDocumentUpdater(io, message.doc_id, message.error, message)
  31. else if message.health_check?
  32. logger.debug {message}, "got health check message in applied ops channel"
  33. _applyUpdateFromDocumentUpdater: (io, doc_id, update) ->
  34. clientList = io.sockets.clients(doc_id)
  35. # avoid unnecessary work if no clients are connected
  36. if clientList.length is 0
  37. return
  38. # send updates to clients
  39. logger.log doc_id: doc_id, version: update.v, source: update.meta?.source, socketIoClients: (client.id for client in clientList), "distributing updates to clients"
  40. seen = {}
  41. # send messages only to unique clients (due to duplicate entries in io.sockets.clients)
  42. for client in clientList when not seen[client.id]
  43. seen[client.id] = true
  44. if client.id == update.meta.source
  45. logger.log doc_id: doc_id, version: update.v, source: update.meta?.source, "distributing update to sender"
  46. client.emit "otUpdateApplied", v: update.v, doc: update.doc
  47. else if !update.dup # Duplicate ops should just be sent back to sending client for acknowledgement
  48. logger.log doc_id: doc_id, version: update.v, source: update.meta?.source, client_id: client.id, "distributing update to collaborator"
  49. client.emit "otUpdateApplied", update
  50. if Object.keys(seen).length < clientList.length
  51. metrics.inc "socket-io.duplicate-clients", 0.1
  52. logger.log doc_id: doc_id, socketIoClients: (client.id for client in clientList), "discarded duplicate clients"
  53. _processErrorFromDocumentUpdater: (io, doc_id, error, message) ->
  54. for client in io.sockets.clients(doc_id)
  55. logger.warn err: error, doc_id: doc_id, client_id: client.id, "error from document updater, disconnecting client"
  56. client.emit "otUpdateError", error, message
  57. client.disconnect()