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()