|
|
|
@@ -71,6 +71,7 @@ type SafeWebsocketClientBuilder struct {
|
|
|
|
rawQuery *string
|
|
|
|
rawQuery *string
|
|
|
|
useTLS *bool
|
|
|
|
useTLS *bool
|
|
|
|
channelSize *int64
|
|
|
|
channelSize *int64
|
|
|
|
|
|
|
|
authenticateFn func() error
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
func NewSafeWebsocketClientBuilder() *SafeWebsocketClientBuilder {
|
|
|
|
func NewSafeWebsocketClientBuilder() *SafeWebsocketClientBuilder {
|
|
|
|
@@ -92,6 +93,11 @@ func (b *SafeWebsocketClientBuilder) UseTLS(useTLS bool) *SafeWebsocketClientBui
|
|
|
|
return b
|
|
|
|
return b
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
func (b *SafeWebsocketClientBuilder) AuthenticateFn(authenticateFn func() error) *SafeWebsocketClientBuilder {
|
|
|
|
|
|
|
|
b.authenticateFn = authenticateFn
|
|
|
|
|
|
|
|
return b
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
func (b *SafeWebsocketClientBuilder) Path(path string) *SafeWebsocketClientBuilder {
|
|
|
|
func (b *SafeWebsocketClientBuilder) Path(path string) *SafeWebsocketClientBuilder {
|
|
|
|
b.path = &path
|
|
|
|
b.path = &path
|
|
|
|
return b
|
|
|
|
return b
|
|
|
|
@@ -136,6 +142,10 @@ func (b *SafeWebsocketClientBuilder) Build(ctx context.Context) (*SafeWebsocketC
|
|
|
|
doneMap: NewSafeMap[string, chan struct{}](),
|
|
|
|
doneMap: NewSafeMap[string, chan struct{}](),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if b.authenticateFn != nil {
|
|
|
|
|
|
|
|
wsClient.authenticateFn = b.authenticateFn
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
go wsClient.reconnectHandler()
|
|
|
|
go wsClient.reconnectHandler()
|
|
|
|
|
|
|
|
|
|
|
|
if err := wsClient.connect(); err != nil {
|
|
|
|
if err := wsClient.connect(); err != nil {
|
|
|
|
@@ -157,9 +167,10 @@ type SafeWebsocketClient struct {
|
|
|
|
cancelFuncs []context.CancelFunc
|
|
|
|
cancelFuncs []context.CancelFunc
|
|
|
|
ctx context.Context
|
|
|
|
ctx context.Context
|
|
|
|
reconnectCh chan struct{}
|
|
|
|
reconnectCh chan struct{}
|
|
|
|
|
|
|
|
reconnectChans []chan struct{}
|
|
|
|
isConnected bool
|
|
|
|
isConnected bool
|
|
|
|
|
|
|
|
|
|
|
|
doneMap *SafeMap[string, chan struct{}]
|
|
|
|
doneMap *SafeMap[string, chan struct{}]
|
|
|
|
|
|
|
|
authenticateFn func() error
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
func (wsClient *SafeWebsocketClient) connect() error {
|
|
|
|
func (wsClient *SafeWebsocketClient) connect() error {
|
|
|
|
@@ -200,15 +211,20 @@ func (wsClient *SafeWebsocketClient) connect() error {
|
|
|
|
return nil
|
|
|
|
return nil
|
|
|
|
})
|
|
|
|
})
|
|
|
|
|
|
|
|
|
|
|
|
wsClient.mu.WriteHandler(func() error {
|
|
|
|
|
|
|
|
if wsClient.conn != nil {
|
|
|
|
if wsClient.conn != nil {
|
|
|
|
wsClient.conn.Close()
|
|
|
|
wsClient.conn.Close()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
wsClient.conn = conn
|
|
|
|
wsClient.conn = conn
|
|
|
|
|
|
|
|
|
|
|
|
wsClient.isConnected = true
|
|
|
|
wsClient.isConnected = true
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if wsClient.authenticateFn != nil {
|
|
|
|
|
|
|
|
if err := wsClient.authenticateFn(); err != nil {
|
|
|
|
|
|
|
|
return err
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
wsClient.mu.WriteHandler(func() error {
|
|
|
|
|
|
|
|
|
|
|
|
pingCtx, pingCancel := context.WithCancel(context.Background())
|
|
|
|
pingCtx, pingCancel := context.WithCancel(context.Background())
|
|
|
|
go wsClient.startPingTicker(pingCtx)
|
|
|
|
go wsClient.startPingTicker(pingCtx)
|
|
|
|
wsClient.cancelFuncs = append(wsClient.cancelFuncs, pingCancel)
|
|
|
|
wsClient.cancelFuncs = append(wsClient.cancelFuncs, pingCancel)
|
|
|
|
@@ -269,7 +285,6 @@ func (wsClient *SafeWebsocketClient) startReceiveHandler(ctx context.Context) {
|
|
|
|
case wsClient.dataChannel <- message:
|
|
|
|
case wsClient.dataChannel <- message:
|
|
|
|
case <-ctx.Done():
|
|
|
|
case <-ctx.Done():
|
|
|
|
log.Println("Reconnect handler stopped")
|
|
|
|
log.Println("Reconnect handler stopped")
|
|
|
|
return nil
|
|
|
|
|
|
|
|
default:
|
|
|
|
default:
|
|
|
|
log.Println("Data channel full, dropping message")
|
|
|
|
log.Println("Data channel full, dropping message")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
@@ -301,7 +316,6 @@ func (wsClient *SafeWebsocketClient) reconnectHandler() {
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
wsClient.isConnected = false
|
|
|
|
wsClient.isConnected = false
|
|
|
|
|
|
|
|
|
|
|
|
time.Sleep(100 * time.Millisecond)
|
|
|
|
time.Sleep(100 * time.Millisecond)
|
|
|
|
|
|
|
|
|
|
|
|
isInnerLoop := true
|
|
|
|
isInnerLoop := true
|
|
|
|
@@ -324,7 +338,11 @@ func (wsClient *SafeWebsocketClient) reconnectHandler() {
|
|
|
|
return
|
|
|
|
return
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
if wsClient.reconnectChans != nil {
|
|
|
|
|
|
|
|
for _, reconnectCh := range wsClient.reconnectChans {
|
|
|
|
|
|
|
|
reconnectCh <- struct{}{}
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
case <-wsClient.ctx.Done():
|
|
|
|
case <-wsClient.ctx.Done():
|
|
|
|
log.Println("reconnect handler stopped due to client shutdown")
|
|
|
|
log.Println("reconnect handler stopped due to client shutdown")
|
|
|
|
wsClient.Close()
|
|
|
|
wsClient.Close()
|
|
|
|
@@ -333,6 +351,16 @@ func (wsClient *SafeWebsocketClient) reconnectHandler() {
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
func (wsClient *SafeWebsocketClient) ReconnectChannel() <-chan struct{} {
|
|
|
|
|
|
|
|
reconnectCh := make(chan struct{})
|
|
|
|
|
|
|
|
wsClient.mu.WriteHandler(func() error {
|
|
|
|
|
|
|
|
wsClient.reconnectChans = append(wsClient.reconnectChans, reconnectCh)
|
|
|
|
|
|
|
|
return nil
|
|
|
|
|
|
|
|
})
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
return reconnectCh
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
func (wsClient *SafeWebsocketClient) DataChannel() <-chan []byte {
|
|
|
|
func (wsClient *SafeWebsocketClient) DataChannel() <-chan []byte {
|
|
|
|
return wsClient.dataChannel
|
|
|
|
return wsClient.dataChannel
|
|
|
|
}
|
|
|
|
}
|
|
|
|
@@ -350,6 +378,12 @@ func (wsClient *SafeWebsocketClient) Close() error {
|
|
|
|
cancel()
|
|
|
|
cancel()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if wsClient.reconnectChans != nil {
|
|
|
|
|
|
|
|
for _, reconnectChan := range wsClient.reconnectChans {
|
|
|
|
|
|
|
|
close(reconnectChan)
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
if wsClient.conn != nil {
|
|
|
|
if wsClient.conn != nil {
|
|
|
|
wsClient.conn.Close()
|
|
|
|
wsClient.conn.Close()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|