본문으로 건너뛰기
버전: 11.x

구독

소개​

구독은 클라이언트와 서버 간의 실시간 이벤트 스트림의 한 유형입니다. 실시간 업데이트를 클라이언트로 전송해야 할 때 구독을 사용하세요.

tRPC의 구독을 사용하면 클라이언트가 서버와 영속적 연결을 설정하고 유지하며, tracked() 이벤트의 도움으로 연결이 끊겨도 자동으로 재연결을 시도하고 정상적으로 복구합니다.

WebSocket과 서버 전송 이벤트 중 무엇을 선택할까요?​

tRPC의 실시간 구독에는 WebSocket 또는 서버 전송 이벤트(SSE)를 사용할 수 있습니다.

어떤 것을 사용할지 확신이 없다면, 설정이 더 쉽고 WebSocket 서버 설정이 필요하지 않으므로 구독에 SSE 사용을 권장합니다.

참고 프로젝트​

유형예제 유형링크
WebSockets최소한의 Node.js WebSockets 예제/examples/standalone-server
SSE풀스택 SSE 구현github.com/trpc/examples-next-sse-chat
WebSockets풀스택 WebSockets 구현github.com/trpc/examples-next-prisma-websockets-starter

기본 예제​

팁

풀스택 예제의 경우 풀스택 SSE 예제를 참조하세요.

server.ts
ts
import EventEmitter, { on } from 'node:events';
import { initTRPC } from '@trpc/server';
 
const t = initTRPC.create();
 
type Post = { id: string; title: string };
const ee = new EventEmitter();
 
export const appRouter = t.router({
  onPostAdd: t.procedure.subscription(async function* (opts) {
    // listen for new events
    for await (const [data] of on(ee, 'add', {
      // Passing the AbortSignal from the request automatically cancels the event emitter when the request is aborted
      signal: opts.signal,
    })) {
      const post = data as Post;
      yield post;
    }
  }),
});
server.ts
ts
import EventEmitter, { on } from 'node:events';
import { initTRPC } from '@trpc/server';
 
const t = initTRPC.create();
 
type Post = { id: string; title: string };
const ee = new EventEmitter();
 
export const appRouter = t.router({
  onPostAdd: t.procedure.subscription(async function* (opts) {
    // listen for new events
    for await (const [data] of on(ee, 'add', {
      // Passing the AbortSignal from the request automatically cancels the event emitter when the request is aborted
      signal: opts.signal,
    })) {
      const post = data as Post;
      yield post;
    }
  }),
});

tracked()를 사용한 id 자동 추적 (권장)​

yield를 사용하여 tracked() 헬퍼로 이벤트를 보내고 id를 포함하면, 클라이언트는 연결이 끊겨도 자동으로 재연결하여 마지막 알려진 ID를 전송합니다.

구독을 초기화할 때 초기 lastEventId를 전송할 수 있으며, 브라우저가 데이터를 수신함에 따라 자동으로 업데이트됩니다.

  • SSE의 경우, 이는 EventSource-spec의 일부이며 .input()의 lastEventId를 통해 전달됩니다.
  • WebSockets의 경우, wsLink가 자동으로 마지막 알려진 ID를 전송하고 브라우저가 데이터를 수신함에 따라 업데이트합니다.
팁

lastEventId에 기반하여 데이터를 가져오고 모든 이벤트를 캡처하는 것이 중요하다면, 풀스택 SSE 예제에서 수행하듯 데이터베이스에서 이벤트를 가져오기 전에 이벤트 리스너를 설정해야 합니다. 이렇게 하면 lastEventId에 기반하여 원래 배치의 yield 처리 중에 방금 방출된 이벤트가 무시되는 것을 방지할 수 있습니다.

ts
import EventEmitter, { on } from 'node:events';
import { initTRPC, tracked } from '@trpc/server';
import { z } from 'zod';
 
class IterableEventEmitter extends EventEmitter {
  toIterable(eventName: string, opts?: { signal?: AbortSignal }) {
    return on(this, eventName, opts);
  }
}
 
type Post = { id: string; title: string };
 
const t = initTRPC.create();
const publicProcedure = t.procedure;
const router = t.router;
 
const ee = new IterableEventEmitter();
 
export const subRouter = router({
  onPostAdd: publicProcedure
    .input(
      z
        .object({
          // lastEventId is the last event id that the client has received
          // On the first call, it will be whatever was passed in the initial setup
          // If the client reconnects, it will be the last event id that the client received
          lastEventId: z.string().nullish(),
        })
        .optional(),
    )
    .subscription(async function* (opts) {
      // We start by subscribing to the ee so that we don't miss any new events while fetching
      const iterable = ee.toIterable('add', {
        // Passing the AbortSignal from the request automatically cancels the event emitter when the request is aborted
        signal: opts.signal,
      });
 
      if (opts.input?.lastEventId) {
        // [...] get the posts since the last event id and yield them
        // const items = await db.post.findMany({ ... })
        // for (const item of items) {
        //   yield tracked(item.id, item);
        // }
      }
      // listen for new events from the iterable we set up above
      for await (const [data] of iterable) {
        const post = data as Post;
        // tracking the post id ensures the client can reconnect at any time and get the latest events since this id
        yield tracked(post.id, post);
      }
    }),
});
ts
import EventEmitter, { on } from 'node:events';
import { initTRPC, tracked } from '@trpc/server';
import { z } from 'zod';
 
class IterableEventEmitter extends EventEmitter {
  toIterable(eventName: string, opts?: { signal?: AbortSignal }) {
    return on(this, eventName, opts);
  }
}
 
type Post = { id: string; title: string };
 
const t = initTRPC.create();
const publicProcedure = t.procedure;
const router = t.router;
 
const ee = new IterableEventEmitter();
 
export const subRouter = router({
  onPostAdd: publicProcedure
    .input(
      z
        .object({
          // lastEventId is the last event id that the client has received
          // On the first call, it will be whatever was passed in the initial setup
          // If the client reconnects, it will be the last event id that the client received
          lastEventId: z.string().nullish(),
        })
        .optional(),
    )
    .subscription(async function* (opts) {
      // We start by subscribing to the ee so that we don't miss any new events while fetching
      const iterable = ee.toIterable('add', {
        // Passing the AbortSignal from the request automatically cancels the event emitter when the request is aborted
        signal: opts.signal,
      });
 
      if (opts.input?.lastEventId) {
        // [...] get the posts since the last event id and yield them
        // const items = await db.post.findMany({ ... })
        // for (const item of items) {
        //   yield tracked(item.id, item);
        // }
      }
      // listen for new events from the iterable we set up above
      for await (const [data] of iterable) {
        const post = data as Post;
        // tracking the post id ensures the client can reconnect at any time and get the latest events since this id
        yield tracked(post.id, post);
      }
    }),
});

루프에서 데이터 가져오기​

이 방식은 데이터베이스 같은 소스에서 새 데이터를 주기적으로 확인해 클라이언트로 전송할 때 유용합니다.

server.ts
ts
import { tracked } from '@trpc/server';
import { z } from 'zod';
import { publicProcedure, router } from './trpc';
 
export const subRouter = router({
  onPostAdd: publicProcedure
    .input(
      z.object({
        // lastEventId is the last event id that the client has received
        // On the first call, it will be whatever was passed in the initial setup
        // If the client reconnects, it will be the last event id that the client received
        // The id is the createdAt of the post
        lastEventId: z.coerce.date().nullish(),
      }),
    )
    .subscription(async function* (opts) {
      // `opts.signal` is an AbortSignal that will be aborted when the client disconnects.
      let lastEventId = opts.input?.lastEventId ?? null;
 
      // We use a `while` loop that checks `!opts.signal.aborted`
      while (!opts.signal!.aborted) {
        const posts = await db.post.findMany({
          // If we have a `lastEventId`, we only fetch posts created after it.
          where: lastEventId
            ? {
                createdAt: {
                  gt: lastEventId,
                },
              }
            : undefined,
          orderBy: {
            createdAt: 'asc',
          },
        });
 
        for (const post of posts) {
          // `tracked` is a helper that sends an `id` with each event.
          // This allows the client to resume from the last received event upon reconnection.
          yield tracked(post.createdAt.toJSON(), post);
          lastEventId = post.createdAt;
        }
 
        // Wait for a bit before polling again to avoid hammering the database.
        await sleep(1_000);
      }
    }),
});
server.ts
ts
import { tracked } from '@trpc/server';
import { z } from 'zod';
import { publicProcedure, router } from './trpc';
 
export const subRouter = router({
  onPostAdd: publicProcedure
    .input(
      z.object({
        // lastEventId is the last event id that the client has received
        // On the first call, it will be whatever was passed in the initial setup
        // If the client reconnects, it will be the last event id that the client received
        // The id is the createdAt of the post
        lastEventId: z.coerce.date().nullish(),
      }),
    )
    .subscription(async function* (opts) {
      // `opts.signal` is an AbortSignal that will be aborted when the client disconnects.
      let lastEventId = opts.input?.lastEventId ?? null;
 
      // We use a `while` loop that checks `!opts.signal.aborted`
      while (!opts.signal!.aborted) {
        const posts = await db.post.findMany({
          // If we have a `lastEventId`, we only fetch posts created after it.
          where: lastEventId
            ? {
                createdAt: {
                  gt: lastEventId,
                },
              }
            : undefined,
          orderBy: {
            createdAt: 'asc',
          },
        });
 
        for (const post of posts) {
          // `tracked` is a helper that sends an `id` with each event.
          // This allows the client to resume from the last received event upon reconnection.
          yield tracked(post.createdAt.toJSON(), post);
          lastEventId = post.createdAt;
        }
 
        // Wait for a bit before polling again to avoid hammering the database.
        await sleep(1_000);
      }
    }),
});

서버에서 구독 중지​

서버에서 구독을 중지해야 한다면, 생성기 함수에서 단순히 return을 실행하세요.

ts
export const subRouter = router({
  onPostAdd: publicProcedure
    .input(
      z.object({
        lastEventId: z.coerce.number().min(0).optional(),
      }),
    )
    .subscription(async function* (opts) {
      let index = opts.input.lastEventId ?? 0;
      while (!opts.signal!.aborted) {
        const idx = index++;
        if (idx > 100) {
          // With this, the subscription will stop and the client will disconnect
          return;
        }
        await new Promise((resolve) => setTimeout(resolve, 10));
      }
    }),
});
ts
export const subRouter = router({
  onPostAdd: publicProcedure
    .input(
      z.object({
        lastEventId: z.coerce.number().min(0).optional(),
      }),
    )
    .subscription(async function* (opts) {
      let index = opts.input.lastEventId ?? 0;
      while (!opts.signal!.aborted) {
        const idx = index++;
        if (idx > 100) {
          // With this, the subscription will stop and the client will disconnect
          return;
        }
        await new Promise((resolve) => setTimeout(resolve, 10));
      }
    }),
});

클라이언트에서는 단순히 구독을 .unsubscribe()합니다.

부수 효과 정리​

구독의 부수 효과를 정리해야 한다면 try...finally 패턴을 사용할 수 있습니다. trpc는 구독이 어떤 이유로든 중지될 때 Generator Instance의 .return()을 호출하기 때문입니다.

ts
import EventEmitter, { on } from 'events';
import { initTRPC } from '@trpc/server';
 
type Post = { id: string; title: string };
 
const t = initTRPC.create();
const publicProcedure = t.procedure;
const router = t.router;
 
const ee = new EventEmitter();
 
export const subRouter = router({
  onPostAdd: publicProcedure.subscription(async function* (opts) {
    let timeout: ReturnType<typeof setTimeout> | undefined;
    try {
      for await (const [data] of on(ee, 'add', {
        signal: opts.signal,
      })) {
        timeout = setTimeout(() => console.log('Pretend like this is useful'));
        const post = data as Post;
        yield post;
      }
    } finally {
      if (timeout) clearTimeout(timeout);
    }
  }),
});
ts
import EventEmitter, { on } from 'events';
import { initTRPC } from '@trpc/server';
 
type Post = { id: string; title: string };
 
const t = initTRPC.create();
const publicProcedure = t.procedure;
const router = t.router;
 
const ee = new EventEmitter();
 
export const subRouter = router({
  onPostAdd: publicProcedure.subscription(async function* (opts) {
    let timeout: ReturnType<typeof setTimeout> | undefined;
    try {
      for await (const [data] of on(ee, 'add', {
        signal: opts.signal,
      })) {
        timeout = setTimeout(() => console.log('Pretend like this is useful'));
        const post = data as Post;
        yield post;
      }
    } finally {
      if (timeout) clearTimeout(timeout);
    }
  }),
});

오류 처리​

생성기 함수에서 오류를 throw하면 백엔드의 trpc의 onError()로 전달됩니다.

throw된 오류가 5xx 오류인 경우, 클라이언트는 tracked()를 사용하여 추적된 마지막 이벤트 ID에 기반하여 자동으로 재연결을 시도합니다. 다른 오류의 경우, 구독이 취소되고 onError() 콜백으로 전달됩니다.

출력 검증​

구독은 비동기 반복자(async iterator)이므로, 출력을 검증하려면 반복자를 통해 진행해야 합니다.

Zod v4 사용 예제​

zAsyncIterable.ts
ts
import type { TrackedEnvelope } from '@trpc/server';
import { isTrackedEnvelope, tracked } from '@trpc/server';
import { z } from 'zod';
 
function isAsyncIterable<TValue, TReturn = unknown>(
  value: unknown,
): value is AsyncIterable<TValue, TReturn> {
  return !!value && typeof value === 'object' && Symbol.asyncIterator in value;
}
const trackedEnvelopeSchema =
  z.custom<TrackedEnvelope<unknown>>(isTrackedEnvelope);
 
/**
 * A Zod schema helper designed specifically for validating async iterables. This schema ensures that:
 * 1. The value being validated is an async iterable.
 * 2. Each item yielded by the async iterable conforms to a specified type.
 * 3. The return value of the async iterable, if any, also conforms to a specified type.
 */
export function zAsyncIterable<
  TYieldIn,
  TYieldOut,
  TReturnIn = void,
  TReturnOut = void,
  Tracked extends boolean = false,
>(opts: {
  /**
   * Validate the value yielded by the async generator
   */
  yield: z.ZodType<TYieldOut, TYieldIn>;
  /**
   * Validate the return value of the async generator
   * @remarks not applicable for subscriptions
   */
  return?: z.ZodType<TReturnOut, TReturnIn>;
  /**
   * Whether the yielded values are tracked
   * @remarks only applicable for subscriptions
   */
  tracked?: Tracked;
}) {
  return z
    .custom<
      AsyncIterable<
        Tracked extends true ? TrackedEnvelope<TYieldIn> : TYieldIn,
        TReturnIn
      >
    >((val) => isAsyncIterable(val))
    .transform(async function* (iter) {
      const iterator = iter[Symbol.asyncIterator]();
 
      try {
        let next;
        while ((next = await iterator.next()) && !next.done) {
          if (opts.tracked) {
            const [id, data] = trackedEnvelopeSchema.parse(next.value);
            yield tracked(id, await opts.yield.parseAsync(data));
            continue;
          }
          yield opts.yield.parseAsync(next.value);
        }
        if (opts.return) {
          return await opts.return.parseAsync(next.value);
        }
        return;
      } finally {
        await iterator.return?.();
      }
    }) as z.ZodType<
    AsyncIterable<
      Tracked extends true ? TrackedEnvelope<TYieldIn> : TYieldIn,
      TReturnIn,
      unknown
    >,
    AsyncIterable<
      Tracked extends true ? TrackedEnvelope<TYieldOut> : TYieldOut,
      TReturnOut,
      unknown
    >
  >;
}
zAsyncIterable.ts
ts
import type { TrackedEnvelope } from '@trpc/server';
import { isTrackedEnvelope, tracked } from '@trpc/server';
import { z } from 'zod';
 
function isAsyncIterable<TValue, TReturn = unknown>(
  value: unknown,
): value is AsyncIterable<TValue, TReturn> {
  return !!value && typeof value === 'object' && Symbol.asyncIterator in value;
}
const trackedEnvelopeSchema =
  z.custom<TrackedEnvelope<unknown>>(isTrackedEnvelope);
 
/**
 * A Zod schema helper designed specifically for validating async iterables. This schema ensures that:
 * 1. The value being validated is an async iterable.
 * 2. Each item yielded by the async iterable conforms to a specified type.
 * 3. The return value of the async iterable, if any, also conforms to a specified type.
 */
export function zAsyncIterable<
  TYieldIn,
  TYieldOut,
  TReturnIn = void,
  TReturnOut = void,
  Tracked extends boolean = false,
>(opts: {
  /**
   * Validate the value yielded by the async generator
   */
  yield: z.ZodType<TYieldOut, TYieldIn>;
  /**
   * Validate the return value of the async generator
   * @remarks not applicable for subscriptions
   */
  return?: z.ZodType<TReturnOut, TReturnIn>;
  /**
   * Whether the yielded values are tracked
   * @remarks only applicable for subscriptions
   */
  tracked?: Tracked;
}) {
  return z
    .custom<
      AsyncIterable<
        Tracked extends true ? TrackedEnvelope<TYieldIn> : TYieldIn,
        TReturnIn
      >
    >((val) => isAsyncIterable(val))
    .transform(async function* (iter) {
      const iterator = iter[Symbol.asyncIterator]();
 
      try {
        let next;
        while ((next = await iterator.next()) && !next.done) {
          if (opts.tracked) {
            const [id, data] = trackedEnvelopeSchema.parse(next.value);
            yield tracked(id, await opts.yield.parseAsync(data));
            continue;
          }
          yield opts.yield.parseAsync(next.value);
        }
        if (opts.return) {
          return await opts.return.parseAsync(next.value);
        }
        return;
      } finally {
        await iterator.return?.();
      }
    }) as z.ZodType<
    AsyncIterable<
      Tracked extends true ? TrackedEnvelope<TYieldIn> : TYieldIn,
      TReturnIn,
      unknown
    >,
    AsyncIterable<
      Tracked extends true ? TrackedEnvelope<TYieldOut> : TYieldOut,
      TReturnOut,
      unknown
    >
  >;
}

이제 이 헬퍼를 사용하여 구독 프로시저의 출력을 검증할 수 있습니다:

_app.ts
ts
import { tracked } from '@trpc/server';
import { z } from 'zod';
import { publicProcedure, router } from './trpc';
import { zAsyncIterable } from './zAsyncIterable';
 
export const appRouter = router({
  mySubscription: publicProcedure
    .input(
      z.object({
        lastEventId: z.coerce.number().min(0).optional(),
      }),
    )
    .output(
      zAsyncIterable({
        yield: z.object({
          count: z.number(),
        }),
        tracked: true,
      }),
    )
    .subscription(async function* (opts) {
      let index = opts.input.lastEventId ?? 0;
      while (true) {
        index++;
        yield tracked(String(index), {
          count: index,
        });
 
        await new Promise((resolve) => setTimeout(resolve, 1000));
      }
    }),
});
_app.ts
ts
import { tracked } from '@trpc/server';
import { z } from 'zod';
import { publicProcedure, router } from './trpc';
import { zAsyncIterable } from './zAsyncIterable';
 
export const appRouter = router({
  mySubscription: publicProcedure
    .input(
      z.object({
        lastEventId: z.coerce.number().min(0).optional(),
      }),
    )
    .output(
      zAsyncIterable({
        yield: z.object({
          count: z.number(),
        }),
        tracked: true,
      }),
    )
    .subscription(async function* (opts) {
      let index = opts.input.lastEventId ?? 0;
      while (true) {
        index++;
        yield tracked(String(index), {
          count: index,
        });
 
        await new Promise((resolve) => setTimeout(resolve, 1000));
      }
    }),
});