Compare commits

...

1 Commits

Author SHA1 Message Date
7f21b733ed feat: adding ping & pong handler 2026-02-01 19:05:33 +07:00
4 changed files with 33 additions and 22 deletions

View File

@@ -169,7 +169,6 @@ type SafeWebsocketClient struct {
doneMap *safemap.SafeMap[string, chan struct{}] doneMap *safemap.SafeMap[string, chan struct{}]
writeChan chan Message writeChan chan Message
pongChan chan error
} }
func (wsClient *SafeWebsocketClient) connect() error { func (wsClient *SafeWebsocketClient) connect() error {
@@ -196,7 +195,6 @@ func (wsClient *SafeWebsocketClient) connect() error {
if wsClient.headers != nil { if wsClient.headers != nil {
for k, v := range *wsClient.headers { for k, v := range *wsClient.headers {
fmt.Println(k, v)
header.Set(k, v) header.Set(k, v)
} }
} }
@@ -220,23 +218,30 @@ func (wsClient *SafeWebsocketClient) connect() error {
wsClient.conn = conn wsClient.conn = conn
wsClient.isConnected = true wsClient.isConnected = true
conn.SetPingHandler(func(pingData string) error {
if err := conn.SetReadDeadline(time.Now().Add(readDeadline)); err != nil {
log.Printf("error on read deadline: %v\n", err)
return err
}
wsClient.writeChan <- Message{
MessageType: MessageTypePong,
Data: []byte(pingData),
}
return nil
})
conn.SetPongHandler(func(pingData string) error {
if err := conn.SetReadDeadline(time.Now().Add(readDeadline)); err != nil {
log.Printf("error on read deadline: %v\n", err)
return err
}
return nil
})
go wsClient.writePump() go wsClient.writePump()
go wsClient.readPump() go wsClient.readPump()
// 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 return nil
} }
@@ -265,9 +270,6 @@ func (wsClient *SafeWebsocketClient) writePump() {
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)
if data.MessageType == MessageTypePong {
wsClient.pongChan <- err
}
return return
} }
} }
@@ -450,7 +452,7 @@ func (wsClient *SafeWebsocketClient) Close() error {
wsClient.conn.Close() wsClient.conn.Close()
} }
wsClient.isConnected = false wsClient.isConnected = false
// close(wsClient.dataChannel) close(wsClient.dataChannel)
return nil return nil
} }

View File

@@ -8,10 +8,18 @@ import (
"os/signal" "os/signal"
"syscall" "syscall"
"net/http"
_ "net/http/pprof"
"git.neurocipta.com/rogerferdinan/safe-web-socket/v1/client" "git.neurocipta.com/rogerferdinan/safe-web-socket/v1/client"
) )
func main() { func main() {
go func() {
log.Println("Starting pprof server on :6060")
log.Println(http.ListenAndServe(":6060", nil))
}()
sigChan := make(chan os.Signal, 1) sigChan := make(chan os.Signal, 1)
signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM) signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

View File

@@ -19,7 +19,7 @@ func main() {
BasePort(8080). BasePort(8080).
ApiKey("abcd"). ApiKey("abcd").
HandleFuncWebsocket("/ws/test/", "data_1", func(c chan []byte) { HandleFuncWebsocket("/ws/test/", "data_1", func(c chan []byte) {
ticker := time.NewTicker(10 * time.Millisecond) ticker := time.NewTicker(100 * time.Millisecond)
for range ticker.C { for range ticker.C {
jsonBytes, err := json.Marshal(ExampleData{ jsonBytes, err := json.Marshal(ExampleData{
Time: time.Now(), Time: time.Now(),
@@ -32,7 +32,7 @@ func main() {
} }
}). }).
HandleFuncWebsocket("/ws/test/", "data_2", func(c chan []byte) { HandleFuncWebsocket("/ws/test/", "data_2", func(c chan []byte) {
ticker := time.NewTicker(10 * time.Millisecond) ticker := time.NewTicker(100 * time.Millisecond)
for range ticker.C { for range ticker.C {
jsonBytes, err := json.Marshal(ExampleData{ jsonBytes, err := json.Marshal(ExampleData{
Time: time.Now(), Time: time.Now(),

View File

@@ -115,6 +115,7 @@ func (s *SafeWebsocketServer) AuthMiddleware(next http.Handler) http.Handler {
StatusCode(http.StatusForbidden). StatusCode(http.StatusForbidden).
Message("X-MBX-APIKEY is missing"). Message("X-MBX-APIKEY is missing").
Build()) Build())
return
} }
next.ServeHTTP(w, r) next.ServeHTTP(w, r)
}) })