bigqueue

node-bigqueue

A BigQueue module for node.js-based apps

Installation

Install global & local dependencies.

$ [sudo] npm install -g coffee coffeegulp [docco]
$ npm install

Configuration

Module is self configured to point to a local instance of BigQueue. It uses the node-config module, so you should place your config files on ./config folder in order to change the defaults.

Configuration Defaults

These are the default configuration values. More information on how override it is available here.

---
bigqueue:
  # Base URL. You may place #{cluster} in this value to point 
  # to specific BigQueue instances. 
  baseUri: 'http://127.0.0.1:8081'
 
  producer:
    # Path to produce messages. 
    postUri: '/messages'
 
    # Number of retries before failing. 
    defaultRetries: 3
 
    # Timeout between fails. 
    defaultTimeout: 1000
 
  consumer:
    # Reading path. 
    readUri: '/topics/#{topic}/consumers/#{consumer}/messages'
 
    # Acknowledgement path. 
    ackUri: '/topics/#{topic}/consumers/#{consumer}/messages/#{recipient}'
 
    worker:
      # Time before start-up fetching messages. 
      startDelay: 2000
 
      # Waiting time when the queue gets empty. 
      sleepOnEmpty: 1000
 
      # Waiting time when fetching fails. 
      sleepOnError: 2000
 
  client:
    # Time module users has to process a message before timing out 
    # without acknowledgment. 
    processTimeout: 500
 
  logger:
    # Enable log messages. 
    enabled: true

Running Tests

$ npm test

Gulp Tasks

$ coffeegulp [task = default]
    lint     -- run lint checks
    test     -- run tests
    coffee   -- compile scripts into .js
    default  -- perform all tasks above
    docco    -- generate docs

Usage

{Consumer, Producer} = require 'bigqueue'

Sending a Message

# Create Producer, with max_retries, timeout. 
producer = new Producer 3, 2000
 
# Send 'hello world!' message. 
producer.send 'cluster', ['test_topic', 'another'], 'hello world!'
.then(
  (value) ->
    console.log "message sent: #{JSON.stringify(value)}"
  (error) ->
    console.log "message failed: #{error}"
)

Fetching Messages

consumer = new Consumer 'cluster', 'test_consumer',
  ['test_topic', 'another']
 
consumer.on 'start', (consumer) ->
  console.log "we're up and running!"
 
consumer.on 'stop', (consumer) ->
  console.log 'stop!'
 
# Don't forget to notify Consumer whether we've been able to process the 
# message or not! 
consumer.on 'message', (message, ack) ->
  console.log "got a message!: #{JSON.stringify(message)}"
 
  # Process the message. We shouldn't take much time in here, otherwise we may 
  # timeout. 
  db.save message, (error) ->
    ack(error)
 
consumer.on 'ack', (message) ->
  console.log "message acknowledged!: #{message.recipientCallback}"
 
consumer.on 'uptodate', (consumer) ->
  console.log "we're up to date!"
 
# Start consuming events. 
consumer.start()