func (s *Stream) forceClose() {
	s.stateLock.Lock()
	s.state = streamClosed
	s.stateLock.Unlock()
	s.notifyWaiting()
}
forceClose is used for when the session is exiting
package yamux
import (
	"bytes"
	"io"
	"sync"
	"sync/atomic"
	"time"
)
type streamState int
const (
	streamInit streamState = iota
	streamSYNSent
	streamSYNReceived
	streamEstablished
	streamLocalClose
	streamRemoteClose
	streamClosed
	streamReset
)
type Stream struct {
	recvWindow uint32
	sendWindow uint32
	id      uint32
	session *Session
	state     streamState
	stateLock sync.Mutex
	recvBuf  bytes.Buffer
	recvLock sync.Mutex
	controlHdr     header
	controlErr     chan error
	controlHdrLock sync.Mutex
	sendHdr  header
	sendErr  chan error
	sendLock sync.Mutex
	recvNotifyCh chan struct{}
	sendNotifyCh chan struct{}
	readDeadline  time.Time
	writeDeadline time.Time
}
func newStream(session *Session, id uint32, state streamState) *Stream {
	s := &Stream{
		id:           id,
		session:      session,
		state:        state,
		controlHdr:   header(make([]byte, headerSize)),
		controlErr:   make(chan error, 1),
		sendHdr:      header(make([]byte, headerSize)),
		sendErr:      make(chan error, 1),
		recvWindow:   initialStreamWindow,
		sendWindow:   initialStreamWindow,
		recvNotifyCh: make(chan struct{}, 1),
		sendNotifyCh: make(chan struct{}, 1),
	}
	return s
}
func (s *Stream) Session() *Session {
	return s.session
}
func (s *Stream) StreamID() uint32 {
	return s.id
}
func (s *Stream) Read(b []byte) (n int, err error) {
	defer asyncNotify(s.recvNotifyCh)
START:
	s.stateLock.Lock()
	switch s.state {
	case streamRemoteClose:
		fallthrough
	case streamClosed:
		if s.recvBuf.Len() == 0 {
			s.stateLock.Unlock()
			return 0, io.EOF
		}
	case streamReset:
		s.stateLock.Unlock()
		return 0, ErrConnectionReset
	}
	s.stateLock.Unlock()
	s.recvLock.Lock()
	if s.recvBuf.Len() == 0 {
		s.recvLock.Unlock()
		goto WAIT
	}
	n, _ = s.recvBuf.Read(b)
	s.recvLock.Unlock()
	err = s.sendWindowUpdate()
	return n, err
WAIT:
	var timeout <-chan time.Time
	if !s.readDeadline.IsZero() {
		delay := s.readDeadline.Sub(time.Now())
		timeout = time.After(delay)
	}
	select {
	case <-s.recvNotifyCh:
		goto START
	case <-timeout:
		return 0, ErrTimeout
	}
}
func (s *Stream) Write(b []byte) (n int, err error) {
	s.sendLock.Lock()
	defer s.sendLock.Unlock()
	total := 0
	for total < len(b) {
		n, err := s.write(b[total:])
		total += n
		if err != nil {
			return total, err
		}
	}
	return total, nil
}
func (s *Stream) write(b []byte) (n int, err error) {
	var flags uint16
	var max uint32
	var body io.Reader
START:
	s.stateLock.Lock()
	switch s.state {
	case streamLocalClose:
		fallthrough
	case streamClosed:
		s.stateLock.Unlock()
		return 0, ErrStreamClosed
	case streamReset:
		s.stateLock.Unlock()
		return 0, ErrConnectionReset
	}
	s.stateLock.Unlock()
	window := atomic.LoadUint32(&s.sendWindow)
	if window == 0 {
		goto WAIT
	}
	flags = s.sendFlags()
	max = min(window, uint32(len(b)))
	body = bytes.NewReader(b[:max])
	s.sendHdr.encode(typeData, flags, s.id, max)
	if err := s.session.waitForSendErr(s.sendHdr, body, s.sendErr); err != nil {
		return 0, err
	}
	atomic.AddUint32(&s.sendWindow, ^uint32(max-1))
	return int(max), err
WAIT:
	var timeout <-chan time.Time
	if !s.writeDeadline.IsZero() {
		delay := s.writeDeadline.Sub(time.Now())
		timeout = time.After(delay)
	}
	select {
	case <-s.sendNotifyCh:
		goto START
	case <-timeout:
		return 0, ErrTimeout
	}
	return 0, nil
}
func (s *Stream) sendFlags() uint16 {
	s.stateLock.Lock()
	defer s.stateLock.Unlock()
	var flags uint16
	switch s.state {
	case streamInit:
		flags |= flagSYN
		s.state = streamSYNSent
	case streamSYNReceived:
		flags |= flagACK
		s.state = streamEstablished
	}
	return flags
}
func (s *Stream) sendWindowUpdate() error {
	s.controlHdrLock.Lock()
	defer s.controlHdrLock.Unlock()
	max := s.session.config.MaxStreamWindowSize
	delta := max - atomic.LoadUint32(&s.recvWindow)
	flags := s.sendFlags()
	if delta < (max/2) && flags == 0 {
		return nil
	}
	atomic.AddUint32(&s.recvWindow, delta)
	s.controlHdr.encode(typeWindowUpdate, flags, s.id, delta)
	if err := s.session.waitForSendErr(s.controlHdr, nil, s.controlErr); err != nil {
		return err
	}
	return nil
}
func (s *Stream) sendClose() error {
	s.controlHdrLock.Lock()
	defer s.controlHdrLock.Unlock()
	flags := s.sendFlags()
	flags |= flagFIN
	s.controlHdr.encode(typeWindowUpdate, flags, s.id, 0)
	if err := s.session.waitForSendErr(s.controlHdr, nil, s.controlErr); err != nil {
		return err
	}
	return nil
}
func (s *Stream) Close() error {
	s.stateLock.Lock()
	switch s.state {
	case streamSYNSent:
		fallthrough
	case streamSYNReceived:
		fallthrough
	case streamEstablished:
		s.state = streamLocalClose
		goto SEND_CLOSE
	case streamLocalClose:
	case streamRemoteClose:
		s.state = streamClosed
		s.session.closeStream(s.id)
		goto SEND_CLOSE
	case streamClosed:
	case streamReset:
	default:
		panic("unhandled state")
	}
	s.stateLock.Unlock()
	return nil
SEND_CLOSE:
	s.stateLock.Unlock()
	s.sendClose()
	s.notifyWaiting()
	return nil
}