| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393 |
- define [
- "utils/EventEmitter"
- "ide/editor/ShareJsDoc"
- "ide/review-panel/RangesTracker"
- ], (EventEmitter, ShareJsDoc, RangesTracker) ->
- class Document extends EventEmitter
- @getDocument: (ide, doc_id) ->
- @openDocs ||= {}
- if !@openDocs[doc_id]?
- sl_console.log "[getDocument] Creating new document instance for #{doc_id}"
- @openDocs[doc_id] = new Document(ide, doc_id)
- else
- sl_console.log "[getDocument] Returning existing document instance for #{doc_id}"
- return @openDocs[doc_id]
- @hasUnsavedChanges: () ->
- for doc_id, doc of (@openDocs or {})
- return true if doc.hasBufferedOps()
- return false
- @flushAll: () ->
- for doc_id, doc of @openDocs
- doc.flush()
- constructor: (@ide, @doc_id) ->
- @connected = @ide.socket.socket.connected
- @joined = false
- @wantToBeJoined = false
- @_checkConsistency = _.bind(@_checkConsistency, @)
- @inconsistentCount = 0
- @_bindToEditorEvents()
- @_bindToSocketEvents()
- attachToAce: (@ace) ->
- @doc?.attachToAce(@ace)
- editorDoc = @ace.getSession().getDocument()
- editorDoc.on "change", @_checkConsistency
- @ide.$scope.$emit 'document:opened', @doc
- detachFromAce: () ->
- @doc?.detachFromAce()
- editorDoc = @ace?.getSession().getDocument()
- editorDoc?.off "change", @_checkConsistency
- @ide.$scope.$emit 'document:closed', @doc
-
- submitOp: (args...) -> @doc?.submitOp(args...)
- _checkConsistency: () ->
- # We've been seeing a lot of errors when I think there shouldn't be
- # any, which may be related to this check happening before the change is
- # applied. If we use a timeout, hopefully we can reduce this.
- setTimeout () =>
- editorValue = @ace?.getValue()
- sharejsValue = @doc?.getSnapshot()
- if editorValue != sharejsValue
- @inconsistentCount++
- else
- @inconsistentCount = 0
- if @inconsistentCount >= 3
- @_onError new Error("Editor text does not match server text")
- , 0
- getSnapshot: () ->
- @doc?.getSnapshot()
- getType: () ->
- @doc?.getType()
- getInflightOp: () ->
- @doc?.getInflightOp()
- getPendingOp: () ->
- @doc?.getPendingOp()
- getRecentAck: () ->
- @doc?.getRecentAck()
- getOpSize: (op) ->
- @doc?.getOpSize(op)
- hasBufferedOps: () ->
- @doc?.hasBufferedOps()
-
- setTrackingChanges: (track_changes) ->
- @doc.track_changes = track_changes
-
- getTrackingChanges: () ->
- !!@doc.track_changes
-
- setTrackChangesIdSeeds: (id_seeds) ->
- @doc.track_changes_id_seeds = id_seeds
- _bindToSocketEvents: () ->
- @_onUpdateAppliedHandler = (update) => @_onUpdateApplied(update)
- @ide.socket.on "otUpdateApplied", @_onUpdateAppliedHandler
- @_onErrorHandler = (error, update) => @_onError(error, update)
- @ide.socket.on "otUpdateError", @_onErrorHandler
- @_onDisconnectHandler = (error) => @_onDisconnect(error)
- @ide.socket.on "disconnect", @_onDisconnectHandler
- _bindToEditorEvents: () ->
- onReconnectHandler = (update) =>
- @_onReconnect(update)
- @_unsubscribeReconnectHandler = @ide.$scope.$on "project:joined", onReconnectHandler
- _unBindFromEditorEvents: () ->
- @_unsubscribeReconnectHandler()
- _unBindFromSocketEvents: () ->
- @ide.socket.removeListener "otUpdateApplied", @_onUpdateAppliedHandler
- @ide.socket.removeListener "otUpdateError", @_onErrorHandler
- @ide.socket.removeListener "disconnect", @_onDisconnectHandler
- leaveAndCleanUp: () ->
- @leave (error) =>
- @_cleanUp()
- join: (callback = (error) ->) ->
- @wantToBeJoined = true
- @_cancelLeave()
- if @connected
- return @_joinDoc callback
- else
- @_joinCallbacks ||= []
- @_joinCallbacks.push callback
- leave: (callback = (error) ->) ->
- @wantToBeJoined = false
- @_cancelJoin()
- if (@doc? and @doc.hasBufferedOps())
- sl_console.log "[leave] Doc has buffered ops, pushing callback for later"
- @_leaveCallbacks ||= []
- @_leaveCallbacks.push callback
- else if !@connected
- sl_console.log "[leave] Not connected, returning now"
- callback()
- else
- sl_console.log "[leave] Leaving now"
- @_leaveDoc(callback)
- flush: () ->
- @doc?.flushPendingOps()
- chaosMonkey: (line = 0, char = "a") ->
- orig = char
- copy = null
- pos = 0
- timer = () =>
- unless copy? and copy.length
- copy = orig.slice() + ' ' + new Date() + '\n'
- line += if Math.random() > 0.1 then 1 else -2
- line = 0 if line < 0
- pos = 0
- char = copy[0]
- copy = copy.slice(1)
- @ace.session.insert({row: line, column: pos}, char)
- pos += 1
- @_cm = setTimeout timer, 100 + if Math.random() < 0.1 then 1000 else 0
- @_cm = timer()
- clearChaosMonkey: () ->
- clearTimeout @_cm
- MAX_PENDING_OP_SIZE: 64 # pending ops bigger than this are always considered unsaved
- pollSavedStatus: () ->
- # returns false if doc has ops waiting to be acknowledged or
- # sent that haven't changed since the last time we checked.
- # Otherwise returns true.
- inflightOp = @getInflightOp()
- pendingOp = @getPendingOp()
- recentAck = @getRecentAck()
- pendingOpSize = pendingOp? && @getOpSize(pendingOp)
- if !inflightOp? and !pendingOp?
- # there's nothing going on, this is ok.
- saved = true
- sl_console.log "[pollSavedStatus] no inflight or pending ops"
- else if inflightOp? and inflightOp == @oldInflightOp
- # The same inflight op has been sitting unacked since we
- # last checked, this is bad.
- saved = false
- sl_console.log "[pollSavedStatus] inflight op is same as before"
- else if pendingOp? and recentAck && pendingOpSize < @MAX_PENDING_OP_SIZE
- # There is an op waiting to go to server but it is small and
- # within the flushDelay, this is ok for now.
- saved = true
- sl_console.log "[pollSavedStatus] pending op (small with recent ack) assume ok", pendingOp, pendingOpSize
- else
- # In any other situation, assume the document is unsaved.
- saved = false
- sl_console.log "[pollSavedStatus] assuming not saved (inflightOp?: #{inflightOp?}, pendingOp?: #{pendingOp?})"
- @oldInflightOp = inflightOp
- return saved
- _cancelLeave: () ->
- if @_leaveCallbacks?
- delete @_leaveCallbacks
- _cancelJoin: () ->
- if @_joinCallbacks?
- delete @_joinCallbacks
- _onUpdateApplied: (update) ->
- @ide.pushEvent "received-update",
- doc_id: @doc_id
- remote_doc_id: update?.doc
- wantToBeJoined: @wantToBeJoined
- update: update
- if window.disconnectOnAck? and Math.random() < window.disconnectOnAck
- sl_console.log "Disconnecting on ack", update
- window._ide.socket.socket.disconnect()
- # Pretend we never received the ack
- return
- if window.dropAcks? and Math.random() < window.dropAcks
- if !update.op? # Only drop our own acks, not collaborator updates
- sl_console.log "Simulating a lost ack", update
- return
- if update?.doc == @doc_id and @doc?
- @doc.processUpdateFromServer update
- if !@wantToBeJoined
- @leave()
- _onDisconnect: () ->
- sl_console.log '[onDisconnect] disconnecting'
- @connected = false
- @joined = false
- @doc?.updateConnectionState "disconnected"
- _onReconnect: () ->
- sl_console.log "[onReconnect] reconnected (joined project)"
- @ide.pushEvent "reconnected:afterJoinProject"
- @connected = true
- if @wantToBeJoined or @doc?.hasBufferedOps()
- sl_console.log "[onReconnect] Rejoining (wantToBeJoined: #{@wantToBeJoined} OR hasBufferedOps: #{@doc?.hasBufferedOps()})"
- @_joinDoc (error) =>
- return @_onError(error) if error?
- @doc.updateConnectionState "ok"
- @doc.flushPendingOps()
- @_callJoinCallbacks()
- _callJoinCallbacks: () ->
- for callback in @_joinCallbacks or []
- callback()
- delete @_joinCallbacks
- _joinDoc: (callback = (error) ->) ->
- if @doc?
- @ide.socket.emit 'joinDoc', @doc_id, @doc.getVersion(), { encodeRanges: true }, (error, docLines, version, updates, ranges) =>
- return callback(error) if error?
- @joined = true
- @doc.catchUp( updates )
- @_decodeRanges(ranges)
- @_catchUpRanges(ranges?.changes, ranges?.comments)
- callback()
- else
- @ide.socket.emit 'joinDoc', @doc_id, { encodeRanges: true }, (error, docLines, version, updates, ranges) =>
- return callback(error) if error?
- @joined = true
- @doc = new ShareJsDoc @doc_id, docLines, version, @ide.socket
- @_decodeRanges(ranges)
- @ranges = new RangesTracker(ranges?.changes, ranges?.comments)
- @_bindToShareJsDocEvents()
- callback()
- _decodeRanges: (ranges) ->
- decodeFromWebsockets = (text) -> decodeURIComponent(escape(text))
- try
- for change in ranges.changes or []
- change.op.i = decodeFromWebsockets(change.op.i) if change.op.i?
- change.op.d = decodeFromWebsockets(change.op.d) if change.op.d?
- for comment in ranges.comments or []
- comment.op.c = decodeFromWebsockets(comment.op.c) if comment.op.c?
- catch err
- console.log(err)
- _leaveDoc: (callback = (error) ->) ->
- sl_console.log '[_leaveDoc] Sending leaveDoc request'
- @ide.socket.emit 'leaveDoc', @doc_id, (error) =>
- return callback(error) if error?
- @joined = false
- for callback in @_leaveCallbacks or []
- sl_console.log '[_leaveDoc] Calling buffered callback', callback
- callback(error)
- delete @_leaveCallbacks
- callback(error)
- _cleanUp: () ->
- if Document.openDocs[@doc_id] == @
- sl_console.log "[_cleanUp] Removing self (#{@doc_id}) from in openDocs"
- delete Document.openDocs[@doc_id]
- else
- # It's possible that this instance has error, and the doc has been reloaded.
- # This creates a new instance in Document.openDoc with the same id. We shouldn't
- # clear it because it's not this instance.
- sl_console.log "[_cleanUp] New instance of (#{@doc_id}) created. Not removing"
- @_unBindFromEditorEvents()
- @_unBindFromSocketEvents()
- _bindToShareJsDocEvents: () ->
- @doc.on "error", (error, meta) => @_onError error, meta
- @doc.on "externalUpdate", (update) =>
- @ide.pushEvent "externalUpdate",
- doc_id: @doc_id
- @trigger "externalUpdate", update
- @doc.on "remoteop", (args...) =>
- @ide.pushEvent "remoteop",
- doc_id: @doc_id
- @trigger "remoteop", args...
- @doc.on "op:sent", (op) =>
- @ide.pushEvent "op:sent",
- doc_id: @doc_id
- op: op
- @trigger "op:sent"
- @doc.on "op:acknowledged", (op) =>
- @ide.pushEvent "op:acknowledged",
- doc_id: @doc_id
- op: op
- @ide.$scope.$emit "ide:opAcknowledged",
- doc_id: @doc_id
- op: op
- @trigger "op:acknowledged"
- @doc.on "op:timeout", (op) =>
- @ide.pushEvent "op:timeout",
- doc_id: @doc_id
- op: op
- @trigger "op:timeout"
- @_onError new Error("op timed out"), {op: op}
- @doc.on "flush", (inflightOp, pendingOp, version) =>
- @ide.pushEvent "flush",
- doc_id: @doc_id,
- inflightOp: inflightOp,
- pendingOp: pendingOp
- v: version
- @doc.on "change", (ops, oldSnapshot, msg) =>
- @_applyOpsToRanges(ops, oldSnapshot, msg)
- @doc.on "flipped_pending_to_inflight", () =>
- @trigger "flipped_pending_to_inflight"
- _onError: (error, meta = {}) ->
- meta.doc_id = @doc_id
- sl_console.log "ShareJS error", error, meta
- ga?('send', 'event', 'error', "shareJsError", "#{error.message} - #{@ide.socket.socket.transport.name}" )
- @doc?.clearInflightAndPendingOps()
- @trigger "error", error, meta
- # The clean up should run after the error is triggered because the error triggers a
- # disconnect. If we run the clean up first, we remove our event handlers and miss
- # the disconnect event, which means we try to leaveDoc when the connection comes back.
- # This could intefere with the new connection of a new instance of this document.
- @_cleanUp()
- _applyOpsToRanges: (ops = [], oldSnapshot, msg) ->
- track_changes_as = null
- remote_op = msg?
- if msg?.meta?.tc?
- old_id_seed = @ranges.getIdSeed()
- @ranges.setIdSeed(msg.meta.tc)
- if remote_op and msg.meta?.tc
- track_changes_as = msg.meta.user_id
- else if !remote_op and @track_changes_as?
- track_changes_as = @track_changes_as
- @ranges.track_changes = track_changes_as?
- for op in ops
- @ranges.applyOp op, { user_id: track_changes_as }
- if old_id_seed?
- @ranges.setIdSeed(old_id_seed)
- if remote_op
- # With remote ops, Ace hasn't been updated when we receive this op,
- # so defer updating track changes until it has
- setTimeout () => @emit "ranges:dirty"
- else
- @emit "ranges:dirty"
-
- _catchUpRanges: (changes = [], comments = []) ->
- # We've just been given the current server's ranges, but need to apply any local ops we have.
- # Reset to the server state then apply our local ops again.
- @emit "ranges:clear"
- @ranges.changes = changes
- @ranges.comments = comments
- @ranges.track_changes = @doc.track_changes
- for op in @doc.getInflightOp() or []
- @ranges.setIdSeed(@doc.track_changes_id_seeds.inflight)
- @ranges.applyOp(op, { user_id: @track_changes_as })
- for op in @doc.getPendingOp() or []
- @ranges.setIdSeed(@doc.track_changes_id_seeds.pending)
- @ranges.applyOp(op, { user_id: @track_changes_as })
- @emit "ranges:redraw"
|