Compression needs work but finished the eqstream abstraction layer otherwise.

common/eq_stream_ident.cpp

@@ -144,7 +144,7 @@ void EQStreamIdentifier::Process() {
 	}	//end foreach stream
 }
 
-void EQStreamIdentifier::AddStream(std::shared_ptr<EQStreamInterface> &eqs) {
+void EQStreamIdentifier::AddStream(std::shared_ptr<EQStreamInterface> eqs) {
 	m_streams.push_back(Record(eqs));
 	eqs = nullptr;
 }

common/eq_stream_ident.h

@@ -22,7 +22,7 @@ public:
 
 	//main processing interface
 	void Process();
-	void AddStream(std::shared_ptr<EQStreamInterface> &eqs);
+	void AddStream(std::shared_ptr<EQStreamInterface> eqs);
 	EQStreamInterface *PopIdentified();
 
 protected:

common/net/daybreak_connection.cpp

@@ -900,7 +900,8 @@ void EQ::Net::DaybreakConnection::Decompress(Packet &p, size_t offset, size_t le
 		new_length = (uint32_t)length - 1;
 	}
 	else {
-		return;
+		memcpy(new_buffer, buffer, length);
+		new_length = length;
 	}
 
 	p.Resize(offset);

common/net/eqstream.cpp

@@ -120,7 +120,7 @@ EQApplicationPacket *EQ::Net::EQStream::PopPacket() {
 		}
 
 		EmuOpcode emu_op = (*m_opcode_manager)->EQToEmu(opcode);
-		EQApplicationPacket *ret = new EQApplicationPacket(emu_op, (unsigned char*)p->Data(), p->Length() - m_owner->m_options.opcode_size);
+		EQApplicationPacket *ret = new EQApplicationPacket(emu_op, (unsigned char*)p->Data() + m_owner->m_options.opcode_size, p->Length() - m_owner->m_options.opcode_size);
 		m_packet_queue.pop_front();
 		return ret;
 	}

common/net/eqstream.h

@@ -75,8 +75,7 @@ namespace EQ
 			virtual void ReleaseFromUse() { };
 			virtual void RemoveData() { };
 			virtual uint32 GetRemoteIP() const;
-			//the code is dumb and assumes this is in network order...
+			virtual uint16 GetRemotePort() const { return m_connection->RemotePort(); }
-			virtual uint16 GetRemotePort() const { return EQ::Net::HostToNetwork(m_connection->RemotePort()); }
 			virtual bool CheckState(EQStreamState state);
 			virtual std::string Describe() const { return "Direct EQStream"; }
 			virtual void SetActive(bool val) { }

world/CMakeLists.txt

@@ -68,16 +68,11 @@ INSTALL(TARGETS world RUNTIME DESTINATION ${CMAKE_INSTALL_PREFIX}/bin)
 
 ADD_DEFINITIONS(-DWORLD)
 
-TARGET_LINK_LIBRARIES(world common ${PERL_LIBRARY} debug ${MySQL_LIBRARY_DEBUG} optimized ${MySQL_LIBRARY_RELEASE} ${ZLIB_LIBRARY})
+TARGET_LINK_LIBRARIES(world common ${PERL_LIBRARY} debug ${MySQL_LIBRARY_DEBUG} optimized ${MySQL_LIBRARY_RELEASE} ${ZLIB_LIBRARY} libuv fmt)
 
-IF(MSVC)
+IF(WIN32)
-	SET_TARGET_PROPERTIES(world PROPERTIES LINK_FLAGS_RELEASE "/OPT:REF /OPT:ICF")
+	TARGET_LINK_LIBRARIES(world "ws2_32" "psapi" "iphlpapi" "userenv")
-	TARGET_LINK_LIBRARIES(world "Ws2_32.lib")
+ENDIF(WIN32)
-ENDIF(MSVC)
-
-IF(MINGW)
-	TARGET_LINK_LIBRARIES(world "WS2_32")
-ENDIF(MINGW)
 
 IF(UNIX)
 	TARGET_LINK_LIBRARIES(world "${CMAKE_DL_LIBS}")

world/net.cpp

@@ -35,6 +35,9 @@
 #include "../common/eqtime.h"
 #include "../common/timeoutmgr.h"
 
+#include "../common/event/event_loop.h"
+#include "../common/net/eqstream.h"
+
 #include "../common/opcodemgr.h"
 #include "../common/guilds.h"
 #include "../common/eq_stream_ident.h"
@@ -88,7 +91,6 @@
 #include "queryserv.h"
 
 TimeoutManager timeout_manager;
-EQStreamFactory eqsf(WorldStream,9000);
 EmuTCPServer tcps;
 ClientList client_list;
 GroupLFPList LFPGroupList;
@@ -395,12 +397,12 @@ int main(int argc, char** argv) {
 		Log.Out(Logs::General, Logs::World_Server,"        %s",errbuf);
 		return 1;
 	}
-	if (eqsf.Open()) {
+
-		Log.Out(Logs::General, Logs::World_Server,"Client (UDP) listener started.");
+	EQ::Net::EQStreamManagerOptions opts(true, false);
-	} else {
+	opts.daybreak_options.port = 9000;
-		Log.Out(Logs::General, Logs::World_Server,"Failed to start client (UDP) listener (port 9000)");
+	opts.opcode_size = 2;
-		return 1;
+
-	}
+	EQ::Net::EQStreamManager eqsm(opts);
 
 	//register all the patches we have avaliable with the stream identifier.
 	EQStreamIdentifier stream_identifier;
@@ -416,20 +418,13 @@ int main(int argc, char** argv) {
 	EmuTCPConnection* tcpc;
 	EQStreamInterface *eqsi;
 
+	eqsm.OnNewConnection([&stream_identifier](std::shared_ptr<EQ::Net::EQStream> stream) {
+		stream_identifier.AddStream(stream);
+		Log.OutF(Logs::Detail, Logs::World_Server, "New connection from IP {0}:{1}", stream->RemoteEndpoint(), ntohs(stream->GetRemotePort()));
+	});
+
 	while(RunLoops) {
 		Timer::SetCurrentTime();
-
-		//check the factory for any new incoming streams.
-		while ((eqs = eqsf.Pop())) {
-			//pull the stream out of the factory and give it to the stream identifier
-			//which will figure out what patch they are running, and set up the dynamic
-			//structures and opcodes for that patch.
-			struct in_addr	in;
-			in.s_addr = eqs->GetRemoteIP();
-			Log.Out(Logs::Detail, Logs::World_Server, "New connection from IP %s:%d", inet_ntoa(in),ntohs(eqs->GetRemotePort()));
-			stream_identifier.AddStream(eqs);	//takes the stream
-		}
-
 		eqs = nullptr;
 
 		//give the stream identifier a chance to do its work....
@@ -528,11 +523,9 @@ int main(int argc, char** argv) {
 			}
 			
 		}
-		if (numclients == 0) {
+		
-			Sleep(50);
+		EQ::EventLoop::Get().Process();
-			continue;
+		Sleep(1);
-		}
-		Sleep(20);
 	}
 	Log.Out(Logs::General, Logs::World_Server, "World main loop completed.");
 	Log.Out(Logs::General, Logs::World_Server, "Shutting down console connections (if any).");
@@ -541,8 +534,6 @@ int main(int argc, char** argv) {
 	zoneserver_list.KillAll();
 	Log.Out(Logs::General, Logs::World_Server, "Zone (TCP) listener stopped.");
 	tcps.Close();
-	Log.Out(Logs::General, Logs::World_Server, "Client (UDP) listener stopped.");
-	eqsf.Close();
 	Log.Out(Logs::General, Logs::World_Server, "Signaling HTTP service to stop...");
 	http_server.Stop();
 	Log.CloseFileLogs();