配置 GraphQL 订阅支持

GraphOS 路由器 支持以下 GraphQL 订阅操作：

1
subscription OnStockPricesChanged {
2
  stockPricesChanged {
3
    symbol
4
    price
5
  }
6
}

启用 订阅 支持，您可以为任何支持通用 WebSocket 协议进行订阅通信的 子图 模式添加 Subscription 字段：

1
type Subscription {
2
  stockPricesChanged: [Stock!]!
3
}

订阅的用途是什么？

GraphQL 订阅 允许客户端在可用的任何时候接收到持续、实时的更新。与查询和 变更 不同，订阅是持续存在的。这意味着客户端可以从单个订阅接收多个更新：

订阅非常适合那些依赖频繁变化、时间敏感的数据的应用，例如股票价格、物联网传感器读取、实时聊天或体育比分。

如何实现订阅

1. 客户端在GraphQL 订阅 操作与您的路由器通过HTTP进行：

   GraphQL

   示例订阅

   1
   subscription OnStockPricesChanged {
   2
     stockPricesChanged {
   3
       symbol
   4
       price
   5
     }
   6
   }

   拷贝
   • 客户端不使用WebSocket协议。相反，它通过多部分HTTP响应。
   • 使用HTTP进行订阅，客户端可以执行所有GraphQL 操作类型，而不是使用两种不同的协议。
   • Apollo Client、Apollo Kotlin以及Apollo iOS都支持通过HTTP配置最少数量的GraphQL订阅。有关详细信息，请参阅每个库的文档。Apollo Client还提供针对Relay和urql库的网络适配器。

2. 当您的路由器接收到订阅时，它将执行相同的订阅操作，针对请求的

   • 这种通信通常使用WebSocket子协议，以便与大多数子图库兼容。
   • 使用自托管路由器，您还可以配置一个

3. 子图周期性地向您的路由器发送新数据。每次发送时，路由器都会在额外的HTTP响应部分中将该数据返回给客户端。

   • 订阅可以包含在其它子图中定义的联邦实体字段。如果确实包含，路由器将首先通过查询相应的子图来检索这些字段，如上图中的Portfolios子图。这些查询通常使用HTTP。

特殊考虑

无论何时您的路由器在运行时更新其supergraph模式，它都会终止所有活动订阅。客户端可以通过错误代码检测这种特殊情况的终止并执行新的订阅。请参阅在模式更新时终止。

先决条件

在向您的子图添加Subscription字段之前，请按照以下顺序完成所有操作以防止模式组合错误：

1. 将您的路由器实例更新到版本1.22.0或更高版本。

   • 路由器的早期版本不支持订阅操作。

2. 确保您的路由器连接到一个GraphOS Enterprise组织。

   • 订阅支持是自托管路由器的Enterprise功能。

3. 如果您使用GraphOS（而不是Rover CLI）组合路由器的supergraph模式，请更新您的构建管道以使用Apollo Federation 2.4或更高版本。

   • Apollo Federation 的早期版本不支持 订阅操作。

4. 修改您的 子图模式，使用 Apollo Federation 2.4 或更高版本：

   GraphQL

   stocks.graphql

   1
   extend schema
   2
   @link(url: "https://specs.apollo.dev/federation/v2.4", 
   3
         import: ["@key", "@shareable"])
   4
   
   5
   type Subscription {
   6
     stockPricesChanged: [Stock!]!
   7
   }

   拷贝
   • 您可以跳过修改 未定义任何 订阅 字段的 子图模式。

5. 如果您使用 Apollo Server 来实现子图，请将您的 Apollo Server 实例更新到版本 4 或更高版本：更新 Apollo Server 实例。

   • 根据 Apollo Server 4 指南启用订阅。
   • 更新 @apollo/subgraph。

在完成所有这些前提条件后，您可以安全地配置 路由器以支持订阅。

路由器设置

在完成所有 前提条件后，在您路由器 YAML 配置文件中，您配置路由器如何在与执行 GraphQL 订阅时与每个子图通信。

路由器支持两种流行的 WebSocket 协议用于订阅，并还提供基于 HTTP-callback 协议的支持。您的路由器必须使用每个子图期望的协议。

WebSocket 设置

以下是一个示例路由器配置片段，它通过 WebSocket 设置子图订阅：

1
subscription:
2
  enabled: true
3
  mode:
4
    passthrough:
5
      all: # The router uses these subscription settings UNLESS overridden per-subgraph
6
        path: /subscriptions # The absolute URL path to use for subgraph subscription endpoints (Default: /ws)
7
      subgraphs: # Overrides subscription settings for individual subgraphs
8
        reviews: # Overrides settings for the 'reviews' subgraph
9
          path: /ws # Absolute path that overrides the preceding '/subscriptions' path for 'all'
10
          protocol: graphql_ws # The WebSocket-based subprotocol to use for subscription communication (Default: graphql_ws)
11
          heartbeat_interval: 10s # Optional and 'disable' by default, also supports 'enable' (set 5s interval) and custom values for intervals, e.g. '100ms', '10s', '1m'.

此示例在 透传模式中启用订阅，该模式使用持久的 WebSocket 连接。

路由器支持以下WebSocket子协议，通过protocol选项指定：

• graphql_ws
  • 由graphql-ws库使用，
  • 该子协议是默认值，适用于实现基于WebSocket的订阅的GraphQL服务器库。
• graphql_transport_ws
  • 由subscriptions-transport-ws库使用的旧子协议，该库已不维护，但提供以实现向下兼容。

默认情况下，路由器为所有子图使用graphql_ws协议选项。您可以更改此全局默认值，并通过设置如上所示的protocol键来覆盖它。

您的路由器为每个客户端订阅创建一个单独的WebSocket连接，除非它能够执行订阅去重。

HTTP回调设置

该路由器提供通过HTTP回调接收子图订阅事件的支持，而不是通过持久的WebSocket连接。这种回调模式相对于基于WebSocket的订阅具有以下优势：

• 路由器不需要为每个不同的订阅维护持久连接。
• 您可以直接从pubsub系统向路由器发布事件，而不是通过子图路由这些事件。

回调模式要求您的子图库支持路由器的HTTP回调协议。

以下是设置回调模式的子图订阅的示例配置：

1
subscription:
2
  enabled: true
3
  mode:
4
    callback:
5
      public_url: https://example.com:4000/callback # The router's public URL, which your subgraphs access, must include the path configured on the router
6
      listen: 127.0.0.1:4000 # The IP address and port the router will listen on for subscription callbacks, for security reasons it might be better to expose on another port that is only available from your internal network
7
      path: /callback # The path of the router's callback endpoint
8
      heartbeat_interval: 5s # Optional (default: 5secs)
9
      subgraphs: # The list of subgraphs that use the HTTP callback protocol
10
        - accounts

您可以通过设置heartbeat_interval: disabled来禁用心跳。这很有用，例如，如果您正在在基于lambda函数的基础设施中以回调模式运行，您既不想发送心跳，又不希望为了只向扩展数据（包括包含回调URL和验证器的回调URL数据）发送心跳

使用组合模式

如果一些子图需要透传模式而另一些需要回调模式处理订阅，您可以在配置中对不同的子图应用不同的模式：

1
subscription:
2
  enabled: true
3
  mode:
4
    passthrough:
5
      subgraphs:
6
        reviews: 
7
          path: /ws
8
          protocol: graphql_ws
9
    callback:
10
      public_url: http://public_url_of_my_router_instance:4000/callback # This must include the path configured on the router
11
      listen: 127.0.0.1:4000
12
      path: /callback
13
      subgraphs:
14
        - accounts 

在这个例子中，reviews 子图使用WebSocket，而accounts 子图使用基于HTTP的回调。

示例执行

假设我们的超级图包含以下子图和部分模式：

type Product @key(fields: "id") {
  id: ID!
  name: String!
  price: Int!
}

type Subscription {
  productPriceChanged: Product!
}

type Product @key(fields: "id") {
  id: ID!
  reviews: [Review!]!
}

type Review {
  score: Int!
}

现在，假设某个客户端对我们的路由器执行以下订阅（通过HTTP）：

1
subscription OnProductPriceChanged {
2
  productPriceChanged {
3
    # Defined in Products subgraph
4
    name
5
    price
6
    reviews {
7
      # Defined in Reviews subgraph
8
      score
9
    }
10
  }
11
}

当我们的路由器收到此操作时，它将针对产品子图执行相应的订阅操作（通过新的WebSocket连接）：

1
subscription {
2
  productPriceChanged {
3
    id # Added for entity fetching
4
    name
5
    price
6
    # Reviews fields removed
7
  }
8
}

在订阅开始后，产品子图可能会向我们路由器发送更新数据。在此之后，每当发生这种情况时，路由器不会立即将数据返回给客户端，因为它缺少请求的评价子图中的字段。

相反，我们的路由器执行一个标准的GraphQL查询，针对评论子图以获取缺失的实体字段：

1
query {
2
  entities(representations: [...]) {
3
    ... on Product {
4
      reviews {
5
        score
6
      }
7
    }
8
  }
9
}

收到来自评论子图的这个查询结果后，我们的路由器将其与产品数据合并，并将组合数据返回给订阅客户。

使用curl尝试订阅

为了快速尝试GraphOS路由器的HTTP订阅，而不需要安装Apollo客户端库，您可以执行以下格式的curl命令针对您的路由器： curl命令：

1
curl 'https://127.0.0.1:4000/' -v \
2
  -H 'accept: multipart/mixed;subscriptionSpec=1.0, application/json' \
3
  -H 'content-type: application/json' \
4
  --data-raw '{"query":"subscription OnProductPriceChanged { productPriceChanged { name price reviews { score } } }","operationName":"OnProductPriceChanged"}'

此命令创建一个HTTP多部分请求，并保持开放的连接，以接收响应块中的新订阅数据：

1
--graphql
2
content-type: application/json
3

4
{}
5
--graphql
6
content-type: application/json
7

8
{"payload":{"data":{"productPriceChanged":{"name":"Croissant","price":400,"reviews":[{"score":5}]}}}}
9
--graphql
10
content-type: application/json
11

12
{"payload":{"data":{"productPriceChanged":{"name":"Croissant","price":375,"reviews":[{"score":5}]}}}}
13
--graphql
14
content-type: application/json
15

16
{"payload":{"data":{"productPriceChanged":{"name":"Croissant","price":425,"reviews":[{"score":5}]}}}}
17
--graphql--

此示例订阅仅发出三个事件，然后直接关闭连接。

有关此多部分HTTP订阅协议的更多信息，请参阅此文章。

订阅去重

默认情况下，路由器去重相同的订阅。这可以显著降低路由器和您的子图的压力，因为如果现有连接已经处理了确切的同一个订阅，路由器不需要打开新的连接。

例如，如果有成千上万的客户端都订阅了同一场体育比赛的实时得分更新，您的路由器只需要维护到sportsgames子图的一个连接，以接收所有这些订阅的事件。

路由器认为以下条件全部满足时，订阅操作是相同的：

• 发送到子图的操作具有相同的 GraphQL选择集（即请求的字段）。
• 这些操作为路由器发送到子图的所有头部提供相同的值。

禁用去重

您可以通过在新路由器的 YAML 配置文件中的subscription键下添加以下内容来禁用订阅去重：

1
subscription:
2
  enabled: true
3
  enable_deduplication: false # default: true

请注意，这是一个全局设置（不是每个子图或每个操作的设置）。

为什么禁用去重？

禁用去重适用于您需要为每个客户端发起的订阅创建单独连接的情况。例如：

• 您的子图需要在每个新客户端订阅其数据时触发一个重要事件。
  • 此事件在路由器重用现有连接时不会触发。
• 您的订阅需要从特定序列的第一个值开始接收，而不是从最新的值开始。
  • 如果订阅重用现有连接，它将从接收该连接的下一个值开始。
  • 作为一个基本的例子，让我们假设一个订阅始终顺序返回整数0到1000的事件。如果新的订阅重用现有的子图连接，它将从原始连接的下一个值开始接收，这几乎肯定不是0。

高级配置

在模式更新时终止

每次您的路由器的 supergraph 模式更新时，路由器将终止所有活动订阅。

您的路由器的 supergraph 模式在以下情况下更新：

• 您的路由器定期轮询 GraphOS 以获取其 supergraph 模式，并且更新的模式变得可用。
• 如果您将路由器的 supergraph 模式从本地文件获取，并且已设置--hot-reload选项，则路由器将监视该文件的更新。

当路由器以这种方式终止订阅时，它将向所有活跃的订阅客户端发送以下最终响应有效载荷：

1
{
2
  "errors": [
3
    {
4
      "message": "subscription has been closed due to a schema reload",
5
      "extensions": {
6
        "code": "SUBSCRIPTION_SCHEMA_RELOAD"
7
      }
8
    }
9
  ]
10
}

收到此SUBSCRIPTION_SCHEMA_RELOAD错误代码的客户端可以通过执行新的订阅操作来重新连接。

WebSocket身份验证支持

默认情况下，如果您已将路由器配置为传播HTTP Authorization头到子图，则路由器在启动到该子图的WebSocket连接时将自动设置相应的connectionParams。

例如，当您的路由器向子图发送connection_init消息时，它将通过以下有效负载包含Authorization头值：

1
{
2
  "connectionParams": {
3
    "token": "CONTENTS_OF_AUTHORIZATION_HEADER"
4
  }
5
}

要为connection_init消息指定自定义有效负载，您可以使用Rhai脚本来直接使用上下文：

1
fn subgraph_service(service, subgraph) {
2
  let params = Router.APOLLO_SUBSCRIPTION_WS_CUSTOM_CONNECTION_PARAMS;
3
  let f = |request| {
4
    request.context[params] = #{
5
            my_token: "here is my token"
6
        };
7
  };
8

9
  service.map_request(f);
10
}

扩展事件队列容量

如果您的路由器收到针对特定订阅的大量事件，它可能会积累这些事件以发送给客户端的后备。为了处理此备用，路由器维护一个未发送事件的内存队列。

路由器为每个活跃的订阅子图连接维护一个独立的事件队列。

您可以在路由器的YAML配置文件中配置每个事件队列的大小，如下所示：

1
subscription:
2
  enabled: true
3
  queue_capacity: 100000 # Default: 128

“queue_capacity”的值对应于每个队列的订阅事件的最大数量，而不是这些事件的总大小。

当您的路由器在队列满时收到新的订阅事件，它会丢弃队列中最旧的未发送事件，并将新接收的事件入队。被丢弃的事件不会发送给订阅客户端。

如果客户端绝对需要接收每个订阅事件，则根据需要增加事件队列的大小。

限制客户端连接数

客户端订阅 是长期存在的HTTP连接，这意味着它们可能会无限期地保持打开状态。您可以在路由器的YAML配置文件中限制同时客户端订阅连接数，如下所示：

1
subscription:
2
  enabled: true
3
  max_opened_subscriptions: 150 # Only 150 simultaneous connections allowed

当客户端尝试在max_open_subscriptions已满时执行路由器上的订阅，路由器会报错拒绝客户端请求。