Compare commits

...

2 Commits

Author SHA1 Message Date
967b8a98b3 fix: write & reader posiiton 2025-09-30 07:12:38 +07:00
d09d389011 fix: initialize write channel 2025-09-30 06:59:01 +07:00

View File

@@ -156,6 +156,7 @@ func (b *SafeWebsocketClientBuilder) Build(ctx context.Context) (*SafeWebsocketC
reconnectCh: make(chan struct{}, 1),
isConnected: false,
doneMap: NewSafeMap[string, chan struct{}](),
writeChan: make(chan Message),
}
if b.authenticateFn != nil {
@@ -258,6 +259,11 @@ func (wsClient *SafeWebsocketClient) connect() error {
wsClient.cancelFuncs = append(wsClient.cancelFuncs, cancel)
return nil
})
writer, err := conn.NextWriter(websocket.TextMessage)
if err != nil {
return
}
for {
select {
case <-ctx.Done():
@@ -268,19 +274,6 @@ func (wsClient *SafeWebsocketClient) connect() error {
wsClient.triggerReconnect()
return
}
messageType := websocket.TextMessage
switch data.MessageType {
case MessageTypePing:
messageType = websocket.PingMessage
case MessageTypePong:
messageType = websocket.PongMessage
case MessageTypeClose:
messageType = websocket.CloseMessage
}
writer, err := conn.NextWriter(messageType)
if err != nil {
return
}
if _, err := writer.Write(data.Data); err != nil {
return
@@ -295,6 +288,10 @@ func (wsClient *SafeWebsocketClient) connect() error {
wsClient.cancelFuncs = append(wsClient.cancelFuncs, cancel)
return nil
})
_, reader, err := conn.NextReader()
if err != nil {
return
}
for {
select {
case <-ctx.Done():
@@ -308,12 +305,6 @@ func (wsClient *SafeWebsocketClient) connect() error {
fmt.Printf("error on read deadline: %v\n", err)
return
}
_, reader, err := conn.NextReader()
if err != nil {
fmt.Printf("Next Reader Closed: %v\n", err)
wsClient.triggerReconnect()
return
}
readerBytes, err := io.ReadAll(reader)
if err != nil {
@@ -404,11 +395,15 @@ func (wsClient *SafeWebsocketClient) reconnectHandler() {
case <-wsClient.reconnectCh:
log.Println("Reconnect triggered")
if wsClient.cancelFuncs != nil {
for _, cancel := range wsClient.cancelFuncs {
cancel()
wsClient.mu.ReadHandler(func() error {
if wsClient.cancelFuncs != nil {
for _, cancel := range wsClient.cancelFuncs {
cancel()
}
}
}
return nil
})
wsClient.isConnected = false
time.Sleep(100 * time.Millisecond)
@@ -470,11 +465,14 @@ func (wsClient *SafeWebsocketClient) Write(data []byte) error {
}
func (wsClient *SafeWebsocketClient) Close() error {
if wsClient.cancelFuncs != nil {
for _, cancel := range wsClient.cancelFuncs {
cancel()
wsClient.mu.ReadHandler(func() error {
if wsClient.cancelFuncs != nil {
for _, cancel := range wsClient.cancelFuncs {
cancel()
}
}
}
return nil
})
if wsClient.reconnectChans != nil {
for _, reconnectChan := range wsClient.reconnectChans {