diff --git a/matterclient.go b/matterclient.go index 60100ee..ef8a750 100644 --- a/matterclient.go +++ b/matterclient.go @@ -11,6 +11,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "github.com/gorilla/websocket" @@ -78,7 +79,8 @@ type Client struct { lruCache *lru.Cache aliveChan chan bool loginCancel context.CancelFunc - lastPong time.Time + + lastWsActivity atomic.Int64 } var Matterircd bool @@ -526,7 +528,7 @@ func (m *Client) wsConnect() { m.WsClient.Listen() - m.lastPong = time.Now() + m.lastWsActivity.Store(time.Now().Unix()) m.logger.Debug("WsClient: connected") @@ -535,25 +537,25 @@ func (m *Client) wsConnect() { } func (m *Client) doCheckAlive() error { - if _, _, err := m.Client.GetPing(context.TODO()); err != nil { - return err + if m.WsClient != nil && m.WsClient.ListenError != nil { + return fmt.Errorf("websocket listen error: %w", m.WsClient.ListenError) } - if m.reconnectBusy { + lastActiveUnix := m.lastWsActivity.Load() + timeSinceActivity := time.Since(time.Unix(lastActiveUnix, 0)) + + if timeSinceActivity < 90*time.Second { + m.logger.Debugf("websocket is active (last event %v ago), skipping HTTP ping", timeSinceActivity.Round(time.Second)) return nil } - if m.WsClient.ListenError == nil { - m.WsClient.SendMessage("ping", nil) - } else { - m.logger.Errorf("got a listen error: %#v", m.WsClient.ListenError) + m.logger.Debug("websocket has been quiet, falling back to HTTP GetPing") - return m.WsClient.ListenError + if _, _, err := m.Client.GetPing(context.TODO()); err != nil { + return fmt.Errorf("fallback HTTP ping failed: %w", err) } - if time.Since(m.lastPong) > 90*time.Second { - return errors.New("no pong received in 90 seconds") - } + m.lastWsActivity.Store(time.Now().Unix()) return nil } @@ -568,11 +570,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++ { + err = m.doCheckAlive() + 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 @@ -620,6 +636,8 @@ func (m *Client) WsReceiver(ctx context.Context) { continue } + m.lastWsActivity.Store(time.Now().Unix()) + m.logger.Debugf("WsReceiver event: %#v", event) msg := &Message{ @@ -631,19 +649,20 @@ func (m *Client) WsReceiver(ctx context.Context) { m.parseMessage(msg) } - m.MessageChan <- msg + select { + case m.MessageChan <- msg: + // Message sent successfully + case <-time.After(15*time.Second): + 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 } - m.logger.Debugf("WsReceiver response: %#v", response) + m.lastWsActivity.Store(time.Now().Unix()) - if text, ok := response.Data["text"].(string); ok { - if text == "pong" { - m.lastPong = time.Now() - } - } + m.logger.Debugf("WsReceiver response: %#v", response) m.parseResponse(response) case <-m.WsClient.PingTimeoutChannel: