defmodule NSQ.Producer do
@moduledoc """
A producer is a process that connects to one or many NSQDs and publishes
messages.
## Interface
To initialize:
{:ok, producer} = NSQ.Producer.Supervisor.start_link("the-default-topic", %NSQ.Config{
nsqds: ["127.0.0.1:6750"]
})
The default `topic` argument is _required_, even if you plan on explicitly
publishing to a different topic. If you don't plan on using, you can set it
to something like `_default_topic_`.
If you provide more than one nsqd, each `pub`/`mpub` will choose one
randomly.
Note that, unlike consumers, producers _cannot_ be configured to use
discovery with nsqlookupd. This is because discovery requires a topic and
channel, and an nsqd will only appear in nsqlookupd if it has already
published messages on that topic. So there's a chicken-and-egg problem. The
recommended solution is to run NSQD on the same box where you're publishing,
so your address is always 127.0.0.1 with a static port.
### pub
Publish a single message to NSQD.
NSQ.Producer.pub(producer, "a message")
NSQ.Producer.pub(producer, "different-topic", "a message")
### mpub
Publish a bunch of messages to NSQD atomically.
NSQ.Producer.mpub(producer, ["one", "two"])
NSQ.Producer.mpub(producer, "different-topic", ["one", "two"])
"""
# ------------------------------------------------------- #
# Directives #
# ------------------------------------------------------- #
import NSQ.Protocol
use GenServer
# ------------------------------------------------------- #
# Module Attributes #
# ------------------------------------------------------- #
@initial_state %{
topic: nil,
channel: nil,
conn_sup_pid: nil,
event_manager_pid: nil,
config: nil,
conn_info_pid: nil
}
# ------------------------------------------------------- #
# Type Definitions #
# ------------------------------------------------------- #
@typedoc """
A tuple with a host and a port.
"""
@type host_with_port :: {String.t(), integer}
@typedoc """
A tuple with a string ID (used to target the connection in
NSQ.Connection.Supervisor) and a PID of the connection.
"""
@type connection :: {String.t(), pid}
@typedoc """
A map, but we can be more specific by asserting some entries that should be
set for a connection's state map.
"""
@type pro_state :: %{conn_sup_pid: pid, config: NSQ.Config.t()}
# ------------------------------------------------------- #
# Behaviour Implementation #
# ------------------------------------------------------- #
@spec init(pro_state) :: {:ok, pro_state}
def init(pro_state) do
{:ok, conn_sup_pid} = NSQ.Connection.Supervisor.start_link()
pro_state = %{pro_state | conn_sup_pid: conn_sup_pid}
{:ok, conn_info_pid} = Agent.start_link(fn -> %{} end)
pro_state = %{pro_state | conn_info_pid: conn_info_pid}
manager =
if pro_state.config.event_manager do
pro_state.config.event_manager
else
{:ok, manager} = :gen_event.start_link()
manager
end
pro_state = %{pro_state | event_manager_pid: manager}
{:ok, _pro_state} = connect_to_nsqds(pro_state.config.nsqds, self(), pro_state)
end
@spec handle_call({:pub, binary}, any, pro_state) ::
{:reply, {:ok, binary}, pro_state}
def handle_call({:pub, data}, _from, pro_state) do
do_pub(pro_state.topic, data, pro_state)
end
@spec handle_call({:pub, binary, binary}, any, pro_state) ::
{:reply, {:ok, binary}, pro_state}
def handle_call({:pub, topic, data}, _from, pro_state) do
do_pub(topic, data, pro_state)
end
@spec handle_call({:mpub, binary}, any, pro_state) ::
{:reply, {:ok, binary}, pro_state}
def handle_call({:mpub, data}, _from, pro_state) do
do_mpub(pro_state.topic, data, pro_state)
end
@spec handle_call({:mpub, binary, binary}, any, pro_state) ::
{:reply, {:ok, binary}, pro_state}
def handle_call({:mpub, topic, data}, _from, pro_state) do
do_mpub(topic, data, pro_state)
end
@spec handle_call(:state, any, pro_state) :: {:reply, pro_state, pro_state}
def handle_call(:state, _from, state) do
{:reply, state, state}
end
# ------------------------------------------------------- #
# API Definitions #
# ------------------------------------------------------- #
@spec start_link({binary(), NSQ.Config.t()}) :: {:ok, pid}
def start_link({topic, config}), do: start_link({topic, config, []})
@spec start_link({binary, NSQ.Config.t(), GenServer.options()}) :: {:ok, pid}
def start_link({topic, config, genserver_options}) do
{:ok, config} = NSQ.Config.validate(config || %NSQ.Config{})
{:ok, config} = NSQ.Config.normalize(config)
unless is_valid_topic_name?(topic), do: raise("Invalid topic name #{topic}")
state = %{@initial_state | topic: topic, config: config}
GenServer.start_link(__MODULE__, state, genserver_options)
end
@spec get_connections(pro_state) :: [connection]
def get_connections(pro_state) when is_map(pro_state) do
children = Supervisor.which_children(pro_state.conn_sup_pid)
Enum.map(children, fn {child_id, pid, _, _} -> {child_id, pid} end)
end
@spec get_connections(pid, pro_state) :: [connection]
def get_connections(pro, pro_state \\ nil) when is_pid(pro) do
pro_state = pro_state || get_state(pro)
get_connections(pro_state)
end
@spec random_connection_pid(pro_state) :: pid
def random_connection_pid(pro_state) do
{_child_id, pid} = Enum.random(get_connections(pro_state))
pid
end
@doc """
Create supervised connections to NSQD.
"""
@spec connect_to_nsqds([host_with_port], pid, pro_state) :: {:ok, pro_state}
def connect_to_nsqds(nsqds, pro, pro_state) do
Enum.map(nsqds, fn nsqd ->
{:ok, _conn} =
NSQ.Connection.Supervisor.start_child(
pro,
nsqd,
pro_state,
restart: :permanent
)
end)
{:ok, pro_state}
end
@doc """
Get the current state of a producer. Used in tests. Not for external use.
"""
@spec get_state(pid) :: pro_state
def get_state(producer) do
GenServer.call(producer, :state)
end
@doc """
Publish data to whatever topic is the default.
"""
@spec pub(pid, binary) :: {:ok, binary}
def pub(sup_pid, data) do
{:ok, _resp} = GenServer.call(get(sup_pid), {:pub, data})
end
@doc """
Publish data to a specific topic.
"""
@spec pub(pid, binary, binary) :: {:ok, binary}
def pub(sup_pid, topic, data) do
{:ok, _resp} = GenServer.call(get(sup_pid), {:pub, topic, data})
end
@doc """
Publish data to whatever topic is the default.
"""
@spec mpub(pid, binary) :: {:ok, binary}
def mpub(sup_pid, data) do
GenServer.call(get(sup_pid), {:mpub, data})
end
@doc """
Publish data to a specific topic.
"""
@spec mpub(pid, binary, binary) :: {:ok, binary}
def mpub(sup_pid, topic, data) do
{:ok, _resp} = GenServer.call(get(sup_pid), {:mpub, topic, data})
end
@doc """
The end-user will be targeting the supervisor, but it's the producer that
can actually handle the command.
"""
@spec get(pid) :: pid
def get(sup_pid) do
child =
Supervisor.which_children(sup_pid)
|> Enum.find(fn {kind, _, _, _} -> kind == NSQ.Producer end)
{_, pid, _, _} = child
pid
end
# ------------------------------------------------------- #
# Private Functions #
# ------------------------------------------------------- #
# Used to DRY up handle_call({:pub, ...).
@spec do_pub(binary, binary, pro_state) :: {:reply, {:ok, binary}, pro_state}
defp do_pub(topic, data, pro_state) do
{:ok, resp} =
random_connection_pid(pro_state)
|> NSQ.Connection.cmd({:pub, topic, data})
{:reply, {:ok, resp}, pro_state}
end
# Used to DRY up handle_call({:mpub, ...).
@spec do_mpub(binary, binary, pro_state) :: {:reply, {:ok, binary}, pro_state}
defp do_mpub(topic, data, pro_state) do
{:ok, resp} =
random_connection_pid(pro_state)
|> NSQ.Connection.cmd({:mpub, topic, data})
{:reply, {:ok, resp}, pro_state}
end
end
