defmodule NSQ.Message do
# ------------------------------------------------------- #
# Directives #
# ------------------------------------------------------- #
import NSQ.Protocol
alias NSQ.Connection.Buffer
use GenServer
# ------------------------------------------------------- #
# Struct Definition #
# ------------------------------------------------------- #
defstruct [
:id,
:reader,
:writer,
:timestamp,
:attempts,
:body,
:connection,
:consumer,
:config,
:parent,
:processing_pid,
:event_manager_pid,
:msg_timeout
]
# ------------------------------------------------------- #
# Behaviour Implementation #
# ------------------------------------------------------- #
def start_link(message, opts \\ []) do
{:ok, _pid} = GenServer.start_link(__MODULE__, message, opts)
end
def init(message) do
# Process the message asynchronously after init.
GenServer.cast(self(), :process)
{:ok, message}
end
def handle_cast(:process, message) do
# `process/1` will send the return value to `message.connection` as part of
# its standard procedure.
process(message)
{:noreply, message}
end
# ------------------------------------------------------- #
# API Definitions #
# ------------------------------------------------------- #
@doc """
Given a raw NSQ message with frame_type = message, return a struct with its
parsed data.
"""
def from_data(data) do
{:ok, message} = decode_as_message(data)
Map.merge(%NSQ.Message{}, message)
end
@doc """
This is the main entry point when processing a message. It starts the message
GenServer and immediately kicks of a processing call.
"""
def process(message) do
# Kick off processing in a separate process, so we can kill it if it takes
# too long.
message = %{message | parent: self()}
{:ok, pid} =
Task.start_link(fn ->
process_without_timeout(message)
end)
message = %{message | processing_pid: pid}
{:ok, ret_val} = wait_for_msg_done(message)
# Even if we've killed the message processing, we're still "done"
# processing it for now. That is, we should free up a spot on
# messages_in_flight.
result = {:message_done, message, ret_val}
send(message.connection, result)
# Nothing more for this process to do.
Process.exit(self(), :normal)
end
@doc """
Tells NSQD that we're done processing this message. This is called
automatically when the handler returns successfully, or when all retries have
been exhausted.
"""
def fin(message) do
NSQ.Logger.debug("(#{inspect(message.connection)}) fin msg ID #{message.id}")
message |> Buffer.send!(encode({:fin, message.id}))
:gen_event.notify(message.event_manager_pid, {:message_finished, message})
GenServer.call(message.consumer, {:start_stop_continue_backoff, :resume})
end
@doc """
Tells NSQD to requeue the message, with delay and backoff. According to the
go-nsq client (but doc'ed nowhere), a delay of -1 is a special value that
means nsqd will calculate the duration itself based on the default requeue
timeout. If backoff=true is set, then the connection will go into "backoff"
mode, where it stops receiving messages for a fixed duration.
"""
def req(message, delay \\ -1, backoff \\ false) do
delay =
if delay == -1 do
delay =
calculate_delay(
message.attempts,
message.config.max_requeue_delay
)
NSQ.Logger.debug(
"(#{inspect(message.connection)}) requeue msg ID #{message.id}, delay #{delay} (auto-calculated with attempts #{message.attempts}), backoff #{backoff}"
)
delay
else
NSQ.Logger.debug(
"(#{inspect(message.connection)}) requeue msg ID #{message.id}, delay #{delay}, backoff #{backoff}"
)
delay
end
delay =
if delay > message.config.max_requeue_delay do
NSQ.Logger.warn(
"Invalid requeue delay #{delay}. Must be between 0 and #{message.config.max_requeue_delay}. Sending with max delay #{message.config.max_requeue_delay} instead."
)
message.config.max_requeue_delay
else
delay
end
if backoff do
GenServer.call(message.consumer, {:start_stop_continue_backoff, :backoff})
else
GenServer.call(message.consumer, {:start_stop_continue_backoff, :continue})
end
message |> Buffer.send!(encode({:req, message.id, delay}))
:gen_event.notify(message.event_manager_pid, {:message_requeued, message})
end
@doc """
This function is intended to be used by the handler for long-running
functions. They can set up a separate process that periodically touches the
message until the process finishes.
"""
def touch(message) do
NSQ.Logger.debug("(#{inspect(message.connection)}) touch msg ID #{message.id}")
message |> Buffer.send!(encode({:touch, message.id}))
send(message.parent, {:message_touch, message})
end
# ------------------------------------------------------- #
# Private Functions #
# ------------------------------------------------------- #
defp process_without_timeout(message) do
ret_val =
if should_fail_message?(message) do
:fail |> respond_to_nsq(message)
else
run_handler_safely(message) |> respond_to_nsq(message)
end
send(message.parent, {:message_done, message, ret_val})
end
defp should_fail_message?(message) do
message.config.max_attempts > 0 &&
message.attempts > message.config.max_attempts
end
# Handler can be either an anonymous function or a module that implements the
# `handle_message\2` function.
defp run_handler(handler, message) do
if is_function(handler) do
handler.(message.body, message)
else
handler.handle_message(message.body, message)
end
end
defp run_handler_safely(message) do
Process.flag(:trap_exit, true)
try do
result = run_handler(message.config.message_handler, message)
Process.flag(:trap_exit, false)
result
rescue
e ->
NSQ.Logger.error("Error running message handler: #{inspect(e)}")
NSQ.Logger.error(inspect(__STACKTRACE__))
{:req, -1, true}
catch
:exit, b ->
NSQ.Logger.error("Caught exit running message handler: :exit, #{inspect(b)}")
NSQ.Logger.error(inspect(__STACKTRACE__))
{:req, -1, true}
a, b ->
NSQ.Logger.error("Caught exception running message handler: #{inspect(a)}, #{inspect(b)}")
NSQ.Logger.error(inspect(__STACKTRACE__))
{:req, -1, true}
end
end
defp respond_to_nsq(ret_val, message) do
case ret_val do
:ok ->
fin(message)
:fail ->
NSQ.Logger.warn("msg #{message.id} attempted #{message.attempts} times, giving up")
fin(message)
:req ->
req(message)
{:req, delay} ->
req(message, delay)
{:req, delay, backoff} ->
req(message, delay, backoff)
_ ->
NSQ.Logger.error(
"Unexpected handler result #{inspect(ret_val)}, requeueing message #{message.id}"
)
req(message)
end
ret_val
end
# We expect our function will send us a message when it's done. Block until
# that happens. If it takes too long, requeue the message and cancel
# processing.
defp wait_for_msg_done(message) do
receive do
{:message_done, _msg, ret_val} ->
{:ok, ret_val}
{:message_touch, _msg} ->
NSQ.Logger.debug("Msg #{message.id} received TOUCH, starting a new wait...")
# If NSQ.Message.touch(msg) is called, we will send TOUCH msg_id to
# NSQD, but we also need to reset our timeout on the client to avoid
# processes that hang forever.
wait_for_msg_done(message)
after
message.msg_timeout ->
# If we've waited this long, we can assume NSQD will requeue the
# message on its own.
NSQ.Logger.warn(
"Msg #{message.id} timed out, quit processing it and expect nsqd to requeue"
)
:gen_event.notify(message.event_manager_pid, {:message_requeued, message})
unlink_and_exit(message.parent)
{:ok, :req}
end
end
defp unlink_and_exit(pid) do
Process.unlink(pid)
Process.exit(pid, :kill)
end
defp calculate_delay(attempts, max_requeue_delay) do
exponential_backoff = :math.pow(2, attempts) * 1000
jitter = round(0.3 * :rand.uniform() * exponential_backoff)
min(
exponential_backoff + jitter,
max_requeue_delay
)
|> round
end
end
