feat: make MqttMediator config separate from the SOURCE envvar

This allows us to add new environment variables for the MQTT broker,
without breaking existing API instances which expect the source to be an
HTTPS URL.

apps/state_mediator/config/config.exs

@@ -29,6 +29,8 @@ config :state_mediator, State.Vehicle,
     "MBTA_VEHICLE_SOURCE",
     "https://cdn.mbta.com/realtime/VehiclePositions_enhanced.json"
   },
+  broker: {:system, "MBTA_VEHICLE_BROKER", nil},
+  topic: {:system, "MBTA_VEHICLE_TOPIC", nil},
   username: {:system, "MBTA_VEHICLE_USERNAME", nil},
   password: {:system, "MBTA_VEHICLE_PASSWORD", nil}
 

apps/state_mediator/lib/state_mediator.ex

@@ -36,7 +36,7 @@ defmodule StateMediator do
           interval: 10_000
         ]
       },
-      vehicle_mediator_child(source_url(State.Vehicle)),
+      vehicle_mediator_child(app_value(State.Vehicle, :broker), source_url(State.Vehicle)),
       {
         StateMediator.Mediator,
         [
@@ -65,46 +65,35 @@ defmodule StateMediator do
     []
   end
 
-  defp vehicle_mediator_child("mqtt" <> _ = url) do
+  defp vehicle_mediator_child(no_broker, url) when no_broker in ["", nil] do
     {
-      StateMediator.MqttMediator,
+      StateMediator.Mediator,
       [
         spec_id: :vehicle_mediator,
         state: State.Vehicle,
         url: url,
-        username: app_value(State.Vehicle, :username),
-        password: app_value(State.Vehicle, :password),
-        sync_timeout: 30_000
+        opts: [timeout: 10_000],
+        sync_timeout: 30_000,
+        interval: 1_000
       ]
     }
   end
 
-  defp vehicle_mediator_child(["mqtt" <> _ | _] = url) do
+  defp vehicle_mediator_child(broker, _url) do
     {
       StateMediator.MqttMediator,
       [
         spec_id: :vehicle_mediator,
         state: State.Vehicle,
-        url: url,
+        url: broker,
+        topic: app_value(State.Vehicle, :topic),
+        username: app_value(State.Vehicle, :username),
+        password: app_value(State.Vehicle, :password),
         sync_timeout: 30_000
       ]
     }
   end
 
-  defp vehicle_mediator_child(url) do
-    {
-      StateMediator.Mediator,
-      [
-        spec_id: :vehicle_mediator,
-        state: State.Vehicle,
-        url: url,
-        opts: [timeout: 10_000],
-        sync_timeout: 30_000,
-        interval: 1_000
-      ]
-    }
-  end
-
   @spec crowding_children(boolean()) :: [:supervisor.child_spec() | {module(), term()} | module()]
   defp crowding_children(true) do
     Logger.info("#{__MODULE__} CR_CROWDING_ENABLED=true")

apps/state_mediator/lib/state_mediator/mqtt_mediator.ex

@@ -50,7 +50,7 @@ defmodule StateMediator.MqttMediator do
     url = Keyword.fetch!(options, :url)
     configs = configs_from_url(url, options)
 
-    "/" <> topic = URI.decode_www_form(URI.parse(url).path)
+    topic = Keyword.fetch!(options, :topic)
 
     sync_timeout = options |> Keyword.get(:sync_timeout, 5000)
 

apps/state_mediator/test/state_mediator/mqtt_mediator_test.exs

@@ -28,7 +28,8 @@ defmodule StateMediator.MqttMediatorTest do
 
   # @moduletag capture_log: true
   @opts [
-    url: "mqtt://test.mosquitto.org/#{URI.encode_www_form("home/#")}",
+    url: "mqtt://test.mosquitto.org",
+    topic: "home/#",
     state: __MODULE__.StateModule
   ]