我希望在Phoenix应用程序中反复运行相同的操作序列,而不会在工作进程中发生任何故障时使整个Web应用程序崩溃,但我真的不知道是否应该使用GenServer、Elixir的Tasks、代理或者其他一些我还没有想到的东西。
当我启动我的Phoenix应用程序时,一个工作进程也应该启动,定期从串行连接中提取一些值,通过Phoenix通道广播这些值,收集它们,直到达到@save_interval
,然后计算中位数,通过另一个通道广播该中位数并将其写入InfluxDB。目前我有类似以下的代码(基本上可以工作):
def do_your_thing(serial_pid) do
Stream.interval(@interval_live)
|> get_new_values_from_serial(serial_pid)
|> broadcast!("live-channel:#{@name}")
|> Enum.take(div(@interval_save, @interval_live))
|> calculate_medians()
|> broadcast!("update-channel:#{@name}")
|> write_to_database()
do_your_thing(serial_pid) # repeat
end
我才刚开始理解OTP相关的内容,希望有人能够帮助我朝着正确的方向前进。