GraphQL Subscriptions: Ein tiefer Einblick in WebSocket- und SSE-Transportschichten

Einleitung

In der heutigen dynamischen Weblandschaft sind Echtzeit-Datenaktualisierungen kein Luxus mehr, sondern eine grundlegende Erwartung. Von Live-Chat-Anwendungen und kollaborativen Dokumenten bis hin zu Finanz-Dashboards und IoT-Überwachung verlangen Benutzer sofortiges Feedback und aktuelle Informationen. GraphQL mit seinen leistungsstarken Datenabruffunktionen erfüllt diesen Bedarf durch seine Abonnementfunktion, die esclients ermöglicht, Echtzeitaktualisierungen vom Server zu empfangen. Doch unter der eleganten GraphQL-Abonnement-API liegt eine entscheidende Entscheidung: Wie werden diese Echtzeitnachrichten transportiert? Dies führt uns zu einem grundlegenden Kampf zwischen zwei prominenten Webtechnologien: WebSockets und Server-Sent Events (SSE). Das Verständnis ihrer Nuancen, Stärken und Schwächen ist für Backend-Entwickler, die robuste, skalierbare und effiziente Echtzeit-GraphQL-Anwendungen erstellen möchten, von größter Bedeutung. Dieser Artikel wird die Kernunterschiede zwischen diesen Transportschichten untersuchen und ihre Prinzipien, Implementierungsstrategien und praktischen Anwendungsszenarien erkunden, um Ihnen bei der fundierten Auswahl zu helfen.

Kernkonzepte

Bevor wir uns in den Kampf stürzen, wollen wir ein klares Verständnis der Schlüsseltechnologien entwickeln, die in dieser Diskussion enthalten sind.

GraphQL-Abos

GraphQL-Abos sind eine leistungsstarke Funktion in GraphQL, die es clients ermöglicht, Ereignisse vom Server zu abonnieren. Im Gegensatz zu Abfragen (die Daten einmal abrufen) und Mutationen (die Daten ändern) aufrechterhalten Abos eine persistente Verbindung zwischen Client und Server. Wenn auf dem Server ein bestimmtes Ereignis eintritt, wird eine Nachricht in Echtzeit an alle abonnierten clients gesendet. Dies wird durch die Definition eines Subscription-Typs in Ihrem GraphQL-Schema erreicht, das Felder verfügbar macht, die clients abonnieren können.

type Subscription {
  commentAdded(postId: ID!): Comment
  postLiked(postId: ID!): Post
}

Wenn ein client commentAdded abonniert, sendet der Server neue Comment-Objekte, wann immer ein neuer Kommentar zur angegebenen postId hinzugefügt wird.

WebSockets

WebSockets bieten einen vollduplexen, persistenten Kommunikationskanal über eine einzige TCP-Verbindung. Das bedeutet, dass, sobald eine WebSocket-Verbindung hergestellt ist, sowohl der client als auch der Server Nachrichten unabhängig und gleichzeitig senden und empfangen können. Diese bidirektionale Fähigkeit macht WebSockets ideal für Anwendungen, die häufige, latenzarme Zwei-Wege-Kommunikation erfordern, wie z. B. Sofortnachrichten, Online-Spiele und Live-Kollaborationstools.

Server-Sent Events (SSE)

Server-Sent Events (SSE) sind ein Standard zum Pushen von unidirektionalen Event-Daten von einem Server an einen client über eine einzige HTTP-Verbindung. Im Gegensatz zu WebSockets ist SSE unidirektional – Daten fließen nur vom Server zum client. Dies macht SSE besonders gut geeignet für Szenarien, in denen der client hauptsächlich Updates empfangen muss, ohne unbedingt häufig Daten an den Server zurücksenden zu müssen. Denken Sie an Börsenticker, Newsfeeds oder Echtzeit-Dashboards, bei denen der Server die primäre Informationsquelle ist. SSE profitiert auch davon, auf HTTP aufzubauen, was es Firewall-freundlich macht und oft einfacher für das Server-Pushing zu implementieren ist.

Der Transportschlacht

Vergleichen wir nun WebSockets und SSE im Kontext von GraphQL-Abos und untersuchen ihre Prinzipien, Implementierung und Anwendungsszenarien.

WebSocket-Prinzipien und Implementierung mit GraphQL-Abos

WebSockets bieten einen hocheffizienten und vielseitigen Transport für GraphQL-Abos aufgrund ihrer bidirektionalen Natur.

Prinzipien:

• Persistente Verbindung: Eine einzige TCP-Verbindung wird hergestellt und offen gehalten.
• Full-Duplex: Client und Server können gleichzeitig Senden und Empfangen.
• Geringer Overhead: Nach Abschluss des Handshakes sind die Datenframes kleiner als HTTP-Anfragen.
• Protokollagnostisch: WebSockets können jede Art von Daten (Text, Binär) transportieren.

Implementierung:

Die Implementierung von GraphQL-Abos über WebSockets beinhaltet typischerweise einen dedizierten WebSocket-Server oder einen kompatiblen HTTP-Server, der WebSocket-Upgrades unterstützt. Bibliotheken wie graphql-ws oder subscriptions-transport-ws (obwohl letzteres zugunsten von graphql-ws als veraltet gilt) werden häufig auf der Serverseite verwendet, um das GraphQL-über-WebSocket-Protokoll zu handhaben.

Schauen wir uns ein vereinfachtes Node.js-Beispiel mit graphql-ws und Express und ws an:

// server.js
import express from 'express';
import { createServer } from 'http';
import { WebSocketServer } from 'ws';
import { useServer } from 'graphql-ws/lib/use/ws';
import { execute, subscribe } from 'graphql';
import { makeExecutableSchema } from '@graphql-tools/schema';
import { PubSub } from 'graphql-subscriptions'; // Für einfaches Veröffentlichen von Ereignissen

const pubsub = new PubSub();
const COMMENTS_CHANNEL = 'COMMENTS_CHANNEL';

const typeDefs = `
  type Comment {
    id: ID!
    content: String!
  }

  type Query {
    hello: String
  }

  type Mutation {
    addComment(content: String!): Comment
  }

  type Subscription {
    commentAdded: Comment
  }
`;

const resolvers = {
  Query: {
    hello: () => 'Hallo GraphQL!',
  },
  Mutation: {
    addComment: (parent, { content }) => {
      const newComment = { id: String(Date.now()), content };
      pubsub.publish(COMMENTS_CHANNEL, { commentAdded: newComment });
      return newComment;
    },
  },
  Subscription: {
    commentAdded: {
      subscribe: () => pubsub.asyncIterator(COMMENTS_CHANNEL),
    },
  },
};

const schema = makeExecutableSchema({ typeDefs, resolvers });

const app = express();
const httpServer = createServer(app);

// WebSocket-Server für GraphQL-Abos erstellen
const wsServer = new WebSocketServer({
  server: httpServer,
  path: '/graphql', // Der WebSocket-Endpunkt
});

useServer(
  {
    schema,
    execute,
    subscribe,
    onConnect: (ctx) => {
      console.log('Client verbunden für GraphQL-Abo');
      // Sie können hier Authentifizierung oder Autorisierung implementieren
    },
    onDisconnect: (ctx, code, reason) => {
      console.log('Client von GraphQL-Abo getrennt');
    },
  },
  wsServer
);

httpServer.listen(4000, () => {
  console.log('GraphQL-Server läuft auf http://localhost:4000');
  console.log('GraphQL-Abos verfügbar unter ws://localhost:4000/graphql');
});

Auf Client-Seite wird eine WebSocket-Client-Bibliothek wie subscriptions-transport-ws (oder graphql-ws-Client) verwendet.

// client.js (Beispiel mit vereinfachtem Client-Setup)
import { createClient } from 'graphql-ws';

const client = createClient({
  url: 'ws://localhost:4000/graphql',
});

const COMMENT_SUBSCRIPTION = `
  subscription OnCommentAdded {
    commentAdded {
      id
      content
    }
  }
`;

(async () => {
  const onNext = ({ data }) => {
    console.log('Kommentar empfangen:', data.commentAdded);
  };

  const onError = (error) => {
    console.error('Abo-Fehler:', error);
  };

  const unsubscribe = client.subscribe(
    { query: COMMENT_SUBSCRIPTION },
    { next: onNext, error: onError, complete: () => console.log('Abo abgeschlossen') }
  );

  // Zur Demonstrationszwecken könnten Sie nach einiger Zeit oder auf Benutzeraktion abbestellen
  // setTimeout(() => {
  //   unsubscribe();
  // }, 10000);
})();

Anwendungsszenarien für WebSockets:

• Echtzeit-Chat-Anwendungen: Bidirektionale Kommunikation ist für das Senden und Empfangen von Nachrichten unerlässlich.
• Kollaborative Editoren: Mehrere Benutzer, die dasselbe Dokument aktualisieren, erfordern eine sofortige bidirektionale Synchronisation.
• Online-Spiele: Latenzarme Zwei-Wege-Kommunikation für Spielstatus-Updates und Spieleraktionen.
• Finanzhandelsplattformen: Hochfrequente Updates und die Platzierung von Benutzeraufträgen.

SSE-Prinzipien und Implementierung mit GraphQL-Abos

SSE bietet einen einfacheren, HTTP-basierten Ansatz zum Pushen von Daten vom Server.

Prinzipien:

• Unidirektional: Daten fließen nur vom Server zum client.
• HTTP-basiert: Verwendet Standard-HTTP für Verbindungen, was es Firewall-freundlich macht.
• Automatische Wiederverbindung: Browser stellen Verbindungen automatisch wieder her, wenn sie unterbrochen werden.
• Einfachheit: Einfachere API als WebSockets für die Server-zu-Client-Kommunikation.

Implementierung:

Um SSE für GraphQL-Abos zu verwenden, hätten Sie typischerweise einen HTTP-Endpunkt, der text/event-stream-Daten streamt. Jedes Ereignis wird als `data: {json_payload}

` formatiert. GraphQL-Server, die SSE für Abos unterstützen, würden normalerweise eine spezielle HTTP-POST-Anfrage dieser SSE-Stream-Daten zuordnen.

Hier ist ein konzeptionelles (und vereinfachtes) serverseitiges Beispiel für SSE mit GraphQL, das eine benutzerdefinierte Implementierung oder eine Bibliothek annimmt, die SSE-Unterstützung für GraphQL-Abos bietet:

// server-sse.js (konzeptionelle SSE-Implementierung für GraphQL-Abos)
import express from 'express';
import { makeExecutableSchema } from '@graphql-tools/schema';
import { PubSub } from 'graphql-subscriptions';
import { execute, subscribe } from 'graphql';

const pubsub = new PubSub();
const COMMENTS_CHANNEL = 'COMMENTS_CHANNEL';

const typeDefs = `
  type Comment {
    id: ID!
    content: String!
  }
  type Query { hello: String }
  type Mutation { addComment(content: String!): Comment }
  type Subscription { commentAdded: Comment }
`;

const resolvers = {
  Query: { hello: () => 'Hallo GraphQL SSE!' },
  Mutation: {
    addComment: (parent, { content }) => {
      const newComment = { id: String(Date.now()), content };
      pubsub.publish(COMMENTS_CHANNEL, { commentAdded: newComment });
      return newComment;
    },
  },
  Subscription: {
    commentAdded: {
      subscribe: () => pubsub.asyncIterator(COMMENTS_CHANNEL),
    },
  },
};

const schema = makeExecutableSchema({ typeDefs, resolvers });
const app = express();
app.use(express.json());

// Mutationsendpunkt (reguläres HTTP POST)
app.post('/graphql', async (req, res) => {
  const { query, variables } = req.body;
  const result = await execute({ schema, document: query, variableValues: variables });
  res.json(result);
});

// SSE-Endpunkt für Abos
app.post('/graphql-sse', async (req, res) => {
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');
  res.flushHeaders(); // Header an den Client senden

  const { query, variables, operationName } = req.body;

  try {
    const subscriber = await subscribe({
      schema,
      document: query,
      variableValues: variables,
      operationName,
      contextValue: {}, // Kontext hinzufügen, falls benötigt
    });

    if (subscriber.errors) {
      res.write(`event: error\ndata: ${JSON.stringify(subscriber.errors)}\n\n`);
      res.end();
      return;
    }

    // Async-Iterator abonnieren
    const iterator = subscriber[Symbol.asyncIterator]();

    const sendEvent = async () => {
      try {
        const { value, done } = await iterator.next();
        if (done) {
          res.write('event: complete\ndata: Abo abgeschlossen\n\n');
          res.end();
          return;
        }
        res.write(`event: message\ndata: ${JSON.stringify(value)}\n\n`);
        // Nächsten Sendevorgang planen
        process.nextTick(sendEvent); // Oder eine kontrolliertere Schleife/Emitter verwenden
      } catch (error) {
        res.write(`event: error\ndata: ${JSON.stringify({ message: error.message })}\n\n`);
        res.end();
      }
    };

    sendEvent(); // Ereignissendungen starten

    req.on('close', () => {
      // Abo bereinigen, wenn der Client sich trennt
      if (subscriber.return) {
        subscriber.return(); // Async-Iterator beenden
      }
      console.log('Client von SSE-Abo getrennt');
    });
  } catch (error) {
    res.write(`event: error\ndata: ${JSON.stringify({ message: error.message })}\n\n`);
    res.end();
  }
});

app.listen(4001, () => {
  console.log('GraphQL-Server mit SSE läuft auf http://localhost:4001 und SSE unter http://localhost:4001/graphql-sse');
});

Auf Client-Seite wird die native EventSource-API des Browsers verwendet:

// client-sse.js
const COMMENT_SUBSCRIPTION_QUERY = `
  subscription OnCommentAdded {
    commentAdded {
      id
      content
    }
  }
`;

// Simulieren einer POST-Anfrage zur Initiierung des SSE-Streams
async function subscribeWithSSE() {
  const response = await fetch('http://localhost:4001/graphql-sse', {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'Accept': 'text/event-stream' // Anzeigen, dass der Client SSE erwartet
    },
    body: JSON.stringify({ query: COMMENT_SUBSCRIPTION_QUERY })
  });

  if (!response.ok) {
    console.error('SSE-Abo konnte nicht initiiert werden:', response.statusText);
    return;
  }

  // Verwenden Sie einen ReadableStreamReader zur Verarbeitung des Event-Streams
  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let buffer = '';

  while (true) {
    const { done, value } = await reader.read();
    if (done) {
      console.log('SSE-Stream beendet.');
      break;
    }

    buffer += decoder.decode(value, { stream: true });
    
    // Vollständige Ereignisse aus dem Puffer verarbeiten
    let eventBoundary;
    while ((eventBoundary = buffer.indexOf('

')) !== -1) {
      const eventData = buffer.substring(0, eventBoundary).trim();
      buffer = buffer.substring(eventBoundary + 2); // Puffer hinter dem Ereignis verschieben

      if (eventData.startsWith('event: message')) {
        const dataPayload = eventData.substring('event: message
data: '.length);
        try {
          const parsedData = JSON.parse(dataPayload);
          console.log('SSE-Kommentar empfangen:', parsedData.data.commentAdded);
        } catch (e) {
          console.error('SSE-Daten konnten nicht geparst werden:', e, dataPayload);
        }
      } else if (eventData.startsWith('event: error')) {
        console.error('SSE-Fehler:', eventData);
      } else if (eventData.startsWith('event: complete')) {
        console.log('SSE-Abo-Abschlussbenachrichtigung.');
        reader.cancel(); // Lesen stoppen
        break;
      }
    }
  }
}

subscribeWithSSE();

Hinweis: Während die EventSource-API die clientseitige Konsumtion vereinfacht, erfordert das serverseitige GraphQL-über-SSE-Protokoll zur Initiierung von Abos normalerweise eine anfängliche HTTP-POST-Anfrage, die die Abo-Abfrage angibt, und der Server streamt dann die text/event-stream-Antwort. Die Bibliothek @graphql-yoga/graphql-sse oder ähnliche kann eine robustere serverseitige Implementierung bieten.

Anwendungsszenarien für SSE:

• Echtzeit-Dashboards: Anzeigen von Metriken, Analysen oder Aktienkursen, bei denen Daten hauptsächlich vom Server zum client fließen.
• Nachrichten-Feeds: Pushen neuer Artikel oder Updates an Benutzer.
• Live-Spielstände: Aktualisieren von Spielständen und Ereignissen in Echtzeit.
• Benachrichtigungssysteme: Senden von Push-Benachrichtigungen an Benutzer.

Fazit

Die Wahl zwischen WebSockets und SSE als Transportschicht für Ihre GraphQL-Abos hängt letztendlich von den spezifischen Anforderungen Ihrer Anwendung ab. WebSockets bieten eine überlegene bidirektionale Kommunikation und sind damit die erste Wahl für interaktive Szenarien wie Chat und kollaboratives Bearbeiten, die latenzarme Zwei-Wege-Datenaustausch erfordern. Im Gegensatz dazu bietet SSE eine einfachere, HTTP-freundliche Lösung für die unidirektionale Server-zu-Client-Datenübertragung, die sich perfekt für Echtzeit-Dashboards, Nachrichten-Feeds und Benachrichtigungssysteme eignet, bei denen der client hauptsächlich Updates konsumiert. Durch sorgfältige Bewertung der Art Ihres Echtzeit-Datenflusses können Sie die Transportschicht auswählen, die die Leistung, Komplexität und Ressourcennutzung Ihrer GraphQL-Abos am besten optimiert.

Im Wesentlichen ermöglichen WebSockets interaktive Echtzeiterlebnisse, während SSE für effizienten passiven Datenkonsum hervorragend geeignet ist.