#include "network/Network.h" #include "data/ByteData.h" #include "network/Channel.h" #include "network/HttpRequest.h" #include "network/NetHost.h" #include "network/NetRpc.h" #include "network/NetStream.h" #include "network/NetWorker.h" #include "network/Session.h" #include "network/TcpSocket.h" #include "network/UdpLink.h" #include "network/UdpSocket.h" #include "platform_event/PlatformEvent.h" #include #include #include #include #include #include #include #include #include #include #include namespace eve::network { namespace { void callScript2(const ssq::Object& obj, int64_t a, const std::string& s) { if (obj.isEmpty()) return; ssq::Function f = obj.toFunction(); if (f.isEmpty()) return; HSQUIRRELVM raw = f.getHandle(); SQInteger top = sq_gettop(raw); sq_pushobject(raw, f.getRaw()); sq_pushroottable(raw); ssq::detail::pushValue(raw, a); ssq::detail::pushValue(raw, s); sq_call(raw, 3, SQFalse, SQTrue); sq_settop(raw, top); } void callScript3(const ssq::Object& obj, int64_t a, int64_t b, const std::string& s) { if (obj.isEmpty()) return; ssq::Function f = obj.toFunction(); if (f.isEmpty()) return; HSQUIRRELVM raw = f.getHandle(); SQInteger top = sq_gettop(raw); sq_pushobject(raw, f.getRaw()); sq_pushroottable(raw); ssq::detail::pushValue(raw, a); ssq::detail::pushValue(raw, b); ssq::detail::pushValue(raw, s); sq_call(raw, 4, SQFalse, SQTrue); sq_settop(raw, top); } } // namespace Module_IMPL(Network, new Network()); Network::Network() { worker_ = std::make_unique(this); worker_->start(); eve::cap::provide(this); } Network::~Network() { if (worker_) worker_->stop(); worker_.reset(); } std::unique_ptr Network::makeTcp() { return std::make_unique(this); } TcpSocket* Network::newTcp() { return makeTcp().release(); } std::unique_ptr Network::makeUdp() { return std::make_unique(this); } UdpSocket* Network::newUdp() { return makeUdp().release(); } std::unique_ptr Network::makeHttp(std::string method, std::string url) { return std::make_unique(this, std::move(method), std::move(url)); } HttpRequest* Network::newHttp(std::string method, std::string url) { return makeHttp(std::move(method), std::move(url)).release(); } bool Network::httpRequest(const std::string& method, const std::string& url, const std::string& body, int timeoutMs, int& status, std::string& responseBody) { auto ownedRequest = makeHttp(method, url); HttpRequest* req = ownedRequest.get(); if (!body.empty()) req->setBodyString(body); req->setTimeout(timeoutMs > 0 ? timeoutMs : 10000); if (!req->submit()) return false; const auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(timeoutMs > 0 ? timeoutMs : 1); while (std::chrono::steady_clock::now() < deadline) { std::vector out; drainCompletions(out); for (const auto& c : out) { if (c.handle != req) continue; if (c.type == NetEvType::HttpResp) { status = c.status; if (c.bytes) responseBody.assign(c.bytes->begin(), c.bytes->end()); return true; } if (c.type == NetEvType::Err) return false; } std::this_thread::sleep_for(std::chrono::milliseconds(10)); } return false; } std::unique_ptr Network::makeChannel(TcpSocket* socket) { if (!socket) return nullptr; auto ch = std::make_unique(socket); bindChannel(socket, ch.get()); return ch; } Channel* Network::newChannel(TcpSocket* socket) { return makeChannel(socket).release(); } std::unique_ptr Network::makeSession() { return std::make_unique(); } Session* Network::newSession() { return makeSession().release(); } std::unique_ptr Network::makeWriter() { return std::make_unique(); } NetWriter* Network::newWriter() { return makeWriter().release(); } std::unique_ptr Network::makeReader(std::string bytes) { auto r = std::make_unique(); r->setBytes(std::move(bytes)); return r; } NetReader* Network::newReader(std::string bytes) { return makeReader(std::move(bytes)).release(); } std::unique_ptr Network::makeUdpLink(UdpSocket* socket) { if (!socket) return nullptr; auto link = std::make_unique(this, socket); bindUdpLink(socket, link.get()); return link; } UdpLink* Network::newUdpLink(UdpSocket* socket) { return makeUdpLink(socket).release(); } std::unique_ptr Network::makeRpc(UdpLink* link) { return link ? std::make_unique(link) : nullptr; } NetRpc* Network::newRpc(UdpLink* link) { return makeRpc(link).release(); } std::unique_ptr Network::makeHost() { return std::make_unique(this); } NetHost* Network::newHost() { return makeHost().release(); } void Network::bindUdpLink(UdpSocket* sock, UdpLink* link) { if (sock) udpLinks_[sock] = link; } void Network::unbindUdpLink(UdpSocket* sock) { if (sock) udpLinks_.erase(sock); } void Network::bindUdpHost(UdpSocket* sock, NetHost* host) { if (sock) udpHosts_[sock] = host; } void Network::unbindUdpHost(UdpSocket* sock) { if (sock) udpHosts_.erase(sock); } 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) { ++completions_; ++telemetryRevision_; if (c.bytes && (c.type == NetEvType::Data || c.type == NetEvType::HttpResp)) receivedBytes_ += c.bytes->size(); if (c.type == NetEvType::Err) ++errors_; if (c.type == NetEvType::Conn && c.reason == "ok") ++connections_; if (worker_) worker_->post(std::move(c)); } void Network::recordSent(size_t bytes) { sentBytes_ += bytes; ++telemetryRevision_; } NetTelemetrySnapshot Network::telemetrySnapshot() const { NetTelemetrySnapshot result; result.revision = telemetryRevision_.load(); result.sentBytes = sentBytes_.load(); result.receivedBytes = receivedBytes_.load(); result.completions = completions_.load(); result.errors = errors_.load(); result.connections = connections_.load(); { std::lock_guard lock(watchMu_); result.watchedTcp = watchedTcp_.size(); result.watchedUdp = watchedUdp_.size(); for (const auto* socket : watchedTcp_) if (socket) result.queuedTcpBytes += socket->pendingSendBytes(); } { std::lock_guard lock(channelMu_); result.channels = channels_.size(); } return result; } void Network::resetTelemetry() { sentBytes_ = receivedBytes_ = completions_ = errors_ = connections_ = 0; ++telemetryRevision_; } void Network::drainCompletions(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); ++telemetryRevision_; } } void Network::unwatchTcp(TcpSocket* sock) { std::lock_guard lock(watchMu_); const auto before=watchedTcp_.size(); watchedTcp_.erase(std::remove(watchedTcp_.begin(), watchedTcp_.end(), sock), watchedTcp_.end()); if(before!=watchedTcp_.size())++telemetryRevision_; } void Network::watchUdp(UdpSocket* sock) { std::lock_guard lock(watchMu_); if (std::find(watchedUdp_.begin(), watchedUdp_.end(), sock) == watchedUdp_.end()) { watchedUdp_.push_back(sock); ++telemetryRevision_; } } void Network::unwatchUdp(UdpSocket* sock) { std::lock_guard lock(watchMu_); const auto before=watchedUdp_.size(); watchedUdp_.erase(std::remove(watchedUdp_.begin(), watchedUdp_.end(), sock), watchedUdp_.end()); if(before!=watchedUdp_.size())++telemetryRevision_; } void Network::bindChannel(TcpSocket* sock, Channel* ch) { std::lock_guard lock(channelMu_); channels_[sock] = ch; ++telemetryRevision_; } void Network::unbindChannel(TcpSocket* sock) { std::lock_guard lock(channelMu_); if(channels_.erase(sock))++telemetryRevision_; } 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()) { sock->flushSend(); 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; for (int i = 0; i < 64; ++i) { try { char buf[64 * 1024]; Poco::Net::SocketAddress sender; int n = sock->datagram()->receiveFrom(buf, sizeof(buf), sender); if (n <= 0) break; // would-block 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&) { break; } catch (...) { break; } } } } void Network::emitCompletion(const NetCompletion& c) { auto* ev = eve::ModuleManager::getInstance("PlatformEvent"); if (!ev) return; using eve::platform_event::Message; using eve::platform_event::Variant; 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(std::make_unique("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::makeOwnedPtr(bd)); args.push_back(Variant::makeString(c.peer)); ev->push(std::make_unique("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(std::make_unique("chclose", args)); } args.clear(); args.push_back(Variant::makePtr(c.handle)); args.push_back(Variant::makeString(c.reason)); ev->push(std::make_unique("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::makeOwnedPtr(bd)); ev->push(std::make_unique("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::makeOwnedPtr(bd)); ev->push(std::make_unique("chmsg", args)); break; } case NetEvType::ChClose: args.push_back(Variant::makePtr(c.handle)); args.push_back(Variant::makeString(c.reason)); ev->push(std::make_unique("chclose", args)); break; } } void Network::pump() { std::vector batch; if (worker_) worker_->drain(batch); const int64_t now = std::chrono::duration_cast( std::chrono::steady_clock::now().time_since_epoch()) .count(); for (auto& c : batch) { if (c.type == NetEvType::Data && c.kind == NetKind::Udp) { auto* sock = static_cast(c.handle); auto hit = udpHosts_.find(sock); if (hit != udpHosts_.end()) { if (c.bytes) hit->second->onDatagram(*c.bytes, c.peer); continue; } auto lit = udpLinks_.find(sock); if (lit != udpLinks_.end()) { if (c.bytes) lit->second->onDatagram(*c.bytes, c.peer); continue; } } emitCompletion(c); } for (auto& kv : udpHosts_) kv.second->pump(now); for (auto& kv : udpLinks_) kv.second->pump(now); } 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); auto writer = table.addClass( "NetWriter", std::function([]() { return new NetWriter(); }), true); writer.addFunc("writeU8", [](NetWriter* w, int64_t v) { w->writeU8(static_cast(v)); }); writer.addFunc("writeI8", [](NetWriter* w, int64_t v) { w->writeI8(static_cast(v)); }); writer.addFunc("writeU16", [](NetWriter* w, int64_t v) { w->writeU16(static_cast(v)); }); writer.addFunc("writeI16", [](NetWriter* w, int64_t v) { w->writeI16(static_cast(v)); }); writer.addFunc("writeU32", [](NetWriter* w, int64_t v) { w->writeU32(static_cast(v)); }); writer.addFunc("writeI32", [](NetWriter* w, int64_t v) { w->writeI32(static_cast(v)); }); writer.addFunc("writeU64", [](NetWriter* w, int64_t v) { w->writeU64(static_cast(v)); }); writer.addFunc("writeI64", [](NetWriter* w, int64_t v) { w->writeI64(static_cast(v)); }); writer.addFunc("writeF32", &NetWriter::writeF32); writer.addFunc("writeF64", &NetWriter::writeF64); writer.addFunc("writeBool", &NetWriter::writeBool); writer.addFunc("writeString", &NetWriter::writeString); writer.addFunc("writeBytes", [](NetWriter* w, eve::data::ByteData* d) { if (d) w->writeBytes(d->getData(), d->getSize()); }); writer.addFunc("toString", [](NetWriter* w) { return w->toString(); }); writer.addFunc("size", [](NetWriter* w) { return static_cast(w->size()); }); auto reader = table.addClass( "NetReader", std::function([]() { return new NetReader(); }), true); reader.addFunc("init", [](NetReader* r, eve::data::ByteData* d) { return d ? r->init(d->getData(), d->getSize()) : false; }); reader.addFunc("initString", [](NetReader* r, const std::string& s) { return r->setBytes(s); }); reader.addFunc("u8", [](NetReader* r) { return static_cast(r->u8()); }); reader.addFunc("i8", [](NetReader* r) { return static_cast(r->i8()); }); reader.addFunc("u16", [](NetReader* r) { return static_cast(r->u16()); }); reader.addFunc("i16", [](NetReader* r) { return static_cast(r->i16()); }); reader.addFunc("u32", [](NetReader* r) { return static_cast(r->u32()); }); reader.addFunc("i32", [](NetReader* r) { return static_cast(r->i32()); }); reader.addFunc("u64", [](NetReader* r) { return static_cast(r->u64()); }); reader.addFunc("i64", [](NetReader* r) { return static_cast(r->i64()); }); reader.addFunc("f32", &NetReader::f32); reader.addFunc("f64", &NetReader::f64); reader.addFunc("bool", &NetReader::b); reader.addFunc("str", [](NetReader* r) { return r->str(); }); reader.addFunc("bytes", [](NetReader* r, int64_t n) { auto v = r->bytes(static_cast(n)); return std::string(v.data(), v.size()); }); reader.addFunc("remaining", [](NetReader* r) { return static_cast(r->remaining()); }); reader.addFunc("pos", [](NetReader* r) { return static_cast(r->pos()); }); reader.addFunc("ok", [](NetReader* r) { return r->ok(); }); auto link = table.addClass( "UdpLink", std::function([]() { return new UdpLink(nullptr, nullptr); }), true); link.addFunc("setRemote", &UdpLink::setRemote); link.addFunc("setRemoteString", &UdpLink::setRemoteString); link.addFunc("sendReliable", [](UdpLink* l, int64_t ch, const std::string& s) { l->sendString(UdpLink::MsgType::Reliable, static_cast(ch), s); }); link.addFunc("sendUnreliable", [](UdpLink* l, int64_t ch, const std::string& s) { l->sendString(UdpLink::MsgType::Unreliable, static_cast(ch), s); }); link.addFunc("sendOrdered", [](UdpLink* l, int64_t ch, const std::string& s) { l->sendString(UdpLink::MsgType::UnreliableOrdered, static_cast(ch), s); }); link.addFunc("onMessage", [](UdpLink* l, ssq::Object fn) { if (!l) return; l->setMessageHandler( [fn](UdpLink::MsgType, uint8_t ch, const char* d, size_t n) { callScript2(fn, ch, std::string(d, n)); }); }); link.addFunc("setLossRate", &UdpLink::setLossRate); link.addFunc("setTimeoutMs", &UdpLink::setTimeoutMs); link.addFunc("isAlive", [](UdpLink* l) { return l && l->isAlive(); }); link.addFunc("peer", &UdpLink::peer); link.addFunc("peerId", [](UdpLink* l) { return static_cast(l->peerId()); }); link.addFunc("pendingReliable", [](UdpLink* l) { return static_cast(l->pendingReliable()); }); link.addFunc("pendingFragments", [](UdpLink* l) { return static_cast(l->pendingFragments()); }); auto rpc = table.addClass( "NetRpc", std::function([]() { return new NetRpc(nullptr); }), true); rpc.addFunc("callRpc", [](NetRpc* r, int64_t msgId, const std::string& payload, bool reliable) { r->callString(static_cast(msgId), payload, reliable); }); rpc.addFunc("registerRpc", [](NetRpc* r, int64_t msgId, ssq::Object fn) { r->registerScript(static_cast(msgId), fn); }); auto host = table.addClass( "NetHost", std::function([]() { return new NetHost(nullptr); }), true); host.addFunc("start", &NetHost::start); host.addFunc("onMessage", [](NetHost* h, ssq::Object fn) { if (!h) return; h->setMessageHandler( [fn](uint32_t peerId, UdpLink::MsgType, uint8_t ch, const char* d, size_t n) { callScript3(fn, peerId, ch, std::string(d, n)); }); }); host.addFunc("onPeerConnected", [](NetHost* h, ssq::Object fn) { if (!h) return; h->setPeerConnectedHandler([fn](uint32_t id) { callScript2(fn, id, ""); }); }); host.addFunc("onPeerDisconnected", [](NetHost* h, ssq::Object fn) { if (!h) return; h->setPeerDisconnectedHandler([fn](uint32_t id) { callScript2(fn, id, ""); }); }); host.addFunc("sendReliable", [](NetHost* h, int64_t peerId, int64_t ch, const std::string& s) { h->sendStringTo(static_cast(peerId), UdpLink::MsgType::Reliable, static_cast(ch), s); }); host.addFunc("sendUnreliable", [](NetHost* h, int64_t peerId, int64_t ch, const std::string& s) { h->sendStringTo(static_cast(peerId), UdpLink::MsgType::Unreliable, static_cast(ch), s); }); host.addFunc("sendOrdered", [](NetHost* h, int64_t peerId, int64_t ch, const std::string& s) { h->sendStringTo(static_cast(peerId), UdpLink::MsgType::UnreliableOrdered, static_cast(ch), s); }); host.addFunc("link", &NetHost::linkByPeerId); host.addFunc("peerCount", [](NetHost* h) { return static_cast(h->peerCount()); }); host.addFunc("setLossRate", &NetHost::setLossRate); host.addFunc("setTimeoutMs", &NetHost::setTimeoutMs); } 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("newWriter", &Network::newWriter); cls.addFunc("newReader", &Network::newReader); cls.addFunc("newUdpLink", &Network::newUdpLink); cls.addFunc("newRpc", &Network::newRpc); cls.addFunc("newHost", &Network::newHost); cls.addFunc("pump", &Network::pump); cls.addFunc("setTimeout", &Network::setTimeout); cls.addFunc("setVerifySsl", &Network::setVerifySsl); } } // namespace eve::network