mm: Close book feeds (#2907)

* mm: Close book feeds

Book feeds were not being closed, leading to warning logs.

client/mm/exchange_adaptor.go

@@ -2297,6 +2297,7 @@ func (u *unifiedExchangeAdaptor) cancelAllOrders(ctx context.Context) {
 		u.tryCancelOrders(ctx, nil, true)
 		return
 	}
+	defer bookFeed.Close()
 
 	mktCfg, err := u.clientCore.ExchangeMarket(u.host, u.baseID, u.quoteID)
 	if err != nil {

client/mm/mm_arb_market_maker.go

@@ -422,7 +422,6 @@ func (a *arbMarketMaker) registerFeeGap() {
 }
 
 func (a *arbMarketMaker) botLoop(ctx context.Context) (*sync.WaitGroup, error) {
-
 	book, bookFeed, err := a.core.SyncBook(a.host, a.baseID, a.quoteID)
 	if err != nil {
 		return nil, fmt.Errorf("failed to sync book: %v", err)
@@ -431,6 +430,7 @@ func (a *arbMarketMaker) botLoop(ctx context.Context) (*sync.WaitGroup, error) {
 
 	err = a.cex.SubscribeMarket(ctx, a.baseID, a.quoteID)
 	if err != nil {
+		bookFeed.Close()
 		return nil, fmt.Errorf("failed to subscribe to cex market: %v", err)
 	}
 
@@ -440,6 +440,7 @@ func (a *arbMarketMaker) botLoop(ctx context.Context) (*sync.WaitGroup, error) {
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
+		defer bookFeed.Close()
 		for {
 			select {
 			case ni := <-bookFeed.Next():

client/mm/mm_basic.go

@@ -387,6 +387,7 @@ func (m *basicMarketMaker) botLoop(ctx context.Context) (*sync.WaitGroup, error)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
+		defer bookFeed.Close()
 		for {
 			select {
 			case ni := <-bookFeed.Next():

client/mm/mm_simple_arb.go

@@ -448,6 +448,7 @@ func (a *simpleArbMarketMaker) botLoop(ctx context.Context) (*sync.WaitGroup, er
 
 	err = a.cex.SubscribeMarket(ctx, a.baseID, a.quoteID)
 	if err != nil {
+		bookFeed.Close()
 		return nil, fmt.Errorf("failed to subscribe to cex market: %v", err)
 	}
 
@@ -457,6 +458,7 @@ func (a *simpleArbMarketMaker) botLoop(ctx context.Context) (*sync.WaitGroup, er
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
+		defer bookFeed.Close()
 		for {
 			select {
 			case ni := <-bookFeed.Next():