concurrent-couch-follower

a couch follower wrapper that you can use to be sure you don't miss any documents even if you process them asynchronously.

var changes = require('concurrent-couch-follower')
var someAction = require(.....)

var dataHandler = function(data, done) {
    someAction(data, function() {
      done()
    })
}

var configOptions = {
  db: 'https://url.to.couchdb/registry/_changes',
  include_docs:true,
  sequence:'.sequence',
  now:false,
  concurrency:5
}

changes(dataHandler, configOptions)

API

changes(handler,options)

• handler = function(change,done)

is a function that is called for every document - change, the change from couchdb {seq:sequence,doc:the document,....} - done, you must call this function when you are done processing the document.

• options

a config object as passed to changes-stream but including these additional properties. - db, the connection string url pointing to the CouchDB registry to be followed. - sequence, the name of the file to persist the sequence id, if this is a function this is passed as a persist function to concurrent-seq-file. - concurrency, the maximum number of documents to process at a time. - the changes-stream property since is populated by the value of the sequence file and cannot be set from outside except if now is set to true. - now, if true, set the changes-stream property since to "now" (instead of 0) on the first start (before .sequence has been created) - since only used, but is required, if you are using a custom backend to save the sequence ids. when you pass a function as sequence.

• stream = changes(handle,options)

- sream , return value is a readable object stream of data passed back with done(err,data)

• stream.sequence()

- returns current sequence id saved to disk. useful for logging.

• WARNING! stream.end()

- this calls destroy on the changes-stream and the through instead of properly ending them.

- this triggers a "premature close" error from `changes-stream` and is something that just has to be worked on. bind `error` or use `end-of-stream`

more examples

save the sequence ids in a database.

var changes = require('concurrent-couch-follower')
var someAction = require(.....)

var dataHandler = function(data, done) {
    someAction(data, function() {
      done()
    })
}

loadSequenceFromDB(function(err,sequence){
  var configOptions = {
    db: 'https://url.to.couchdb/registry/_changes',
    include_docs:true,
    sequence:function(seq,cb){
      saveInDB(seq,cb)
    },
    since:sequence,
    concurrency:5
  }

  changes(dataHandler, configOptions)
})