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
}
write is used to write to the stream, may return on
a short write.
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
}