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

clean EQStreamFactory

Browse Source
This commit is contained in:
Sky Johnson 2025-08-31 22:18:35 -05:00
parent a8c533f012
commit 7e9f7eb061
2 changed files with 442 additions and 301 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

390
source/common/EQStreamFactory.cpp
View File

@ -1,138 +1,163 @@
/* // 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/>.
*/
#include "EQStreamFactory.h" #include "EQStreamFactory.h"
#include "Log.h" #include "Log.h"
#ifdef WIN32 // Unix/Linux networking headers
#include <WinSock2.h>
#include <windows.h>
#include <process.h>
#include <io.h>
#include <stdio.h>
#else
#include <sys/socket.h> #include <sys/socket.h>
#include <netinet/in.h> #include <netinet/in.h>
#include <sys/select.h> #include <sys/select.h>
#include <arpa/inet.h> #include <arpa/inet.h>
#include <netdb.h> #include <netdb.h>
#include <pthread.h> #include <pthread.h>
#endif
#include <fcntl.h> #include <fcntl.h>
#include <unistd.h>
// Standard library headers
#include <fstream> #include <fstream>
#include <iostream> #include <iostream>
#include <cstring>
#include <deque>
#include <vector>
#include <algorithm>
// Project headers
#include "op_codes.h" #include "op_codes.h"
#include "EQStream.h" #include "EQStream.h"
#include "packet_dump.h" #include "packet_dump.h"
#ifdef WORLD #ifdef WORLD
#include "../WorldServer/client.h" #include "../WorldServer/client.h"
#endif #endif
using namespace std; using namespace std;
#ifdef WORLD #ifdef WORLD
extern ClientList client_list; extern ClientList client_list;
#endif #endif
/**
* Thread entry point for the packet reader loop.
*
* @param eqfs - Pointer to EQStreamFactory instance
* @return Thread return value
*/
ThreadReturnType EQStreamFactoryReaderLoop(void* eqfs) ThreadReturnType EQStreamFactoryReaderLoop(void* eqfs)
{ {
if (eqfs) { if (eqfs) {
EQStreamFactory *fs=(EQStreamFactory *)eqfs; auto fs = static_cast<EQStreamFactory*>(eqfs);
fs->ReaderLoop(); fs->ReaderLoop();
} }
THREAD_RETURN(NULL); THREAD_RETURN(nullptr);
} }
/**
* Thread entry point for the packet writer loop.
*
* @param eqfs - Pointer to EQStreamFactory instance
* @return Thread return value
*/
ThreadReturnType EQStreamFactoryWriterLoop(void* eqfs) ThreadReturnType EQStreamFactoryWriterLoop(void* eqfs)
{ {
if (eqfs) { if (eqfs) {
EQStreamFactory *fs=(EQStreamFactory *)eqfs; auto fs = static_cast<EQStreamFactory*>(eqfs);
fs->WriterLoop(); fs->WriterLoop();
} }
THREAD_RETURN(NULL); THREAD_RETURN(nullptr);
} }
/**
* Thread entry point for the packet combining loop.
*
* @param eqfs - Pointer to EQStreamFactory instance
* @return Thread return value
*/
ThreadReturnType EQStreamFactoryCombinePacketLoop(void* eqfs) ThreadReturnType EQStreamFactoryCombinePacketLoop(void* eqfs)
{ {
if (eqfs) { if (eqfs) {
EQStreamFactory *fs=(EQStreamFactory *)eqfs; auto fs = static_cast<EQStreamFactory*>(eqfs);
fs->CombinePacketLoop(); fs->CombinePacketLoop();
} }
THREAD_RETURN(NULL); THREAD_RETURN(nullptr);
} }
/**
* EQStreamFactory constructor with stream type and port.
*
* @param type - Type of streams to create (login/world)
* @param port - Port number to listen on
*/
EQStreamFactory::EQStreamFactory(EQStreamType type, int port) EQStreamFactory::EQStreamFactory(EQStreamType type, int port)
{ {
StreamType = type; StreamType = type;
Port = port; Port = port;
listen_ip_address = 0; listen_ip_address = nullptr;
sock = -1;
ReaderRunning = false;
WriterRunning = false;
CombinePacketRunning = false;
DecayTimer = nullptr;
} }
/**
* Close the stream factory and clean up all resources.
* Stops all threads, closes socket, and removes all streams.
*/
void EQStreamFactory::Close() void EQStreamFactory::Close()
{ {
CheckTimeout(true); CheckTimeout(true);
Stop(); Stop();
if (sock != -1) { if (sock != -1) {
#ifdef WIN32
closesocket(sock);
#else
close(sock); close(sock);
#endif
sock = -1; sock = -1;
} }
} }
/**
* Open the UDP socket and start worker threads.
* Creates reader, writer, and packet combiner threads.
*
* @return true if successful, false on error
*/
bool EQStreamFactory::Open() bool EQStreamFactory::Open()
{ {
struct sockaddr_in address; struct sockaddr_in address;
#ifndef WIN32
pthread_t t1, t2, t3; pthread_t t1, t2, t3;
#endif
/* Setup internet address information. // Setup socket address structure
This is used with the bind() call */ memset(reinterpret_cast<char*>(&address), 0, sizeof(address));
memset((char *) &address, 0, sizeof(address));
address.sin_family = AF_INET; address.sin_family = AF_INET;
address.sin_port = htons(Port); address.sin_port = htons(Port);
// Set bind address based on configuration
#if defined(LOGIN) || defined(MINILOGIN) #if defined(LOGIN) || defined(MINILOGIN)
if(listen_ip_address) if (listen_ip_address) {
address.sin_addr.s_addr = inet_addr(listen_ip_address); address.sin_addr.s_addr = inet_addr(listen_ip_address);
else } else {
address.sin_addr.s_addr = htonl(INADDR_ANY); address.sin_addr.s_addr = htonl(INADDR_ANY);
}
#else #else
address.sin_addr.s_addr = htonl(INADDR_ANY); address.sin_addr.s_addr = htonl(INADDR_ANY);
#endif #endif
/* Setting up UDP port for new clients */
// Create UDP socket
sock = socket(AF_INET, SOCK_DGRAM, 0); sock = socket(AF_INET, SOCK_DGRAM, 0);
if (sock < 0) { if (sock < 0) {
return false; return false;
} }
if (::bind(sock, (struct sockaddr *) &address, sizeof(address)) < 0) { // Bind socket to address
//close(sock); if (::bind(sock, reinterpret_cast<struct sockaddr*>(&address), sizeof(address)) < 0) {
close(sock);
sock = -1; sock = -1;
return false; return false;
} }
#ifdef WIN32
unsigned long nonblock = 1; // Set socket to non-blocking mode
ioctlsocket(sock, FIONBIO, &nonblock);
#else
fcntl(sock, F_SETFL, O_NONBLOCK); fcntl(sock, F_SETFL, O_NONBLOCK);
#endif
//moved these because on windows the output was delayed and causing the console window to look bad // Log thread startup
#ifdef LOGIN #ifdef LOGIN
LogWrite(LOGIN__DEBUG, 0, "Login", "Starting factory Reader"); LogWrite(LOGIN__DEBUG, 0, "Login", "Starting factory Reader");
LogWrite(LOGIN__DEBUG, 0, "Login", "Starting factory Writer"); LogWrite(LOGIN__DEBUG, 0, "Login", "Starting factory Writer");
@ -140,28 +165,31 @@ struct sockaddr_in address;
LogWrite(WORLD__DEBUG, 0, "World", "Starting factory Reader"); LogWrite(WORLD__DEBUG, 0, "World", "Starting factory Reader");
LogWrite(WORLD__DEBUG, 0, "World", "Starting factory Writer"); LogWrite(WORLD__DEBUG, 0, "World", "Starting factory Writer");
#endif #endif
#ifdef WIN32
_beginthread(EQStreamFactoryReaderLoop,0, this); // Create and detach worker threads
_beginthread(EQStreamFactoryWriterLoop,0, this); pthread_create(&t1, nullptr, EQStreamFactoryReaderLoop, this);
_beginthread(EQStreamFactoryCombinePacketLoop,0, this); pthread_create(&t2, nullptr, EQStreamFactoryWriterLoop, this);
#else pthread_create(&t3, nullptr, EQStreamFactoryCombinePacketLoop, this);
pthread_create(&t1,NULL,EQStreamFactoryReaderLoop,this);
pthread_create(&t2,NULL,EQStreamFactoryWriterLoop,this);
pthread_create(&t3,NULL,EQStreamFactoryCombinePacketLoop,this);
pthread_detach(t1); pthread_detach(t1);
pthread_detach(t2); pthread_detach(t2);
pthread_detach(t3); pthread_detach(t3);
#endif
return true; return true;
} }
/**
* Get the next new stream from the queue.
* Thread-safe method to retrieve newly created streams.
*
* @return Next EQStream from queue, or nullptr if none available
*/
EQStream* EQStreamFactory::Pop() EQStream* EQStreamFactory::Pop()
{ {
if (!NewStreams.size()) if (!NewStreams.size()) {
return NULL; return nullptr;
}
EQStream *s=NULL; EQStream* s = nullptr;
//cout << "Pop():Locking MNewStreams" << endl;
MNewStreams.lock(); MNewStreams.lock();
if (NewStreams.size()) { if (NewStreams.size()) {
s = NewStreams.front(); s = NewStreams.front();
@ -169,20 +197,27 @@ EQStream *s=NULL;
s->PutInUse(); s->PutInUse();
} }
MNewStreams.unlock(); MNewStreams.unlock();
//cout << "Pop(): Unlocking MNewStreams" << endl;
return s; return s;
} }
/**
* Add a new stream to the queue for processing.
* Thread-safe method to queue newly created streams.
*
* @param s - EQStream to add to queue
*/
void EQStreamFactory::Push(EQStream* s) void EQStreamFactory::Push(EQStream* s)
{ {
//cout << "Push():Locking MNewStreams" << endl;
MNewStreams.lock(); MNewStreams.lock();
NewStreams.push(s); NewStreams.push(s);
MNewStreams.unlock(); MNewStreams.unlock();
//cout << "Push(): Unlocking MNewStreams" << endl;
} }
/**
* Main packet reading loop - runs in separate thread.
* Receives UDP packets and routes them to appropriate streams.
*/
void EQStreamFactory::ReaderLoop() void EQStreamFactory::ReaderLoop()
{ {
fd_set readset; fd_set readset;
@ -191,43 +226,66 @@ int num;
int length; int length;
unsigned char buffer[2048]; unsigned char buffer[2048];
sockaddr_in from; sockaddr_in from;
int socklen=sizeof(sockaddr_in); socklen_t socklen = sizeof(sockaddr_in);
timeval sleep_time; timeval sleep_time;
ReaderRunning = true; ReaderRunning = true;
while (sock != -1) { while (sock != -1) {
MReaderRunning.lock(); MReaderRunning.lock();
if (!ReaderRunning) if (!ReaderRunning) {
MReaderRunning.unlock();
break; break;
}
MReaderRunning.unlock(); MReaderRunning.unlock();
// Setup select() for socket monitoring
FD_ZERO(&readset); FD_ZERO(&readset);
FD_SET(sock, &readset); FD_SET(sock, &readset);
sleep_time.tv_sec = 30; sleep_time.tv_sec = 30;
sleep_time.tv_usec = 0; sleep_time.tv_usec = 0;
if ((num=select(sock+1,&readset,NULL,NULL,&sleep_time))<0) {
// What do we wanna do?
} else if (num==0)
continue;
// Wait for incoming data or timeout
num = select(sock + 1, &readset, nullptr, nullptr, &sleep_time);
if (num < 0) {
// Select error - could log this
continue;
} else if (num == 0) {
// Timeout - continue loop
continue;
}
// Check if our socket has data
if (FD_ISSET(sock, &readset)) { if (FD_ISSET(sock, &readset)) {
#ifdef WIN32 length = recvfrom(sock, buffer, 2048, 0,
if ((length=recvfrom(sock,(char*)buffer,sizeof(buffer),0,(struct sockaddr*)&from,(int *)&socklen))<2) reinterpret_cast<struct sockaddr*>(&from),
#else &socklen);
if ((length=recvfrom(sock,buffer,2048,0,(struct sockaddr *)&from,(socklen_t *)&socklen))<2)
#endif if (length < 2) {
{ // Packet too small - ignore
// What do we wanna do? continue;
} else { }
// Create address:port string for stream identification
char temp[25]; char temp[25];
sprintf(temp,"%u.%d",ntohl(from.sin_addr.s_addr),ntohs(from.sin_port)); snprintf(temp, sizeof(temp), "%u.%d",
ntohl(from.sin_addr.s_addr), ntohs(from.sin_port));
MStreams.lock(); MStreams.lock();
if ((stream_itr=Streams.find(temp))==Streams.end() || buffer[1]==OP_SessionRequest) { stream_itr = Streams.find(temp);
// Handle new connections or session requests
if (stream_itr == Streams.end() || buffer[1] == OP_SessionRequest) {
MStreams.unlock(); MStreams.unlock();
if (buffer[1] == OP_SessionRequest) { if (buffer[1] == OP_SessionRequest) {
if(stream_itr != Streams.end() && stream_itr->second) // Close existing stream if present
if (stream_itr != Streams.end() && stream_itr->second) {
stream_itr->second->SetState(CLOSED); stream_itr->second->SetState(CLOSED);
EQStream *s=new EQStream(from); }
// Create new stream
auto s = new EQStream(from);
s->SetFactory(this); s->SetFactory(this);
s->SetStreamType(StreamType); s->SetStreamType(StreamType);
Streams[temp] = s; Streams[temp] = s;
@ -237,12 +295,15 @@ timeval sleep_time;
s->SetLastPacketTime(Timer::GetCurrentTime2()); s->SetLastPacketTime(Timer::GetCurrentTime2());
} }
} else { } else {
// Route packet to existing stream
EQStream* curstream = stream_itr->second; EQStream* curstream = stream_itr->second;
//dont bother processing incoming packets for closed connections
if(curstream->CheckClosed()) // Skip closed connections
curstream = NULL; if (curstream->CheckClosed()) {
else curstream = nullptr;
} else {
curstream->PutInUse(); curstream->PutInUse();
}
MStreams.unlock(); MStreams.unlock();
if (curstream) { if (curstream) {
@ -254,11 +315,15 @@ timeval sleep_time;
} }
} }
} }
}
/**
* Check for timed out streams and clean up closed connections.
*
* @param remove_all - If true, remove all streams regardless of state
*/
void EQStreamFactory::CheckTimeout(bool remove_all) void EQStreamFactory::CheckTimeout(bool remove_all)
{ {
//lock streams the entire time were checking timeouts, it should be fast. // Lock streams for the entire timeout check - should be fast
MStreams.lock(); MStreams.lock();
unsigned long now = Timer::GetCurrentTime2(); unsigned long now = Timer::GetCurrentTime2();
@ -268,10 +333,12 @@ void EQStreamFactory::CheckTimeout(bool remove_all)
EQStream* s = stream_itr->second; EQStream* s = stream_itr->second;
EQStreamState state = s->GetState(); EQStreamState state = s->GetState();
// Transition CLOSING streams to CLOSED when no outgoing data
if (state == CLOSING && !s->HasOutgoingData()) { if (state == CLOSING && !s->HasOutgoingData()) {
stream_itr->second->SetState(CLOSED); stream_itr->second->SetState(CLOSED);
state = CLOSED; state = CLOSED;
} else if (s->CheckTimeout(now, STREAM_TIMEOUT)) { } else if (s->CheckTimeout(now, STREAM_TIMEOUT)) {
// Handle timeout based on current state
const char* stateString; const char* stateString;
switch (state) { switch (state) {
case ESTABLISHED: case ESTABLISHED:
@ -290,36 +357,39 @@ void EQStreamFactory::CheckTimeout(bool remove_all)
stateString = "Unknown"; stateString = "Unknown";
break; break;
} }
LogWrite(WORLD__DEBUG, 0, "World", "Timeout up!, state=%s (%u)", stateString, state);
LogWrite(WORLD__DEBUG, 0, "World", "Timeout up!, state=%s (%u)",
stateString, state);
if (state == ESTABLISHED) { if (state == ESTABLISHED) {
s->Close(); s->Close();
} } else if (state == WAIT_CLOSE) {
else if (state == WAIT_CLOSE) {
s->SetState(CLOSING); s->SetState(CLOSING);
state = CLOSING; state = CLOSING;
} } else if (state == CLOSING) {
else if (state == CLOSING) { // If we timeout in closing state, force close
//if we time out in the closing state, just give up
s->SetState(CLOSED); s->SetState(CLOSED);
state = CLOSED; state = CLOSED;
} }
} }
//not part of the else so we check it right away on state change
// Remove closed streams (check immediately after state changes)
if (remove_all || state == CLOSED) { if (remove_all || state == CLOSED) {
if (!remove_all && s->getTimeoutDelays() < 2) { if (!remove_all && s->getTimeoutDelays() < 2) {
s->addTimeoutDelay(); s->addTimeoutDelay();
//give it a little time for everybody to finish with it // Give other threads time to finish with this stream
++stream_itr;
} else { } else {
//everybody is done, we can delete it now // Safe to delete now
#ifdef LOGIN #ifdef LOGIN
LogWrite(LOGIN__DEBUG, 0, "Login", "Removing connection..."); LogWrite(LOGIN__DEBUG, 0, "Login", "Removing connection...");
#else #else
LogWrite(WORLD__DEBUG, 0, "World", "Removing connection..."); LogWrite(WORLD__DEBUG, 0, "World", "Removing connection...");
#endif #endif
map<string,EQStream *>::iterator temp=stream_itr; auto temp = stream_itr;
stream_itr++; ++stream_itr;
//let whoever has the stream outside delete it
// Let client list handle cleanup if in world server
#ifdef WORLD #ifdef WORLD
client_list.RemoveConnection(temp->second); client_list.RemoveConnection(temp->second);
#endif #endif
@ -328,47 +398,70 @@ void EQStreamFactory::CheckTimeout(bool remove_all)
delete stream; delete stream;
continue; continue;
} }
} else {
++stream_itr;
}
} }
stream_itr++;
}
MStreams.unlock(); MStreams.unlock();
} }
void EQStreamFactory::CombinePacketLoop(){ /**
* Packet combining optimization loop - runs in separate thread.
* Combines multiple small packets for efficient transmission.
*/
void EQStreamFactory::CombinePacketLoop()
{
deque<EQStream*> combine_que; deque<EQStream*> combine_que;
CombinePacketRunning = true; CombinePacketRunning = true;
bool packets_waiting = false; bool packets_waiting = false;
while (sock != -1) { while (sock != -1) {
if (!CombinePacketRunning) if (!CombinePacketRunning) {
break; break;
}
MStreams.lock(); MStreams.lock();
map<string,EQStream *>::iterator stream_itr;
for(stream_itr=Streams.begin();stream_itr!=Streams.end();stream_itr++) { // Check all streams for combine timer expiration
if(!stream_itr->second){ for (auto& stream_pair : Streams) {
if (!stream_pair.second) {
continue; continue;
} }
if(stream_itr->second->combine_timer && stream_itr->second->combine_timer->Check())
combine_que.push_back(stream_itr->second); if (stream_pair.second->combine_timer &&
stream_pair.second->combine_timer->Check()) {
combine_que.push_back(stream_pair.second);
} }
EQStream* stream = 0; }
// Process streams that need packet combining
packets_waiting = false; packets_waiting = false;
while(combine_que.size()){ while (!combine_que.empty()) {
stream = combine_que.front(); EQStream* stream = combine_que.front();
if (stream->CheckActive()) { if (stream->CheckActive()) {
if(!stream->CheckCombineQueue()) if (!stream->CheckCombineQueue()) {
packets_waiting = true; packets_waiting = true;
} }
}
combine_que.pop_front(); combine_que.pop_front();
} }
MStreams.unlock(); MStreams.unlock();
if(!packets_waiting)
Sleep(25);
Sleep(1); // Sleep longer if no packets are waiting
if (!packets_waiting) {
usleep(25000); // 25ms
}
usleep(1000); // 1ms
} }
} }
/**
* Main packet writing loop - runs in separate thread.
* Handles outgoing packet transmission and resends.
*/
void EQStreamFactory::WriterLoop() void EQStreamFactory::WriterLoop()
{ {
map<string, EQStream*>::iterator stream_itr; map<string, EQStream*>::iterator stream_itr;
@ -382,61 +475,70 @@ Timer DecayTimer(20);
WriterRunning = true; WriterRunning = true;
DecayTimer.Enable(); DecayTimer.Enable();
while (sock != -1) { while (sock != -1) {
Timer::SetCurrentTime(); Timer::SetCurrentTime();
//if (!havework) {
//WriterWork.Wait();
//}
MWriterRunning.lock(); MWriterRunning.lock();
if (!WriterRunning) if (!WriterRunning) {
MWriterRunning.unlock();
break; break;
}
MWriterRunning.unlock(); MWriterRunning.unlock();
wants_write.clear(); wants_write.clear();
resend_que.clear();
decay = DecayTimer.Check(); decay = DecayTimer.Check();
//copy streams into a seperate list so we dont have to keep // Copy streams into separate list to minimize lock time
//MStreams locked while we are writting
MStreams.lock(); MStreams.lock();
for(stream_itr=Streams.begin();stream_itr!=Streams.end();stream_itr++) { for (stream_itr = Streams.begin(); stream_itr != Streams.end(); ++stream_itr) {
// If it's time to decay the bytes sent, then let's do it before we try to write
if (!stream_itr->second) { if (!stream_itr->second) {
Streams.erase(stream_itr); // This shouldn't happen, but handle gracefully
break; continue;
} }
if (decay)
stream_itr->second->Decay();
// Apply bandwidth decay if it's time
if (decay) {
stream_itr->second->Decay();
}
// Queue streams with outgoing data
if (stream_itr->second->HasOutgoingData()) { if (stream_itr->second->HasOutgoingData()) {
stream_itr->second->PutInUse(); stream_itr->second->PutInUse();
wants_write.push_back(stream_itr->second); wants_write.push_back(stream_itr->second);
} }
if(stream_itr->second->resend_que_timer->Check())
// Queue streams that need resend processing
if (stream_itr->second->resend_que_timer->Check()) {
resend_que.push_back(stream_itr->second); resend_que.push_back(stream_itr->second);
} }
}
MStreams.unlock(); MStreams.unlock();
//do the actual writes // Perform actual packet writes
cur = wants_write.begin(); for (cur = wants_write.begin(), end = wants_write.end();
end = wants_write.end(); cur != end; ++cur) {
for(; cur != end; cur++) {
(*cur)->Write(sock); (*cur)->Write(sock);
(*cur)->ReleaseFromUse(); (*cur)->ReleaseFromUse();
} }
while(resend_que.size()){
// Handle packet resends
while (!resend_que.empty()) {
resend_que.front()->CheckResend(sock); resend_que.front()->CheckResend(sock);
resend_que.pop_front(); resend_que.pop_front();
} }
Sleep(10);
usleep(10000); // 10ms sleep
// Check if we have any streams - wait if not
MStreams.lock(); MStreams.lock();
stream_count = Streams.size(); stream_count = Streams.size();
MStreams.unlock(); MStreams.unlock();
if (!stream_count) { if (!stream_count) {
//cout << "No streams, waiting on condition" << endl;
WriterWork.Wait(); WriterWork.Wait();
//cout << "Awake from condition, must have a stream now" << endl;
} }
} }
} }

151
source/common/EQStreamFactory.h
View File

@ -1,86 +1,125 @@
/* // 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 _EQSTREAMFACTORY_H #ifndef _EQSTREAMFACTORY_H
#define _EQSTREAMFACTORY_H #define _EQSTREAMFACTORY_H
#include <queue> #include <queue>
#include <map> #include <map>
#include <string>
#include <memory>
#include "../common/EQStream.h" #include "../common/EQStream.h"
#include "../common/Condition.h" #include "../common/Condition.h"
#include "../common/opcodemgr.h" #include "../common/opcodemgr.h"
#include "../common/timer.h" #include "../common/timer.h"
#define STREAM_TIMEOUT 45000 //in ms #define STREAM_TIMEOUT 45000 // Stream timeout in milliseconds
/**
* Factory class for creating and managing EverQuest network streams.
* Handles UDP socket communication, stream lifecycle, and packet processing
* for both login and world server connections.
*/
class EQStreamFactory { class EQStreamFactory {
private: private:
int sock; // Network socket and configuration
int Port; int sock; // UDP socket file descriptor
int Port; // Port number to listen on
bool ReaderRunning; // Thread management flags and mutexes
Mutex MReaderRunning; bool ReaderRunning; // Reader thread active flag
bool WriterRunning; Mutex MReaderRunning; // Mutex for reader thread flag
Mutex MWriterRunning; bool WriterRunning; // Writer thread active flag
bool CombinePacketRunning; Mutex MWriterRunning; // Mutex for writer thread flag
Mutex MCombinePacketRunning; bool CombinePacketRunning; // Packet combiner thread active flag
Mutex MCombinePacketRunning; // Mutex for combiner thread flag
Condition WriterWork; // Thread synchronization
Condition WriterWork; // Condition variable for writer thread
EQStreamType StreamType; // Stream management
EQStreamType StreamType; // Type of streams this factory creates
std::queue<EQStream*> NewStreams; // Queue of new streams waiting for processing
Mutex MNewStreams; // Mutex for new streams queue
std::map<std::string, EQStream*> Streams; // Active streams mapped by address:port
Mutex MStreams; // Mutex for streams map
queue<EQStream *> NewStreams; // Cleanup timer
Mutex MNewStreams; Timer* DecayTimer; // Timer for periodic cleanup operations
map<string,EQStream *> Streams;
Mutex MStreams;
Timer *DecayTimer;
public: public:
char* listen_ip_address; // Network configuration
char* listen_ip_address; // IP address to bind to (nullptr = any)
// Stream lifecycle management
void CheckTimeout(bool remove_all = false); void CheckTimeout(bool remove_all = false);
EQStreamFactory(EQStreamType type) { ReaderRunning=false; WriterRunning=false; StreamType=type; }
// Constructors and destructor
EQStreamFactory(EQStreamType type) {
ReaderRunning = false;
WriterRunning = false;
CombinePacketRunning = false;
StreamType = type;
sock = -1;
Port = 0;
listen_ip_address = nullptr;
DecayTimer = nullptr;
}
EQStreamFactory(EQStreamType type, int port); EQStreamFactory(EQStreamType type, int port);
~EQStreamFactory() { ~EQStreamFactory() {
safe_delete_array(listen_ip_address); safe_delete_array(listen_ip_address);
} }
EQStream *Pop(); // Stream queue management
void Push(EQStream *s); EQStream* Pop(); // Get next new stream from queue
void Push(EQStream* s); // Add new stream to queue
bool loadPublicKey(); // Network operations
bool Open(); bool loadPublicKey(); // Load encryption keys (if needed)
bool Open(unsigned long port) { Port=port; return Open(); } bool Open(); // Open socket and start threads
void Close(); bool Open(unsigned long port) {
void ReaderLoop(); Port = port;
void WriterLoop(); return Open();
void CombinePacketLoop(); }
void Stop() { StopReader(); StopWriter(); StopCombinePacket(); } void Close(); // Close socket and stop threads
void StopReader() { MReaderRunning.lock(); ReaderRunning=false; MReaderRunning.unlock(); }
void StopWriter() { MWriterRunning.lock(); WriterRunning=false; MWriterRunning.unlock(); WriterWork.Signal(); }
void StopCombinePacket() { MCombinePacketRunning.lock(); CombinePacketRunning=false; MCombinePacketRunning.unlock(); }
void SignalWriter() { WriterWork.Signal(); }
// Main thread loops
void ReaderLoop(); // Main packet reading loop
void WriterLoop(); // Main packet writing loop
void CombinePacketLoop(); // Packet combining optimization loop
// Thread control
void Stop() {
StopReader();
StopWriter();
StopCombinePacket();
}
void StopReader() {
MReaderRunning.lock();
ReaderRunning = false;
MReaderRunning.unlock();
}
void StopWriter() {
MWriterRunning.lock();
WriterRunning = false;
MWriterRunning.unlock();
WriterWork.Signal();
}
void StopCombinePacket() {
MCombinePacketRunning.lock();
CombinePacketRunning = false;
MCombinePacketRunning.unlock();
}
void SignalWriter() {
WriterWork.Signal();
}
}; };
#endif #endif