twitter - spawn 関数から GenServer 子プロセスを作成するには?

Question

ExTwitter Streamを使用してリアルタイムでツイートを追跡し、チャネル エンドポイント経由でブロードキャストしています。イベントごとに 1 つのプロセスを作成し、それに 1 つの Twitter ストリーム リスナーを割り当てたいと思います。次に、新しいサブスクライバーが同じイベントに参加すると、以前のストリーム状態を取得し、新しいツイートを受信して​​ブロードキャストします。

以下から GenServer プロセスを作成する方法:

stream = ExTwitter.stream_filter(track: hashtags)
pid = spawn(fn ->
  for tweet <- stream do
    IO.puts tweet.text
    MyApp.Endpoint.broadcast! "stream", "tweet", %{tweet: tweet.text}
  end
end)

次のモジュールの子として event_id で割り当てます。

defmodule MyApp.TwitterStream.Monitor do
  require IEx
  @moduledoc """
  Store twitter stream per event_id
  """
  use GenServer

  def create(event_id, hashtags, coords) do
    case GenServer.whereis(ref(event_id)) do
      nil ->
        Supervisor.start_child(MyApp.TwitterStream.Supervisor, [event_id, hashtags, coords])
      _twitter_stream ->
        IEx.pry
        # return previous ExTwitter stream state and broadcast new tweets
        {:error, :twitter_stream_already_exists}
    end
  end

  def start_link(event_id, hashtags, coords) do
    # stream = ExTwitter.stream_filter(track: hashtags)
    # pid = spawn(fn ->
    #   for tweet <- stream do
    #     IO.puts tweet.text
    #     MyApp.Endpoint.broadcast! "stream", "tweet", %{tweet: tweet.text}
    #   end
    # end)
    GenServer.start_link(__MODULE__, %{hashtags: hashtags, coords: coords}, name: ref(event_id))
  end

  def stop(event_id) do
    try_call(event_id, :stop)
  end

  def info(event_id) do
    try_call(event_id, :info)
  end

  def handle_call(:stop, _from, state) do
    # ExTwitter.stream_control(pid, :stop)
    {:stop, :normal, :ok, state}
  end

  def handle_call(:info, _from, state) do
    {:reply, state, state}
  end

  defp try_call(event_id, call_function) do
    case GenServer.whereis(ref(event_id)) do
      nil ->
        {:error, :invalid_twitter_stream}
      twitter_stream ->
        GenServer.call(twitter_stream, call_function)
    end
  end

  defp ref(event_id) do
    {:global, {:twitter_stream, event_id}}
  end
end

新しいツイートを受信する方法、または最終的にイベント モニターの外部で Twitter ストリームを停止する方法は?

スーパーバイザー:

defmodule MyApp.TwitterStream.Supervisor do
  use Supervisor

  def start_link do
    Supervisor.start_link(__MODULE__, :ok, name: __MODULE__)
  end

  def init(:ok) do
    children = [
      worker(MyApp.TwitterStream.Monitor, [], restart: :temporary)
    ]

    supervise(children, strategy: :simple_one_for_one)
  end
end