code: function consume_one (shared_ctx, self_ctx, cb) {
    if (shared_ctx.pop_count > shared_ctx.pop_max) {
      console.log ('pop consumer %s ended', self_ctx.id);
      shared_ctx.q.cancel ();
      return cb ();
    }
  
    shared_ctx.pop_count ++;
    self_ctx.pop_count ++;
  
    shared_ctx.q.pop (self_ctx.id, shared_ctx.pop_opts, (err, res) => {
      if (err) return cb (err);
      if ((shared_ctx.pop_count % 100000) == 0) console.log ('%s : consumed #%d elems', self_ctx.id, shared_ctx.pop_count);
      consume_one (shared_ctx, self_ctx, cb);
    });
  }
docs: ////////////////////////////////////////////////////////////////////////////////////////////////////////////