Document.coffee 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370
  1. define [
  2. "utils/EventEmitter"
  3. "ide/editor/ShareJsDoc"
  4. "ide/review-panel/RangesTracker"
  5. ], (EventEmitter, ShareJsDoc, RangesTracker) ->
  6. class Document extends EventEmitter
  7. @getDocument: (ide, doc_id) ->
  8. @openDocs ||= {}
  9. if !@openDocs[doc_id]?
  10. sl_console.log "[getDocument] Creating new document instance for #{doc_id}"
  11. @openDocs[doc_id] = new Document(ide, doc_id)
  12. else
  13. sl_console.log "[getDocument] Returning existing document instance for #{doc_id}"
  14. return @openDocs[doc_id]
  15. @hasUnsavedChanges: () ->
  16. for doc_id, doc of (@openDocs or {})
  17. return true if doc.hasBufferedOps()
  18. return false
  19. @flushAll: () ->
  20. for doc_id, doc of @openDocs
  21. doc.flush()
  22. constructor: (@ide, @doc_id) ->
  23. @connected = @ide.socket.socket.connected
  24. @joined = false
  25. @wantToBeJoined = false
  26. @_checkConsistency = _.bind(@_checkConsistency, @)
  27. @inconsistentCount = 0
  28. @_bindToEditorEvents()
  29. @_bindToSocketEvents()
  30. attachToAce: (@ace) ->
  31. @doc?.attachToAce(@ace)
  32. editorDoc = @ace.getSession().getDocument()
  33. editorDoc.on "change", @_checkConsistency
  34. detachFromAce: () ->
  35. @doc?.detachFromAce()
  36. editorDoc = @ace?.getSession().getDocument()
  37. editorDoc?.off "change", @_checkConsistency
  38. @ide.$scope.$emit 'document:closed', @doc
  39. submitOp: (args...) -> @doc?.submitOp(args...)
  40. _checkConsistency: () ->
  41. # We've been seeing a lot of errors when I think there shouldn't be
  42. # any, which may be related to this check happening before the change is
  43. # applied. If we use a timeout, hopefully we can reduce this.
  44. setTimeout () =>
  45. editorValue = @ace?.getValue()
  46. sharejsValue = @doc?.getSnapshot()
  47. if editorValue != sharejsValue
  48. @inconsistentCount++
  49. else
  50. @inconsistentCount = 0
  51. if @inconsistentCount >= 3
  52. @_onError new Error("Editor text does not match server text")
  53. , 0
  54. getSnapshot: () ->
  55. @doc?.getSnapshot()
  56. getType: () ->
  57. @doc?.getType()
  58. getInflightOp: () ->
  59. @doc?.getInflightOp()
  60. getPendingOp: () ->
  61. @doc?.getPendingOp()
  62. getRecentAck: () ->
  63. @doc?.getRecentAck()
  64. getOpSize: (op) ->
  65. @doc?.getOpSize(op)
  66. hasBufferedOps: () ->
  67. @doc?.hasBufferedOps()
  68. setTrackingChanges: (track_changes) ->
  69. @doc.track_changes = track_changes
  70. getTrackingChanges: () ->
  71. !!@doc.track_changes
  72. setTrackChangesIdSeeds: (id_seeds) ->
  73. @doc.track_changes_id_seeds = id_seeds
  74. _bindToSocketEvents: () ->
  75. @_onUpdateAppliedHandler = (update) => @_onUpdateApplied(update)
  76. @ide.socket.on "otUpdateApplied", @_onUpdateAppliedHandler
  77. @_onErrorHandler = (error, update) => @_onError(error, update)
  78. @ide.socket.on "otUpdateError", @_onErrorHandler
  79. @_onDisconnectHandler = (error) => @_onDisconnect(error)
  80. @ide.socket.on "disconnect", @_onDisconnectHandler
  81. _bindToEditorEvents: () ->
  82. onReconnectHandler = (update) =>
  83. @_onReconnect(update)
  84. @_unsubscribeReconnectHandler = @ide.$scope.$on "project:joined", onReconnectHandler
  85. _unBindFromEditorEvents: () ->
  86. @_unsubscribeReconnectHandler()
  87. _unBindFromSocketEvents: () ->
  88. @ide.socket.removeListener "otUpdateApplied", @_onUpdateAppliedHandler
  89. @ide.socket.removeListener "otUpdateError", @_onErrorHandler
  90. @ide.socket.removeListener "disconnect", @_onDisconnectHandler
  91. leaveAndCleanUp: () ->
  92. @leave (error) =>
  93. @_cleanUp()
  94. join: (callback = (error) ->) ->
  95. @wantToBeJoined = true
  96. @_cancelLeave()
  97. if @connected
  98. return @_joinDoc callback
  99. else
  100. @_joinCallbacks ||= []
  101. @_joinCallbacks.push callback
  102. leave: (callback = (error) ->) ->
  103. @wantToBeJoined = false
  104. @_cancelJoin()
  105. if (@doc? and @doc.hasBufferedOps())
  106. sl_console.log "[leave] Doc has buffered ops, pushing callback for later"
  107. @_leaveCallbacks ||= []
  108. @_leaveCallbacks.push callback
  109. else if !@connected
  110. sl_console.log "[leave] Not connected, returning now"
  111. callback()
  112. else
  113. sl_console.log "[leave] Leaving now"
  114. @_leaveDoc(callback)
  115. flush: () ->
  116. @doc?.flushPendingOps()
  117. chaosMonkey: (line = 0, char = "a") ->
  118. orig = char
  119. copy = null
  120. pos = 0
  121. timer = () =>
  122. unless copy? and copy.length
  123. copy = orig.slice() + ' ' + new Date() + '\n'
  124. line += if Math.random() > 0.1 then 1 else -2
  125. line = 0 if line < 0
  126. pos = 0
  127. char = copy[0]
  128. copy = copy.slice(1)
  129. @ace.session.insert({row: line, column: pos}, char)
  130. pos += 1
  131. @_cm = setTimeout timer, 100 + if Math.random() < 0.1 then 1000 else 0
  132. @_cm = timer()
  133. clearChaosMonkey: () ->
  134. clearTimeout @_cm
  135. MAX_PENDING_OP_SIZE: 30 # pending ops bigger than this are always considered unsaved
  136. pollSavedStatus: () ->
  137. # returns false if doc has ops waiting to be acknowledged or
  138. # sent that haven't changed since the last time we checked.
  139. # Otherwise returns true.
  140. inflightOp = @getInflightOp()
  141. pendingOp = @getPendingOp()
  142. recentAck = @getRecentAck()
  143. pendingOpSize = pendingOp? && @getOpSize(pendingOp)
  144. if !inflightOp? and !pendingOp?
  145. # there's nothing going on, this is ok.
  146. saved = true
  147. sl_console.log "[pollSavedStatus] no inflight or pending ops"
  148. else if inflightOp? and inflightOp == @oldInflightOp
  149. # The same inflight op has been sitting unacked since we
  150. # last checked, this is bad.
  151. saved = false
  152. sl_console.log "[pollSavedStatus] inflight op is same as before"
  153. else if pendingOp? and recentAck && pendingOpSize < @MAX_PENDING_OP_SIZE
  154. # There is an op waiting to go to server but it is small and
  155. # within the flushDelay, this is ok for now.
  156. saved = true
  157. sl_console.log "[pollSavedStatus] pending op (small with recent ack) assume ok", pendingOp, pendingOpSize
  158. else
  159. # In any other situation, assume the document is unsaved.
  160. saved = false
  161. sl_console.log "[pollSavedStatus] assuming not saved (inflightOp?: #{inflightOp?}, pendingOp?: #{pendingOp?})"
  162. @oldInflightOp = inflightOp
  163. return saved
  164. _cancelLeave: () ->
  165. if @_leaveCallbacks?
  166. delete @_leaveCallbacks
  167. _cancelJoin: () ->
  168. if @_joinCallbacks?
  169. delete @_joinCallbacks
  170. _onUpdateApplied: (update) ->
  171. @ide.pushEvent "received-update",
  172. doc_id: @doc_id
  173. remote_doc_id: update?.doc
  174. wantToBeJoined: @wantToBeJoined
  175. update: update
  176. if window.disconnectOnAck? and Math.random() < window.disconnectOnAck
  177. sl_console.log "Disconnecting on ack", update
  178. window._ide.socket.socket.disconnect()
  179. # Pretend we never received the ack
  180. return
  181. if window.dropAcks? and Math.random() < window.dropAcks
  182. if !update.op? # Only drop our own acks, not collaborator updates
  183. sl_console.log "Simulating a lost ack", update
  184. return
  185. if update?.doc == @doc_id and @doc?
  186. @doc.processUpdateFromServer update
  187. if !@wantToBeJoined
  188. @leave()
  189. _onDisconnect: () ->
  190. sl_console.log '[onDisconnect] disconnecting'
  191. @connected = false
  192. @joined = false
  193. @doc?.updateConnectionState "disconnected"
  194. _onReconnect: () ->
  195. sl_console.log "[onReconnect] reconnected (joined project)"
  196. @ide.pushEvent "reconnected:afterJoinProject"
  197. @connected = true
  198. if @wantToBeJoined or @doc?.hasBufferedOps()
  199. sl_console.log "[onReconnect] Rejoining (wantToBeJoined: #{@wantToBeJoined} OR hasBufferedOps: #{@doc?.hasBufferedOps()})"
  200. @_joinDoc (error) =>
  201. return @_onError(error) if error?
  202. @doc.updateConnectionState "ok"
  203. @doc.flushPendingOps()
  204. @_callJoinCallbacks()
  205. _callJoinCallbacks: () ->
  206. for callback in @_joinCallbacks or []
  207. callback()
  208. delete @_joinCallbacks
  209. _joinDoc: (callback = (error) ->) ->
  210. if @doc?
  211. @ide.socket.emit 'joinDoc', @doc_id, @doc.getVersion(), (error, docLines, version, updates, ranges) =>
  212. return callback(error) if error?
  213. @joined = true
  214. @doc.catchUp( updates )
  215. @_catchUpRanges( ranges?.changes, ranges?.comments )
  216. callback()
  217. else
  218. @ide.socket.emit 'joinDoc', @doc_id, (error, docLines, version, updates, ranges) =>
  219. return callback(error) if error?
  220. @joined = true
  221. @doc = new ShareJsDoc @doc_id, docLines, version, @ide.socket
  222. @ranges = new RangesTracker(ranges?.changes, ranges?.comments)
  223. @_bindToShareJsDocEvents()
  224. callback()
  225. _leaveDoc: (callback = (error) ->) ->
  226. sl_console.log '[_leaveDoc] Sending leaveDoc request'
  227. @ide.socket.emit 'leaveDoc', @doc_id, (error) =>
  228. return callback(error) if error?
  229. @joined = false
  230. for callback in @_leaveCallbacks or []
  231. sl_console.log '[_leaveDoc] Calling buffered callback', callback
  232. callback(error)
  233. delete @_leaveCallbacks
  234. callback(error)
  235. _cleanUp: () ->
  236. if Document.openDocs[@doc_id] == @
  237. sl_console.log "[_cleanUp] Removing self (#{@doc_id}) from in openDocs"
  238. delete Document.openDocs[@doc_id]
  239. else
  240. # It's possible that this instance has error, and the doc has been reloaded.
  241. # This creates a new instance in Document.openDoc with the same id. We shouldn't
  242. # clear it because it's not this instance.
  243. sl_console.log "[_cleanUp] New instance of (#{@doc_id}) created. Not removing"
  244. @_unBindFromEditorEvents()
  245. @_unBindFromSocketEvents()
  246. _bindToShareJsDocEvents: () ->
  247. @doc.on "error", (error, meta) => @_onError error, meta
  248. @doc.on "externalUpdate", (update) =>
  249. @ide.pushEvent "externalUpdate",
  250. doc_id: @doc_id
  251. @trigger "externalUpdate", update
  252. @doc.on "remoteop", (args...) =>
  253. @ide.pushEvent "remoteop",
  254. doc_id: @doc_id
  255. @trigger "remoteop", args...
  256. @doc.on "op:sent", (op) =>
  257. @ide.pushEvent "op:sent",
  258. doc_id: @doc_id
  259. op: op
  260. @trigger "op:sent"
  261. @doc.on "op:acknowledged", (op) =>
  262. @ide.pushEvent "op:acknowledged",
  263. doc_id: @doc_id
  264. op: op
  265. @trigger "op:acknowledged"
  266. @doc.on "op:timeout", (op) =>
  267. @ide.pushEvent "op:timeout",
  268. doc_id: @doc_id
  269. op: op
  270. @trigger "op:timeout"
  271. @_onError new Error("op timed out"), {op: op}
  272. @doc.on "flush", (inflightOp, pendingOp, version) =>
  273. @ide.pushEvent "flush",
  274. doc_id: @doc_id,
  275. inflightOp: inflightOp,
  276. pendingOp: pendingOp
  277. v: version
  278. @doc.on "change", (ops, oldSnapshot, msg) =>
  279. @_applyOpsToRanges(ops, oldSnapshot, msg)
  280. @doc.on "flipped_pending_to_inflight", () =>
  281. @trigger "flipped_pending_to_inflight"
  282. _onError: (error, meta = {}) ->
  283. meta.doc_id = @doc_id
  284. sl_console.log "ShareJS error", error, meta
  285. ga?('send', 'event', 'error', "shareJsError", "#{error.message} - #{@ide.socket.socket.transport.name}" )
  286. @doc?.clearInflightAndPendingOps()
  287. @trigger "error", error, meta
  288. # The clean up should run after the error is triggered because the error triggers a
  289. # disconnect. If we run the clean up first, we remove our event handlers and miss
  290. # the disconnect event, which means we try to leaveDoc when the connection comes back.
  291. # This could intefere with the new connection of a new instance of this document.
  292. @_cleanUp()
  293. _applyOpsToRanges: (ops = [], oldSnapshot, msg) ->
  294. track_changes_as = null
  295. remote_op = msg?
  296. if msg?.meta?.tc?
  297. old_id_seed = @ranges.getIdSeed()
  298. @ranges.setIdSeed(msg.meta.tc)
  299. if remote_op and msg.meta?.tc
  300. track_changes_as = msg.meta.user_id
  301. else if !remote_op and @track_changes_as?
  302. track_changes_as = @track_changes_as
  303. @ranges.track_changes = track_changes_as?
  304. for op in ops
  305. @ranges.applyOp op, { user_id: track_changes_as }
  306. if old_id_seed?
  307. @ranges.setIdSeed(old_id_seed)
  308. _catchUpRanges: (changes = [], comments = []) ->
  309. # We've just been given the current server's ranges, but need to apply any local ops we have.
  310. # Reset to the server state then apply our local ops again.
  311. @ranges.emit "clear"
  312. @ranges.changes = changes
  313. @ranges.comments = comments
  314. @ranges.track_changes = @doc.track_changes
  315. for op in @doc.getInflightOp() or []
  316. @ranges.setIdSeed(@doc.track_changes_id_seeds.inflight)
  317. @ranges.applyOp(op, { user_id: @track_changes_as })
  318. for op in @doc.getPendingOp() or []
  319. @ranges.setIdSeed(@doc.track_changes_id_seeds.pending)
  320. @ranges.applyOp(op, { user_id: @track_changes_as })
  321. @ranges.emit "redraw"