Current section
4 Versions
Jump to
Current section
4 Versions
Compare versions
7
files changed
+211
additions
-89
deletions
| @@ -10,7 +10,7 @@ by adding `phoenix_pubsub_nats` to your list of dependencies in `mix.exs`: | |
| 10 10 | ```elixir |
| 11 11 | def deps do |
| 12 12 | [ |
| 13 | - {:phoenix_pubsub_nats, "~> 0.1.0"} |
| 13 | + {:phoenix_pubsub_nats, "~> 0.2.1"} |
| 14 14 | ] |
| 15 15 | end |
| 16 16 | ``` |
| @@ -42,4 +42,4 @@ An example usage (add this to your supervision tree): | |
| 42 42 | * `Nats.Serializer.Native`: Convert messages using Erlang's native serialization mechanism. |
| 43 43 | * `Nats.Serializer.Json`: Convert messages using Json format. |
| 44 44 | |
| 45 | - |
| 45 | + For more details on how to use it, see the [Phoenix.PubSub documentation](https://hexdocs.pm/phoenix_pubsub/Phoenix.PubSub.html). |
| @@ -4,6 +4,7 @@ | |
| 4 4 | {<<"elixir">>,<<"~> 1.14">>}. |
| 5 5 | {<<"files">>, |
| 6 6 | [<<"lib">>,<<"lib/nats">>,<<"lib/nats/nats_pubsub.ex">>, |
| 7 | + <<"lib/nats/consumer.ex">>,<<"lib/nats/consumer_supervisor.ex">>, |
| 7 8 | <<"lib/nats/serializer">>,<<"lib/nats/serializer/native.ex">>, |
| 8 9 | <<"lib/nats/serializer/json.ex">>,<<"lib/nats/serializer.ex">>, |
| 9 10 | <<"lib/phoenix_pubsub_nats.ex">>,<<"mix.exs">>,<<"README.md">>, |
| @@ -28,4 +29,4 @@ | |
| 28 29 | {<<"optional">>,false}, |
| 29 30 | {<<"repository">>,<<"hexpm">>}, |
| 30 31 | {<<"requirement">>,<<"~> 1.6">>}]]}. |
| 31 | - {<<"version">>,<<"0.1.0">>}. |
| 32 | + {<<"version">>,<<"0.2.1">>}. |
| @@ -0,0 +1,111 @@ | |
| 1 | + defmodule Nats.Consumer do |
| 2 | + @moduledoc """ |
| 3 | + Node subscriber |
| 4 | + """ |
| 5 | + use Gnat.Server |
| 6 | + require Logger |
| 7 | + |
| 8 | + def request( |
| 9 | + %{ |
| 10 | + body: body, |
| 11 | + headers: headers, |
| 12 | + topic: topic |
| 13 | + } = event |
| 14 | + ) do |
| 15 | + Logger.debug("Received request #{inspect(event)}") |
| 16 | + current_node = to_string(node()) |
| 17 | + |
| 18 | + with {:convert_headers, %{"serializer_type" => serializer_type} = headers} <- |
| 19 | + {:convert_headers, conver_headers_to_map(headers)}, |
| 20 | + {:decode, |
| 21 | + %{ |
| 22 | + name: name, |
| 23 | + dispatcher: dispatcher, |
| 24 | + from: from, |
| 25 | + payload: message, |
| 26 | + pubsub_name: _pubsub_name |
| 27 | + }} <- {:decode, decode(serializer_type, body)} do |
| 28 | + target = get_target(headers) |
| 29 | + real_topic = get_real_topic(topic) |
| 30 | + |
| 31 | + cond do |
| 32 | + from == current_node -> |
| 33 | + # This node is the source, nothing to do, because local dispatch already |
| 34 | + # happened. |
| 35 | + :ok |
| 36 | + |
| 37 | + not is_nil(target) and target != current_node -> |
| 38 | + # when not is_nil(target) and target != current_node -> |
| 39 | + # Direct broadcast and this is not the destination node. |
| 40 | + :ok |
| 41 | + |
| 42 | + true -> |
| 43 | + res = |
| 44 | + Phoenix.PubSub.local_broadcast( |
| 45 | + name, |
| 46 | + real_topic, |
| 47 | + message, |
| 48 | + maybe_convert_to_existing_atom(dispatcher) |
| 49 | + ) |
| 50 | + |
| 51 | + res |
| 52 | + end |
| 53 | + else |
| 54 | + {:convert_headers, msg} -> |
| 55 | + Logger.warning("Unable to parse headers. Details: #{inspect(msg)}") |
| 56 | + :ok |
| 57 | + |
| 58 | + {:decode, msg} -> |
| 59 | + Logger.warning("Unable to decode the message. Details: #{inspect(msg)}") |
| 60 | + :ok |
| 61 | + |
| 62 | + error -> |
| 63 | + Logger.warning("Unable to process pubsub message. Details: #{inspect(error)}") |
| 64 | + :ok |
| 65 | + end |
| 66 | + end |
| 67 | + |
| 68 | + def error(%{gnat: _gnat, reply_to: _reply_to}, error) do |
| 69 | + Logger.error( |
| 70 | + "Error on #{inspect(__MODULE__)} during handle incoming message. Error #{inspect(error)}" |
| 71 | + ) |
| 72 | + end |
| 73 | + |
| 74 | + defp decode(serializer_type, body) do |
| 75 | + serializer = |
| 76 | + case serializer_type do |
| 77 | + "json" -> |
| 78 | + Nats.Serializer.Json |
| 79 | + |
| 80 | + _ -> |
| 81 | + Nats.Serializer.Native |
| 82 | + end |
| 83 | + |
| 84 | + case serializer.decode(body, []) do |
| 85 | + {:ok, payload} -> |
| 86 | + payload |
| 87 | + |
| 88 | + _ -> |
| 89 | + raise ArgumentError, "Could not deserialize data: #{inspect(body)}" |
| 90 | + end |
| 91 | + end |
| 92 | + |
| 93 | + defp get_target(headers) do |
| 94 | + if Map.has_key?(headers, "target") do |
| 95 | + Map.get(headers, "target", nil) |
| 96 | + else |
| 97 | + nil |
| 98 | + end |
| 99 | + end |
| 100 | + |
| 101 | + defp get_real_topic(topic) do |
| 102 | + String.replace(topic, "nodes.pubsub.", "") |
| 103 | + end |
| 104 | + |
| 105 | + defp conver_headers_to_map(headers), do: Enum.into(headers, %{}) |
| 106 | + |
| 107 | + defp maybe_convert_to_existing_atom(string) when is_binary(string), |
| 108 | + do: String.to_existing_atom(string) |
| 109 | + |
| 110 | + defp maybe_convert_to_existing_atom(atom) when is_atom(atom), do: atom |
| 111 | + end |
| @@ -0,0 +1,38 @@ | |
| 1 | + defmodule Nats.ConsumerSupervisor do |
| 2 | + @moduledoc false |
| 3 | + use Supervisor |
| 4 | + require Logger |
| 5 | + |
| 6 | + @main_topic "nodes.pubsub.*" |
| 7 | + |
| 8 | + def start_link(state) do |
| 9 | + Supervisor.start_link(__MODULE__, state, name: __MODULE__) |
| 10 | + end |
| 11 | + |
| 12 | + def child_spec(state) do |
| 13 | + %{ |
| 14 | + id: __MODULE__, |
| 15 | + start: {__MODULE__, :start_link, [state]} |
| 16 | + } |
| 17 | + end |
| 18 | + |
| 19 | + @impl true |
| 20 | + def init(_state) do |
| 21 | + Logger.debug("Initializing Nats PubSub consumer for Node #{inspect(node())}") |
| 22 | + connection_name = Module.concat(PhoenixPubsubNats, "Connection") |
| 23 | + |
| 24 | + connection_params = %{ |
| 25 | + connection_name: connection_name, |
| 26 | + module: Nats.Consumer, |
| 27 | + subscription_topics: [ |
| 28 | + %{topic: @main_topic} |
| 29 | + ] |
| 30 | + } |
| 31 | + |
| 32 | + children = [ |
| 33 | + {Gnat.ConsumerSupervisor, connection_params} |
| 34 | + ] |
| 35 | + |
| 36 | + Supervisor.init(children, strategy: :one_for_one) |
| 37 | + end |
| 38 | + end |
| @@ -7,19 +7,16 @@ defmodule Nats.NatsPubsub do | |
| 7 7 | require Logger |
| 8 8 | |
| 9 9 | defmodule State do |
| 10 | - defstruct id: nil, pubsub_name: nil, gnat: nil, subscription: nil, serializer: nil |
| 10 | + defstruct id: nil, name: nil, gnat: nil, serializer: nil |
| 11 11 | |
| 12 12 | @type t :: %__MODULE__{ |
| 13 13 | id: Node.t(), |
| 14 | - pubsub_name: String.t(), |
| 14 | + name: String.t(), |
| 15 15 | gnat: any(), |
| 16 | - subscription: String.t(), |
| 17 16 | serializer: module() |
| 18 17 | } |
| 19 18 | end |
| 20 19 | |
| 21 | - @main_topic "nodes.pubsub.*" |
| 22 | - |
| 23 20 | @doc false |
| 24 21 | def start_link(opts) do |
| 25 22 | GenServer.start_link(__MODULE__, opts, name: opts[:adapter_name]) |
| @@ -43,59 +40,62 @@ defmodule Nats.NatsPubsub do | |
| 43 40 | metadata = Keyword.put(metadata, :from, node()) |
| 44 41 | metadata = Keyword.put(metadata, :dispatcher, dispatcher) |
| 45 42 | |
| 46 | - GenServer.call(adapter_name, {:broadcast, topic, message, metadata}) |
| 43 | + GenServer.call(adapter_name, {:broadcast, topic, message, metadata, adapter_name}) |
| 47 44 | end |
| 48 45 | |
| 49 46 | ## GenServer Callbacks |
| 50 47 | @impl true |
| 51 48 | @doc false |
| 52 | - def init(state) do |
| 49 | + def init(opts) do |
| 53 50 | Process.flag(:trap_exit, true) |
| 54 | - nats_conn = Keyword.fetch!(state, :connection) |
| 55 | - pubsub_name = Keyword.get(state, :adapter_name, __MODULE__) |
| 56 | - serializer = Keyword.get(state, :serializer, Nats.Serializer.Native) |
| 57 | - {:ok, %State{pubsub_name: pubsub_name}, {:continue, {:setup, nats_conn, serializer}}} |
| 51 | + name = Keyword.fetch!(opts, :name) |
| 52 | + nats_conn = Keyword.fetch!(opts, :connection) |
| 53 | + serializer = Keyword.get(opts, :serializer, Nats.Serializer.Native) |
| 54 | + {:ok, %State{name: name}, {:continue, {:setup, nats_conn, serializer}}} |
| 58 55 | end |
| 59 56 | |
| 60 57 | @impl true |
| 61 58 | def handle_continue({:setup, nats_conn, serializer}, state) do |
| 62 | - {gnat, subscription} = |
| 59 | + {:ok, gnat} = |
| 63 60 | case Gnat.start_link(nats_conn) do |
| 64 61 | {:ok, gnat} -> |
| 65 | - subscription = sub(gnat) |
| 66 | - {gnat, subscription} |
| 62 | + {:ok, gnat} |
| 67 63 | |
| 68 64 | _ -> |
| 69 65 | raise RuntimeError, "Could not connect to Nats with options #{inspect(nats_conn)}" |
| 70 66 | end |
| 71 67 | |
| 72 | - {:noreply, |
| 73 | - %State{state | id: node(), gnat: gnat, subscription: subscription, serializer: serializer}} |
| 74 | - end |
| 75 | - |
| 76 | - defp sub(gnat) do |
| 77 | - case Gnat.sub(gnat, self(), @main_topic) do |
| 78 | - {:ok, subscription} -> |
| 79 | - subscription |
| 80 | - |
| 81 | - _ -> |
| 82 | - raise RuntimeError, |
| 83 | - "Unable to subscribe Node #{inspect(node())} to channel #{@main_topic}" |
| 84 | - end |
| 68 | + {:noreply, %State{state | id: node(), gnat: gnat, serializer: serializer}} |
| 85 69 | end |
| 86 70 | |
| 87 71 | @impl true |
| 88 72 | def handle_call( |
| 89 | - {:broadcast, topic, message, metadata}, |
| 73 | + {:broadcast, topic, message, metadata, adapter_name} = _event, |
| 90 74 | _from, |
| 91 | - %State{gnat: gnat, serializer: serializer} = state |
| 75 | + %State{name: name, gnat: gnat, serializer: serializer} = state |
| 92 76 | ) do |
| 77 | + from = Keyword.get(metadata, :from, node()) |
| 78 | + dispatcher = Keyword.get(metadata, :dispatcher) |
| 79 | + serializer_type = get_serializer_type(serializer) |
| 93 80 | headers = build_headers(metadata) |
| 94 81 | |
| 82 | + headers = |
| 83 | + (headers ++ [{"serializer_type", serializer_type}]) |
| 84 | + |> List.flatten() |
| 85 | + |
| 86 | + msg = %{ |
| 87 | + dispatcher: dispatcher, |
| 88 | + from: to_string(from), |
| 89 | + name: name, |
| 90 | + payload: message, |
| 91 | + pubsub_name: adapter_name |
| 92 | + } |
| 93 | + |
| 95 94 | res = |
| 96 | - case serializer.encode(message) do |
| 95 | + case serializer.encode(msg) do |
| 97 96 | {:ok, payload} -> |
| 98 | - Gnat.pub(gnat, "nodes.pubsub.#{topic}", payload, headers: headers) |
| 97 | + topic = "nodes.pubsub.#{topic}" |
| 98 | + Gnat.pub(gnat, topic, payload, headers: headers) |
| 99 99 | |
| 100 100 | _ -> |
| 101 101 | :error |
| @@ -105,38 +105,6 @@ defmodule Nats.NatsPubsub do | |
| 105 105 | end |
| 106 106 | |
| 107 107 | @impl true |
| 108 | - def handle_info( |
| 109 | - {:msg, %{topic: topic, reply_to: _to, headers: headers} = event}, |
| 110 | - %State{pubsub_name: pubsub_name, serializer: serializer} = state |
| 111 | - ) do |
| 112 | - current_node = to_string(node()) |
| 113 | - |
| 114 | - case conver_headers_to_map(headers) do |
| 115 | - %{target: target} |
| 116 | - when not is_nil(target) and target != current_node -> |
| 117 | - # Direct broadcast and this is not the destination node. |
| 118 | - :ok |
| 119 | - |
| 120 | - %{from: ^current_node} -> |
| 121 | - # This node is the source, nothing to do, because local dispatch already |
| 122 | - # happened. |
| 123 | - :ok |
| 124 | - |
| 125 | - %{dispatcher: dispatcher} -> |
| 126 | - real_topic = get_real_topic(topic) |
| 127 | - payload = decode(serializer, event) |
| 128 | - |
| 129 | - Phoenix.PubSub.local_broadcast( |
| 130 | - pubsub_name, |
| 131 | - real_topic, |
| 132 | - payload, |
| 133 | - maybe_convert_to_existing_atom(dispatcher) |
| 134 | - ) |
| 135 | - end |
| 136 | - |
| 137 | - {:noreply, state} |
| 138 | - end |
| 139 | - |
| 140 108 | def handle_info(event, state) do |
| 141 109 | Logger.warning("An unexpected event has occurred. Event: #{inspect(event)}") |
| 142 110 | {:noreply, state} |
| @@ -147,6 +115,16 @@ defmodule Nats.NatsPubsub do | |
| 147 115 | Logger.warning("#{inspect(__MODULE__)} terminate with reason: #{inspect(reason)}") |
| 148 116 | end |
| 149 117 | |
| 118 | + defp get_serializer_type(serializer) do |
| 119 | + case serializer do |
| 120 | + Nats.Serializer.Json -> |
| 121 | + "json" |
| 122 | + |
| 123 | + _ -> |
| 124 | + "native" |
| 125 | + end |
| 126 | + end |
| 127 | + |
| 150 128 | defp build_headers([]), do: [] |
| 151 129 | |
| 152 130 | defp build_headers(metadata) when is_list(metadata) do |
| @@ -158,25 +136,4 @@ defmodule Nats.NatsPubsub do | |
| 158 136 | end |
| 159 137 | |
| 160 138 | defp build_headers(_), do: [] |
| 161 | - |
| 162 | - defp decode(serializer, %{body: body} = event) do |
| 163 | - case serializer.decode(body) do |
| 164 | - {:ok, payload} -> |
| 165 | - payload |
| 166 | - |
| 167 | - _ -> |
| 168 | - raise ArgumentError, "Could not deserialize event: #{inspect(event)}" |
| 169 | - end |
| 170 | - end |
| 171 | - |
| 172 | - defp get_real_topic(topic) do |
| 173 | - String.replace(topic, "nodes.pubsub.", "") |
| 174 | - end |
| 175 | - |
| 176 | - defp conver_headers_to_map(headers), do: Enum.into(headers, %{}) |
| 177 | - |
| 178 | - defp maybe_convert_to_existing_atom(string) when is_binary(string), |
| 179 | - do: String.to_existing_atom(string) |
| 180 | - |
| 181 | - defp maybe_convert_to_existing_atom(atom) when is_atom(atom), do: atom |
| 182 139 | end |
Loading more files…