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; +}