mirror of
https://github.com/AlexxIT/go2rtc.git
synced 2025-09-27 12:42:18 +08:00
feat(nest): add retry logic for 429 and 409 errors with exponential backoff
This commit is contained in:
@@ -166,6 +166,11 @@ func (a *API) ExchangeSDP(projectID, deviceID, offer string) (string, error) {
|
|||||||
|
|
||||||
uri := "https://smartdevicemanagement.googleapis.com/v1/enterprises/" +
|
uri := "https://smartdevicemanagement.googleapis.com/v1/enterprises/" +
|
||||||
projectID + "/devices/" + deviceID + ":executeCommand"
|
projectID + "/devices/" + deviceID + ":executeCommand"
|
||||||
|
|
||||||
|
maxRetries := 3
|
||||||
|
retryDelay := time.Second * 30
|
||||||
|
|
||||||
|
for attempt := 0; attempt < maxRetries; attempt++ {
|
||||||
req, err := http.NewRequest("POST", uri, bytes.NewReader(b))
|
req, err := http.NewRequest("POST", uri, bytes.NewReader(b))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
@@ -178,6 +183,27 @@ func (a *API) ExchangeSDP(projectID, deviceID, offer string) (string, error) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Handle 409 (Conflict) and 429 (Too Many Requests)
|
||||||
|
if res.StatusCode == 409 || res.StatusCode == 429 {
|
||||||
|
res.Body.Close()
|
||||||
|
if attempt < maxRetries-1 {
|
||||||
|
log.Info().
|
||||||
|
Int("status", res.StatusCode).
|
||||||
|
Float64("delay", retryDelay.Seconds()).
|
||||||
|
Int("attempt", attempt+1).
|
||||||
|
Int("max_retries", maxRetries-1).
|
||||||
|
Msg("API request failed, retrying")
|
||||||
|
// Get new token from Google
|
||||||
|
if err := a.refreshToken(); err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
time.Sleep(retryDelay)
|
||||||
|
retryDelay *= 2 // exponential backoff
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
defer res.Body.Close()
|
defer res.Body.Close()
|
||||||
|
|
||||||
if res.StatusCode != 200 {
|
if res.StatusCode != 200 {
|
||||||
@@ -204,6 +230,44 @@ func (a *API) ExchangeSDP(projectID, deviceID, offer string) (string, error) {
|
|||||||
return resv.Results.Answer, nil
|
return resv.Results.Answer, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
return "", errors.New("nest: max retries exceeded")
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a *API) refreshToken() error {
|
||||||
|
// Get the cached API with matching token to get credentials
|
||||||
|
var refreshKey string
|
||||||
|
cacheMu.Lock()
|
||||||
|
for key, api := range cache {
|
||||||
|
if api.Token == a.Token {
|
||||||
|
refreshKey = key
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
cacheMu.Unlock()
|
||||||
|
|
||||||
|
if refreshKey == "" {
|
||||||
|
return errors.New("nest: unable to find cached credentials")
|
||||||
|
}
|
||||||
|
|
||||||
|
// Parse credentials from cache key
|
||||||
|
parts := strings.Split(refreshKey, ":")
|
||||||
|
if len(parts) != 3 {
|
||||||
|
return errors.New("nest: invalid cache key format")
|
||||||
|
}
|
||||||
|
clientID, clientSecret, refreshToken := parts[0], parts[1], parts[2]
|
||||||
|
|
||||||
|
// Get new API instance which will refresh the token
|
||||||
|
newAPI, err := NewAPI(clientID, clientSecret, refreshToken)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// Update current API with new token
|
||||||
|
a.Token = newAPI.Token
|
||||||
|
a.ExpiresAt = newAPI.ExpiresAt
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
func (a *API) ExtendStream() error {
|
func (a *API) ExtendStream() error {
|
||||||
var reqv struct {
|
var reqv struct {
|
||||||
Command string `json:"command"`
|
Command string `json:"command"`
|
||||||
|
@@ -4,13 +4,17 @@ import (
|
|||||||
"errors"
|
"errors"
|
||||||
"net/url"
|
"net/url"
|
||||||
"strings"
|
"strings"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/AlexxIT/go2rtc/internal/app"
|
||||||
"github.com/AlexxIT/go2rtc/pkg/core"
|
"github.com/AlexxIT/go2rtc/pkg/core"
|
||||||
"github.com/AlexxIT/go2rtc/pkg/rtsp"
|
"github.com/AlexxIT/go2rtc/pkg/rtsp"
|
||||||
"github.com/AlexxIT/go2rtc/pkg/webrtc"
|
"github.com/AlexxIT/go2rtc/pkg/webrtc"
|
||||||
pion "github.com/pion/webrtc/v3"
|
pion "github.com/pion/webrtc/v3"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
var log = app.GetLogger("nest")
|
||||||
|
|
||||||
type WebRTCClient struct {
|
type WebRTCClient struct {
|
||||||
conn *webrtc.Conn
|
conn *webrtc.Conn
|
||||||
api *API
|
api *API
|
||||||
@@ -38,9 +42,32 @@ func Dial(rawURL string) (core.Producer, error) {
|
|||||||
return nil, errors.New("nest: wrong query")
|
return nil, errors.New("nest: wrong query")
|
||||||
}
|
}
|
||||||
|
|
||||||
nestAPI, err := NewAPI(cliendID, cliendSecret, refreshToken)
|
maxRetries := 3
|
||||||
if err != nil {
|
retryDelay := time.Second * 30
|
||||||
return nil, err
|
|
||||||
|
var nestAPI *API
|
||||||
|
var lastErr error
|
||||||
|
|
||||||
|
for attempt := 0; attempt < maxRetries; attempt++ {
|
||||||
|
nestAPI, err = NewAPI(cliendID, cliendSecret, refreshToken)
|
||||||
|
if err == nil {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
lastErr = err
|
||||||
|
if attempt < maxRetries-1 {
|
||||||
|
log.Info().
|
||||||
|
Float64("delay", retryDelay.Seconds()).
|
||||||
|
Int("attempt", attempt+1).
|
||||||
|
Int("max_retries", maxRetries-1).
|
||||||
|
Err(err).
|
||||||
|
Msg("API initialization failed, retrying")
|
||||||
|
time.Sleep(retryDelay)
|
||||||
|
retryDelay *= 2 // exponential backoff
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if nestAPI == nil {
|
||||||
|
return nil, lastErr
|
||||||
}
|
}
|
||||||
|
|
||||||
protocols := strings.Split(query.Get("protocols"), ",")
|
protocols := strings.Split(query.Get("protocols"), ",")
|
||||||
@@ -79,6 +106,11 @@ func (c *WebRTCClient) MarshalJSON() ([]byte, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func rtcConn(nestAPI *API, rawURL, projectID, deviceID string) (*WebRTCClient, error) {
|
func rtcConn(nestAPI *API, rawURL, projectID, deviceID string) (*WebRTCClient, error) {
|
||||||
|
maxRetries := 3
|
||||||
|
retryDelay := time.Second * 30
|
||||||
|
var lastErr error
|
||||||
|
|
||||||
|
for attempt := 0; attempt < maxRetries; attempt++ {
|
||||||
rtcAPI, err := webrtc.NewAPI()
|
rtcAPI, err := webrtc.NewAPI()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
@@ -112,6 +144,18 @@ func rtcConn(nestAPI *API, rawURL, projectID, deviceID string) (*WebRTCClient, e
|
|||||||
// 4. Exchange SDP via Hass
|
// 4. Exchange SDP via Hass
|
||||||
answer, err := nestAPI.ExchangeSDP(projectID, deviceID, offer)
|
answer, err := nestAPI.ExchangeSDP(projectID, deviceID, offer)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
lastErr = err
|
||||||
|
if attempt < maxRetries-1 {
|
||||||
|
log.Info().
|
||||||
|
Float64("delay", retryDelay.Seconds()).
|
||||||
|
Int("attempt", attempt+1).
|
||||||
|
Int("max_retries", maxRetries-1).
|
||||||
|
Err(err).
|
||||||
|
Msg("WebRTC connection setup failed, retrying")
|
||||||
|
time.Sleep(retryDelay)
|
||||||
|
retryDelay *= 2
|
||||||
|
continue
|
||||||
|
}
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -123,6 +167,9 @@ func rtcConn(nestAPI *API, rawURL, projectID, deviceID string) (*WebRTCClient, e
|
|||||||
return &WebRTCClient{conn: conn, api: nestAPI}, nil
|
return &WebRTCClient{conn: conn, api: nestAPI}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
return nil, lastErr
|
||||||
|
}
|
||||||
|
|
||||||
func rtspConn(nestAPI *API, rawURL, projectID, deviceID string) (*RTSPClient, error) {
|
func rtspConn(nestAPI *API, rawURL, projectID, deviceID string) (*RTSPClient, error) {
|
||||||
rtspURL, err := nestAPI.GenerateRtspStream(projectID, deviceID)
|
rtspURL, err := nestAPI.GenerateRtspStream(projectID, deviceID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
Reference in New Issue
Block a user