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.

Let a client ask for a board

9c01b9b · Gabriel Morell · 2026-09-03 16:22

16 files +118 -4
Message 
{commit_body(@commit)}

Files changed

modified apps/room_hermes/lib/room_hermes_web/controllers/topic_controller.ex
+16 −2
@@ -54,15 +54,29 @@ defmodule RoomHermesWeb.TopicController do
54 54 def permitted?(topic, "read", routing_key), do: String.starts_with?(routing_key, topic)
55 55
56 56 def permitted?(topic, "write", routing_key),
57 − do: String.starts_with?(routing_key, uplink_prefix(topic))
57 + do: Enum.any?(uplink_prefixes(topic), &String.starts_with?(routing_key, &1))
58 58
59 59 def permitted?(_topic, _permission, _routing_key), do: false
60 60
61 61 @doc """
62 − Where a client publishes to, for a given Ankyra topic.
62 + Where a client may publish to, for a given Ankyra topic.
63 +
64 + Two channels, both inbound:
65 +
66 + * `<topic>.up.` — what the client is, most of it what it knows about
67 + itself. A location, for a foci that travels with the phone.
68 + * `<topic>.publish.` — what the client wants, which is currently a board.
69 + A Pythiae publishes on change and on its own tick, so a client that has
70 + just woken up can be looking at something old; this is how it asks.
63 71
64 72 Named here rather than in each caller so the two ends of the rule cannot
65 73 drift: this is both what the broker allows and what a client should send to.
66 74 """
75 + def uplink_prefixes(topic), do: [topic <> ".up.", topic <> ".publish."]
76 +
77 + @doc "The uplink a client reports itself on."
67 78 def uplink_prefix(topic), do: topic <> ".up."
79 +
80 + @doc "The channel a client asks for something on."
81 + def request_prefix(topic), do: topic <> ".publish."
68 82 end
modified ?
+0 −0

Binary file — not rendered.

modified ?
+0 −0

Binary file — not rendered.

modified ?
+0 −0

Binary file — not rendered.

modified apps/room_sanctum/lib/room_sanctum/configuration.ex
+18 −0
@@ -618,6 +618,24 @@ defmodule RoomSanctum.Configuration do
618 618 """
619 619 def get_pythiae!(id), do: Repo.get!(Pythiae, id)
620 620
621 + @doc """
622 + Every Pythiae that publishes to a given Ankyra.
623 +
624 + The reverse of the list a Pythiae carries, for the Ankyra end: a client's
625 + request arrives on a topic, which identifies the Ankyra, and what has to act
626 + on it is whatever publishes there.
627 + """
628 + def list_pythiae(:ankyra, id) when is_binary(id) do
629 + case Integer.parse(id) do
630 + {n, ""} -> list_pythiae(:ankyra, n)
631 + _ -> []
632 + end
633 + end
634 +
635 + def list_pythiae(:ankyra, id) do
636 + from(p in Pythiae, where: ^id in p.ankyra) |> Repo.all()
637 + end
638 +
621 639 def get_pythiae!(:name, name), do: Repo.get_by(Pythiae, name: name)
622 640
623 641 def get_pythiae(:vision, id) do
modified apps/room_sanctum/lib/room_sanctum/workers/ankyra_worker.ex
+84 −2
@@ -5,6 +5,10 @@ defmodule RoomSanctum.Worker.Ankyra do
5 5 require Logger
6 6
7 7 alias RoomSanctum.Accounts
8 + alias RoomSanctum.Configuration
9 +
10 + # Asking is cheap and answering is not: a vision's queries all run again.
11 + @request_floor_ms 1_000
8 12
9 13 @registry :zeus
10 14 def start_link(opts) do
@@ -30,7 +34,7 @@ defmodule RoomSanctum.Worker.Ankyra do
30 34 initial_delay: 10
31 35 )
32 36
33 − {:ok, %{id: opts[:id], ankyra: nil}}
37 + {:ok, %{id: opts[:id], ankyra: nil, requests: nil, last_request: nil}}
34 38 end
35 39
36 40 defp via_tuple(name), do: {:via, Registry, {@registry, name}}
@@ -75,7 +79,7 @@ defmodule RoomSanctum.Worker.Ankyra do
75 79
76 80 def handle_cast(:refresh_db_cfg, state) do
77 81 p = Accounts.get_rabbit_user!(state[:id])
78 − {:noreply, state |> Map.put(:ankyra, p)}
82 + {:noreply, state |> Map.put(:ankyra, p) |> listen_for_requests()}
79 83 end
80 84
81 85 def handle_cast({:publish, data}, state) do
@@ -104,6 +108,14 @@ defmodule RoomSanctum.Worker.Ankyra do
104 108 {:noreply, state}
105 109 end
106 110
111 + def handle_info({:basic_consume_ok, _meta}, state), do: {:noreply, state}
112 + def handle_info({:basic_cancel, _meta}, state), do: {:noreply, state |> Map.put(:requests, nil)}
113 + def handle_info({:basic_cancel_ok, _meta}, state), do: {:noreply, state}
114 +
115 + def handle_info({:basic_deliver, _payload, %{routing_key: key}}, state) do
116 + {:noreply, serve_request(state, key)}
117 + end
118 +
107 119 def handle_info({:status_result, count}, state) do
108 120 Phoenix.PubSub.broadcast(
109 121 RoomSanctum.PubSub,
@@ -168,6 +180,76 @@ defmodule RoomSanctum.Worker.Ankyra do
168 180 # itself throws -- so a broker that is merely unreachable arrives here as an
169 181 # exception, and unrescued it would kill the spawned checker with no status
170 182 # ever sent and the badge stuck on "checking" forever.
183 + # Listen for a client asking for something.
184 + #
185 + # A Pythiae publishes on change and on its own tick, so a client that has just
186 + # woken up -- a phone out of a pocket, a panel after a power cut -- can be
187 + # looking at a board from some time ago with no way to ask for a fresh one.
188 + # This is the way to ask.
189 + #
190 + # The queue is exclusive and auto-deleting: it belongs to this worker and
191 + # should go when it does, rather than accumulating requests for a process that
192 + # is not there to serve them.
193 + defp listen_for_requests(%{requests: tag} = state) when not is_nil(tag), do: state
194 +
195 + defp listen_for_requests(%{ankyra: nil} = state), do: state
196 +
197 + defp listen_for_requests(state) do
198 + with {:ok, conn} <- AMQP.Application.get_connection(:default),
199 + {:ok, chan} <- AMQP.Channel.open(conn),
200 + {:ok, %{queue: queue}} <- AMQP.Queue.declare(chan, "", exclusive: true, auto_delete: true),
201 + :ok <- AMQP.Queue.bind(chan, queue, "amq.topic", routing_key: request_key(state)),
202 + {:ok, tag} <- AMQP.Basic.consume(chan, queue, self(), no_ack: true) do
203 + Logger.info("ankyra #{state.id} listening for requests on #{request_key(state)}")
204 + state |> Map.put(:requests, tag) |> Map.put(:requests_chan, chan)
205 + else
206 + error ->
207 + Logger.warning("ankyra #{state.id} could not listen for requests: #{inspect(error)}")
208 + state
209 + end
210 + end
211 +
212 + defp request_key(state), do: state.ankyra.topic <> ".publish.#"
213 +
214 + # The consumer's own lifecycle. Nothing to do but let them through.
215 +
216 + # Serve a request, if one is due.
217 + #
218 + # Debounced: asking is cheap and a vision's queries are not, so a client in a
219 + # loop would have the server running them as fast as it could. One a second is
220 + # more than enough to feel immediate, and the Pythiae's own tick covers
221 + # everything else.
222 + defp serve_request(state, key) do
223 + now = DateTime.utc_now()
224 +
225 + if too_soon?(state.last_request, now) do
226 + state
227 + else
228 + for pythiae <- Configuration.list_pythiae(:ankyra, state.id) do
229 + serve(pythiae, key)
230 + end
231 +
232 + Map.put(state, :last_request, now)
233 + end
234 + end
235 +
236 + # What was asked for, from the end of the routing key. `.img` gets the board
237 + # drawn; anything else gets the board itself, since that is what a client
238 + # that has just woken up wants and the reason to guess at all.
239 + defp serve(pythiae, key) do
240 + id = to_string(pythiae.id)
241 +
242 + if String.ends_with?(key, ".img") do
243 + RoomSanctum.Worker.Pythiae.publish_img(id)
244 + else
245 + RoomSanctum.Worker.Pythiae.query_current_now(id)
246 + end
247 + end
248 +
249 + defp too_soon?(nil, _now), do: false
250 +
251 + defp too_soon?(last, now), do: DateTime.diff(now, last, :millisecond) < @request_floor_ms
252 +
171 253 defp fetch_connections(url, cfg) do
172 254 HTTPoison.get(
173 255 String.trim_trailing(url, "/") <> "/api/connections?columns=protocol,user,vhost,client_properties.client_id",

Parents: fb8519f