handle error cases on streams

License: MIT
Signed-off-by: Jeromy <why@ipfs.io>

exchange/bitswap/wantmanager.go

@@ -133,14 +133,6 @@ func (pm *WantManager) startPeerHandler(p peer.ID) *msgQueue {
  mq.out = fullwantlist
  mq.work <- struct{}{}
 
- s, err := pm.network.NewMessageSender(pm.ctx, p)
- if err != nil {
- log.Error("error opening stream to peer: ", err)
- return nil
- }
-
- mq.sender = s
-
  pm.peers[p] = mq
  go mq.runQueue(pm.ctx)
  return mq
@@ -163,7 +155,11 @@ func (pm *WantManager) stopPeerHandler(p peer.ID) {
 }
 
 func (mq *msgQueue) runQueue(ctx context.Context) {
- defer mq.sender.Close()
+ defer func() {
+ if mq.sender != nil {
+ mq.sender.Close()
+ }
+ }()
  for {
  select {
  case <-mq.work: // there is work to be done
@@ -180,14 +176,25 @@ func (mq *msgQueue) doWork(ctx context.Context) {
  // allow ten minutes for connections
  // this includes looking them up in the dht
  // dialing them, and handshaking
- conctx, cancel := context.WithTimeout(ctx, time.Minute*10)
- defer cancel()
+ if mq.sender == nil {
+ conctx, cancel := context.WithTimeout(ctx, time.Minute*10)
+ defer cancel()
+
+ err := mq.network.ConnectTo(conctx, mq.p)
+ if err != nil {
+ log.Infof("cant connect to peer %s: %s", mq.p, err)
+ // TODO: cant connect, what now?
+ return
+ }
 
- err := mq.network.ConnectTo(conctx, mq.p)
- if err != nil {
- log.Infof("cant connect to peer %s: %s", mq.p, err)
- // TODO: cant connect, what now?
- return
+ nsender, err := mq.network.NewMessageSender(ctx, mq.p)
+ if err != nil {
+ log.Infof("cant open new stream to peer %s: %s", mq.p, err)
+ // TODO: cant open stream, what now?
+ return
+ }
+
+ mq.sender = nsender
  }
 
  // grab outgoing message
@@ -201,9 +208,11 @@ func (mq *msgQueue) doWork(ctx context.Context) {
  mq.outlk.Unlock()
 
  // send wantlist updates
- err = mq.sender.SendMsg(wlm)
+ err := mq.sender.SendMsg(wlm)
  if err != nil {
  log.Infof("bitswap send error: %s", err)
+ mq.sender.Close()
+ mq.sender = nil
  // TODO: what do we do if this fails?
  return
  }