Compare commits

...

3 Commits

View File

@@ -71,6 +71,7 @@ type SafeWebsocketClientBuilder struct {
rawQuery *string rawQuery *string
useTLS *bool useTLS *bool
channelSize *int64 channelSize *int64
authenticateFn func(*SafeWebsocketClient) error
} }
func NewSafeWebsocketClientBuilder() *SafeWebsocketClientBuilder { func NewSafeWebsocketClientBuilder() *SafeWebsocketClientBuilder {
@@ -92,6 +93,11 @@ func (b *SafeWebsocketClientBuilder) UseTLS(useTLS bool) *SafeWebsocketClientBui
return b return b
} }
func (b *SafeWebsocketClientBuilder) AuthenticateFn(authenticateFn func(*SafeWebsocketClient) error) *SafeWebsocketClientBuilder {
b.authenticateFn = authenticateFn
return b
}
func (b *SafeWebsocketClientBuilder) Path(path string) *SafeWebsocketClientBuilder { func (b *SafeWebsocketClientBuilder) Path(path string) *SafeWebsocketClientBuilder {
b.path = &path b.path = &path
return b return b
@@ -136,6 +142,10 @@ func (b *SafeWebsocketClientBuilder) Build(ctx context.Context) (*SafeWebsocketC
doneMap: NewSafeMap[string, chan struct{}](), doneMap: NewSafeMap[string, chan struct{}](),
} }
if b.authenticateFn != nil {
wsClient.authenticateFn = b.authenticateFn
}
go wsClient.reconnectHandler() go wsClient.reconnectHandler()
if err := wsClient.connect(); err != nil { if err := wsClient.connect(); err != nil {
@@ -160,6 +170,7 @@ type SafeWebsocketClient struct {
reconnectChans []chan struct{} reconnectChans []chan struct{}
isConnected bool isConnected bool
doneMap *SafeMap[string, chan struct{}] doneMap *SafeMap[string, chan struct{}]
authenticateFn func(*SafeWebsocketClient) error
} }
func (wsClient *SafeWebsocketClient) connect() error { func (wsClient *SafeWebsocketClient) connect() error {
@@ -200,15 +211,19 @@ func (wsClient *SafeWebsocketClient) connect() error {
return nil return nil
}) })
wsClient.mu.WriteHandler(func() error {
if wsClient.conn != nil { if wsClient.conn != nil {
wsClient.conn.Close() wsClient.conn.Close()
} }
wsClient.conn = conn wsClient.conn = conn
wsClient.isConnected = true wsClient.isConnected = true
if wsClient.authenticateFn != nil {
if err := wsClient.authenticateFn(wsClient); err != nil {
return err
}
}
wsClient.mu.WriteHandler(func() error {
pingCtx, pingCancel := context.WithCancel(context.Background()) pingCtx, pingCancel := context.WithCancel(context.Background())
go wsClient.startPingTicker(pingCtx) go wsClient.startPingTicker(pingCtx)
wsClient.cancelFuncs = append(wsClient.cancelFuncs, pingCancel) wsClient.cancelFuncs = append(wsClient.cancelFuncs, pingCancel)
@@ -253,31 +268,32 @@ func (wsClient *SafeWebsocketClient) startReceiveHandler(ctx context.Context) {
log.Println("receive handler stopped") log.Println("receive handler stopped")
return return
default: default:
if err := wsClient.mu.ReadHandler(func() error {
conn := wsClient.conn conn := wsClient.conn
if conn == nil { if conn == nil {
wsClient.triggerReconnect() wsClient.triggerReconnect()
return return fmt.Errorf("connection closed")
// return fmt.Errorf("connection closed")
} }
_, message, err := conn.ReadMessage() _, message, err := conn.ReadMessage()
if err != nil { if err != nil {
wsClient.triggerReconnect() wsClient.triggerReconnect()
// return fmt.Errorf("failed to read message: %v", err) return fmt.Errorf("failed to read message: %v", err)
return
} }
select { select {
case wsClient.dataChannel <- message: case wsClient.dataChannel <- message:
case <-ctx.Done(): case <-ctx.Done():
log.Println("Reconnect handler stopped") log.Println("Reconnect handler stopped")
return
default: default:
log.Println("Data channel full, dropping message") log.Println("Data channel full, dropping message")
} }
return nil
}); err != nil {
return return
} }
} }
} }
}
func (wsClient *SafeWebsocketClient) triggerReconnect() { func (wsClient *SafeWebsocketClient) triggerReconnect() {
select { select {