Merge pull request #54 from latesun/master

fix: no longer try websocket ping after disconnect

go/README.md

@@ -26,18 +26,19 @@ import (
 	"os"
 	"os/signal"
 	"syscall"
+	"time"
 
 	gate "github.com/gateio/gatews/go"
 )
 
 func main() {
 	// create WsService with ConnConf, this is recommended, key and secret will be needed by some channels
 	// ctx and logger could be nil, they'll be initialized by default
-	// ws, err := gate.NewWsService(nil, nil, gate.NewConnConf("",
-	// 	"YOUR_API_KEY", "YOUR_API_SECRET", 10))
-	// RECOMMEND this way to get a ConnConf
 	ws, err := gate.NewWsService(nil, nil, gate.NewConnConfFromOption(&gate.ConfOptions{
-		Key: "YOUR_API_KEY", Secret: "YOUR_API_SECRET", MaxRetryConn: 10, SkipTlsVerify: false,
+		Key:           "YOUR_API_KEY",
+		Secret:        "YOUR_API_SECRET",
+		MaxRetryConn:  10, // default value is math.MaxInt64, set it when needs
+		SkipTlsVerify: false,
 	}))
 	// we can also do nothing to get a WsService, all parameters will be initialized by default and default url is spot
 	// but some channels need key and secret for auth, we can also use set function to set key and secret
@@ -49,6 +50,15 @@ func main() {
 		return
 	}
 
+	// checkout connection status when needs
+	go func() {
+		ticker := time.NewTicker(time.Second)
+		for {
+			<-ticker.C
+			log.Println("connetion status:", ws.Status())
+		}
+	}()
+
 	// create callback functions for receive messages
 	callOrder := gate.NewCallBack(func(msg *gate.UpdateMsg) {
 		// parse the message to struct we need
@@ -58,22 +68,24 @@ func main() {
 		}
 		log.Printf("%+v", order)
 	})
+
 	callTrade := gate.NewCallBack(func(msg *gate.UpdateMsg) {
 		var trade gate.SpotTradeMsg
 		if err := json.Unmarshal(msg.Result, &trade); err != nil {
 			log.Printf("trade Unmarshal err:%s", err.Error())
 		}
 		log.Printf("%+v", trade)
 	})
+
 	// first, we need set callback function
 	ws.SetCallBack(gate.ChannelSpotOrder, callOrder)
 	ws.SetCallBack(gate.ChannelSpotPublicTrade, callTrade)
 	// second, after set callback function, subscribe to any channel you are interested into
-	if err := ws.Subscribe(gate.ChannelSpotPublicTrade, []string{"BCH_USDT"}); err != nil {
+	if err := ws.Subscribe(gate.ChannelSpotPublicTrade, []string{"BTC_USDT"}); err != nil {
 		log.Printf("Subscribe err:%s", err.Error())
 		return
 	}
-	if err := ws.Subscribe(gate.ChannelSpotOrder, []string{"BCH_USDT"}); err != nil {
+	if err := ws.Subscribe(gate.ChannelSpotBookTicker, []string{"BTC_USDT"}); err != nil {
 		log.Printf("Subscribe err:%s", err.Error())
 		return
 	}
@@ -94,4 +106,4 @@ We provide some demo applications in the [examples](_examples) directory, which
 
 ## ChangeLog
 
-[changelog](changelog.md) 
+[changelog](changelog.md)

go/_examples/main.go

@@ -6,6 +6,7 @@ import (
 	"os"
 	"os/signal"
 	"syscall"
+	"time"
 
 	gate "github.com/gateio/gatews/go"
 )
@@ -14,7 +15,10 @@ func main() {
 	// create WsService with ConnConf, this is recommended, key and secret will be needed by some channels
 	// ctx and logger could be nil, they'll be initialized by default
 	ws, err := gate.NewWsService(nil, nil, gate.NewConnConfFromOption(&gate.ConfOptions{
-		Key: "YOUR_API_KEY", Secret: "YOUR_API_SECRET", MaxRetryConn: 10, SkipTlsVerify: false,
+		Key:           "YOUR_API_KEY",
+		Secret:        "YOUR_API_SECRET",
+		MaxRetryConn:  10, // default value is math.MaxInt64, set it when needs
+		SkipTlsVerify: false,
 	}))
 	// we can also do nothing to get a WsService, all parameters will be initialized by default and default url is spot
 	// but some channels need key and secret for auth, we can also use set function to set key and secret
@@ -26,6 +30,15 @@ func main() {
 		return
 	}
 
+	// checkout connection status when needs
+	go func() {
+		ticker := time.NewTicker(time.Second)
+		for {
+			<-ticker.C
+			log.Println("connetion status:", ws.Status())
+		}
+	}()
+
 	// create callback functions for receive messages
 	callOrder := gate.NewCallBack(func(msg *gate.UpdateMsg) {
 		// parse the message to struct we need
@@ -35,15 +48,17 @@ func main() {
 		}
 		log.Printf("%+v", order)
 	})
+
 	callTrade := gate.NewCallBack(func(msg *gate.UpdateMsg) {
 		var trade gate.SpotTradeMsg
 		if err := json.Unmarshal(msg.Result, &trade); err != nil {
 			log.Printf("trade Unmarshal err:%s", err.Error())
 		}
 		log.Printf("%+v", trade)
 	})
+
 	// first, we need set callback function
-	ws.SetCallBack(gate.ChannelSpotBookTicker, callOrder)
+	ws.SetCallBack(gate.ChannelSpotOrder, callOrder)
 	ws.SetCallBack(gate.ChannelSpotPublicTrade, callTrade)
 	// second, after set callback function, subscribe to any channel you are interested into
 	if err := ws.Subscribe(gate.ChannelSpotPublicTrade, []string{"BTC_USDT"}); err != nil {

go/changelog.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## v0.4.2
+
+2023-05-09
+
+- no longer try websocket ping after disconnect
+- support querying websocket connection status
+
 ## v0.4.1
 
 2023-03-16
@@ -128,7 +135,7 @@
 
 - Support futures websocket.
 - Modify channels name to with flag `Spot` or `Future`.
-- Add field `TimestampInMilli`  in models `SpotBalancesMsg`, `SpotFundingBalancesMsg`, `SpotMarginBalancesMsg`. Add
+- Add field `TimestampInMilli` in models `SpotBalancesMsg`, `SpotFundingBalancesMsg`, `SpotMarginBalancesMsg`. Add
   field `TimeInMilli` in model `SpotBookTickerMsg`
 - Add new method `NewConnConfFromOption` to get a ConnConf flexible.
 - Add new method `SubscribeWithOption` to support futures subscribe with id.
@@ -152,4 +159,4 @@
 
 2021-04-12
 
-- Support spot websocket function.
+- Support spot websocket function.

go/channel.go

@@ -6,8 +6,6 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
-	"io"
-	"net"
 	"strings"
 	"time"
 
@@ -30,12 +28,7 @@ func (ws *WsService) Subscribe(channel string, payload []string) error {
 		go ws.receiveCallMsg(channel, msgCh.(chan *UpdateMsg))
 	}
 
-	err := ws.newBaseChannel(channel, payload, msgCh.(chan *UpdateMsg), nil)
-	if err != nil {
-		return err
-	}
-
-	return nil
+	return ws.newBaseChannel(channel, payload, msgCh.(chan *UpdateMsg), nil)
 }
 
 func (ws *WsService) SubscribeWithOption(channel string, payload []string, op *SubscribeOptions) error {
@@ -49,12 +42,7 @@ func (ws *WsService) SubscribeWithOption(channel string, payload []string, op *S
 		go ws.receiveCallMsg(channel, msgCh.(chan *UpdateMsg))
 	}
 
-	err := ws.newBaseChannel(channel, payload, msgCh.(chan *UpdateMsg), op)
-	if err != nil {
-		return err
-	}
-
-	return nil
+	return ws.newBaseChannel(channel, payload, msgCh.(chan *UpdateMsg), op)
 }
 
 func (ws *WsService) UnSubscribe(channel string, payload []string) error {
@@ -105,9 +93,10 @@ func (ws *WsService) baseSubscribe(event string, channel string, payload []strin
 	defer func() {
 		ws.mu.Unlock()
 	}()
+
 	err = ws.Client.WriteMessage(websocket.TextMessage, byteReq)
 	if err != nil {
-		ws.Logger.Printf("wsWrite err:%s", err.Error())
+		ws.Logger.Printf("wsWrite [%s] err:%s", channel, err.Error())
 		return err
 	}
 
@@ -156,23 +145,16 @@ func (ws *WsService) readMsg() {
 				default:
 					_, message, err := ws.Client.ReadMessage()
 					if err != nil {
-						ne, hasNetErr := err.(net.Error)
-						noe, hasNetOpErr := err.(*net.OpError)
-						if websocket.IsUnexpectedCloseError(err) || (hasNetErr && ne.Timeout()) || (hasNetOpErr && noe != nil) ||
-							websocket.IsCloseError(err) || io.ErrUnexpectedEOF == err {
-							ws.Logger.Printf("websocket err:%s", err.Error())
-							if e := ws.reconnect(); e != nil {
-								ws.Logger.Printf("reconnect err:%s", err.Error())
-								return
-							} else {
-								ws.Logger.Printf("reconnect success, continue read message")
-								continue
-							}
-						} else {
-							ws.Logger.Printf("wsRead err:%s, type:%T", err.Error(), err)
+						ws.Logger.Printf("websocket err: %s", err.Error())
+						if e := ws.reconnect(); e != nil {
+							ws.Logger.Printf("reconnect err:%s", err.Error())
 							return
+						} else {
+							ws.Logger.Println("reconnect success, continue read message")
+							continue
 						}
 					}
+
 					var rawTrade UpdateMsg
 					if err := json.Unmarshal(message, &rawTrade); err != nil {
 						ws.Logger.Printf("Unmarshal err:%s, body:%s", err.Error(), string(message))

go/client.go

@@ -3,7 +3,6 @@ package gatews
 import (
 	"context"
 	"crypto/tls"
-	"fmt"
 	"log"
 	"os"
 	"strings"
@@ -14,15 +13,25 @@ import (
 	"github.com/gorilla/websocket"
 )
 
+type status int
+
+const (
+	disconnected status = iota
+	connected
+	reconnecting
+)
+
 type WsService struct {
-	mu     *sync.Mutex
-	Logger *log.Logger
-	Ctx    context.Context
-	Client *websocket.Conn
-	once   *sync.Once
-	msgChs *sync.Map // business chan
-	calls  *sync.Map
-	conf   *ConnConf
+	mu       *sync.Mutex
+	Logger   *log.Logger
+	Ctx      context.Context
+	Client   *websocket.Conn
+	once     *sync.Once
+	msgChs   *sync.Map // business chan
+	calls    *sync.Map
+	conf     *ConnConf
+	status   status
+	clientMu *sync.Mutex
 }
 
 // ConnConf default URL is spot websocket
@@ -49,7 +58,7 @@ type ConfOptions struct {
 
 func NewWsService(ctx context.Context, logger *log.Logger, conf *ConnConf) (*WsService, error) {
 	if logger == nil {
-		logger = log.New(os.Stdout, "", 0)
+		logger = log.New(os.Stdout, "", log.LstdFlags|log.Lshortfile)
 	}
 	if ctx == nil {
 		ctx = context.Background()
@@ -91,14 +100,16 @@ func NewWsService(ctx context.Context, logger *log.Logger, conf *ConnConf) (*WsS
 	}
 
 	ws := &WsService{
-		mu:     new(sync.Mutex),
-		conf:   conf,
-		Logger: logger,
-		Ctx:    ctx,
-		Client: conn,
-		calls:  new(sync.Map),
-		msgChs: new(sync.Map),
-		once:   new(sync.Once),
+		mu:       new(sync.Mutex),
+		conf:     conf,
+		Logger:   logger,
+		Ctx:      ctx,
+		Client:   conn,
+		calls:    new(sync.Map),
+		msgChs:   new(sync.Map),
+		once:     new(sync.Once),
+		status:   connected,
+		clientMu: new(sync.Mutex),
 	}
 
 	go ws.activePing()
@@ -160,6 +171,20 @@ func (ws *WsService) GetConnConf() *ConnConf {
 }
 
 func (ws *WsService) reconnect() error {
+	// avoid repeated reconnection
+	if ws.status == reconnecting {
+		return nil
+	}
+
+	ws.clientMu.Lock()
+	defer ws.clientMu.Unlock()
+
+	if ws.Client != nil {
+		ws.Client.Close()
+	}
+
+	ws.status = reconnecting
+
 	stop := false
 	retry := 0
 	for !stop {
@@ -179,6 +204,8 @@ func (ws *WsService) reconnect() error {
 		}
 	}
 
+	ws.status = connected
+
 	// resubscribe after reconnect
 	ws.conf.subscribeMsg.Range(func(key, value interface{}) bool {
 		// key is channel, value is []requestHistory
@@ -298,9 +325,26 @@ func (ws *WsService) activePing() {
 				return true
 			})
 
+			if ws.status != connected {
+				continue
+			}
+
 			for app := range subscribeMap {
-				ws.Subscribe(fmt.Sprintf("%s.ping", app), nil)
+				channel := app + ".ping"
+				if err := ws.Subscribe(channel, nil); err != nil {
+					ws.Logger.Printf("subscribe channel[%s] failed: %v", channel, err)
+				}
 			}
 		}
 	}
 }
+
+var statusString = map[status]string{
+	disconnected: "disconnected",
+	connected:    "connected",
+	reconnecting: "reconnecting",
+}
+
+func (ws *WsService) Status() string {
+	return statusString[ws.status]
+}