From eada3e1d2caafbc953ed58bb3bb45cfd7f98d16e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=A5=BF=E9=A3=8E=E9=80=8D=E9=81=A5=E6=B8=B8?= Date: Wed, 5 Aug 2026 14:01:09 +0800 Subject: [PATCH] feat: implement network module with TCP/UDP support and event handling Add a new Network module that includes TCP and UDP socket management, HTTP request handling, and session management. Introduce event emission for network events and integrate with the existing event system. Update CMake configurations to include the new module and its dependencies. --- src/engine/CMakeLists.txt | 16 +- src/modules/CMakeLists.txt | 4 +- src/modules/event/Event.cpp | 19 +- src/modules/event/Event.h | 55 ++++- src/modules/event/sdl/Event.cpp | 5 +- src/modules/network/Channel.cpp | 120 ++++++++++ src/modules/network/Channel.h | 35 +++ src/modules/network/HttpRequest.cpp | 113 ++++++++++ src/modules/network/HttpRequest.h | 43 ++++ src/modules/network/NetTypes.h | 34 +++ src/modules/network/NetWorker.cpp | 66 ++++++ src/modules/network/NetWorker.h | 41 ++++ src/modules/network/Network.cpp | 336 ++++++++++++++++++++++++++++ src/modules/network/Network.h | 68 +++++- src/modules/network/Session.cpp | 31 +++ src/modules/network/Session.h | 24 ++ src/modules/network/TcpSocket.cpp | 207 +++++++++++++++++ src/modules/network/TcpSocket.h | 63 ++++++ src/modules/network/UdpSocket.cpp | 136 +++++++++++ src/modules/network/UdpSocket.h | 47 ++++ test/CMakeLists.txt | 25 ++- test/network.cpp | 236 +++++++++++++++++++ 22 files changed, 1677 insertions(+), 47 deletions(-) create mode 100644 src/modules/network/Channel.cpp create mode 100644 src/modules/network/Channel.h create mode 100644 src/modules/network/HttpRequest.cpp create mode 100644 src/modules/network/HttpRequest.h create mode 100644 src/modules/network/NetTypes.h create mode 100644 src/modules/network/NetWorker.cpp create mode 100644 src/modules/network/NetWorker.h create mode 100644 src/modules/network/Session.cpp create mode 100644 src/modules/network/Session.h create mode 100644 src/modules/network/TcpSocket.cpp create mode 100644 src/modules/network/TcpSocket.h create mode 100644 src/modules/network/UdpSocket.cpp create mode 100644 src/modules/network/UdpSocket.h create mode 100644 test/network.cpp diff --git a/src/engine/CMakeLists.txt b/src/engine/CMakeLists.txt index 6fd4a88d..706f0fe5 100644 --- a/src/engine/CMakeLists.txt +++ b/src/engine/CMakeLists.txt @@ -7,24 +7,30 @@ configure_file(${CMAKE_CURRENT_SOURCE_DIR}/common/config.h.in common/config.h) add_executable(eve main.cpp) if(WIN32) - set(SystemLibraries ${SystemLibraries} winmm version imm32 Setupapi) + set(SystemLibraries ${SystemLibraries} winmm version imm32 Setupapi ws2_32 iphlpapi) elseif(UNIX) set(SystemLibraries ${SystemLibraries} dl X11 pthread) endif() if(WIN32 AND (${CMAKE_BUILD_TYPE} STREQUAL "Debug")) set(ThirdParty ${ThirdParty} simplesquirrel_static sqstdlib_static squirrel_static SDL2d SDL2maind - medialoader_imagemdd zlibstaticd ${SystemLibraries}) + medialoader_imagemdd zlibstaticd physfsmdd + PocoFoundationmdd PocoNetmdd + ${SystemLibraries}) else() if(WIN32) set(ThirdParty ${ThirdParty} simplesquirrel_static sqstdlib_static squirrel_static SDL2 SDL2main - medialoader_imagemd zlibstatic ${SystemLibraries}) + medialoader_imagemd zlibstatic physfsmd + PocoFoundationmd PocoNetmd + ${SystemLibraries}) else() set(ThirdParty ${ThirdParty} simplesquirrel_static sqstdlib_static squirrel_static SDL2 SDL2main - medialoader_image z ${SystemLibraries}) + medialoader_image z physfs + PocoFoundation PocoNet + ${SystemLibraries}) endif() endif() -set(EVELIBS EVWindow EVMouse EVEvent EVCmdLine EVDevTools EVScripts EVCommon EVGraphics EVImage EVFileSystem) +set(EVELIBS EVWindow EVMouse EVEvent EVCmdLine EVDevTools EVScripts EVCommon EVGraphics EVImage EVFileSystem EVNetwork) target_link_libraries(eve ${EVELIBS} zeroerr ${ThirdParty} ${EVENGINE_VULKAN_LIB}) \ No newline at end of file diff --git a/src/modules/CMakeLists.txt b/src/modules/CMakeLists.txt index 4a2a2d27..8e1f3164 100644 --- a/src/modules/CMakeLists.txt +++ b/src/modules/CMakeLists.txt @@ -6,4 +6,6 @@ create_module(EVEvent event) create_module(EVMouse mouse) create_module(EVTouch touch) create_module(EVData data) -create_module(EVImage image) \ No newline at end of file +create_module(EVImage image) +create_module(EVNetwork network) +target_compile_definitions(EVNetwork PUBLIC POCO_STATIC POCO_NO_AUTOMATIC_LIBS) \ No newline at end of file diff --git a/src/modules/event/Event.cpp b/src/modules/event/Event.cpp index e8123bf8..bb44f4ca 100644 --- a/src/modules/event/Event.cpp +++ b/src/modules/event/Event.cpp @@ -1,34 +1,31 @@ - #include "event/Event.h" #include "event/sdl/Event.h" +#include + namespace eve::event { -Message::Message(const std::string &name, const std::vector &vargs) : name(name), args(vargs) {} +Message::Message(const std::string& name, const std::vector& vargs) : name(name), args(vargs) {} Message::~Message() {} Module_IMPL(Event, new sdl::Event()); - -void Event::expose(ssq::Table &table) { +void Event::expose(ssq::Table& table) { auto cls = table.addClass(name, Event::create, false); expose(cls); } -void Event::expose(ssq::Class &cls) { +void Event::expose(ssq::Class& cls) { cls.addFunc("getName", &Event::getName); } Event::~Event() {} -void Event::push(Message *msg) { - // Lock lock(mutex); - // msg->retain(); +void Event::push(Message* msg) { queue.push(msg); } -Message *Event::poll() { - // Lock lock(mutex); +Message* Event::poll() { if (queue.empty()) return nullptr; auto msg = queue.front(); queue.pop(); @@ -36,9 +33,7 @@ Message *Event::poll() { } void Event::clear() { - // Lock lock(mutex); while (!queue.empty()) { - // std::queue::pop will remove the first (front) element. delete queue.front(); queue.pop(); } diff --git a/src/modules/event/Event.h b/src/modules/event/Event.h index 28ce3984..86db9734 100644 --- a/src/modules/event/Event.h +++ b/src/modules/event/Event.h @@ -1,41 +1,72 @@ #pragma once +#include #include #include #include -#include #include "common/Module.h" +namespace ssq { +class Table; +class Class; +} // namespace ssq + namespace eve { namespace event { +struct Variant { + enum class Type { Nil, Int, String, Ptr }; + Type type = Type::Nil; + int64_t i = 0; + std::string s; + void* p = nullptr; + + static Variant makeNil() { return {}; } + static Variant makeInt(int64_t v) { + Variant x; + x.type = Type::Int; + x.i = v; + return x; + } + static Variant makeString(std::string v) { + Variant x; + x.type = Type::String; + x.s = std::move(v); + return x; + } + static Variant makePtr(void* v) { + Variant x; + x.type = Type::Ptr; + x.p = v; + return x; + } +}; + class Message { public: - Message(const std::string &name, const std::vector &vargs = {}); + Message(const std::string& name, const std::vector& vargs = {}); ~Message(); - const std::string name; - - const std::vector args; -}; // Message + const std::string name; + const std::vector args; +}; class Event : public Module { public: Module_REG(Event); virtual ~Event(); - void push(Message *msg); - Message * poll(); + void push(Message* msg); + Message* poll(); virtual void clear(); virtual void pump() = 0; - virtual Message *wait() = 0; + virtual Message* wait() = 0; protected: - std::queue queue; - -}; // Event + std::queue queue; +}; } // namespace event } // namespace eve diff --git a/src/modules/event/sdl/Event.cpp b/src/modules/event/sdl/Event.cpp index 511ec0b5..978d5c4b 100644 --- a/src/modules/event/sdl/Event.cpp +++ b/src/modules/event/sdl/Event.cpp @@ -138,10 +138,7 @@ void Event::exceptionIfInRenderPass(const char *name) Message *Event::convert(const SDL_Event &e) { Message *msg = nullptr; - - std::vector vargs; - vargs.reserve(4); - + (void)e; return msg; } diff --git a/src/modules/network/Channel.cpp b/src/modules/network/Channel.cpp new file mode 100644 index 00000000..45570e0e --- /dev/null +++ b/src/modules/network/Channel.cpp @@ -0,0 +1,120 @@ +#include "network/Channel.h" +#include "network/TcpSocket.h" +#include "network/Network.h" +#include "network/NetTypes.h" +#include "data/ByteData.h" +#include "event/Event.h" +#include "common/Module.h" + +#include + +namespace eve::network { + +namespace { + +void writeBe32(uint32_t v, char out[4]) { + out[0] = static_cast((v >> 24) & 0xff); + out[1] = static_cast((v >> 16) & 0xff); + out[2] = static_cast((v >> 8) & 0xff); + out[3] = static_cast(v & 0xff); +} + +uint32_t readBe32(const char* p) { + return (uint32_t(uint8_t(p[0])) << 24) | (uint32_t(uint8_t(p[1])) << 16) | + (uint32_t(uint8_t(p[2])) << 8) | uint32_t(uint8_t(p[3])); +} + +} // namespace + +Channel::Channel(TcpSocket* socket) : socket_(socket) {} + +Channel::~Channel() { + if (socket_ && socket_->network()) socket_->network()->unbindChannel(socket_); +} + +bool Channel::sendMsg(eve::data::ByteData* data) { + if (!data || !socket_) return false; + size_t n = data->getSize(); + if (n > kMaxFrameSize) { + if (socket_->network()) { + NetCompletion c; + c.type = NetEvType::Err; + c.kind = NetKind::Channel; + c.handle = this; + c.reason = "limit"; + socket_->network()->post(std::move(c)); + } + return false; + } + std::vector frame(4 + n); + writeBe32(static_cast(n), frame.data()); + std::memcpy(frame.data() + 4, data->getData(), n); + eve::data::ByteData wrapped(frame.data(), frame.size()); + return socket_->send(&wrapped); +} + +bool Channel::sendMsgString(std::string s) { + if (s.empty()) return false; + eve::data::ByteData data(s.data(), s.size()); + return sendMsg(&data); +} + +void Channel::feed(const std::vector& bytes) { + buffer_.insert(buffer_.end(), bytes.begin(), bytes.end()); + auto* net = socket_ ? socket_->network() : nullptr; + while (true) { + if (!hasLen_) { + if (buffer_.size() < 4) return; + pendingLen_ = readBe32(buffer_.data()); + hasLen_ = true; + buffer_.erase(buffer_.begin(), buffer_.begin() + 4); + if (pendingLen_ > kMaxFrameSize) { + if (net) { + NetCompletion c; + c.type = NetEvType::ChClose; + c.kind = NetKind::Channel; + c.handle = this; + c.reason = "limit"; + net->post(std::move(c)); + NetCompletion e; + e.type = NetEvType::Err; + e.kind = NetKind::Channel; + e.handle = this; + e.reason = "limit"; + net->post(std::move(e)); + } + if (socket_) socket_->close(); + buffer_.clear(); + hasLen_ = false; + return; + } + } + if (buffer_.size() < pendingLen_) return; + auto payload = std::make_shared>(buffer_.begin(), buffer_.begin() + pendingLen_); + buffer_.erase(buffer_.begin(), buffer_.begin() + pendingLen_); + hasLen_ = false; + pendingLen_ = 0; + if (net) { + NetCompletion c; + c.type = NetEvType::ChMsg; + c.kind = NetKind::Channel; + c.handle = this; + c.bytes = std::move(payload); + // Emit directly via event on main thread — feed is called from pump + using eve::event::Variant; + using eve::event::Message; + auto* ev = eve::ModuleManager::getInstance("Event"); + if (ev) { + eve::data::ByteData* bd = nullptr; + if (c.bytes && !c.bytes->empty()) + bd = new eve::data::ByteData(c.bytes->data(), c.bytes->size()); + std::vector args; + args.push_back(Variant::makePtr(this)); + args.push_back(Variant::makePtr(bd)); + ev->push(new Message("chmsg", args)); + } + } + } +} + +} // namespace eve::network diff --git a/src/modules/network/Channel.h b/src/modules/network/Channel.h new file mode 100644 index 00000000..a3af398e --- /dev/null +++ b/src/modules/network/Channel.h @@ -0,0 +1,35 @@ +#pragma once + +#include +#include +#include + +namespace eve::data { +class ByteData; +} + +namespace eve::network { + +class TcpSocket; +class Network; + +class Channel { +public: + explicit Channel(TcpSocket* socket); + ~Channel(); + + bool sendMsg(eve::data::ByteData* data); + bool sendMsgString(std::string s); + TcpSocket* getSocket() const { return socket_; } + + // Called on main thread from Network::pump / emitCompletion path + void feed(const std::vector& bytes); + +private: + TcpSocket* socket_ = nullptr; + std::vector buffer_; + uint32_t pendingLen_ = 0; + bool hasLen_ = false; +}; + +} // namespace eve::network diff --git a/src/modules/network/HttpRequest.cpp b/src/modules/network/HttpRequest.cpp new file mode 100644 index 00000000..5fdfe6ec --- /dev/null +++ b/src/modules/network/HttpRequest.cpp @@ -0,0 +1,113 @@ +#include "network/HttpRequest.h" +#include "network/Network.h" +#include "network/NetWorker.h" +#include "data/ByteData.h" + +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +namespace eve::network { + +HttpRequest::HttpRequest(Network* net, std::string method, std::string url) + : net_(net), method_(std::move(method)), url_(std::move(url)) {} + +void HttpRequest::setHeader(std::string k, std::string v) { + headers_[std::move(k)] = std::move(v); +} + +void HttpRequest::setBody(eve::data::ByteData* data) { + if (!data) { + body_.reset(); + return; + } + body_ = std::make_shared>( + static_cast(data->getData()), + static_cast(data->getData()) + data->getSize()); +} + +void HttpRequest::setBodyString(std::string s) { + body_ = std::make_shared>(s.begin(), s.end()); +} + +void HttpRequest::setTimeout(int ms) { + timeoutMs_ = ms; +} + +void HttpRequest::setVerifySsl(bool verify) { + verifySsl_ = verify ? 1 : 0; +} + +bool HttpRequest::submit() { + if (!net_) return false; + auto* self = this; + auto method = method_; + auto url = url_; + auto headers = headers_; + auto body = body_; + int timeoutMs = timeoutMs_ >= 0 ? timeoutMs_ : net_->getTimeout(); + (void)verifySsl_; + + // Dedicated thread: HTTP would block NetWorker and prevent accept/poll. + std::thread([self, method, url, headers, body, timeoutMs]() { + NetCompletion c; + c.kind = NetKind::Http; + c.handle = self; + try { + Poco::URI uri(url); + if (uri.getScheme() == "https") { + c.type = NetEvType::Err; + c.reason = "tls"; + self->net_->post(std::move(c)); + return; + } + Poco::Net::HTTPClientSession session(uri.getHost(), uri.getPort()); + session.setTimeout(Poco::Timespan(timeoutMs / 1000, (timeoutMs % 1000) * 1000)); + std::string path = uri.getPathAndQuery(); + if (path.empty()) path = "/"; + Poco::Net::HTTPRequest req(method, path, Poco::Net::HTTPMessage::HTTP_1_1); + req.setHost(uri.getHost()); + for (auto& kv : headers) req.set(kv.first, kv.second); + if (body && !body->empty()) { + req.setContentLength(static_cast(body->size())); + } + std::ostream& os = session.sendRequest(req); + if (body && !body->empty()) { + os.write(body->data(), static_cast(body->size())); + } + Poco::Net::HTTPResponse resp; + std::istream& rs = session.receiveResponse(resp); + std::ostringstream oss; + Poco::StreamCopier::copyStream(rs, oss); + std::string content = oss.str(); + c.type = NetEvType::HttpResp; + c.status = static_cast(resp.getStatus()); + c.bytes = std::make_shared>(content.begin(), content.end()); + self->net_->post(std::move(c)); + } catch (const Poco::Exception& ex) { + c.type = NetEvType::Err; + std::string msg = ex.displayText(); + if (msg.find("timed") != std::string::npos || msg.find("Timeout") != std::string::npos) + c.reason = "timeout"; + else if (msg.find("host") != std::string::npos || msg.find("DNS") != std::string::npos) + c.reason = "dns"; + else + c.reason = "refused"; + self->net_->post(std::move(c)); + } catch (...) { + c.type = NetEvType::Err; + c.reason = "refused"; + self->net_->post(std::move(c)); + } + }).detach(); + return true; +} + +} // namespace eve::network diff --git a/src/modules/network/HttpRequest.h b/src/modules/network/HttpRequest.h new file mode 100644 index 00000000..f086b0df --- /dev/null +++ b/src/modules/network/HttpRequest.h @@ -0,0 +1,43 @@ +#pragma once + +#include "NetTypes.h" + +#include +#include +#include +#include +#include + +namespace eve::data { +class ByteData; +} + +namespace eve::network { + +class Network; + +class HttpRequest { +public: + HttpRequest(Network* net, std::string method, std::string url); + ~HttpRequest() = default; + + void setHeader(std::string k, std::string v); + void setBody(eve::data::ByteData* data); + void setBodyString(std::string s); + void setTimeout(int ms); + void setVerifySsl(bool verify); + bool submit(); + + Network* network() const { return net_; } + +private: + Network* net_ = nullptr; + std::string method_; + std::string url_; + std::map headers_; + std::shared_ptr> body_; + int timeoutMs_ = -1; + int verifySsl_ = -1; // -1 = use module default +}; + +} // namespace eve::network diff --git a/src/modules/network/NetTypes.h b/src/modules/network/NetTypes.h new file mode 100644 index 00000000..9b9b5936 --- /dev/null +++ b/src/modules/network/NetTypes.h @@ -0,0 +1,34 @@ +#pragma once + +#include +#include +#include +#include + +namespace eve::network { + +constexpr size_t kMaxSendBuffer = 1024 * 1024; +constexpr size_t kMaxFrameSize = 1024 * 1024; + +enum class NetKind { Tcp, Udp, Http, Channel }; + +enum class NetEvType { + Conn, + Data, + Err, + HttpResp, + ChMsg, + ChClose +}; + +struct NetCompletion { + NetEvType type = NetEvType::Err; + NetKind kind = NetKind::Tcp; + void* handle = nullptr; // TcpSocket* / UdpSocket* / HttpRequest* / Channel* + std::shared_ptr> bytes; + std::string peer; + std::string reason; + int status = 0; +}; + +} // namespace eve::network diff --git a/src/modules/network/NetWorker.cpp b/src/modules/network/NetWorker.cpp new file mode 100644 index 00000000..cb034da6 --- /dev/null +++ b/src/modules/network/NetWorker.cpp @@ -0,0 +1,66 @@ +#include "network/NetWorker.h" +#include "network/Network.h" + +#include + +namespace eve::network { + +NetWorker::NetWorker(Network* owner) : owner_(owner) {} + +NetWorker::~NetWorker() { + stop(); +} + +void NetWorker::start() { + if (running_) return; + running_ = true; + thread_ = std::thread([this] { threadMain(); }); +} + +void NetWorker::stop() { + if (!running_) return; + { + std::lock_guard lock(mu_); + running_ = false; + } + cv_.notify_all(); + if (thread_.joinable()) thread_.join(); +} + +void NetWorker::post(NetCompletion c) { + std::lock_guard lock(mu_); + completions_.push_back(std::move(c)); +} + +void NetWorker::drain(std::vector& out) { + std::lock_guard lock(mu_); + out.swap(completions_); +} + +void NetWorker::submit(std::function job) { + { + std::lock_guard lock(mu_); + jobs_.push_back(std::move(job)); + } + cv_.notify_one(); +} + +void NetWorker::threadMain() { + while (true) { + std::vector> batch; + { + std::unique_lock lock(mu_); + cv_.wait_for(lock, std::chrono::milliseconds(5), [this] { + return !running_ || !jobs_.empty(); + }); + if (!running_ && jobs_.empty()) break; + batch.swap(jobs_); + } + for (auto& job : batch) { + if (job) job(); + } + if (owner_) owner_->pollSockets(); + } +} + +} // namespace eve::network diff --git a/src/modules/network/NetWorker.h b/src/modules/network/NetWorker.h new file mode 100644 index 00000000..421278b8 --- /dev/null +++ b/src/modules/network/NetWorker.h @@ -0,0 +1,41 @@ +#pragma once + +#include "NetTypes.h" + +#include +#include +#include +#include +#include +#include + +namespace eve::network { + +class Network; + +class NetWorker { +public: + explicit NetWorker(Network* owner); + ~NetWorker(); + + void start(); + void stop(); + + void post(NetCompletion c); + void drain(std::vector& out); + + void submit(std::function job); + +private: + void threadMain(); + + Network* owner_ = nullptr; + std::mutex mu_; + std::condition_variable cv_; + std::vector completions_; + std::vector> jobs_; + std::thread thread_; + std::atomic running_{false}; +}; + +} // namespace eve::network diff --git a/src/modules/network/Network.cpp b/src/modules/network/Network.cpp index e69de29b..178cc7f7 100644 --- a/src/modules/network/Network.cpp +++ b/src/modules/network/Network.cpp @@ -0,0 +1,336 @@ +#include "network/Network.h" +#include "network/NetWorker.h" +#include "network/TcpSocket.h" +#include "network/UdpSocket.h" +#include "network/HttpRequest.h" +#include "network/Channel.h" +#include "network/Session.h" +#include "data/ByteData.h" +#include "event/Event.h" + +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +namespace eve::network { + +Module_IMPL(Network, new Network()); + +Network::Network() { + worker_ = std::make_unique(this); + worker_->start(); +} + +Network::~Network() { + if (worker_) worker_->stop(); + worker_.reset(); +} + +TcpSocket* Network::newTcp() { + return new TcpSocket(this); +} + +UdpSocket* Network::newUdp() { + return new UdpSocket(this); +} + +HttpRequest* Network::newHttp(std::string method, std::string url) { + return new HttpRequest(this, std::move(method), std::move(url)); +} + +Channel* Network::newChannel(TcpSocket* socket) { + if (!socket) return nullptr; + auto* ch = new Channel(socket); + bindChannel(socket, ch); + return ch; +} + +Session* Network::newSession() { + return new Session(); +} + +void Network::setTimeout(int ms) { + timeoutMs_ = ms; +} + +int Network::getTimeout() const { + return timeoutMs_; +} + +void Network::setVerifySsl(bool verify) { + verifySsl_ = verify; +} + +bool Network::getVerifySsl() const { + return verifySsl_; +} + +void Network::post(NetCompletion c) { + if (worker_) worker_->post(std::move(c)); +} + +void Network::drainForTest(std::vector& out) { + if (worker_) worker_->drain(out); +} + +void Network::watchTcp(TcpSocket* sock) { + std::lock_guard lock(watchMu_); + if (std::find(watchedTcp_.begin(), watchedTcp_.end(), sock) == watchedTcp_.end()) + watchedTcp_.push_back(sock); +} + +void Network::unwatchTcp(TcpSocket* sock) { + std::lock_guard lock(watchMu_); + watchedTcp_.erase(std::remove(watchedTcp_.begin(), watchedTcp_.end(), sock), watchedTcp_.end()); +} + +void Network::watchUdp(UdpSocket* sock) { + std::lock_guard lock(watchMu_); + if (std::find(watchedUdp_.begin(), watchedUdp_.end(), sock) == watchedUdp_.end()) + watchedUdp_.push_back(sock); +} + +void Network::unwatchUdp(UdpSocket* sock) { + std::lock_guard lock(watchMu_); + watchedUdp_.erase(std::remove(watchedUdp_.begin(), watchedUdp_.end(), sock), watchedUdp_.end()); +} + +void Network::bindChannel(TcpSocket* sock, Channel* ch) { + std::lock_guard lock(channelMu_); + channels_[sock] = ch; +} + +void Network::unbindChannel(TcpSocket* sock) { + std::lock_guard lock(channelMu_); + channels_.erase(sock); +} + +Channel* Network::channelFor(TcpSocket* sock) const { + std::lock_guard lock(channelMu_); + auto it = channels_.find(sock); + return it == channels_.end() ? nullptr : it->second; +} + +void Network::pollSockets() { + std::vector tcpCopy; + std::vector udpCopy; + { + std::lock_guard lock(watchMu_); + tcpCopy = watchedTcp_; + udpCopy = watchedUdp_; + } + for (TcpSocket* sock : tcpCopy) { + if (!sock) continue; + if (sock->isListening() && sock->server()) { + try { + Poco::Net::StreamSocket ss = sock->server()->acceptConnection(); + ss.setBlocking(false); + auto peer = std::make_unique(this); + auto stream = std::make_unique(ss); + peer->setConnectedSocket(std::move(stream)); + TcpSocket* raw = peer.get(); + sock->pushAccepted(std::move(peer)); + watchTcp(raw); + NetCompletion c; + c.type = NetEvType::Conn; + c.kind = NetKind::Tcp; + c.handle = raw; + c.reason = "ok"; + c.peer = raw->getPeer(); + post(std::move(c)); + } catch (...) { + } + } + if (sock->isConnected() && sock->stream()) { + try { + char buf[64 * 1024]; + int n = sock->stream()->receiveBytes(buf, sizeof(buf)); + if (n > 0) { + NetCompletion c; + c.type = NetEvType::Data; + c.kind = NetKind::Tcp; + c.handle = sock; + c.bytes = std::make_shared>(buf, buf + n); + c.peer = sock->getPeer(); + post(std::move(c)); + } else if (n == 0) { + NetCompletion c; + c.type = NetEvType::Err; + c.kind = NetKind::Tcp; + c.handle = sock; + c.reason = "closed"; + post(std::move(c)); + unwatchTcp(sock); + } + } catch (const Poco::Exception&) { + } catch (...) { + } + } + } + for (UdpSocket* sock : udpCopy) { + if (!sock || !sock->datagram()) continue; + try { + char buf[64 * 1024]; + Poco::Net::SocketAddress sender; + int n = sock->datagram()->receiveFrom(buf, sizeof(buf), sender); + if (n > 0) { + NetCompletion c; + c.type = NetEvType::Data; + c.kind = NetKind::Udp; + c.handle = sock; + c.bytes = std::make_shared>(buf, buf + n); + c.peer = sender.toString(); + post(std::move(c)); + } + } catch (const Poco::Exception&) { + } catch (...) { + } + } +} + +void Network::emitCompletion(const NetCompletion& c) { + auto* ev = eve::ModuleManager::getInstance("Event"); + if (!ev) return; + + using eve::event::Variant; + using eve::event::Message; + + std::vector args; + switch (c.type) { + case NetEvType::Conn: + args.push_back(Variant::makePtr(c.handle)); + args.push_back(Variant::makeString(c.reason.empty() ? "ok" : c.reason)); + ev->push(new Message("netconn", args)); + break; + case NetEvType::Data: { + Channel* ch = channelFor(static_cast(c.handle)); + if (ch && c.bytes) { + ch->feed(*c.bytes); + return; + } + eve::data::ByteData* bd = nullptr; + if (c.bytes && !c.bytes->empty()) + bd = new eve::data::ByteData(c.bytes->data(), c.bytes->size()); + args.push_back(Variant::makePtr(c.handle)); + args.push_back(Variant::makePtr(bd)); + args.push_back(Variant::makeString(c.peer)); + ev->push(new Message("netdata", args)); + break; + } + case NetEvType::Err: { + Channel* ch = channelFor(static_cast(c.handle)); + if (ch) { + args.push_back(Variant::makePtr(ch)); + args.push_back(Variant::makeString(c.reason)); + ev->push(new Message("chclose", args)); + } + args.clear(); + args.push_back(Variant::makePtr(c.handle)); + args.push_back(Variant::makeString(c.reason)); + ev->push(new Message("neterr", args)); + break; + } + case NetEvType::HttpResp: { + eve::data::ByteData* bd = nullptr; + if (c.bytes && !c.bytes->empty()) + bd = new eve::data::ByteData(c.bytes->data(), c.bytes->size()); + args.push_back(Variant::makePtr(c.handle)); + args.push_back(Variant::makeInt(c.status)); + args.push_back(Variant::makePtr(bd)); + ev->push(new Message("httpresp", args)); + break; + } + case NetEvType::ChMsg: { + eve::data::ByteData* bd = nullptr; + if (c.bytes && !c.bytes->empty()) + bd = new eve::data::ByteData(c.bytes->data(), c.bytes->size()); + args.push_back(Variant::makePtr(c.handle)); + args.push_back(Variant::makePtr(bd)); + ev->push(new Message("chmsg", args)); + break; + } + case NetEvType::ChClose: + args.push_back(Variant::makePtr(c.handle)); + args.push_back(Variant::makeString(c.reason)); + ev->push(new Message("chclose", args)); + break; + } +} + +void Network::pump() { + std::vector batch; + if (worker_) worker_->drain(batch); + for (auto& c : batch) emitCompletion(c); +} + +void Network::expose(ssq::Table& table) { + auto cls = table.addClass(name, Network::create, false); + expose(cls); + + auto tcp = table.addClass( + "TcpSocket", std::function([]() { return new TcpSocket(Network::create()); }), true); + tcp.addFunc("connect", &TcpSocket::connect); + tcp.addFunc("listen", &TcpSocket::listen); + tcp.addFunc("accept", &TcpSocket::accept); + tcp.addFunc("send", &TcpSocket::send); + tcp.addFunc("sendString", &TcpSocket::sendString); + tcp.addFunc("close", &TcpSocket::close); + tcp.addFunc("isConnected", &TcpSocket::isConnected); + tcp.addFunc("getPeer", &TcpSocket::getPeer); + + auto udp = table.addClass( + "UdpSocket", std::function([]() { return new UdpSocket(Network::create()); }), true); + udp.addFunc("bind", &UdpSocket::bind); + udp.addFunc("connect", &UdpSocket::connect); + udp.addFunc("sendTo", &UdpSocket::sendTo); + udp.addFunc("sendToString", &UdpSocket::sendToString); + udp.addFunc("send", &UdpSocket::send); + udp.addFunc("sendString", &UdpSocket::sendString); + udp.addFunc("close", &UdpSocket::close); + + auto http = table.addClass( + "HttpRequest", + std::function( + []() { return new HttpRequest(Network::create(), "GET", "http://127.0.0.1/"); }), + true); + http.addFunc("setHeader", &HttpRequest::setHeader); + http.addFunc("setBody", &HttpRequest::setBody); + http.addFunc("setBodyString", &HttpRequest::setBodyString); + http.addFunc("setTimeout", &HttpRequest::setTimeout); + http.addFunc("setVerifySsl", &HttpRequest::setVerifySsl); + http.addFunc("submit", &HttpRequest::submit); + + auto ch = table.addClass( + "Channel", std::function([]() { return new Channel(nullptr); }), true); + ch.addFunc("sendMsg", &Channel::sendMsg); + ch.addFunc("sendMsgString", &Channel::sendMsgString); + ch.addFunc("getSocket", &Channel::getSocket); + + auto sess = table.addClass( + "Session", std::function([]() { return new Session(); }), true); + sess.addFunc("add", &Session::add); + sess.addFunc("get", &Session::get); + sess.addFunc("remove", &Session::remove); + sess.addFunc("closeAll", &Session::closeAll); +} + +void Network::expose(ssq::Class& cls) { + cls.addFunc("getName", &Network::getName); + cls.addFunc("newTcp", &Network::newTcp); + cls.addFunc("newUdp", &Network::newUdp); + cls.addFunc("newHttp", &Network::newHttp); + cls.addFunc("newChannel", &Network::newChannel); + cls.addFunc("newSession", &Network::newSession); + cls.addFunc("pump", &Network::pump); + cls.addFunc("setTimeout", &Network::setTimeout); + cls.addFunc("setVerifySsl", &Network::setVerifySsl); +} + +} // namespace eve::network diff --git a/src/modules/network/Network.h b/src/modules/network/Network.h index d30ef2ef..fc8a809f 100644 --- a/src/modules/network/Network.h +++ b/src/modules/network/Network.h @@ -1,14 +1,70 @@ #pragma once #include "common/Module.h" +#include "NetTypes.h" -namespace eve::network -{ +#include +#include +#include +#include +#include +namespace eve::network { -class Network : public Module -{ +class TcpSocket; +class UdpSocket; +class HttpRequest; +class Channel; +class Session; +class NetWorker; + +class Network : public Module { +public: + Module_REG(Network); + Network(); + ~Network() override; + + TcpSocket* newTcp(); + UdpSocket* newUdp(); + HttpRequest* newHttp(std::string method, std::string url); + Channel* newChannel(TcpSocket* socket); + Session* newSession(); + + void setTimeout(int ms); + int getTimeout() const; + void setVerifySsl(bool verify); + bool getVerifySsl() const; + + void pump(); + + void post(NetCompletion c); + void drainForTest(std::vector& out); + + NetWorker* worker() const { return worker_.get(); } + + void watchTcp(TcpSocket* sock); + void unwatchTcp(TcpSocket* sock); + void watchUdp(UdpSocket* sock); + void unwatchUdp(UdpSocket* sock); + void pollSockets(); // called from NetWorker thread + + void bindChannel(TcpSocket* sock, Channel* ch); + void unbindChannel(TcpSocket* sock); + Channel* channelFor(TcpSocket* sock) const; + +private: + void emitCompletion(const NetCompletion& c); + + int timeoutMs_ = 10000; + bool verifySsl_ = true; + std::unique_ptr worker_; + + mutable std::mutex watchMu_; + std::vector watchedTcp_; + std::vector watchedUdp_; + + mutable std::mutex channelMu_; + std::unordered_map channels_; }; - -} \ No newline at end of file +} // namespace eve::network diff --git a/src/modules/network/Session.cpp b/src/modules/network/Session.cpp new file mode 100644 index 00000000..f6019830 --- /dev/null +++ b/src/modules/network/Session.cpp @@ -0,0 +1,31 @@ +#include "network/Session.h" +#include "network/Channel.h" +#include "network/TcpSocket.h" + +namespace eve::network { + +Session::~Session() { + closeAll(); +} + +void Session::add(std::string name, Channel* ch) { + channels_[std::move(name)] = ch; +} + +Channel* Session::get(std::string name) { + auto it = channels_.find(name); + return it == channels_.end() ? nullptr : it->second; +} + +void Session::remove(std::string name) { + channels_.erase(name); +} + +void Session::closeAll() { + for (auto& kv : channels_) { + if (kv.second && kv.second->getSocket()) kv.second->getSocket()->close(); + } + channels_.clear(); +} + +} // namespace eve::network diff --git a/src/modules/network/Session.h b/src/modules/network/Session.h new file mode 100644 index 00000000..dc15345a --- /dev/null +++ b/src/modules/network/Session.h @@ -0,0 +1,24 @@ +#pragma once + +#include +#include + +namespace eve::network { + +class Channel; + +class Session { +public: + Session() = default; + ~Session(); + + void add(std::string name, Channel* ch); + Channel* get(std::string name); + void remove(std::string name); + void closeAll(); + +private: + std::unordered_map channels_; +}; + +} // namespace eve::network diff --git a/src/modules/network/TcpSocket.cpp b/src/modules/network/TcpSocket.cpp new file mode 100644 index 00000000..a8d5587c --- /dev/null +++ b/src/modules/network/TcpSocket.cpp @@ -0,0 +1,207 @@ +#include "network/TcpSocket.h" +#include "network/Network.h" +#include "network/NetWorker.h" +#include "data/ByteData.h" + +#include +#include +#include +#include +#include + +namespace eve::network { + +TcpSocket::TcpSocket(Network* net) : net_(net) {} + +TcpSocket::~TcpSocket() { + close(); +} + +void TcpSocket::setConnectedSocket(std::unique_ptr sock) { + stream_ = std::move(sock); + connected_ = stream_ != nullptr; + if (stream_) { + try { + peer_ = stream_->peerAddress().toString(); + } catch (...) { + peer_.clear(); + } + } +} + +Poco::Net::StreamSocket* TcpSocket::stream() { + return stream_.get(); +} + +Poco::Net::ServerSocket* TcpSocket::server() { + return server_.get(); +} + +void TcpSocket::pushAccepted(std::unique_ptr sock) { + std::lock_guard lock(acceptMu_); + accepted_.push_back(std::move(sock)); +} + +bool TcpSocket::takeAccepted(std::unique_ptr& out) { + std::lock_guard lock(acceptMu_); + if (accepted_.empty()) return false; + out = std::move(accepted_.front()); + accepted_.erase(accepted_.begin()); + return true; +} + +bool TcpSocket::connect(std::string host, uint16_t port) { + if (!net_ || !net_->worker()) return false; + auto* self = this; + net_->worker()->submit([self, host, port]() { + NetCompletion c; + c.kind = NetKind::Tcp; + c.handle = self; + try { + auto sock = std::make_unique(); + sock->connect(Poco::Net::SocketAddress(host, port), Poco::Timespan(self->net_->getTimeout(), 0)); + sock->setBlocking(false); + self->setConnectedSocket(std::move(sock)); + c.type = NetEvType::Conn; + c.reason = "ok"; + c.peer = self->getPeer(); + self->net_->post(std::move(c)); + self->net_->watchTcp(self); + } catch (const Poco::Net::ConnectionRefusedException&) { + c.type = NetEvType::Err; + c.reason = "refused"; + self->net_->post(std::move(c)); + NetCompletion fail; + fail.type = NetEvType::Conn; + fail.kind = NetKind::Tcp; + fail.handle = self; + fail.reason = "fail"; + self->net_->post(std::move(fail)); + } catch (const Poco::Exception& ex) { + c.type = NetEvType::Err; + std::string msg = ex.displayText(); + if (msg.find("timed") != std::string::npos || msg.find("Timeout") != std::string::npos) + c.reason = "timeout"; + else if (msg.find("host") != std::string::npos || msg.find("DNS") != std::string::npos) + c.reason = "dns"; + else + c.reason = "refused"; + self->net_->post(std::move(c)); + NetCompletion fail; + fail.type = NetEvType::Conn; + fail.kind = NetKind::Tcp; + fail.handle = self; + fail.reason = "fail"; + self->net_->post(std::move(fail)); + } + }); + return true; +} + +bool TcpSocket::listen(uint16_t port) { + try { + server_ = std::make_unique(port); + server_->setBlocking(false); + listening_ = true; + if (net_) net_->watchTcp(this); + return true; + } catch (...) { + listening_ = false; + server_.reset(); + if (net_) { + NetCompletion c; + c.type = NetEvType::Err; + c.kind = NetKind::Tcp; + c.handle = this; + c.reason = "refused"; + net_->post(std::move(c)); + } + return false; + } +} + +TcpSocket* TcpSocket::accept() { + std::unique_ptr sock; + if (!takeAccepted(sock)) return nullptr; + return sock.release(); +} + +bool TcpSocket::send(eve::data::ByteData* data) { + if (!data || !stream_ || !connected_ || !net_ || !net_->worker()) return false; + size_t n = data->getSize(); + if (sendQueued_ + n > kMaxSendBuffer) { + NetCompletion c; + c.type = NetEvType::Err; + c.kind = NetKind::Tcp; + c.handle = this; + c.reason = "limit"; + net_->post(std::move(c)); + return false; + } + auto buf = std::make_shared>( + static_cast(data->getData()), + static_cast(data->getData()) + n); + addSendQueued(n); + auto* self = this; + net_->worker()->submit([self, buf, n]() { + try { + if (!self->stream_) { + self->subSendQueued(n); + return; + } + int sent = self->stream_->sendBytes(buf->data(), static_cast(buf->size())); + (void)sent; + self->subSendQueued(n); + } catch (...) { + self->subSendQueued(n); + NetCompletion c; + c.type = NetEvType::Err; + c.kind = NetKind::Tcp; + c.handle = self; + c.reason = "closed"; + self->net_->post(std::move(c)); + } + }); + return true; +} + +bool TcpSocket::sendString(std::string s) { + if (s.empty()) return false; + eve::data::ByteData data(s.data(), s.size()); + return send(&data); +} + +void TcpSocket::close() { + if (net_) net_->unwatchTcp(this); + connected_ = false; + listening_ = false; + try { + if (stream_) stream_->close(); + } catch (...) { + } + stream_.reset(); + try { + if (server_) server_->close(); + } catch (...) { + } + server_.reset(); +} + +bool TcpSocket::isConnected() const { + return connected_ && stream_ != nullptr; +} + +std::string TcpSocket::getPeer() const { + return peer_; +} + +uint16_t TcpSocket::getLocalPort() const { + try { + if (server_) return static_cast(server_->address().port()); + if (stream_) return static_cast(stream_->address().port()); + } catch (...) { + } + return 0; +} + +} // namespace eve::network diff --git a/src/modules/network/TcpSocket.h b/src/modules/network/TcpSocket.h new file mode 100644 index 00000000..b39cdbb4 --- /dev/null +++ b/src/modules/network/TcpSocket.h @@ -0,0 +1,63 @@ +#pragma once + +#include "NetTypes.h" + +#include +#include +#include +#include +#include + +namespace eve::data { +class ByteData; +} + +namespace Poco::Net { +class StreamSocket; +class ServerSocket; +} + +namespace eve::network { + +class Network; + +class TcpSocket { +public: + explicit TcpSocket(Network* net); + ~TcpSocket(); + + bool connect(std::string host, uint16_t port); + bool listen(uint16_t port); + TcpSocket* accept(); + bool send(eve::data::ByteData* data); + bool sendString(std::string s); + void close(); + bool isConnected() const; + std::string getPeer() const; + uint16_t getLocalPort() const; + + // Internal: used by Network::pump / NetWorker + Network* network() const { return net_; } + void setConnectedSocket(std::unique_ptr sock); + Poco::Net::StreamSocket* stream(); + Poco::Net::ServerSocket* server(); + bool isListening() const { return listening_; } + bool takeAccepted(std::unique_ptr& out); + void pushAccepted(std::unique_ptr sock); + size_t pendingSendBytes() const { return sendQueued_; } + void addSendQueued(size_t n) { sendQueued_ += n; } + void subSendQueued(size_t n) { sendQueued_ = sendQueued_ > n ? sendQueued_ - n : 0; } + +private: + Network* net_ = nullptr; + std::unique_ptr stream_; + std::unique_ptr server_; + bool listening_ = false; + bool connected_ = false; + size_t sendQueued_ = 0; + std::mutex acceptMu_; + std::vector> accepted_; + std::string peer_; +}; + +} // namespace eve::network diff --git a/src/modules/network/UdpSocket.cpp b/src/modules/network/UdpSocket.cpp new file mode 100644 index 00000000..6bd4c1c9 --- /dev/null +++ b/src/modules/network/UdpSocket.cpp @@ -0,0 +1,136 @@ +#include "network/UdpSocket.h" +#include "network/Network.h" +#include "network/NetWorker.h" +#include "data/ByteData.h" + +#include +#include +#include + +namespace eve::network { + +UdpSocket::UdpSocket(Network* net) : net_(net) {} + +UdpSocket::~UdpSocket() { + close(); +} + +Poco::Net::DatagramSocket* UdpSocket::datagram() { + return sock_.get(); +} + +bool UdpSocket::bind(uint16_t port) { + try { + sock_ = std::make_unique(); + sock_->bind(Poco::Net::SocketAddress(port), true); + sock_->setBlocking(false); + bound_ = true; + if (net_) net_->watchUdp(this); + return true; + } catch (...) { + sock_.reset(); + bound_ = false; + if (net_) { + NetCompletion c; + c.type = NetEvType::Err; + c.kind = NetKind::Udp; + c.handle = this; + c.reason = "refused"; + net_->post(std::move(c)); + } + return false; + } +} + +bool UdpSocket::connect(std::string host, uint16_t port) { + try { + if (!sock_) sock_ = std::make_unique(); + sock_->connect(Poco::Net::SocketAddress(host, port)); + sock_->setBlocking(false); + connected_ = true; + if (net_) net_->watchUdp(this); + return true; + } catch (...) { + if (net_) { + NetCompletion c; + c.type = NetEvType::Err; + c.kind = NetKind::Udp; + c.handle = this; + c.reason = "refused"; + net_->post(std::move(c)); + } + return false; + } +} + +bool UdpSocket::sendTo(eve::data::ByteData* data, std::string host, uint16_t port) { + if (!data || !sock_ || !net_ || !net_->worker()) return false; + size_t n = data->getSize(); + auto buf = std::make_shared>( + static_cast(data->getData()), + static_cast(data->getData()) + n); + auto* self = this; + net_->worker()->submit([self, buf, host, port]() { + try { + if (!self->sock_) return; + self->sock_->sendTo(buf->data(), static_cast(buf->size()), + Poco::Net::SocketAddress(host, port)); + } catch (...) { + NetCompletion c; + c.type = NetEvType::Err; + c.kind = NetKind::Udp; + c.handle = self; + c.reason = "closed"; + self->net_->post(std::move(c)); + } + }); + return true; +} + +bool UdpSocket::sendToString(std::string s, std::string host, uint16_t port) { + if (s.empty()) return false; + eve::data::ByteData data(s.data(), s.size()); + return sendTo(&data, host, port); +} + +bool UdpSocket::send(eve::data::ByteData* data) { + if (!data || !sock_ || !connected_ || !net_ || !net_->worker()) return false; + size_t n = data->getSize(); + auto buf = std::make_shared>( + static_cast(data->getData()), + static_cast(data->getData()) + n); + auto* self = this; + net_->worker()->submit([self, buf]() { + try { + if (!self->sock_) return; + self->sock_->sendBytes(buf->data(), static_cast(buf->size())); + } catch (...) { + NetCompletion c; + c.type = NetEvType::Err; + c.kind = NetKind::Udp; + c.handle = self; + c.reason = "closed"; + self->net_->post(std::move(c)); + } + }); + return true; +} + +bool UdpSocket::sendString(std::string s) { + if (s.empty()) return false; + eve::data::ByteData data(s.data(), s.size()); + return send(&data); +} + +void UdpSocket::close() { + if (net_) net_->unwatchUdp(this); + bound_ = false; + connected_ = false; + try { + if (sock_) sock_->close(); + } catch (...) { + } + sock_.reset(); +} + +} // namespace eve::network diff --git a/src/modules/network/UdpSocket.h b/src/modules/network/UdpSocket.h new file mode 100644 index 00000000..238db7bb --- /dev/null +++ b/src/modules/network/UdpSocket.h @@ -0,0 +1,47 @@ +#pragma once + +#include "NetTypes.h" + +#include +#include +#include +#include +#include + +namespace eve::data { +class ByteData; +} + +namespace Poco::Net { +class DatagramSocket; +} + +namespace eve::network { + +class Network; + +class UdpSocket { +public: + explicit UdpSocket(Network* net); + ~UdpSocket(); + + bool bind(uint16_t port); + bool connect(std::string host, uint16_t port); + bool sendTo(eve::data::ByteData* data, std::string host, uint16_t port); + bool sendToString(std::string s, std::string host, uint16_t port); + bool send(eve::data::ByteData* data); + bool sendString(std::string s); + void close(); + + Poco::Net::DatagramSocket* datagram(); + bool isBound() const { return bound_; } + Network* network() const { return net_; } + +private: + Network* net_ = nullptr; + std::unique_ptr sock_; + bool bound_ = false; + bool connected_ = false; +}; + +} // namespace eve::network diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 5e797bae..5121094a 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -6,6 +6,7 @@ file(GLOB all_test_cpp graphic.cpp model.cpp mouse.cpp + network.cpp RenderSystem.cpp simplesquirrel.cpp window.cpp @@ -35,28 +36,38 @@ add_executable( if(WIN32) - set(SystemLibraries ${SystemLibraries} winmm version imm32 Setupapi) + set(SystemLibraries ${SystemLibraries} winmm version imm32 Setupapi ws2_32 iphlpapi) elseif(UNIX) set(SystemLibraries ${SystemLibraries} dl X11 pthread) endif() if(WIN32) if(${CMAKE_BUILD_TYPE} STREQUAL "Debug") - set(ThirdParty ${ThirdParty} physfsmdd medialoader_imagemdd zlibstaticd) + set(ThirdParty ${ThirdParty} physfsmdd medialoader_imagemdd zlibstaticd lz4mdd) else() - set(ThirdParty ${ThirdParty} physfsmd medialoader_imagemd zlibstatic) + set(ThirdParty ${ThirdParty} physfsmd medialoader_imagemd zlibstatic lz4md) endif() else() - set(ThirdParty ${ThirdParty} physfs medialoader_image z) + set(ThirdParty ${ThirdParty} physfs medialoader_image z lz4) endif() if(WIN32 AND (${CMAKE_BUILD_TYPE} STREQUAL "Debug")) - set(ThirdParty ${ThirdParty} simplesquirrel_static sqstdlib_static squirrel_static SDL2d ${SystemLibraries}) + set(ThirdParty ${ThirdParty} simplesquirrel_static sqstdlib_static squirrel_static SDL2d + PocoFoundationmdd PocoNetmdd + ${SystemLibraries}) else() - set(ThirdParty ${ThirdParty} simplesquirrel_static sqstdlib_static squirrel_static SDL2 ${SystemLibraries}) + if(WIN32) + set(ThirdParty ${ThirdParty} simplesquirrel_static sqstdlib_static squirrel_static SDL2 + PocoFoundationmd PocoNetmd + ${SystemLibraries}) + else() + set(ThirdParty ${ThirdParty} simplesquirrel_static sqstdlib_static squirrel_static SDL2 + PocoFoundation PocoNet + ${SystemLibraries}) + endif() endif() -set(EVELIBS EVCmdLine EVScripts EVWindow EVFileSystem EVMouse EVGraphics EVImage EVDevTools EVCommon) +set(EVELIBS EVCmdLine EVScripts EVWindow EVFileSystem EVMouse EVGraphics EVImage EVDevTools EVCommon EVEvent EVNetwork EVData) target_link_libraries(unit_test ${EVELIBS} zeroerr ${ThirdParty} ${EVENGINE_VULKAN_LIB}) diff --git a/test/network.cpp b/test/network.cpp new file mode 100644 index 00000000..d1f0c74b --- /dev/null +++ b/test/network.cpp @@ -0,0 +1,236 @@ +#include "zeroerr/assert.h" +#include "zeroerr/unittest.h" + +#include "event/Event.h" +#include "network/Network.h" +#include "network/TcpSocket.h" +#include "network/UdpSocket.h" +#include "network/HttpRequest.h" +#include "network/Channel.h" +#include "network/Session.h" +#include "data/ByteData.h" + +#include +#include +#include + +using eve::event::Event; +using eve::event::Message; +using eve::event::Variant; + +TEST_CASE("event.VariantMessage") { + auto* ev = eve::event::Event::create(); + std::vector args; + args.push_back(Variant::makeInt(200)); + args.push_back(Variant::makeString("ok")); + args.push_back(Variant::makePtr(nullptr)); + ev->push(new Message("httpresp", args)); + Message* m = ev->poll(); + REQUIRE(m != nullptr); + CHECK(m->name == "httpresp"); + REQUIRE(m->args.size() == 3); + CHECK(static_cast(m->args[0].type) == static_cast(Variant::Type::Int)); + CHECK(m->args[0].i == 200); + CHECK(m->args[1].s == "ok"); + delete m; +} + +TEST_CASE("network.NetWorkerQueue") { + auto* net = eve::network::Network::create(); + eve::network::NetCompletion c; + c.type = eve::network::NetEvType::Err; + c.reason = "timeout"; + net->post(std::move(c)); + std::vector out; + net->drainForTest(out); + REQUIRE(out.size() == 1); + CHECK(out[0].reason == "timeout"); +} + +TEST_CASE("network.TcpEcho") { + auto* net = eve::network::Network::create(); + auto* ev = eve::event::Event::create(); + REQUIRE(net != nullptr); + REQUIRE(ev != nullptr); + + const uint16_t port = 39281; + auto* server = net->newTcp(); + REQUIRE(server->listen(port)); + + auto* client = net->newTcp(); + REQUIRE(client->connect("127.0.0.1", port)); + + bool clientOk = false; + eve::network::TcpSocket* peer = nullptr; + for (int i = 0; i < 400 && (!clientOk || !peer); ++i) { + net->pump(); + while (auto* m = ev->poll()) { + if (m->name == "netconn" && m->args.size() >= 2 && m->args[1].s == "ok") { + if (m->args[0].p == client) clientOk = true; + } + delete m; + } + if (!peer) peer = server->accept(); + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + REQUIRE(clientOk); + REQUIRE(peer != nullptr); + + eve::data::ByteData payload("ping", 4); + REQUIRE(client->send(&payload)); + + bool got = false; + for (int i = 0; i < 400 && !got; ++i) { + net->pump(); + while (auto* m = ev->poll()) { + if (m->name == "netdata") { + got = true; + if (m->args.size() >= 2 && m->args[1].p) { + delete static_cast(m->args[1].p); + } + } + delete m; + } + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + CHECK(got); + + client->close(); + peer->close(); + server->close(); + delete client; + delete peer; + delete server; +} + +TEST_CASE("network.UdpSendTo") { + auto* net = eve::network::Network::create(); + auto* ev = eve::event::Event::create(); + auto* a = net->newUdp(); + auto* b = net->newUdp(); + REQUIRE(a->bind(39301)); + REQUIRE(b->bind(39302)); + eve::data::ByteData payload("udp!", 4); + REQUIRE(a->sendTo(&payload, "127.0.0.1", 39302)); + + bool got = false; + for (int i = 0; i < 400 && !got; ++i) { + net->pump(); + while (auto* m = ev->poll()) { + if (m->name == "netdata" && m->args.size() >= 1 && m->args[0].p == b) { + got = true; + if (m->args.size() >= 2 && m->args[1].p) + delete static_cast(m->args[1].p); + } + delete m; + } + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + CHECK(got); + a->close(); + b->close(); + delete a; + delete b; +} + +TEST_CASE("network.HttpGetLocal") { + auto* net = eve::network::Network::create(); + auto* ev = eve::event::Event::create(); + const uint16_t port = 39311; + auto* server = net->newTcp(); + REQUIRE(server->listen(port)); + + auto* req = net->newHttp("GET", "http://127.0.0.1:" + std::to_string(port) + "/"); + REQUIRE(req->submit()); + + eve::network::TcpSocket* peer = nullptr; + for (int i = 0; i < 400 && !peer; ++i) { + net->pump(); + while (auto* m = ev->poll()) { delete m; } + peer = server->accept(); + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + REQUIRE(peer != nullptr); + + const char* httpResp = + "HTTP/1.1 200 OK\r\nContent-Length: 5\r\nConnection: close\r\n\r\nhello"; + eve::data::ByteData respData(httpResp, std::strlen(httpResp)); + REQUIRE(peer->send(&respData)); + + bool got = false; + int status = 0; + for (int i = 0; i < 400 && !got; ++i) { + net->pump(); + while (auto* m = ev->poll()) { + if (m->name == "httpresp" && m->args.size() >= 2) { + got = true; + status = static_cast(m->args[1].i); + if (m->args.size() >= 3 && m->args[2].p) + delete static_cast(m->args[2].p); + } + delete m; + } + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + CHECK(got); + CHECK(status == 200); + + peer->close(); + server->close(); + delete peer; + delete server; + delete req; +} + +TEST_CASE("network.ChannelFrames") { + auto* net = eve::network::Network::create(); + auto* ev = eve::event::Event::create(); + const uint16_t port = 39321; + auto* server = net->newTcp(); + REQUIRE(server->listen(port)); + auto* client = net->newTcp(); + REQUIRE(client->connect("127.0.0.1", port)); + + eve::network::TcpSocket* peer = nullptr; + for (int i = 0; i < 400 && !peer; ++i) { + net->pump(); + while (auto* m = ev->poll()) { delete m; } + peer = server->accept(); + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + REQUIRE(peer != nullptr); + + auto* chSend = net->newChannel(client); + auto* chRecv = net->newChannel(peer); + REQUIRE(chSend->sendMsgString("ab")); + REQUIRE(chSend->sendMsgString("cd")); + + int msgs = 0; + for (int i = 0; i < 400 && msgs < 2; ++i) { + net->pump(); + while (auto* m = ev->poll()) { + if (m->name == "chmsg") { + ++msgs; + if (m->args.size() >= 2 && m->args[1].p) + delete static_cast(m->args[1].p); + } + if (m->name == "netdata" && m->args.size() >= 2 && m->args[1].p) + delete static_cast(m->args[1].p); + delete m; + } + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + CHECK(msgs == 2); + + auto* sess = net->newSession(); + sess->add("main", chRecv); + CHECK(sess->get("main") == chRecv); + sess->closeAll(); + + delete sess; + delete chSend; + delete chRecv; + delete client; + delete peer; + delete server; +}