[Go] go-socket.io의 Redis Adapter

socket.io에서는 redis adapter를 사용할 수 있다.

Adapter가 무엇일까?

https://socket.io/docs/v4/adapter/

정의

Adapter는 client의 부분집합 혹은 전체 client에게 event를 broadcast하는 server-side component이다.

왜 필요할까?

기본적으로 socket.io에서 in-memory adapter가 broadcast를 수행하는데,

socket.io 서버가 여러대로 분산되어있을 때, in-memory adapter를 이용하면 문제가 생긴다.

아래 그림처럼 두 개의 socket.io 서버가 있을 때, in-memory adapter를 이용하면

server A에서 server B에 연결되어있는 client들에게 메세지를 보내줄 수 없기 때문이다.

그래서 기본 in-memory adapter를 다른 adapter로 교체할 필요가 있다.

지원하는 Adapter의 종류

기본 in-memory adapter를 제외하고

js의 socket.io에서는 공식적으로 4종류의 adapter가 존재한다.

• Redis adapter
• MongoDB adapter
• Postgres adapter
• Cluster adapter

go-socket.io는 따로 정리되어있는 문서는 없지만,

github 소스코드를 봤을 때, redis-adapter만 존재한다. (22년 3월 기준)

 

Redis Adapter

https://socket.io/docs/v4/redis-adapter/

Redis Adapter는 Redis의 Pub/Sub 매커니즘에 의존한다.

 

Redis Pub/Sub 매커니즘에 대해 잘 모른다면 먼저 공부하고 오는 것을 추천한다!

 

redis adapter에 연결된 여러 socket.io서버 중 한 서버에서 emit을 하면,

1. 현재 서버에 연결된 클라이언트로 전송되는 것은 당연하고
2. 추가적으로 Redis 채널에 메세지가 publish되고, 클러스터의 다른 socket.io 서버에서 이를 수신해 해당 서버에 연결된 클라이언트에게도 전송된다.

여기까지가 문서의 설명이다.

문서에서는 socket.io 서버가 여러 대로 분산되어 있을 때 사용한다는 내용만 적혀있다.

 

하지만 이것을 응용하면, socket.io서버가 아닌 일반적인 서버를 socket.io 서버에 대한 trigger처럼 사용할 수도 있다. 

 

특정 서버에서 Redis에 특정 형식의 메세지를 publish했을 때,
동일한 Redis에 연결된 socket.io 서버의 Redis Adapter를 통해 emit하게 만들 수 있다는 것이다!

 

 

이 과정을 구현하기 위해서는,

Redis의 어떤 채널에 어떤 형식의 메세지를 publish 해야 socket.io의 redis adapter가 알아챌 수 있는지 알아야 한다.

 

js의 경우에는 가이드라인을 제공하는데, (js의 redis adapter github)

go-socket.io는 가이드라인을 제공하지 않아서 직접 찾아야 했다.

 

삽질 🛠️

postman으로 socket을 연결해 놓고,

redis-cli를 이용해, 터미널에서 psubscribe socket.io* 명령어를 이용해 구독하게 한 뒤

서버에서 broadcast와 emit을 하면서 터미널에 어떻게 로그가 찍히는지 확인해보았다.

 

socket.io*를 구독한 이유는, redis adapter 설정에 prefix의 기본값이 socket.io였기 때문이다. redis 채널 이름이 redis adapter의 prefix로 시작한다는 것을 여기에서 확인할 수 있었다.

 

go로 작성된 socket.io 서버에서

server.BroadcastToNamespace("/test", "hi", "hello")의 결과로 아래와 같은 로그가 찍혔다.

1) "pmessage"
2) "socket.io*"
3) "socket.io#/test#c3e57b2d-2400-45d9-9cda-6db7817929df"
4) "{\"args\":[\"hello\"],\"opts\":[\"\",\"hi\"]}"

채널 : socket.io#/test#${uid}

메세지: {\"args\":[\"hello\"],\"opts\":[\"\",\"hi\"]}

 

이렇게 몇 차례 로그를 찍어본 결과,

prefix#namespace# 채널에 아래와 같은 형식의 메세지를 publish해야 한다는 것을 알게되었다.

{
    args: [payload...],
    opts: ["roomId", "event name"],
}

• args: 말 그대로 내용(payload)
• opts는 2칸짜리 배열인데,
  • opts[0] : roomId (roomId가 필요 없는 경우에는 “” 이렇게 빈 값으로 작성)
  • opts[1] : event name 으로 작성하면 된다!

 

이렇게 publish하니 매우 잘 작동했다 :)

코드 분석

하지만.. 뭔가 주먹구구식으로 찾은 것 같은 느낌이 들어, go-socket.io코드를 분석해보았다.

그 전에도 코드 분석을 시도했으나 명확하게 보이지 않았는데, 답을 알고나서 코드를 보니까 더 잘 읽혔던 것 같다.

그리고 redis adapter관련 코드가 한 파일에 들어있어서 그리 복잡하지 않게 볼 수 있었다.

 

코드 링크

 

1. 채널 이름

아래 코드에서 key field를 보면

prefix#nsp(namespace)#uid 라고 되어있는데, 위에서 확인한 redis 로그의 형식과 일치한다.

socket.io#/test#c3e57b2d-2400-45d9-9cda-6db7817929df

func newRedisBroadcast(nsp string, opts *RedisAdapterOptions) (*redisBroadcast, error) {
    ...
    rbc := &redisBroadcast{
        rooms:      make(map[string]map[string]Conn),
        requests:   make(map[string]interface{}),
        sub:        subConn,
        pub:        pubConn,
        key:        fmt.Sprintf("%s#%s#%s", opts.Prefix, nsp, uid),
        reqChannel: fmt.Sprintf("%s-request#%s", opts.Prefix, nsp),
        resChannel: fmt.Sprintf("%s-response#%s", opts.Prefix, nsp),
        nsp:        nsp,
        uid:        uid,
    }

    ...
}

 

 

다만, 이건 socket.io 서버에서 broadcast했을 때 생기는 redis message이고,

그냥 redis에 바로 publish할 때는 uid 부분을 생략해도 되었다.

그리고 defalut namespace (/)는 namespace 자리에 /를 적는 것이 아니라 아예 생략해야 했다.

 

결론 ⭐️

따라서, 내가 만든 프로젝트에서는

• prefix : socket.io
• namespace : defalut (/)

였기 때문에 socket.io## 채널에 publish를 해야 했다.

 

js는 여기에 채널 네이밍 관련 내용이 나와있는데,

이에 따르면 socket.io#namespace#roomId 로 publish 해야하는데, go에서는 채널 이름에 roomId를 적는 방법이 먹히지 않았다. (설계가 다르게 되어있나보다..)

go-socket.io에서는 위에서 말한 것 처럼 opts[0]에 roomId를 적어줘야 한다.

 

2. 메세지 형식

아래 코드는 redis adapter가 redis에서 subscribe를 통해 받은 메세지를 처리하는 메소드이다.

func (bc *redisBroadcast) onMessage(channel string, msg []byte) error {
    channelParts := strings.Split(channel, "#")
    nsp := channelParts[len(channelParts)-2]
    if bc.nsp != nsp {
        return nil
    }

    uid := channelParts[len(channelParts)-1]
    if bc.uid == uid {
        return nil
    }

    var bcMessage map[string][]interface{}
    err := json.Unmarshal(msg, &bcMessage)
    if err != nil {
        return errors.New("invalid broadcast message")
    }

    args := bcMessage["args"]
    opts := bcMessage["opts"]

    room, ok := opts[0].(string)
    if !ok {
        return errors.New("invalid room")
    }

    event, ok := opts[1].(string)
    if !ok {
        return errors.New("invalid event")
    }

    if room != "" {
        bc.send(room, event, args...)
    } else {
        bc.sendAll(event, args...)
    }

    return nil
}

코드 link

 

onMessage()에서 msg를 처리할 때,

args, opts 필드를 사용하는 것을 확인할 수 있고, 위에서 redis 로그를 통해서 유추한 것과 정확히 일치했다.

opts[0]번 room에 opts[1] event로 args를 broadcast

• room단위가 아니라 broadcast하고 싶을 때에도 opts 길이는 2여야함 (roomid를 빈 값(””)으로 넣어 보내기)
  • opts: [””, “event name”]

socket.io 문서에 따르면,

adapter를 통해서 단순 메세지 emit 뿐 아니라

join room이나, leave room 등의 기능도 가능하다고 하는데 어떻게 하는지 아직 모른다!

 

다만 redis_broadcast.go의 dispatch() 에서 onMessage 말고도, onResponse, onRequest도 수행하고 있기 때문에 이쪽 코드를 더 자세하게 보면 알 수 있을 것 같다. 

func (bc *redisBroadcast) dispatch() {
    for {
        switch m := bc.sub.Receive().(type) {
        case redis.Message:
            if m.Channel == bc.reqChannel {
                bc.onRequest(m.Data)
                break
            } else if m.Channel == bc.resChannel {
                bc.onResponse(m.Data)
                break
            }

            err := bc.onMessage(m.Channel, m.Data)
            if err != nil {
                return
            }

        case redis.Subscription:
            if m.Count == 0 {
                return
            }

        case error:
            return
        }
    }
}

 

추가

go-socket.io에서 쓰는 redis package : redigo

내가 개발한 socket server에서 쓰는 redis package : go-redis

• redigo vs redis

https://go.libhunt.com/compare-redigo-vs-redis