feat: Add new multi-route vehicles channel

.tool-versions

@@ -1,2 +1,3 @@
 elixir 1.15.7-otp-26
 erlang 26.1.2
+direnv 2.32.3

lib/mobile_app_backend_web/channels/vehicles_for_route_channel.ex

@@ -2,6 +2,7 @@ defmodule MobileAppBackendWeb.VehiclesForRouteChannel do
   use MobileAppBackendWeb, :channel
 
   alias MBTAV3API.JsonApi
+  alias MBTAV3API.Stream
   alias MBTAV3API.Vehicle
 
   @throttle_ms 500
@@ -13,14 +14,14 @@ defmodule MobileAppBackendWeb.VehiclesForRouteChannel do
       {:ok, throttler} =
         MobileAppBackend.Throttler.start_link(target: self(), cast: :send_data, ms: @throttle_ms)
 
-      {:ok, vehicle_data} = MBTAV3API.Stream.StaticInstance.subscribe("vehicles")
+      {:ok, vehicle_data} = Stream.StaticInstance.subscribe("vehicles")
 
-      vehicle_data = filter_data(vehicle_data, route_id, direction_id)
+      vehicle_data = filter_data(vehicle_data, [route_id], direction_id)
 
       {:ok, vehicle_data,
        assign(socket,
          data: vehicle_data,
-         route_id: route_id,
+         route_ids: [route_id],
          direction_id: direction_id,
          throttler: throttler
        )}
@@ -29,10 +30,38 @@ defmodule MobileAppBackendWeb.VehiclesForRouteChannel do
     end
   end
 
+  @impl true
+  def join("vehicles:routes:" <> topic_param_concat, _payload, socket) do
+    case parse_params(topic_param_concat) do
+      {:ok, route_ids, direction_id} ->
+        {:ok, throttler} =
+          MobileAppBackend.Throttler.start_link(
+            target: self(),
+            cast: :send_data,
+            ms: @throttle_ms
+          )
+
+        {:ok, vehicle_data} = Stream.StaticInstance.subscribe("vehicles")
+
+        vehicle_data = filter_data(vehicle_data, route_ids, direction_id)
+
+        {:ok, vehicle_data,
+         assign(socket,
+           data: vehicle_data,
+           route_ids: route_ids,
+           direction_id: direction_id,
+           throttler: throttler
+         )}
+
+      _ ->
+        {:error, %{code: :missing_param}}
+    end
+  end
+
   @impl true
   def handle_info({:stream_data, "vehicles", data}, socket) do
     old_data = socket.assigns.data
-    new_data = filter_data(data, socket.assigns.route_id, socket.assigns.direction_id)
+    new_data = filter_data(data, socket.assigns.route_ids, socket.assigns.direction_id)
 
     if old_data != new_data do
       MobileAppBackend.Throttler.request(socket.assigns.throttler)
@@ -48,11 +77,28 @@ defmodule MobileAppBackendWeb.VehiclesForRouteChannel do
     {:noreply, socket}
   end
 
-  @spec filter_data(JsonApi.Object.full_map(), String.t(), 0 | 1) :: JsonApi.Object.full_map()
-  defp filter_data(vehicle_data, route_id, direction_id) do
+  @spec parse_params(String.t()) :: {:ok, [String.t()], 0 | 1} | :error
+  defp parse_params(param_string) do
+    if param_string == "" || !String.contains?(param_string, ":") do
+      :error
+    else
+      [route_string, direction_string] = String.split(param_string, ":", parts: 2)
+
+      case Integer.parse(direction_string) do
+        {direction_id, _} when direction_id in [0, 1] ->
+          {:ok, String.split(route_string, ","), direction_id}
+
+        _ ->
+          :error
+      end
+    end
+  end
+
+  @spec filter_data(JsonApi.Object.full_map(), [String.t()], 0 | 1) :: JsonApi.Object.full_map()
+  defp filter_data(vehicle_data, route_ids, direction_id) do
     update_in(vehicle_data.vehicles, fn vehicles ->
       Map.filter(vehicles, fn {_, %Vehicle{} = vehicle} ->
-        vehicle.route_id == route_id and vehicle.direction_id == direction_id
+        Enum.member?(route_ids, vehicle.route_id) and vehicle.direction_id == direction_id
       end)
     end)
   end

test/mobile_app_backend_web/channels/vehicles_for_route_channel_test.exs

@@ -67,4 +67,68 @@ defmodule MobileAppBackendWeb.VehiclesForRouteChannelTest do
     assert_push "stream_data", data
     assert data == to_full_map([good_vehicle1])
   end
+
+  test "joins multi route ok", %{socket: socket} do
+    route_ids = ["123", "456", "789"]
+    direction_id = 1
+    vehicle1 = build(:vehicle, route_id: Enum.at(route_ids, 0), direction_id: direction_id)
+    vehicle2 = build(:vehicle, route_id: Enum.at(route_ids, 1), direction_id: direction_id)
+    vehicle3 = build(:vehicle, route_id: Enum.at(route_ids, 2), direction_id: direction_id)
+
+    start_link_supervised!(
+      {FakeStaticInstance, topic: "vehicles", data: to_full_map([vehicle1, vehicle2, vehicle3])}
+    )
+
+    {:ok, reply, _socket} =
+      subscribe_and_join(
+        socket,
+        "vehicles:routes:#{Enum.join(route_ids, ",")}:#{direction_id}",
+        %{}
+      )
+
+    assert reply == to_full_map([vehicle1, vehicle2, vehicle3])
+  end
+
+  test "filters multi route to requested data", %{socket: socket} do
+    route_id_1 = "123"
+    route_id_2 = "456"
+    route_ids = [route_id_1, route_id_2]
+    direction_id = 0
+    good_vehicle1 = build(:vehicle, route_id: route_id_1, direction_id: direction_id)
+    good_vehicle2 = build(:vehicle, route_id: route_id_2, direction_id: direction_id)
+    bad_vehicle1 = build(:vehicle, route_id: "NOT-#{route_id_1}", direction_id: direction_id)
+    bad_vehicle2 = build(:vehicle, route_id: route_id_1, direction_id: 1 - direction_id)
+    bad_vehicle3 = build(:vehicle, route_id: "NOT-#{route_id_2}", direction_id: 1 - direction_id)
+
+    start_link_supervised!(
+      {FakeStaticInstance,
+       topic: "vehicles",
+       data: to_full_map([good_vehicle1, good_vehicle2, bad_vehicle1, bad_vehicle2, bad_vehicle3])}
+    )
+
+    {:ok, reply, _socket} =
+      subscribe_and_join(
+        socket,
+        "vehicles:routes:#{Enum.join(route_ids, ",")}:#{direction_id}",
+        %{}
+      )
+
+    assert reply == to_full_map([good_vehicle1, good_vehicle2])
+
+    Stream.PubSub.broadcast!(
+      "vehicles",
+      {:stream_data, "vehicles",
+       to_full_map([good_vehicle1, good_vehicle2, bad_vehicle1, bad_vehicle2])}
+    )
+
+    refute_push "stream_data", _
+
+    Stream.PubSub.broadcast!(
+      "vehicles",
+      {:stream_data, "vehicles", to_full_map([good_vehicle1, bad_vehicle2])}
+    )
+
+    assert_push "stream_data", data
+    assert data == to_full_map([good_vehicle1])
+  end
 end