Document.coffee 13 KB

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