Absinthe实时订阅Skill absinthe-subscriptions

---

名称: absinthe-subscriptions 用户可调用: false 描述: 在实现实时GraphQL订阅与Absinthe时使用。涵盖Phoenix通道、PubSub和订阅模式。

Absinthe - 订阅

实现实时GraphQL订阅与Absinthe和Phoenix的指南。

关键概念

基本设置

# 在您的Phoenix端点中
defmodule MyAppWeb.Endpoint do
  use Phoenix.Endpoint, otp_app: :my_app
  use Absinthe.Phoenix.Endpoint

  socket "/socket", MyAppWeb.UserSocket,
    websocket: true,
    longpoll: false
end

# 套接字配置
defmodule MyAppWeb.UserSocket do
  use Phoenix.Socket
  use Absinthe.Phoenix.Socket, schema: MyApp.Schema

  def connect(params, socket, _connect_info) do
    current_user = get_user_from_token(params["token"])
    socket = Absinthe.Phoenix.Socket.put_options(socket,
      context: %{current_user: current_user}
    )
    {:ok, socket}
  end

  def id(socket), do: "user_socket:#{socket.assigns.user_id}"
end

定义订阅

defmodule MyApp.Schema.Subscriptions do
  use Absinthe.Schema.Notation

  object :post_subscriptions do
    field :post_created, :post do
      config fn _args, _resolution ->
        {:ok, topic: "posts"}
      end

      trigger :create_post, topic: fn _post ->
        "posts"
      end
    end

    field :post_updated, :post do
      arg :id, non_null(:id)

      config fn %{id: id}, _resolution ->
        {:ok, topic: "post:#{id}"}
      end

      trigger :update_post, topic: fn post ->
        "post:#{post.id}"
      end
    end
  end
end

从突变发布

defmodule MyApp.Resolvers.Post do
  def create_post(_parent, %{input: input}, _resolution) do
    case MyApp.Posts.create_post(input) do
      {:ok, post} ->
        # 发布到订阅
        Absinthe.Subscription.publish(
          MyAppWeb.Endpoint,
          post,
          post_created: "posts"
        )
        {:ok, post}
      {:error, changeset} ->
        {:error, changeset}
    end
  end
end

用户特定订阅

field :user_notification, :notification do
  config fn _args, %{context: %{current_user: user}} ->
    {:ok, topic: "user:#{user.id}:notifications"}
  end
end

# 发布
Absinthe.Subscription.publish(
  MyAppWeb.Endpoint,
  notification,
  user_notification: "user:#{user_id}:notifications"
)

最佳实践

1. 范围订阅 - 使用主题限制数据暴露
2. 认证连接 - 在套接字连接中验证用户
3. 使用触发器 - 在突变时自动发布
4. 处理断开连接 - 断开时清理资源
5. 限制订阅速率 - 防止滥用

PubSub配置

# config/config.exs
config :my_app, MyAppWeb.Endpoint,
  pubsub_server: MyApp.PubSub

# application.ex
children = [
  {Phoenix.PubSub, name: MyApp.PubSub},
  MyAppWeb.Endpoint,
  {Absinthe.Subscription, MyAppWeb.Endpoint}
]

订阅中的授权

field :private_messages, :message do
  config fn _args, %{context: context} ->
    case context do
      %{current_user: %{id: user_id}} ->
        {:ok, topic: "user:#{user_id}:messages"}
      _ ->
        {:error, "未授权"}
    end
  end
end

反模式

• 不要将敏感数据发布到广泛主题
• 避免未经认证的订阅
• 不要跳过连接级授权
• 避免过度细粒度的主题（影响性能）