2

我正在使用ExTwitter Stream实时跟踪推文并通过频道端点广播它们。我想为每个事件创建一个进程并为其分配一个 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

如何在事件监视器之外接收新推文或最终停止推特流？

导师：

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
4

0 回答 0