-
Notifications
You must be signed in to change notification settings - Fork 9
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: create module to publish trip modifications to MQTT (#2679)
* feat: derive `Jason.Encoder` for `TripModification` structs * feat(ex/mix/deps): add `emqtt_fallover` library * feat(ex/config/runtime): get MQTT variables from env in prod * init(ex/Skate.MqttConnection): source connection adapter from `ride_along` * feat(ex/Skate.MqttConnection): prefix message topic before publishing * init: trip_modification_publisher * init(ex/detours/trip_modification_publisher): tests init: trip_modification_publisher * feat(gh/actions/ci): add `:mqtt` integration tests to CI mbta/api@6382198
- Loading branch information
Showing
13 changed files
with
335 additions
and
3 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,147 @@ | ||
| defmodule Skate.Detours.TripModificationPublisher do | ||
| @moduledoc """ | ||
| Connects to the MQTT Broker, then allows sending `Realtime.TripModification`'s | ||
| to the Broker. | ||
| References: https://github.com/mbta/ride_along/blob/fb440fb15dce22921fc4a141a125f3645da98b26/lib/ride_along/sql_publisher.ex | ||
| ## Testing | ||
| To "unit test" this code, you must have an MQTT broker running locally, or | ||
| configure URL's for a MQTT Broker which you have access too. | ||
| By default, these "integration tests" are excluded from the test suite, | ||
| under the tag `"Test.Integration": :mqtt`. So to run these tests | ||
| `--include 'Test.Integration:mqtt'` to run only the MQTT tests; | ||
| or you can use `--include 'Test.Integration'` to target all instances of | ||
| this tag. | ||
| ### Example: | ||
| > #### Note {: .tip} | ||
| > This example requires that you have the `mosquitto` mqtt broker binary in | ||
| > your `$PATH`. | ||
| > | ||
| > Ex: `brew install mosquitto` | ||
| ```sh | ||
| # Start and run `mosquitto` broker in the background | ||
| mosquitto & | ||
| # Wait for `mosquitto` to launch then launch integration tests with | ||
| mix test --include 'Test.Integration' test/skate/detours/trip_modification_publisher_test.exs | ||
| ``` | ||
| Alternatively, instead of running something locally with the default | ||
| `config :skate, Skate.MqttConnection` config in `config/config.exs` configuration; | ||
| You can configure the `config :skate, Skate.MqttConnection` in | ||
| `config/runtime.exs` to run in the `Mix` env `:dev` and configure those | ||
| environment variables locally to connect to a different Broker; | ||
| Or you can edit the config `config/dev.exs` to statically configure it for | ||
| local development | ||
| """ | ||
| use GenServer | ||
|
|
||
| @default_name __MODULE__ | ||
| def start_link(opts) do | ||
| if opts[:start] do | ||
| name = Keyword.get(opts, :name, @default_name) | ||
| GenServer.start_link(__MODULE__, opts, name: name) | ||
| else | ||
| :ignore | ||
| end | ||
| end | ||
|
|
||
| @doc """ | ||
| Publishes a `Realtime.TripModification` to the configured `MQTT` server. | ||
| MQTT is optional for Skate, and callers should remember to handle both the | ||
| `:ok` and `:error` return values | ||
| """ | ||
| def publish_modification( | ||
| %Realtime.TripModification{} = modification, | ||
| opts \\ [] | ||
| ) do | ||
| is_draft? = Keyword.get(opts, :is_draft?, false) | ||
| server = Keyword.get(opts, :server, @default_name) | ||
|
|
||
| GenServer.call( | ||
| server, | ||
| { | ||
| :new_modification, | ||
| %{ | ||
| is_draft?: is_draft?, | ||
| modification: modification | ||
| } | ||
| } | ||
| ) | ||
| end | ||
|
|
||
| @type t :: %{ | ||
| connection: Skate.MqttConnection.on_start() | nil, | ||
| on_connect_subscribers: [pid()] | ||
| } | ||
| defstruct connection: nil, on_connect_subscribers: [] | ||
|
|
||
| @impl GenServer | ||
| def init(opts) do | ||
| state = %__MODULE__{ | ||
| on_connect_subscribers: Keyword.get_values(opts, :on_connect) | ||
| } | ||
|
|
||
| {:ok, state, {:continue, :connect}} | ||
| end | ||
|
|
||
| @impl GenServer | ||
| def handle_continue(:connect, state) do | ||
| {:ok, connection} = Skate.MqttConnection.start_link() | ||
|
|
||
| {:noreply, | ||
| %{ | ||
| state | ||
| | connection: connection | ||
| }} | ||
| end | ||
|
|
||
| @impl GenServer | ||
| def handle_info({:connected, _connection}, %__MODULE__{} = state) do | ||
| Enum.each(state.on_connect_subscribers, &send(&1, {:connected, self()})) | ||
|
|
||
| {:noreply, state} | ||
| end | ||
|
|
||
| def handle_info({:disconnected, _, _reason}, state) do | ||
| {:noreply, state} | ||
| end | ||
|
|
||
| @impl GenServer | ||
| def handle_call( | ||
| {:new_modification, %{is_draft?: is_draft?, modification: modification}}, | ||
| _from, | ||
| %__MODULE__{connection: connection} = state | ||
| ) | ||
| when not is_nil(connection) do | ||
| id = Ecto.UUID.generate() | ||
|
|
||
| res = | ||
| Skate.MqttConnection.publish(connection, %EmqttFailover.Message{ | ||
| topic: trip_modification_topic(id), | ||
| payload: | ||
| Jason.encode!(%{ | ||
| data: modification, | ||
| meta: %{ | ||
| is_draft?: is_draft? | ||
| } | ||
| }), | ||
| # Send at least once | ||
| qos: 1 | ||
| }) | ||
|
|
||
| { | ||
| :reply, | ||
| {res, id}, | ||
| state | ||
| } | ||
| end | ||
|
|
||
| def trip_modification_topic(id), do: "#{trip_modifications_topic(id)}/trip_modification" | ||
| defp trip_modifications_topic(id), do: "trip_modifications/#{id}" | ||
| end |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,42 @@ | ||
| defmodule Skate.MqttConnection do | ||
| @moduledoc """ | ||
| Shared functionality to connect to the MQTT broker. | ||
| Sourced From: https://github.com/mbta/ride_along/blob/ffdb1007332c3eaf6f36244531468f67333bc292/lib/ride_along/mqtt_connection.ex | ||
| """ | ||
|
|
||
| @type on_start :: GenServer.on_start() | ||
| @type server :: GenServer.server() | ||
|
|
||
| @spec start_link(listen_topics :: [String.t()]) :: on_start() | ||
| def start_link(topics \\ []) do | ||
| app_config = app_config() | ||
|
|
||
| EmqttFailover.Connection.start_link( | ||
| configs: app_config[:broker_configs], | ||
| client_id: EmqttFailover.client_id(prefix: app_config[:broker_client_prefix]), | ||
| backoff: {1_000, 60_000, :jitter}, | ||
| handler: {EmqttFailover.ConnectionHandler.Parent, parent: self(), topics: topics} | ||
| ) | ||
| end | ||
|
|
||
| @spec publish(server(), EmqttFailover.Message.t()) :: :ok | {:error, term()} | ||
| def publish(connection, message) do | ||
| EmqttFailover.Connection.publish(connection, prefix_topic(message)) | ||
| end | ||
|
|
||
| @spec prefix_topic(EmqttFailover.Message.t()) :: EmqttFailover.Message.t() | ||
| @spec prefix_topic(binary()) :: binary() | ||
| def prefix_topic(%EmqttFailover.Message{topic: topic} = message) do | ||
| %{ | ||
| message | ||
| | topic: prefix_topic(topic) | ||
| } | ||
| end | ||
|
|
||
| def prefix_topic(topic) when is_binary(topic), do: (topic_prefix() || "") <> topic | ||
|
|
||
| def topic_prefix, do: app_config()[:broker_topic_prefix] | ||
|
|
||
| defp app_config, do: Application.get_env(:skate, __MODULE__) | ||
| end |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,2 @@ | ||
| listener 1883 | ||
| allow_anonymous true |
Oops, something went wrong.