From 239e0050a11a5c022e64ce89432737fd480ad83b Mon Sep 17 00:00:00 2001 From: Adrian Zuercher Date: Wed, 23 Jul 2025 08:56:43 +0200 Subject: [PATCH] change ping pong handler and fix blocking channel --- driver/artNet.go | 8 ++++- main.go | 1 + websocket/client.go | 75 ++++++++++++++------------------------------- 3 files changed, 31 insertions(+), 53 deletions(-) diff --git a/driver/artNet.go b/driver/artNet.go index fe0c8fd..f8498c4 100644 --- a/driver/artNet.go +++ b/driver/artNet.go @@ -102,15 +102,18 @@ func (d *ArtNetDriver) Connect(ip, id string, port uint) error { } client.OnMessage = func(data []byte) { - request := json_data.NewResponse() + err = json.Unmarshal(data, &request) + if err != nil { return } + if request.Subscribe != nil { d.Subscribe(request.Subscribe...) } + if request.Publish != nil { d.Publish(request.Publish...) } @@ -138,6 +141,9 @@ func (d *ArtNetDriver) Connect(ip, id string, port uint) error { // send data to all buses that the send flage is true func (d *ArtNetDriver) SendData() { for _, bus := range d.Buses { + if !bus.Reachable { + continue + } bus.Send <- bus.Data } } diff --git a/main.go b/main.go index 5209f88..c40bdcd 100644 --- a/main.go +++ b/main.go @@ -75,6 +75,7 @@ func main() { if err := artNetDriver.Connect(*serverIp, DriverName, *serverPort); err != nil { artNetDriver.Log.Error("main", err) } + artNetDriver.Log.Info("main", "next reconnecting attempt in 10 seconds") time.Sleep(10 * time.Second) } diff --git a/websocket/client.go b/websocket/client.go index bd86f2c..d1c3edc 100644 --- a/websocket/client.go +++ b/websocket/client.go @@ -16,9 +16,6 @@ const ( // Time allowed to read the next pong message from the peer. pongWait = 10 * time.Second - - // Send pings to peer with this period. Must be less than pongWait. - pingPeriod = (pongWait * 9) / 10 ) type Client struct { @@ -32,7 +29,6 @@ type Client struct { OnError func(err error) OnPing func() OnPong func() - sendPong chan string send chan []byte unregister chan []byte } @@ -45,9 +41,8 @@ func NewClient(ip, id string, port uint) (*Client, error) { ip: ip, port: port, Connected: true, - sendPong: make(chan string), - send: make(chan []byte), - unregister: make(chan []byte), + send: make(chan []byte, 256), + unregister: make(chan []byte, 256), } dialer := websocket.DefaultDialer @@ -60,18 +55,21 @@ func NewClient(ip, id string, port uint) (*Client, error) { } c.conn = conn - // Setup control handlers + //Setup control handlers conn.SetPingHandler(func(appData string) error { if c.OnPing != nil { c.OnPing() } - conn.SetWriteDeadline(time.Now().Add(writeWait)) - conn.SetReadDeadline(time.Now().Add(writeWait)) - c.sendPong <- appData + conn.SetReadDeadline(time.Now().Add(pongWait)) + conn.SetWriteDeadline(time.Now().Add(pongWait)) + if err := conn.WriteControl(websocket.PongMessage, []byte(appData), time.Now().Add(2*pongWait)); err != nil { + c.OnError(err) + return err + } return nil }) - conn.SetPongHandler(func(string) error { + conn.SetPongHandler(func(appData string) error { conn.SetReadDeadline(time.Now().Add(pongWait)) if c.OnPong != nil { c.OnPong() @@ -79,17 +77,6 @@ func NewClient(ip, id string, port uint) (*Client, error) { return nil }) - conn.SetCloseHandler(func(code int, text string) error { - if c.OnClose != nil { - c.OnClose(code, text) - } - return nil - }) - - if c.OnOpen != nil { - c.OnOpen() - } - // Start reading messages from client go c.Read() go c.Write() @@ -101,13 +88,16 @@ func (c *Client) Read() { c.OnOpen() } - c.conn.SetReadDeadline(time.Now().Add(writeWait)) + c.conn.SetReadDeadline(time.Now().Add(pongWait)) for c.Connected { + c.conn.SetReadDeadline(time.Now().Add(pongWait)) msgType, msg, err := c.conn.ReadMessage() + if err != nil { - c.handleError(fmt.Errorf("read error (id:%s:%d): %w", c.ip, c.port, err)) + c.handleError(fmt.Errorf("read error (ip:%s): %w", c.ip, err)) return } + switch msgType { case websocket.CloseMessage: c.Close(websocket.CloseNormalClosure, "Client closed") @@ -116,32 +106,29 @@ func (c *Client) Read() { if c.OnMessage != nil { c.OnMessage(msg) } else { - log.Printf("Received message but no handler set (id:%s:%d): %s", c.ip, c.port, string(msg)) + log.Printf("Received message but no handler set (ip:%s): %s", c.ip, string(msg)) } default: - log.Printf("Unhandled message type %d (id:%s:%d)", msgType, c.ip, c.port) + log.Printf("Unhandled message type %d (ip:%s)", msgType, c.ip) } } } func (c *Client) Write() { - ticker := time.NewTicker(pingPeriod) - defer func() { - ticker.Stop() - c.conn.Close() - }() - for { + defer c.conn.Close() + c.conn.SetWriteDeadline(time.Now().Add(writeWait)) + for { select { case message, ok := <-c.send: c.conn.SetWriteDeadline(time.Now().Add(writeWait)) if !ok { // The hub closed the channel. - if err := c.conn.WriteMessage(websocket.CloseMessage, []byte{}); err != nil { + if err := c.conn.WriteMessage(websocket.CloseMessage, []byte("ping")); err != nil { c.handleError(err) return } - c.handleError(fmt.Errorf("server %s:%d closed channel", c.ip, c.port)) + c.handleError(fmt.Errorf("server %s closed channel", c.ip)) return } else { if err := c.conn.WriteMessage(websocket.TextMessage, message); err != nil { @@ -149,26 +136,10 @@ func (c *Client) Write() { return } } - case <-ticker.C: - c.conn.SetWriteDeadline(time.Now().Add(writeWait)) - if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil { - c.handleError(err) - return - } - - if c.OnPing != nil { - c.OnPing() - } - - case message, ok := <-c.sendPong: - if ok { - c.conn.WriteMessage(websocket.PongMessage, []byte(message)) - } case message := <-c.unregister: c.conn.WriteMessage(websocket.CloseMessage, message) c.Connected = false close(c.send) - close(c.sendPong) close(c.unregister) return } @@ -217,7 +188,7 @@ func (c *Client) handleError(err error) { if c.OnError != nil { c.OnError(err) } else { - fmt.Println("error: ", err) + log.Println("error:", err) } } }