merge

lib/mux/conn.go

@@ -499,8 +499,16 @@ start:
 
 func (Self *SendWindow) waitReceiveWindow() (err error) {
 	t := Self.timeout.Sub(time.Now())
-	if t < 0 {
+	if t < 0 { // not set the timeout, wait for it as long as connection close
-		t = time.Minute * 5
+		select {
+		case _, ok := <-Self.setSizeCh:
+			if !ok {
+				return errors.New("conn.writeWindow: window closed")
+			}
+			return nil
+		case <-Self.closeOpCh:
+			return errors.New("conn.writeWindow: window closed")
+		}
 	}
 	timer := time.NewTimer(t)
 	defer timer.Stop()

lib/mux/queue.go

@@ -300,8 +300,14 @@ func (Self *ReceiveWindowQueue) waitPush() (err error) {
 	//logs.Warn("wait push")
 	//defer logs.Warn("wait push finish")
 	t := Self.timeout.Sub(time.Now())
-	if t <= 0 {
+	if t <= 0 { // not set the timeout, so wait for it without timeout, just like a tcp connection
-		t = time.Minute * 5
+		select {
+		case <-Self.readOp:
+			return nil
+		case <-Self.stopOp:
+			err = io.EOF
+			return
+		}
 	}
 	timer := time.NewTimer(t)
 	defer timer.Stop()