elixir_nsq
is a client library for NSQ. Use it in your Elixir or Erlang
applications to handle messages asynchronously. This library seeks to be
complete, well-tested, and easy to use.
This library used go-nsq and pynsq for reference, but is structured to better fit common Elixir workflows.
To use this, you will need to have NSQ.
{:ok, producer} = NSQ.Producer.Supervisor.start_link("my-topic", %NSQ.Config{
nsqds: ["127.0.0.1:4150", "127.0.0.1:4151"]
})
# publish to the default topic "my-topic"
NSQ.Producer.pub(producer, "a message")
NSQ.Producer.mpub(producer, ["one", "two", "three"])
# specify a topic
NSQ.Producer.pub(producer, "different-topic", "another message")
NSQ.Producer.mpub(producer, "different-topic", ["four", "five", "six"])
NSQ.Producer.close(producer)
defp deps do
[{:elixir_nsq, "~> 1.0.3"}]
end
defp applications do
[:logger, :elixir_nsq]
end
The handler should return :ok
to finish normally, :req
to requeue the
message, or {:req, delay}
to specify your own requeue delay. If your message
handler throws an exception, it will automatically be requeued and delayed with
a timeout based on attempts.
{:ok, consumer} = NSQ.Consumer.Supervisor.start_link("my-topic", "my-channel", %NSQ.Config{
nsqlookupds: ["127.0.0.1:4160", "127.0.0.1:4161"],
message_handler: fn(body, msg) ->
IO.puts "id: #{msg.id}"
IO.puts "attempts: #{msg.attempts}"
IO.puts "timestamp: #{msg.timestamp}"
:ok
end
})
The message handler can also be a module that implements handle_message/2
:
defmodule MyMsgHandler do
def handle_message(body, msg) do
IO.puts "Handled in a module! #{msg.id}"
:ok
end
end
{:ok, consumer} = NSQ.Consumer.Supervisor.start_link("my-topic", "my-channel", %NSQ.Config{
nsqlookupds: ["127.0.0.1:4160", "127.0.0.1:4161"],
message_handler: MyMsgHandler
})
If your message is especially long-running and you know it's not dead, you can touch it so that NSQ doesn't automatically fail and requeue it.
def MyMsgHandler do
def handle_message(body, msg) do
Task.start_link fn ->
:timer.sleep(30_000)
NSQ.Message.touch(msg)
end
long_running_operation()
:ok
end
end
If you're not using nsqlookupd, you can specify nsqds directly:
{:ok, consumer} = NSQ.Consumer.Supervisor.start_link("my-topic", "my-channel", %NSQ.Config{
nsqds: ["127.0.0.1:4150", "127.0.0.1:4151"],
message_handler: fn(body, msg) ->
:ok
end
})
Check https://github.com/wistia/elixir_nsq/blob/master/lib/nsq/config.ex for supported config values.
NSQ.Consumer and NSQ.Producer provide the function event_manager/1
so that
you can receive events from the NSQ client. You can keep your own stats/logs
and perform actions based on that info.
defmodule EventForwarder do
use GenEvent
def handle_event(event, parent) do
send parent, event
{:ok, parent}
end
end
def setup_consumer do
{:ok, consumer} = NSQ.Consumer.Supervisor.start_link("my-topic", "my-channel", %NSQ.Config{
nsqds: ["127.0.0.1:4150", "127.0.0.1:4151"],
message_handler: fn(body, msg) ->
:ok
end
})
# subscribe to events from the event manager
NSQ.Consumer.event_manager(consumer)
|> GenEvent.add_handler(EventForwarder, self)
end
Potential event formats are:
{:message, NSQ.Message.t}
{:message_finished, NSQ.Message.t}
{:message_requeued, NSQ.Message.t}
:resume
:continue
:backoff
:heartbeat
{:response, binary}
{:error, String.t, binary}
For your convenience, this is the overall process structure of elixir_nsq
.
In practice, the Connection.Supervisors and Task.Supervisors don't do much
automatic restarting because NSQ itself is built to handle that. But they are
useful for propagating exit commands and keeping track of all running
processes.
Consumer Supervisor
Consumer
ConnInfo Agent
Connection.Supervisor
Connection
Message.Supervisor
Message
Message
Connection
Message.Supervisor
Message
Message
Connection discovery loop
RDY redistribution loop
Producer Supervisor
Producer
ConnInfo Agent
Connection.Supervisor
Connection
Connection
The included tests require two nsqds and two nsqlookupds. A Procfile for use with foreman is included to start these up. If you don't have foreman, you'll need to find a way to run those commands if you want to run the tests.
foreman start
mix test
Note that some tests intentionally cause processes to exit, so you might see some error logging as part of the tests. As long as they're still passing, that is considered normal behavior.
- Snappy cannot be supported because existing NIFs cannot correctly decompress the nsqd stream. I believe they need support for skipping the checksum.