|
|
@@ -0,0 +1,603 @@
|
|
|
+# The model of all the ops. Responsible for applying & transforming remote deltas
|
|
|
+# and managing the storage layer.
|
|
|
+#
|
|
|
+# Actual storage is handled by the database wrappers in db/*, wrapped by DocCache
|
|
|
+
|
|
|
+{EventEmitter} = require 'events'
|
|
|
+
|
|
|
+queue = require './syncqueue'
|
|
|
+types = require '../types'
|
|
|
+
|
|
|
+isArray = (o) -> Object.prototype.toString.call(o) == '[object Array]'
|
|
|
+
|
|
|
+# This constructor creates a new Model object. There will be one model object
|
|
|
+# per server context.
|
|
|
+#
|
|
|
+# The model object is responsible for a lot of things:
|
|
|
+#
|
|
|
+# - It manages the interactions with the database
|
|
|
+# - It maintains (in memory) a set of all active documents
|
|
|
+# - It calls out to the OT functions when necessary
|
|
|
+#
|
|
|
+# The model is an event emitter. It emits the following events:
|
|
|
+#
|
|
|
+# create(docName, data): A document has been created with the specified name & data
|
|
|
+module.exports = Model = (db, options) ->
|
|
|
+ # db can be null if the user doesn't want persistance.
|
|
|
+
|
|
|
+ return new Model(db, options) if !(this instanceof Model)
|
|
|
+
|
|
|
+ model = this
|
|
|
+
|
|
|
+ options ?= {}
|
|
|
+
|
|
|
+ # This is a cache of 'live' documents.
|
|
|
+ #
|
|
|
+ # The cache is a map from docName -> {
|
|
|
+ # ops:[{op, meta}]
|
|
|
+ # snapshot
|
|
|
+ # type
|
|
|
+ # v
|
|
|
+ # meta
|
|
|
+ # eventEmitter
|
|
|
+ # reapTimer
|
|
|
+ # committedVersion: v
|
|
|
+ # snapshotWriteLock: bool to make sure writeSnapshot isn't re-entrant
|
|
|
+ # dbMeta: database specific data
|
|
|
+ # opQueue: syncQueue for processing ops
|
|
|
+ # }
|
|
|
+ #
|
|
|
+ # The ops list contains the document's last options.numCachedOps ops. (Or all
|
|
|
+ # of them if we're using a memory store).
|
|
|
+ #
|
|
|
+ # Documents are stored in this set so long as the document has been accessed in
|
|
|
+ # the last few seconds (options.reapTime) OR at least one client has the document
|
|
|
+ # open. I don't know if I should keep open (but not being edited) documents live -
|
|
|
+ # maybe if a client has a document open but the document isn't being edited, I should
|
|
|
+ # flush it from the cache.
|
|
|
+ #
|
|
|
+ # In any case, the API to model is designed such that if we want to change that later
|
|
|
+ # it should be pretty easy to do so without any external-to-the-model code changes.
|
|
|
+ docs = {}
|
|
|
+
|
|
|
+ # This is a map from docName -> [callback]. It is used when a document hasn't been
|
|
|
+ # cached and multiple getSnapshot() / getVersion() requests come in. All requests
|
|
|
+ # are added to the callback list and called when db.getSnapshot() returns.
|
|
|
+ #
|
|
|
+ # callback(error, snapshot data)
|
|
|
+ awaitingGetSnapshot = {}
|
|
|
+
|
|
|
+ # The time that documents which no clients have open will stay in the cache.
|
|
|
+ # Should be > 0.
|
|
|
+ options.reapTime ?= 3000
|
|
|
+
|
|
|
+ # The number of operations the cache holds before reusing the space
|
|
|
+ options.numCachedOps ?= 10
|
|
|
+
|
|
|
+ # This option forces documents to be reaped, even when there's no database backend.
|
|
|
+ # This is useful when you don't care about persistance and don't want to gradually
|
|
|
+ # fill memory.
|
|
|
+ #
|
|
|
+ # You might want to set reapTime to a day or something.
|
|
|
+ options.forceReaping ?= false
|
|
|
+
|
|
|
+ # Until I come up with a better strategy, we'll save a copy of the document snapshot
|
|
|
+ # to the database every ~20 submitted ops.
|
|
|
+ options.opsBeforeCommit ?= 20
|
|
|
+
|
|
|
+ # It takes some processing time to transform client ops. The server will punt ops back to the
|
|
|
+ # client to transform if they're too old.
|
|
|
+ options.maximumAge ?= 40
|
|
|
+
|
|
|
+ # **** Cache API methods
|
|
|
+
|
|
|
+ # Its important that all ops are applied in order. This helper method creates the op submission queue
|
|
|
+ # for a single document. This contains the logic for transforming & applying ops.
|
|
|
+ makeOpQueue = (docName, doc) -> queue (opData, callback) ->
|
|
|
+ return callback 'Version missing' unless opData.v >= 0
|
|
|
+ return callback 'Op at future version' if opData.v > doc.v
|
|
|
+
|
|
|
+ # Punt the transforming work back to the client if the op is too old.
|
|
|
+ return callback 'Op too old' if opData.v + options.maximumAge < doc.v
|
|
|
+
|
|
|
+ opData.meta ||= {}
|
|
|
+ opData.meta.ts = Date.now()
|
|
|
+
|
|
|
+ # We'll need to transform the op to the current version of the document. This
|
|
|
+ # calls the callback immediately if opVersion == doc.v.
|
|
|
+ getOps docName, opData.v, doc.v, (error, ops) ->
|
|
|
+ return callback error if error
|
|
|
+
|
|
|
+ unless doc.v - opData.v == ops.length
|
|
|
+ # This should never happen. It indicates that we didn't get all the ops we
|
|
|
+ # asked for. Its important that the submitted op is correctly transformed.
|
|
|
+ console.error "Could not get old ops in model for document #{docName}"
|
|
|
+ console.error "Expected ops #{opData.v} to #{doc.v} and got #{ops.length} ops"
|
|
|
+ return callback 'Internal error'
|
|
|
+
|
|
|
+ if ops.length > 0
|
|
|
+ try
|
|
|
+ # If there's enough ops, it might be worth spinning this out into a webworker thread.
|
|
|
+ for oldOp in ops
|
|
|
+ # Dup detection works by sending the id(s) the op has been submitted with previously.
|
|
|
+ # If the id matches, we reject it. The client can also detect the op has been submitted
|
|
|
+ # already if it sees its own previous id in the ops it sees when it does catchup.
|
|
|
+ if oldOp.meta.source and opData.dupIfSource and oldOp.meta.source in opData.dupIfSource
|
|
|
+ return callback 'Op already submitted'
|
|
|
+
|
|
|
+ opData.op = doc.type.transform opData.op, oldOp.op, 'left'
|
|
|
+ opData.v++
|
|
|
+ catch error
|
|
|
+ console.error error.stack
|
|
|
+ return callback error.message
|
|
|
+
|
|
|
+ try
|
|
|
+ snapshot = doc.type.apply doc.snapshot, opData.op
|
|
|
+ catch error
|
|
|
+ console.error error.stack
|
|
|
+ return callback error.message
|
|
|
+
|
|
|
+ # The op data should be at the current version, and the new document data should be at
|
|
|
+ # the next version.
|
|
|
+ #
|
|
|
+ # This should never happen in practice, but its a nice little check to make sure everything
|
|
|
+ # is hunky-dory.
|
|
|
+ unless opData.v == doc.v
|
|
|
+ # This should never happen.
|
|
|
+ console.error "Version mismatch detected in model. File a ticket - this is a bug."
|
|
|
+ console.error "Expecting #{opData.v} == #{doc.v}"
|
|
|
+ return callback 'Internal error'
|
|
|
+
|
|
|
+ #newDocData = {snapshot, type:type.name, v:opVersion + 1, meta:docData.meta}
|
|
|
+ writeOp = db?.writeOp or (docName, newOpData, callback) -> callback()
|
|
|
+
|
|
|
+ writeOp docName, opData, (error) ->
|
|
|
+ if error
|
|
|
+ # The user should probably know about this.
|
|
|
+ console.warn "Error writing ops to database: #{error}"
|
|
|
+ return callback error
|
|
|
+
|
|
|
+ options.stats?.writeOp?()
|
|
|
+
|
|
|
+ # This is needed when we emit the 'change' event, below.
|
|
|
+ oldSnapshot = doc.snapshot
|
|
|
+
|
|
|
+ # All the heavy lifting is now done. Finally, we'll update the cache with the new data
|
|
|
+ # and (maybe!) save a new document snapshot to the database.
|
|
|
+
|
|
|
+ doc.v = opData.v + 1
|
|
|
+ doc.snapshot = snapshot
|
|
|
+
|
|
|
+ doc.ops.push opData
|
|
|
+ doc.ops.shift() if db and doc.ops.length > options.numCachedOps
|
|
|
+
|
|
|
+ model.emit 'applyOp', docName, opData, snapshot, oldSnapshot
|
|
|
+ doc.eventEmitter.emit 'op', opData, snapshot, oldSnapshot
|
|
|
+
|
|
|
+ # The callback is called with the version of the document at which the op was applied.
|
|
|
+ # This is the op.v after transformation, and its doc.v - 1.
|
|
|
+ callback null, opData.v
|
|
|
+
|
|
|
+ # I need a decent strategy here for deciding whether or not to save the snapshot.
|
|
|
+ #
|
|
|
+ # The 'right' strategy looks something like "Store the snapshot whenever the snapshot
|
|
|
+ # is smaller than the accumulated op data". For now, I'll just store it every 20
|
|
|
+ # ops or something. (Configurable with doc.committedVersion)
|
|
|
+ if !doc.snapshotWriteLock and doc.committedVersion + options.opsBeforeCommit <= doc.v
|
|
|
+ tryWriteSnapshot docName, (error) ->
|
|
|
+ console.warn "Error writing snapshot #{error}. This is nonfatal" if error
|
|
|
+
|
|
|
+ # Add the data for the given docName to the cache. The named document shouldn't already
|
|
|
+ # exist in the doc set.
|
|
|
+ #
|
|
|
+ # Returns the new doc.
|
|
|
+ add = (docName, error, data, committedVersion, ops, dbMeta) ->
|
|
|
+ callbacks = awaitingGetSnapshot[docName]
|
|
|
+ delete awaitingGetSnapshot[docName]
|
|
|
+
|
|
|
+ if error
|
|
|
+ callback error for callback in callbacks if callbacks
|
|
|
+ else
|
|
|
+ doc = docs[docName] =
|
|
|
+ snapshot: data.snapshot
|
|
|
+ v: data.v
|
|
|
+ type: data.type
|
|
|
+ meta: data.meta
|
|
|
+
|
|
|
+ # Cache of ops
|
|
|
+ ops: ops or []
|
|
|
+
|
|
|
+ eventEmitter: new EventEmitter
|
|
|
+
|
|
|
+ # Timer before the document will be invalidated from the cache (if the document has no
|
|
|
+ # listeners)
|
|
|
+ reapTimer: null
|
|
|
+
|
|
|
+ # Version of the snapshot thats in the database
|
|
|
+ committedVersion: committedVersion ? data.v
|
|
|
+ snapshotWriteLock: false
|
|
|
+ dbMeta: dbMeta
|
|
|
+
|
|
|
+ doc.opQueue = makeOpQueue docName, doc
|
|
|
+
|
|
|
+ refreshReapingTimeout docName
|
|
|
+ model.emit 'add', docName, data
|
|
|
+ callback null, doc for callback in callbacks if callbacks
|
|
|
+
|
|
|
+ doc
|
|
|
+
|
|
|
+ # This is a little helper wrapper around db.getOps. It does two things:
|
|
|
+ #
|
|
|
+ # - If there's no database set, it returns an error to the callback
|
|
|
+ # - It adds version numbers to each op returned from the database
|
|
|
+ # (These can be inferred from context so the DB doesn't store them, but its useful to have them).
|
|
|
+ getOpsInternal = (docName, start, end, callback) ->
|
|
|
+ return callback? 'Document does not exist' unless db
|
|
|
+
|
|
|
+ db.getOps docName, start, end, (error, ops) ->
|
|
|
+ return callback? error if error
|
|
|
+
|
|
|
+ v = start
|
|
|
+ op.v = v++ for op in ops
|
|
|
+
|
|
|
+ callback? null, ops
|
|
|
+
|
|
|
+ # Load the named document into the cache. This function is re-entrant.
|
|
|
+ #
|
|
|
+ # The callback is called with (error, doc)
|
|
|
+ load = (docName, callback) ->
|
|
|
+ if docs[docName]
|
|
|
+ # The document is already loaded. Return immediately.
|
|
|
+ options.stats?.cacheHit? 'getSnapshot'
|
|
|
+ return callback null, docs[docName]
|
|
|
+
|
|
|
+ # We're a memory store. If we don't have it, nobody does.
|
|
|
+ return callback 'Document does not exist' unless db
|
|
|
+
|
|
|
+ callbacks = awaitingGetSnapshot[docName]
|
|
|
+
|
|
|
+ # The document is being loaded already. Add ourselves as a callback.
|
|
|
+ return callbacks.push callback if callbacks
|
|
|
+
|
|
|
+ options.stats?.cacheMiss? 'getSnapshot'
|
|
|
+
|
|
|
+ # The document isn't loaded and isn't being loaded. Load it.
|
|
|
+ awaitingGetSnapshot[docName] = [callback]
|
|
|
+ db.getSnapshot docName, (error, data, dbMeta) ->
|
|
|
+ return add docName, error if error
|
|
|
+
|
|
|
+ type = types[data.type]
|
|
|
+ unless type
|
|
|
+ console.warn "Type '#{data.type}' missing"
|
|
|
+ return callback "Type not found"
|
|
|
+ data.type = type
|
|
|
+
|
|
|
+ committedVersion = data.v
|
|
|
+
|
|
|
+ # The server can close without saving the most recent document snapshot.
|
|
|
+ # In this case, there are extra ops which need to be applied before
|
|
|
+ # returning the snapshot.
|
|
|
+ getOpsInternal docName, data.v, null, (error, ops) ->
|
|
|
+ return callback error if error
|
|
|
+
|
|
|
+ if ops.length > 0
|
|
|
+ console.log "Catchup #{docName} #{data.v} -> #{data.v + ops.length}"
|
|
|
+
|
|
|
+ try
|
|
|
+ for op in ops
|
|
|
+ data.snapshot = type.apply data.snapshot, op.op
|
|
|
+ data.v++
|
|
|
+ catch e
|
|
|
+ # This should never happen - it indicates that whats in the
|
|
|
+ # database is invalid.
|
|
|
+ console.error "Op data invalid for #{docName}: #{e.stack}"
|
|
|
+ return callback 'Op data invalid'
|
|
|
+
|
|
|
+ model.emit 'load', docName, data
|
|
|
+ add docName, error, data, committedVersion, ops, dbMeta
|
|
|
+
|
|
|
+ # This makes sure the cache contains a document. If the doc cache doesn't contain
|
|
|
+ # a document, it is loaded from the database and stored.
|
|
|
+ #
|
|
|
+ # Documents are stored so long as either:
|
|
|
+ # - They have been accessed within the past #{PERIOD}
|
|
|
+ # - At least one client has the document open
|
|
|
+ refreshReapingTimeout = (docName) ->
|
|
|
+ doc = docs[docName]
|
|
|
+ return unless doc
|
|
|
+
|
|
|
+ # I want to let the clients list be updated before this is called.
|
|
|
+ process.nextTick ->
|
|
|
+ # This is an awkward way to find out the number of clients on a document. If this
|
|
|
+ # causes performance issues, add a numClients field to the document.
|
|
|
+ #
|
|
|
+ # The first check is because its possible that between refreshReapingTimeout being called and this
|
|
|
+ # event being fired, someone called delete() on the document and hence the doc is something else now.
|
|
|
+ if doc == docs[docName] and
|
|
|
+ doc.eventEmitter.listeners('op').length == 0 and
|
|
|
+ (db or options.forceReaping) and
|
|
|
+ doc.opQueue.busy is false
|
|
|
+
|
|
|
+ clearTimeout doc.reapTimer
|
|
|
+ doc.reapTimer = reapTimer = setTimeout ->
|
|
|
+ tryWriteSnapshot docName, ->
|
|
|
+ # If the reaping timeout has been refreshed while we're writing the snapshot, or if we're
|
|
|
+ # in the middle of applying an operation, don't reap.
|
|
|
+ delete docs[docName] if docs[docName].reapTimer is reapTimer and doc.opQueue.busy is false
|
|
|
+ , options.reapTime
|
|
|
+
|
|
|
+ tryWriteSnapshot = (docName, callback) ->
|
|
|
+ return callback?() unless db
|
|
|
+
|
|
|
+ doc = docs[docName]
|
|
|
+
|
|
|
+ # The doc is closed
|
|
|
+ return callback?() unless doc
|
|
|
+
|
|
|
+ # The document is already saved.
|
|
|
+ return callback?() if doc.committedVersion is doc.v
|
|
|
+
|
|
|
+ return callback? 'Another snapshot write is in progress' if doc.snapshotWriteLock
|
|
|
+
|
|
|
+ doc.snapshotWriteLock = true
|
|
|
+
|
|
|
+ options.stats?.writeSnapshot?()
|
|
|
+
|
|
|
+ writeSnapshot = db?.writeSnapshot or (docName, docData, dbMeta, callback) -> callback()
|
|
|
+
|
|
|
+ data =
|
|
|
+ v: doc.v
|
|
|
+ meta: doc.meta
|
|
|
+ snapshot: doc.snapshot
|
|
|
+ # The database doesn't know about object types.
|
|
|
+ type: doc.type.name
|
|
|
+
|
|
|
+ # Commit snapshot.
|
|
|
+ writeSnapshot docName, data, doc.dbMeta, (error, dbMeta) ->
|
|
|
+ doc.snapshotWriteLock = false
|
|
|
+
|
|
|
+ # We have to use data.v here because the version in the doc could
|
|
|
+ # have been updated between the call to writeSnapshot() and now.
|
|
|
+ doc.committedVersion = data.v
|
|
|
+ doc.dbMeta = dbMeta
|
|
|
+
|
|
|
+ callback? error
|
|
|
+
|
|
|
+ # *** Model interface methods
|
|
|
+
|
|
|
+ # Create a new document.
|
|
|
+ #
|
|
|
+ # data should be {snapshot, type, [meta]}. The version of a new document is 0.
|
|
|
+ @create = (docName, type, meta, callback) ->
|
|
|
+ [meta, callback] = [{}, meta] if typeof meta is 'function'
|
|
|
+
|
|
|
+ return callback? 'Invalid document name' if docName.match /\//
|
|
|
+ return callback? 'Document already exists' if docs[docName]
|
|
|
+
|
|
|
+ type = types[type] if typeof type == 'string'
|
|
|
+ return callback? 'Type not found' unless type
|
|
|
+
|
|
|
+ data =
|
|
|
+ snapshot:type.create()
|
|
|
+ type:type.name
|
|
|
+ meta:meta or {}
|
|
|
+ v:0
|
|
|
+
|
|
|
+ done = (error, dbMeta) ->
|
|
|
+ # dbMeta can be used to cache extra state needed by the database to access the document, like an ID or something.
|
|
|
+ return callback? error if error
|
|
|
+
|
|
|
+ # From here on we'll store the object version of the type name.
|
|
|
+ data.type = type
|
|
|
+ add docName, null, data, 0, [], dbMeta
|
|
|
+ model.emit 'create', docName, data
|
|
|
+ callback?()
|
|
|
+
|
|
|
+ if db
|
|
|
+ db.create docName, data, done
|
|
|
+ else
|
|
|
+ done()
|
|
|
+
|
|
|
+ # Perminantly deletes the specified document.
|
|
|
+ # If listeners are attached, they are removed.
|
|
|
+ #
|
|
|
+ # The callback is called with (error) if there was an error. If error is null / undefined, the
|
|
|
+ # document was deleted.
|
|
|
+ #
|
|
|
+ # WARNING: This isn't well supported throughout the code. (Eg, streaming clients aren't told about the
|
|
|
+ # deletion. Subsequent op submissions will fail).
|
|
|
+ @delete = (docName, callback) ->
|
|
|
+ doc = docs[docName]
|
|
|
+
|
|
|
+ if doc
|
|
|
+ clearTimeout doc.reapTimer
|
|
|
+ delete docs[docName]
|
|
|
+
|
|
|
+ done = (error) ->
|
|
|
+ model.emit 'delete', docName unless error
|
|
|
+ callback? error
|
|
|
+
|
|
|
+ if db
|
|
|
+ db.delete docName, doc?.dbMeta, done
|
|
|
+ else
|
|
|
+ done (if !doc then 'Document does not exist')
|
|
|
+
|
|
|
+ # This gets all operations from [start...end]. (That is, its not inclusive.)
|
|
|
+ #
|
|
|
+ # end can be null. This means 'get me all ops from start'.
|
|
|
+ #
|
|
|
+ # Each op returned is in the form {op:o, meta:m, v:version}.
|
|
|
+ #
|
|
|
+ # Callback is called with (error, [ops])
|
|
|
+ #
|
|
|
+ # If the document does not exist, getOps doesn't necessarily return an error. This is because
|
|
|
+ # its awkward to figure out whether or not the document exists for things
|
|
|
+ # like the redis database backend. I guess its a bit gross having this inconsistant
|
|
|
+ # with the other DB calls, but its certainly convenient.
|
|
|
+ #
|
|
|
+ # Use getVersion() to determine if a document actually exists, if thats what you're
|
|
|
+ # after.
|
|
|
+ @getOps = getOps = (docName, start, end, callback) ->
|
|
|
+ # getOps will only use the op cache if its there. It won't fill the op cache in.
|
|
|
+ throw new Error 'start must be 0+' unless start >= 0
|
|
|
+
|
|
|
+ [end, callback] = [null, end] if typeof end is 'function'
|
|
|
+
|
|
|
+ ops = docs[docName]?.ops
|
|
|
+
|
|
|
+ if ops
|
|
|
+ version = docs[docName].v
|
|
|
+
|
|
|
+ # Ops contains an array of ops. The last op in the list is the last op applied
|
|
|
+ end ?= version
|
|
|
+ start = Math.min start, end
|
|
|
+
|
|
|
+ return callback null, [] if start == end
|
|
|
+
|
|
|
+ # Base is the version number of the oldest op we have cached
|
|
|
+ base = version - ops.length
|
|
|
+
|
|
|
+ # If the database is null, we'll trim to the ops we do have and hope thats enough.
|
|
|
+ if start >= base or db is null
|
|
|
+ refreshReapingTimeout docName
|
|
|
+ options.stats?.cacheHit 'getOps'
|
|
|
+
|
|
|
+ return callback null, ops[(start - base)...(end - base)]
|
|
|
+
|
|
|
+ options.stats?.cacheMiss 'getOps'
|
|
|
+
|
|
|
+ getOpsInternal docName, start, end, callback
|
|
|
+
|
|
|
+ # Gets the snapshot data for the specified document.
|
|
|
+ # getSnapshot(docName, callback)
|
|
|
+ # Callback is called with (error, {v: <version>, type: <type>, snapshot: <snapshot>, meta: <meta>})
|
|
|
+ @getSnapshot = (docName, callback) ->
|
|
|
+ load docName, (error, doc) ->
|
|
|
+ callback error, if doc then {v:doc.v, type:doc.type, snapshot:doc.snapshot, meta:doc.meta}
|
|
|
+
|
|
|
+ # Gets the latest version # of the document.
|
|
|
+ # getVersion(docName, callback)
|
|
|
+ # callback is called with (error, version).
|
|
|
+ @getVersion = (docName, callback) ->
|
|
|
+ load docName, (error, doc) -> callback error, doc?.v
|
|
|
+
|
|
|
+ # Apply an op to the specified document.
|
|
|
+ # The callback is passed (error, applied version #)
|
|
|
+ # opData = {op:op, v:v, meta:metadata}
|
|
|
+ #
|
|
|
+ # Ops are queued before being applied so that the following code applies op C before op B:
|
|
|
+ # model.applyOp 'doc', OPA, -> model.applyOp 'doc', OPB
|
|
|
+ # model.applyOp 'doc', OPC
|
|
|
+ @applyOp = (docName, opData, callback) ->
|
|
|
+ # All the logic for this is in makeOpQueue, above.
|
|
|
+ load docName, (error, doc) ->
|
|
|
+ return callback error if error
|
|
|
+
|
|
|
+ process.nextTick -> doc.opQueue opData, (error, newVersion) ->
|
|
|
+ refreshReapingTimeout docName
|
|
|
+ callback? error, newVersion
|
|
|
+
|
|
|
+ # TODO: store (some) metadata in DB
|
|
|
+ # TODO: op and meta should be combineable in the op that gets sent
|
|
|
+ @applyMetaOp = (docName, metaOpData, callback) ->
|
|
|
+ {path, value} = metaOpData.meta
|
|
|
+
|
|
|
+ return callback? "path should be an array" unless isArray path
|
|
|
+
|
|
|
+ load docName, (error, doc) ->
|
|
|
+ if error?
|
|
|
+ callback? error
|
|
|
+ else
|
|
|
+ applied = false
|
|
|
+ switch path[0]
|
|
|
+ when 'shout'
|
|
|
+ doc.eventEmitter.emit 'op', metaOpData
|
|
|
+ applied = true
|
|
|
+
|
|
|
+ model.emit 'applyMetaOp', docName, path, value if applied
|
|
|
+ callback? null, doc.v
|
|
|
+
|
|
|
+ # Listen to all ops from the specified version. If version is in the past, all
|
|
|
+ # ops since that version are sent immediately to the listener.
|
|
|
+ #
|
|
|
+ # The callback is called once the listener is attached, but before any ops have been passed
|
|
|
+ # to the listener.
|
|
|
+ #
|
|
|
+ # This will _not_ edit the document metadata.
|
|
|
+ #
|
|
|
+ # If there are any listeners, we don't purge the document from the cache. But be aware, this behaviour
|
|
|
+ # might change in a future version.
|
|
|
+ #
|
|
|
+ # version is the document version at which the document is opened. It can be left out if you want to open
|
|
|
+ # the document at the most recent version.
|
|
|
+ #
|
|
|
+ # listener is called with (opData) each time an op is applied.
|
|
|
+ #
|
|
|
+ # callback(error, openedVersion)
|
|
|
+ @listen = (docName, version, listener, callback) ->
|
|
|
+ [version, listener, callback] = [null, version, listener] if typeof version is 'function'
|
|
|
+
|
|
|
+ load docName, (error, doc) ->
|
|
|
+ return callback? error if error
|
|
|
+
|
|
|
+ clearTimeout doc.reapTimer
|
|
|
+
|
|
|
+ if version?
|
|
|
+ getOps docName, version, null, (error, data) ->
|
|
|
+ return callback? error if error
|
|
|
+
|
|
|
+ doc.eventEmitter.on 'op', listener
|
|
|
+ callback? null, version
|
|
|
+ for op in data
|
|
|
+ listener op
|
|
|
+
|
|
|
+ # The listener may well remove itself during the catchup phase. If this happens, break early.
|
|
|
+ # This is done in a quite inefficient way. (O(n) where n = #listeners on doc)
|
|
|
+ break unless listener in doc.eventEmitter.listeners 'op'
|
|
|
+
|
|
|
+ else # Version is null / undefined. Just add the listener.
|
|
|
+ doc.eventEmitter.on 'op', listener
|
|
|
+ callback? null, doc.v
|
|
|
+
|
|
|
+ # Remove a listener for a particular document.
|
|
|
+ #
|
|
|
+ # removeListener(docName, listener)
|
|
|
+ #
|
|
|
+ # This is synchronous.
|
|
|
+ @removeListener = (docName, listener) ->
|
|
|
+ # The document should already be loaded.
|
|
|
+ doc = docs[docName]
|
|
|
+ throw new Error 'removeListener called but document not loaded' unless doc
|
|
|
+
|
|
|
+ doc.eventEmitter.removeListener 'op', listener
|
|
|
+ refreshReapingTimeout docName
|
|
|
+
|
|
|
+ # Flush saves all snapshot data to the database. I'm not sure whether or not this is actually needed -
|
|
|
+ # sharejs will happily replay uncommitted ops when documents are re-opened anyway.
|
|
|
+ @flush = (callback) ->
|
|
|
+ return callback?() unless db
|
|
|
+
|
|
|
+ pendingWrites = 0
|
|
|
+
|
|
|
+ for docName, doc of docs
|
|
|
+ if doc.committedVersion < doc.v
|
|
|
+ pendingWrites++
|
|
|
+ # I'm hoping writeSnapshot will always happen in another thread.
|
|
|
+ tryWriteSnapshot docName, ->
|
|
|
+ process.nextTick ->
|
|
|
+ pendingWrites--
|
|
|
+ callback?() if pendingWrites is 0
|
|
|
+
|
|
|
+ # If nothing was queued, terminate immediately.
|
|
|
+ callback?() if pendingWrites is 0
|
|
|
+
|
|
|
+ # Close the database connection. This is needed so nodejs can shut down cleanly.
|
|
|
+ @closeDb = ->
|
|
|
+ db?.close?()
|
|
|
+ db = null
|
|
|
+
|
|
|
+ return
|
|
|
+
|
|
|
+# Model inherits from EventEmitter.
|
|
|
+Model:: = new EventEmitter
|
|
|
+
|