mirror of
https://git.sr.ht/~ashkeel/strimertul
synced 2024-09-18 01:50:50 +00:00
fix: don't block local client on Wait call
This commit is contained in:
parent
0f93ac1c17
commit
3c107b9a32
2 changed files with 28 additions and 22 deletions
|
@ -5,6 +5,7 @@ import (
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/gorilla/websocket"
|
"github.com/gorilla/websocket"
|
||||||
|
|
||||||
jsoniter "github.com/json-iterator/go"
|
jsoniter "github.com/json-iterator/go"
|
||||||
"github.com/nicklaw5/helix/v2"
|
"github.com/nicklaw5/helix/v2"
|
||||||
"go.uber.org/zap"
|
"go.uber.org/zap"
|
||||||
|
@ -12,22 +13,36 @@ import (
|
||||||
|
|
||||||
const websocketEndpoint = "wss://eventsub-beta.wss.twitch.tv/ws"
|
const websocketEndpoint = "wss://eventsub-beta.wss.twitch.tv/ws"
|
||||||
|
|
||||||
func (c *Client) connectWebsocket(userClient *helix.Client) {
|
func (c *Client) eventSubLoop(userClient *helix.Client) {
|
||||||
connection, _, err := websocket.DefaultDialer.Dial(websocketEndpoint, nil)
|
endpoint := websocketEndpoint
|
||||||
|
var err error
|
||||||
|
for endpoint != "" {
|
||||||
|
endpoint, err = c.connectWebsocket(endpoint, userClient)
|
||||||
|
if err != nil {
|
||||||
|
c.logger.Error("eventsub ws read error", zap.Error(err))
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *Client) connectWebsocket(url string, userClient *helix.Client) (string, error) {
|
||||||
|
connection, _, err := websocket.DefaultDialer.Dial(url, nil)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
c.logger.Error("could not connect to eventsub ws", zap.Error(err))
|
c.logger.Error("could not connect to eventsub ws", zap.Error(err))
|
||||||
return
|
return "", err
|
||||||
}
|
}
|
||||||
defer connection.Close()
|
defer connection.Close()
|
||||||
|
|
||||||
received := make(chan []byte)
|
received := make(chan []byte, 10)
|
||||||
wsErr := make(chan error)
|
wsErr := make(chan error, 1)
|
||||||
readFromWS := func(connection *websocket.Conn, recv chan<- []byte) {
|
|
||||||
|
readFromWS := func(connection *websocket.Conn, recv chan<- []byte, wsErr chan<- error) {
|
||||||
for {
|
for {
|
||||||
messageType, messageData, err := connection.ReadMessage()
|
messageType, messageData, err := connection.ReadMessage()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
c.logger.Error("eventsub ws read error", zap.Error(err))
|
|
||||||
wsErr <- err
|
wsErr <- err
|
||||||
|
close(recv)
|
||||||
|
close(wsErr)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
if messageType != websocket.TextMessage {
|
if messageType != websocket.TextMessage {
|
||||||
|
@ -37,16 +52,16 @@ func (c *Client) connectWebsocket(userClient *helix.Client) {
|
||||||
recv <- messageData
|
recv <- messageData
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
go readFromWS(connection, received)
|
go readFromWS(connection, received, wsErr)
|
||||||
|
|
||||||
for {
|
for {
|
||||||
// Wait for next message or closing/error
|
// Wait for next message or closing/error
|
||||||
var messageData []byte
|
var messageData []byte
|
||||||
select {
|
select {
|
||||||
case <-c.ctx.Done():
|
case <-c.ctx.Done():
|
||||||
return
|
return "", nil
|
||||||
case <-wsErr:
|
case err = <-wsErr:
|
||||||
return
|
return "", err
|
||||||
case messageData = <-received:
|
case messageData = <-received:
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -83,16 +98,7 @@ func (c *Client) connectWebsocket(userClient *helix.Client) {
|
||||||
}
|
}
|
||||||
c.logger.Info("eventsub ws connection reset requested", zap.String("session-id", reconnectData.Session.Id), zap.String("reconnect-url", reconnectData.Session.ReconnectUrl))
|
c.logger.Info("eventsub ws connection reset requested", zap.String("session-id", reconnectData.Session.Id), zap.String("reconnect-url", reconnectData.Session.ReconnectUrl))
|
||||||
|
|
||||||
// Try reconnecting to the new URL
|
return reconnectData.Session.ReconnectUrl, nil
|
||||||
newConnection, _, err := websocket.DefaultDialer.Dial(reconnectData.Session.ReconnectUrl, nil)
|
|
||||||
if err != nil {
|
|
||||||
c.logger.Error("eventsub ws reconnect error", zap.Error(err))
|
|
||||||
break
|
|
||||||
} else {
|
|
||||||
_ = connection.Close()
|
|
||||||
go readFromWS(newConnection, received)
|
|
||||||
connection = newConnection
|
|
||||||
}
|
|
||||||
case "notification":
|
case "notification":
|
||||||
go c.processEvent(wsMessage)
|
go c.processEvent(wsMessage)
|
||||||
case "revocation":
|
case "revocation":
|
||||||
|
|
|
@ -216,7 +216,7 @@ func newClient(config Config, db *database.LocalDBClient, server *http.Server, l
|
||||||
client.logger.Error("no users found, please authenticate in Twitch configuration -> Events")
|
client.logger.Error("no users found, please authenticate in Twitch configuration -> Events")
|
||||||
} else {
|
} else {
|
||||||
client.User = users.Data.Users[0]
|
client.User = users.Data.Users[0]
|
||||||
go client.connectWebsocket(userClient)
|
go client.eventSubLoop(userClient)
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
client.logger.Warn("twitch user not identified, this will break most features")
|
client.logger.Warn("twitch user not identified, this will break most features")
|
||||||
|
|
Loading…
Reference in a new issue