GraphQL 订阅的 HTTP 回调协议

本参考描述了一种协议，用于GraphQL服务器（或子图）通过HTTP回调将订阅数据发送到订阅的图形路由器（如GraphOS路由器）。

对于具有许多同时打开的订阅的路由器，此协议比基于WebSocket的协议扩展性更好，后者需要长时间打开连接。

GraphOS路由器为此协议提供支持，这是其对联邦订阅支持的一部分：

协议流程

以下步骤中描述的所有动作都由路由器、子图或发射器之一执行。

初始化

1. 在执行订阅操作之前，路由器生成一个唯一的ID来表示该操作。

2. 路由器通过HTTP POST以标准子图。

   JSON

   1
   {
   2
     "query": "subscription { userWasCreated { name reviews { body } } }",
   3
     "extensions": {
   4
       "subscription": {
   5
         "callbackUrl": "https://127.0.0.1:4000/callback/c4a9d1b8-dc57-44ab-9e5a-6e6189b2b945",
   6
         "subscriptionId": "c4a9d1b8-dc57-44ab-9e5a-6e6189b2b945",
   7
         "verifier": "XXX",
   8
         "heartbeatIntervalMs": 5000
   9
       }
   10
     }
   11
   }

   复制

   此有效载荷的 extensions 属性包含一个 subscription 对象，内容包括以下内容：

   • callbackUrl： }}Emitter 将向其发送 subscription 数据的 URL
   • subscriptionId：为 subscription 操作生成的唯一 ID
   • verifier：一个字符串，Emitter 将包括在所有 HTTP 回调请求中，以验证其身份
   • heartbeatIntervalMs：向 subscription 操作的回调端点发送心跳的毫秒数，如果 0 则表示禁用心跳

3. 在 Subgraph 返回 Router 的请求之前， 它向提供的 callbackUrl 发送一个带有回调协议版本头（subscription-protocol: callback/1.0）的 check 消息 以从 Router 确认：

   JSON

   1
   {
   2
     "kind": "subscription",
   3
     "action": "check",
   4
     "id": "c4a9d1b8-dc57-44ab-9e5a-6e6189b2b945",
   5
     "verifier": "XXX"
   6
   }

   复制

   发送到 callbackUrl 的所有消息都是 HTTP POST 请求。以下列出消息类型：

   这有助于确保 子图 能够成功发送回调，并且 id 和 verifier 字段 是正确的。

4. 路由器使用其 check 消息的 id 和 verifier 字段来验证消息。

   再次强调，子图 尚未响应当前 路由器' 的原始请求！

   • 如果 check 消息有效，路由器 将以以下详细信息响应：
     • 204 HTTP 状态码
     • 空响应体
     • 指定最大支持的协议版本的响应头 subscription-protocol: callback/1.0
   • 如果 check 消息是 无效的，路由器 将以 204 之外的状态码响应（推荐使用 400 级别）。
     • 如果发生这种情况，子图 将以 400 级别状态码响应 路由器 的请求，并取消订阅。

5. 如果验证成功，子图 将启动一个后台进程或通知一个单独的系统（发射器）以开始监听订阅事件。

6. 子图 最后以 200 级别状态码和空数据 GraphQL 响应（{ "data": null }）响应 路由器。这表示订阅已被初始化。

初始化完成后，协议开始执行其 主循环。

主循环

在订阅的整个过程中，协议的主循环保持活跃。在主循环期间，以下所有操作都会发生：

• 如果启用了心跳（heartbeatIntervalMs > 0），发射器 必须在每 heartbeatIntervalMs 毫秒周期内向 路由器 发送 check 消息以确认 路由器 仍在监听。（注意，heartbeatIntervalMs 的值由 路由器 发送的初始负载中的 extensions 设置。)
• 每当有新的订阅数据可用时，发射器 将向 路由器 发送 next 消息，其中包含新数据。
• 如果发生错误并且必须终止订阅，发射器 将向 路由器 发送 complete 消息，并包括 errors 字段。

消息类型

1
{
2
  "kind": "subscription",
3
  "action": "check",
4
  "id": "c4a9d1b8-dc57-44ab-9e5a-6e6189b2b945",
5
  "verifier": "XXX"
6
}

检测

只要订阅处于活动状态并启用心跳（heartbeatIntervalMs > 0），Emitter
必须每隔heartbeatIntervalMs
（毫秒）向Router发送一条check信息。这个信息是从extensions中获取的初始载荷发送给Router的值。）这使Emitter
能够确认它仍然可以访问Router的回调端点，并且订阅仍然有效。

一条check信息仅包含基本消息字段：

1
{
2
  "kind": "subscription",
3
  "action": "check",
4
  "id": "c4a9d1b8-dc57-44ab-9e5a-6e6189b2b945",
5
  "verifier": "XXX"
6
}

如果id
和verifier都匹配Router提供的值，Router应以下详细信息作出响应：

• 204 HTTP 状态码
• 空响应体
• 指定最大支持的协议版本的响应头 subscription-protocol: callback/1.0

否则，Router应响应一个错误，并subgraph应终止相关的订阅。

下一个

每当发生新的订阅事件时，Emitter将相关数据作为Router的next消息发送。

“next”消息包含一个payload字段，其中包含订阅数据，格式为标准的GraphQLJSON响应格式：

1
{
2
  "kind": "subscription",
3
  "action": "next",
4
  "id": "c4a9d1b8-dc57-44ab-9e5a-6e6189b2b945",
5
  "verifier": "XXX",
6
  "payload": {
7
    "data": {
8
      "getLivePriceUpdates": {
9
        "__typename": "Stock",
10
        "price": 5
11
      }
12
    }
13
  }
14
}

• 根级键必须是订阅操作的名称。

• 初始订阅查询将包含Router预期的所有字段。

  上述两项都是服务器库自动处理的事情，因此在手动实现协议时可能不会很明显。

完成

Emitter向Router发送“complete
”消息来终止一个活跃的订阅。Emitter可能因为以下原因终止订阅：

• 订阅已达到流末尾并且没有新的数据提供。
• Emitter发生错误导致订阅失败。

一个complete消息可以包含包含GraphQL错误数组的errors字段。如果订阅失败，该字段是必需的；如果成功完成（在这种情况下通常是空列表）：

1
{
2
    "kind": "subscription",
3
    "action": "complete",
4
    "id": "c4a9d1b8-dc57-44ab-9e5a-6e6189b2b945",
5
    "verifier": "XXX",
6
    "errors": [{ // Optional if subscription completed successfully
7
      "message": "Something went wrong"
8
    }]
9
}

在收到complete消息后，Router终止相关subscription。

错误状态

以下是可能与本协议相关的一些常见错误状态：

• Emitter无法与Router的回调端点进行通信，这可能是由于端点不可用或其提供的凭据（id和/或verifier）无效。
• Emitter从Router的回调端点收到错误HTTP状态码。
  • 如果错误码是404，Emitter应考虑关联的subscription已被Router终止。
  • 在其他错误情况下，Emitter应考虑subscription由于意外错误而终止。
• Emitter无法以每heartbeatIntervalMs毫秒（来自Router发送的初始payload中的extensions的值）向Router发送check消息，导致Router终止该subscription。