Подписка GraphQL с использованием отправленных сервером событий и EventSource

Подписки в GraphQL обычно реализуются с использованием WebSockets, а не SSE. И Apollo, и Relay поддерживают использование подписки-транспорт-ws на стороне клиента для прослушивания событий. Apollo Server включает встроенная поддержка для подписок с использованием WebSockets. Если вы просто пытаетесь внедрить подписки, было бы лучше использовать одно из этих существующих решений.

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

Если вы используете Apollo, они поддерживают автоматические постоянные запросы (сокращенно APQ в документации). Если вы не используете Apollo, реализация не должна быть слишком плохой на любом языке. Я бы рекомендовал следовать их соглашениям, чтобы ваши клиенты могли использовать Apollo, если захотят.

Первый раз, когда какой-либо клиент делает запрос EventSource с хэшем запроса, он завершается ошибкой, а затем повторяет запрос с полной полезной нагрузкой к обычной конечной точке GraphQL. Если на сервере включен APQ, последующие запросы GET от всех клиентов с параметрами запроса будут выполняться в соответствии с планом.

После того, как вы решили эту проблему, вам просто нужно создать транспорт событий, отправляемых сервером, для GraphQL (должно быть легко, учитывая, что функция subscribe просто возвращает асинхронный итератор)

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

Здесь есть две вещи: соединение SSE и конечная точка GraphQL. У конечной точки есть спецификация, которой нужно следовать, поэтому простой возврат SSE из запроса на подписку не выполняется, и в любом случае требуется запрос GET. Так что эти двое должны быть разделены.

Как насчет того, чтобы позволить клиенту открыть канал SSE через /graphql-sse, который создает токен канала. Используя этот токен, клиент может затем запрашивать подписки, и события будут поступать по выбранному каналу.

Токен может быть отправлен в качестве первого события на канале SSE, а для передачи токена в запрос он может быть предоставлен клиентом в файле cookie, заголовке запроса или даже в неиспользуемой переменной запроса.

В качестве альтернативы сервер может сохранить последний открытый канал в хранилище сеансов (ограничив клиента одним каналом).

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

На данный момент у вас есть подписка на несколько пакетов для GraphQL через SSE.

graphql-sse

Предоставляет как клиент, так и сервер для использования подписки GraphQL через SSE. Этот пакет имеет специальный обработчик для подписки.

Вот пример использования с экспрессом.

import express from 'express'; // yarn add express
import { createHandler } from 'graphql-sse';

// Create the GraphQL over SSE handler
const handler = createHandler({ schema });

// Create an express app serving all methods on `/graphql/stream`
const app = express();
app.use('/graphql/stream', handler);

app.listen(4000);
console.info('Listening to port 4000');

@graphql-sse/сервер

Предоставляет обработчик сервера для подписки GraphQL. Однако обработка HTTP зависит от используемой вами инфраструктуры.

Отказ от ответственности: Я автор Пакеты @graphql-sse

Вот пример с экспрессом.

import express, { RequestHandler } from "express";
import {
  getGraphQLParameters,
  processSubscription,
} from "@graphql-sse/server";
import { schema } from "./schema";

const app = express();

app.use(express.json());

app.post(path, async (req, res, next) => {
    const request = {
        body: req.body,
        headers: req.headers,
        method: req.method,
        query: req.query,
    };

    const { operationName, query, variables } = getGraphQLParameters(request);
    if (!query) {
        return next();
    }
    const result = await processSubscription({
        operationName,
        query,
        variables,
        request: req,
        schema,
    });

    if (result.type === RESULT_TYPE.NOT_SUBSCRIPTION) {
        return next();
    } else if (result.type === RESULT_TYPE.ERROR) {
        result.headers.forEach(({ name, value }) => res.setHeader(name, value));
        res.status(result.status);
        res.json(result.payload);
    } else if (result.type === RESULT_TYPE.EVENT_STREAM) {
        res.writeHead(200, {
            'Content-Type': 'text/event-stream',
            Connection: 'keep-alive',
            'Cache-Control': 'no-cache',
        });

        result.subscribe((data) => {
            res.write(`data: ${JSON.stringify(data)}\n\n`);
        });

        req.on('close', () => {
            result.unsubscribe();
        });
    }
});

Клиенты

У двух упомянутых выше пакетов есть клиенты-компаньоны. Из-за ограничений EventSource API оба пакета реализуют собственный клиент, который предоставляет опции для отправки заголовков HTTP, полезной нагрузки с публикацией, что не поддерживает EvenSource API. graphql-sse поставляется вместе с клиентом, а @graphql-sse/server имеет дополнительные клиенты в отдельных пакетах.

пример клиента graphql-sse

import { createClient } from 'graphql-sse';

const client = createClient({
  // singleConnection: true, use "single connection mode" instead of the default "distinct connection mode"
  url: 'http://localhost:4000/graphql/stream',
});

// query

  const result = await new Promise((resolve, reject) => {
    let result;
    client.subscribe(
      {
        query: '{ hello }',
      },
      {
        next: (data) => (result = data),
        error: reject,
        complete: () => resolve(result),
      },
    );
  });


 // subscription

  const onNext = () => {
    /* handle incoming values */
  };

  let unsubscribe = () => {
    /* complete the subscription */
  };

  await new Promise((resolve, reject) => {
    unsubscribe = client.subscribe(
      {
        query: 'subscription { greetings }',
      },
      {
        next: onNext,
        error: reject,
        complete: resolve,
      },
    );
  });

;

@graphql-sse/клиент

Спутник @graphql-sse/server.

Пример

import {
  SubscriptionClient,
  SubscriptionClientOptions,
} from '@graphql-sse/client';

const subscriptionClient = SubscriptionClient.create({
    graphQlSubscriptionUrl: 'http://some.host/graphl/subscriptions'
});

const subscription = subscriptionClient.subscribe(
    {
        query: 'subscription { greetings }',
    }
)

const onNext = () => {
    /* handle incoming values */
  };

const onError = () => {
    /* handle incoming errors */
  };

subscription.susbscribe(onNext, onError)

@gaphql-sse/аполло-клиент

Сопутствующий пакет пакета @graph-sse/server для клиента Apollo.

import { split, HttpLink, ApolloClient, InMemoryCache } from '@apollo/client';
import { getMainDefinition } from '@apollo/client/utilities';
import { ServerSentEventsLink } from '@graphql-sse/apollo-client';

const httpLink = new HttpLink({
  uri: 'http://localhost:4000/graphql',
});

const sseLink = new ServerSentEventsLink({
  graphQlSubscriptionUrl: 'http://localhost:4000/graphql',
});

const splitLink = split(
  ({ query }) => {
    const definition = getMainDefinition(query);
    return (
      definition.kind === 'OperationDefinition' &&
      definition.operation === 'subscription'
    );
  },
  sseLink,
  httpLink
);

export const client = new ApolloClient({
  link: splitLink,
  cache: new InMemoryCache(),
});