@@ -10,6 +10,12 @@ defmodule RoomSanctum.Worker.Ankyra do
|
10
|
10
|
# Asking is cheap and answering is not: a vision's queries all run again. |
|
11
|
11
|
@request_floor_ms 1_000 |
|
12
|
12
|
|
|
|
13
|
+ # How long a reported position is worth showing, and how many to keep. Five |
|
|
14
|
+ # minutes is a trail rather than a history: long enough to see which way |
|
|
15
|
+ # somebody is going, short enough that it is gone before it is a record. |
|
|
16
|
+ @positions_ttl_s 300 |
|
|
17
|
+ @positions_kept 20 |
|
|
18
|
+ |
|
13
|
19
|
@registry :zeus |
|
14
|
20
|
def start_link(opts) do |
|
15
|
21
|
GenServer.start_link(__MODULE__, opts, name: via_tuple("ankyra" <> opts[:id])) |
@@ -34,7 +40,8 @@ defmodule RoomSanctum.Worker.Ankyra do
|
34
|
40
|
initial_delay: 10 |
|
35
|
41
|
) |
|
36
|
42
|
|
|
37
|
|
− {:ok, %{id: opts[:id], ankyra: nil, requests: nil, last_request: nil}} |
|
|
43
|
+ {:ok, |
|
|
44
|
+ %{id: opts[:id], ankyra: nil, requests: nil, last_request: nil, positions: []}} |
|
38
|
45
|
end |
|
39
|
46
|
|
|
40
|
47
|
defp via_tuple(name), do: {:via, Registry, {@registry, name}} |
@@ -68,15 +75,21 @@ defmodule RoomSanctum.Worker.Ankyra do
|
68
|
75
|
|
|
69
|
76
|
def handle_cast(:meta_check, %{ankyra: nil} = state), do: {:noreply, state} |
|
70
|
77
|
|
|
71
|
|
− def handle_cast(:meta_check, state) do |
|
72
|
|
− parent = self() |
|
73
|
|
− username = state.ankyra.username |
|
|
78
|
+ # Positions age out whether or not anything is arriving, so the pruning |
|
|
79
|
+ # rides the check that already runs every couple of seconds. Without it a |
|
|
80
|
+ # client that stopped moving would sit on the page indefinitely. |
|
|
81
|
+ def handle_cast(:meta_check, %{positions: [_ | _]} = state) do |
|
|
82
|
+ pruned = prune(state.positions) |
|
74
|
83
|
|
|
75
|
|
− spawn(fn -> send(parent, {:status_result, connected_clients(username)}) end) |
|
|
84
|
+ if pruned != state.positions do |
|
|
85
|
+ broadcast_positions(state.id, pruned) |
|
|
86
|
+ end |
|
76
|
87
|
|
|
77
|
|
− {:noreply, state} |
|
|
88
|
+ do_meta_check(%{state | positions: pruned}) |
|
78
|
89
|
end |
|
79
|
90
|
|
|
|
91
|
+ def handle_cast(:meta_check, state), do: do_meta_check(state) |
|
|
92
|
+ |
|
80
|
93
|
def handle_cast(:refresh_db_cfg, state) do |
|
81
|
94
|
p = Accounts.get_rabbit_user!(state[:id]) |
|
82
|
95
|
{:noreply, state |> Map.put(:ankyra, p) |> listen_for_requests()} |
@@ -112,8 +125,11 @@ defmodule RoomSanctum.Worker.Ankyra do
|
112
|
125
|
def handle_info({:basic_cancel, _meta}, state), do: {:noreply, state |> Map.put(:requests, nil)} |
|
113
|
126
|
def handle_info({:basic_cancel_ok, _meta}, state), do: {:noreply, state} |
|
114
|
127
|
|
|
115
|
|
− def handle_info({:basic_deliver, _payload, %{routing_key: key}}, state) do |
|
116
|
|
− {:noreply, serve_request(state, key)} |
|
|
128
|
+ def handle_info({:basic_deliver, payload, %{routing_key: key}}, state) do |
|
|
129
|
+ cond do |
|
|
130
|
+ String.contains?(key, ".up.") -> {:noreply, remember_position(state, payload)} |
|
|
131
|
+ true -> {:noreply, serve_request(state, key)} |
|
|
132
|
+ end |
|
117
|
133
|
end |
|
118
|
134
|
|
|
119
|
135
|
def handle_info({:status_result, count}, state) do |
@@ -199,6 +215,7 @@ defmodule RoomSanctum.Worker.Ankyra do
|
199
|
215
|
{:ok, chan} <- AMQP.Channel.open(conn), |
|
200
|
216
|
{:ok, %{queue: queue}} <- AMQP.Queue.declare(chan, "", exclusive: true, auto_delete: true), |
|
201
|
217
|
:ok <- AMQP.Queue.bind(chan, queue, "amq.topic", routing_key: request_key(state)), |
|
|
218
|
+ :ok <- AMQP.Queue.bind(chan, queue, "amq.topic", routing_key: uplink_key(state)), |
|
202
|
219
|
{:ok, tag} <- AMQP.Basic.consume(chan, queue, self(), no_ack: true) do |
|
203
|
220
|
Logger.info("ankyra #{state.id} listening for requests on #{request_key(state)}") |
|
204
|
221
|
state |> Map.put(:requests, tag) |> Map.put(:requests_chan, chan) |
@@ -211,6 +228,8 @@ defmodule RoomSanctum.Worker.Ankyra do
|
211
|
228
|
|
|
212
|
229
|
defp request_key(state), do: state.ankyra.topic <> ".publish.#" |
|
213
|
230
|
|
|
|
231
|
+ defp uplink_key(state), do: state.ankyra.topic <> ".up.#" |
|
|
232
|
+ |
|
214
|
233
|
# The consumer's own lifecycle. Nothing to do but let them through. |
|
215
|
234
|
|
|
216
|
235
|
# Serve a request, if one is due. |
@@ -246,6 +265,63 @@ defmodule RoomSanctum.Worker.Ankyra do
|
246
|
265
|
end |
|
247
|
266
|
end |
|
248
|
267
|
|
|
|
268
|
+ defp do_meta_check(state) do |
|
|
269
|
+ parent = self() |
|
|
270
|
+ username = state.ankyra.username |
|
|
271
|
+ |
|
|
272
|
+ spawn(fn -> send(parent, {:status_result, connected_clients(username)}) end) |
|
|
273
|
+ |
|
|
274
|
+ {:noreply, state} |
|
|
275
|
+ end |
|
|
276
|
+ |
|
|
277
|
+ # Where a client said it was, kept for as long as it is worth showing. |
|
|
278
|
+ # |
|
|
279
|
+ # In memory and nowhere else: this is a trail on a page somebody is looking |
|
|
280
|
+ # at, not a record of where anybody has been. It goes when the worker does, |
|
|
281
|
+ # and each fix goes five minutes after it arrived. |
|
|
282
|
+ defp remember_position(state, payload) do |
|
|
283
|
+ case decode_position(payload) do |
|
|
284
|
+ nil -> |
|
|
285
|
+ state |
|
|
286
|
+ |
|
|
287
|
+ position -> |
|
|
288
|
+ positions = prune([position | state.positions]) |> Enum.take(@positions_kept) |
|
|
289
|
+ broadcast_positions(state.id, positions) |
|
|
290
|
+ %{state | positions: positions} |
|
|
291
|
+ end |
|
|
292
|
+ end |
|
|
293
|
+ |
|
|
294
|
+ defp decode_position(payload) do |
|
|
295
|
+ with {:ok, %{"lat" => lat, "lon" => lon} = fix} <- Poison.decode(payload), |
|
|
296
|
+ true <- is_number(lat) and is_number(lon) do |
|
|
297
|
+ %{ |
|
|
298
|
+ lat: lat, |
|
|
299
|
+ lon: lon, |
|
|
300
|
+ accuracy_m: Map.get(fix, "accuracy_m"), |
|
|
301
|
+ speed_mps: Map.get(fix, "speed_mps"), |
|
|
302
|
+ client_id: Map.get(fix, "client_id"), |
|
|
303
|
+ # When it reached us, not when the phone says it was taken: a clock |
|
|
304
|
+ # that is wrong should not put a fix in the future or age it out early. |
|
|
305
|
+ at: DateTime.utc_now() |
|
|
306
|
+ } |
|
|
307
|
+ else |
|
|
308
|
+ _ -> nil |
|
|
309
|
+ end |
|
|
310
|
+ end |
|
|
311
|
+ |
|
|
312
|
+ defp prune(positions) do |
|
|
313
|
+ cutoff = DateTime.add(DateTime.utc_now(), -@positions_ttl_s, :second) |
|
|
314
|
+ Enum.filter(positions, fn p -> DateTime.compare(p.at, cutoff) == :gt end) |
|
|
315
|
+ end |
|
|
316
|
+ |
|
|
317
|
+ defp broadcast_positions(id, positions) do |
|
|
318
|
+ Phoenix.PubSub.broadcast( |
|
|
319
|
+ RoomSanctum.PubSub, |
|
|
320
|
+ topic(id), |
|
|
321
|
+ {:ankyra_positions, positions} |
|
|
322
|
+ ) |
|
|
323
|
+ end |
|
|
324
|
+ |
|
249
|
325
|
defp too_soon?(nil, _now), do: false |
|
250
|
326
|
|
|
251
|
327
|
defp too_soon?(last, now), do: DateTime.diff(now, last, :millisecond) < @request_floor_ms |