binance: update book stream url

client/comms/wsconn.go

@@ -98,6 +98,7 @@ type WsConn interface {
  RequestWithTimeout(msg *msgjson.Message, respHandler func(*msgjson.Message), expireTime time.Duration, expire func()) error
  Connect(ctx context.Context) (*sync.WaitGroup, error)
  MessageSource() <-chan *msgjson.Message
+ UpdateURL(string)
 }
 
 // When the DEX sends a request to the client, a responseHandler is created
@@ -161,6 +162,7 @@ type wsConn struct {
  cfg *WsCfg
  tlsCfg *tls.Config
  readCh chan *msgjson.Message
+ urlV atomic.Value // string
 
  wsMtx sync.Mutex
  ws *websocket.Conn
@@ -203,14 +205,25 @@ func NewWsConn(cfg *WsCfg) (WsConn, error) {
  ServerName: uri.Hostname(),
  }
 
- return &wsConn{
+ conn := &wsConn{
  cfg: cfg,
  log: cfg.Logger,
  tlsCfg: tlsConfig,
  readCh: make(chan *msgjson.Message, readBuffSize),
  respHandlers: make(map[uint64]*responseHandler),
  reconnectCh: make(chan struct{}, 1),
- }, nil
+ }
+ conn.urlV.Store(cfg.URL)
+
+ return conn, nil
+}
+
+func (conn *wsConn) UpdateURL(uri string) {
+ conn.urlV.Store(uri)
+}
+
+func (conn *wsConn) url() string {
+ return conn.urlV.Load().(string)
 }
 
 // IsDown indicates if the connection is known to be down.
@@ -240,7 +253,7 @@ func (conn *wsConn) connect(ctx context.Context) error {
  dialer.Proxy = http.ProxyFromEnvironment
  }
 
- ws, _, err := dialer.DialContext(ctx, conn.cfg.URL, conn.cfg.ConnectHeaders)
+ ws, _, err := dialer.DialContext(ctx, conn.url(), conn.cfg.ConnectHeaders)
  if err != nil {
  if isErrorInvalidCert(err) {
  conn.setConnectionStatus(InvalidCert)
@@ -331,7 +344,7 @@ func (conn *wsConn) handleReadError(err error) {
 
  var netErr net.Error
  if errors.As(err, &netErr) && netErr.Timeout() {
- conn.log.Errorf("Read timeout on connection to %s.", conn.cfg.URL)
+ conn.log.Errorf("Read timeout on connection to %s.", conn.url())
  reconnect()
  return
  }
@@ -457,11 +470,11 @@ func (conn *wsConn) keepAlive(ctx context.Context) {
  return
  }
 
- conn.log.Infof("Attempting to reconnect to %s...", conn.cfg.URL)
+ conn.log.Infof("Attempting to reconnect to %s...", conn.url())
  err := conn.connect(ctx)
  if err != nil {
  conn.log.Errorf("Reconnect failed. Scheduling reconnect to %s in %.1f seconds.",
- conn.cfg.URL, rcInt.Seconds())
+ conn.url(), rcInt.Seconds())
  time.AfterFunc(rcInt, func() {
  conn.reconnectCh <- struct{}{}
  })

client/core/core_test.go

@@ -346,6 +346,8 @@ func (conn *TWebsocket) Connect(context.Context) (*sync.WaitGroup, error) {
  return &sync.WaitGroup{}, conn.connectErr
 }
 
+func (conn *TWebsocket) UpdateURL(string) {}
+
 type TDB struct {
  updateWalletErr error
  acct *db.AccountInfo

client/mm/libxc/binance.go

@@ -1618,7 +1618,7 @@ func (bnc *binance) getOrderbookSnapshot(ctx context.Context, mktSymbol string)
 // subscribeToAdditionalMarketDataStream is called when a new market is
 // subscribed to after the market data stream connection has already been
 // established.
-func (bnc *binance) subscribeToAdditionalMarketDataStream(ctx context.Context, baseID, quoteID uint32) error {
+func (bnc *binance) subscribeToAdditionalMarketDataStream(ctx context.Context, baseID, quoteID uint32) (err error) {
  baseCfg, quoteCfg, err := bncAssetCfgs(baseID, quoteID)
  if err != nil {
  return fmt.Errorf("error getting asset cfg for %d: %w", baseID, err)
@@ -1627,6 +1627,10 @@ func (bnc *binance) subscribeToAdditionalMarketDataStream(ctx context.Context, b
  mktID := binanceMktID(baseCfg, quoteCfg)
  streamID := marketDataStreamID(mktID)
 
+ defer func() {
+ bnc.marketStream.UpdateURL(bnc.streamURL())
+ }()
+
  bnc.booksMtx.Lock()
  defer bnc.booksMtx.Unlock()
 
@@ -1662,6 +1666,10 @@ func (bnc *binance) streams() []string {
  return streamNames
 }
 
+func (bnc *binance) streamURL() string {
+ return fmt.Sprintf("%s/stream?streams=%s", bnc.wsURL, strings.Join(bnc.streams(), "/"))
+}
+
 // checkSubs will query binance for current market subscriptions and compare
 // that to what subscriptions we should have. If there is a discrepancy a
 // warning is logged and the market subbed or unsubbed.
@@ -1756,8 +1764,7 @@ func (bnc *binance) connectToMarketDataStream(ctx context.Context, baseID, quote
  reconnectC := make(chan struct{})
  checkSubsC := make(chan struct{})
 
- newConnection := func() (comms.WsConn, *dex.ConnectionMaster, error) {
- addr := fmt.Sprintf("%s/stream?streams=%s", bnc.wsURL, strings.Join(bnc.streams(), "/"))
+ newConnection := func() (*dex.ConnectionMaster, error) {
  // Need to send key but not signature
  connectEventFunc := func(cs comms.ConnectionStatus) {
  if cs != comms.Disconnected && cs != comms.Connected {
@@ -1776,7 +1783,7 @@ func (bnc *binance) connectToMarketDataStream(ctx context.Context, baseID, quote
  }
  }
  conn, err := comms.NewWsConn(&comms.WsCfg{
- URL: addr,
+ URL: bnc.streamURL(),
  // Binance Docs: The websocket server will send a ping frame every 3
  // minutes. If the websocket server does not receive a pong frame
  // back from the connection within a 10 minute period, the connection
@@ -1795,16 +1802,16 @@ func (bnc *binance) connectToMarketDataStream(ctx context.Context, baseID, quote
  RawHandler: bnc.handleMarketDataNote,
  })
  if err != nil {
- return nil, nil, err
+ return nil, err
  }
 
  bnc.marketStream = conn
  cm := dex.NewConnectionMaster(conn)
  if err = cm.ConnectOnce(ctx); err != nil {
- return nil, nil, fmt.Errorf("websocketHandler remote connect: %v", err)
+ return nil, fmt.Errorf("websocketHandler remote connect: %v", err)
  }
 
- return conn, cm, nil
+ return cm, nil
  }
 
  // Add the initial book to the books map
@@ -1822,32 +1829,27 @@ func (bnc *binance) connectToMarketDataStream(ctx context.Context, baseID, quote
  bnc.booksMtx.Unlock()
 
  // Create initial connection to the market data stream
- conn, cm, err := newConnection()
+ cm, err := newConnection()
  if err != nil {
  return fmt.Errorf("error connecting to market data stream : %v", err)
  }
 
- bnc.marketStream = conn
-
  book.sync(ctx)
 
  // Start a goroutine to reconnect every 12 hours
  go func() {
  reconnect := func() error {
  bnc.marketStreamMtx.Lock()
  defer bnc.marketStreamMtx.Unlock()
-
  oldCm := cm
- conn, cm, err = newConnection()
+ cm, err = newConnection()
  if err != nil {
  return err
  }
 
  if oldCm != nil {
  oldCm.Disconnect()
  }
-
- bnc.marketStream = conn
  return nil
  }
 
@@ -1922,6 +1924,8 @@ func (bnc *binance) UnsubscribeMarket(baseID, quoteID uint32) (err error) {
  defer func() {
  bnc.booksMtx.Unlock()
 
+ conn.UpdateURL(bnc.streamURL())
+
  if closer != nil {
  closer.Disconnect()
  }