fix binance emitAuth event trigger

pkg/exchange/binance/stream.go

@@ -148,7 +148,9 @@ func (s *Stream) handleConnect() {
 	if !s.PublicOnly {
 		// Emit Auth before establishing the connection to prevent the caller from missing the Update data after
 		// creating the order.
-		s.EmitAuth()
+
+		// spawn a goroutine to emit auth event to prevent blocking the main event loop
+		go s.EmitAuth()
 		return
 	}
 
@@ -450,6 +452,24 @@ func (s *Stream) closeListenKey(ctx context.Context, listenKey string) (err erro
 	return err
 }
 
+func (s *Stream) String() string {
+	ss := "binance.Stream"
+
+	if s.PublicOnly {
+		ss += " (public only)"
+	} else {
+		ss += " (user data)"
+	}
+
+	if s.MarginSettings.IsMargin {
+		ss += " (margin)"
+	} else if s.FuturesSettings.IsFutures {
+		ss += " (futures)"
+	}
+
+	return ss
+}
+
 // listenKeyKeepAlive
 // From Binance
 // Keepalive a user data stream to prevent a time out. User data streams will close after 60 minutes.

pkg/exchange/max/stream.go

@@ -254,6 +254,22 @@ func (s *Stream) handleBookEvent(ex *Exchange) func(e max.BookEvent) {
 	}
 }
 
+func (s *Stream) String() string {
+	ss := "max.Stream"
+
+	if s.PublicOnly {
+		ss += " (public only)"
+	} else {
+		ss += " (user data)"
+	}
+
+	if s.MarginSettings.IsMargin {
+		ss += " (margin)"
+	}
+
+	return ss
+}
+
 func (s *Stream) handleAccountSnapshotEvent(e max.AccountSnapshotEvent) {
 	snapshot := map[string]types.Balance{}
 	for _, bm := range e.Balances {

pkg/types/connectivitygroup.go

@@ -97,7 +97,6 @@ func NewConnectivityGroup(cons ...*Connectivity) *ConnectivityGroup {
 	sumState := sumStates(states)
 	g := &ConnectivityGroup{
 		Connectivity: NewConnectivity(),
-		connections:  cons,
 		states:       states,
 		sumState:     sumState,
 	}
@@ -206,7 +205,8 @@ func (g *ConnectivityGroup) waitAllAuthed(ctx context.Context, c chan struct{},
 			return
 
 		default:
-			if g.GetState() == ConnectivityStateAuthed {
+			state := g.GetState()
+			if state == ConnectivityStateAuthed {
 				close(c)
 				return
 			}

pkg/types/stream.go

@@ -2,6 +2,7 @@ package types
 
 import (
 	"context"
+	"fmt"
 	"net"
 	"net/http"
 	"sync"
@@ -533,6 +534,12 @@ func (s *StandardStream) Close() error {
 	return nil
 }
 
+func (s *StandardStream) String() string {
+	ss := "StandardStream"
+	ss += fmt.Sprintf("(%p)", s)
+	return ss
+}
+
 // SetHeartBeat sets the custom heart beat implementation if needed
 func (s *StandardStream) SetHeartBeat(fn HeartBeat) {
 	s.heartBeat = fn