GITGUD

neiam neiam /apollos_crib

It's Apollo's Crib

public · Issues · Pulls 1 · Labels · Forks · Compare · Actions failure · Badges · Packages

⭐

Log in to mark this repository.

url_rt_shared: one URL for an agency that publishes one feed

58b065b · Gabriel Morell · 2026-08-25 21:21

18 files +113 -91
Message 
{commit_body(@commit)}

Files changed

modified ?
+0 −0

Binary file — not rendered.

modified ?
+0 −0

Binary file — not rendered.

modified apps/room_gtfs/lib/room_gtfs/gtfs_worker.ex
+91 −90
@@ -329,23 +329,18 @@ defmodule RoomGtfs.Worker.RT do
329 329 # whether it worked or not. The logs already say when one fails; what they
330 330 # cannot say is that a feed stopped being fetched at all, which is the failure
331 331 # that looks exactly like silence. See RoomGtfs.FeedHealth.
332 − defp fetch_recorded(state, kind, url) do
333 − started = System.monotonic_time(:millisecond)
334 −
335 − # A source polling every three minutes can happily use data up to three
336 − # minutes old, and several sources sharing one URL -- which is the whole
337 − # point of the regional feed -- then fetch it once between them instead of
338 − # once each.
339 − result =
340 − case rt_period_ms(state) do
341 − nil -> FeedCache.get(url)
342 − period -> FeedCache.get(url, period - :timer.seconds(5))
343 − end
344 − took = System.monotonic_time(:millisecond) - started
332 + # A source polling every three minutes can happily use data up to three
333 + # minutes old, and several sources sharing one URL -- which is the whole point
334 + # of the regional feed -- then fetch it once between them instead of once each.
335 + defp cached_get(state, url) do
336 + case rt_period_ms(state) do
337 + nil -> FeedCache.get(url)
338 + period -> FeedCache.get(url, period - :timer.seconds(5))
339 + end
340 + end
345 341
342 + defp record_health(state, kind, url, result, took) do
346 343 RoomGtfs.FeedHealth.record(state.id, state.inst && state.inst.name, kind, url, result, took)
347 −
348 − result
349 344 end
350 345
351 346 @doc """
@@ -784,93 +779,99 @@ defmodule RoomGtfs.Worker.RT do
784 779
785 780 defp do_update_realtime(state) do
786 781 state =
787 − case state.inst.enabled do
788 − true ->
789 − state =
790 − case state.inst.config |> Map.get(:url_rt_sa) do
791 − nil ->
792 − state
782 + if state.inst.enabled do
783 + Enum.reduce([:sa, :tu, :vp], state, &fetch_kind/2)
784 + else
785 + bcast(state.id, :disabled)
786 + state
787 + end
793 788
794 − val ->
795 − case fetch_recorded(state, :sa, val) do
796 − {:ok, data_sa} when is_struct(data_sa, TransitRealtime.FeedMessage) ->
797 − state |> Map.put(:rt_sa, for_this_agency(data_sa, state))
789 + {:noreply, state}
790 + end
798 791
799 − {:error, error} ->
800 − Logger.info(
801 − "failed to fetch gtfs-rt url[sa] for '#{state.inst.name}', reason: #{inspect(error)}"
802 − )
792 + # Where each kind comes from. A kind with no URL of its own falls back to the
793 + # combined feed, so an agency publishing everything in one message -- the MTA
794 + # subway feeds are 67 trip updates, 45 vehicle positions and an alert in a
795 + # single URL -- names it once instead of three times. Specific beats general,
796 + # so a combined feed plus a dedicated alerts feed works too.
797 + defp rt_url(config, :sa), do: Map.get(config, :url_rt_sa) || Map.get(config, :url_rt_shared)
798 + defp rt_url(config, :tu), do: Map.get(config, :url_rt_tu) || Map.get(config, :url_rt_shared)
799 + defp rt_url(config, :vp), do: Map.get(config, :url_rt_vp) || Map.get(config, :url_rt_shared)
803 800
804 − state
805 − end
806 − end
801 + defp fetch_kind(kind, state) do
802 + case rt_url(state.inst.config, kind) do
803 + nil ->
804 + state
805 +
806 + url ->
807 + started = System.monotonic_time(:millisecond)
808 + result = cached_get(state, url)
809 + took = System.monotonic_time(:millisecond) - started
810 +
811 + case result do
812 + {:ok, feed} when is_struct(feed, TransitRealtime.FeedMessage) ->
813 + # Narrowed before it is recorded, so health reports what this kind
814 + # actually got. Recording the raw feed would have every kind of a
815 + # combined source claim the same entity count -- exactly the reading
816 + # that means "the same URL is in all three fields by mistake", and
817 + # would say it about a correct configuration.
818 + feed = feed |> for_this_agency(state) |> slice(entity_field(kind))
819 + record_health(state, kind, url, {:ok, feed}, took)
820 + store_feed(state, kind, feed)
821 +
822 + {:error, error} ->
823 + record_health(state, kind, url, result, took)
824 +
825 + Logger.info(
826 + "failed to fetch gtfs-rt url[#{kind}] for '#{state.inst.name}', reason: #{inspect(error)}"
827 + )
807 828
808 − state =
809 − case state.inst.config |> Map.get(:url_rt_tu) do
810 − nil ->
811 − state
829 + state
830 + end
831 + end
832 + end
812 833
813 − val ->
814 − case fetch_recorded(state, :tu, val) do
815 − {:ok, data_tu} when is_struct(data_tu, TransitRealtime.FeedMessage) ->
816 − state |> Map.put(:rt_tu, for_this_agency(data_tu, state))
834 + defp entity_field(:sa), do: :alert
835 + defp entity_field(:tu), do: :trip_update
836 + defp entity_field(:vp), do: :vehicle
817 837
818 − {:error, error} ->
819 − Logger.info(
820 − "failed to fetch gtfs-rt url[tu] for '#{state.inst.name}', reason: #{inspect(error)}"
821 − )
838 + # Sliced to the kind it is being held as, always -- not only for a combined
839 + # feed. Three sources pointing at one mixed URL used to store the whole feed
840 + # three times, so the health metrics reported the same entity count for all
841 + # three kinds, which is indistinguishable from having pasted the same URL into
842 + # all three fields by mistake. Sliced, subway reads 67/45/1 and says what it
843 + # actually has. The query handlers filter by kind anyway, so this changes
844 + # nothing they see.
845 + defp store_feed(state, :sa, feed), do: Map.put(state, :rt_sa, feed)
822 846
823 − state
824 − end
825 − end
847 + defp store_feed(state, :tu, feed), do: Map.put(state, :rt_tu, feed)
826 848
827 − state =
828 − case state.inst.config |> Map.get(:url_rt_vp) do
829 − nil ->
830 − state
831 −
832 − val ->
833 − case fetch_recorded(state, :vp, val) do
834 − {:ok, data_vp} when is_struct(data_vp, TransitRealtime.FeedMessage) ->
835 − data_vp = for_this_agency(data_vp, state)
836 − new_state = state |> Map.put(:rt_vp, data_vp) |> ensure_stops()
837 −
838 − # Broadcast vehicle position updates
839 − vehicles = vehicle_positions_from(data_vp, new_state.stops)
840 −
841 − # Broadcast to specific source channel for source page
842 − Phoenix.PubSub.broadcast(
843 − RoomSanctum.PubSub,
844 − "gtfs_vehicle_positions:#{state.id}",
845 − {:vehicle_positions_updated, state.id, vehicles}
846 − )
847 −
848 − # Also broadcast to general channel for query pages
849 − Phoenix.PubSub.broadcast(
850 − RoomSanctum.PubSub,
851 − "gtfs_vehicle_positions",
852 − {:vehicle_positions_updated, vehicles}
853 − )
854 −
855 − new_state
856 −
857 − {:error, error} ->
858 − Logger.info(
859 − "failed to fetch gtfs-rt url[vp] for '#{state.inst.name}', reason: #{inspect(error)}"
860 − )
861 −
862 − state
863 − end
864 − end
849 + defp store_feed(state, :vp, feed) do
850 + state = state |> Map.put(:rt_vp, feed) |> ensure_stops()
851 + vehicles = vehicle_positions_from(feed, state.stops)
865 852
866 − false ->
867 − bcast(state.id, :disabled)
868 − state
869 − end
853 + # The source page watches its own channel; the query pages watch the shared
854 + # one and sort out which vehicles are theirs.
855 + Phoenix.PubSub.broadcast(
856 + RoomSanctum.PubSub,
857 + "gtfs_vehicle_positions:#{state.id}",
858 + {:vehicle_positions_updated, state.id, vehicles}
859 + )
870 860
871 − {:noreply, state}
861 + Phoenix.PubSub.broadcast(
862 + RoomSanctum.PubSub,
863 + "gtfs_vehicle_positions",
864 + {:vehicle_positions_updated, vehicles}
865 + )
866 +
867 + state
872 868 end
873 869
870 + defp slice(feed, field) do
871 + %{feed | entity: Enum.filter(feed.entity, &Map.get(&1, field))}
872 + end
873 +
874 +
874 875 def handle_call({:query_realtime, trips, stop}, _from, state) do
875 876 # IO.inspect(trips)
876 877
modified apps/room_sanctum/lib/room_sanctum/configuration/configs/gtfs.ex
+12 −1
@@ -9,6 +9,17 @@ defmodule RoomSanctum.Configuration.Configs.GTFS do
9 9 field :url_rt_sa, :string
10 10 field :url_rt_tu, :string
11 11 field :url_rt_vp, :string
12 +
13 + # One feed carrying more than one kind of entity. Plenty of agencies publish
14 + # trip updates, vehicle positions and alerts in three files; some put them
15 + # in one message, and the MTA's subway feeds are the clearest case -- a
16 + # single URL that is 67 trip updates, 45 vehicle positions and an alert.
17 + #
18 + # Any kind without a URL of its own is taken from here, so a source with a
19 + # combined feed sets this alone, and one with a combined feed plus a
20 + # separate alerts feed sets this and url_rt_sa. Specific beats general.
21 + field :url_rt_shared, :string
22 +
12 23 field :tz, :string
13 24
14 25 # Seconds between realtime polls. Nil polls at the default 30s, which is
@@ -28,7 +39,7 @@ defmodule RoomSanctum.Configuration.Configs.GTFS do
28 39
29 40 def changeset(source, params) do
30 41 source
31 − |> cast(params, ~w(url url_rt_sa url_rt_tu url_rt_vp tz rt_period rt_agency)a)
42 + |> cast(params, ~w(url url_rt_sa url_rt_tu url_rt_vp url_rt_shared tz rt_period rt_agency)a)
32 43 |> validate_required([:url, :tz])
33 44 # Below about ten seconds the poller cannot keep up and the feed will not
34 45 # have changed anyway; the upper bound is a day, past which it is not
modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/form_component.html.heex
+10 −0
@@ -79,6 +79,16 @@
79 79 <div class="form-control">
80 80 <.input field={gtfs_form[:url_rt_vp]} type="text" label="Realtime URL Vehicle Positions"/>
81 81 </div>
82 + <div class="form-control">
83 + <.input field={gtfs_form[:url_rt_shared]} type="text" label="Realtime URL (combined feed)"/>
84 + <label class="label">
85 + <span class="label-text-alt opacity-70">
86 + For an agency that publishes trip updates, vehicle positions and alerts in
87 + one feed &mdash; MTA subway does. Fills in whichever of the three above are
88 + left blank, and each is split out by entity type.
89 + </span>
90 + </label>
91 + </div>
82 92 <div class="form-control">
83 93 <.input field={gtfs_form[:rt_period]} type="number" label="Realtime poll interval (seconds)"/>
84 94 <label class="label">

Parents: d328973