Trigger PA Messages from RTS #764
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,191 @@ | ||
| defmodule Engine.PaMessages do | ||
| use GenServer | ||
| require Logger | ||
|
|
||
| alias PaMessages.PaMessage | ||
|
|
||
| @type state :: %{ | ||
| pa_message_timers_table: :ets.tab() | ||
| } | ||
|
|
||
| @pa_message_timers_table :pa_message_timers | ||
| @minute_in_ms 1000 * 60 | ||
|
|
||
| def start_link([]) do | ||
| GenServer.start_link(__MODULE__, [], name: __MODULE__) | ||
| end | ||
|
|
||
| @impl true | ||
| def init(_) do | ||
| # Add some delay to wait for sign processes to come up before sending PA messages | ||
| Process.send_after(self(), :update, 10000) | ||
| state = %{pa_message_timers_table: @pa_message_timers_table} | ||
| create_table(state) | ||
| {:ok, state} | ||
| end | ||
|
|
||
| def create_table(state) do | ||
| :ets.new(state.pa_message_timers_table, [:named_table, read_concurrency: true]) | ||
|
There was a problem hiding this comment. Do we gain anything from using ETS here? It looks like the only access is inside this module, so there shouldn't be any contention to worry about. There was a problem hiding this comment. Oh that is true. Originally I had planned that the sign processes would be making requests to access the table so started with this and left it, but we can simplify things and go without the table |
||
| end | ||
|
|
||
| @impl true | ||
| def handle_info(:update, state) do | ||
| schedule_update(self()) | ||
|
|
||
| case get_active_pa_messages() do | ||
| {:ok, pa_messages} -> | ||
| schedule_pa_messages(pa_messages, state.pa_message_timers_table) | ||
| |> handle_inactive_pa_messages(state.pa_message_timers_table) | ||
|
|
||
| {:error, %HTTPoison.Response{status_code: status_code, body: body}} -> | ||
| Logger.error("pa_messages_response_error: status_code=#{status_code} body=#{body}") | ||
|
|
||
| {:error, %HTTPoison.Error{reason: reason}} -> | ||
| Logger.error("pa_messages_response_error: reason=#{reason}") | ||
|
|
||
| {:error, error} -> | ||
| Logger.error("pa_messages_response_error: error=#{inspect(error)}") | ||
| end | ||
|
|
||
| {:noreply, state} | ||
| end | ||
|
|
||
| @impl true | ||
| def handle_info({:send_pa_message, pa_message}, state) do | ||
| Enum.each(pa_message.sign_ids, fn sign_id -> | ||
| send( | ||
| String.to_existing_atom("Signs/#{sign_id}"), | ||
| {:play_pa_message, pa_message} | ||
| ) | ||
| end) | ||
|
|
||
| {:noreply, state} | ||
| end | ||
|
|
||
| defp schedule_pa_messages(pa_messages, table) do | ||
| for %{ | ||
| "id" => pa_id, | ||
| "visual_text" => visual_text, | ||
| "audio_text" => audio_text, | ||
| "interval_in_minutes" => interval_in_minutes, | ||
| "priority" => priority, | ||
| "sign_ids" => sign_ids | ||
| } <- pa_messages do | ||
| active_pa_message = %PaMessage{ | ||
| id: pa_id, | ||
| visual_text: visual_text, | ||
| audio_text: audio_text, | ||
| priority: priority, | ||
| sign_ids: sign_ids, | ||
| interval_in_ms: interval_in_minutes * @minute_in_ms | ||
| } | ||
|
|
||
| case get_pa_message_timer(pa_id, table) do | ||
| {timer_ref, existing_pa_message} | ||
| when existing_pa_message.interval_in_ms != active_pa_message.interval_in_ms -> | ||
| case Process.read_timer(timer_ref) do | ||
| false -> | ||
| schedule_pa_message(active_pa_message, active_pa_message.interval_in_ms, table) | ||
|
|
||
| remaining_ms -> | ||
| ms_elapsed = existing_pa_message.interval_in_ms - remaining_ms | ||
| temp_interval = (active_pa_message.interval_in_ms - ms_elapsed) |> max(0) | ||
| cancel_pa_timer(timer_ref, pa_id) | ||
| schedule_pa_message(active_pa_message, temp_interval, table) | ||
|
There was a problem hiding this comment. The timer-based approach will certainly work, but now that I'm looking at the code, I wonder if it's a little more complex than we need. I know the original motivation was to avoid excessive polling, but now that all the logic is centralized in this engine (which I think is good), we could also just keep a map of id-to-last-playtime in state, and check all the deltas against the current time once per second (especially since we're already doing an There was a problem hiding this comment. Hmm yeah I think that approach works nicely and will make the "edited interval" piece of the logic a good deal simpler. And it removes the need to cancel timers and such as you hinted at. I think we also would get the "handling" of inactive messages for free as well since if they're not included in the response, we won't be checking for the last playtime anyway. |
||
| end | ||
|
|
||
| {timer, _} -> | ||
| if Process.read_timer(timer) == false do | ||
| schedule_pa_message(active_pa_message, active_pa_message.interval_in_ms, table) | ||
| end | ||
|
|
||
| nil -> | ||
| schedule_pa_message(active_pa_message, 0, table) | ||
| end | ||
|
|
||
| pa_id | ||
| end | ||
| end | ||
|
|
||
| defp schedule_pa_message(pa_message, interval_in_ms, table) do | ||
| Logger.info( | ||
| "pa_message: action=scheduled id=#{pa_message.id} interval_ms=#{interval_in_ms} sign_ids=#{inspect(pa_message.sign_ids)}" | ||
| ) | ||
|
|
||
| timer_ref = | ||
| Process.send_after( | ||
| self(), | ||
| {:send_pa_message, pa_message}, | ||
| interval_in_ms | ||
| ) | ||
|
|
||
| :ets.insert( | ||
| table, | ||
| {pa_message.id, {timer_ref, pa_message}} | ||
| ) | ||
| end | ||
|
|
||
| defp handle_inactive_pa_messages(active_pa_ids, table) do | ||
| :ets.tab2list(table) | ||
| |> Enum.each(fn {pa_id, {timer_ref, pa_message}} -> | ||
| if pa_id not in active_pa_ids do | ||
| cancel_pa_timer(timer_ref, pa_id) | ||
| delete_pa_message(pa_message, table) | ||
| end | ||
| end) | ||
| end | ||
|
|
||
| defp cancel_pa_timer(timer_ref, pa_id) do | ||
| Logger.info("pa_message: action=timer_canceled id=#{pa_id}") | ||
| Process.cancel_timer(timer_ref) | ||
| end | ||
|
|
||
| defp delete_pa_message(pa_message, table) do | ||
| Logger.info("pa_message: action=message_deleted id=#{pa_message.id}") | ||
|
|
||
| Enum.each(pa_message.sign_ids, fn sign_id -> | ||
| send( | ||
| String.to_existing_atom("Signs/#{sign_id}"), | ||
| {:delete_pa_message, pa_message.id} | ||
| ) | ||
| end) | ||
|
|
||
| :ets.delete(table, pa_message.id) | ||
| end | ||
|
|
||
| defp get_pa_message_timer(pa_id, table) do | ||
| case :ets.lookup(table, pa_id) do | ||
| [{^pa_id, timer}] -> timer | ||
| _ -> nil | ||
| end | ||
| end | ||
|
|
||
| defp get_active_pa_messages() do | ||
| active_pa_messages_url = | ||
| Application.get_env(:realtime_signs, :screenplay_url) <> | ||
| Application.get_env(:realtime_signs, :active_pa_messages_path) | ||
|
|
||
| http_client = Application.get_env(:realtime_signs, :http_client) | ||
|
|
||
| with {:ok, response} <- | ||
| http_client.get( | ||
| active_pa_messages_url, | ||
| [ | ||
| {"x-api-key", Application.get_env(:realtime_signs, :screenplay_api_key)} | ||
| ], | ||
| timeout: 2000, | ||
| recv_timeout: 2000 | ||
| ), | ||
| %{status_code: 200, body: body} <- response, | ||
| {:ok, pa_messages} <- Jason.decode(body) do | ||
| {:ok, pa_messages} | ||
| else | ||
| error -> | ||
| {:error, error} | ||
| end | ||
| end | ||
|
|
||
| defp schedule_update(pid) do | ||
| Process.send_after(pid, :update, 1_000) | ||
| end | ||
| end | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,27 @@ | ||
| defmodule PaMessages.PaMessage do | ||
| defstruct id: nil, | ||
| visual_text: nil, | ||
| audio_text: nil, | ||
| priority: nil, | ||
| sign_ids: [], | ||
| interval_in_ms: nil | ||
|
|
||
| @type t :: %__MODULE__{ | ||
| id: integer(), | ||
| visual_text: String.t(), | ||
| audio_text: String.t(), | ||
| priority: integer(), | ||
| sign_ids: [String.t()], | ||
| interval_in_ms: non_neg_integer() | ||
| } | ||
|
|
||
| defimpl Content.Audio do | ||
| def to_params(%PaMessages.PaMessage{visual_text: visual_text}) do | ||
| {:ad_hoc, {visual_text, :audio_visual}} | ||
| end | ||
|
|
||
| def to_tts(%PaMessages.PaMessage{visual_text: visual_text, audio_text: audio_text}) do | ||
| {audio_text, PaEss.Utilities.paginate_text(visual_text)} | ||
| end | ||
| end | ||
| end |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
What's the rationale for this being a runtime environment variable? If we just need a testing hook, then just using the application environment should be sufficient.
As a side note, it would be nice at some point to convert the
Fakehttp module to useMoxinstead, which would make testing more straightforward, but that will probably take a bit of effort.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Hmm yeah that's a good point. I think my rationale was just that the path could change since the screenplay api is relatively new and still being developed, but I guess it would be just as much effort to make the devops change rather than just make a code change in RTS or probably even less. I'll make a change to remove this one