当websocket连接关闭时，关闭redis订阅并结束go例程

Question

I'm pushing events from a redis subscription to a client who is connected via websocket. I'm having trouble unsubscribing and exiting the redis go routine when the client disconnects the websocket.

Inspired by this post, here's what I have thus far. I'm able to receive subscription events and send messages to the client via websocket, but when the client closes the websocket and the defer close(done) code fires, my case b, ok := <-done: doesn't fire. It seems to be overloaded by the default case???

package api

import (
    ...

    "github.com/garyburd/redigo/redis"
    "github.com/gorilla/websocket"
)

func wsHandler(w http.ResponseWriter, r *http.Request) {
    var upgrader = websocket.Upgrader{
        ReadBufferSize:  1024,
        WriteBufferSize: 1024,
    }

    conn, err := upgrader.Upgrade(w, r, nil)
    if err != nil {
        HandleError(w, err)
        return
    }
    defer conn.Close()

    done := make(chan bool)
    defer close(done)

    for {
        var req WSRequest
        err := conn.ReadJSON(&req)
        if err != nil {
            HandleWSError(conn, err)
            return
        }
        defer conn.Close()

        go func(done chan bool, req *WSRequest, conn *websocket.Conn) {
            rc := redisPool.Get()
            defer rc.Close()

            psc := redis.PubSubConn{Conn: rc}
            if err := psc.PSubscribe(req.chanName); err != nil {
                HandleWSError(conn, err)
                return
            }
            defer psc.PUnsubscribe()

            for {
                select {
                case b, ok := <-done:
                    if !ok || b == true {
                        return
                    }
                default:
                    switch v := psc.Receive().(type) {
                    case redis.PMessage:
                        err := handler(conn, req, v)
                        if err != nil {
                            HandleWSError(conn, err)
                        }

                    case redis.Subscription:
                         log.Printf("%s: %s %d
", v.Channel, v.Kind, v.Count)

                    case error:
                         log.Printf("error in redis subscription; err:
%v
", v)
                         HandleWSError(conn, v)

                    default:
                        // do nothing...
                        log.Printf("unknown redis subscription event type; %s
", reflect.TypeOf(v))
                    }
                }
            }
        }(done, &req, conn)
    }
}

Answer 1

Make these changes to break out of the read loop when done serving the websocket connection:

• Maintain a slice of the Redis connections created for this websocket connection.

• Unsubscribe all connections when done.

• Modify the read loop to return when the subscription count is zero.

Here's the code:

func wsHandler(w http.ResponseWriter, r *http.Request) {
    var upgrader = websocket.Upgrader{
        ReadBufferSize:  1024,
        WriteBufferSize: 1024,
    }

    conn, err := upgrader.Upgrade(w, r, nil)
    if err != nil {
        HandleError(w, err)
        return
    }
    defer conn.Close()

    // Keep slice of all connections. Unsubscribe all connections on exit.
    var pscs []redis.PubSubConn
    defer func() {
        for _, psc := range rcs {
           psc.Unsubscribe() // unsubscribe with no args unsubs all channels
        }
    }()

    for {
        var req WSRequest
        err := conn.ReadJSON(&req)
        if err != nil {
            HandleWSError(conn, err)
            return
        }

        rc := redisPool.Get()
        psc := redis.PubSubConn{Conn: rc}
        pscs = append(pscs, psc)

        if err := psc.PSubscribe(req.chanName); err != nil {
            HandleWSError(conn, err)
            return
        }

        go func(req *WSRequest, conn *websocket.Conn) {
            defer rc.Close()
            for {
                switch v := psc.Receive().(type) {
                case redis.PMessage:
                    err := handler(conn, req, v)
                    if err != nil {
                        HandleWSError(conn, err)
                    }

                case redis.Subscription:
                     log.Printf("%s: %s %d
", v.Channel, v.Kind, v.Count)
                     if v.Count == 0 {
                         return
                     }

                case error:
                     log.Printf("error in redis subscription; err:
%v
", v)
                     HandleWSError(conn, v)

                default:
                    // do nothing...
                    log.Printf("unknown redis subscription event type; %s
", reflect.TypeOf(v))
                }
            }
        }(&req, conn)
    }
}

The code in the question and this answer dial multiple Redis connections for each websocket client. A more typical and scalable approach is to share a single Redis pubsub connection across multiple clients. The typical approach may be appropriate for your application given the high-level description, but I am still unsure of what you are trying to do given the code in the question.