diff --git a/.vscode/launch.json b/.vscode/launch.json new file mode 100644 index 0000000..4e571f6 --- /dev/null +++ b/.vscode/launch.json @@ -0,0 +1,17 @@ +{ + // Use IntelliSense to learn about possible attributes. + // Hover to view descriptions of existing attributes. + // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387 + "version": "0.2.0", + "configurations": [ + { + "name": "Launch", + "type": "go", + "request": "launch", + "mode": "auto", + "program": "${fileDirname}", + "env": {}, + "args": [] + } + ] +} \ No newline at end of file diff --git a/serv.go b/serv.go index 9bab4a0..564bf7a 100644 --- a/serv.go +++ b/serv.go @@ -42,6 +42,8 @@ var ( ReadBufferSize: 1024, WriteBufferSize: 1024, } + hubMap = make(map[string]Hub) + run = true ) // Client is a middleman between the websocket connection and the hub. @@ -55,25 +57,8 @@ type Client struct { send chan []byte } -// readPump pumps messages from the websocket connection to the hub. -// -// The application runs readPump in a per-connection goroutine. The application -// ensures that there is at most one reader on a connection by executing all -// reads from this goroutine. -func (c *Client) readPump(topic string) { - defer func() { - c.hub.unregister <- c - c.conn.Close() - }() - c.conn.SetReadLimit(maxMessageSize) - c.conn.SetReadDeadline(time.Now().Add(pongWait)) - c.conn.SetPongHandler(func(string) error { c.conn.SetReadDeadline(time.Now().Add(pongWait)); return nil }) - - sigchan := make(chan os.Signal, 1) - signal.Notify(sigchan, syscall.SIGINT, syscall.SIGTERM) - - // group, err := uuid.Must(uuid.NewV4()) - group := uuid.NewV4() +func create(topic string) *kafka.Reader { + group, _ := uuid.NewV4() log.WithFields(log.Fields{ "topic": topic, @@ -90,11 +75,26 @@ func (c *Client) readPump(topic string) { }) reader.SetOffset(-1) //latest offset - log.WithFields(log.Fields{ - "reader": reader, - }).Info("Created ") + return reader +} - run := true +// readPump pumps messages from the websocket connection to the hub. +// +// The application runs readPump in a per-connection goroutine. The application +// ensures that there is at most one reader on a connection by executing all +// reads from this goroutine. +func (c *Client) readPump() { + defer func() { + c.hub.unregister <- c + c.conn.Close() + log.WithField("client", c).Warn("exit") + }() + c.conn.SetReadLimit(maxMessageSize) + c.conn.SetReadDeadline(time.Now().Add(pongWait)) + c.conn.SetPongHandler(func(string) error { c.conn.SetReadDeadline(time.Now().Add(pongWait)); return nil }) + + sigchan := make(chan os.Signal, 1) + signal.Notify(sigchan, syscall.SIGINT, syscall.SIGTERM) for run == true { select { @@ -102,15 +102,15 @@ func (c *Client) readPump(topic string) { log.WithFields(log.Fields{"signal": sig}).Info(" terminating") run = false log.Fatal("break ") - default: - for { - m, err := reader.ReadMessage(context.Background()) - if err != nil { - log.WithFields(log.Fields{"error": err}).Error("read error") - } - log.WithFields(log.Fields{"offset": m.Offset, "message": string(m.Value)}).Info("got message") - c.hub.broadcast <- m.Value - } + // default: + // for { + // m, err := c.hub.reader.ReadMessage(context.Background()) + // if err != nil { + // log.WithFields(log.Fields{"error": err}).Error("read error") + // } + // log.WithFields(log.Fields{"offset": m.Offset, "message": string(m.Value)}).Info("got message") + // c.hub.broadcast <- m.Value + // } } } @@ -163,7 +163,7 @@ func (c *Client) writePump() { } // serveWs handles websocket requests from the peer. -func serveWs(hub *Hub, w http.ResponseWriter, r *http.Request) { +func serveWs(w http.ResponseWriter, r *http.Request) { vars := mux.Vars(r) topic := vars["topic"] @@ -172,18 +172,29 @@ func serveWs(hub *Hub, w http.ResponseWriter, r *http.Request) { log.Error("upgrade connection failed ", err) return } - client := &Client{hub: hub, conn: conn, send: make(chan []byte, 256)} + h, ok := hubMap[topic] + if !ok { + log.WithField("topic", topic).Info("create new hub ") + h = *newHub(topic) + go h.run() + hubMap[topic] = h + } else { + log.Info("join hub") + } + client := &Client{hub: &h, conn: conn, send: make(chan []byte, 256)} client.hub.register <- client // Allow collection of memory referenced by the caller by doing all work in // new goroutines. go client.writePump() - go client.readPump(topic) + go client.readPump() } // Hub maintains the set of active clients and broadcasts messages to the // clients. type Hub struct { + reader *kafka.Reader + // Registered clients. clients map[*Client]bool @@ -197,17 +208,30 @@ type Hub struct { unregister chan *Client } -func newHub() *Hub { +func newHub(topic string) *Hub { return &Hub{ + reader: create(topic), broadcast: make(chan []byte), register: make(chan *Client), unregister: make(chan *Client), clients: make(map[*Client]bool), } } +func (h *Hub) read() { + log.WithField("reader", h.reader).Info("start to read") + for run == true { + m, err := h.reader.ReadMessage(context.Background()) + if err != nil { + log.WithFields(log.Fields{"error": err}).Error("read error") + } + log.WithFields(log.Fields{"offset": m.Offset, "message": string(m.Value)}).Info("got message") + h.broadcast <- m.Value + } +} func (h *Hub) run() { - for { + go h.read() + for run == true { select { case client := <-h.register: h.clients[client] = true @@ -242,17 +266,17 @@ func main() { Handler: loggedRouter, // Pass our instance of gorilla/mux in. } - hub := newHub() - go hub.run() + // hub := newHub() + // go hub.run() r.HandleFunc("/ws/{topic}", func(w http.ResponseWriter, r *http.Request) { - serveWs(hub, w, r) + serveWs(w, r) }) // Run our server in a goroutine so that it doesn't block. go func() { log.WithFields(log.Fields{"address": *addr}).Info("starting server") if err := srv.ListenAndServe(); err != nil { - log.WithFields(log.Fields{"error": err}).Error("startup error") + log.WithFields(log.Fields{"error": err}).Error("error") } }()