diff --git a/matterclient.go b/matterclient.go index b02ef35..af03717 100644 --- a/matterclient.go +++ b/matterclient.go @@ -12,6 +12,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "github.com/gorilla/websocket" @@ -82,7 +83,9 @@ type Client struct { lruCache *lru.Cache aliveChan chan bool loginCancel context.CancelFunc - lastPong time.Time + + lastWsActivity atomic.Int64 + connectedAt atomic.Int64 } var Matterircd bool @@ -569,7 +572,8 @@ func (m *Client) wsConnect() { m.WsClient.Listen() - m.lastPong = time.Now() + m.lastWsActivity.Store(time.Now().Unix()) + m.connectedAt.Store(time.Now().Unix()) m.logger.Debug("WsClient: connected") @@ -577,32 +581,43 @@ func (m *Client) wsConnect() { m.WsConnected = true } -func (m *Client) doCheckAlive() error { - if _, _, err := m.Client.GetPing(context.TODO()); err != nil { - return err +func (m *Client) doCheckAlive(ctx context.Context) error { + if m.WsClient != nil && m.WsClient.ListenError != nil { + return fmt.Errorf("websocket listen error: %w", m.WsClient.ListenError) } - if m.reconnectBusy { + connectedUnix := m.connectedAt.Load() + uptime := time.Since(time.Unix(connectedUnix, 0)).Round(time.Second) + lastActiveUnix := m.lastWsActivity.Load() + timeSinceActivity := time.Since(time.Unix(lastActiveUnix, 0)) + + if timeSinceActivity < 20*time.Second { + m.logger.Tracef("websocket is active (last event %v ago; up %s), skipping ping", timeSinceActivity.Round(time.Second), uptime) return nil } - if m.WsClient.ListenError == nil { - m.WsClient.SendMessage("ping", nil) - } else { - m.logger.Errorf("got a listen error: %#v", m.WsClient.ListenError) - - return m.WsClient.ListenError + if timeSinceActivity < 55*time.Second { + // Send a ping down the websocket to try to keep it active/alive + if m.WsClient != nil { + m.logger.Tracef("websocket has been quiet (last event %v ago; up %s), sending websocket ping", timeSinceActivity.Round(time.Second), uptime) + m.WsClient.SendMessage("ping", nil) + return nil + } } - if time.Since(m.lastPong) > 90*time.Second { - return errors.New("no pong received in 90 seconds") + m.logger.Tracef("websocket has been quiet (last event %v ago; up %s), falling back to HTTP GetPing", timeSinceActivity.Round(time.Second), uptime) + if _, _, err := m.Client.GetPing(ctx); err != nil { + m.logger.Warnf("fallback HTTP ping failed (up %s): %s", uptime, err) + return fmt.Errorf("fallback HTTP ping failed (up %s): %w", uptime, err) } + m.lastWsActivity.Store(time.Now().Unix()) + return nil } func (m *Client) checkAlive(ctx context.Context) { - ticker := time.NewTicker(time.Second * 45) + ticker := time.NewTicker(time.Second * 30) for { select { @@ -611,11 +626,25 @@ func (m *Client) checkAlive(ctx context.Context) { return case <-ticker.C: + var err error + // check if session still is valid - err := m.doCheckAlive() + for i := 0; i < 3; i++ { //nolint:intrange + err = m.doCheckAlive(ctx) + if err == nil { + break + } + + if i < 2 { + m.logger.Warnf("alive check failed, retrying %d/3: %s", i+1, err) + time.Sleep(time.Second * 2) + } + } + if err != nil { m.logger.Errorf("connection not alive: %s", err) m.aliveChan <- false + continue } m.aliveChan <- true @@ -632,7 +661,7 @@ func (m *Client) checkConnection(ctx context.Context) { if !alive { time.Sleep(time.Second * 10) - if m.doCheckAlive() != nil { + if m.doCheckAlive(ctx) != nil { m.Reconnect() } } @@ -663,6 +692,8 @@ func (m *Client) WsReceiver(ctx context.Context) { continue } + m.lastWsActivity.Store(time.Now().Unix()) + m.logger.Debugf("WsReceiver event: %#v", event) eventType := event.EventType() @@ -680,7 +711,13 @@ func (m *Client) WsReceiver(ctx context.Context) { m.parseMessage(msg) } - m.MessageChan <- msg + select { + case m.MessageChan <- msg: + // Message sent successfully + default: + // Message channel buffer full, drop/discard + m.logger.Errorf("CRITICAL: MessageChan is blocked! Downstream processor is hung. Dropping event: %s", event.EventType()) + } case response := <-m.WsClient.ResponseChannel: if response == nil || !response.IsValid() { continue @@ -688,11 +725,7 @@ func (m *Client) WsReceiver(ctx context.Context) { m.logger.Debugf("WsReceiver response: %#v", response) - if text, ok := response.Data["text"].(string); ok { - if text == "pong" { - m.lastPong = time.Now() - } - } + m.lastWsActivity.Store(time.Now().Unix()) m.parseResponse(response) case <-m.WsClient.PingTimeoutChannel: