rogerferdinan/safe-web-socket
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: seperete goroutine into function for readability

Browse Source
This commit is contained in:
Roger Ferdinan
2025-10-11 10:16:02 +07:00
parent 88ae539f54
commit 0d0de32d21
3 changed files with 104 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
go.mod
View File

@@ -5,3 +5,5 @@ go 1.24.5
require github.com/gorilla/websocket v1.5.3
require git.neurocipta.com/rogerferdinan/custom-rwmutex v1.0.0
require git.neurocipta.com/rogerferdinan/safe-map v0.0.0-20251011004629-ab0b119a7c48 // indirect

2
go.sum
View File

@@ -1,4 +1,6 @@
git.neurocipta.com/rogerferdinan/custom-rwmutex v1.0.0 h1:KnNc40SrYsg0cksIIcQy/ca6bunkGADQOs1u7O/E+iY=
git.neurocipta.com/rogerferdinan/custom-rwmutex v1.0.0/go.mod h1:9DvvHc2UZhBwEs63NgO4IhiuHnBNtTuBkTJgiMnnCss=
git.neurocipta.com/rogerferdinan/safe-map v0.0.0-20251011004629-ab0b119a7c48 h1:4wXSbEuwFd2gycaaGP35bjUkKEEO6WcVfJ6cetEyT5s=
git.neurocipta.com/rogerferdinan/safe-map v0.0.0-20251011004629-ab0b119a7c48/go.mod h1:QtIxG0BYCCq8a5qyklpSHA8qWUvKr+mfl42qF9QxTc0=
github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=

116
v1/client/client.go
View File

@@ -6,10 +6,10 @@ import (
"log"
"net/url"
"strings"
"sync"
"time"
custom_rwmutex "git.neurocipta.com/rogerferdinan/custom-rwmutex"
safemap "git.neurocipta.com/rogerferdinan/safe-map"
"git.neurocipta.com/rogerferdinan/safe-web-socket/internal"
"github.com/gorilla/websocket"
)
@@ -19,52 +19,6 @@ const (
readDeadline = 30 * time.Second
)
type SafeMap[K comparable, V any] struct {
m sync.Map
}
func NewSafeMap[K comparable, V any]() *SafeMap[K, V] {
return &SafeMap[K, V]{
m: sync.Map{},
}
}
func (sm *SafeMap[K, V]) Store(key K, value V) {
sm.m.Store(key, value)
}
func (sm *SafeMap[K, V]) Load(key K) (value V, ok bool) {
val, loaded := sm.m.Load(key)
if !loaded {
return *new(V), false
}
return val.(V), true
}
func (sm *SafeMap[K, V]) Delete(key K) {
sm.m.Delete(key)
}
func (sm *SafeMap[K, V]) Range(f func(K, V) bool) {
sm.m.Range(func(key, value any) bool {
k, ok1 := key.(K)
v, ok2 := value.(V)
if !ok1 || !ok2 {
return true
}
return f(k, v)
})
}
func (sm *SafeMap[K, V]) Len() int {
count := 0
sm.Range(func(_ K, _ V) bool {
count++
return true
})
return count
}
type MessageType uint
const (
@@ -86,7 +40,6 @@ type SafeWebsocketClientBuilder struct {
rawQuery *string
useTLS *bool
channelSize *int64
authenticateFn func(*SafeWebsocketClient) error
}
func NewSafeWebsocketClientBuilder() *SafeWebsocketClientBuilder {
@@ -108,11 +61,6 @@ func (b *SafeWebsocketClientBuilder) UseTLS(useTLS bool) *SafeWebsocketClientBui
return b
}
func (b *SafeWebsocketClientBuilder) AuthenticateFn(authenticateFn func(*SafeWebsocketClient) error) *SafeWebsocketClientBuilder {
b.authenticateFn = authenticateFn
return b
}
func (b *SafeWebsocketClientBuilder) Path(path string) *SafeWebsocketClientBuilder {
b.path = &path
return b
@@ -154,14 +102,10 @@ func (b *SafeWebsocketClientBuilder) Build(ctx context.Context) (*SafeWebsocketC
ctx: ctx,
reconnectCh: make(chan struct{}, 1),
isConnected: false,
doneMap: NewSafeMap[string, chan struct{}](),
doneMap: safemap.NewSafeMap[string, chan struct{}](),
writeChan: make(chan Message, 1),
}
if b.authenticateFn != nil {
wsClient.authenticateFn = b.authenticateFn
}
go wsClient.reconnectHandler()
if err := wsClient.connect(); err != nil {
@@ -189,8 +133,7 @@ type SafeWebsocketClient struct {
reconnectCh chan struct{}
reconnectChans []chan struct{}
isConnected bool
doneMap *SafeMap[string, chan struct{}]
authenticateFn func(*SafeWebsocketClient) error
doneMap *safemap.SafeMap[string, chan struct{}]
writeChan chan Message
pongChan chan error
@@ -235,13 +178,27 @@ func (wsClient *SafeWebsocketClient) connect() error {
wsClient.conn = conn
wsClient.isConnected = true
if wsClient.authenticateFn != nil {
if err := wsClient.authenticateFn(wsClient); err != nil {
return err
}
go wsClient.writePump()
go wsClient.readPump()
conn.SetPingHandler(func(pingData string) error {
wsClient.writeChan <- Message{
MessageType: MessageTypePong,
Data: []byte(pingData),
}
go func() {
select {
case err := <-wsClient.pongChan:
return err
default:
}
return nil
})
return nil
}
func (wsClient *SafeWebsocketClient) writePump() {
ctx, cancel := context.WithCancel(context.Background())
wsClient.mu.WriteHandler(func() error {
wsClient.cancelFuncs = append(wsClient.cancelFuncs, cancel)
@@ -250,7 +207,7 @@ func (wsClient *SafeWebsocketClient) connect() error {
var c *websocket.Conn
wsClient.mu.ReadHandler(func() error {
c = conn
c = wsClient.conn
return nil
})
@@ -267,6 +224,7 @@ 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)
c.Close()
wsClient.triggerReconnect()
if data.MessageType == MessageTypePong {
wsClient.pongChan <- err
@@ -275,9 +233,9 @@ func (wsClient *SafeWebsocketClient) connect() error {
}
}
}
}()
}
go func() {
func (wsClient *SafeWebsocketClient) readPump() {
ctx, cancel := context.WithCancel(context.Background())
wsClient.mu.WriteHandler(func() error {
wsClient.cancelFuncs = append(wsClient.cancelFuncs, cancel)
@@ -287,13 +245,14 @@ func (wsClient *SafeWebsocketClient) connect() error {
var c *websocket.Conn
wsClient.mu.ReadHandler(func() error {
c = conn
c = wsClient.conn
return nil
})
if c == nil {
wsClient.triggerReconnect()
return
}
for {
select {
case <-ctx.Done():
@@ -325,23 +284,6 @@ 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
}
func (wsClient *SafeWebsocketClient) startPingTicker(ctx context.Context) {
@@ -387,7 +329,7 @@ func (wsClient *SafeWebsocketClient) reconnectHandler() {
})
wsClient.isConnected = false
time.Sleep(100 * time.Millisecond)
// time.Sleep(100 * time.Millisecond)
isInnerLoop := true
for isInnerLoop {