Concurrent starting concurrent eqstream

CMakeLists.txt

@@ -349,6 +349,7 @@ INCLUDE_DIRECTORIES(SYSTEM "${CMAKE_CURRENT_SOURCE_DIR}/submodules/recastnavigat
 INCLUDE_DIRECTORIES(SYSTEM "${CMAKE_CURRENT_SOURCE_DIR}/submodules/recastnavigation/DetourCrowd/Include")
 INCLUDE_DIRECTORIES(SYSTEM "${CMAKE_CURRENT_SOURCE_DIR}/submodules/recastnavigation/DetourTileCache/Include")
 INCLUDE_DIRECTORIES(SYSTEM "${CMAKE_CURRENT_SOURCE_DIR}/submodules/recastnavigation/Recast/Include")
+INCLUDE_DIRECTORIES(SYSTEM "${CMAKE_CURRENT_SOURCE_DIR}/submodules/concurrentqueue")
 
 IF(EQEMU_BUILD_SERVER OR EQEMU_BUILD_LOGIN OR EQEMU_BUILD_TESTS OR EQEMU_BUILD_HC)
 	ADD_SUBDIRECTORY(common)

common/CMakeLists.txt

@@ -79,6 +79,7 @@ SET(common_sources
 	net/crc32.cpp
 	net/daybreak_connection.cpp
 	net/eqstream.cpp
+	net/eqstream_concurrent.cpp
 	net/packet.cpp
 	net/servertalk_client_connection.cpp
 	net/servertalk_legacy_client_connection.cpp
@@ -222,6 +223,7 @@ SET(common_headers
 	net/dns.h
 	net/endian.h
 	net/eqstream.h
+	net/eqstream_concurrent.h
 	net/packet.h
 	net/servertalk_client_connection.h
 	net/servertalk_legacy_client_connection.h
@@ -294,6 +296,8 @@ SOURCE_GROUP(Net FILES
 	net/eqmq.h
 	net/eqstream.cpp
 	net/eqstream.h
+	net/eqstream_concurrent.cpp
+	net/eqstream_concurrent.h
 	net/packet.cpp
 	net/packet.h
 	net/servertalk_client_connection.cpp

common/net/eqstream_concurrent.cpp

@@ -0,0 +1,36 @@
+#include "eqstream_concurrent.h"
+
+struct EQ::Net::ConcurrentEQStreamManager::Impl
+{
+
+};
+
+EQ::Net::ConcurrentEQStreamManager::ConcurrentEQStreamManager(const EQStreamManagerInterfaceOptions &options)
+	: EQStreamManagerInterface(options)
+{
+	_impl.reset(new Impl());
+}
+
+EQ::Net::ConcurrentEQStreamManager::~ConcurrentEQStreamManager()
+{
+}
+
+void EQ::Net::ConcurrentEQStreamManager::OnNewConnection(std::function<void(std::shared_ptr<ConcurrentEQStream>)> func)
+{
+}
+
+void EQ::Net::ConcurrentEQStreamManager::OnConnectionStateChange(std::function<void(std::shared_ptr<ConcurrentEQStream>, DbProtocolStatus, DbProtocolStatus)> func)
+{
+}
+
+void EQ::Net::ConcurrentEQStreamManager::DaybreakNewConnection(std::shared_ptr<DaybreakConnection> connection)
+{
+}
+
+void EQ::Net::ConcurrentEQStreamManager::DaybreakConnectionStateChange(std::shared_ptr<DaybreakConnection> connection, DbProtocolStatus from, DbProtocolStatus to)
+{
+}
+
+void EQ::Net::ConcurrentEQStreamManager::DaybreakPacketRecv(std::shared_ptr<DaybreakConnection> connection, const Packet &p)
+{
+}

common/net/eqstream_concurrent.h

@@ -0,0 +1,60 @@
+#pragma once
+
+#include "../eq_stream_intf.h"
+#include <memory>
+
+namespace EQ
+{
+	namespace Net
+	{
+		class ConcurrentEQStream;
+		class ConcurrentEQStreamManager : public EQStreamManagerInterface
+		{
+		public:
+			ConcurrentEQStreamManager(const EQStreamManagerInterfaceOptions &options);
+			~ConcurrentEQStreamManager();
+
+			void OnNewConnection(std::function<void(std::shared_ptr<ConcurrentEQStream>)> func);
+			void OnConnectionStateChange(std::function<void(std::shared_ptr<ConcurrentEQStream>, DbProtocolStatus, DbProtocolStatus)> func);
+		private:
+			struct Impl;
+			std::unique_ptr<Impl> _impl;
+
+			void DaybreakNewConnection(std::shared_ptr<DaybreakConnection> connection);
+			void DaybreakConnectionStateChange(std::shared_ptr<DaybreakConnection> connection, DbProtocolStatus from, DbProtocolStatus to);
+			void DaybreakPacketRecv(std::shared_ptr<DaybreakConnection> connection, const Packet &p);
+			friend class EQStream;
+		};
+
+		class ConcurrentEQStream : public EQStreamInterface
+		{
+		public:
+			ConcurrentEQStream(EQStreamManagerInterface *parent, uint64_t id);
+			~ConcurrentEQStream();
+
+			virtual void QueuePacket(const EQApplicationPacket *p, bool ack_req = true);
+			virtual void FastQueuePacket(EQApplicationPacket **p, bool ack_req = true);
+			virtual EQApplicationPacket *PopPacket();
+			virtual void Close();
+			virtual void ReleaseFromUse();
+			virtual void RemoveData();
+			virtual std::string GetRemoteAddr() const;
+			virtual uint32 GetRemoteIP() const;
+			virtual uint16 GetRemotePort() const;
+			virtual bool CheckState(EQStreamState state);
+			virtual std::string Describe() const;
+			virtual void SetActive(bool val);
+			virtual MatchState CheckSignature(const Signature *sig);
+			virtual EQStreamState GetState();
+			virtual void SetOpcodeManager(OpcodeManager **opm);
+			virtual Stats GetStats() const;
+			virtual void ResetStats();
+			virtual EQStreamManagerInterface* GetManager() const;
+		private:
+			struct Impl;
+
+			std::unique_ptr<Impl> _impl;
+			friend class ConcurrentEQStreamManager;
+		};
+	}
+}