make sure streamLock is non-blocking due to receiveBytes

session.go

@@ -248,20 +248,18 @@ func (s *Session) recvLoop() {
 				}
 				s.streamLock.Unlock()
 			case cmdPSH:
-				var written int64
-				var err error
 				s.streamLock.Lock()
-				if stream, ok := s.streams[sid]; ok {
+				stream := s.streams[sid]
-					written, err = stream.receiveBytes(s.conn, int64(hdr.Length()))
-					atomic.AddInt32(&s.bucket, -int32(written))
-					stream.notifyReadEvent()
-				}
 				s.streamLock.Unlock()
 
-				// read data error
+				if stream != nil {
-				if err != nil {
+					written, err := stream.receiveBytes(s.conn, int64(hdr.Length()))
-					s.Close()
+					atomic.AddInt32(&s.bucket, -int32(written))
-					return
+					stream.notifyReadEvent()
+					if err != nil {
+						s.Close()
+						return
+					}
 				}
 			default:
 				s.Close()