Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 88ae539f54 | |||
| 78a5b21531 | |||
| e011931436 | |||
| 70d9a37a4e |
2
go.mod
2
go.mod
@@ -4,4 +4,4 @@ go 1.24.5
|
|||||||
|
|
||||||
require github.com/gorilla/websocket v1.5.3
|
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
|
||||||
|
|||||||
@@ -23,9 +23,9 @@ type Client struct {
|
|||||||
func NewClient(conn *websocket.Conn, subscribedPath string) *Client {
|
func NewClient(conn *websocket.Conn, subscribedPath string) *Client {
|
||||||
return &Client{
|
return &Client{
|
||||||
Conn: conn,
|
Conn: conn,
|
||||||
Send: make(chan []byte, 64),
|
Send: make(chan []byte, 2),
|
||||||
SubscribedPath: subscribedPath,
|
SubscribedPath: subscribedPath,
|
||||||
done: make(chan struct{}),
|
done: make(chan struct{}, 1),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -38,10 +38,10 @@ type Hub struct {
|
|||||||
|
|
||||||
func NewHub() *Hub {
|
func NewHub() *Hub {
|
||||||
return &Hub{
|
return &Hub{
|
||||||
Broadcast: make(chan []byte),
|
Broadcast: make(chan []byte, 1),
|
||||||
Register: make(chan *Client),
|
Register: make(chan *Client, 1),
|
||||||
Unregister: make(chan *Client),
|
Unregister: make(chan *Client, 1),
|
||||||
Clients: make(map[*Client]bool),
|
Clients: make(map[*Client]bool, 1),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -117,7 +117,7 @@ func ReadPump(c *Client, h *Hub) {
|
|||||||
for {
|
for {
|
||||||
messageType, message, err := c.Conn.ReadMessage()
|
messageType, message, err := c.Conn.ReadMessage()
|
||||||
if err != nil {
|
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)
|
log.Printf("WebSocket error: %v", err)
|
||||||
}
|
}
|
||||||
break
|
break
|
||||||
|
|||||||
@@ -16,7 +16,7 @@ import (
|
|||||||
|
|
||||||
const (
|
const (
|
||||||
pingPeriod = 10 * time.Second
|
pingPeriod = 10 * time.Second
|
||||||
readDeadline = 120 * time.Second
|
readDeadline = 30 * time.Second
|
||||||
)
|
)
|
||||||
|
|
||||||
type SafeMap[K comparable, V any] struct {
|
type SafeMap[K comparable, V any] struct {
|
||||||
@@ -68,10 +68,10 @@ func (sm *SafeMap[K, V]) Len() int {
|
|||||||
type MessageType uint
|
type MessageType uint
|
||||||
|
|
||||||
const (
|
const (
|
||||||
MessageTypeText MessageType = iota + 1
|
MessageTypeText MessageType = websocket.TextMessage
|
||||||
MessageTypePing MessageType = iota
|
MessageTypePing MessageType = websocket.PingMessage
|
||||||
MessageTypePong MessageType = iota
|
MessageTypePong MessageType = websocket.PongMessage
|
||||||
MessageTypeClose MessageType = iota
|
MessageTypeClose MessageType = websocket.CloseMessage
|
||||||
)
|
)
|
||||||
|
|
||||||
type Message struct {
|
type Message struct {
|
||||||
@@ -155,7 +155,7 @@ func (b *SafeWebsocketClientBuilder) Build(ctx context.Context) (*SafeWebsocketC
|
|||||||
reconnectCh: make(chan struct{}, 1),
|
reconnectCh: make(chan struct{}, 1),
|
||||||
isConnected: false,
|
isConnected: false,
|
||||||
doneMap: NewSafeMap[string, chan struct{}](),
|
doneMap: NewSafeMap[string, chan struct{}](),
|
||||||
writeChan: make(chan Message),
|
writeChan: make(chan Message, 1),
|
||||||
}
|
}
|
||||||
|
|
||||||
if b.authenticateFn != nil {
|
if b.authenticateFn != nil {
|
||||||
@@ -193,6 +193,7 @@ type SafeWebsocketClient struct {
|
|||||||
authenticateFn func(*SafeWebsocketClient) error
|
authenticateFn func(*SafeWebsocketClient) error
|
||||||
|
|
||||||
writeChan chan Message
|
writeChan chan Message
|
||||||
|
pongChan chan error
|
||||||
}
|
}
|
||||||
|
|
||||||
func (wsClient *SafeWebsocketClient) connect() error {
|
func (wsClient *SafeWebsocketClient) connect() error {
|
||||||
@@ -267,6 +268,9 @@ func (wsClient *SafeWebsocketClient) connect() error {
|
|||||||
if err := c.WriteMessage(int(data.MessageType), data.Data); err != nil {
|
if err := c.WriteMessage(int(data.MessageType), data.Data); err != nil {
|
||||||
log.Printf("error on write message: %v\n", err)
|
log.Printf("error on write message: %v\n", err)
|
||||||
wsClient.triggerReconnect()
|
wsClient.triggerReconnect()
|
||||||
|
if data.MessageType == MessageTypePong {
|
||||||
|
wsClient.pongChan <- err
|
||||||
|
}
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -324,18 +328,15 @@ func (wsClient *SafeWebsocketClient) connect() error {
|
|||||||
}()
|
}()
|
||||||
|
|
||||||
conn.SetPingHandler(func(pingData string) error {
|
conn.SetPingHandler(func(pingData string) error {
|
||||||
// wsClient.writeChan <- Message{
|
wsClient.writeChan <- Message{
|
||||||
// MessageType: MessageTypePong,
|
MessageType: MessageTypePong,
|
||||||
// Data: []byte(pingData),
|
Data: []byte(pingData),
|
||||||
// }
|
}
|
||||||
if err := conn.WriteMessage(websocket.PongMessage, []byte(pingData)); err != nil {
|
|
||||||
if err == websocket.ErrCloseSent {
|
select {
|
||||||
return nil
|
case err := <-wsClient.pongChan:
|
||||||
}
|
|
||||||
if netErr, ok := err.(interface{ Timeout() bool }); ok && netErr.Timeout() {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
return err
|
return err
|
||||||
|
default:
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
})
|
})
|
||||||
@@ -424,7 +425,7 @@ func (wsClient *SafeWebsocketClient) reconnectHandler() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (wsClient *SafeWebsocketClient) ReconnectChannel() <-chan struct{} {
|
func (wsClient *SafeWebsocketClient) ReconnectChannel() <-chan struct{} {
|
||||||
reconnectCh := make(chan struct{})
|
reconnectCh := make(chan struct{}, 1)
|
||||||
wsClient.mu.WriteHandler(func() error {
|
wsClient.mu.WriteHandler(func() error {
|
||||||
wsClient.reconnectChans = append(wsClient.reconnectChans, reconnectCh)
|
wsClient.reconnectChans = append(wsClient.reconnectChans, reconnectCh)
|
||||||
return nil
|
return nil
|
||||||
|
|||||||
Reference in New Issue
Block a user