구독 / WebSockets

비공식 베타 번역

이 페이지는 PageTurner AI로 번역되었습니다(베타). 프로젝트 공식 승인을 받지 않았습니다. 오류를 발견하셨나요? 문제 신고 →

구독 사용하기

팁

• 전체 스택 예시는 /examples/next-prisma-starter-websockets를 참조하세요.
• 최소한의 Node.js 예시는 /examples/standalone-server를 확인하세요.

구독 프로시저 추가하기

server/router.ts
tsx
import { EventEmitter } from 'events';
import * as trpc from '@trpc/server';
// create a global event emitter (could be replaced by redis, etc)
const ee = new EventEmitter();
export const appRouter = trpc
  .router()
  .subscription('onAdd', {
    resolve({ ctx }) {
      // `resolve()` is triggered for each client when they start subscribing `onAdd`
      // return a `Subscription` with a callback which is triggered immediately
      return new trpc.Subscription<Post>((emit) => {
        const onAdd = (data: Post) => {
          // emit data to client
          emit.data(data);
        };
        // trigger `onAdd()` when `add` is triggered in our event emitter
        ee.on('add', onAdd);
        // unsubscribe function when client disconnects or stops subscribing
        return () => {
          ee.off('add', onAdd);
        };
      });
    },
  })
  .mutation('add', {
    input: z.object({
      id: z.string().uuid().optional(),
      text: z.string().min(1),
    }),
    async resolve({ ctx, input }) {
      const post = { ...input }; /* [..] add to db */
      ee.emit('add', post);
      return post;
    },
  });

server/router.ts
tsx
import { EventEmitter } from 'events';
import * as trpc from '@trpc/server';
// create a global event emitter (could be replaced by redis, etc)
const ee = new EventEmitter();
export const appRouter = trpc
  .router()
  .subscription('onAdd', {
    resolve({ ctx }) {
      // `resolve()` is triggered for each client when they start subscribing `onAdd`
      // return a `Subscription` with a callback which is triggered immediately
      return new trpc.Subscription<Post>((emit) => {
        const onAdd = (data: Post) => {
          // emit data to client
          emit.data(data);
        };
        // trigger `onAdd()` when `add` is triggered in our event emitter
        ee.on('add', onAdd);
        // unsubscribe function when client disconnects or stops subscribing
        return () => {
          ee.off('add', onAdd);
        };
      });
    },
  })
  .mutation('add', {
    input: z.object({
      id: z.string().uuid().optional(),
      text: z.string().min(1),
    }),
    async resolve({ ctx, input }) {
      const post = { ...input }; /* [..] add to db */
      ee.emit('add', post);
      return post;
    },
  });

WebSocket 서버 생성

bash
yarn add ws

bash
yarn add ws

server/wsServer.ts
ts
import { applyWSSHandler } from '@trpc/server/adapters/ws';
import ws from 'ws';
import { appRouter } from './routers/app';
import { createContext } from './trpc';
const wss = new ws.Server({
  port: 3001,
});
const handler = applyWSSHandler({ wss, router: appRouter, createContext });
wss.on('connection', (ws) => {
  console.log(`➕➕ Connection (${wss.clients.size})`);
  ws.once('close', () => {
    console.log(`➖➖ Connection (${wss.clients.size})`);
  });
});
console.log('✅ WebSocket Server listening on ws://localhost:3001');
process.on('SIGTERM', () => {
  console.log('SIGTERM');
  handler.broadcastReconnectNotification();
  wss.close();
});

server/wsServer.ts
ts
import { applyWSSHandler } from '@trpc/server/adapters/ws';
import ws from 'ws';
import { appRouter } from './routers/app';
import { createContext } from './trpc';
const wss = new ws.Server({
  port: 3001,
});
const handler = applyWSSHandler({ wss, router: appRouter, createContext });
wss.on('connection', (ws) => {
  console.log(`➕➕ Connection (${wss.clients.size})`);
  ws.once('close', () => {
    console.log(`➖➖ Connection (${wss.clients.size})`);
  });
});
console.log('✅ WebSocket Server listening on ws://localhost:3001');
process.on('SIGTERM', () => {
  console.log('SIGTERM');
  handler.broadcastReconnectNotification();
  wss.close();
});

TRPCClient를 WebSockets로 설정하기

팁

링크 사용을 통해 쿼리 및/또는 뮤테이션은 HTTP 전송으로, 구독은 WebSockets를 통해 라우팅할 수 있습니다.

client.ts
tsx
import { httpBatchLink } from '@trpc/client/links/httpBatchLink';
import { createWSClient, wsLink } from '@trpc/client/links/wsLink';
// create persistent WebSocket connection
const wsClient = createWSClient({
  url: `ws://localhost:3001`,
});
// configure TRPCClient to use WebSockets transport
const client = createTRPCClient<AppRouter>({
  links: [
    wsLink({
      client: wsClient,
    }),
  ],
});

client.ts
tsx
import { httpBatchLink } from '@trpc/client/links/httpBatchLink';
import { createWSClient, wsLink } from '@trpc/client/links/wsLink';
// create persistent WebSocket connection
const wsClient = createWSClient({
  url: `ws://localhost:3001`,
});
// configure TRPCClient to use WebSockets transport
const client = createTRPCClient<AppRouter>({
  links: [
    wsLink({
      client: wsClient,
    }),
  ],
});

React 사용하기

참조: /examples/next-prisma-starter-websockets

WebSockets RPC 사양

자세한 내용은 TypeScript 정의를 확인하세요:

• /packages/server/src/unstable-core-do-not-import/rpc/envelopes.ts
• /packages/server/src/unstable-core-do-not-import/rpc/codes.ts.

query / mutation

요청

ts
{
  id: number | string;
  jsonrpc?: '2.0';
  method: 'query' | 'mutation';
  params: {
    path: string;
    input?: unknown; // <-- pass input of procedure, serialized by transformer
  };
}

ts
{
  id: number | string;
  jsonrpc?: '2.0';
  method: 'query' | 'mutation';
  params: {
    path: string;
    input?: unknown; // <-- pass input of procedure, serialized by transformer
  };
}

응답

... 또는 아래 오류 발생

ts
{
  id: number | string;
  jsonrpc: '2.0';
  result: {
    type: 'data'; // always 'data' for mutation / queries
    data: TOutput; // output from procedure
  }
}

ts
{
  id: number | string;
  jsonrpc: '2.0';
  result: {
    type: 'data'; // always 'data' for mutation / queries
    data: TOutput; // output from procedure
  }
}

subscription / subscription.stop

구독 시작

ts
{
  id: number | string;
  jsonrpc?: '2.0';
  method: 'subscription';
  params: {
    path: string;
    input?: unknown; // <-- pass input of procedure, serialized by transformer
  };
}

ts
{
  id: number | string;
  jsonrpc?: '2.0';
  method: 'subscription';
  params: {
    path: string;
    input?: unknown; // <-- pass input of procedure, serialized by transformer
  };
}

구독 취소를 위해 subscription.stop 호출

ts
{
  id: number | string; // <-- id of your created subscription
  jsonrpc?: '2.0';
  method: 'subscription.stop';
}

ts
{
  id: number | string; // <-- id of your created subscription
  jsonrpc?: '2.0';
  method: 'subscription.stop';
}

구독 응답 형태

... 또는 아래 오류 발생

ts
{
  id: number | string;
  jsonrpc: '2.0';
  result: (
    | {
      type: 'data';
        data: TData; // subscription emitted data
      }
    | {
        type: 'started'; // sub started
      }
    | {
        type: 'stopped'; // sub stopped
      }
  )
}

ts
{
  id: number | string;
  jsonrpc: '2.0';
  result: (
    | {
      type: 'data';
        data: TData; // subscription emitted data
      }
    | {
        type: 'started'; // sub started
      }
    | {
        type: 'stopped'; // sub stopped
      }
  )
}

오류

참조: https://www.jsonrpc.org/specification#error_object 또는 오류 형식

서버→클라이언트 알림

{id: null, type: 'reconnect' }

서버 종료 전에 클라이언트에게 재연결하라고 알립니다. wssHandler.broadcastReconnectNotification()에 의해 호출됩니다.