 
   # Register the 'next' event, and listen for 'resume' messages.
   constructor: (@options) ->
+    @queue_id = @_random_queue_id()
     @_workers = {}
     @_mixins = {}
     @_runners = {}
+    @_job_queues = @options.queues ? ['jobs']
     @_kv = db.key_value @options
     @_in_queue = db.queue @options
     @_out_queue = db.queue @options

  _run: (key) ->
    try
      @_worker_count++
      @_kv.del "sources:#{key}", "targets:#{key}"
      @_workers[key] = new Worker key, this
      @_workers[key].run()
    catch e

    else
      @_workers[key].resume()

  _multi: ->
    @_kv.redis.multi()

  # on pop job key
  #   record which queue for key
  #   if dirty flag
  #     enqueue key
  #   else
  #     as multi
  #       set self as owner
  #       add to set of owned jobs
  #       when done
  #         start running job
  _handle_job_key: (queue, key) ->
    @_queue_for_key[key] = queue
    if @_dirty
      @_enqueue_job queue, key
    else
      @_multi()
        .set('_lock:'+key, @queue_id)
        .sadd('_active:'+@queue_id, key)
        .exec (err) =>
          return @error err if err
          @_run key
    @_watch_queues()

  # on pop dirty
  #   set dirty flag
  #   delete info key
  #   look up targets
  #     push targets to dirty
  #       reset dirty timeout
  #       call popper
  _handle_dirty_key: (key) ->
    @_dirty = true
    @_kv.del '_lock:'+key
    @_kv.smembers '_targets:'+key, (err, targets) =>
      return @error err if err
      @_out_queue.rpush_all 'dirty', targets, (err) =>
        @error err if err
        @_reset_dirty_timeout()
        @_watch_queues()

  # on enqueue key
  #   acquire lock
  #     if success
  #       push key to job queue
  _enqueue_job: (queue, key) ->
    @_kv.setnx '_lock:'+key, 'queued', (err, set) =>
      return @error err if err
      return unless set
      @_out_queue.lpush queue, key

  # on dirty timeout
  #   clear dirty flag
  #   take/clear all pending keys
  #   get info for pending keys
  #     for each key
  #       if missing
  #         enqueue key
  #       else
  #         next tick: call resume function
  _post_dirty: ->
    @_dirty = false
    @_kv.del '_dirty'
    pending = @_pending
    @_pending = []
    keys = _.map pending, (p) -> '_lock:'+p
    @_kv.mget keys, (err, locks) ->
      return @error err if err
      for lock, i in locks
        f = @_resume_for_key keys[i]
        delete @_resume_for_key keys[i]
        if lock
          process.nextTick f
        else
          @_enqueue_job @_queue_for_key[key], key


  _reset_dirty_timeout: ->
    clearTimeout @_dirty_timeout
    @_dirty_timeout = setTimeout (=> @_post_dirty()), 1000

  _handle:

    dirty: ->
      @_dirty = true
      @_reset_dirty_timeout()

    ready: (key) ->
      clearTimeout @_cycle_timeouts[key]
