redis - 如何设置 Phoenix PubSub 订阅者回调

Question

我有一个相当简单的要求，主要围绕在 Phoenix 中构建的 2 个服务(目前):

ServiceA 负责注册用户。当用户注册时，ServiceA 广播一条消息，其中包含有关新创建用户的信息。现在正在 Controller 操作中使用以下代码完成此操作:

ServiceA.Endpoint.broadcast("activity:all", "new:user", %{email: "test@test.com"})

ServiceB 负责监听所有这些事件广播并对其进行处理(实质上是建立事件提要)。

我遇到了一个绊脚石，因为我可以看到 ServiceA 将消息广播到 Redis(使用 Phoenix.PubSub.Redis)，但不完全理解如何让 ServiceB 上的订阅者处理它...

以下代码是我设法得到的，它在消息广播时执行某些操作，然后引发异常。

部分订阅者模块

defmodule ServiceB.UserSubscriber do

  def start_link do
    sub = spawn_link &(process_feed/0)
    ServiceB.Endpoint.subscribe(:user_pubsub, "activity:all")
    {:ok, sub}
  end

  def process_feed do
    receive do
      params ->
        IO.inspect "processing goes here..."
    end
    process_feed
  end

end

异常

[error] GenServer :user_pubsub terminating
** (FunctionClauseError) no function clause matching in Phoenix.PubSub.RedisServer.handle_info/2

我猜我错过了某个地方的整个 GenServer 工作负载，但似乎无法在网上找到任何提示 where 的内容。

Answer 1

问题(正如预期的那样)是我的订阅者模块没有作为 GenServer 实现，但我试图复制相同的功能(而且很糟糕!)。如下更新我的订阅者模型已经成功:

defmodule SubscriberService.ActivitySubscriber do
  use GenServer

  def start_link(channel) do
    GenServer.start_link(__MODULE__, channel)
  end

  def init(channel) do
    pid = self
    ref = SubscriberService.Endpoint.subscribe(pid, channel)
    {:ok, {pid, channel, ref}}
  end

  def handle_info(%{event: "new:user"} = message, state) do
    IO.inspect "#######################"
    IO.inspect "New User - Received Message:"
    IO.inspect message
    IO.inspect "#######################"
    {:noreply, state}
  end

  def handle_info(message, state) do
    IO.inspect "#######################"
    IO.inspect "Catch All - Received Message:"
    IO.inspect message
    IO.inspect "#######################"
    {:noreply, state}
  end
end

如您所见，init/1 触发订阅，handle_info/2 函数接收传入的消息。

如果你想看看它是如何工作的(发布者和订阅者服务)，take a look at the repo .