func (s *KCPConn) GetConv() uint32 {
    return s.kcp.conv
}
GetConv gets conversation id of a session
package kcp
import (
    "crypto/rand"
    "encoding/binary"
    "net"
    "sync"
    "sync/atomic"
    "time"
    "errors"
    "log"
    "math"
    "io"
)
type errTimeout struct {
    error
}
func (errTimeout) Timeout() bool   { return true }
func (errTimeout) Temporary() bool { return true }
func (errTimeout) Error() string   { return "i/o timeout" }
const (
    defaultWndSize           = 128  
    udpPacketSizeLimit       = 2048
    rxQueueLimit             = 8192
    defaultKeepAliveInterval = 10
    kcpSendBufferLimit = 1024 * 1024
)
const (
    errBrokenPipe       = "broken pipe"
    errInvalidOperation = "invalid operation"
)
var (
    udpPacketPool sync.Pool
)
func init() {
    udpPacketPool.New = func() interface{} {
        return make([]byte, udpPacketSizeLimit)
    }
}
type (
    KCPConn struct {
        kcp               *KCP            
        listener          *Listener       
        isClient          bool
        conn              net.PacketConn  
        remoteAddr        net.Addr
        deadlineRead      time.Time
        deadlineWrite     time.Time
        bufRead           *Buffer
        die               chan struct{}
        chReadEvent       chan struct{}
        chWriteEvent      chan struct{}
        chKcpFlushEvent   chan struct{}
        chUdpInput        chan []byte
        keepAliveInterval int32
        mu                sync.Mutex
    }
    setReadBuffer interface {
        SetReadBuffer(bytes int) error
    }
    setWriteBuffer interface {
        SetWriteBuffer(bytes int) error
    }
)
func newKCPConn() *KCPConn {
    s := new(KCPConn)
    s.bufRead = new(Buffer)
    s.die = make(chan struct{})
    s.chReadEvent = make(chan struct{}, 1)
    s.chWriteEvent = make(chan struct{}, 1)
    s.chKcpFlushEvent = make(chan struct{}, 1)
    s.chUdpInput = make(chan []byte, rxQueueLimit)
    s.keepAliveInterval = defaultKeepAliveInterval
    s.kcp = newKCP()
    s.kcp.setWndSize(defaultWndSize, defaultWndSize)
    s.kcp.setMtu(IKCP_MTU_DEF)
    s.kcp.stream = 1
    s.kcp.output = s.kcpOutput
    s.kcp.stats = &KcpConnStats{}
    return s
}
func (s *KCPConn) accept(conv uint32, l *Listener, conn net.PacketConn, remote net.Addr) {
    s.listener = l
    s.isClient = false
    s.conn = conn
    s.remoteAddr = remote
    s.kcp.conv = conv
    atomic.AddInt64(&Stats.TotalAccept, 1)
    go s.run();
}
func (s *KCPConn) connect(conv uint32, conn net.PacketConn, remote net.Addr) {
    s.isClient = true
    s.conn = conn
    s.remoteAddr = remote
    s.kcp.conv = conv
    atomic.AddInt64(&Stats.TotalConnect, 1)
    s.goRunClientRecv()
    go s.run();
    s.mu.Lock()
    s.kcp.sendConnectFlush(currentTickMs())
    s.mu.Unlock()
    <- s.chWriteEvent
}
func (s *KCPConn) dump() {
    s.mu.Lock()
    role := "server"
    if s.isClient {
        role = "client"
    }
    log.Printf("-- role=%s (%p) --\n", role, s)
    log.Printf("stats: %+v\n", s.kcp.stats)
    log.Printf("isLocalOpen=%v, isRemoteOpen=%v\n", !s.kcp.isStateLocalClosed(), s.kcp.isRemoteOpen())
    log.Printf("snd_nxt=%d, WaitSnd=%d\n", s.kcp.snd_nxt, s.kcp.waitSnd())
    log.Printf("rcv_nxt=%d, rcv_queue=%d, rcv_buf=%d\n", s.kcp.rcv_nxt, len(s.kcp.rcv_queue), len(s.kcp.rcv_buf))
    log.Println()
    s.mu.Unlock()
}
func (s *KCPConn) debug() {
    loop:
    for {
        select {
        case <-s.die:
            log.Printf("%p stats: %+v\n", s, s.kcp.stats)
            log.Printf("%p close\n", s)
            break loop
        case <-time.After(time.Second):
            s.dump()
        }
    }
}
func (s *KCPConn) Read(b []byte) (int, error) {
    for {
        s.mu.Lock()
        if s.bufRead.Len() < len(b) {
            for {
                n := s.kcp.recvSize();
                if n <= 0 {
                    break
                }
                s.kcp.recv(s.bufRead.Extend(n))
                s.kcp.stats.ByteRx += int64(n)
                atomic.AddInt64(&Stats.ByteRx, int64(n))
            }
        }
        shouldClose := s.kcp.shouldClose()
        if shouldClose {
            s.closeInternal()
        }
        if s.bufRead.Len() > 0 {
            n, err := s.bufRead.Read(b)
            s.mu.Unlock()
            return n, err
        }
        if s.kcp.isStateLocalClosed() {
            s.mu.Unlock()
            return 0, io.EOF
        }
        var timeout *time.Timer
        var deadline <-chan time.Time
        if !s.deadlineRead.IsZero() {
            delay := s.deadlineRead.Sub(time.Now())
            if delay <= 0 {
                s.mu.Unlock()
                return 0, errTimeout{}
            }
            timeout = time.NewTimer(delay)
            deadline = timeout.C
        }
        s.mu.Unlock()
        select {
        case <-s.chReadEvent:
        case <-deadline:
        case <-s.die:
        }
        if timeout != nil {
            timeout.Stop()
        }
    }
}
func (s *KCPConn) canKcpSendInternal() bool {
    return s.kcp.waitSnd() < int(s.kcp.snd_wnd) && s.kcp.isStateConnected()
}
func (s *KCPConn) Write(b []byte) (int, error) {
    for {
        s.mu.Lock()
        if s.kcp.isStateLocalClosed() {
            s.mu.Unlock()
            return 0, errors.New(errBrokenPipe)
        }
        if s.canKcpSendInternal() {
            ret := s.kcp.send(b)
            s.mu.Unlock()
            if ret < 0 {
                log.Panicf("internal bug, kcp send: %d", ret)
            }
            s.kcp.stats.ByteTx += int64(len(b))
            atomic.AddInt64(&Stats.ByteTx, int64(len(b)))
            if s.kcp.sndBufAvail() > 0 {
            s.notifyFlushEvent()
            }
            return len(b), nil
        }
        var timeout *time.Timer
        var deadline <-chan time.Time
        if !s.deadlineWrite.IsZero() {
            delay := s.deadlineWrite.Sub(time.Now())
            if delay <= 0 {
                s.mu.Unlock()
                return 0, errTimeout{}
            }
            timeout = time.NewTimer(delay)
            deadline = timeout.C
        }
        s.mu.Unlock()
        select {
        case <-s.chWriteEvent:
        case <-deadline:
        case <-s.die:
        }
        if timeout != nil {
            timeout.Stop()
        }
    }
}
func (s *KCPConn) closeInternal() error {
    if s.conn == nil || s.kcp.isStateLocalClosed() {
        return errors.New(errBrokenPipe)
    }
    close(s.die)
    s.kcp.sendCloseFlush(currentTickMs())
    return nil
}
func (s *KCPConn) Close() error {
    s.mu.Lock()
    defer s.mu.Unlock()
    return s.closeInternal()
}
func (s *KCPConn) doKcpInput(data []byte) bool {
    s.mu.Lock()
    defer s.mu.Unlock()
    alreadyConnected := s.kcp.isStateConnected()
    if ret := s.kcp.input(currentTickMs(), data, true); ret != 0 {
        s.kcp.stats.ErrorInput += 1
        atomic.AddInt64(&Stats.ErrorInput, 1)
    }
    if !alreadyConnected && s.kcp.isStateConnected(){
        s.kcp.sendConnectFlush(currentTickMs())
        s.notifyWriteEvent()
    }
    n := s.kcp.recvSize()
    if n > 0 || s.kcp.shouldClose() {
        s.notifyReadEvent()
    }
    udpPacketPool.Put(data)
    s.kcp.stats.PacketIn += 1
    s.kcp.stats.ByteIn += int64(len(data))
    atomic.AddInt64(&Stats.PacketIn, 1)
    atomic.AddInt64(&Stats.ByteIn, int64(len(data)))
    return true
}
func (s *KCPConn) goRunClientRecv() {
    go func() {
        for {
            data := udpPacketPool.Get().([]byte)[:udpPacketSizeLimit]
            if n, _, err := s.conn.ReadFrom(data); err == nil {
                select {
                case s.chUdpInput <- data[:n]:
                case <-s.die:
                }
            } else if err != nil {
                log.Printf("client=%v udp read error: %v\n", s.isClient, err)
                s.kcp.stats.ErrorRead += 1
                atomic.AddInt64(&Stats.ErrorRead, 1)
                return
            }
        }
    }()
}
func (s *KCPConn) run() {
    connCurrent := atomic.AddInt64(&Stats.ConnCurrent, 1)
    connMax := atomic.LoadInt64(&Stats.ConnMax)
    if connCurrent > connMax {
        atomic.CompareAndSwapInt64(&Stats.ConnMax, connMax, connCurrent)
    }
    updateDelayMax := 1000 * time.Millisecond
    updateDelayMin := 10 * time.Millisecond
    updateDelay := updateDelayMin
    doKcpFlush := false
    loopMain:
    for s.kcp.isAllOpen() {
        if updateDelay > updateDelayMax {
            updateDelay = updateDelayMax
        }
        if updateDelay < updateDelayMin {
            updateDelay = updateDelayMin
        }
        s.mu.Lock()
        if s.kcp.sndBufAvail() > 0 {
            doKcpFlush = true
        }
        if doKcpFlush {
            s.kcp.flush(currentTickMs())
        } else {
            s.kcp.update(currentTickMs())
        }
        if s.canKcpSendInternal() {
            s.notifyWriteEvent()
        }
        s.mu.Unlock()
        doKcpFlush = false
        select {
        case data := <-s.chUdpInput:
            s.doKcpInput(data)
            updateDelay = updateDelay / 2
        case <-s.chKcpFlushEvent:
            doKcpFlush = true
            updateDelay = updateDelay / 2
        case <-time.After(updateDelay):
            updateDelay = updateDelay * 2
        case <-s.die:
            break loopMain
        }
    }
    atomic.AddInt64(&Stats.ConnClosing, 1)
    if !s.kcp.isStateLocalClosed() {
        s.kcp.sendCloseFlush(currentTickMs())
    }
    closeWaitStartTime := time.Now()
    dangling := true
    loopClose:
    for {
        s.mu.Lock()
        s.kcp.update(currentTickMs())
        isRemoteClosed := s.kcp.isStateRemoteClosed()
        isDead := s.kcp.isStateDead()
        if s.kcp.waitSnd() > 0 || !isRemoteClosed {
            updateDelay = 100 * time.Millisecond
        } else {
            updateDelay = updateDelayMax
            dangling = false
        }
        s.mu.Unlock()
        isCloseWaitTimeout := math.Abs(time.Now().Sub(closeWaitStartTime).Seconds()) >= 5
        if isDead || isCloseWaitTimeout {
            break loopClose
        }
        select {
        case data := <-s.chUdpInput:
            s.doKcpInput(data)
        case <-time.After(updateDelay):
        }
    }
    if s.listener == nil {
        s.conn.Close()
    } else {
        select {
        case s.listener.chDeadConns <- s:
        case <-s.listener.die:
        }
    }
    atomic.AddInt64(&Stats.ConnCurrent, -1)
    atomic.AddInt64(&Stats.ConnClosing, -1)
    atomic.AddInt64(&Stats.TotalClose, 1)
    if dangling {
        atomic.AddInt64(&Stats.TotalCloseDangling, 1)
    }
}
func (s *KCPConn) LocalAddr() net.Addr {
    return s.conn.LocalAddr()
}
func (s *KCPConn) RemoteAddr() net.Addr {
    return s.remoteAddr
}
func (s *KCPConn) SetDeadline(t time.Time) error {
    s.mu.Lock()
    defer s.mu.Unlock()
    s.deadlineRead = t
    s.deadlineWrite = t
    s.notifyReadEvent()
    s.notifyWriteEvent()
    return nil
}
func (s *KCPConn) SetReadDeadline(t time.Time) error {
    s.mu.Lock()
    defer s.mu.Unlock()
    s.deadlineRead = t
    s.notifyReadEvent()
    return nil
}
func (s *KCPConn) SetWriteDeadline(t time.Time) error {
    s.mu.Lock()
    defer s.mu.Unlock()
    s.deadlineWrite = t
    s.notifyWriteEvent()
    return nil
}
func (s *KCPConn) SetWindowSize(sndwnd, rcvwnd int) {
    s.mu.Lock()
    defer s.mu.Unlock()
    s.kcp.setWndSize(sndwnd, rcvwnd)
}
func (s *KCPConn) SetMtu(mtu int) {
    s.mu.Lock()
    defer s.mu.Unlock()
    s.kcp.setMtu(mtu)
}
func (s *KCPConn) SetStreamMode(enable bool) {
    s.mu.Lock()
    defer s.mu.Unlock()
    if enable {
        s.kcp.stream = 1
    } else {
        s.kcp.stream = 0
    }
}
func (s *KCPConn) SetNoDelay(nodelay, interval, resend, nc int) {
    s.mu.Lock()
    defer s.mu.Unlock()
    s.kcp.setNoDelay(nodelay, interval, resend, nc)
}
func (s *KCPConn) SetReadBuffer(bytes int) error {
    s.mu.Lock()
    defer s.mu.Unlock()
    if s.listener == nil {
        if nc, ok := s.conn.(setReadBuffer); ok {
            return nc.SetReadBuffer(bytes)
        }
    }
    return errors.New(errInvalidOperation)
}
func (s *KCPConn) SetWriteBuffer(bytes int) error {
    s.mu.Lock()
    defer s.mu.Unlock()
    if s.listener == nil {
        if nc, ok := s.conn.(setWriteBuffer); ok {
            return nc.SetWriteBuffer(bytes)
        }
    }
    return errors.New(errInvalidOperation)
}
func (s *KCPConn) SetKeepAlive(interval int) {
    atomic.StoreInt32(&s.keepAliveInterval, int32(interval))
}