EQ2/Protocol
1
0
Fork 0
Code Releases Activity

reset

Browse Source
This commit is contained in:
Sky Johnson 2025-09-02 09:41:39 -05:00
parent 1a9f194f6a
commit 44b0b86a2e
16 changed files with 6 additions and 4086 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

0
EQPACKET.md → DOCS/EQPACKET.md
View File

0
EQSTREAM.md → DOCS/EQSTREAM.md
View File

817
LOGIN.md
View File

@ -1,817 +0,0 @@
# EverQuest II Login Server - Complete Technical Documentation
## Table of Contents
1. [Overview](#overview)
2. [Architecture](#architecture)
3. [Network Protocol Stack](#network-protocol-stack)
4. [Encryption and Security](#encryption-and-security)
5. [Packet Structure](#packet-structure)
6. [Complete Login Flow](#complete-login-flow)
7. [Character Management](#character-management)
8. [World Server Communication](#world-server-communication)
9. [Error Handling](#error-handling)
10. [Database Operations](#database-operations)
## Overview
The EverQuest II Login Server is responsible for:
- Client authentication and authorization
- World server discovery and status management
- Character listing and management
- Secure handoff to world servers
- Account management and creation
### Key Components
- **NetConnection**: Main network manager
- **Client**: Individual client connection handler
- **ClientList**: Thread-safe client container
- **LWorld**: World server representation
- **LWorldList**: World server manager
- **LoginDatabase**: Database interface
- **EQStream**: Network stream handler with encryption
## Architecture
```
┌─────────────┐ UDP ┌──────────────┐
│ EQ2 Client ├─────────────────►│ Login Server │
└─────────────┘ └──────┬───────┘
│ TCP
▼
┌──────────────────┐
│ World Servers │
└──────────────────┘
```
### Port Configuration
- **Default Login Port**: 5999 (UDP for game clients)
- **Web API Port**: Configurable (TCP/HTTPS)
- **World Server Communication**: TCP (internal)
- **World Server Game Port**: Dynamic per world (UDP for game clients)
## Network Protocol Stack
### Layer Structure
```
Application Layer: [Game Packets]
Presentation Layer: [PacketStruct Serialization]
Session Layer: [EQStream with RC4/CRC]
Transport Layer: [UDP Datagram Protocol]
Network Layer: [IP Protocol]
```
**Note**: EQStream implements its own reliability layer on top of UDP, providing:
- Sequence numbers for ordering
- Acknowledgments for reliability
- Retransmission for lost packets
- This gives the benefits of TCP while maintaining lower latency
### EQStream Protocol
The EQStream protocol provides:
1. **Sequencing**: Packet ordering and acknowledgment
2. **Fragmentation**: Large packet splitting
3. **Compression**: zlib compression for efficiency
4. **Encryption**: RC4 stream cipher
5. **Integrity**: CRC32 checksums
#### Packet Types
```cpp
enum EQStreamOp {
OP_SessionRequest = 0x0001, // Initial connection
OP_SessionResponse = 0x0002, // Server accepts connection
OP_Combined = 0x0003, // Multiple packets in one
OP_SessionDisconnect = 0x0005, // Connection termination
OP_KeepAlive = 0x0006, // Connection heartbeat
OP_SessionStatRequest = 0x0007, // Statistics request
OP_SessionStatResponse = 0x0008,// Statistics response
OP_Packet = 0x0009, // Application data
OP_Fragment = 0x000d, // Fragmented packet piece
OP_Ack = 0x0015, // Acknowledgment
OP_AckFuture = 0x0016, // Future acknowledgment
OP_AckPast = 0x0017 // Past acknowledgment
};
```
## Encryption and Security
### RC4 Encryption
The login server uses RC4 stream cipher for packet encryption:
1. **Key Exchange**:
```cpp
// Initial session setup
struct SessionRequest {
uint32_t unknown; // Protocol version
uint32_t session_id; // Client session ID
uint32_t max_length; // Max packet size
};
struct SessionResponse {
uint32_t session_id; // Server session ID
uint32_t key; // RC4 key seed
uint8_t crc_length; // CRC bytes (2)
uint8_t compression; // Compression flag
uint32_t unknown; // Reserved
uint32_t max_length; // Max packet size
};
```
2. **RC4 Key Generation**:
```cpp
// Key is derived from session response
void GenerateRC4Key(uint32_t key_seed) {
// Initialize RC4 state with key_seed
RC4_KEY encrypt_key, decrypt_key;
unsigned char key_bytes[4];
memcpy(key_bytes, &key_seed, 4);
// Setup encryption/decryption keys
RC4_set_key(&encrypt_key, 4, key_bytes);
RC4_set_key(&decrypt_key, 4, key_bytes);
}
```
3. **Packet Encryption**:
```cpp
void EncryptPacket(uint8_t* data, size_t length) {
// Skip protocol header (2 bytes)
RC4(&encrypt_key, length - 2, data + 2, data + 2);
}
```
### CRC32 Integrity Check
Every packet includes a CRC32 checksum:
```cpp
struct EQProtocolPacket {
uint16_t opcode; // Operation code
uint8_t data[]; // Packet data
uint32_t crc32; // CRC32 of opcode + data
};
uint32_t CalculateCRC(const uint8_t* data, size_t length) {
// Standard CRC32 calculation
return crc32(0, data, length);
}
```
### CRC Verification Process:
1. Extract CRC from packet end
2. Calculate CRC of packet data
3. Compare calculated vs received CRC
4. Reject packet if mismatch
## Packet Structure
### Base Packet Header
```cpp
struct EQ2Packet {
uint16_t opcode; // Operation identifier
uint16_t sequence; // Packet sequence number
uint32_t size; // Payload size
uint8_t compressed; // Compression flag
uint8_t data[]; // Actual payload
};
```
### Application Packet Structure
```cpp
struct EQApplicationPacket {
uint16_t opcode; // Game operation code
uint32_t version; // Client version
uint8_t data[]; // Serialized data
};
```
## Complete Login Flow
### Phase 1: Connection Establishment
1. **Client Initiates UDP Communication**:
```
Client → Server: UDP Datagram with OP_SessionRequest
(No TCP handshake - connectionless protocol)
```
2. **EQStream Session**:
```
Client → Server: OP_SessionRequest
{
protocol_version: 0x0003
session_id: random()
max_packet_size: 512
}
Server → Client: OP_SessionResponse
{
session_id: server_session
rc4_key: generated_key
crc_length: 2
compression: true
max_packet_size: 512
}
```
3. **Encryption Initialization**:
- Both sides initialize RC4 with the shared key
- All subsequent packets are encrypted
### Phase 2: Authentication
1. **Login Request**:
```cpp
Client → Server: OP_LoginRequestMsg
{
version: client_version // e.g., 1208
username: "account_name" // EQ2_16BitString
password: "hashed_password" // EQ2_16BitString
unknown3: cl_eqversion // From eq2_defaults.ini
}
```
2. **Version Validation**:
```cpp
// Server checks version compatibility
if (!EQOpcodeManager.count(GetOpcodeVersion(version))) {
// Send incompatible version error
SendLoginDeniedBadVersion();
return;
}
```
3. **Account Verification**:
```cpp
// Database lookup
LoginAccount* acct = database.LoadAccount(username, password);
// Check for duplicate login
Client* existing = client_list.FindByLSID(acct->getLoginAccountID());
if (existing) {
existing->getConnection()->SendDisconnect();
}
// Update account info
database.UpdateAccountIPAddress(acct->id, client_ip);
database.UpdateAccountClientDataVersion(acct->id, version);
```
4. **Login Response**:
```cpp
Server → Client: OP_LoginReplyMsg
{
account_id: account_id
login_response: 0 // 0 = success
sub_level: 0xFFFFFFFF // Subscription level
race_flag: 0x1FFFFF // Available races
class_flag: 0x7FFFFFE // Available classes
expansion_flag: 0x7CFF // Enabled expansions
cities_flag: 0xFF // Starting cities
enabled_races: 0xFFFF // Race availability
}
```
### Phase 3: World List Request
1. **Client Requests World List**:
```cpp
Client → Server: OP_AllWSDescRequestMsg
{} // Empty request
```
2. **Server Builds World List**:
```cpp
Server → Client: OP_AllWSDescReplyMsg
{
num_worlds: count
worlds[]: {
world_id: id
world_name: "ServerName"
world_status: 1 // 0=down, 1=up, 2=locked
num_players: current
max_players: maximum
language: "en"
recommended: false
}
}
```
3. **World Status Updates**:
- Server sends periodic updates (every 10 seconds)
- Updates include player counts and status changes
### Phase 4: Character List
1. **Character Loading**:
```cpp
// Server loads characters from database
database.LoadCharacters(GetLoginAccount(), GetVersion());
```
2. **Character List Response**:
```cpp
Server → Client: LS_CharSelectList
{
account_id: account_id
num_characters: count
characters[]: {
char_id: database_id
server_id: world_id
name: "CharacterName"
race: race_id
class: class_id
level: current_level
zone: "ZoneName"
gender: 0/1
deity: deity_id
body_size: size
body_age: age
soga_race_type: soga_id
x: position_x
y: position_y
z: position_z
}
}
```
### Phase 5: Character Creation
1. **Creation Request**:
```cpp
Client → Server: OP_CreateCharacterRequestMsg
{
server_id: target_world
name: "NewCharacter"
race: selected_race
class: selected_class
gender: 0/1
deity: selected_deity
body_size: size_value
body_age: age_value
// Appearance data...
}
```
2. **Forward to World Server**:
```cpp
Login → World: ServerOP_CharacterCreate
{
version: client_version
account_id: account_id
[original_request_data]
}
```
3. **World Server Validation**:
- Check name availability
- Validate race/class combination
- Check character limits
4. **Creation Response**:
```cpp
World → Login: ServerOP_CharacterCreateResponse
{
success: true/false
char_id: new_character_id // If successful
reason: error_code // If failed
}
Login → Client: OP_CreateCharacterReplyMsg
{
response: CREATESUCCESS_REPLY or error_code
name: "NewCharacter"
account_id: account_id
}
```
### Phase 6: World Entry
1. **Play Request**:
```cpp
Client → Server: OP_PlayCharacterRequestMsg
{
char_id: selected_character
server_id: world_server // Version > 283
name: "CharacterName" // Version <= 283
}
```
2. **World Server Handoff**:
```cpp
Login → World: ServerOP_UsertoWorldReq
{
char_id: character_id
lsaccountid: account_id
worldid: server_id
ip_address: client_ip
}
```
3. **World Server Response**:
```cpp
World → Login: ServerOP_UsertoWorldResp
{
worldid: server_id
response: 1 // 1 = success
ip_address: world_ip
port: world_port
access_key: session_key // For authentication
}
```
4. **Client Redirect**:
```cpp
Login → Client: OP_PlayCharacterReplyMsg
{
response: 1 // Success
server: "world.ip.address"
port: world_port
account_id: account_id
access_code: session_key
}
```
5. **Client Connects to World**:
- Client disconnects from login server
- Connects to world server using provided IP:port
- Authenticates using access_key
## Character Management
### Character Deletion
1. **Delete Request**:
```cpp
Client → Server: OP_DeleteCharacterRequestMsg
{
char_id: character_id
server_id: world_id
name: "CharacterName"
}
```
2. **Verification**:
```cpp
// Verify ownership
bool valid = database.VerifyDelete(account_id, char_id, name);
```
3. **World Server Notification**:
```cpp
Login → World: ServerOP_DeleteCharacter
{
char_id: character_id
account_id: account_id
}
```
4. **Delete Response**:
```cpp
Server → Client: OP_DeleteCharacterReplyMsg
{
response: 1 // 1 = success
char_id: deleted_id
server_id: world_id
name: "CharacterName"
}
```
## World Server Communication
### Inter-Server Protocol
1. **World Server Registration**:
```cpp
World → Login: ServerOP_LSInfo
{
name: "WorldName"
address: "public_ip"
port: game_port
admin_port: admin_port
status: 1 // 0=down, 1=up, 2=locked
players: current_count
max_players: maximum
}
```
2. **Heartbeat**:
```cpp
World → Login: ServerOP_LSStatus
{
status: current_status
players: current_count
zones: active_zones
}
// Sent every 30 seconds
```
3. **Player Updates**:
```cpp
World → Login: ServerOP_UsertoWorldResp
{
lsaccountid: account_id
worldid: server_id
response: status
// 0 = rejected
// 1 = accepted
// -1 = world full
// -2 = character not found
// -3 = world locked
}
```
## Error Handling
### Login Error Codes
```cpp
enum LoginResponseCodes {
LOGIN_SUCCESS = 0,
LOGIN_BADPASS = 1, // Invalid username/password
LOGIN_BADVERSION = 6, // Client version mismatch
LOGIN_SUSPENDED = 7, // Account suspended
LOGIN_BANNED = 9, // Account banned
LOGIN_WORLDFULL = 10, // World at capacity
LOGIN_DISCONNECT = 100 // Generic disconnect
};
```
### Play Error Codes
```cpp
enum PlayResponseCodes {
PLAY_SUCCESS = 1,
PLAY_ERROR_PROBLEM = 2, // Generic error
PLAY_ERROR_WORLDFULL = 3, // World full
PLAY_ERROR_LOCKED = 4, // World locked
PLAY_ERROR_BANNED = 5, // Character banned
PLAY_ERROR_SUSPENDED = 6, // Character suspended
PLAY_ERROR_SERVER_TIMEOUT = 7, // Timeout waiting for world
PLAY_ERROR_CHAR_NOT_LOADED = 8 // Character data issue
};
```
### Character Creation Error Codes
```cpp
enum CreateResponseCodes {
CREATESUCCESS_REPLY = 1,
INVALIDRACE_REPLY = 2, // Invalid race selection
INVALIDGENDER_REPLY = 3, // Invalid gender
BADNAMELENGTH_REPLY = 9, // Name too short/long
NAMEINVALID_REPLY = 10, // Invalid characters
NAMEFILTER_REPLY = 11, // Profanity filter
NAMETAKEN_REPLY = 12, // Name already exists
OVERLOADEDSERVER_REPLY = 13, // Server overloaded
UNKNOWNERROR_REPLY = 20 // Generic error
};
```
## Database Operations
### Account Management
1. **Account Loading**:
```sql
SELECT id, name, password, suspended, banned
FROM login_accounts
WHERE name = ? AND password = SHA2(?, 256)
```
2. **Account Creation** (if enabled):
```sql
INSERT INTO login_accounts (name, password, created)
VALUES (?, SHA2(?, 256), NOW())
```
3. **IP Address Update**:
```sql
UPDATE login_accounts
SET last_ip = ?, last_login = NOW()
WHERE id = ?
```
### Character Operations
1. **Character List**:
```sql
SELECT c.id, c.name, c.server_id, c.level, c.race,
c.class, c.zone, c.gender, c.deity
FROM characters c
WHERE c.account_id = ? AND c.deleted = 0
ORDER BY c.server_id, c.name
```
2. **Character Creation**:
```sql
INSERT INTO characters
(account_id, server_id, name, race, class, gender,
deity, body_size, body_age, created)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())
```
3. **Character Deletion**:
```sql
UPDATE characters
SET deleted = 1, deleted_date = NOW()
WHERE id = ? AND account_id = ?
```
### World Server Tracking
1. **World Registration**:
```sql
INSERT INTO login_worldservers
(name, ip_address, port, status)
VALUES (?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
ip_address = VALUES(ip_address),
port = VALUES(port),
status = VALUES(status)
```
2. **Status Updates**:
```sql
UPDATE login_worldservers
SET status = ?, players = ?, last_update = NOW()
WHERE id = ?
```
## Security Considerations
### Password Handling
- Passwords are hashed using SHA256 before transmission
- Never store plaintext passwords
- Use prepared statements to prevent SQL injection
### Session Management
- Generate unique session keys for world handoff
- Session keys expire after use or timeout
- IP address validation between login and world entry
### Rate Limiting
- Limit login attempts per IP
- Throttle character creation requests
- Monitor for abnormal packet patterns
### Packet Validation
- Verify packet size limits
- Check sequence numbers for replay attacks
- Validate all input data ranges
## Performance Optimizations
### Connection Pooling
- Maintain persistent database connections
- Reuse world server connections
- Implement connection timeout and retry logic
### Caching
- Cache world server status for 10 seconds
- Cache character lists until modified
- Cache opcode mappings per version
### Threading Model
- Main thread handles network I/O
- Database operations on thread pool
- World server communication async
- Client processing single-threaded per client
## Monitoring and Logging
### Key Metrics
- Active client connections
- Login success/failure rates
- World server availability
- Database query performance
- Packet processing latency
### Log Levels
```cpp
enum LogLevel {
LOGIN__INFO, // General information
LOGIN__DEBUG, // Detailed debugging
LOGIN__ERROR, // Error conditions
LOGIN__WARNING, // Warning conditions
WORLD__INFO, // World server info
WORLD__ERROR, // World server errors
OPCODE__DEBUG, // Packet opcodes
INIT__INFO, // Initialization
INIT__ERROR // Init failures
};
```
### Critical Events to Log
- Failed login attempts
- Account creation
- Character creation/deletion
- World server registration/deregistration
- Network errors and timeouts
- Database connection issues
## Configuration
### Main Configuration (config.json)
```json
{
"loginconfig": {
"serverport": 5999,
"serverip": "",
"accountcreation": 1,
"expansionflag": 32463,
"citiesflag": 255,
"defaultsubscriptionlevel": -1,
"enabledraces": 65535,
"webloginaddress": "0.0.0.0",
"webloginport": 8080,
"webcertfile": "",
"webkeyfile": "",
"webkeypassword": "",
"webhardcodeuser": "admin",
"webhardcodepassword": "password"
}
}
```
### Database Configuration
```json
{
"database": {
"host": "localhost",
"port": 3306,
"username": "eq2login",
"password": "password",
"database": "eq2login",
"max_connections": 10,
"connection_timeout": 5
}
}
```
## Troubleshooting
### Common Issues
1. **"Version Mismatch" Error**:
- Client version not in opcode database
- Solution: Update opcodes.sql
2. **"Cannot Connect to Login Server"**:
- Firewall blocking port 5999
- Server not running
- Network configuration issue
3. **"World Server Unavailable"**:
- World server not registered
- World server crashed
- Network issue between login and world
4. **Character Creation Fails**:
- Name already taken
- Invalid race/class combination
- World server communication timeout
5. **Cannot Enter World**:
- Session key mismatch
- World server full
- Character data corruption
### Debug Commands
The login server supports console commands (Linux):
- `l` - List all connected world servers
- `v` - Display version information
- `h` - Show help menu
- `q` - Quit server gracefully
## Protocol Evolution
### Version Differences
**Version <= 283**:
- Server ID not included in play request
- Character name used for lookup
**Version <= 561**:
- Auto-enter world after character creation
- Different packet structure alignment
**Version >= 546**:
- Enhanced error logging format
- Larger packet size support
**Version >= 1208**:
- New login request structure
- Additional client version field
## Conclusion
The EverQuest II login server implements a sophisticated authentication and routing system with multiple layers of security and error handling. The combination of RC4 encryption, CRC32 integrity checking, and session-based authentication provides a robust framework for managing player connections and character data.
Key design principles:
- **Security First**: All communications encrypted
- **Scalability**: Support for multiple world servers
- **Reliability**: Comprehensive error handling
- **Performance**: Efficient caching and threading
- **Maintainability**: Clear separation of concerns
The login server acts as the gateway to the game world, ensuring only authenticated players can access their characters while maintaining the integrity and security of the game environment.

209
compression.go
View File

@ -1,209 +0,0 @@
package eq2net
import (
"bytes"
"compress/zlib"
"encoding/binary"
"io"
)
const (
// Compression flags
CompressionFlagZlib = 0x5A // Zlib compression
CompressionFlagSimple = 0xA5 // Simple encoding (no actual compression)
// Compression threshold - packets larger than this use zlib
CompressionThreshold = 30
)
// CompressPacket compresses a packet using zlib or simple encoding
func CompressPacket(data []byte) ([]byte, error) {
if len(data) < 2 {
return data, nil
}
// Determine opcode size
flagOffset := 1
if data[0] == 0 {
flagOffset = 2 // Two-byte opcode
}
// Don't compress if too small
if len(data) <= flagOffset {
return data, nil
}
result := make([]byte, 0, len(data)+1)
// Copy opcode bytes
result = append(result, data[:flagOffset]...)
if len(data) > CompressionThreshold {
// Use zlib compression for larger packets
result = append(result, CompressionFlagZlib)
// Compress the data after opcode
var compressed bytes.Buffer
w := zlib.NewWriter(&compressed)
if _, err := w.Write(data[flagOffset:]); err != nil {
return nil, err
}
if err := w.Close(); err != nil {
return nil, err
}
result = append(result, compressed.Bytes()...)
} else {
// Use simple encoding for smaller packets
result = append(result, CompressionFlagSimple)
result = append(result, data[flagOffset:]...)
}
return result, nil
}
// DecompressPacket decompresses a packet
func DecompressPacket(data []byte) ([]byte, error) {
if len(data) < 3 {
return data, nil
}
// Determine opcode size and compression flag position
flagOffset := 1
if data[0] == 0 {
flagOffset = 2
}
if len(data) <= flagOffset {
return data, nil
}
compressionFlag := data[flagOffset]
// Check compression type
switch compressionFlag {
case CompressionFlagZlib:
// Zlib decompression
result := make([]byte, 0, len(data)*2)
// Copy opcode
result = append(result, data[:flagOffset]...)
// Decompress data (skip flag byte)
r, err := zlib.NewReader(bytes.NewReader(data[flagOffset+1:]))
if err != nil {
return nil, err
}
defer r.Close()
decompressed, err := io.ReadAll(r)
if err != nil {
return nil, err
}
result = append(result, decompressed...)
return result, nil
case CompressionFlagSimple:
// Simple encoding - just remove the flag byte
result := make([]byte, 0, len(data)-1)
result = append(result, data[:flagOffset]...)
result = append(result, data[flagOffset+1:]...)
return result, nil
default:
// No compression
return data, nil
}
}
// IsCompressed checks if a packet is compressed
func IsCompressed(data []byte) bool {
if len(data) < 2 {
return false
}
flagOffset := 1
if data[0] == 0 {
flagOffset = 2
}
if len(data) <= flagOffset {
return false
}
flag := data[flagOffset]
return flag == CompressionFlagZlib || flag == CompressionFlagSimple
}
// ChatEncode encodes chat data using XOR encryption with rolling key
func ChatEncode(data []byte, encodeKey uint32) []byte {
// Skip certain packet types
if len(data) >= 2 && (data[1] == 0x01 || data[0] == 0x02 || data[0] == 0x1d) {
return data
}
// Work with data after opcode
if len(data) <= 2 {
return data
}
result := make([]byte, len(data))
copy(result[:2], data[:2]) // Copy opcode
key := encodeKey
offset := 2
// Process 4-byte blocks with rolling key
for i := offset; i+4 <= len(data); i += 4 {
block := binary.LittleEndian.Uint32(data[i : i+4])
encrypted := block ^ key
binary.LittleEndian.PutUint32(result[i:i+4], encrypted)
key = encrypted // Update key with encrypted data
}
// Handle remaining bytes
keyByte := byte(key & 0xFF)
alignedEnd := offset + ((len(data)-offset)/4)*4
for i := alignedEnd; i < len(data); i++ {
result[i] = data[i] ^ keyByte
}
return result
}
// ChatDecode decodes chat data using XOR encryption with rolling key
func ChatDecode(data []byte, decodeKey uint32) []byte {
// Skip certain packet types
if len(data) >= 2 && (data[1] == 0x01 || data[0] == 0x02 || data[0] == 0x1d) {
return data
}
// Work with data after opcode
if len(data) <= 2 {
return data
}
result := make([]byte, len(data))
copy(result[:2], data[:2]) // Copy opcode
key := decodeKey
offset := 2
// Process 4-byte blocks with rolling key
for i := offset; i+4 <= len(data); i += 4 {
encrypted := binary.LittleEndian.Uint32(data[i : i+4])
decrypted := encrypted ^ key
binary.LittleEndian.PutUint32(result[i:i+4], decrypted)
key = encrypted // Update key with encrypted data (before decryption)
}
// Handle remaining bytes
keyByte := byte(key & 0xFF)
alignedEnd := offset + ((len(data)-offset)/4)*4
for i := alignedEnd; i < len(data); i++ {
result[i] = data[i] ^ keyByte
}
return result
}

93
crc16.go
View File

@ -1,93 +0,0 @@
package eq2net
// CRC16 table for CCITT polynomial (0x1021)
var crc16Table [256]uint16
func init() {
// Initialize CRC16 lookup table
for i := 0; i < 256; i++ {
crc := uint16(i << 8)
for j := 0; j < 8; j++ {
if (crc & 0x8000) != 0 {
crc = (crc << 1) ^ 0x1021
} else {
crc = crc << 1
}
}
crc16Table[i] = crc
}
}
// CRC16 calculates the CRC16-CCITT checksum with a key
func CRC16(data []byte, length int, key uint32) uint16 {
if length <= 0 || len(data) < length {
return 0
}
// Mix the key into initial CRC value
crc := uint16(0xFFFF)
keyBytes := []byte{
byte(key),
byte(key >> 8),
byte(key >> 16),
byte(key >> 24),
}
// Process key bytes first
for _, b := range keyBytes {
tableIndex := (uint8(crc>>8) ^ b) & 0xFF
crc = (crc << 8) ^ crc16Table[tableIndex]
}
// Process data
for i := 0; i < length; i++ {
tableIndex := (uint8(crc>>8) ^ data[i]) & 0xFF
crc = (crc << 8) ^ crc16Table[tableIndex]
}
return crc ^ 0xFFFF
}
// ValidateCRC checks if a packet has a valid CRC
func ValidateCRC(buffer []byte, key uint32) bool {
if len(buffer) < 3 {
return false
}
// Check for CRC-exempt packets
if len(buffer) >= 2 && buffer[0] == 0x00 {
switch buffer[1] {
case byte(OPSessionRequest), byte(OPSessionResponse), byte(OPOutOfSession):
return true // Session packets don't require CRC
}
}
// Check for combined application packets (also CRC-exempt)
if len(buffer) >= 4 && buffer[2] == 0x00 && buffer[3] == 0x19 {
return true
}
// Calculate CRC for the packet (excluding last 2 CRC bytes)
dataLen := len(buffer) - 2
calculatedCRC := CRC16(buffer, dataLen, key)
// Extract packet CRC (big-endian in last 2 bytes)
packetCRC := uint16(buffer[dataLen])<<8 | uint16(buffer[dataLen+1])
// Valid if no CRC present (0) or CRCs match
return packetCRC == 0 || calculatedCRC == packetCRC
}
// AppendCRC adds CRC to the end of a packet
func AppendCRC(buffer []byte, key uint32) []byte {
// Calculate CRC for current buffer
crc := CRC16(buffer, len(buffer), key)
// Append CRC in big-endian format
result := make([]byte, len(buffer)+2)
copy(result, buffer)
result[len(buffer)] = byte(crc >> 8)
result[len(buffer)+1] = byte(crc)
return result
}

1
go.mod
View File

@ -5,6 +5,7 @@ go 1.21.0
require (
filippo.io/edwards25519 v1.1.0 // indirect
github.com/go-sql-driver/mysql v1.9.3 // indirect
github.com/jmoiron/sqlx v1.4.0 // indirect
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

5
go.sum
View File

@ -1,7 +1,12 @@
filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA=
filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4=
github.com/go-sql-driver/mysql v1.8.1/go.mod h1:wEBSXgmK//2ZFJyE+qWnIsVGmvmEKlqwuVSjsCm7DZg=
github.com/go-sql-driver/mysql v1.9.3 h1:U/N249h2WzJ3Ukj8SowVFjdtZKfu9vlLZxjPXV1aweo=
github.com/go-sql-driver/mysql v1.9.3/go.mod h1:qn46aNg1333BRMNU69Lq93t8du/dwxI64Gl8i5p1WMU=
github.com/jmoiron/sqlx v1.4.0 h1:1PLqN7S1UYp5t4SrVVnt4nUVNemrDAtxlulVe+Qgm3o=
github.com/jmoiron/sqlx v1.4.0/go.mod h1:ZrZ7UsYB/weZdl2Bxg6jCRO9c3YHl8r3ahlKmRT4JLY=
github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
github.com/mattn/go-sqlite3 v1.14.22/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y=
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=

272
opcodes.go
View File

@ -1,272 +0,0 @@
package eq2net
import (
"fmt"
"sync"
)
// VersionRange represents a client version range mapped to an opcode version
type VersionRange struct {
MinVersion uint16 // version_range1 in database
MaxVersion uint16 // version_range2 in database
OpcodeVersion uint16 // The opcode version for this range (usually MinVersion)
}
// OpcodeVersionMap maps version ranges for opcode lookups
// This replaces the C++ EQOpcodeVersions map<int16, int16>
type OpcodeVersionMap map[uint16]uint16 // Key: version_range1, Value: version_range2
// GetOpcodeVersion returns the opcode version for a given client version
// This is a direct port of the C++ GetOpcodeVersion function
func GetOpcodeVersion(clientVersion uint16, versionMap OpcodeVersionMap) uint16 {
ret := clientVersion
// Iterate through version ranges to find a match
for minVersion, maxVersion := range versionMap {
if clientVersion >= minVersion && clientVersion <= maxVersion {
ret = minVersion
break
}
}
return ret
}
// EmuOpcode represents an emulator-side opcode
type EmuOpcode uint16
// Common emulator opcodes - these match the C++ emu_opcodes.h
const (
OP_Unknown EmuOpcode = 0x0000
OP_LoginRequestMsg EmuOpcode = 0x0001
OP_LoginByNumRequestMsg EmuOpcode = 0x0002
OP_WSLoginRequestMsg EmuOpcode = 0x0003
OP_ESLoginRequestMsg EmuOpcode = 0x0004
OP_LoginReplyMsg EmuOpcode = 0x0005
OP_WSStatusReplyMsg EmuOpcode = 0x0006
OP_WorldListMsg EmuOpcode = 0x0007
OP_WorldStatusMsg EmuOpcode = 0x0008
OP_DeleteCharacterRequestMsg EmuOpcode = 0x0009
OP_DeleteCharacterReplyMsg EmuOpcode = 0x000A
OP_CreateCharacterRequestMsg EmuOpcode = 0x000B
OP_CreateCharacterReplyMsg EmuOpcode = 0x000C
OP_PlayCharacterRequestMsg EmuOpcode = 0x000D
OP_PlayCharacterReplyMsg EmuOpcode = 0x000E
OP_ServerListRequestMsg EmuOpcode = 0x000F
OP_ServerListReplyMsg EmuOpcode = 0x0010
OP_CharacterListRequestMsg EmuOpcode = 0x0011
OP_CharacterListReplyMsg EmuOpcode = 0x0012
// Add more opcodes as needed
)
// OpcodeNames maps emulator opcodes to their string names
// This matches the C++ OpcodeNames array
var OpcodeNames = map[EmuOpcode]string{
OP_Unknown: "OP_Unknown",
OP_LoginRequestMsg: "OP_LoginRequestMsg",
OP_LoginByNumRequestMsg: "OP_LoginByNumRequestMsg",
OP_WSLoginRequestMsg: "OP_WSLoginRequestMsg",
OP_ESLoginRequestMsg: "OP_ESLoginRequestMsg",
OP_LoginReplyMsg: "OP_LoginReplyMsg",
OP_WSStatusReplyMsg: "OP_WSStatusReplyMsg",
OP_WorldListMsg: "OP_WorldListMsg",
OP_WorldStatusMsg: "OP_WorldStatusMsg",
OP_DeleteCharacterRequestMsg: "OP_DeleteCharacterRequestMsg",
OP_DeleteCharacterReplyMsg: "OP_DeleteCharacterReplyMsg",
OP_CreateCharacterRequestMsg: "OP_CreateCharacterRequestMsg",
OP_CreateCharacterReplyMsg: "OP_CreateCharacterReplyMsg",
OP_PlayCharacterRequestMsg: "OP_PlayCharacterRequestMsg",
OP_PlayCharacterReplyMsg: "OP_PlayCharacterReplyMsg",
OP_ServerListRequestMsg: "OP_ServerListRequestMsg",
OP_ServerListReplyMsg: "OP_ServerListReplyMsg",
OP_CharacterListRequestMsg: "OP_CharacterListRequestMsg",
OP_CharacterListReplyMsg: "OP_CharacterListReplyMsg",
}
// RegularOpcodeManager manages opcode mappings for a specific version
// This is equivalent to the C++ RegularOpcodeManager class
type RegularOpcodeManager struct {
version uint16
emuToEQ map[EmuOpcode]uint16
eqToEmu map[uint16]EmuOpcode
mu sync.RWMutex
}
// NewRegularOpcodeManager creates a new opcode manager
func NewRegularOpcodeManager(version uint16) *RegularOpcodeManager {
return &RegularOpcodeManager{
version: version,
emuToEQ: make(map[EmuOpcode]uint16),
eqToEmu: make(map[uint16]EmuOpcode),
}
}
// LoadOpcodes loads opcode mappings from a map
// Input format matches database: map[opcode_name]opcode_value
func (om *RegularOpcodeManager) LoadOpcodes(opcodes map[string]uint16) bool {
om.mu.Lock()
defer om.mu.Unlock()
// Clear existing mappings
om.emuToEQ = make(map[EmuOpcode]uint16)
om.eqToEmu = make(map[uint16]EmuOpcode)
// Build bidirectional mappings
for name, eqOpcode := range opcodes {
// Find the emulator opcode by name
var emuOpcode EmuOpcode = OP_Unknown
for emu, opcName := range OpcodeNames {
if opcName == name {
emuOpcode = emu
break
}
}
if emuOpcode != OP_Unknown {
om.emuToEQ[emuOpcode] = eqOpcode
om.eqToEmu[eqOpcode] = emuOpcode
}
}
return true
}
// EmuToEQ converts an emulator opcode to EQ network opcode
func (om *RegularOpcodeManager) EmuToEQ(emu EmuOpcode) uint16 {
om.mu.RLock()
defer om.mu.RUnlock()
if eq, exists := om.emuToEQ[emu]; exists {
return eq
}
return 0xCDCD // Invalid opcode marker (matches C++)
}
// EQToEmu converts an EQ network opcode to emulator opcode
func (om *RegularOpcodeManager) EQToEmu(eq uint16) EmuOpcode {
om.mu.RLock()
defer om.mu.RUnlock()
if emu, exists := om.eqToEmu[eq]; exists {
return emu
}
return OP_Unknown
}
// EmuToName returns the name of an emulator opcode
func (om *RegularOpcodeManager) EmuToName(emu EmuOpcode) string {
if name, exists := OpcodeNames[emu]; exists {
return name
}
return "OP_Unknown"
}
// EQToName returns the name of an EQ network opcode
func (om *RegularOpcodeManager) EQToName(eq uint16) string {
emu := om.EQToEmu(eq)
return om.EmuToName(emu)
}
// NameSearch finds an emulator opcode by name
func NameSearch(name string) EmuOpcode {
for opcode, opcName := range OpcodeNames {
if opcName == name {
return opcode
}
}
return OP_Unknown
}
// EQOpcodeManager is the global opcode manager map
// Maps opcode version to manager instance
// This replaces the C++ map<int16, OpcodeManager*> EQOpcodeManager
type EQOpcodeManagerMap map[uint16]*RegularOpcodeManager
// NewEQOpcodeManager creates and initializes the global opcode manager
func NewEQOpcodeManager() EQOpcodeManagerMap {
return make(EQOpcodeManagerMap)
}
// LoadFromDatabase simulates loading opcodes from database results
// This would be called by your application after querying the database
func (m EQOpcodeManagerMap) LoadFromDatabase(versions OpcodeVersionMap, opcodesByVersion map[uint16]map[string]uint16) error {
// For each version range, create an opcode manager
for minVersion := range versions {
manager := NewRegularOpcodeManager(minVersion)
// Load opcodes for this version
if opcodes, exists := opcodesByVersion[minVersion]; exists {
if !manager.LoadOpcodes(opcodes) {
return fmt.Errorf("failed to load opcodes for version %d", minVersion)
}
} else {
return fmt.Errorf("no opcodes found for version %d", minVersion)
}
m[minVersion] = manager
}
return nil
}
// GetManagerForClient returns the appropriate opcode manager for a client version
func (m EQOpcodeManagerMap) GetManagerForClient(clientVersion uint16, versionMap OpcodeVersionMap) *RegularOpcodeManager {
opcodeVersion := GetOpcodeVersion(clientVersion, versionMap)
return m[opcodeVersion]
}
// Example helper functions for database integration
// These would be implemented by the application using this library
// LoadVersionsFromDB would execute:
// SELECT DISTINCT version_range1, version_range2 FROM opcodes
func LoadVersionsFromDB() OpcodeVersionMap {
// This is just an example - actual implementation would query the database
return OpcodeVersionMap{
1: 546, // Version range 1-546 uses opcode version 1
547: 889, // Version range 547-889 uses opcode version 547
890: 1027, // etc.
1028: 1048,
1049: 1095,
1096: 1184,
1185: 1197,
1198: 1207,
1208: 1211,
1212: 9999,
}
}
// LoadOpcodesFromDB would execute:
// SELECT name, opcode FROM opcodes WHERE ? BETWEEN version_range1 AND version_range2
func LoadOpcodesFromDB(version uint16) map[string]uint16 {
// This is just an example - actual implementation would query the database
return map[string]uint16{
"OP_LoginRequestMsg": 0x00B3,
"OP_LoginReplyMsg": 0x00B6,
// ... etc
}
}
// InitializeOpcodeSystem shows how to initialize the opcode system
// This would be called during server startup
func InitializeOpcodeSystem() (EQOpcodeManagerMap, OpcodeVersionMap, error) {
// Load version ranges from database
versions := LoadVersionsFromDB()
// Create the global opcode manager
opcodeManager := NewEQOpcodeManager()
// Load opcodes for each version
opcodesByVersion := make(map[uint16]map[string]uint16)
for minVersion := range versions {
opcodes := LoadOpcodesFromDB(minVersion)
opcodesByVersion[minVersion] = opcodes
}
// Initialize the manager
if err := opcodeManager.LoadFromDatabase(versions, opcodesByVersion); err != nil {
return nil, nil, err
}
return opcodeManager, versions, nil
}

252
opcodes_db_example.go
View File

@ -1,252 +0,0 @@
package eq2net
import (
"database/sql"
"fmt"
"log"
"time"
_ "github.com/go-sql-driver/mysql"
)
// OpcodeDBLoader handles loading opcodes from a MySQL database
// This keeps database concerns separate from the core opcode system
type OpcodeDBLoader struct {
db *sql.DB
}
// NewOpcodeDBLoader creates a new database loader
func NewOpcodeDBLoader(db *sql.DB) *OpcodeDBLoader {
return &OpcodeDBLoader{db: db}
}
// LoadVersions loads version ranges from the database
// Executes: SELECT DISTINCT version_range1, version_range2 FROM opcodes
func (l *OpcodeDBLoader) LoadVersions() (OpcodeVersionMap, error) {
query := `SELECT DISTINCT version_range1, version_range2 FROM opcodes ORDER BY version_range1`
rows, err := l.db.Query(query)
if err != nil {
return nil, fmt.Errorf("failed to query version ranges: %w", err)
}
defer rows.Close()
versions := make(OpcodeVersionMap)
for rows.Next() {
var minVersion, maxVersion uint16
if err := rows.Scan(&minVersion, &maxVersion); err != nil {
return nil, fmt.Errorf("failed to scan version range: %w", err)
}
versions[minVersion] = maxVersion
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("error iterating version rows: %w", err)
}
return versions, nil
}
// LoadOpcodes loads opcodes for a specific version
// Executes: SELECT name, opcode FROM opcodes WHERE ? BETWEEN version_range1 AND version_range2
func (l *OpcodeDBLoader) LoadOpcodes(version uint16) (map[string]uint16, error) {
query := `
SELECT name, opcode
FROM opcodes
WHERE ? BETWEEN version_range1 AND version_range2
ORDER BY version_range1, id
`
rows, err := l.db.Query(query, version)
if err != nil {
return nil, fmt.Errorf("failed to query opcodes for version %d: %w", version, err)
}
defer rows.Close()
opcodes := make(map[string]uint16)
for rows.Next() {
var name string
var opcode uint16
if err := rows.Scan(&name, &opcode); err != nil {
return nil, fmt.Errorf("failed to scan opcode row: %w", err)
}
opcodes[name] = opcode
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("error iterating opcode rows: %w", err)
}
return opcodes, nil
}
// LoadAllOpcodes loads all opcodes for all versions at once
// More efficient for server initialization
func (l *OpcodeDBLoader) LoadAllOpcodes() (map[uint16]map[string]uint16, error) {
// First get all unique version ranges
versions, err := l.LoadVersions()
if err != nil {
return nil, err
}
// Load opcodes for each version
result := make(map[uint16]map[string]uint16)
for minVersion := range versions {
opcodes, err := l.LoadOpcodes(minVersion)
if err != nil {
return nil, fmt.Errorf("failed to load opcodes for version %d: %w", minVersion, err)
}
result[minVersion] = opcodes
}
return result, nil
}
// InitializeOpcodeSystemFromDB initializes the opcode system from a database
func InitializeOpcodeSystemFromDB(db *sql.DB) (EQOpcodeManagerMap, OpcodeVersionMap, error) {
loader := NewOpcodeDBLoader(db)
// Load version ranges
versions, err := loader.LoadVersions()
if err != nil {
return nil, nil, fmt.Errorf("failed to load version ranges: %w", err)
}
// Load all opcodes
opcodesByVersion, err := loader.LoadAllOpcodes()
if err != nil {
return nil, nil, fmt.Errorf("failed to load opcodes: %w", err)
}
// Create and initialize the opcode manager
opcodeManager := NewEQOpcodeManager()
if err := opcodeManager.LoadFromDatabase(versions, opcodesByVersion); err != nil {
return nil, nil, fmt.Errorf("failed to initialize opcode manager: %w", err)
}
log.Printf("Loaded opcodes for %d version ranges", len(versions))
for minVersion, maxVersion := range versions {
if opcodes, exists := opcodesByVersion[minVersion]; exists {
log.Printf(" Version %d-%d: %d opcodes", minVersion, maxVersion, len(opcodes))
}
}
return opcodeManager, versions, nil
}
// Example usage showing how to use the opcode system with a database
func ExampleDatabaseUsage() {
// Connect to database
dsn := "root:Root12!@tcp(localhost:3306)/eq2db?parseTime=true"
db, err := sql.Open("mysql", dsn)
if err != nil {
log.Fatalf("Failed to open database: %v", err)
}
defer db.Close()
// Configure connection pool
db.SetMaxOpenConns(25)
db.SetMaxIdleConns(5)
db.SetConnMaxLifetime(5 * time.Minute)
// Test connection
if err := db.Ping(); err != nil {
log.Fatalf("Failed to ping database: %v", err)
}
// Initialize the opcode system
opcodeManager, versionMap, err := InitializeOpcodeSystemFromDB(db)
if err != nil {
log.Fatalf("Failed to initialize opcode system: %v", err)
}
// Example: Handle a client with version 1193
clientVersion := uint16(1193)
// Get the appropriate opcode manager for this client
manager := opcodeManager.GetManagerForClient(clientVersion, versionMap)
if manager == nil {
log.Fatalf("No opcode manager available for client version %d", clientVersion)
}
// Convert opcodes as needed
emuOpcode := OP_LoginRequestMsg
eqOpcode := manager.EmuToEQ(emuOpcode)
log.Printf("Client %d: %s -> 0x%04X", clientVersion, manager.EmuToName(emuOpcode), eqOpcode)
// Reverse conversion
emuOpcode = manager.EQToEmu(eqOpcode)
log.Printf("Client %d: 0x%04X -> %s", clientVersion, eqOpcode, manager.EmuToName(emuOpcode))
}
// CreateOpcodeTableSQL returns the SQL to create the opcodes table
// This matches the existing EQ2 schema
func CreateOpcodeTableSQL() string {
return `
CREATE TABLE IF NOT EXISTS opcodes (
id INT AUTO_INCREMENT PRIMARY KEY,
version_range1 INT NOT NULL,
version_range2 INT NOT NULL,
name VARCHAR(64) NOT NULL,
opcode INT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
INDEX idx_version_range (version_range1, version_range2),
INDEX idx_name (name)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
`
}
// InsertExampleOpcodes inserts example opcodes for testing
func InsertExampleOpcodes(db *sql.DB) error {
// Example data for version 1185-1197 (includes client version 1193)
opcodes := []struct {
versionMin uint16
versionMax uint16
name string
opcode uint16
}{
{1185, 1197, "OP_LoginRequestMsg", 0x00B3},
{1185, 1197, "OP_LoginReplyMsg", 0x00B6},
{1185, 1197, "OP_WorldListMsg", 0x00B8},
{1185, 1197, "OP_PlayCharacterRequestMsg", 0x00BE},
{1185, 1197, "OP_PlayCharacterReplyMsg", 0x00BF},
{1185, 1197, "OP_DeleteCharacterRequestMsg", 0x00BA},
{1185, 1197, "OP_CreateCharacterRequestMsg", 0x00BC},
// Example data for version 1198-1207
{1198, 1207, "OP_LoginRequestMsg", 0x00C3},
{1198, 1207, "OP_LoginReplyMsg", 0x00C6},
{1198, 1207, "OP_WorldListMsg", 0x00C8},
{1198, 1207, "OP_PlayCharacterRequestMsg", 0x00CE},
{1198, 1207, "OP_PlayCharacterReplyMsg", 0x00CF},
// Example data for version 1208-1211
{1208, 1211, "OP_LoginRequestMsg", 0x00D3},
{1208, 1211, "OP_LoginReplyMsg", 0x00D6},
{1208, 1211, "OP_WorldListMsg", 0x00D8},
{1208, 1211, "OP_PlayCharacterRequestMsg", 0x00DE},
{1208, 1211, "OP_PlayCharacterReplyMsg", 0x00DF},
}
// Prepare insert statement
stmt, err := db.Prepare(`
INSERT INTO opcodes (version_range1, version_range2, name, opcode)
VALUES (?, ?, ?, ?)
ON DUPLICATE KEY UPDATE opcode = VALUES(opcode)
`)
if err != nil {
return fmt.Errorf("failed to prepare statement: %w", err)
}
defer stmt.Close()
// Insert all opcodes
for _, op := range opcodes {
if _, err := stmt.Exec(op.versionMin, op.versionMax, op.name, op.opcode); err != nil {
return fmt.Errorf("failed to insert opcode %s: %w", op.name, err)
}
}
log.Printf("Inserted %d example opcodes", len(opcodes))
return nil
}

263
opcodes_test.go
View File

@ -1,263 +0,0 @@
package eq2net
import (
"fmt"
"testing"
)
func TestGetOpcodeVersionV2(t *testing.T) {
// Create a version map that matches the C++ implementation
versionMap := OpcodeVersionMap{
1: 546,
547: 889,
890: 1027,
1028: 1048,
1049: 1095,
1096: 1184,
1185: 1197,
1198: 1207,
1208: 1211,
1212: 9999,
}
tests := []struct {
clientVersion uint16
expectedOpcode uint16
}{
// Test first range
{1, 1},
{100, 1},
{546, 1},
// Test second range
{547, 547},
{700, 547},
{889, 547},
// Test middle ranges
{890, 890},
{1000, 890},
{1027, 890},
{1028, 1028},
{1048, 1028},
{1096, 1096},
{1100, 1096},
{1184, 1096},
{1185, 1185},
{1193, 1185},
{1197, 1185},
{1198, 1198},
{1200, 1198},
{1207, 1198},
{1208, 1208},
{1210, 1208},
{1211, 1208},
// Test last range
{1212, 1212},
{2000, 1212},
{9999, 1212},
// Test out of range (should return client version)
{10000, 10000},
}
for _, tt := range tests {
result := GetOpcodeVersion(tt.clientVersion, versionMap)
if result != tt.expectedOpcode {
t.Errorf("GetOpcodeVersion(%d) = %d, want %d",
tt.clientVersion, result, tt.expectedOpcode)
}
}
}
func TestRegularOpcodeManagerV2(t *testing.T) {
manager := NewRegularOpcodeManager(1193)
// Test loading opcodes (simulating database results)
opcodes := map[string]uint16{
"OP_LoginRequestMsg": 0x0001,
"OP_LoginReplyMsg": 0x0002,
"OP_WorldListMsg": 0x0003,
"OP_PlayCharacterRequestMsg": 0x0004,
}
if !manager.LoadOpcodes(opcodes) {
t.Fatal("Failed to load opcodes")
}
// Test EmuToEQ conversion
eq := manager.EmuToEQ(OP_LoginRequestMsg)
if eq != 0x0001 {
t.Errorf("Expected EQ opcode 0x0001, got 0x%04x", eq)
}
// Test invalid opcode returns 0xCDCD
eq = manager.EmuToEQ(OP_Unknown)
if eq != 0xCDCD {
t.Errorf("Expected 0xCDCD for unknown opcode, got 0x%04x", eq)
}
// Test EQToEmu conversion
emu := manager.EQToEmu(0x0002)
if emu != OP_LoginReplyMsg {
t.Errorf("Expected emu opcode %v, got %v", OP_LoginReplyMsg, emu)
}
// Test unknown EQ opcode
emu = manager.EQToEmu(0xFFFF)
if emu != OP_Unknown {
t.Errorf("Expected OP_Unknown for unknown EQ opcode, got %v", emu)
}
// Test name lookups
name := manager.EmuToName(OP_LoginRequestMsg)
if name != "OP_LoginRequestMsg" {
t.Errorf("Expected 'OP_LoginRequestMsg', got '%s'", name)
}
name = manager.EQToName(0x0003)
if name != "OP_WorldListMsg" {
t.Errorf("Expected 'OP_WorldListMsg', got '%s'", name)
}
}
func TestEQOpcodeManagerMapV2(t *testing.T) {
// Create version map
versions := OpcodeVersionMap{
1185: 1197,
1198: 1207,
1208: 1211,
}
// Create opcodes for each version
opcodesByVersion := map[uint16]map[string]uint16{
1185: {
"OP_LoginRequestMsg": 0x00B3,
"OP_LoginReplyMsg": 0x00B6,
},
1198: {
"OP_LoginRequestMsg": 0x00C1,
"OP_LoginReplyMsg": 0x00C4,
},
1208: {
"OP_LoginRequestMsg": 0x00D1,
"OP_LoginReplyMsg": 0x00D4,
},
}
// Initialize the manager map
managerMap := NewEQOpcodeManager()
err := managerMap.LoadFromDatabase(versions, opcodesByVersion)
if err != nil {
t.Fatalf("Failed to load from database: %v", err)
}
// Test getting manager for client version 1193 (should use 1185 opcodes)
manager := managerMap.GetManagerForClient(1193, versions)
if manager == nil {
t.Fatal("Failed to get manager for client version 1193")
}
eq := manager.EmuToEQ(OP_LoginRequestMsg)
if eq != 0x00B3 {
t.Errorf("Expected 0x00B3 for version 1193, got 0x%04x", eq)
}
// Test getting manager for client version 1200 (should use 1198 opcodes)
manager = managerMap.GetManagerForClient(1200, versions)
if manager == nil {
t.Fatal("Failed to get manager for client version 1200")
}
eq = manager.EmuToEQ(OP_LoginRequestMsg)
if eq != 0x00C1 {
t.Errorf("Expected 0x00C1 for version 1200, got 0x%04x", eq)
}
// Test getting manager for client version 1210 (should use 1208 opcodes)
manager = managerMap.GetManagerForClient(1210, versions)
if manager == nil {
t.Fatal("Failed to get manager for client version 1210")
}
eq = manager.EmuToEQ(OP_LoginRequestMsg)
if eq != 0x00D1 {
t.Errorf("Expected 0x00D1 for version 1210, got 0x%04x", eq)
}
}
func TestNameSearch(t *testing.T) {
// Test finding opcodes by name
opcode := NameSearch("OP_LoginRequestMsg")
if opcode != OP_LoginRequestMsg {
t.Errorf("Expected %v, got %v", OP_LoginRequestMsg, opcode)
}
opcode = NameSearch("OP_WorldListMsg")
if opcode != OP_WorldListMsg {
t.Errorf("Expected %v, got %v", OP_WorldListMsg, opcode)
}
// Test unknown name
opcode = NameSearch("OP_NonExistent")
if opcode != OP_Unknown {
t.Errorf("Expected OP_Unknown for non-existent name, got %v", opcode)
}
}
// BenchmarkOpcodeConversionV2 benchmarks the new implementation
func BenchmarkOpcodeConversionV2(b *testing.B) {
manager := NewRegularOpcodeManager(1193)
opcodes := make(map[string]uint16)
// Add many opcodes for benchmarking
for i := uint16(1); i <= 100; i++ {
name := OpcodeNames[EmuOpcode(i)]
if name == "" {
name = fmt.Sprintf("OP_Test%d", i)
OpcodeNames[EmuOpcode(i)] = name
}
opcodes[name] = i
}
manager.LoadOpcodes(opcodes)
b.ResetTimer()
b.Run("EmuToEQ", func(b *testing.B) {
for i := 0; i < b.N; i++ {
_ = manager.EmuToEQ(EmuOpcode(i%100 + 1))
}
})
b.Run("EQToEmu", func(b *testing.B) {
for i := 0; i < b.N; i++ {
_ = manager.EQToEmu(uint16(i%100 + 1))
}
})
b.Run("GetOpcodeVersion", func(b *testing.B) {
versionMap := OpcodeVersionMap{
1: 546,
547: 889,
890: 1027,
1028: 1048,
1049: 1095,
1096: 1184,
1185: 1197,
1198: 1207,
1208: 1211,
1212: 9999,
}
for i := 0; i < b.N; i++ {
_ = GetOpcodeVersion(uint16(i%2000+1), versionMap)
}
})
}

299
packet.go
View File

@ -1,299 +0,0 @@
// Package eq2net implements the EverQuest 2 network protocol
package eq2net
import (
"encoding/binary"
"fmt"
"net"
"time"
)
// Protocol opcodes for low-level packet control
const (
OPSessionRequest = 0x0001
OPSessionResponse = 0x0002
OPCombined = 0x0003
OPSessionDisconnect = 0x0005
OPKeepAlive = 0x0006
OPSessionStatRequest = 0x0007
OPSessionStatResponse = 0x0008
OPPacket = 0x0009
OPFragment = 0x000D
OPOutOfOrderAck = 0x0011
OPAck = 0x0015
OPAppCombined = 0x0019
OPOutOfSession = 0x001D
)
// EQPacket is the base packet type for all EverQuest packets
type EQPacket struct {
// Core packet data
Buffer []byte
Size uint32
Opcode uint16
// Network information
SrcIP net.IP
DstIP net.IP
SrcPort uint16
DstPort uint16
// Metadata
Priority uint32
Timestamp time.Time
Version int16
}
// NewEQPacket creates a new packet with the specified opcode and data
func NewEQPacket(opcode uint16, data []byte) *EQPacket {
p := &EQPacket{
Opcode: opcode,
Timestamp: time.Now(),
}
if len(data) > 0 {
p.Buffer = make([]byte, len(data))
copy(p.Buffer, data)
p.Size = uint32(len(data))
}
return p
}
// TotalSize returns the total packet size including opcode
func (p *EQPacket) TotalSize() uint32 {
return p.Size + 2 // +2 for opcode
}
// SetNetworkInfo sets the source and destination network information
func (p *EQPacket) SetNetworkInfo(srcIP net.IP, srcPort uint16, dstIP net.IP, dstPort uint16) {
p.SrcIP = srcIP
p.SrcPort = srcPort
p.DstIP = dstIP
p.DstPort = dstPort
}
// CopyInfo copies network and timing information from another packet
func (p *EQPacket) CopyInfo(other *EQPacket) {
p.SrcIP = other.SrcIP
p.SrcPort = other.SrcPort
p.DstIP = other.DstIP
p.DstPort = other.DstPort
p.Timestamp = other.Timestamp
p.Version = other.Version
}
// EQProtocolPacket handles low-level protocol operations
type EQProtocolPacket struct {
*EQPacket
// Protocol state flags
Compressed bool
Prepared bool
Encrypted bool
Acked bool
// Reliability tracking
SentTime time.Time
AttemptCount uint8
Sequence uint32
}
// NewEQProtocolPacket creates a new protocol packet
func NewEQProtocolPacket(opcode uint16, data []byte) *EQProtocolPacket {
return &EQProtocolPacket{
EQPacket: NewEQPacket(opcode, data),
}
}
// NewEQProtocolPacketFromBuffer creates a protocol packet from raw buffer
func NewEQProtocolPacketFromBuffer(buffer []byte, opcodeOverride int) (*EQProtocolPacket, error) {
if len(buffer) < 2 {
return nil, fmt.Errorf("buffer too small for opcode")
}
var opcode uint16
var dataOffset int
if opcodeOverride >= 0 {
opcode = uint16(opcodeOverride)
dataOffset = 0
} else {
opcode = binary.BigEndian.Uint16(buffer[:2])
dataOffset = 2
}
var data []byte
if len(buffer) > dataOffset {
data = buffer[dataOffset:]
}
return NewEQProtocolPacket(opcode, data), nil
}
// Serialize writes the protocol packet to a byte buffer
func (p *EQProtocolPacket) Serialize(offset int) []byte {
// Allocate buffer for opcode + data
result := make([]byte, 2+len(p.Buffer)-offset)
// Write opcode (big-endian)
if p.Opcode > 0xFF {
binary.BigEndian.PutUint16(result[0:2], p.Opcode)
} else {
result[0] = 0
result[1] = byte(p.Opcode)
}
// Copy packet data
if len(p.Buffer) > offset {
copy(result[2:], p.Buffer[offset:])
}
return result
}
// IsProtocolPacket checks if the opcode is a valid protocol packet
func IsProtocolPacket(opcode uint16) bool {
switch opcode {
case OPSessionRequest, OPSessionDisconnect, OPKeepAlive,
OPSessionStatResponse, OPPacket, OPCombined, OPFragment,
OPAck, OPOutOfOrderAck, OPOutOfSession:
return true
default:
return false
}
}
// Copy creates a deep copy of the protocol packet
func (p *EQProtocolPacket) Copy() *EQProtocolPacket {
newPacket := &EQProtocolPacket{
EQPacket: NewEQPacket(p.Opcode, p.Buffer),
Compressed: p.Compressed,
Prepared: p.Prepared,
Encrypted: p.Encrypted,
Acked: p.Acked,
SentTime: p.SentTime,
AttemptCount: p.AttemptCount,
Sequence: p.Sequence,
}
newPacket.CopyInfo(p.EQPacket)
return newPacket
}
// EQApplicationPacket represents high-level application packets
type EQApplicationPacket struct {
*EQPacket
// Cached emulator opcode
EmuOpcode uint16
// Opcode size (1 or 2 bytes)
OpcodeSize uint8
}
// DefaultOpcodeSize is the default size for application opcodes
var DefaultOpcodeSize uint8 = 2
// NewEQApplicationPacket creates a new application packet
func NewEQApplicationPacket(opcode uint16, data []byte) *EQApplicationPacket {
return &EQApplicationPacket{
EQPacket: NewEQPacket(opcode, data),
EmuOpcode: opcode,
OpcodeSize: DefaultOpcodeSize,
}
}
// Serialize writes the application packet to a byte buffer
func (p *EQApplicationPacket) Serialize() []byte {
opcodeBytes := p.OpcodeSize
// Special handling for opcodes with low byte = 0x00
if p.OpcodeSize == 2 && (p.Opcode&0x00FF) == 0 {
opcodeBytes = 3
}
result := make([]byte, uint32(opcodeBytes)+p.Size)
if p.OpcodeSize == 1 {
result[0] = byte(p.Opcode)
} else {
if (p.Opcode & 0x00FF) == 0 {
result[0] = 0
binary.BigEndian.PutUint16(result[1:3], p.Opcode)
} else {
binary.BigEndian.PutUint16(result[0:2], p.Opcode)
}
}
// Copy data after opcode
if p.Size > 0 {
copy(result[opcodeBytes:], p.Buffer)
}
return result
}
// Copy creates a deep copy of the application packet
func (p *EQApplicationPacket) Copy() *EQApplicationPacket {
newPacket := &EQApplicationPacket{
EQPacket: NewEQPacket(p.Opcode, p.Buffer),
EmuOpcode: p.EmuOpcode,
OpcodeSize: p.OpcodeSize,
}
newPacket.CopyInfo(p.EQPacket)
return newPacket
}
// PacketCombiner handles combining multiple packets for efficient transmission
type PacketCombiner struct {
maxSize int
}
// NewPacketCombiner creates a new packet combiner with max size limit
func NewPacketCombiner(maxSize int) *PacketCombiner {
return &PacketCombiner{maxSize: maxSize}
}
// CombineProtocolPackets combines multiple protocol packets into one
func (c *PacketCombiner) CombineProtocolPackets(packets []*EQProtocolPacket) *EQProtocolPacket {
if len(packets) == 0 {
return nil
}
if len(packets) == 1 {
return packets[0]
}
// Calculate total size needed
totalSize := 0
for _, p := range packets {
totalSize += 1 + int(p.TotalSize()) // 1 byte for size prefix
}
if totalSize > c.maxSize {
return nil // Too large to combine
}
// Build combined packet buffer
buffer := make([]byte, totalSize)
offset := 0
for _, p := range packets {
// Write size prefix
buffer[offset] = byte(p.TotalSize())
offset++
// Serialize packet
serialized := p.Serialize(0)
copy(buffer[offset:], serialized)
offset += len(serialized)
}
// Create combined packet
combined := NewEQProtocolPacket(OPCombined, buffer)
if len(packets) > 0 {
combined.CopyInfo(packets[0].EQPacket)
}
return combined
}

207
packet_test.go
View File

@ -1,207 +0,0 @@
package eq2net
import (
"bytes"
"testing"
)
func TestEQPacket(t *testing.T) {
data := []byte("Hello, World!")
packet := NewEQPacket(OPPacket, data)
if packet.Opcode != OPPacket {
t.Errorf("Expected opcode %04x, got %04x", OPPacket, packet.Opcode)
}
if packet.Size != uint32(len(data)) {
t.Errorf("Expected size %d, got %d", len(data), packet.Size)
}
if !bytes.Equal(packet.Buffer, data) {
t.Errorf("Buffer mismatch")
}
}
func TestProtocolPacketSerialization(t *testing.T) {
data := []byte("Test Data")
packet := NewEQProtocolPacket(OPPacket, data)
serialized := packet.Serialize(0)
// Check opcode (first 2 bytes)
if len(serialized) < 2 {
t.Fatal("Serialized packet too small")
}
// OPPacket = 0x0009, should be [0x00, 0x09] in big-endian
if serialized[0] != 0x00 || serialized[1] != 0x09 {
t.Errorf("Opcode not serialized correctly: %02x %02x", serialized[0], serialized[1])
}
// Check data
if !bytes.Equal(serialized[2:], data) {
t.Error("Data not serialized correctly")
}
}
func TestCRC16(t *testing.T) {
tests := []struct {
data []byte
key uint32
}{
{[]byte{0x00, 0x09, 0x00, 0x00, 0x00}, 0x12345678},
{[]byte{0x00, 0x01}, 0},
{[]byte("Hello"), 0},
}
for _, tt := range tests {
// Just test that CRC16 produces consistent results
got1 := CRC16(tt.data, len(tt.data), tt.key)
got2 := CRC16(tt.data, len(tt.data), tt.key)
if got1 != got2 {
t.Errorf("CRC16 not consistent: %04x != %04x", got1, got2)
}
// Test that different keys produce different CRCs
if tt.key != 0 {
got3 := CRC16(tt.data, len(tt.data), 0)
if got1 == got3 {
t.Errorf("Different keys should produce different CRCs")
}
}
}
}
func TestValidateCRC(t *testing.T) {
// Test session packet (CRC exempt)
sessionPacket := []byte{0x00, byte(OPSessionRequest), 0x00, 0x00}
if !ValidateCRC(sessionPacket, 0) {
t.Error("Session packet should be CRC exempt")
}
// Test packet with valid CRC
data := []byte{0x00, 0x09, 0x48, 0x65, 0x6C, 0x6C, 0x6F} // "Hello"
crc := CRC16(data, len(data), 0x1234)
dataWithCRC := append(data, byte(crc>>8), byte(crc))
if !ValidateCRC(dataWithCRC, 0x1234) {
t.Error("Packet with valid CRC should validate")
}
// Test packet with invalid CRC
dataWithCRC[len(dataWithCRC)-1] ^= 0xFF // Corrupt CRC
if ValidateCRC(dataWithCRC, 0x1234) {
t.Error("Packet with invalid CRC should not validate")
}
}
func TestCompression(t *testing.T) {
// Test simple encoding (small packet)
smallData := []byte{0x00, 0x09, 0x01, 0x02, 0x03}
compressed, err := CompressPacket(smallData)
if err != nil {
t.Fatal(err)
}
if compressed[2] != CompressionFlagSimple {
t.Errorf("Small packet should use simple encoding, got flag %02x", compressed[2])
}
decompressed, err := DecompressPacket(compressed)
if err != nil {
t.Fatal(err)
}
if !bytes.Equal(decompressed, smallData) {
t.Error("Decompressed data doesn't match original")
}
// Test zlib compression (large packet)
largeData := make([]byte, 100)
largeData[0] = 0x00
largeData[1] = 0x09
for i := 2; i < len(largeData); i++ {
largeData[i] = byte(i)
}
compressed, err = CompressPacket(largeData)
if err != nil {
t.Fatal(err)
}
if compressed[2] != CompressionFlagZlib {
t.Errorf("Large packet should use zlib compression, got flag %02x", compressed[2])
}
decompressed, err = DecompressPacket(compressed)
if err != nil {
t.Fatal(err)
}
if !bytes.Equal(decompressed, largeData) {
t.Error("Decompressed large data doesn't match original")
}
}
func TestChatEncryption(t *testing.T) {
// Test chat encoding/decoding
original := []byte{0x00, 0x09, 0x48, 0x65, 0x6C, 0x6C, 0x6F, 0x21} // "Hello!"
key := uint32(0x12345678)
encoded := ChatEncode(original, key)
if bytes.Equal(encoded[2:], original[2:]) {
t.Error("Encoded data should differ from original")
}
decoded := ChatDecode(encoded, key)
if !bytes.Equal(decoded, original) {
t.Errorf("Decoded data doesn't match original\nOriginal: %v\nDecoded: %v", original, decoded)
}
// Test exempt packet types
exemptPacket := []byte{0x00, 0x01, 0x12, 0x34}
encoded = ChatEncode(exemptPacket, key)
if !bytes.Equal(encoded, exemptPacket) {
t.Error("Exempt packet should not be encoded")
}
}
func TestPacketCombiner(t *testing.T) {
combiner := NewPacketCombiner(256)
p1 := NewEQProtocolPacket(OPPacket, []byte{0x01, 0x02})
p2 := NewEQProtocolPacket(OPAck, []byte{0x03, 0x04})
p3 := NewEQProtocolPacket(OPKeepAlive, []byte{0x05})
combined := combiner.CombineProtocolPackets([]*EQProtocolPacket{p1, p2, p3})
if combined == nil {
t.Fatal("Failed to combine packets")
}
if combined.Opcode != OPCombined {
t.Errorf("Combined packet should have opcode %04x, got %04x", OPCombined, combined.Opcode)
}
// Verify combined packet structure
buffer := combined.Buffer
offset := 0
// First packet
if buffer[offset] != byte(p1.TotalSize()) {
t.Errorf("First packet size incorrect: %d", buffer[offset])
}
offset++
offset += int(p1.TotalSize())
// Second packet
if buffer[offset] != byte(p2.TotalSize()) {
t.Errorf("Second packet size incorrect: %d", buffer[offset])
}
offset++
offset += int(p2.TotalSize())
// Third packet
if buffer[offset] != byte(p3.TotalSize()) {
t.Errorf("Third packet size incorrect: %d", buffer[offset])
}
}

407
server.go
View File

@ -1,407 +0,0 @@
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
View File

@ -1,567 +0,0 @@
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
View File

@ -1,506 +0,0 @@
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
View File

@ -1,194 +0,0 @@
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
}