worker: avoid illegal zero value of channel (#868)

dm/worker/server.go

@@ -353,7 +353,10 @@ func (s *Server) handleSourceBound(ctx context.Context, boundCh chan ha.SourceBo
 		case <-ctx.Done():
 			log.L().Info("worker server is closed, handleSourceBound will quit now")
 			return nil
-		case bound := <-boundCh:
+		case bound, ok := <-boundCh:
+			if !ok {
+				continue
+			}
 			err := s.operateSourceBound(bound)
 			s.setSourceStatus(bound.Source, err, true)
 			if err != nil {

go.mod

@@ -16,7 +16,7 @@ require (
 	github.com/grpc-ecosystem/go-grpc-middleware v1.1.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway v1.14.3
 	github.com/kami-zh/go-capturer v0.0.0-20171211120116-e492ea43421d
-	github.com/lance6716/retool v1.3.7
+	github.com/lance6716/retool v1.3.8-0.20200806070832-3469f70b2afe
 	github.com/pingcap/check v0.0.0-20200212061837-5e12011dc712
 	github.com/pingcap/dumpling v0.0.0-20200605144140-0175843056a6
 	github.com/pingcap/errcode v0.3.0 // indirect

go.sum

@@ -374,8 +374,8 @@ github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/pty v1.1.5/go.mod h1:9r2w37qlBe7rQ6e1fg1S/9xpWHSnaqNdHD3WcMdbPDA=
 github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
-github.com/lance6716/retool v1.3.7 h1:KDdLkDqp/8+Wgnr0OQ6C8Pru8upEAreYay41hyazUw4=
-github.com/lance6716/retool v1.3.7/go.mod h1:9nFHbMjlFhh2msJ6vuHgpz7OM7G9RO0wSLO/w1u2nhM=
+github.com/lance6716/retool v1.3.8-0.20200806070832-3469f70b2afe h1:TzBkezwbsiqkytxItaOYwjtXymmwCRAfoyPGAbPfYGw=
+github.com/lance6716/retool v1.3.8-0.20200806070832-3469f70b2afe/go.mod h1:9nFHbMjlFhh2msJ6vuHgpz7OM7G9RO0wSLO/w1u2nhM=
 github.com/leodido/go-urn v1.1.0/go.mod h1:+cyI34gQWZcE1eQU7NVgKkkzdXDQHr1dBMtdAPozLkw=
 github.com/leodido/go-urn v1.2.0/go.mod h1:+8+nEpDfqqsY+g338gtMEUOtuK+4dEMhiQEgxpxOKII=
 github.com/lib/pq v1.1.1/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=