EQ2/Emu
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

clean EQStream

Browse Source
This commit is contained in:
Sky Johnson 2025-08-31 22:51:17 -05:00
parent 7e9f7eb061
commit d00e7cce75
2 changed files with 1953 additions and 1046 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2221
source/common/EQStream.cpp
View File

File diff suppressed because it is too large Load Diff

682
source/common/EQStream.h
View File

@ -1,22 +1,6 @@
/* // EQ2Emulator: Everquest II Server Emulator
EQ2Emulator: Everquest II Server Emulator // Copyright (C) 2007 EQ2EMulator Development Team
Copyright (C) 2007 EQ2EMulator Development Team (http://www.eq2emulator.net) // Licensed under GPL v3 - see <http://www.gnu.org/licenses/>
This file is part of EQ2Emulator.
EQ2Emulator is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
EQ2Emulator is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with EQ2Emulator. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _EQPROTOCOL_H #ifndef _EQPROTOCOL_H
#define _EQPROTOCOL_H #define _EQPROTOCOL_H
@ -24,12 +8,15 @@
#include <vector> #include <vector>
#include <deque> #include <deque>
#include <queue> #include <queue>
#include <map> #include <map>
#include <set> #include <set>
#ifndef WIN32 #include <memory>
#include <cstdint>
// Unix networking headers
#include <netinet/in.h> #include <netinet/in.h>
#endif
// Project headers
#include "EQPacket.h" #include "EQPacket.h"
#include "Mutex.h" #include "Mutex.h"
#include "opcodemgr.h" #include "opcodemgr.h"
@ -38,338 +25,499 @@
#include "Crypto.h" #include "Crypto.h"
#include "zlib.h" #include "zlib.h"
#include "timer.h" #include "timer.h"
#ifdef WRITE_PACKETS #ifdef WRITE_PACKETS
#include <stdarg.h> #include <stdarg.h>
#endif #endif
using namespace std; using namespace std;
// Forward declarations
class OpcodeManager;
class EQStreamFactory;
/**
* EverQuest stream connection states.
* Represents the current state of a network stream connection.
*/
typedef enum { typedef enum {
ESTABLISHED, ESTABLISHED, // Active connection ready for data
WAIT_CLOSE, WAIT_CLOSE, // Waiting for graceful close
CLOSING, CLOSING, // In process of closing
DISCONNECTING, DISCONNECTING, // Actively disconnecting
CLOSED CLOSED // Connection fully closed
} EQStreamState; } EQStreamState;
#define FLAG_COMPRESSED 0x01 // Packet flags
#define FLAG_ENCODED 0x04 #define FLAG_COMPRESSED 0x01 // Packet is compressed
#define FLAG_ENCODED 0x04 // Packet is encoded/encrypted
#define RATEBASE 1048576 // 1 MB // Rate limiting and bandwidth constants
#define DECAYBASE 78642 // RATEBASE/10 #define RATEBASE 1048576 // Base rate: 1 MB
#define DECAYBASE 78642 // Decay rate: RATEBASE/10
// Retransmission timing constants
#ifndef RETRANSMIT_TIMEOUT_MULT #ifndef RETRANSMIT_TIMEOUT_MULT
#define RETRANSMIT_TIMEOUT_MULT 3.0 #define RETRANSMIT_TIMEOUT_MULT 3.0 // Timeout multiplier
#endif #endif
#ifndef RETRANSMIT_TIMEOUT_MAX #ifndef RETRANSMIT_TIMEOUT_MAX
#define RETRANSMIT_TIMEOUT_MAX 5000 #define RETRANSMIT_TIMEOUT_MAX 5000 // Maximum retransmit timeout (ms)
#endif #endif
#ifndef AVERAGE_DELTA_MAX #ifndef AVERAGE_DELTA_MAX
#define AVERAGE_DELTA_MAX 2500 #define AVERAGE_DELTA_MAX 2500 // Maximum average delta (ms)
#endif #endif
#pragma pack(1) #pragma pack(1)
/**
* Session request packet structure.
* Sent by client to initiate a new session.
*/
struct SessionRequest { struct SessionRequest {
uint32 UnknownA; uint32 UnknownA; // Unknown field A
uint32 Session; uint32 Session; // Requested session ID
uint32 MaxLength; uint32 MaxLength; // Maximum packet length
}; };
/**
* Session response packet structure.
* Sent by server in response to session request.
*/
struct SessionResponse { struct SessionResponse {
uint32 Session; uint32 Session; // Assigned session ID
uint32 Key; uint32 Key; // Encryption/authentication key
uint8 UnknownA; uint8 UnknownA; // Unknown field A
uint8 Format; uint8 Format; // Packet format version
uint8 UnknownB; uint8 UnknownB; // Unknown field B
uint32 MaxLength; uint32 MaxLength; // Maximum packet length
uint32 UnknownD; uint32 UnknownD; // Unknown field D
}; };
//Deltas are in ms, representing round trip times /**
* Client-side session statistics.
* Deltas are in milliseconds, representing round trip times.
*/
struct ClientSessionStats { struct ClientSessionStats {
/*000*/ uint16 RequestID; /*000*/ uint16 RequestID; // Statistics request ID
/*002*/ uint32 last_local_delta; /*002*/ uint32 last_local_delta; // Last local round trip time
/*006*/ uint32 average_delta; /*006*/ uint32 average_delta; // Average round trip time
/*010*/ uint32 low_delta; /*010*/ uint32 low_delta; // Lowest recorded round trip time
/*014*/ uint32 high_delta; /*014*/ uint32 high_delta; // Highest recorded round trip time
/*018*/ uint32 last_remote_delta; /*018*/ uint32 last_remote_delta; // Last remote round trip time
/*022*/ uint64 packets_sent; /*022*/ uint64 packets_sent; // Total packets sent
/*030*/ uint64 packets_recieved; /*030*/ uint64 packets_recieved; // Total packets received
/*038*/ /*038*/
}; };
/**
* Server-side session statistics.
* Provides server perspective on connection quality.
*/
struct ServerSessionStats { struct ServerSessionStats {
uint16 RequestID; uint16 RequestID; // Statistics request ID
uint32 current_time; uint32 current_time; // Current server time
uint32 unknown1; uint32 unknown1; // Unknown field 1
uint32 received_packets; uint32 received_packets; // Packets received by server
uint32 unknown2; uint32 unknown2; // Unknown field 2
uint32 sent_packets; uint32 sent_packets; // Packets sent by server
uint32 unknown3; uint32 unknown3; // Unknown field 3
uint32 sent_packets2; uint32 sent_packets2; // Duplicate sent packets count
uint32 unknown4; uint32 unknown4; // Unknown field 4
uint32 received_packets2; uint32 received_packets2; // Duplicate received packets count
}; };
#pragma pack() #pragma pack()
class OpcodeManager; // External opcode manager
extern OpcodeManager *EQNetworkOpcodeManager; extern OpcodeManager* EQNetworkOpcodeManager;
class EQStreamFactory;
/**
* Types of EverQuest network streams.
* Each type handles different aspects of the game protocol.
*/
typedef enum { typedef enum {
UnknownStream=0, UnknownStream = 0, // Unidentified stream type
LoginStream, LoginStream, // Login server authentication
WorldStream, WorldStream, // World server communication
ZoneStream, ZoneStream, // Zone server gameplay
ChatOrMailStream, ChatOrMailStream, // Combined chat/mail (legacy)
ChatStream, ChatStream, // Chat system only
MailStream, MailStream, // Mail system only
EQ2Stream, EQ2Stream, // EverQuest 2 specific stream
} EQStreamType; } EQStreamType;
/**
* EverQuest network stream class.
* Handles reliable UDP communication with sequence numbers, acknowledgments,
* compression, encryption, and packet combining for EverQuest protocols.
*/
class EQStream { class EQStream {
protected: protected:
typedef enum { /**
SeqPast, * Sequence number ordering enumeration.
SeqInOrder, * Used to determine packet sequence relationships.
SeqFuture */
} SeqOrder; typedef enum {
SeqPast, // Sequence is from the past (duplicate/old)
SeqInOrder, // Sequence is the expected next sequence
SeqFuture // Sequence is from the future (out of order)
} SeqOrder;
uint32 received_packets; // Packet statistics
uint32 sent_packets; uint32 received_packets; // Total packets received
uint32 remote_ip; uint32 sent_packets; // Total packets sent
uint16 remote_port;
uint8 buffer[8192];
unsigned char *oversize_buffer;
uint32 oversize_offset,oversize_length;
unsigned char *rogue_buffer;
uint32 roguebuf_offset,roguebuf_size;
uint8 app_opcode_size;
EQStreamType StreamType;
bool compressed,encoded;
unsigned char write_buffer[2048]; // Remote endpoint information
uint32 remote_ip; // Remote IP address
uint16 remote_port; // Remote port number
uint32 retransmittimer; // Packet buffers
uint32 retransmittimeout; uint8 buffer[8192]; // Main packet buffer
//uint32 buffer_len; unsigned char* oversize_buffer; // Buffer for oversized packets
uint32 oversize_offset; // Offset in oversize buffer
uint32 oversize_length; // Length of oversize buffer data
unsigned char* rogue_buffer; // Buffer for rogue/malformed packets
uint32 roguebuf_offset; // Offset in rogue buffer
uint32 roguebuf_size; // Size of rogue buffer
uint16 sessionAttempts; // Protocol configuration
uint16 reconnectAttempt; uint8 app_opcode_size; // Size of application opcodes
bool streamactive; EQStreamType StreamType; // Type of this stream
bool compressed; // Stream supports compression
bool encoded; // Stream supports encoding/encryption
uint32 Session, Key; // Write buffer for outgoing packets
uint16 NextInSeq; unsigned char write_buffer[2048];
uint16 NextOutSeq;
uint16 SequencedBase; //the sequence number of SequencedQueue[0]
uint32 MaxLen;
uint16 MaxSends;
int8 timeout_delays;
uint8 active_users; //how many things are actively using this // Retransmission timing
Mutex MInUse; uint32 retransmittimer; // Current retransmit timer
uint32 retransmittimeout; // Retransmit timeout value
// Session management
uint16 sessionAttempts; // Number of session attempts
uint16 reconnectAttempt; // Number of reconnect attempts
bool streamactive; // Stream is actively connected
// Session state
uint32 Session; // Session ID
uint32 Key; // Session encryption key
uint16 NextInSeq; // Next expected incoming sequence
uint16 NextOutSeq; // Next outgoing sequence number
uint16 SequencedBase; // Base sequence for SequencedQueue[0]
uint32 MaxLen; // Maximum packet length
uint16 MaxSends; // Maximum send attempts
int8 timeout_delays; // Number of timeout delays accumulated
// Thread safety for stream usage
uint8 active_users; // Number of active users of this stream
Mutex MInUse; // Mutex for usage tracking
#ifdef WRITE_PACKETS #ifdef WRITE_PACKETS
FILE* write_packets = NULL; // Packet logging for debugging
char GetChar(uchar in); FILE* write_packets = nullptr; // File handle for packet dumps
void WriteToFile(char* pFormat, ...); char GetChar(uchar in); // Convert byte to printable character
void WritePackets(const char* opcodeName, uchar* data, int32 size, bool outgoing); void WriteToFile(char* pFormat, ...); // Write formatted data to file
void WritePackets(EQ2Packet* app, bool outgoing); void WritePackets(const char* opcodeName, uchar* data, int32 size, bool outgoing);
Mutex MWritePackets; void WritePackets(EQ2Packet* app, bool outgoing);
Mutex MWritePackets; // Mutex for packet writing
#endif #endif
EQStreamState State; // Stream connection state
Mutex MState; EQStreamState State; // Current connection state
Mutex MState; // Mutex for state access
uint32 LastPacket; // Timing and general variables
Mutex MVarlock; uint32 LastPacket; // Timestamp of last packet activity
Mutex MVarlock; // General variable lock
EQApplicationPacket* CombinedAppPacket; // Combined application packet handling
Mutex MCombinedAppPacket; EQApplicationPacket* CombinedAppPacket; // Current combined packet
Mutex MCombinedAppPacket; // Mutex for combined packet access
long LastSeqSent; // Sequence number tracking
Mutex MLastSeqSent; long LastSeqSent; // Last sequence number sent
void SetLastSeqSent(uint32); Mutex MLastSeqSent; // Mutex for last sequence sent
void SetLastSeqSent(uint32 seq); // Set last sent sequence number
// Ack sequence tracking. // Acknowledgment sequence tracking
long MaxAckReceived,NextAckToSend,LastAckSent; long MaxAckReceived; // Highest ack sequence received
long GetMaxAckReceived(); long NextAckToSend; // Next ack sequence to send
long GetNextAckToSend(); long LastAckSent; // Last ack sequence sent
long GetLastAckSent();
void SetMaxAckReceived(uint32 seq);
void SetNextAckToSend(uint32);
void SetLastAckSent(uint32);
Mutex MAcks; // Acknowledgment accessor methods
long GetMaxAckReceived();
long GetNextAckToSend();
long GetLastAckSent();
void SetMaxAckReceived(uint32 seq);
void SetNextAckToSend(uint32 seq);
void SetLastAckSent(uint32 seq);
// Packets waiting to be sent Mutex MAcks; // Mutex for acknowledgment data
queue<EQProtocolPacket*> NonSequencedQueue;
deque<EQProtocolPacket*> SequencedQueue;
map<uint16, EQProtocolPacket *> OutOfOrderpackets;
Mutex MOutboundQueue;
// Packes waiting to be processed // Outbound packet queues
deque<EQApplicationPacket *> InboundQueue; queue<EQProtocolPacket*> NonSequencedQueue; // Non-sequenced packets
Mutex MInboundQueue; deque<EQProtocolPacket*> SequencedQueue; // Sequenced packets
map<uint16, EQProtocolPacket*> OutOfOrderpackets; // Out-of-order packets
Mutex MOutboundQueue; // Mutex for outbound queues
static uint16 MaxWindowSize; // Inbound packet queue
deque<EQApplicationPacket*> InboundQueue; // Packets waiting processing
Mutex MInboundQueue; // Mutex for inbound queue
sint32 BytesWritten; // Static configuration
static uint16 MaxWindowSize; // Maximum window size for flow control
Mutex MRate; // Rate limiting and flow control
sint32 RateThreshold; sint32 BytesWritten; // Bytes written this period
sint32 DecayRate; Mutex MRate; // Mutex for rate limiting
uint32 AverageDelta; sint32 RateThreshold; // Rate limiting threshold
sint32 DecayRate; // Rate decay per time period
uint32 AverageDelta; // Average round-trip time
EQStreamFactory *Factory; // Factory reference
EQStreamFactory* Factory; // Factory that created this stream
public: public:
Mutex MCombineQueueLock; // EQ2-specific packet combining
bool CheckCombineQueue(); Mutex MCombineQueueLock; // Mutex for combine queue operations
deque<EQ2Packet*> combine_queue; bool CheckCombineQueue(); // Check and process combine queue
Timer* combine_timer; deque<EQ2Packet*> combine_queue; // Queue of packets to combine
Timer* combine_timer; // Timer for combine operations
Crypto* crypto; // Encryption and compression
int8 EQ2_Compress(EQ2Packet* app, int8 offset = 3); Crypto* crypto; // Cryptographic handler
z_stream stream; int8 EQ2_Compress(EQ2Packet* app, int8 offset = 3); // Compress EQ2 packet
uchar* stream_buffer; z_stream stream; // zlib compression stream
int32 stream_buffer_size; uchar* stream_buffer; // Compression buffer
bool eq2_compressed; int32 stream_buffer_size; // Size of compression buffer
int8 compressed_offset; bool eq2_compressed; // Stream uses EQ2 compression
int16 client_version; int8 compressed_offset; // Offset for compressed data
int16 GetClientVersion(){ return client_version; }
void SetClientVersion(int16 version){ client_version = version; } // Client version management
void ResetSessionAttempts() { reconnectAttempt = 0; } int16 client_version; // Client protocol version
bool HasSessionAttempts() { return reconnectAttempt>0; } int16 GetClientVersion() { return client_version; }
EQStream() { init(); remote_ip = 0; remote_port = 0; State = CLOSED; StreamType = UnknownStream; compressed = true; void SetClientVersion(int16 version) { client_version = version; }
encoded = false; app_opcode_size = 2;}
EQStream(sockaddr_in addr); // Session attempt management
virtual ~EQStream() { void ResetSessionAttempts() { reconnectAttempt = 0; }
MOutboundQueue.lock(); bool HasSessionAttempts() { return reconnectAttempt > 0; }
SetState(CLOSED);
MOutboundQueue.unlock(); // Constructors
RemoveData(); EQStream() {
safe_delete(crypto); init();
safe_delete(combine_timer); remote_ip = 0;
safe_delete(resend_que_timer); remote_port = 0;
safe_delete_array(oversize_buffer); State = CLOSED;
safe_delete_array(rogue_buffer); StreamType = UnknownStream;
deque<EQ2Packet*>::iterator cmb; compressed = true;
MCombineQueueLock.lock(); encoded = false;
for (cmb = combine_queue.begin(); cmb != combine_queue.end(); cmb++){ app_opcode_size = 2;
safe_delete(*cmb); }
}
MCombineQueueLock.unlock(); EQStream(sockaddr_in addr);
deflateEnd(&stream); // Destructor
map<int16, EQProtocolPacket*>::iterator oop; virtual ~EQStream() {
for (oop = OutOfOrderpackets.begin(); oop != OutOfOrderpackets.end(); oop++){ // Close stream safely
safe_delete(oop->second); MOutboundQueue.lock();
} SetState(CLOSED);
#ifdef WRITE_PACKETS MOutboundQueue.unlock();
if (write_packets)
fclose(write_packets); // Clean up data structures
#endif RemoveData();
// Clean up allocated resources
safe_delete(crypto);
safe_delete(combine_timer);
safe_delete(resend_que_timer);
safe_delete_array(oversize_buffer);
safe_delete_array(rogue_buffer);
// Clean up combine queue
MCombineQueueLock.lock();
for (auto cmb = combine_queue.begin(); cmb != combine_queue.end(); ++cmb) {
safe_delete(*cmb);
} }
inline void SetFactory(EQStreamFactory *f) { Factory=f; } MCombineQueueLock.unlock();
void init(bool resetSession = true);
void SetMaxLen(uint32 length) { MaxLen=length; }
int8 getTimeoutDelays(){ return timeout_delays; }
void addTimeoutDelay(){ timeout_delays++; }
void EQ2QueuePacket(EQ2Packet* app, bool attempted_combine = false);
void PreparePacket(EQ2Packet* app, int8 offset = 0);
void UnPreparePacket(EQ2Packet* app);
void EncryptPacket(EQ2Packet* app, int8 compress_offset, int8 offset);
void FlushCombinedPacket();
void SendPacket(EQApplicationPacket *p);
void QueuePacket(EQProtocolPacket *p);
void SendPacket(EQProtocolPacket *p);
vector<EQProtocolPacket *> convert(EQApplicationPacket *p);
void NonSequencedPush(EQProtocolPacket *p);
void SequencedPush(EQProtocolPacket *p);
Mutex MResendQue; // Clean up compression stream
Mutex MCompressData; deflateEnd(&stream);
deque<EQProtocolPacket*>resend_que;
void CheckResend(int eq_fd);
void AckPackets(uint16 seq); // Clean up out-of-order packets
void Write(int eq_fd); for (auto oop = OutOfOrderpackets.begin(); oop != OutOfOrderpackets.end(); ++oop) {
safe_delete(oop->second);
}
void SetActive(bool val) { streamactive = val; } #ifdef WRITE_PACKETS
if (write_packets) {
fclose(write_packets);
}
#endif
}
// Factory and initialization
void SetFactory(EQStreamFactory* f) { Factory = f; }
void init(bool resetSession = true);
void WritePacket(int fd,EQProtocolPacket *p); // Configuration
void SetMaxLen(uint32 length) { MaxLen = length; }
int8 getTimeoutDelays() { return timeout_delays; }
void addTimeoutDelay() { timeout_delays++; }
void EncryptPacket(uchar* data, int16 size); // EQ2-specific packet handling
uint32 GetKey() { return Key; } void EQ2QueuePacket(EQ2Packet* app, bool attempted_combine = false);
void SetKey(uint32 k) { Key=k; } void PreparePacket(EQ2Packet* app, int8 offset = 0);
void SetSession(uint32 s) { Session=s; } void UnPreparePacket(EQ2Packet* app);
void SetLastPacketTime(uint32 t) {LastPacket=t;} void EncryptPacket(EQ2Packet* app, int8 compress_offset, int8 offset);
void FlushCombinedPacket();
void Process(const unsigned char *data, const uint32 length); // General packet transmission
void ProcessPacket(EQProtocolPacket *p, EQProtocolPacket* lastp=NULL); void SendPacket(EQApplicationPacket* p);
void QueuePacket(EQProtocolPacket* p);
void SendPacket(EQProtocolPacket* p);
vector<EQProtocolPacket*> convert(EQApplicationPacket* p);
void NonSequencedPush(EQProtocolPacket* p);
void SequencedPush(EQProtocolPacket* p);
bool ProcessEmbeddedPacket(uchar* pBuffer, uint16 length, int8 opcode = OP_Packet); // Resend queue management
bool HandleEmbeddedPacket(EQProtocolPacket *p, int16 offset = 2, int16 length = 0); Mutex MResendQue; // Mutex for resend queue
Mutex MCompressData; // Mutex for compression operations
deque<EQProtocolPacket*> resend_que; // Queue of packets needing resend
void CheckResend(int eq_fd); // Check and handle packet resends
EQProtocolPacket * ProcessEncryptedPacket(EQProtocolPacket *p); // Acknowledgment and writing
EQProtocolPacket * ProcessEncryptedData(uchar* data, int32 size, int16 opcode); void AckPackets(uint16 seq); // Acknowledge received packets
void Write(int eq_fd); // Write packets to socket
virtual void DispatchPacket(EQApplicationPacket *p) { p->DumpRaw(); } // Stream state management
void SetActive(bool val) { streamactive = val; }
void SendSessionResponse(); // Low-level packet operations
void SendSessionRequest(); void WritePacket(int fd, EQProtocolPacket* p);
void SendDisconnect(bool setstate = true); void EncryptPacket(uchar* data, int16 size);
void SendAck(uint16 seq);
void SendOutOfOrderAck(uint16 seq);
bool CheckTimeout(uint32 now, uint32 timeout=30) { return (LastPacket && (now-LastPacket) > timeout); } // Session management
bool Stale(uint32 now, uint32 timeout=30) { return (LastPacket && (now-LastPacket) > timeout); } uint32 GetKey() { return Key; }
void SetKey(uint32 k) { Key = k; }
void SetSession(uint32 s) { Session = s; }
void SetLastPacketTime(uint32 t) { LastPacket = t; }
void InboundQueuePush(EQApplicationPacket *p); // Packet processing
EQApplicationPacket *PopPacket(); // InboundQueuePop void Process(const unsigned char* data, uint32 length);
void InboundQueueClear(); void ProcessPacket(EQProtocolPacket* p, EQProtocolPacket* lastp = nullptr);
void OutboundQueueClear(); // Embedded packet handling
bool HasOutgoingData(); bool ProcessEmbeddedPacket(uchar* pBuffer, uint16 length, int8 opcode = OP_Packet);
void SendKeyRequest(); bool HandleEmbeddedPacket(EQProtocolPacket* p, int16 offset = 2, int16 length = 0);
int16 processRSAKey(EQProtocolPacket *p, uint16 subpacket_length = 0);
void RemoveData() { InboundQueueClear(); OutboundQueueClear(); if (CombinedAppPacket) delete CombinedAppPacket; }
// // Encryption handling
inline bool IsInUse() { bool flag; MInUse.lock(); flag=(active_users>0); MInUse.unlock(); return flag; } EQProtocolPacket* ProcessEncryptedPacket(EQProtocolPacket* p);
inline void PutInUse() { MInUse.lock(); active_users++; MInUse.unlock(); } EQProtocolPacket* ProcessEncryptedData(uchar* data, int32 size, int16 opcode);
inline void ReleaseFromUse() { MInUse.lock(); if(active_users > 0) active_users--; MInUse.unlock(); }
static SeqOrder CompareSequence(uint16 expected_seq, uint16 seq); // Virtual packet dispatch (override in derived classes)
virtual void DispatchPacket(EQApplicationPacket* p) { p->DumpRaw(); }
inline EQStreamState GetState() { return State; } // Session protocol messages
inline void SetState(EQStreamState state) { MState.lock(); State = state; MState.unlock(); } void SendSessionResponse();
void SendSessionRequest();
void SendDisconnect(bool setstate = true);
void SendAck(uint16 seq);
void SendOutOfOrderAck(uint16 seq);
inline uint32 GetRemoteIP() { return remote_ip; } // Connection health checks
inline uint32 GetrIP() { return remote_ip; } bool CheckTimeout(uint32 now, uint32 timeout = 30) {
inline uint16 GetRemotePort() { return remote_port; } return (LastPacket && (now - LastPacket) > timeout);
inline uint16 GetrPort() { return remote_port; } }
bool Stale(uint32 now, uint32 timeout = 30) {
return (LastPacket && (now - LastPacket) > timeout);
}
// Inbound queue management
void InboundQueuePush(EQApplicationPacket* p);
EQApplicationPacket* PopPacket(); // Pop packet from inbound queue
void InboundQueueClear();
static EQProtocolPacket *Read(int eq_fd, sockaddr_in *from); // Outbound queue management
void OutboundQueueClear();
bool HasOutgoingData();
void Close() { SendDisconnect(); } // Key exchange and RSA
bool CheckActive() { return (GetState()==ESTABLISHED); } void SendKeyRequest();
bool CheckClosed() { return GetState()==CLOSED; } int16 processRSAKey(EQProtocolPacket* p, uint16 subpacket_length = 0);
void SetOpcodeSize(uint8 s) { app_opcode_size = s; }
void SetStreamType(EQStreamType t);
inline const EQStreamType GetStreamType() const { return StreamType; }
void ProcessQueue(); // Data cleanup
EQProtocolPacket* RemoveQueue(uint16 seq); void RemoveData() {
InboundQueueClear();
OutboundQueueClear();
if (CombinedAppPacket) {
delete CombinedAppPacket;
CombinedAppPacket = nullptr;
}
}
void Decay(); // Usage tracking (thread-safe reference counting)
void AdjustRates(uint32 average_delta); bool IsInUse() {
Timer* resend_que_timer; bool flag;
MInUse.lock();
flag = (active_users > 0);
MInUse.unlock();
return flag;
}
void PutInUse() {
MInUse.lock();
active_users++;
MInUse.unlock();
}
void ReleaseFromUse() {
MInUse.lock();
if (active_users > 0) {
active_users--;
}
MInUse.unlock();
}
// Sequence number utilities
static SeqOrder CompareSequence(uint16 expected_seq, uint16 seq);
// State management
EQStreamState GetState() { return State; }
void SetState(EQStreamState state) {
MState.lock();
State = state;
MState.unlock();
}
// Remote endpoint access
uint32 GetRemoteIP() { return remote_ip; }
uint32 GetrIP() { return remote_ip; } // Legacy alias
uint16 GetRemotePort() { return remote_port; }
uint16 GetrPort() { return remote_port; } // Legacy alias
// Static packet reading
static EQProtocolPacket* Read(int eq_fd, sockaddr_in* from);
// Connection management
void Close() { SendDisconnect(); }
bool CheckActive() { return (GetState() == ESTABLISHED); }
bool CheckClosed() { return GetState() == CLOSED; }
// Stream configuration
void SetOpcodeSize(uint8 s) { app_opcode_size = s; }
void SetStreamType(EQStreamType t);
EQStreamType GetStreamType() const { return StreamType; }
// Queue processing
void ProcessQueue();
EQProtocolPacket* RemoveQueue(uint16 seq);
// Rate limiting and flow control
void Decay();
void AdjustRates(uint32 average_delta);
// Resend timer
Timer* resend_que_timer; // Timer for checking resend queue
}; };
#endif #endif