Subscriptions

query คืนค่าครั้งเดียว mutation ก็คืนค่าครั้งเดียว ส่วน subscription ต่างออกไป client subscribe แค่ครั้งเดียว จากนั้น server จะ push payload ใหม่ให้เรื่อย ๆ ตราบใดที่ connection ยังเปิดอยู่ นี่คือ root type ตัวที่สาม และเป็นวิธีที่ GraphQL ใช้จำลองเรียลไทม์ เช่น มีรีวิวใหม่เข้ามา เพลงประมวลผลเสร็จ หรือมีคนเริ่มพิมพ์ โดย client ไม่ต้อง poll ซ้ำ ๆ

ตัว runner ใน browser ที่ใช้ในบทเรียนอื่นรัน operation ครั้งเดียวแล้วหยุด จึงสาธิตสตรีมสด ๆ ไม่ได้ บทเรียนนี้เลยอธิบาย subscription ผ่าน code กับ sequence diagram แล้วชี้ทางไปที่ server Yoga ที่รันจริงได้ ให้คุณเปิดดูการสตรีมด้วยตัวเอง

การประกาศ subscription

field ของ subscription อยู่บน root type Subscription หน้าตาเหมือน field ของ query ทุกอย่าง คือมีชื่อ มี argument มี type คืนค่า ต่างกันที่ความหมาย ซึ่งแปลว่า “ทุกครั้งที่ event ที่ตรงเงื่อนไขเกิดขึ้น ให้ push อันนี้มาให้หนึ่งครั้ง”

type Subscription {
  reviewAdded(trackId: ID!): Review!
}

argument trackId ทำหน้าที่จำกัดขอบเขตของสตรีม เพราะ client อยากได้รีวิวของเพลงเดียว ไม่ใช่ทั้ง catalogue ส่วน type คืนค่า Review! คือรูปร่างของ payload แต่ละก้อนที่ push ออกไป

resolver ของ subscription ไม่ใช่ function เดียว แต่เป็นสอง function และการแยกสองตัวนี้ให้ขาดคือแนวคิดทั้งหมดของบทนี้

subscribe คืน async iterator ที่เป็นแหล่งของ event ทุกครั้งที่มีอะไร push เข้าไปบน iterator นั้น server จะตื่นขึ้นมาแล้วผลิตผลลัพธ์หนึ่งชุดให้ client
resolve คือ field resolver ธรรมดาที่รัน หนึ่งครั้งต่อหนึ่ง event รับค่าที่ iterator yield ออกมา แล้ว map ไปเป็น payload ตามที่ client เลือกไว้

ใน graphql-yoga และ server ส่วนใหญ่ แหล่งของ event คือบัส pub/sub ฝั่ง mutation publish event ออกไป แล้ว iterator ของ subscribe ในทุก subscription ที่ตรงเงื่อนไขก็ yield event นั้นต่อ

import { createPubSub } from 'graphql-yoga';

const pubSub = createPubSub<{ reviewAdded: [trackId: string, review: Review] }>();

export const resolvers = {
  Mutation: {
    addReview: (_parent, { input }) => {
      const review = saveReview(input);
      // Publish the event — every matching subscriber is woken up.
      pubSub.publish('reviewAdded', input.trackId, review);
      return { review, userErrors: [] };
    },
  },
  Subscription: {
    reviewAdded: {
      // subscribe: the event source, scoped to one track.
      subscribe: (_parent, { trackId }) => pubSub.subscribe('reviewAdded', trackId),
      // resolve: map each published event to the payload.
      resolve: (review: Review) => review,
    },
  },
};

ลำดับการทำงานคือ mutation เขียนข้อมูลแล้ว publish จากนั้นบัส pub/sub กระจาย event ออกไป iterator ของ subscribe ในแต่ละ subscriber yield event ออกมา resolve จัดรูปร่างให้ แล้ว client ก็ได้รับ payload การเขียนชุดเดียวกับที่บทเรียนก่อนคืนค่าแบบ synchronous ตอนนี้ ยัง แจ้งทุกคนที่ฟังอยู่ไปพร้อมกันด้วย

sequenceDiagram
  participant C as Client
  participant S as GraphQL Server
  participant B as Pub/Sub bus
  participant W as Writer (another client)
  C->>S: subscription reviewAdded(trackId: "t1")
  S->>B: subscribe to "reviewAdded" for t1
  Note over C,S: stream stays open
  W->>S: mutation addReview(input)
  S->>B: publish "reviewAdded" event
  B-->>S: yield event to subscriber
  S-->>C: push Review payload
  W->>S: mutation addReview(input) again
  S->>B: publish again
  B-->>S: yield event
  S-->>C: push next Review payload

subscribe ครั้งเดียวเปิดสตรีม แล้วทุก mutation ที่ตามมา publish event ที่ push ไปหา client

Transport: สตรีมไปถึง client ได้อย่างไร

query และ mutation วิ่งบน HTTP request/response เดียวจบ ส่วน subscription ต้องการช่องทางที่เปิดค้างและ push ได้ จึงใช้ transport คนละแบบ

Server-Sent Events (SSE) — HTTP response ที่เปิดค้างยาว ๆ แล้วสตรีม event แบบข้อความไปทางเดียวจาก server ถึง client วิ่งบน HTTP ธรรมดา ผ่าน proxy ส่วนใหญ่ได้ และเป็น default ยุคใหม่ของ graphql-yoga ผ่าน protocol graphql-sse แม้จะเป็นทางเดียว แต่เท่านี้ก็พอสำหรับ subscription แล้ว
WebSocket — connection แบบ full-duplex ที่ใช้คู่กับ protocol graphql-ws มาแต่เดิม ทำได้มากกว่าเพราะสื่อสารสองทาง และ client รองรับกันแพร่หลาย แต่ก็เป็น connection แยกอีกเส้นที่ต้องดูแล และ route ผ่าน infrastructure บางแบบยากกว่า

ทั้งสองแบบพา subscription operation ของ GraphQL ตัวเดียวกัน ต่างกันแค่ระบบท่อข้างใน เริ่มที่ SSE ไว้ก่อน ยกเว้นจะมีเหตุผลชัดเจนว่าต้องใช้ WebSocket

เมื่อไหร่ที่ไม่ควรใช้ subscription

subscription คือ connection ที่มี state และเปิดค้างยาว กินหน่วยความจำ server และทำให้การ scale ยุ่งขึ้น หยิบมาใช้เฉพาะตอนที่ข้อมูลเปลี่ยนตามจังหวะของ server จริง ๆ กรณีอื่นให้เลือกทางอื่นแทน

ผลลัพธ์จาก mutation ก็พอแล้ว ถ้าการเปลี่ยนแปลงเกิดจากการกระทำของ client เอง ก็ให้ mutation คืนข้อมูลที่กระทบกลับไปเลย อย่า subscribe การเขียนของตัวเอง
ข้อมูลเปลี่ยนไม่บ่อย หรือ client poll เอาได้ dashboard ที่รีเฟรชทุก 30 วินาทีไม่ต้องมี connection ถาวร ยิง query เป็นระยะ (หรือ refetch ตอน focus) ง่ายกว่าและถูกกว่า
ต้องการผลลัพธ์ async แค่ครั้งเดียว งานแบบ “สร้างรายงานนี้ เสร็จแล้วบอกด้วย” ใช้การ poll สถานะงานหรือ webhook ก็จบ ไม่ต้องถึงมือ subscription

subscription จะคุ้มกับความซับซ้อนก็ต่อเมื่อ client หลายตัวต้องเห็น event เดียวกันในวินาทีที่เกิดขึ้น เช่น ฟีดสด สถานะออนไลน์ หรือราคาที่ขยับตลอด

server สตรีมมิงที่คุณรันเองได้

schema ด้านล่างเป็นโปรเจกต์ Yoga เต็มตัว export ทั้ง typeDefs และ resolvers โดยมี mutation ที่ publish และ subscription ที่สตรีม ตัว runner ใน browser เปิดสตรีมค้างไม่ได้ ให้เปิดโปรเจกต์นี้ใน StackBlitz แทน แล้วยิง addReview ใน GraphiQL แท็บหนึ่ง พร้อมดู reviewAdded push payload ในอีกแท็บ

Node.js

import { createPubSub } from 'graphql-yoga';

type Review = { id: string; trackId: string; rating: number; body: string | null };

export const typeDefs = /* GraphQL */ `
  type Review {
    id: ID!
    trackId: ID!
    rating: Int!
    body: String
  }

  input AddReviewInput {
    trackId: ID!
    rating: Int!
    body: String
  }

  type AddReviewPayload {
    review: Review
  }

  type Query {
    reviews(trackId: ID!): [Review!]!
  }

  type Mutation {
    addReview(input: AddReviewInput!): AddReviewPayload!
  }

  type Subscription {
    reviewAdded(trackId: ID!): Review!
  }
`;

const store: Review[] = [];
let nextId = 1;

const pubSub = createPubSub<{ reviewAdded: [trackId: string, review: Review] }>();

export const resolvers = {
  Query: {
    reviews: (_p: unknown, { trackId }: { trackId: string }) =>
      store.filter((r) => r.trackId === trackId),
  },
  Mutation: {
    addReview: (
      _p: unknown,
      { input }: { input: { trackId: string; rating: number; body?: string } },
    ) => {
      const review: Review = {
        id: String(nextId++),
        trackId: input.trackId,
        rating: input.rating,
        body: input.body ?? null,
      };
      store.push(review);
      // Notify every subscriber listening on this track.
      pubSub.publish('reviewAdded', input.trackId, review);
      return { review };
    },
  },
  Subscription: {
    reviewAdded: {
      subscribe: (_p: unknown, { trackId }: { trackId: string }) =>
        pubSub.subscribe('reviewAdded', trackId),
      resolve: (review: Review) => review,
    },
  },
};

Needs the Node.js runtime — open in StackBlitz to run.

วิธีดูการสตรีมคือ เปิดโปรเจกต์ขึ้นมา แล้วใน GraphiQL แท็บแรกรัน subscription { reviewAdded(trackId: "t1") { id rating } } ปล่อยค้างไว้ จากนั้นแท็บที่สองรัน mutation addReview ด้วย trackId: "t1" แท็บ subscription จะอัปเดตทันทีที่ mutation publish นี่แหละคือคุณค่าทั้งหมดของสตรีม และไม่มีการ poll แบบไหนสู้ latency ระดับนี้ได้

ข้อดี	ข้อแลกเปลี่ยน
real-time push — client รับข้อมูลทันทีโดยไม่ต้อง poll	persistent connection — server ต้องรักษา state ต่อ subscriber ทุกคน
latency ต่ำกว่า polling มาก — event มาถึง client ในชั่วขณะ	WebSocket ไม่ผ่าน HTTP cache — CDN ช่วยไม่ได้
client กำหนด field ที่ต้องการใน subscription เหมือน query	ซับซ้อนกว่า deployment — ต้องการ sticky session หรือ pub/sub bus กลาง

ข้อผิดพลาดที่พบบ่อย

ใช้ subscription แทน query ทุกอย่าง

อาการ:

subscribe userProfile เพื่อดูข้อมูลที่แทบไม่เปลี่ยน
server แบกรับ connection จำนวนมากโดยไม่จำเป็น
query ธรรมดาเร็วกว่าและ cache ได้

ไม่กรอง event ด้วย argument

อาการ:

subscription { reviewAdded } รับรีวิวทุกเพลง แทนที่จะเป็นเพลงที่ client สนใจ
client ต้อง filter ฝั่งตัวเองทุกครั้ง — traffic สูงโดยไม่จำเป็น
resolver ควรใช้ argument เช่น trackId เพื่อ scope subscription ตั้งแต่ต้น

💡 ตัวอย่างจากของจริง

Slack:

ใช้ WebSocket subscription สำหรับ message real-time ใน channel

subscribe แยกต่อ channel เพื่อไม่ให้รับ event จาก workspace ทั้งหมด

mutation sendMessage publish event — subscription resolver ส่งต่อให้ subscriber ที่ตรงกัน

Figma:

subscription สำหรับ cursor position และ object selection ของ collaborator

กรอง event ด้วย documentId + userId เพื่อลด noise

◈ subscription คือช่องทางที่เปิดอยู่ ไม่ใช่การอ่านเพิ่ม

subscription คือ connection ที่เปิดค้างและมี state คือ subscribe ครั้งเดียว แล้วมี payload ถูก push ตามมาหลายครั้ง จนกว่า client จะตัด connection เวลา implement ให้แบ่งเป็นสองครึ่ง คือ subscribe ผลิตสตรีมของ event (ปกติมาจากบัส pub/sub ที่ mutation publish เข้าไป) ส่วน resolve จัดรูปร่างแต่ละ event ให้เป็น payload ส่งผ่าน SSE เป็นค่าเริ่มต้น หรือ WebSocket เมื่อจำเป็นต้องสื่อสารสองทาง และอย่าหลงไป subscribe ทุกอย่าง ถ้าการเปลี่ยนแปลงเกิดจาก mutation ของ client เอง ก็แค่คืนผลลัพธ์กลับไป เก็บ subscription ไว้ใช้กับ event ที่มาตามนาฬิกาของ server ไม่ใช่ของ client

Subscriptions

การประกาศ subscription

subscribe เทียบกับ resolve

Transport: สตรีมไปถึง client ได้อย่างไร

เมื่อไหร่ที่ไม่ควรใช้ subscription

server สตรีมมิงที่คุณรันเองได้

ข้อผิดพลาดที่พบบ่อย