how to ensure redis subscribers receive all messages from redis pubsub? #2014

brian-goo · 2022-02-01T07:52:20Z

brian-goo
Feb 1, 2022

I am trying to publish messages to dynamically generated channels in redis while subscribing all messages from the existing channels. The following seems to work, but it fails to receive some messages depending on the timing of requests from the client (browser).

package main

import (
	"context"
	"log"
	"net/http"
	"strings"

	"github.com/go-redis/redis/v8"
	"github.com/gorilla/websocket"
)

var upgrader = websocket.Upgrader{}
var rd = redis.NewClient(&redis.Options{
	Addr: "localhost:6379",
})
var ctx = context.Background()

func echo(w http.ResponseWriter, r *http.Request) {
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Println("websocket connection err:", err)
		return
	}
	defer conn.Close()

	room := make(chan string)
	start := make(chan string)

	go func() {
	loop:
		for {
			sub := rd.Subscribe(ctx)
			defer sub.Close()

       			channels := []string{}

			for {
				select {
                                 //it seems that messages are not received when executing this case sometimes
				case channel := <-room:
					log.Println("channel", channel)
					channels = append(channels, channel)
					sub = rd.Subscribe(ctx, channels...)
					start <- "ok"
				case msg := <-sub.Channel():
					log.Println("msg", msg)
					err := conn.WriteMessage(websocket.TextMessage, []byte(msg.Payload))
					if err != nil {
						log.Println("websocket write err:", err)
						break loop
					}
				}
			}

		}
	}()

	for {
		_, msg, err := conn.ReadMessage()
		if err != nil {
			log.Println("websocket read err:", err)
			break
		}
		log.Println(string(msg))

		chPrefix := strings.Split(string(msg), ":")[0]
		ch := chPrefix + "-channel"
		if string(msg) == "test" || string(msg) == "yeah" {
			room <- ch
			log.Println(ch)
			log.Println(<-start)
		}

		if err := rd.Publish(ctx, ch, msg).Err(); err != nil {
			log.Println("redis publish err:", err)
			break
		}
	}

}

func main() {
	http.Handle("/", http.FileServer(http.Dir("./js")))
	http.HandleFunc("/ws", echo)

	log.Println("server starting...", "http://localhost:5000")
	log.Fatal(http.ListenAndServe("localhost:5000", nil))
}

<html lang="en">
<body>
    <script>
        const ws = new WebSocket('ws://localhost:5000/ws')
        ws.onopen = e => {
            console.log('open')
            ws.send('test')
            setTimeout(() => {
                ws.send(`test:hello from clent ${new Date().getTime()}`)
            }, 200);

            setTimeout(() => {
                ws.send('yeah')
            }, 1000);
            setTimeout(() => {
                ws.send(`yeah:hello from clent ${new Date().getTime()}`)
            }, 1200);
        }
        ws.onclose = e => {
            console.log('close')
        }
        ws.onmessage = e => console.log(e.data)
        ws.onerror = e => console.log(e.data)
    </script>
</body>
</html>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

how to ensure redis subscribers receive all messages from redis pubsub? #2014

{{title}}

{{editor}}'s edit

{{editor}}'s edit

Replies: 0 comments

Select a reply

how to ensure redis subscribers receive all messages from redis pubsub? #2014

brian-goo Feb 1, 2022

Replies: 0 comments

brian-goo
Feb 1, 2022