Overview

This module can be used to instantiate a job-consumer that is able to receive jobs from multiple types of broker backends. Each consumer connects to a single queue to receive jobs from. "queue" is a general term used to represent a job pipeline (Pub/Sub channel, AMQP channel/queue, SQS queue).

Currently supported backends:

• AMQP protocol via RabbitMQ
• AWS Lambda via AWS SQS - Also supports AWS SQS directly
• Redis via Bull.js

Broker Configuration

All broker backend configurations are handled by environment variables. Below you can find all environment variables and the expected environment variables for each broker backend type.

General

NODE_ENV

NODE_PATH

JOB_CONSUMER_NAME

JOB_BROKER_QUEUE

The queue to subscribe to.

JOB_BROKER_API_URL

URL used by @abeai/job-broker API itself. This is used to retry jobs. See 'Options, retryStrategy' below for more information.

JOB_BROKER_HOST

The host that is supporting your desired broker backend.

JOB_BROKER_AUTO_GENERATE_QUEUE

Set to true to automatically generate any queues that are attempted but do not exist. Currently this only applies to SQS broker backends.

JOB_BROKER_USER

Username used by broker backend if authentication is supported.

JOB_BROKER_PASSWORD

Password used by broker backend if authentication is supported.

Required

JOB_BROKER_BACKEND

Desired broker backend. Currently supports the following strings.

• AMQP
• SQS
• LAMBDA
• REDIS

AMQP

JOB_BROKER_HOST

amqp://{YOUR AMQP HOST}

SQS

JOB_BROKER_HOST

https://sqs.{YOUR AWS REGION}.amazonaws.com/{YOUR AWS ACCOUNT ID}/

JOB_BROKER_USER

AWS shared credentials profile.

AWS_REGION

AWS region.

LAMBDA

Lambda handler is called _lambdaExecute. No required environment variables.

REDIS

JOB_BROKER_HOST

redis://{YOUR REDIS HOST}

Interface

Constructor

const consumerFactory = require('@abeai/job-consumer');

consumerFactory(options);

Options

consume(job) (function)

This function will be executed when a job is received. Upon completion the job will be removed from its respective queue.

job

Arbitrary parsed JSON data passed into the consumer's respective queue.

error(error) (function)

This function provides custom error handling for a failed job. Generally this will only include logging as retry logic is handled by the below retryStrategy option.

removeOnError (boolean)

If set to true any failed jobs will be acknowledge with the broker and subsequently removed from its respective queue. Defaults to false.

retryStrategy(options) (function)

This method will be used as a job's retry strategy in the event of an error (and with removeOnError not set to true). You must return either an integer defining the amount of time to wait before retrying the job or null to cancel the current retry and allow the job to be acknowledged with its broker.

If this option is not set and removeOnError is not set to true the job will be retried with a progressively longer wait time between retries.

options (object)

• attempts: Current number of attempts for the job being retried.
• error: The error which caused the retry attempts.