A consumer is a process that creates connections to NSQD to receive messages for a specific topic and channel. It has three primary functions:

1. Provide a simple interface for a user to setup and configure message handlers.
2. Balance RDY across all available connections.
3. Add/remove connections as they are discovered.

Simple Interface

In standard practice, the only function a user should need to know about is NSQ.Consumer.Supervisor.start_link/3. It takes a topic, a channel, and an NSQ.Config struct, which has possible values defined and explained in nsq/config.ex.

{:ok, consumer} = NSQ.Consumer.Supervisor.start_link("my-topic", "my-channel", %NSQ.Config{
  nsqlookupds: ["127.0.0.1:6751", "127.0.0.1:6761"],
  message_handler: fn(body, msg) ->
    # handle them message
    :ok
  end
})

Message handler return values

The return value of the message handler determines how we will respond to NSQ.

:ok

The message was handled and should not be requeued. This sends a FIN command to NSQD.

:req

This message should be requeued. With no delay specified, it will calculate delay exponentially based on the number of attempts. Refer to Message.calculate_delay for the exact formula.

{:req, delay}

This message should be requeued. Use the delay specified. A positive integer is expected.

{:req, delay, backoff}

This message should be requeued. Use the delay specified. If backoff is truthy, the consumer will temporarily set RDY to 0 in order to stop receiving messages. It will use a standard strategy to resume from backoff mode.

This type of return value is only meant for exceptional cases, such as internal network partitions, where stopping message handling briefly could be beneficial. Only use this return value if you know what you’re doing.

A message handler that throws an unhandled exception will automatically requeue and enter backoff mode.

NSQ.Message.touch(msg)

NSQ.Config has a property called msg_timeout, which configures the NSQD server to wait that long before assuming the message failed and requeueing it. If you expect your message handler to take longer than that, you can call NSQ.Message.touch(msg) from the message handler to reset the server-side timer.

NSQ.Consumer.change_max_in_flight(consumer, max_in_flight)

If you’d like to manually change the max in flight of a consumer, use this function. It will cause the consumer’s connections to rebalance to the new value. If the new max_in_flight is smaller than the current messages in flight, it must wait for the existing handlers to finish or requeue before it can fully rebalance.