|
| 1 | +defmodule NervesHub.AWSIoT do |
| 2 | + @moduledoc """ |
| 3 | + Support for common AWS IOT infrastructure including MQTT and SQS |
| 4 | +
|
| 5 | + Requires `:queues` to be defined in the application config or |
| 6 | + the supervisor is simply ignored |
| 7 | +
|
| 8 | + See docs.nerves-hub.org for a general overview of the architecture |
| 9 | + """ |
| 10 | + use Supervisor |
| 11 | + |
| 12 | + alias NervesHub.Tracker |
| 13 | + |
| 14 | + @type opt :: {:queues, [keyword()]} |
| 15 | + @spec start_link([opt]) :: Supervisor.on_start() |
| 16 | + def start_link(opts) do |
| 17 | + Supervisor.start_link(__MODULE__, opts, name: __MODULE__) |
| 18 | + end |
| 19 | + |
| 20 | + @impl Supervisor |
| 21 | + def init(opts) do |
| 22 | + opts = |
| 23 | + Application.get_env(:nerves_hub, __MODULE__, []) |
| 24 | + |> Keyword.merge(opts) |
| 25 | + |
| 26 | + case opts[:queues] do |
| 27 | + queues when is_list(queues) and length(queues) > 0 -> |
| 28 | + children = |
| 29 | + Enum.map(queues, &{__MODULE__.SQS, &1}) |
| 30 | + |> maybe_add_local_broker(opts) |
| 31 | + |
| 32 | + Supervisor.init(children, strategy: :one_for_one) |
| 33 | + |
| 34 | + _ -> |
| 35 | + :ignore |
| 36 | + end |
| 37 | + end |
| 38 | + |
| 39 | + defp maybe_add_local_broker(children, opts) do |
| 40 | + if broker_spec = opts[:local_broker] do |
| 41 | + [broker_spec | children] |
| 42 | + else |
| 43 | + children |
| 44 | + end |
| 45 | + end |
| 46 | + |
| 47 | + if Application.compile_env(:nerves_hub, [__MODULE__, :local_broker], false) do |
| 48 | + def publish(serial, event, payload) do |
| 49 | + data = Jason.encode!(%{event: event, payload: payload}) |
| 50 | + PintBroker.publish(__MODULE__.PintBroker, "nh/#{serial}", data) |
| 51 | + end |
| 52 | + else |
| 53 | + def publish(serial, event, payload) do |
| 54 | + # TODO: Topic and data may change soon |
| 55 | + # Stubbing out initial idea here for now |
| 56 | + data = %{event: event, payload: payload} |
| 57 | + topic = "/topics/nh/#{serial}" |
| 58 | + |
| 59 | + ExAws.Operation.JSON.new(:iot_data, %{path: topic, data: data}) |
| 60 | + |> ExAws.request() |
| 61 | + end |
| 62 | + end |
| 63 | + |
| 64 | + defmodule SQS do |
| 65 | + @moduledoc """ |
| 66 | + Consumer for AWS SQS messages |
| 67 | +
|
| 68 | + This is the ingestion point of devices coming from the MQTT |
| 69 | + broker. A message from a device must include the `"identifier"` |
| 70 | + key either in the payload or pulled from the topic via the |
| 71 | + AWS IoT rule that forwards to the queue. |
| 72 | +
|
| 73 | + The system must also be setup with a rule to forward [AWS Lifecycle |
| 74 | + events](https://docs.aws.amazon.com/iot/latest/developerguide/life-cycle-events.html) |
| 75 | + to a queue for tracking device online/offline presence |
| 76 | +
|
| 77 | + Right now, all configured queues are handled by this module. |
| 78 | + In the future, we may want to separate handling for each |
| 79 | + queue in it's own module. |
| 80 | + """ |
| 81 | + use Broadway |
| 82 | + |
| 83 | + alias Broadway.Message |
| 84 | + alias NervesHub.Devices |
| 85 | + |
| 86 | + require Logger |
| 87 | + |
| 88 | + def start_link(opts), do: Broadway.start_link(__MODULE__, opts) |
| 89 | + |
| 90 | + @impl Broadway |
| 91 | + def handle_message(_processor, %{data: raw} = msg, _context) do |
| 92 | + case Jason.decode(raw) do |
| 93 | + {:ok, data} -> |
| 94 | + Message.put_data(msg, data) |
| 95 | + |> process_message() |
| 96 | + |
| 97 | + _ -> |
| 98 | + Message.failed(msg, :malformed) |
| 99 | + end |
| 100 | + end |
| 101 | + |
| 102 | + @impl Broadway |
| 103 | + def handle_batch(_batcher, messages, batch_info, _context) do |
| 104 | + Logger.debug("[SQS] Handled #{inspect(batch_info.size)}") |
| 105 | + messages |
| 106 | + end |
| 107 | + |
| 108 | + defp process_message(%{data: %{"eventType" => "connected"} = data} = msg) do |
| 109 | + # TODO: Maybe use more info from the connection? |
| 110 | + # Example payload of AWS lifecycle connected event |
| 111 | + # principalIdentifier is a SHA256 fingerprint of the certificate that |
| 112 | + # is Base16 encoded |
| 113 | + # { |
| 114 | + # "clientId": "186b5", |
| 115 | + # "timestamp": 1573002230757, |
| 116 | + # "eventType": "connected", |
| 117 | + # "sessionIdentifier": "a4666d2a7d844ae4ac5d7b38c9cb7967", |
| 118 | + # "principalIdentifier": "12345678901234567890123456789012", |
| 119 | + # "ipAddress": "192.0.2.0", |
| 120 | + # "versionNumber": 0 |
| 121 | + # } |
| 122 | + |
| 123 | + with {:ok, device} <- Devices.get_by_identifier(data["clientId"]) do |
| 124 | + Logger.debug("[AWS IoT] device #{device.identifier} connected") |
| 125 | + |
| 126 | + Tracker.online(device) |
| 127 | + |
| 128 | + msg |
| 129 | + else |
| 130 | + _ -> |
| 131 | + Message.failed(msg, :unknown_device) |
| 132 | + end |
| 133 | + end |
| 134 | + |
| 135 | + defp process_message(%{data: %{"eventType" => "disconnected"} = data} = msg) do |
| 136 | + # TODO: Maybe use more of the disconnect data? |
| 137 | + # Example payload of AWS lifecyle disconnect event |
| 138 | + # { |
| 139 | + # "clientId": "186b5", |
| 140 | + # "timestamp": 1573002340451, |
| 141 | + # "eventType": "disconnected", |
| 142 | + # "sessionIdentifier": "a4666d2a7d844ae4ac5d7b38c9cb7967", |
| 143 | + # "principalIdentifier": "12345678901234567890123456789012", |
| 144 | + # "clientInitiatedDisconnect": true, |
| 145 | + # "disconnectReason": "CLIENT_INITIATED_DISCONNECT", |
| 146 | + # "versionNumber": 0 |
| 147 | + # } |
| 148 | + with {:ok, device} <- Devices.get_by_identifier(data["clientId"]) do |
| 149 | + Logger.debug( |
| 150 | + "[AWS IoT] device #{device.identifier} disconnected: #{data["disconnectReason"]}" |
| 151 | + ) |
| 152 | + |
| 153 | + Tracker.offline(device) |
| 154 | + end |
| 155 | + |
| 156 | + msg |
| 157 | + end |
| 158 | + |
| 159 | + defp process_message(msg) do |
| 160 | + # TODO: Track unhandled msg |
| 161 | + msg |
| 162 | + end |
| 163 | + end |
| 164 | +end |
0 commit comments