Compare commits

..

8 Commits

3 changed files with 33 additions and 28 deletions

2
go.mod
View File

@@ -4,4 +4,4 @@ go 1.24.5
require github.com/gorilla/websocket v1.5.3
require git.neurocipta.com/rogerferdinan/custom-rwmutex v1.0.0 // indirect
require git.neurocipta.com/rogerferdinan/custom-rwmutex v1.0.0

View File

@@ -23,9 +23,9 @@ type Client struct {
func NewClient(conn *websocket.Conn, subscribedPath string) *Client {
return &Client{
Conn: conn,
Send: make(chan []byte, 64),
Send: make(chan []byte, 2),
SubscribedPath: subscribedPath,
done: make(chan struct{}),
done: make(chan struct{}, 1),
}
}
@@ -38,10 +38,10 @@ type Hub struct {
func NewHub() *Hub {
return &Hub{
Broadcast: make(chan []byte),
Register: make(chan *Client),
Unregister: make(chan *Client),
Clients: make(map[*Client]bool),
Broadcast: make(chan []byte, 1),
Register: make(chan *Client, 1),
Unregister: make(chan *Client, 1),
Clients: make(map[*Client]bool, 1),
}
}
@@ -117,7 +117,7 @@ func ReadPump(c *Client, h *Hub) {
for {
messageType, message, err := c.Conn.ReadMessage()
if err != nil {
if !websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
log.Printf("WebSocket error: %v", err)
}
break

View File

@@ -68,10 +68,10 @@ func (sm *SafeMap[K, V]) Len() int {
type MessageType uint
const (
MessageTypeText MessageType = iota + 1
MessageTypePing MessageType = iota
MessageTypePong MessageType = iota
MessageTypeClose MessageType = iota
MessageTypeText MessageType = websocket.TextMessage
MessageTypePing MessageType = websocket.PingMessage
MessageTypePong MessageType = websocket.PongMessage
MessageTypeClose MessageType = websocket.CloseMessage
)
type Message struct {
@@ -155,7 +155,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),
writeChan: make(chan Message, 1),
}
if b.authenticateFn != nil {
@@ -193,6 +193,7 @@ type SafeWebsocketClient struct {
authenticateFn func(*SafeWebsocketClient) error
writeChan chan Message
pongChan chan error
}
func (wsClient *SafeWebsocketClient) connect() error {
@@ -220,19 +221,6 @@ func (wsClient *SafeWebsocketClient) connect() error {
return fmt.Errorf("failed to connect to %s: %w", wsClient.baseHost, err)
}
conn.SetPingHandler(func(pingData string) error {
if err := conn.WriteMessage(websocket.PongMessage, []byte(pingData)); err != nil {
if err == websocket.ErrCloseSent {
return nil
}
if netErr, ok := err.(interface{ Timeout() bool }); ok && netErr.Timeout() {
return nil
}
return err
}
return nil
})
pingCtx, pingCancel := context.WithCancel(context.Background())
wsClient.mu.WriteHandler(func() error {
wsClient.cancelFuncs = append(wsClient.cancelFuncs, pingCancel)
@@ -280,6 +268,9 @@ func (wsClient *SafeWebsocketClient) connect() error {
if err := c.WriteMessage(int(data.MessageType), data.Data); err != nil {
log.Printf("error on write message: %v\n", err)
wsClient.triggerReconnect()
if data.MessageType == MessageTypePong {
wsClient.pongChan <- err
}
return
}
}
@@ -336,6 +327,20 @@ func (wsClient *SafeWebsocketClient) connect() error {
}
}()
conn.SetPingHandler(func(pingData string) error {
wsClient.writeChan <- Message{
MessageType: MessageTypePong,
Data: []byte(pingData),
}
select {
case err := <-wsClient.pongChan:
return err
default:
}
return nil
})
return nil
}
@@ -351,7 +356,7 @@ func (wsClient *SafeWebsocketClient) startPingTicker(ctx context.Context) {
case <-ticker.C:
wsClient.writeChan <- Message{
MessageType: websocket.PingMessage,
Data: nil,
Data: []byte{},
}
}
}
@@ -420,7 +425,7 @@ func (wsClient *SafeWebsocketClient) reconnectHandler() {
}
func (wsClient *SafeWebsocketClient) ReconnectChannel() <-chan struct{} {
reconnectCh := make(chan struct{})
reconnectCh := make(chan struct{}, 1)
wsClient.mu.WriteHandler(func() error {
wsClient.reconnectChans = append(wsClient.reconnectChans, reconnectCh)
return nil