graphql subscsription 어떻게 굴러가는지 알아보기

저희 서비스의 "미발행 송장 카운트"에 graphql subscription을 적용하면서, 전반적인 subscription이 어떻게 동작하는지 흐름을 정리한 글입니다.

 

저희 회사는 graphql server를 apollo server + nexus 조합을 사용하고 있습니다. 

 

graphql spec문서에 보면 operation을 3개로 정의하고 있으며 주로 사용하는 query, mutation 이외에도 subscription가 존재합니다. 

subscription은 클라이언트와 서버 간에 장기적인 연결을 유지하며, 서버가 특정 이벤트나 데이터 변경을 감지할 때 클라이언트에 데이터를 푸시하는 방법입니다. 

 

이 글에는 저희 회사 스펙인 apollo server와 nexus를 사용하여 구현해보도록 하겠습니다.

 

---

Subscription 구현

import { RedisPubSub } from "graphql-redis-subscriptions";

const pubsub = new RedisPubSub({
  connection: "redis url",
});

const mutation = extendType({
  type: "Mutation",
  definition(t) {
    t.field("createPost", {
      type: "Post",
      args: {
        content: stringArg(),
      },
      resolve: async (parent, args, ctx, info) => {
        /**
         * DB 게시글 데이터 생성
         */

        await pubsub.publish("createPost", { --- (4)
          data: {
            content: args.content,
          },
        });

        return {
          author: args.author,
          content: args.content,
        };
      },
    });
  },
});

const subscription = subscriptionType({
  definition(t) {
    t.field("createPost", {
      type: "Post",
      subscribe: () => {
        return pubsub.asyncIterator("createPost"); --- (2)
      },
      async resolve(eventPromise) { --- (5)
        const event: any = await eventPromise;
        return event.data;
      },
    });
  },
});

mutation{
  createPost(    --- (3)
    content:"content",
  ) {
    content
  }
}

subscription{    --- (1)
  createPost {   --- (6)
    content
  }
}

 

전체적인 핵심 코드는 위와 같습니다. 서버에서 createPost subscription이 추가되었고, 프론트에서는 createPost를 호출하고 있습니다. 그리고 서버코드에 pubsub이라는 보이는데, pubsub은 이벤트의 중간 매개체 역할을 하게 됩니다.

 

먼저 프론트에서 createPost subscription을 호출하게 되면(1), subscribe 메서드의 pubsub.asyncIterator("createPost")가 실행됩니다(2). 여기서 일어나는 일은 redis의 createPost채널을 subscribe하게 됩니다. 이 부분이 핵심입니다.

 

즉 프론트에서 subscription를 실행한다는 것은 redis의 특정 채널을 subscribe하고 있다는 것이며, 이후에 mutation에서 publish하게 될 때, 실제로 redis의 특정 채널에 publish가 되어 해당 데이터를 읽어오는 것입니다.

 

이후 프론트에서 게시글 생성 요청(createPost mutation)을 호출하면(3) DB에 게시글 생성 후, 데이터와 함께 특정 채널(”createPost”)로 publish하게 됩니다(4). 이때 redis에 위의 채널로 데이터와 함께 publish가 됩니다.

 

앞서 프론트가 createPost subscription을 호출하면서, (2)번에서 redis의 createPost 채널에 팔로우 하고 있었습니다. 이제 해당 채널에 데이터가 들어왔으니, 가져와야 겠죠 ?

 

해당 데이터는 (5)의 resolve함수의 전달인자로 담아져 오게 됩니다. 이후 해당 데이터를 가공 후 return하게 되면 프론트의 6)번에서 받게 됩니다.

 

브라우저 네트워크 탭을 확인하면, createPost mutation을 호출할 때 마다 서버로부터 데이터를 전달받고 있습니다.

---

특정 클라이언트에게만 publish 하기

 

사실 서비스를 구현하다 보면, 모든 프론트에게 publish하는 것보다는 조건에 맞는 프론트에게만 publish하는 경우가 더 많을 것입니다. socket.io에서 room과 같은 기능인데요, graphql에서는 graphql-subscriptions의 withFilter를 사용할 수 있습니다.

 

import { withFilter } from "graphql-subscriptions";

const subscription = subscriptionType({
  definition(t) {
    t.field("createPost", {
      type: "Post",
      args: {
        userId: "String",
      },
      subscribe: withFilter(
        () => {
          return pubsub.asyncIterator("createPost");
        },
        (payload, variables) => {
          return true;
        }
      ),
      async resolve(eventPromise) {
        const event: any = await eventPromise;
        return event.data;
      },
    });
  },
});

 

mutation{
  createPost(    --- (3)
    content:"content",
  ) {
    content
  }
}

subscription{    --- (1)
  createPost(userId:'123') {   --- (6)
    content
  }
}

 

기존의 코드와 달라진 점은 아래와 같습니다.

• 프론트에서 createPost subscription 호출 시, 필터조건(userId)를 전달
• 서버의 createPost subscription 인자로 userId 추가
• subscriptionType의 subscribe 함수에 withFilter 적용

 

여기서 핵심은 withFilter 호출 시, 인자로 전달하는 2개의 함수입니다.

• 첫번째 함수 : 프론트에서 createPost subscription호출 시, redis의 createPost채널 subscribe(기존과 동일)
• 두번째 함수 : 프론트에게 publish 할 지 여부로 true는 프론트로 데이터 publish, false는 publish 하지 않음 여기서 payload는 createPost mutation에서 publish 메서드 호출 시 두번째 값이며, variables는 프론트에서 호출한 createPost subscription의 전달인자(userId) 입니다. 이 두가지 값을 사용하여 비지니스 로직에 맞게 특정 프론트에게만 데이터를 전달할 수 있습니다.

---

PubSub의 종류

apollo server 공식문서에는 예제 코드에 아래의 모듈을 사용하고 있지만, 실제 운영 환경에서 사용하는 것을 지양하고 있습니다. 왜냐하면 PubSub 모듈은 in-memory event publish system을 사용하고 있기 때문에 클러스터 모드나 여러 환경에서 서버를 올려 사용하는 경우 이벤트 간에 동기화가 되지 않습니다.

 

import { PubSub } from 'graphql-subscriptions';

const pubsub = new PubSub();

 

즉 실제로 모든 또는 특정 프론트에게 이벤트가 publish되야 하지만, 특정 서버와 연결된 프론트에게만 이벤트가 갈 수 있습니다.

따라서 실제 운영 환경에서는 메세지/이벤트 브로커를 사용하여 여러 서버간의 이벤트를 동기화할 필요가 있으며, 개발 환경에 맞게 라이브러리를 선택할 수 있습니다.