defmodule Stargate do
@moduledoc """
Stargate provides an Elixir client for the Apache Pulsar distributed message
log service, based on the Pulsar project's websocket API.
### Producer
Create a producer process under your application's supervision tree with the following:
options = [
name: :pulsar_app,
host: [{:"broker-url.com", 8080}],
producer: [
persistence: "non-persistent",
tenant: "marketing",
namespace: "public",
topic: "new-stuff"
]
]
Stargate.Supervisor.start_link(options)
Once the producer is running, pass messages to the client by pid or by the named
registry entry:
Stargate.produce(producer, [{"key", "value"}])
If you won't be producing frequently you can choose to run ad hoc produce commands against
the url of the Pulsar cluster/topic as follows:
url = "ws://broker-url.com:8080/ws/v2/producer/non-persistent/marketing/public/new-stuff"
Stargate.produce(url, [{"key, "value"}])
### Consumer and Reader
Both consumers and readers connected to Pulsar via Stargate process received messages the
same way. Stargate takes care of receiving the messages and sending acknowledgements back
to the cluster so all you need to do is start a process and define a module in your application
that invokes `use Stargate.Receiver.MessageHandler` and has a `handle_message/1` or `handle_message/2`
function as follows:
defmodule Publicize.MessageHandler do
use Stargate.Receiver.MessageHandler
def handle_message(%{context: context, payload: payload}) do
publish_to_channel(payload, context)
:ack
end
defp publish_to_channel(payload, context) do
...do stuff...
end
end
The `handle_message/1` must return either `:ack` or `:continue` in order to ack successful
processing of the message back to the cluster or continue processing without ack (in the event
you want to do a bulk/cumulative ack at a later time). If using the `handle_message/2` callback
for handlers that keep state across messages handled, it must return `{:ack, state}` or
`{:continue, state}`.
Then, create a consumer or reader process under your application's supevision tree with the following:
options = [
name: :pulsar_app,
host: [{:"broker-url.com", 8080}]
consumer: [ <====== replace with `:reader` for a reader client
tenant: "internal",
namespace: "research",
topic: "ready-to-release",
subscription: "rss-feed", <====== required for a `:consumer`
handler: Publicizer.MessageHandler
]
]
Stargate.Supervisor.start_link(options)
Readers and Consumers share the same configuration API with the two key differences that the
`:consumer` key in the options differentiates from the `:reader` key, as well as the requirement
to provide a `"subscription"` to a consumer for the cluster to manage messages.
"""
defdelegate produce(url_or_connection, message), to: Stargate.Producer
defdelegate produce(connection, message, mfa), to: Stargate.Producer
@type tenant :: String.t()
@type namespace :: String.t()
@type topic :: String.t()
@type persistence :: String.t()
@type component :: :producer | :producer_ack | :consumer | :consumer_ack | :reader | :reader_ack
@type key_opt ::
{:persistence, persistence()}
| {:name, atom()}
| {:registry, atom()}
| {:component, component()}
@doc """
Generate the via-tuple needed for addressing a process within the Stargate supervision tree. Expects
at minimum the tenant, namespace, and topic of the process being addressed and assumes by default the
desired process is the Producer of a persistent topic managed by the default supervisor/registry.
iex> Stargate.registry_key("foo", "bar", "baz")
{:via, Registry, {:sg_reg_default, {:producer, "persistent", "foo", "bar", "baz"}}}
iex> Stargate.registry_key("foo", "bar", "baz", registry: MyCustom.Registry, persistence: "non-persistent", component: :producer_ack)
{:via, Registry, {MyCustom.Registry, {:producer_ack, "non-persistent", "foo", "bar", "baz"}}}
"""
@spec registry_key(tenant(), namespace(), topic(), [key_opt]) ::
{:via, Registry, {atom(), {component(), persistence(), tenant(), namespace(), topic()}}}
def registry_key(tenant, namespace, topic, opts \\ []) do
name = Keyword.get(opts, :name, :default)
registry = Keyword.get(opts, :registry) || :"sg_reg_#{name}"
component = Keyword.get(opts, :component, :producer)
persistence = Keyword.get(opts, :persistence, "persistent")
{:via, Registry, {registry, {component, persistence, tenant, namespace, topic}}}
end
defmodule Message do
@moduledoc """
Defines the Elixir Struct that represents the structure of a Pulsar message.
The struct combines the "location" data of the received messages (persistent vs. non-persistent,
tenant, namespace, topic) with the payload, any key and/or properties provided with the message,
and the publication timestamp as an DateTime struct, and the messageId assigned by the cluster.
### Example
message = %Stargate.Message{
topic: "ready-for-release",
namespace: "research",
tenant: "internal",
persistence: "persistent",
message_id: "CAAQAw==",
payload: "Hello World",
key: "1234",
properties: nil,
publish_time: ~U[2020-01-10 18:13:34.443264Z]
}
"""
@type t :: %__MODULE__{
topic: String.t(),
namespace: String.t(),
tenant: String.t(),
persistence: String.t(),
message_id: String.t(),
payload: String.t(),
key: String.t(),
properties: map(),
publish_time: DateTime.t()
}
defstruct [
:topic,
:namespace,
:tenant,
:persistence,
:message_id,
:payload,
:key,
:properties,
:publish_time
]
@doc """
Create a %Stargate.Message{} struct from a list of arguments. Takes the map decoded from
the json message payload received from Pulsar and adds the tenant, namespace, topic, persistence
information to maintain "location awareness" of a message's source topic.
Creating a %Stargate.Message{} via the `new/5` function automatically converts the ISO8601-formatted
publish timestamp to a DateTime struct and decodes the message payload from the Base64 encoding
received from the cluster.
"""
@spec new(map(), String.t(), String.t(), String.t(), String.t()) :: Stargate.Message.t()
def new(message, persistence, tenant, namespace, topic) do
{:ok, timestamp, _} =
message
|> Map.get("publishTime")
|> DateTime.from_iso8601()
payload =
message
|> Map.get("payload")
|> Base.decode64!()
%Message{
topic: topic,
namespace: namespace,
tenant: tenant,
persistence: persistence,
message_id: message["messageId"],
payload: payload,
key: Map.get(message, "key", ""),
properties: Map.get(message, "properties", %{}),
publish_time: timestamp
}
end
end
end
