fix(direct):range err

Author: DiyiWi

internal/adapt/direct_adapter/adapter.go

@@ -1,6 +1,7 @@
 package direct_adapter
 
 import (
+	"context"
 	"fmt"
 	"runtime"
 	"sync"
@@ -33,6 +34,8 @@ type DirectAdapter struct {
 	ibtps           sync.Map
 	appchainID      string
 	remotePierID    string
+	ctx             context.Context
+	cancel          context.CancelFunc
 	gopool          *pool
 }
 
@@ -55,7 +58,7 @@ func (d *DirectAdapter) GetServiceIDList() ([]string, error) {
 func New(peerMgr peermgr.PeerManager, appchainAdapt adapt.Adapt, logger logrus.FieldLogger) (*DirectAdapter, error) {
 
 	appchainID := appchainAdapt.ID()
-
+	ctx, cancel := context.WithCancel(context.Background())
 	da := &DirectAdapter{
 		logger:        logger,
 		peerMgr:       peerMgr,
@@ -64,6 +67,8 @@ func New(peerMgr peermgr.PeerManager, appchainAdapt adapt.Adapt, logger logrus.F
 		ibtpC:         make(chan *pb.IBTP, maxChSize),
 		appchainID:    appchainID,
 		gopool:        NewGoPool(runtime.GOMAXPROCS(runtime.NumCPU())),
+		ctx:           ctx,
+		cancel:        cancel,
 	}
 
 	return da, nil

internal/adapt/direct_adapter/handler.go

@@ -82,31 +82,37 @@ func (d *DirectAdapter) handleSendIBTPMessage(stream network.Stream, msg *pb.Mes
 				}()
 				d.ibtpC <- ibtp
 				index := ibtp.Index
+				ticker := time.NewTicker(time.Second)
+				defer ticker.Stop()
 				for {
-					pool.lock.Lock()
-					if item := pool.ibtps.Min(); item != nil {
-						if item.(*MyTree).index < index+1 {
-							pool.ibtps.DeleteMin()
-						}
-
-						if item.(*MyTree).index == index+1 {
-							d.ibtpC <- item.(*MyTree).ibtp
-							pool.ibtps.DeleteMin()
-							index++
-							pool.time = time.Now()
-						}
-
-						// By default, the index will be equalized after 5 seconds
-						if time.Now().Sub(pool.time).Seconds() > 5.0 {
-							d.ibtpC <- item.(*MyTree).ibtp
-							pool.ibtps.DeleteMin()
-							index = item.(*MyTree).index
-							pool.time = time.Now()
+					select {
+					case <-ticker.C:
+						pool.lock.Lock()
+						if item := pool.ibtps.Min(); item != nil {
+							if item.(*MyTree).index < index+1 {
+								pool.ibtps.DeleteMin()
+							}
+
+							if item.(*MyTree).index == index+1 {
+								d.ibtpC <- item.(*MyTree).ibtp
+								pool.ibtps.DeleteMin()
+								index++
+								pool.time = time.Now()
+							}
+
+							// By default, the index will be equalized after 5 seconds
+							if time.Now().Sub(pool.time).Seconds() > 5.0 {
+								d.ibtpC <- item.(*MyTree).ibtp
+								pool.ibtps.DeleteMin()
+								index = item.(*MyTree).index
+								pool.time = time.Now()
+							}
 						}
+						pool.lock.Unlock()
+					case <-d.ctx.Done():
+						return
 					}
-					pool.lock.Unlock()
 				}
-
 			}(pool, ibtp)
 		} else {
 			pool.lock.Lock()