func (s *Stream) consumeFrame(h frameHeader, payload []byte) {
	s.cond.L.Lock()
	defer s.cond.L.Unlock()
	if s.err != nil {
		return
	}
	if h.flags&flagFinal != 0 {
		err := ErrPeerClosedStream
		if h.flags&flagError != 0 {
			err = errors.New(string(payload))
		}
		s.err = err
		s.cond.Broadcast()  
		return
	} else if len(payload) == 0 {
		return
	}
	s.readBuf = payload
	s.cond.Broadcast()  
	for len(s.readBuf) != 0 && s.err == nil {
		s.cond.Wait()
	}
}
consumeFrame stores a frame in s.read and waits for it to be consumed by
(*Stream).Read calls.
package mux
import (
	"bytes"
	"crypto/cipher"
	"crypto/ed25519"
	"errors"
	"fmt"
	"io"
	"net"
	"os"
	"sync"
	"syscall"
	"time"
)
var (
	ErrClosedConn       = errors.New("underlying connection was closed")
	ErrClosedStream     = errors.New("stream was gracefully closed")
	ErrPeerClosedStream = errors.New("peer closed stream gracefully")
	ErrPeerClosedConn   = errors.New("peer closed underlying connection")
)
type Mux struct {
	conn     net.Conn
	aead     cipher.AEAD
	settings connSettings
	mu      sync.Mutex
	cond    sync.Cond
	streams map[uint32]*Stream
	nextID  uint32
	err     error  
	write   struct {
		header   frameHeader
		payload  []byte
		timedOut bool
		cond     sync.Cond  
	}
}
func (m *Mux) setErr(err error) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.err != nil {
		return m.err
	}
	if errors.Is(err, io.EOF) ||
		errors.Is(err, syscall.ECONNRESET) ||
		errors.Is(err, syscall.EPIPE) ||
		errors.Is(err, syscall.EPROTOTYPE) {
		err = ErrPeerClosedConn
	}
	m.err = err
	for _, s := range m.streams {
		s.cond.L.Lock()
		s.err = err
		s.cond.Broadcast()
		s.cond.L.Unlock()
	}
	m.conn.Close()
	m.cond.Broadcast()
	m.write.cond.Broadcast()
	return err
}
func (m *Mux) bufferFrame(h frameHeader, payload []byte, deadline time.Time) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	if !deadline.IsZero() {
		if !time.Now().Before(deadline) {
			return os.ErrDeadlineExceeded
		}
		timer := time.AfterFunc(time.Until(deadline), m.write.cond.Broadcast)  
		defer timer.Stop()
	}
	for m.write.header.id != 0 && m.err == nil && (deadline.IsZero() || time.Now().Before(deadline)) {
		m.write.cond.Wait()
	}
	if m.err != nil {
		return m.err
	} else if !deadline.IsZero() && !time.Now().Before(deadline) {
		return os.ErrDeadlineExceeded
	}
	m.write.header = h
	m.write.payload = payload
	m.cond.Broadcast()
	return nil
}
func (m *Mux) writeLoop() {
	keepaliveInterval := m.settings.MaxTimeout - m.settings.MaxTimeout/4
	nextKeepalive := time.Now().Add(keepaliveInterval)
	timer := time.AfterFunc(keepaliveInterval, m.cond.Broadcast)
	defer timer.Stop()
	writeBuf := make([]byte, m.settings.maxFrameSize())
	for {
		m.mu.Lock()
		for m.write.header.id == 0 && m.err == nil && time.Now().Before(nextKeepalive) {
			m.cond.Wait()
		}
		if m.err != nil {
			m.mu.Unlock()
			return
		}
		h, payload := m.write.header, m.write.payload
		if h.id == 0 {
			h, payload = frameHeader{id: idKeepalive}, nil
		}
		frame := encryptFrame(writeBuf, h, payload, m.settings.RequestedPacketSize, m.aead)
		m.mu.Unlock()
		timer.Stop()
		timer.Reset(keepaliveInterval)
		nextKeepalive = time.Now().Add(keepaliveInterval)
		if _, err := m.conn.Write(frame); err != nil {
			m.setErr(err)
			return
		}
		m.mu.Lock()
		m.write.header = frameHeader{}
		m.write.payload = nil
		m.write.cond.Signal()
		m.mu.Unlock()
	}
}
func (m *Mux) readLoop() {
	var curStream *Stream  
	buf := make([]byte, m.settings.maxFrameSize())
	for {
		h, payload, err := readEncryptedFrame(m.conn, buf, m.settings.RequestedPacketSize, m.aead)
		if err != nil {
			m.setErr(err)
			return
		}
		switch h.id {
		case idErrorBadInit, idEstablishEncryption, idUpdateSettings:
			m.setErr(errors.New("peer sent invalid frame ID"))
			return
		case idKeepalive:
			continue  
		default:
			if curStream == nil || h.id != curStream.id {
				m.mu.Lock()
				if curStream = m.streams[h.id]; curStream == nil {
					curStream = &Stream{
						m:        m,
						id:       h.id,
						accepted: false,
						cond:     sync.Cond{L: new(sync.Mutex)},
					}
					m.streams[h.id] = curStream
					m.cond.Broadcast()  
				}
				m.mu.Unlock()
			}
			curStream.consumeFrame(h, payload)
		}
	}
}
func (m *Mux) Close() error {
	err := m.setErr(ErrClosedConn)
	if err == ErrClosedConn || err == ErrPeerClosedConn {
		err = nil
	}
	return err
}
func (m *Mux) AcceptStream() (*Stream, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	for {
		if m.err != nil {
			return nil, m.err
		}
		for _, s := range m.streams {
			if !s.accepted {
				s.accepted = true
				return s, nil
			}
		}
		m.cond.Wait()
	}
}
func (m *Mux) DialStream() (*Stream, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.err != nil {
		return nil, m.err
	}
	s := &Stream{
		m:        m,
		accepted: true,
		cond:     sync.Cond{L: new(sync.Mutex)},
	}
	for m.streams[m.nextID] != nil {
		m.nextID += 2
	}
	s.id = m.nextID
	m.nextID += 2
	m.streams[s.id] = s
	return s, nil
}
func newMux(conn net.Conn, aead cipher.AEAD, settings connSettings) *Mux {
	m := &Mux{
		conn:     conn,
		aead:     aead,
		settings: settings,
		streams:  make(map[uint32]*Stream),
		nextID:   1 << 8,  
	}
	m.cond.L = &m.mu
	m.write.cond.L = &m.mu
	go m.readLoop()
	go m.writeLoop()
	return m
}
func Dial(conn net.Conn, theirKey ed25519.PublicKey) (*Mux, error) {
	if err := initiateVersionHandshake(conn); err != nil {
		return nil, fmt.Errorf("version handshake failed: %w", err)
	}
	aead, err := initiateEncryptionHandshake(conn, theirKey)
	if err != nil {
		return nil, fmt.Errorf("encryption handshake failed: %w", err)
	}
	settings, err := initiateSettingsHandshake(conn, defaultConnSettings, aead)
	if err != nil {
		return nil, fmt.Errorf("settings handshake failed: %w", err)
	}
	return newMux(conn, aead, settings), nil
}
func Accept(conn net.Conn, ourKey ed25519.PrivateKey) (*Mux, error) {
	if err := acceptVersionHandshake(conn); err != nil {
		return nil, fmt.Errorf("version handshake failed: %w", err)
	}
	aead, err := acceptEncryptionHandshake(conn, ourKey)
	if err != nil {
		return nil, fmt.Errorf("encryption handshake failed: %w", err)
	}
	settings, err := acceptSettingsHandshake(conn, defaultConnSettings, aead)
	if err != nil {
		return nil, fmt.Errorf("settings handshake failed: %w", err)
	}
	m := newMux(conn, aead, settings)
	m.nextID++  
	return m, nil
}
type Stream struct {
	m        *Mux
	id       uint32
	accepted bool
	cond    sync.Cond  
	err     error
	readBuf []byte
	rd, wd  time.Time  
}
func (s *Stream) LocalAddr() net.Addr { return s.m.conn.LocalAddr() }
func (s *Stream) RemoteAddr() net.Addr { return s.m.conn.RemoteAddr() }
func (s *Stream) SetDeadline(t time.Time) error {
	s.SetReadDeadline(t)
	s.SetWriteDeadline(t)
	return nil
}
func (s *Stream) SetReadDeadline(t time.Time) error {
	s.cond.L.Lock()
	defer s.cond.L.Unlock()
	s.rd = t
	return nil
}
func (s *Stream) SetWriteDeadline(t time.Time) error {
	s.cond.L.Lock()
	defer s.cond.L.Unlock()
	s.wd = t
	return nil
}