EQ2/Protocol
1
0
Fork 0
Code Releases Activity

stream

Browse Source
This commit is contained in:
Sky Johnson 2025-09-01 13:56:01 -05:00
parent fd7126878f
commit 1561c74fdf
6 changed files with 1701 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
go.mod
View File

@ -1,3 +1,14 @@
module git.sharkk.net/EQ2/Protocol
go 1.21
require (
github.com/panjf2000/ants/v2 v2.11.3 // indirect
github.com/panjf2000/gnet/v2 v2.9.3 // indirect
github.com/valyala/bytebufferpool v1.0.0 // indirect
go.uber.org/multierr v1.11.0 // indirect
go.uber.org/zap v1.27.0 // indirect
golang.org/x/sync v0.11.0 // indirect
golang.org/x/sys v0.30.0 // indirect
gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
)

16
go.sum Normal file
View File

@ -0,0 +1,16 @@
github.com/panjf2000/ants/v2 v2.11.3 h1:AfI0ngBoXJmYOpDh9m516vjqoUu2sLrIVgppI9TZVpg=
github.com/panjf2000/ants/v2 v2.11.3/go.mod h1:8u92CYMUc6gyvTIw8Ru7Mt7+/ESnJahz5EVtqfrilek=
github.com/panjf2000/gnet/v2 v2.9.3 h1:auV3/A9Na3jiBDmYAAU00rPhFKnsAI+TnI1F7YUJMHQ=
github.com/panjf2000/gnet/v2 v2.9.3/go.mod h1:WQTxDWYuQ/hz3eccH0FN32IVuvZ19HewEWx0l62fx7E=
github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
golang.org/x/sync v0.11.0 h1:GGz8+XQP4FvTTrjZPzNKTMFtSXH80RAzG+5ghFPgK9w=
golang.org/x/sync v0.11.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc=
golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
gopkg.in/natefinch/lumberjack.v2 v2.2.1 h1:bBRl1b0OH9s/DuPhuXpNl+VtCaJXFZ5/uEFST95x9zc=
gopkg.in/natefinch/lumberjack.v2 v2.2.1/go.mod h1:YD8tP3GAjkrDg1eZH7EGmyESg/lsYskCTPBJVb9jqSc=

407
server.go Normal file
View File

@ -0,0 +1,407 @@
package eq2net
import (
"context"
"fmt"
"net"
"sync"
"time"
"github.com/panjf2000/gnet/v2"
)
// ServerConfig contains configuration for the EQ2 server
type ServerConfig struct {
// Network settings
Address string // Listen address (e.g., ":9000")
MaxConnections int // Maximum concurrent connections
ReadBufferSize int // UDP read buffer size
WriteBufferSize int // UDP write buffer size
// Stream settings
StreamConfig *StreamConfig // Default config for new streams
// Performance settings
NumEventLoops int // Number of gnet event loops (0 = NumCPU)
ReusePort bool // Enable SO_REUSEPORT for load balancing
}
// DefaultServerConfig returns a default server configuration
func DefaultServerConfig() *ServerConfig {
return &ServerConfig{
Address: ":9000",
MaxConnections: 10000,
ReadBufferSize: 65536,
WriteBufferSize: 65536,
StreamConfig: DefaultStreamConfig(),
NumEventLoops: 0,
ReusePort: true,
}
}
// EQ2Server implements a gnet-based EverQuest 2 server
type EQ2Server struct {
gnet.BuiltinEventEngine
config *ServerConfig
engine gnet.Engine
engineSet bool // Track if engine has been set
addr net.Addr
// Connection management
streams map[string]*serverStream // Key: remote address string
streamsMu sync.RWMutex
// Callbacks
onNewConnection func(*EQStream)
onConnectionClosed func(*EQStream, string)
// Lifecycle
ctx context.Context
cancel context.CancelFunc
wg sync.WaitGroup
}
// serverStream wraps an EQStream with server-specific data
type serverStream struct {
stream *EQStream
lastActive time.Time
conn gnet.Conn
}
// NewEQ2Server creates a new EQ2 server
func NewEQ2Server(config *ServerConfig) *EQ2Server {
if config == nil {
config = DefaultServerConfig()
}
ctx, cancel := context.WithCancel(context.Background())
return &EQ2Server{
config: config,
streams: make(map[string]*serverStream),
ctx: ctx,
cancel: cancel,
}
}
// Start begins listening for connections
func (s *EQ2Server) Start() error {
// Configure gnet options
opts := []gnet.Option{
gnet.WithMulticore(true),
gnet.WithReusePort(s.config.ReusePort),
gnet.WithSocketRecvBuffer(s.config.ReadBufferSize),
gnet.WithSocketSendBuffer(s.config.WriteBufferSize),
gnet.WithTicker(true),
}
if s.config.NumEventLoops > 0 {
opts = append(opts, gnet.WithNumEventLoop(s.config.NumEventLoops))
}
// Start cleanup worker
s.wg.Add(1)
go s.cleanupWorker()
// Start gnet server
return gnet.Run(s, "udp://"+s.config.Address, opts...)
}
// Stop gracefully shuts down the server
func (s *EQ2Server) Stop() error {
// Signal shutdown
s.cancel()
// Close all streams
s.streamsMu.Lock()
for _, ss := range s.streams {
ss.stream.Close()
}
s.streamsMu.Unlock()
// Wait for cleanup
done := make(chan struct{})
go func() {
s.wg.Wait()
close(done)
}()
select {
case <-done:
case <-time.After(10 * time.Second):
// Force shutdown after timeout
}
// Stop gnet engine
if s.engineSet {
return s.engine.Stop(s.ctx)
}
return nil
}
// gnet event handlers
// OnBoot is called when the server starts
func (s *EQ2Server) OnBoot(eng gnet.Engine) (action gnet.Action) {
s.engine = eng
s.engineSet = true
// Parse and store the address
addr, err := net.ResolveUDPAddr("udp", s.config.Address)
if err == nil {
s.addr = addr
}
fmt.Printf("EQ2 server started on %s\n", s.config.Address)
return gnet.None
}
// OnShutdown is called when the server stops
func (s *EQ2Server) OnShutdown(eng gnet.Engine) {
fmt.Println("EQ2 server shutting down")
}
// OnTraffic handles incoming UDP packets
func (s *EQ2Server) OnTraffic(c gnet.Conn) (action gnet.Action) {
// Read the packet
buf, err := c.Next(-1)
if err != nil {
return gnet.None
}
// Get remote address
remoteAddr := c.RemoteAddr()
if remoteAddr == nil {
return gnet.None
}
addrStr := remoteAddr.String()
// Look up or create stream
s.streamsMu.RLock()
ss, exists := s.streams[addrStr]
s.streamsMu.RUnlock()
if !exists {
// Check for session request
if len(buf) >= 2 {
opcode := uint16(buf[0])<<8 | uint16(buf[1])
if opcode == OPSessionRequest {
// Create new stream
ss = s.createStream(c, remoteAddr)
if ss == nil {
return gnet.None
}
} else {
// Not a session request, send out-of-session
s.sendOutOfSession(c, remoteAddr)
return gnet.None
}
} else {
return gnet.None
}
}
// Update last activity
ss.lastActive = time.Now()
// Process packet in stream
ss.stream.handleIncomingPacket(buf)
return gnet.None
}
// OnTick is called periodically
func (s *EQ2Server) OnTick() (delay time.Duration, action gnet.Action) {
// Tick interval for maintenance tasks
return 100 * time.Millisecond, gnet.None
}
// createStream creates a new stream for a client
func (s *EQ2Server) createStream(c gnet.Conn, remoteAddr net.Addr) *serverStream {
// Check connection limit
s.streamsMu.Lock()
defer s.streamsMu.Unlock()
if len(s.streams) >= s.config.MaxConnections {
return nil
}
// Create stream config (copy from default)
streamConfig := *s.config.StreamConfig
// Create new stream
stream := NewEQStream(&streamConfig)
// Set up callbacks
stream.SetCallbacks(
func() {
// On connect
if s.onNewConnection != nil {
s.onNewConnection(stream)
}
},
func(reason string) {
// On disconnect
s.removeStream(remoteAddr.String())
if s.onConnectionClosed != nil {
s.onConnectionClosed(stream, reason)
}
},
nil, // Error handler
)
// Create server stream wrapper
ss := &serverStream{
stream: stream,
lastActive: time.Now(),
conn: c,
}
// Store in map
s.streams[remoteAddr.String()] = ss
// Create a custom PacketConn wrapper for this stream
packetConn := &gnetPacketConn{
conn: c,
localAddr: s.addr,
remoteAddr: remoteAddr,
server: s,
}
// Connect the stream (in server mode, this just sets up the connection)
go func() {
if err := stream.Connect(packetConn, remoteAddr); err != nil {
s.removeStream(remoteAddr.String())
}
}()
return ss
}
// removeStream removes a stream from the server
func (s *EQ2Server) removeStream(addrStr string) {
s.streamsMu.Lock()
defer s.streamsMu.Unlock()
if ss, exists := s.streams[addrStr]; exists {
ss.stream.Close()
delete(s.streams, addrStr)
}
}
// sendOutOfSession sends an out-of-session packet
func (s *EQ2Server) sendOutOfSession(c gnet.Conn, remoteAddr net.Addr) {
packet := NewEQProtocolPacket(OPOutOfSession, nil)
data := packet.Serialize(0)
c.AsyncWrite(data, nil)
}
// cleanupWorker periodically cleans up inactive connections
func (s *EQ2Server) cleanupWorker() {
defer s.wg.Done()
ticker := time.NewTicker(30 * time.Second)
defer ticker.Stop()
for {
select {
case <-s.ctx.Done():
return
case <-ticker.C:
s.cleanupInactiveStreams()
}
}
}
// cleanupInactiveStreams removes streams that have been inactive too long
func (s *EQ2Server) cleanupInactiveStreams() {
timeout := 5 * time.Minute
now := time.Now()
s.streamsMu.Lock()
defer s.streamsMu.Unlock()
for addr, ss := range s.streams {
if now.Sub(ss.lastActive) > timeout {
ss.stream.Close()
delete(s.streams, addr)
}
}
}
// SetCallbacks sets server event callbacks
func (s *EQ2Server) SetCallbacks(onNew func(*EQStream), onClosed func(*EQStream, string)) {
s.onNewConnection = onNew
s.onConnectionClosed = onClosed
}
// GetStream returns the stream for a given address
func (s *EQ2Server) GetStream(addr string) *EQStream {
s.streamsMu.RLock()
defer s.streamsMu.RUnlock()
if ss, exists := s.streams[addr]; exists {
return ss.stream
}
return nil
}
// GetAllStreams returns all active streams
func (s *EQ2Server) GetAllStreams() []*EQStream {
s.streamsMu.RLock()
defer s.streamsMu.RUnlock()
streams := make([]*EQStream, 0, len(s.streams))
for _, ss := range s.streams {
streams = append(streams, ss.stream)
}
return streams
}
// gnetPacketConn implements net.PacketConn for gnet connections
type gnetPacketConn struct {
conn gnet.Conn
localAddr net.Addr
remoteAddr net.Addr
server *EQ2Server
}
func (g *gnetPacketConn) ReadFrom(p []byte) (n int, addr net.Addr, err error) {
// This is handled by OnTraffic, not used in server mode
return 0, nil, fmt.Errorf("not implemented for server mode")
}
func (g *gnetPacketConn) WriteTo(p []byte, addr net.Addr) (n int, err error) {
// Write to the gnet connection
err = g.conn.AsyncWrite(p, nil)
if err != nil {
return 0, err
}
return len(p), nil
}
func (g *gnetPacketConn) Close() error {
// Connection lifecycle is managed by server
return nil
}
func (g *gnetPacketConn) LocalAddr() net.Addr {
return g.localAddr
}
func (g *gnetPacketConn) SetDeadline(t time.Time) error {
// Not implemented for UDP
return nil
}
func (g *gnetPacketConn) SetReadDeadline(t time.Time) error {
// Not implemented for UDP
return nil
}
func (g *gnetPacketConn) SetWriteDeadline(t time.Time) error {
// Not implemented for UDP
return nil
}

567
stream.go Normal file
View File

@ -0,0 +1,567 @@
package eq2net
import (
"context"
"encoding/binary"
"fmt"
"net"
"sync"
"sync/atomic"
"time"
)
// StreamState represents the state of an EQStream connection
type StreamState int32
const (
StreamStateDisconnected StreamState = iota
StreamStateConnecting
StreamStateConnected
StreamStateDisconnecting
StreamStateClosed
)
// StreamConfig contains configuration for an EQStream
type StreamConfig struct {
// Network settings
MaxPacketSize int // Maximum packet size (default: 512)
WindowSize uint16 // Sliding window size for flow control (default: 2048)
RetransmitTimeMs int64 // Initial retransmit time in milliseconds (default: 500)
MaxRetransmits int // Maximum retransmission attempts (default: 5)
ConnectTimeout time.Duration // Connection timeout (default: 30s)
KeepAliveTime time.Duration // Keep-alive interval (default: 5s)
// Session settings
SessionID uint32 // Session identifier
MaxBandwidth uint32 // Maximum bandwidth in bytes/sec (0 = unlimited)
CRCKey uint32 // CRC key for packet validation
EncodeKey uint32 // Encryption key for chat packets
DecodeKey uint32 // Decryption key for chat packets
CompressEnable bool // Enable packet compression
// Performance settings
SendBufferSize int // Size of send buffer (default: 1024)
RecvBufferSize int // Size of receive buffer (default: 1024)
}
// DefaultStreamConfig returns a default configuration
func DefaultStreamConfig() *StreamConfig {
return &StreamConfig{
MaxPacketSize: 512,
WindowSize: 2048,
RetransmitTimeMs: 500,
MaxRetransmits: 5,
ConnectTimeout: 30 * time.Second,
KeepAliveTime: 5 * time.Second,
SendBufferSize: 1024,
RecvBufferSize: 1024,
CompressEnable: true,
}
}
// StreamStats tracks stream statistics
type StreamStats struct {
PacketsSent atomic.Uint64
PacketsReceived atomic.Uint64
BytesSent atomic.Uint64
BytesReceived atomic.Uint64
PacketsDropped atomic.Uint64
Retransmits atomic.Uint64
RTT atomic.Int64 // Round-trip time in microseconds
Bandwidth atomic.Uint64
}
// EQStream implements reliable UDP communication for EQ2
type EQStream struct {
// Configuration
config *StreamConfig
// Network
conn net.PacketConn
remoteAddr net.Addr
localAddr net.Addr
// State management
state atomic.Int32 // StreamState
sessionID uint32
nextSeqOut atomic.Uint32
nextSeqIn atomic.Uint32
lastAckSeq atomic.Uint32
// Packet queues - using channels for lock-free operations
sendQueue chan *EQProtocolPacket
recvQueue chan *EQApplicationPacket
ackQueue chan uint32
resendQueue chan *EQProtocolPacket
fragmentQueue map[uint32][]*EQProtocolPacket // Fragments being assembled
// Sliding window for flow control
sendWindow map[uint32]*sendPacket
sendWindowMu sync.RWMutex
recvWindow map[uint32]*EQProtocolPacket
recvWindowMu sync.RWMutex
// Retransmission management
rtt atomic.Int64 // Smoothed RTT in microseconds
rttVar atomic.Int64 // RTT variance
rto atomic.Int64 // Retransmission timeout
// Statistics
stats *StreamStats
// Lifecycle management
ctx context.Context
cancel context.CancelFunc
wg sync.WaitGroup
// Callbacks
onConnect func()
onDisconnect func(reason string)
onError func(error)
}
// sendPacket tracks packets awaiting acknowledgment
type sendPacket struct {
packet *EQProtocolPacket
sentTime time.Time
attempts int
nextRetry time.Time
}
// NewEQStream creates a new EQ2 stream
func NewEQStream(config *StreamConfig) *EQStream {
if config == nil {
config = DefaultStreamConfig()
}
ctx, cancel := context.WithCancel(context.Background())
s := &EQStream{
config: config,
sendQueue: make(chan *EQProtocolPacket, config.SendBufferSize),
recvQueue: make(chan *EQApplicationPacket, config.RecvBufferSize),
ackQueue: make(chan uint32, 256),
resendQueue: make(chan *EQProtocolPacket, 256),
fragmentQueue: make(map[uint32][]*EQProtocolPacket),
sendWindow: make(map[uint32]*sendPacket),
recvWindow: make(map[uint32]*EQProtocolPacket),
stats: &StreamStats{},
ctx: ctx,
cancel: cancel,
}
// Initialize state
s.state.Store(int32(StreamStateDisconnected))
s.sessionID = config.SessionID
// Set initial RTO
s.rto.Store(config.RetransmitTimeMs * 1000) // Convert to microseconds
return s
}
// Connect establishes a connection to the remote endpoint
func (s *EQStream) Connect(conn net.PacketConn, remoteAddr net.Addr) error {
// Check state
if !s.compareAndSwapState(StreamStateDisconnected, StreamStateConnecting) {
return fmt.Errorf("stream not in disconnected state")
}
s.conn = conn
s.remoteAddr = remoteAddr
s.localAddr = conn.LocalAddr()
// Start workers
s.wg.Add(4)
go s.sendWorker()
go s.recvWorker()
go s.retransmitWorker()
go s.keepAliveWorker()
// Send session request
sessionReq := s.createSessionRequest()
if err := s.sendPacket(sessionReq); err != nil {
s.Close()
return fmt.Errorf("failed to send session request: %w", err)
}
// Wait for connection or timeout
timer := time.NewTimer(s.config.ConnectTimeout)
defer timer.Stop()
for {
select {
case <-timer.C:
s.Close()
return fmt.Errorf("connection timeout")
case <-s.ctx.Done():
return fmt.Errorf("connection cancelled")
default:
if s.GetState() == StreamStateConnected {
if s.onConnect != nil {
s.onConnect()
}
return nil
}
time.Sleep(10 * time.Millisecond)
}
}
}
// Send queues an application packet for transmission
func (s *EQStream) Send(packet *EQApplicationPacket) error {
if s.GetState() != StreamStateConnected {
return fmt.Errorf("stream not connected")
}
// Convert to protocol packet
protoPacket := s.applicationToProtocol(packet)
select {
case s.sendQueue <- protoPacket:
return nil
case <-s.ctx.Done():
return fmt.Errorf("stream closed")
default:
return fmt.Errorf("send queue full")
}
}
// Receive gets the next application packet from the receive queue
func (s *EQStream) Receive() (*EQApplicationPacket, error) {
select {
case packet := <-s.recvQueue:
return packet, nil
case <-s.ctx.Done():
return nil, fmt.Errorf("stream closed")
default:
return nil, nil // Non-blocking
}
}
// sendWorker handles outgoing packets
func (s *EQStream) sendWorker() {
defer s.wg.Done()
combiner := NewPacketCombiner(s.config.MaxPacketSize - 10) // Leave room for headers/CRC
combineTimer := time.NewTicker(1 * time.Millisecond)
defer combineTimer.Stop()
var pendingPackets []*EQProtocolPacket
for {
select {
case <-s.ctx.Done():
return
case packet := <-s.sendQueue:
// Add sequence number if needed
if s.needsSequence(packet.Opcode) {
packet.Sequence = s.nextSeqOut.Add(1)
// Track in send window
s.sendWindowMu.Lock()
s.sendWindow[packet.Sequence] = &sendPacket{
packet: packet.Copy(),
sentTime: time.Now(),
attempts: 1,
nextRetry: time.Now().Add(time.Duration(s.rto.Load()) * time.Microsecond),
}
s.sendWindowMu.Unlock()
}
pendingPackets = append(pendingPackets, packet)
case packet := <-s.resendQueue:
// Priority resend
s.sendPacketNow(packet)
case <-combineTimer.C:
// Send any pending combined packets
if len(pendingPackets) > 0 {
s.sendCombined(pendingPackets, combiner)
pendingPackets = nil
}
}
// Try to combine and send if we have enough packets
if len(pendingPackets) >= 3 {
s.sendCombined(pendingPackets, combiner)
pendingPackets = nil
}
}
}
// recvWorker handles incoming packets from the network
func (s *EQStream) recvWorker() {
defer s.wg.Done()
buffer := make([]byte, s.config.MaxPacketSize)
for {
select {
case <-s.ctx.Done():
return
default:
}
// Read from network with timeout
s.conn.SetReadDeadline(time.Now().Add(100 * time.Millisecond))
n, addr, err := s.conn.ReadFrom(buffer)
if err != nil {
if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
continue
}
if s.onError != nil {
s.onError(err)
}
continue
}
// Verify source address
if addr.String() != s.remoteAddr.String() {
continue // Ignore packets from other sources
}
// Process packet
s.handleIncomingPacket(buffer[:n])
}
}
// retransmitWorker handles packet retransmissions
func (s *EQStream) retransmitWorker() {
defer s.wg.Done()
ticker := time.NewTicker(10 * time.Millisecond)
defer ticker.Stop()
for {
select {
case <-s.ctx.Done():
return
case <-ticker.C:
now := time.Now()
s.sendWindowMu.Lock()
for seq, sp := range s.sendWindow {
if now.After(sp.nextRetry) {
if sp.attempts >= s.config.MaxRetransmits {
// Max retransmits reached, connection is dead
delete(s.sendWindow, seq)
s.stats.PacketsDropped.Add(1)
continue
}
// Retransmit
sp.attempts++
sp.nextRetry = now.Add(time.Duration(s.rto.Load()) * time.Microsecond * time.Duration(sp.attempts))
s.stats.Retransmits.Add(1)
// Queue for immediate send
select {
case s.resendQueue <- sp.packet:
default:
// Resend queue full, try next time
}
}
}
s.sendWindowMu.Unlock()
}
}
}
// keepAliveWorker sends periodic keep-alive packets
func (s *EQStream) keepAliveWorker() {
defer s.wg.Done()
ticker := time.NewTicker(s.config.KeepAliveTime)
defer ticker.Stop()
for {
select {
case <-s.ctx.Done():
return
case <-ticker.C:
if s.GetState() == StreamStateConnected {
keepAlive := NewEQProtocolPacket(OPKeepAlive, nil)
select {
case s.sendQueue <- keepAlive:
default:
// Queue full, skip this keep-alive
}
}
}
}
}
// Helper methods
func (s *EQStream) GetState() StreamState {
return StreamState(s.state.Load())
}
func (s *EQStream) compareAndSwapState(old, new StreamState) bool {
return s.state.CompareAndSwap(int32(old), int32(new))
}
func (s *EQStream) createSessionRequest() *EQProtocolPacket {
data := make([]byte, 10)
binary.BigEndian.PutUint32(data[0:4], 2) // Protocol version
binary.BigEndian.PutUint32(data[4:8], s.sessionID)
binary.BigEndian.PutUint16(data[8:10], s.config.WindowSize)
return NewEQProtocolPacket(OPSessionRequest, data)
}
func (s *EQStream) needsSequence(opcode uint16) bool {
switch opcode {
case OPPacket, OPFragment:
return true
default:
return false
}
}
func (s *EQStream) sendPacket(packet *EQProtocolPacket) error {
// Check if connection is established
if s.conn == nil {
return fmt.Errorf("no connection")
}
// Prepare packet data
data := packet.Serialize(0)
// Add CRC if not exempt
if !s.isCRCExempt(packet.Opcode) {
data = AppendCRC(data, s.config.CRCKey)
}
// Send to network
n, err := s.conn.WriteTo(data, s.remoteAddr)
if err != nil {
return err
}
// Update statistics
s.stats.PacketsSent.Add(1)
s.stats.BytesSent.Add(uint64(n))
return nil
}
func (s *EQStream) sendPacketNow(packet *EQProtocolPacket) {
if err := s.sendPacket(packet); err != nil && s.onError != nil {
s.onError(err)
}
}
func (s *EQStream) sendCombined(packets []*EQProtocolPacket, combiner *PacketCombiner) {
if len(packets) == 1 {
s.sendPacketNow(packets[0])
return
}
combined := combiner.CombineProtocolPackets(packets)
if combined != nil {
s.sendPacketNow(combined)
} else {
// Couldn't combine, send individually
for _, p := range packets {
s.sendPacketNow(p)
}
}
}
func (s *EQStream) isCRCExempt(opcode uint16) bool {
switch opcode {
case OPSessionRequest, OPSessionResponse, OPOutOfSession:
return true
default:
return false
}
}
func (s *EQStream) applicationToProtocol(app *EQApplicationPacket) *EQProtocolPacket {
// Serialize application packet
data := app.Serialize()
// Create protocol packet
proto := NewEQProtocolPacket(OPPacket, data)
proto.CopyInfo(app.EQPacket)
// Apply compression if enabled
if s.config.CompressEnable && len(data) > CompressionThreshold {
if compressed, err := CompressPacket(data); err == nil {
proto.Buffer = compressed
proto.Size = uint32(len(compressed))
proto.Compressed = true
}
}
return proto
}
// handleIncomingPacket is implemented in stream_packet_handler.go
// Close gracefully shuts down the stream
func (s *EQStream) Close() error {
if !s.compareAndSwapState(StreamStateConnected, StreamStateDisconnecting) &&
!s.compareAndSwapState(StreamStateConnecting, StreamStateDisconnecting) {
return nil // Already closing or closed
}
// Send disconnect packet if we have a connection
if s.conn != nil {
disconnect := NewEQProtocolPacket(OPSessionDisconnect, nil)
s.sendPacketNow(disconnect)
}
// Cancel context to stop workers
s.cancel()
// Wait for workers to finish
done := make(chan struct{})
go func() {
s.wg.Wait()
close(done)
}()
select {
case <-done:
case <-time.After(5 * time.Second):
// Force close after timeout
}
s.state.Store(int32(StreamStateClosed))
if s.onDisconnect != nil {
s.onDisconnect("closed")
}
return nil
}
// GetStats returns a copy of the current statistics
func (s *EQStream) GetStats() StreamStats {
return StreamStats{
PacketsSent: atomic.Uint64{},
PacketsReceived: atomic.Uint64{},
BytesSent: atomic.Uint64{},
BytesReceived: atomic.Uint64{},
PacketsDropped: atomic.Uint64{},
Retransmits: atomic.Uint64{},
RTT: atomic.Int64{},
Bandwidth: atomic.Uint64{},
}
}
// SetCallbacks sets the stream event callbacks
func (s *EQStream) SetCallbacks(onConnect func(), onDisconnect func(string), onError func(error)) {
s.onConnect = onConnect
s.onDisconnect = onDisconnect
s.onError = onError
}

506
stream_packet_handler.go Normal file
View File

@ -0,0 +1,506 @@
package eq2net
import (
"encoding/binary"
"time"
)
// handleIncomingPacket processes incoming network packets
func (s *EQStream) handleIncomingPacket(data []byte) {
// Update statistics
s.stats.PacketsReceived.Add(1)
s.stats.BytesReceived.Add(uint64(len(data)))
// Validate minimum size
if len(data) < 2 {
return
}
// Extract opcode
opcode := binary.BigEndian.Uint16(data[0:2])
// Check CRC for non-exempt packets
if !s.isCRCExempt(opcode) {
if !ValidateCRC(data, s.config.CRCKey) {
s.stats.PacketsDropped.Add(1)
return
}
// Remove CRC bytes for further processing
if len(data) > 2 {
data = data[:len(data)-2]
}
}
// Create protocol packet
packet, err := NewEQProtocolPacketFromBuffer(data, -1)
if err != nil {
return
}
// Handle based on opcode
switch packet.Opcode {
case OPSessionRequest:
s.handleSessionRequest(packet)
case OPSessionResponse:
s.handleSessionResponse(packet)
case OPSessionDisconnect:
s.handleDisconnect(packet)
case OPKeepAlive:
s.handleKeepAlive(packet)
case OPAck:
s.handleAck(packet)
case OPOutOfOrderAck:
s.handleOutOfOrderAck(packet)
case OPPacket:
s.handleDataPacket(packet)
case OPFragment:
s.handleFragment(packet)
case OPCombined:
s.handleCombined(packet)
case OPOutOfSession:
s.handleOutOfSession(packet)
default:
// Unknown opcode, ignore
}
}
// handleSessionRequest processes incoming session requests
func (s *EQStream) handleSessionRequest(packet *EQProtocolPacket) {
if s.GetState() != StreamStateDisconnected {
// We're not accepting new connections
s.sendSessionResponse(false)
return
}
// Parse request
if len(packet.Buffer) < 10 {
return
}
version := binary.BigEndian.Uint32(packet.Buffer[0:4])
sessionID := binary.BigEndian.Uint32(packet.Buffer[4:8])
maxLength := binary.BigEndian.Uint16(packet.Buffer[8:10])
// Validate version
if version != 2 {
s.sendSessionResponse(false)
return
}
// Update session info
s.sessionID = sessionID
if int(maxLength) < s.config.MaxPacketSize {
s.config.MaxPacketSize = int(maxLength)
}
// Accept connection
s.state.Store(int32(StreamStateConnected))
s.sendSessionResponse(true)
if s.onConnect != nil {
s.onConnect()
}
}
// handleSessionResponse processes session response packets
func (s *EQStream) handleSessionResponse(packet *EQProtocolPacket) {
if s.GetState() != StreamStateConnecting {
return
}
// Parse response
if len(packet.Buffer) < 11 {
return
}
sessionID := binary.BigEndian.Uint32(packet.Buffer[0:4])
crcKey := binary.BigEndian.Uint32(packet.Buffer[4:8])
validation := packet.Buffer[8]
format := packet.Buffer[9]
unknownByte := packet.Buffer[10]
// Check if accepted
if validation == 0 {
// Connection rejected
s.state.Store(int32(StreamStateDisconnected))
if s.onDisconnect != nil {
s.onDisconnect("connection rejected")
}
return
}
// Update session info
s.sessionID = sessionID
s.config.CRCKey = crcKey
_ = format // Store for later use if needed
_ = unknownByte
// Connection established
s.state.Store(int32(StreamStateConnected))
}
// handleDisconnect processes disconnect packets
func (s *EQStream) handleDisconnect(packet *EQProtocolPacket) {
s.state.Store(int32(StreamStateDisconnecting))
// Send acknowledgment
ack := NewEQProtocolPacket(OPSessionDisconnect, nil)
s.sendPacketNow(ack)
// Clean shutdown
if s.onDisconnect != nil {
s.onDisconnect("remote disconnect")
}
s.Close()
}
// handleKeepAlive processes keep-alive packets
func (s *EQStream) handleKeepAlive(packet *EQProtocolPacket) {
// Keep-alives don't require a response, just update last activity time
// This helps detect dead connections
}
// handleAck processes acknowledgment packets
func (s *EQStream) handleAck(packet *EQProtocolPacket) {
if len(packet.Buffer) < 2 {
return
}
ackSeq := binary.BigEndian.Uint16(packet.Buffer[0:2])
s.processAck(uint32(ackSeq))
}
// handleOutOfOrderAck processes out-of-order acknowledgments
func (s *EQStream) handleOutOfOrderAck(packet *EQProtocolPacket) {
if len(packet.Buffer) < 2 {
return
}
ackSeq := binary.BigEndian.Uint16(packet.Buffer[0:2])
s.processAck(uint32(ackSeq))
}
// processAck handles acknowledgment of a sent packet
func (s *EQStream) processAck(seq uint32) {
s.sendWindowMu.Lock()
defer s.sendWindowMu.Unlock()
if sp, exists := s.sendWindow[seq]; exists {
// Calculate RTT and update estimates
rtt := time.Since(sp.sentTime).Microseconds()
s.updateRTT(rtt)
// Remove from send window
delete(s.sendWindow, seq)
// Update last acknowledged sequence
if seq > s.lastAckSeq.Load() {
s.lastAckSeq.Store(seq)
}
}
}
// updateRTT updates the RTT estimates using Jacobson/Karels algorithm
func (s *EQStream) updateRTT(sampleRTT int64) {
// First sample
if s.rtt.Load() == 0 {
s.rtt.Store(sampleRTT)
s.rttVar.Store(sampleRTT / 2)
s.rto.Store(sampleRTT + 4*s.rttVar.Load())
return
}
// Subsequent samples (RFC 6298)
alpha := int64(125) // 1/8 in fixed point (multiply by 1000)
beta := int64(250) // 1/4 in fixed point
// SRTT = (1-alpha) * SRTT + alpha * RTT
srtt := s.rtt.Load()
srtt = ((1000-alpha)*srtt + alpha*sampleRTT) / 1000
// RTTVAR = (1-beta) * RTTVAR + beta * |SRTT - RTT|
var diff int64
if srtt > sampleRTT {
diff = srtt - sampleRTT
} else {
diff = sampleRTT - srtt
}
rttVar := s.rttVar.Load()
rttVar = ((1000-beta)*rttVar + beta*diff) / 1000
// RTO = SRTT + 4 * RTTVAR
rto := srtt + 4*rttVar
// Minimum RTO of 200ms, maximum of 60s
if rto < 200000 {
rto = 200000
} else if rto > 60000000 {
rto = 60000000
}
s.rtt.Store(srtt)
s.rttVar.Store(rttVar)
s.rto.Store(rto)
s.stats.RTT.Store(srtt)
}
// handleDataPacket processes regular data packets
func (s *EQStream) handleDataPacket(packet *EQProtocolPacket) {
// Extract sequence number (first 2 bytes after opcode)
if len(packet.Buffer) < 2 {
return
}
seq := uint32(binary.BigEndian.Uint16(packet.Buffer[0:2]))
// Send acknowledgment
s.sendAck(seq)
// Check if it's in order
expectedSeq := s.nextSeqIn.Load()
if seq == expectedSeq {
// In order, process immediately
s.nextSeqIn.Add(1)
s.processDataPacket(packet.Buffer[2:])
// Check if we have any queued packets that can now be processed
s.processQueuedPackets()
} else if seq > expectedSeq {
// Future packet, queue it
s.recvWindowMu.Lock()
s.recvWindow[seq] = packet
s.recvWindowMu.Unlock()
// Send out-of-order ACK
s.sendOutOfOrderAck(seq)
}
// If seq < expectedSeq, it's a duplicate, ignore (we already sent ACK)
}
// processQueuedPackets processes any queued packets that are now in order
func (s *EQStream) processQueuedPackets() {
for {
expectedSeq := s.nextSeqIn.Load()
s.recvWindowMu.Lock()
packet, exists := s.recvWindow[expectedSeq]
if !exists {
s.recvWindowMu.Unlock()
break
}
delete(s.recvWindow, expectedSeq)
s.recvWindowMu.Unlock()
s.nextSeqIn.Add(1)
if len(packet.Buffer) > 2 {
s.processDataPacket(packet.Buffer[2:])
}
}
}
// processDataPacket processes the data portion of a packet
func (s *EQStream) processDataPacket(data []byte) {
// Decompress if needed
if IsCompressed(data) {
decompressed, err := DecompressPacket(data)
if err != nil {
return
}
data = decompressed
}
// Decrypt chat if needed (check for chat opcodes)
// This would need opcode inspection
// Convert to application packet
if len(data) < 2 {
return
}
app := &EQApplicationPacket{
EQPacket: NewEQPacket(binary.BigEndian.Uint16(data[0:2]), nil),
}
if len(data) > 2 {
app.Buffer = make([]byte, len(data)-2)
copy(app.Buffer, data[2:])
app.Size = uint32(len(app.Buffer))
}
// Queue for application
select {
case s.recvQueue <- app:
default:
// Receive queue full, drop packet
s.stats.PacketsDropped.Add(1)
}
}
// handleFragment processes fragmented packets
func (s *EQStream) handleFragment(packet *EQProtocolPacket) {
if len(packet.Buffer) < 6 {
return
}
// Parse fragment header
seq := uint32(binary.BigEndian.Uint16(packet.Buffer[0:2]))
totalSize := binary.BigEndian.Uint32(packet.Buffer[2:6])
// Send acknowledgment
s.sendAck(seq)
// Store fragment
s.recvWindowMu.Lock()
s.fragmentQueue[seq] = append(s.fragmentQueue[seq], packet)
// Check if we have all fragments
currentSize := uint32(0)
for _, frag := range s.fragmentQueue[seq] {
if len(frag.Buffer) > 6 {
currentSize += uint32(len(frag.Buffer) - 6)
}
}
if currentSize >= totalSize {
// Reassemble packet
reassembled := make([]byte, 0, totalSize)
for _, frag := range s.fragmentQueue[seq] {
if len(frag.Buffer) > 6 {
reassembled = append(reassembled, frag.Buffer[6:]...)
}
}
// Clean up fragment queue
delete(s.fragmentQueue, seq)
s.recvWindowMu.Unlock()
// Process reassembled packet
s.processDataPacket(reassembled)
} else {
s.recvWindowMu.Unlock()
}
}
// handleCombined processes combined packets
func (s *EQStream) handleCombined(packet *EQProtocolPacket) {
data := packet.Buffer
offset := 0
for offset < len(data) {
if offset+1 > len(data) {
break
}
// Get sub-packet size
size := int(data[offset])
offset++
// Handle oversized packets (size == 255)
if size == 255 && offset+2 <= len(data) {
size = int(binary.BigEndian.Uint16(data[offset:offset+2]))
offset += 2
}
if offset+size > len(data) {
break
}
// Process sub-packet
subData := data[offset : offset+size]
s.handleIncomingPacket(subData)
offset += size
}
}
// handleOutOfSession processes out-of-session packets
func (s *EQStream) handleOutOfSession(packet *EQProtocolPacket) {
// Server is telling us we're not in a session
s.state.Store(int32(StreamStateDisconnected))
if s.onDisconnect != nil {
s.onDisconnect("out of session")
}
}
// sendAck sends an acknowledgment packet
func (s *EQStream) sendAck(seq uint32) {
data := make([]byte, 2)
binary.BigEndian.PutUint16(data, uint16(seq))
ack := NewEQProtocolPacket(OPAck, data)
s.sendPacketNow(ack)
}
// sendOutOfOrderAck sends an out-of-order acknowledgment
func (s *EQStream) sendOutOfOrderAck(seq uint32) {
data := make([]byte, 2)
binary.BigEndian.PutUint16(data, uint16(seq))
ack := NewEQProtocolPacket(OPOutOfOrderAck, data)
s.sendPacketNow(ack)
}
// sendSessionResponse sends a session response packet
func (s *EQStream) sendSessionResponse(accept bool) {
data := make([]byte, 11)
binary.BigEndian.PutUint32(data[0:4], s.sessionID)
binary.BigEndian.PutUint32(data[4:8], s.config.CRCKey)
if accept {
data[8] = 1 // Validation byte
} else {
data[8] = 0 // Rejection
}
data[9] = 0 // Format
data[10] = 0 // Unknown
response := NewEQProtocolPacket(OPSessionResponse, data)
s.sendPacketNow(response)
}
// FragmentPacket breaks a large packet into fragments
func (s *EQStream) FragmentPacket(data []byte, maxSize int) []*EQProtocolPacket {
if len(data) <= maxSize {
// No fragmentation needed
return []*EQProtocolPacket{NewEQProtocolPacket(OPPacket, data)}
}
// Calculate fragment sizes
headerSize := 6 // seq(2) + total_size(4)
fragmentDataSize := maxSize - headerSize
numFragments := (len(data) + fragmentDataSize - 1) / fragmentDataSize
fragments := make([]*EQProtocolPacket, 0, numFragments)
totalSize := uint32(len(data))
for offset := 0; offset < len(data); offset += fragmentDataSize {
end := offset + fragmentDataSize
if end > len(data) {
end = len(data)
}
// Build fragment packet
fragData := make([]byte, headerSize+end-offset)
// Sequence will be set by send worker
binary.BigEndian.PutUint32(fragData[2:6], totalSize)
copy(fragData[6:], data[offset:end])
fragments = append(fragments, NewEQProtocolPacket(OPFragment, fragData))
}
return fragments
}

194
stream_test.go Normal file
View File

@ -0,0 +1,194 @@
package eq2net
import (
"net"
"testing"
"time"
)
func TestStreamCreation(t *testing.T) {
config := DefaultStreamConfig()
stream := NewEQStream(config)
if stream == nil {
t.Fatal("Failed to create stream")
}
if stream.GetState() != StreamStateDisconnected {
t.Errorf("Expected disconnected state, got %v", stream.GetState())
}
// Test state transitions
if !stream.compareAndSwapState(StreamStateDisconnected, StreamStateConnecting) {
t.Error("Failed to transition to connecting state")
}
if stream.GetState() != StreamStateConnecting {
t.Errorf("Expected connecting state, got %v", stream.GetState())
}
// Clean up
stream.Close()
}
func TestStreamConfig(t *testing.T) {
config := DefaultStreamConfig()
if config.MaxPacketSize != 512 {
t.Errorf("Expected max packet size 512, got %d", config.MaxPacketSize)
}
if config.WindowSize != 2048 {
t.Errorf("Expected window size 2048, got %d", config.WindowSize)
}
if config.RetransmitTimeMs != 500 {
t.Errorf("Expected retransmit time 500ms, got %d", config.RetransmitTimeMs)
}
}
func TestRTTCalculation(t *testing.T) {
stream := NewEQStream(nil)
// Test first RTT sample
stream.updateRTT(100000) // 100ms in microseconds
if stream.rtt.Load() != 100000 {
t.Errorf("Expected RTT 100000, got %d", stream.rtt.Load())
}
// Test subsequent samples
stream.updateRTT(120000) // 120ms
stream.updateRTT(80000) // 80ms
// RTT should be smoothed
rtt := stream.rtt.Load()
if rtt < 80000 || rtt > 120000 {
t.Errorf("RTT outside expected range: %d", rtt)
}
// RTO should be set
rto := stream.rto.Load()
if rto < 200000 { // Minimum 200ms
t.Errorf("RTO below minimum: %d", rto)
}
}
func TestPacketSequencing(t *testing.T) {
stream := NewEQStream(nil)
// Test sequence number generation
seq1 := stream.nextSeqOut.Add(1)
seq2 := stream.nextSeqOut.Add(1)
seq3 := stream.nextSeqOut.Add(1)
if seq1 != 1 || seq2 != 2 || seq3 != 3 {
t.Errorf("Sequence numbers not incrementing correctly: %d, %d, %d", seq1, seq2, seq3)
}
}
func TestSendWindow(t *testing.T) {
stream := NewEQStream(nil)
// Add packet to send window
packet := NewEQProtocolPacket(OPPacket, []byte("test"))
packet.Sequence = 1
stream.sendWindowMu.Lock()
stream.sendWindow[1] = &sendPacket{
packet: packet,
sentTime: time.Now(),
attempts: 1,
nextRetry: time.Now().Add(500 * time.Millisecond),
}
stream.sendWindowMu.Unlock()
// Process ACK
stream.processAck(1)
// Verify packet removed from window
stream.sendWindowMu.RLock()
_, exists := stream.sendWindow[1]
stream.sendWindowMu.RUnlock()
if exists {
t.Error("Packet not removed from send window after ACK")
}
}
func TestFragmentation(t *testing.T) {
stream := NewEQStream(nil)
// Create large data that needs fragmentation
largeData := make([]byte, 1000)
for i := range largeData {
largeData[i] = byte(i % 256)
}
// Fragment the data
fragments := stream.FragmentPacket(largeData, 100)
if len(fragments) == 0 {
t.Fatal("No fragments created")
}
// Verify fragments
expectedFragments := (len(largeData) + 93) / 94 // 100 - 6 header bytes
if len(fragments) != expectedFragments {
t.Errorf("Expected %d fragments, got %d", expectedFragments, len(fragments))
}
// Verify each fragment has correct opcode
for _, frag := range fragments {
if frag.Opcode != OPFragment {
t.Errorf("Fragment has wrong opcode: %04x", frag.Opcode)
}
}
}
// TestMockConnection tests basic packet flow without real network
func TestMockConnection(t *testing.T) {
// Create mock packet conn
clientConn, serverConn := net.Pipe()
defer clientConn.Close()
defer serverConn.Close()
// Note: net.Pipe creates a stream connection, not packet-based
// For a real test, we'd need to use actual UDP sockets
// This is just to verify compilation
config := DefaultStreamConfig()
stream := NewEQStream(config)
// Verify stream creation
if stream == nil {
t.Fatal("Failed to create stream")
}
stream.Close()
}
func TestServerCreation(t *testing.T) {
config := DefaultServerConfig()
server := NewEQ2Server(config)
if server == nil {
t.Fatal("Failed to create server")
}
// Set callbacks
connectCount := 0
disconnectCount := 0
server.SetCallbacks(
func(s *EQStream) {
connectCount++
},
func(s *EQStream, reason string) {
disconnectCount++
},
)
// Note: We don't actually start the server in unit tests
// as it would require binding to a real port
}