시스템 설계2 - 2장 (feat. Redis adapter, pub/sub 내부 파헤쳐보기)

 /**
   * Adapter constructor.
   *
   * @param nsp - the namespace
   * @param pubClient - a Redis client that will be used to publish messages
   * @param subClient - a Redis client that will be used to receive messages (put in subscribed state)
   * @param opts - additional options
   *
   * @public
   */
  constructor(
    nsp: any,
    readonly pubClient: any,
    readonly subClient: any,
    opts: Partial<RedisAdapterOptions> = {}
  ) {
    super(nsp);

    this.uid = uid2(6);
    this.requestsTimeout = opts.requestsTimeout || 5000;
    this.publishOnSpecificResponseChannel = !!opts.publishOnSpecificResponseChannel;

    const prefix = opts.key || "socket.io";

    this.channel = prefix + "#" + nsp.name + "#";
    this.requestChannel = prefix + "-request#" + this.nsp.name + "#";
    this.responseChannel = prefix + "-response#" + this.nsp.name + "#";
    const specificResponseChannel = this.responseChannel + this.uid + "#";

    const isRedisV4 = typeof this.pubClient.pSubscribe === "function";
    if (isRedisV4) {
      this.subClient.pSubscribe(
        this.channel + "*",
        (msg, channel) => {
          this.onmessage(null, channel, msg);
        },
        true
      );
      this.subClient.subscribe(
        [this.requestChannel, this.responseChannel, specificResponseChannel],
        (msg, channel) => {
          this.onrequest(channel, msg);
        }
      );
    }
    ...
  }

this.uid = uid2(6);

this.subClient.pSubscribe(
        this.channel + "*",
        (msg, channel) => {
          this.onmessage(null, channel, msg);
        },
        true
      );
this.subClient.subscribe(
[this.requestChannel, this.responseChannel, specificResponseChannel],
(msg, channel) => {
  this.onrequest(channel, msg);
}
);

typedef struct client {
    /* 1) "구독자 = 연결"을 보여주는 핵심 */
    uint64_t id;            /* Redis 내부 고유 client id */
    uint64_t flags;         /* CLIENT_* 플래그 (PUBSUB 모드 등) */
    connection *conn;       /* 실제 TCP/TLS/Unix 연결 핸들 */

    /* 2) 프로토콜/인증 같은 연결 상태 */
    int resp;               /* RESP2 or RESP3 */
    int authenticated;      /* 인증 여부(ACL/requirepass 등) */
    time_t lastinteraction; /* 마지막 상호작용 시간(타임아웃/idle 등) */

    /* 3) Pub/Sub: "이 client가 뭘 구독 중인지" */
    dict *pubsub_channels;       /* SUBSCRIBE한 채널 집합(=set) */
    dict *pubsub_patterns;       /* PSUBSCRIBE한 패턴 집합(=set) */
    dict *pubsubshard_channels;  /* SSUBSCRIBE(sharded pubsub) 집합(=set) */

    /* 4) 출력 버퍼 */
    list *reply;                 /* 응답(출력) 큐 */
    unsigned long long reply_bytes; /* reply 큐에 쌓인 바이트 */
    size_t sentlen;              /* 현재 reply에서 이미 보낸 길이 */

    /* 작은 응답은 고정 버퍼로도 나감*/
    char *buf;
    size_t bufpos;
    size_t buf_usable_size;

} client;

struct connection {
    ConnectionType *type;
    ConnectionState state;
    int fd;
    ...
};

static void connSocketAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
    int cport, cfd;
    int max = server.max_new_conns_per_cycle;
    char cip[NET_IP_STR_LEN];
    UNUSED(mask);
    UNUSED(privdata);

    while(max--) {
        cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport);
        ...
        acceptCommonHandler(connCreateAcceptedSocket(el,cfd,NULL), 0, cip);
    }
}

/* 3) Pub/Sub: "이 client가 뭘 구독 중인지" */
    dict *pubsub_channels;       /* SUBSCRIBE한 채널 집합(=set) */
    dict *pubsub_patterns;       /* PSUBSCRIBE한 패턴 집합(=set) */
    dict *pubsubshard_channels;  /* SSUBSCRIBE(sharded pubsub) 집합(=set) */

/* 4) 출력 버퍼 */
    list *reply;                 /* 응답(출력) 큐 */
    unsigned long long reply_bytes; /* reply 큐에 쌓인 바이트 */
    size_t sentlen;              /* 현재 reply에서 이미 보낸 길이 */

    /* 작은 응답은 고정 버퍼로도 나감*/
    char *buf;
    size_t bufpos;
    size_t buf_usable_size;

clientBufferLimitsConfig clientBufferLimitsDefaults[CLIENT_TYPE_OBUF_COUNT] = {
    {0, 0, 0}, /* normal */
    {1024*1024*256, 1024*1024*64, 60}, /* slave : hard 256MB / soft 64MB for 60s */
    {1024*1024*32, 1024*1024*8, 60}  /* pubsub : hard 32MB / soft 8MB for 60s */
};