Возможно, вы думаете, что о подписках говорить особо нечего. Они определены в спецификации GraphQL, и должно быть очень ясно, как они работают и для чего они нужны.

Но на самом деле спецификация мало что говорит о транспортном слое. Фактически, она вообще не указывает транспортный слой. С одной стороны, это преимущество, потому что вы можете использовать GraphQL в самых разных средах. С другой стороны, у нас сейчас есть по крайней мере пять разных реализаций подписок GraphQL.

Это означает, что вы не можете просто использовать любой клиент GraphQL, подключиться к серверу GraphQL и ожидать, что все будет работать. Вам нужно знать, какой протокол поддерживает сервер и какой клиент вам нужно использовать. Это идеальная ситуация? Вероятно, нет, но мы собираемся это изменить!

Мы - создатели WunderGraph (открытый исходный код), первого облачного серверного GraphQL API Gateway. Одной из проблем, с которой мы столкнулись, была поддержка всех различных протоколов подписки GraphQL. Поскольку спецификация GraphQL строго агностична к протоколу, за годы было разработано несколько различных протоколов.

Если клиент хочет использовать подписку GraphQL, ему нужно знать, какой протокол использовать, и реализовать клиентскую сторону этого протокола.

С нашим Open Source API Gateway, мы делаем шаг вперед и объединяем все под одной крышей. Если вы смотрите на использование подписок GraphQL в вашем проекте, этот пост - отличный способ быстро ознакомиться с различными протоколами и их особенностями.

Введение - Что такое подписки GraphQL?

У GraphQL есть три типа операций: Запросы, Мутации и Подписки. Запросы и Мутации используются для получения и изменения данных. Подписки используются для подписки на изменения данных.

Вместо опроса сервера об обновлении, подписки позволяют клиенту подписываться на изменения данных, например, подписываясь на чат-комнату. Когда в чат-комнату отправляется новое сообщение, сервер отправляет сообщение клиенту.

С Запросами и Мутациями, контроль потока находится в руках клиента. Клиент отправляет запрос на сервер и ждет ответа. С Подписками, контроль потока находится в руках сервера.

Вот пример подписки GraphQL:

subscription($roomId: ID!) {
  messages(roomId: $roomId) {
    id
    text
  }
}

Теперь сервер будет отправлять клиенту непрерывный поток сообщений. Вот пример с 2 сообщениями:

{
  "data": {
    "messages": {
      "id": 1,
      "text": "Hello Subscriptions!"
    }
  }
}
{
  "data": {
    "messages": {
      "id": 2,
      "text": "Hello WunderGraph!"
    }
  }
}

Теперь, когда мы понимаем, что такое подписки GraphQL, давайте рассмотрим различные доступные протоколы.

Подписки GraphQL через WebSockets

Наиболее широко используемый транспортный слой для подписок GraphQL - это WebSockets. WebSockets - это двунаправленный протокол связи. Они позволяют клиенту и серверу отправлять друг другу сообщения в любое время.

Существует две реализации подписок GraphQL через WebSockets:

Первая - это subscription-transport-ws от Apollo, вторая - graphql-ws от Дениса Бадурина.

Оба протокола довольно похожи, хотя есть некоторые незначительные различия. Важно отметить, что протокол Apollo устарел в пользу graphql-ws, но он все еще широко используется.

Подписки GraphQL через WebSockets: subscription-transport-ws против graphql-ws

Оба транспорта используют JSON в качестве формата сообщения. Для уникальной идентификации типа сообщения используется поле type. Отдельные подписки идентифицируются по полю id.
Клиенты инициируют соединение, отправляя сообщение connection_init, за которым следует сообщение connection_ack от сервера.

{"type": "connection_init"}
{"type": "connection_ack"}

Мне это кажется странным. Создается впечатление, что мы создаем несколько слоев TCP. Чтобы создать соединение WebSocket, нам сначала нужно создать соединение TCP. Соединение TCP инициируется отправкой пакета SYN, за которым следует пакет ACK от сервера. Таким образом, между клиентом и сервером уже происходит рукопожатие.

Затем мы инициируем соединение WebSocket, отправляя запрос HTTP Upgrade, который сервер принимает, отправляя ответ HTTP Upgrade. Это второе рукопожатие между клиентом и сервером.

Зачем нам третье рукопожатие? Мы недостаточно доверяем протоколу WebSocket?

В любом случае, после этих трех рукопожатий, мы наконец готовы отправлять сообщения. Но прежде чем говорить о запуске и остановке подписок, мы должны убедиться, что наше соединение WebSocket все еще живо. Что касается сердцебиения, то есть некоторые различия между двумя протоколами.

Протокол Apollo использует сообщение {"type": "ka"} для отправки сердцебиения от сервера к клиенту. В протоколе отсутствует определение того, как должен реагировать клиент. Если сервер отправляет клиенту сообщение о поддержании активности, но клиент никогда не отвечает, в чем смысл сообщения о поддержании активности? Но есть еще одна проблема. Протокол указывает, что сервер должен начать отправлять сообщения о поддержании активности только после того, как соединение будет подтверждено. На практике мы обнаружили, что Hasura может отправлять сообщения о поддержании активности до подтверждения соединения. Поэтому, если ваша реализация зависит от строгого порядка сообщений, вам следует быть в курсе этого.

Протокол graphql-ws улучшил это. Вместо одного сообщения о поддержании активности он определяет, что сервер должен периодически отправлять сообщение {"type":"ping"}, на которое клиент должен ответить сообщением {"type":"pong"}. Это обеспечивает для обоих клиента и сервера, что другая сторона все еще жива.

Теперь давайте поговорим о запуске подписки. С протоколом Apollo нам пришлось отправить следующее сообщение:

{"type":"start","id":"1","payload":{"query":"subscription {online_users{id}}"}}

Тип - это start, и мы должны указать id для уникальной идентификации подписки; subscription отправляется как поле запроса на объекте payload. Мне кажется, это сбивает с толку, и это происходит из-за того, что многие люди в сообществе GraphQL называют операции запросами. Это становится еще более запутанным, потому что, хотя "Операция GraphQL" называется запросом (query), вы должны указать поле operationName, если у вас в документе есть несколько именованных операций.

К сожалению, протокол graphql-ws не улучшил это. Я предполагаю, что это потому, что они хотят оставаться в соответствии со спецификацией GraphQL over HTTP, спецификацией, которая пытается унифицировать способ использования GraphQL через HTTP.

В любом случае, вот как мы бы начали подписку с протоколом graphql-ws:

{"type":"subscribe","id":"1","payload":{"query":"subscription {online_users{id}}"}}

Тип start был заменен на subscribe, остальное осталось без изменений.

После того, как мы инициировали соединение и начали подписку, мы теперь должны разобраться, как работает прием сообщений.

Для сообщений подписки протокол Apollo использует тип data, вместе с id подписки, с фактическими данными, которые отправляются в поле payload.

{"type":"data","id":"1","payload":{"data":{"online_users":[{"id":1},{"id":2}]}}}

Протокол graphql-ws использует тип next для сообщений подписки, остальная часть сообщения остается без изменений.

{"type":"next","id":"1","payload":{"data":{"online_users":[{"id":1},{"id":2}]}}}

Теперь, когда мы начали подписку, мы можем захотеть остановить ее в какой-то момент.

Протокол Apollo использует тип stop для этого. Если клиент хочет остановить подписку, он отправляет сообщение stop с id подписки.

{"type":"stop","id":"1"}

Протокол graphql-ws упростил это. Как клиент, так и сервер могут отправить сообщение complete с id подписки, чтобы остановить ее или уведомить другую сторону о том, что подписка была остановлена.

{"type":"complete","id":"1"}

С другой стороны, в протоколе Apollo сообщение complete использовалось только сервером для уведомления клиента о том, что подписка была остановлена, и больше нет данных для отправки.

Это был быстрый обзор различий между двумя протоколами. Но как клиент на самом деле знает, какой протокол использовать, или как сервер может узнать, какой протокол использует клиент?

Здесь вступает в игру согласование содержимого. Когда клиент инициирует соединение WebSocket, он может отправить список поддерживаемых протоколов в заголовке Sec-WebSocket-Protocol. Затем сервер может выбрать один из протоколов и отправить его обратно в заголовке Sec-WebSocket-Protocol ответа HTTP Upgrade.

Вот как может выглядеть такой запрос на обновление:

GET /graphql HTTP/1.1
Host: localhost:8080
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Sec-WebSocket-Version: 13
Sec-WebSocket-Protocol: graphql-ws, graphql-transport-ws

А вот как может отреагировать сервер:

HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
Sec-WebSocket-Protocol: graphql-ws

Это теория. Но работает ли это на практике? Простой ответ - нет, но я думаю, что стоит более подробно разобраться в этом вопросе.

Реализации клиента и сервера GraphQL обычно не поддерживают согласование содержимого. Причина в том, что в течение долгого времени существовал только один протокол, поэтому не было необходимости в переговорах. Теперь, когда есть несколько протоколов, уже слишком поздно добавлять поддержку согласования содержимого в существующие реализации.

Это означает, что даже если клиент отправляет список поддерживаемых протоколов, сервер может просто проигнорировать его и использовать протокол, который он поддерживает. Или, что еще хуже, сервер может выбрать первый протокол в списке, даже если он его не поддерживает, а затем действовать так, как будто был выбран второй протокол.

Так что вам нужно как-то "идентифицировать" клиента и сервер, чтобы понять, какой протокол он поддерживает. Другой вариант - "просто попробовать" и посмотреть, какой протокол работает. Это не идеально, но с этим приходится работать.

Было бы хорошо, если бы у нас было что-то вроде запроса OPTIONS для серверов GraphQL, чтобы клиент и сервер могли узнать друг о друге, чтобы выбрать правильный протокол. Но мы вернемся к этому позже.

А пока давайте подведем итог полного потока двух протоколов. Начнем с протокола Apollo.

C: {"type": "connection_init"}
S: {"type": "connection_ack"}
S: {"type": "ping"}
C: {"type": "pong"}
C: {"type": "subscribe","id":"1","payload":{"query":"subscription {online_users{id}}"}}
S: {"type": "next","id":"1","payload":{"data":{"online_users":[{"id":1},{"id":2}]}}}
C: {"type": "complete","id":"1"}

Для сравнения, вот поток subscriptions-transport-ws:

C: {"type": "connection_init"}
S: {"type": "connection_ack"}
S: {"type": "ka"}
C: {"type": "start","id":"1","payload":{"query":"subscription {online_users{id}}"}}
S: {"type": "data","id":"1","payload":{"data":{"online_users":[{"id":1},{"id":2}]}}}
C: {"type": "stop","id":"1"}

Мультиплексирование подписок GraphQL через WebSocket

Что замечательно в этих двух протоколах, так это то, что они оба поддерживают мультиплексирование нескольких подписок через одно соединение WebSocket. Это означает, что мы можем отправлять несколько подписок через одно и то же соединение и получать несколько сообщений подписки на одном и том же соединении. В то же время, это также огромный недостаток, потому что мультиплексирование реализовано на уровне приложения.

Когда вы реализуете сервер или клиент GraphQL, который использует WebSockets, вам приходится самостоятельно реализовывать мультиплексирование. Не было бы гораздо лучше, если бы этим занимался транспортный уровень? Оказывается, есть протокол, который делает именно это.

GraphQL через Server-Sent Events (SSE)

Протокол Server-Sent Events - это протокол транспортного уровня, который позволяет клиенту получать события от сервера. Это очень простой протокол, который построен поверх HTTP. Вместе с HTTP/2 и HTTP/3 он является одним из наиболее эффективных протоколов для отправки событий от сервера к клиенту. Что самое главное, он решает проблему мультиплексирования нескольких подписок через одно соединение на транспортном уровне. Это означает, что уровню приложения больше не нужно беспокоиться о мультиплексировании.

Давайте посмотрим, как работает протокол, посмотрев на реализацию от GraphQL Yoga:

curl -N -H "accept:text/event-stream" "http://localhost:4000/graphql?query=subscription%20%7B%0A%20%20countdown%28from%3A%205%29%0A%7D"

data: {"data":{"countdown":5}}

data: {"data":{"countdown":4}}

data: {"data":{"countdown":3}}

data: {"data":{"countdown":2}}

data: {"data":{"countdown":1}}

data: {"data":{"countdown":0}}

Это не случайно, что мы здесь используем curl. Протокол Server-Sent Events - это протокол транспортного уровня, который построен поверх HTTP. Он настолько прост, что его можно использовать с любым HTTP-клиентом, поддерживающим потоковую передачу, например, curl. Подписка GraphQL отправляется в виде URL-кодированного параметра запроса.

Подписка начинается, когда клиент подключается к серверу, и заканчивается, когда клиент или сервер закрывает соединение. С HTTP/2 и HTTP/3 одно и то же TCP-соединение может быть использовано для нескольких подписок. Это мультиплексирование на уровне транспорта.

Если клиент не поддерживает HTTP/2, он все равно может использовать кодировку с разбиением на части по HTTP/1.1 в качестве резервного варианта.

На самом деле, этот протокол настолько прост, что его даже не нужно объяснять.

Проксирование подписок GraphQL через "шлюз" SSE

Как мы только что показали, подход Server-Sent Events - это самый простой подход. Именно поэтому мы выбрали его для WunderGraph в качестве основного способа предоставления подписок и живых запросов.

Но как объединить несколько серверов GraphQL с разными протоколами подписки под одним API? Об этом будет последняя часть этого поста...

Мультиплексирование нескольких подписок GraphQL через одно соединение WebSocket

Мы ранее обсуждали, как протоколы WebSocket поддерживают мультиплексирование нескольких подписок через одно соединение WebSocket. Это имеет смысл для клиента, но становится немного сложнее при использовании в прокси/API Gateway.

Мы не можем просто использовать одно и то же соединение WebSocket для всех подписок, потому что нам нужно обрабатывать аутентификацию и авторизацию для каждой подписки.

Поэтому, вместо использования одного соединения WebSocket для всех подписок, мы должны "собрать" все подписки вместе, которые должны быть выполнены в одном "безопасном контексте". Мы делаем это путем хеширования всей информации, связанной с безопасностью, такой как заголовки, происхождение и т.д., чтобы создать уникальный идентификатор для каждого безопасного контекста.

Если соединение WebSocket для этого хеша уже существует, мы его используем. В противном случае мы создаем новое соединение WebSocket для этого безопасного контекста.

Аутентификация для подписок GraphQL через WebSockets

Некоторые API GraphQL, например, от Reddit, ожидают, что клиент отправит заголовок авторизации с соединением WebSocket. Это немного проблематично, потому что браузеры не могут отправлять пользовательские заголовки с запросами на обновление WebSocket, API браузера просто не поддерживает это.

Так как же Reddit обрабатывает это? Перейдите, например, на reddit.com/r/graphql и откройте инструменты разработчика. Если вы отфильтруете соединения по websocket ("ws"), вы должны увидеть соединение WebSocket с wss://gql-realtime.reddit.com/query.

Если вы посмотрите на первое сообщение, вы увидите, что это connection_init с некоторым специальным содержимым:

{"type":"connection_init","payload":{"Authorization":"Bearer XXX-Redacted-XXX"}}

Клиент отправляет “заголовок авторизации” в качестве части полезной нагрузки сообщения connection_init. Мы задавались вопросом, как мы можем реализовать это, не зная, какое сообщение вы хотели бы отправить в connection_init сообщении. Reddit отправляет Bearer Token в поле Authorization, но вы можете захотеть отправить какую-то другую информацию.

Поэтому мы решили разрешить нашим пользователям определять пользовательский хук, который может изменять полезную нагрузку сообщения connection_init так, как они хотят.

Вот пример:

// wundergraph.server.ts
export default configureWunderGraphServer<HooksConfig, InternalClient>(() => ({
  hooks: {
    global: {
      wsTransport: {
        onConnectionInit: {
          // counter is the id of the introspected api (data source id), defined in the wundergraph.config.ts
          enableForDataSources: ['counter'],
          hook: async (hook) => {
            let token = hook.clientRequest.headers.get('Authorization') || ''
            // we can have a different logic for each data source
            if (hook.dataSourceId === 'counter') {
              token = 'secret'
            }
            return {
              // this payload will be passed to the ws `connection_init` message payload
              payload: {
                Authorization: token,
              },
            }
          },
        },
      },
    },
  },
  graphqlServers: [],
}))

Этот хук берет заголовок Authorization из клиентского запроса (SSE) и вставляет его в полезную нагрузку сообщения connection_init.

Это не только упрощает аутентификацию для подписок WebSocket, но и делает реализацию намного безопаснее.

Реализация Reddit предоставляет клиенту Bearer Token. Это означает, что Javascript-клиент в браузере имеет доступ к Bearer Token. Этот токен может быть потерян или может быть доступен для вредоносного Javascript-кода, который был внедрен на страницу.

С реализацией SSE дело обстоит иначе. Мы не раскрываем клиенту никаких токенов. Вместо этого идентификационные данные пользователя хранятся в зашифрованном куки-файле, доступном только по http.

Манипулирование/фильтрация сообщений подписки GraphQL

Еще одной проблемой, с которой вы можете столкнуться, является желание манипулировать/фильтровать сообщения, которые отправляются клиенту. Вы можете захотеть интегрировать API GraphQL от третьей стороны, и перед отправкой сообщений клиенту, вы захотите отфильтровать некоторые поля, которые содержат конфиденциальную информацию.

Мы также реализовали хук для этого:

// wundergraph.server.ts
export default configureWunderGraphServer<HooksConfig, InternalClient>(() => ({
  hooks: {
    global: {},
    queries: {},
    mutations: {},
    subscriptions: {
      Ws: {
        mutatingPreResolve: async (hook) => {
          // here we modify the input before request is sent to the data source
          hook.input.from = 7
          return hook.input
        },
        postResolve: async (hook) => {
          // here we log the response we got from the ws server (not the modified one)
          hook.log.info(`postResolve hook: ${hook.response.data!.ws_countdown}`)
        },
        mutatingPostResolve: async (hook) => {
          // here we modify the response before it gets sent to the client
          let count = hook.response.data!.ws_countdown!
          count++
          hook.response.data!.ws_countdown = count
          return hook.response
        },
        preResolve: async (hook) => {
          // here we log the request input
          hook.log.info(
            `preResolve hook input, counter starts from: ${hook.input.from}`
          )
        },
      },
    },
  },
}))

В вашем инструментарии есть четыре хука, которые позволяют вам манипулировать сообщением подписки до его отправки в источник и до отправки каждого ответа клиенту.

Самый интересный хук, возможно, это mutatingPostResolve, так как он позволяет вам фильтровать и манипулировать ответом, о котором мы говорили ранее.

Проксирование подписок GraphQL к федеративным API GraphQL (Apollo Federation / Supergraph / Subgraph)

Проксирование подписок GraphQL к федеративным API GraphQL добавляет целый новый уровень сложности к проблеме. Вам нужно начать подписку на корневое поле на одном из подграфов, а затем "объединить" ответ из одного или нескольких подграфов в одно сообщение.

Если вам интересно посмотреть пример того, как это работает, ознакомьтесь с примером Apollo Federation в нашем монорепозитории.

Я сделаю более подробное описание этой темы в будущем, но пока позвольте мне дать вам краткий обзор того, как это работает.

Мы разбиваем федеративную подписку GraphQL на несколько операций, одну подписку для корневого поля и один или несколько запросов для остальной части дерева ответа.

Затем мы выполняем подписку, как любую другую подписку, и переключаемся в "обычный" режим выполнения, как только приходит новое сообщение подписки от корневого поля.

Это также позволяет нам использовать "волшебное" поле _join для объединения подписки с REST API или любым другим источником данных.

Как только вы разобрались с частью управления несколькими соединениями WebSocket, остальное - это просто вопрос объединения ответов от различных источников данных, будь то федеративные или нефедеративные API GraphQL, REST API или даже gRPC.

Примеры

Это было довольно много для понимания, поэтому давайте посмотрим на некоторые примеры, чтобы сделать это немного более конкретным.

WunderGraph в качестве API-шлюза перед Hasura

Этот пример показывает, как использовать WunderGraph перед Hasura.

WunderGraph с graphql-ws-subscriptions

Следующий пример объединяет graphql-ws-subscriptions с WunderGraph.

WunderGraph с подписками Apollo GraphQL

Если вы все еще используете устаревшие подписки Apollo GraphQL, мы также покрываем вас.

WunderGraph и подписки GraphQL SSE

Этот пример использует реализацию подписок GraphQL SSE.

WunderGraph с подписками GraphQL Yoga

Одна из самых популярных библиотек GraphQL, GraphQL Yoga, определенно должна быть в списке.

Пример хуков на подписки WunderGraph

Наконец, мы хотели бы завершить это примером хуков на подписки WunderGraph, демонстрирующим доступные хуки.

Заключение

Как вы узнали, понимание и реализация всех различных протоколов подписки GraphQL включает в себя некоторую сложность.

Я думаю, что в сообществе GraphQL действительно не хватает стандартизации на протокол "возможности сервера GraphQL". Этот протокол позволил бы клиенту быстро определить, какие возможности у сервера GraphQL и какие протоколы он поддерживает.

В текущем состоянии не всегда гарантировано, что клиент GraphQL может автоматически определить, как общаться с сервером GraphQL. Если мы хотим, чтобы экосистема GraphQL росла, мы должны установить стандарты, чтобы клиенты могли общаться с серверами GraphQL без вмешательства человека.

Если вы пытаетесь объединить несколько API GraphQL под одним зонтиком, вы, вероятно, столкнулись с теми же проблемами, с которыми мы столкнулись. Мы надеемся, что смогли дать вам некоторые подсказки о том, как решить эти проблемы.

И, конечно, если вы просто ищете готовый программируемый шлюз API GraphQL, который обрабатывает всю сложность за вас, ознакомьтесь с примерами выше и попробуйте WunderGraph. Это Open Source (Apache 2.0) и бесплатно для использования.