diff options
-rw-r--r-- | include/erebos/identity.h | 1 | ||||
-rw-r--r-- | include/erebos/network.h | 36 | ||||
-rw-r--r-- | src/identity.cpp | 7 | ||||
-rw-r--r-- | src/network.cpp | 95 | ||||
-rw-r--r-- | src/network.h | 26 |
5 files changed, 160 insertions, 5 deletions
diff --git a/include/erebos/identity.h b/include/erebos/identity.h index 9ed170a..dce6093 100644 --- a/include/erebos/identity.h +++ b/include/erebos/identity.h @@ -12,6 +12,7 @@ public: std::optional<std::string> name() const; std::optional<Identity> owner() const; + const Identity & finalOwner() const; Stored<class PublicKey> keyMessage() const; diff --git a/include/erebos/network.h b/include/erebos/network.h index 47c7e7a..90c85a6 100644 --- a/include/erebos/network.h +++ b/include/erebos/network.h @@ -2,6 +2,8 @@ #include <erebos/identity.h> +#include <functional> + namespace erebos { class Server @@ -10,10 +12,44 @@ public: Server(const Identity &); ~Server(); + class PeerList & peerList() const; + struct Peer; private: struct Priv; const std::shared_ptr<Priv> p; }; +class Peer +{ +public: + struct Priv; + Peer(const std::shared_ptr<Priv> & p); + ~Peer(); + + std::string name() const; + std::optional<Identity> identity() const; + +private: + std::shared_ptr<Priv> p; }; + +class PeerList +{ +public: + struct Priv; + PeerList(); + PeerList(const std::shared_ptr<Priv> & p); + ~PeerList(); + + size_t size() const; + Peer at(size_t n) const; + + void onUpdate(std::function<void(size_t, const Peer *)>); + +private: + friend Server; + const std::shared_ptr<Priv> p; +}; + +} diff --git a/src/identity.cpp b/src/identity.cpp index 00abf0b..61059ab 100644 --- a/src/identity.cpp +++ b/src/identity.cpp @@ -46,6 +46,13 @@ optional<Identity> Identity::owner() const return p->owner; } +const Identity & Identity::finalOwner() const +{ + if (p->owner) + return p->owner->finalOwner(); + return *this; +} + Stored<PublicKey> Identity::keyMessage() const { return p->keyMessage; diff --git a/src/network.cpp b/src/network.cpp index f778bf9..f3a3651 100644 --- a/src/network.cpp +++ b/src/network.cpp @@ -6,12 +6,14 @@ #include <cstring> #include <iostream> +#include <arpa/inet.h> #include <ifaddrs.h> #include <net/if.h> #include <unistd.h> using std::holds_alternative; using std::scoped_lock; +using std::to_string; using std::unique_lock; using namespace erebos; @@ -23,6 +25,87 @@ Server::Server(const Identity & self): Server::~Server() = default; +PeerList & Server::peerList() const +{ + return p->plist; +} + + +Peer::Peer(const shared_ptr<Priv> & p): p(p) {} +Peer::~Peer() = default; + +string Peer::name() const +{ + if (auto speer = p->speer.lock()) { + if (holds_alternative<Identity>(speer->identity)) + if (auto name = std::get<Identity>(speer->identity).finalOwner().name()) + return *name; + if (holds_alternative<shared_ptr<WaitingRef>>(speer->identity)) + return string(std::get<shared_ptr<WaitingRef>>(speer->identity)->ref.digest()); + + char buf[16]; + if (inet_ntop(AF_INET, &speer->addr.sin_addr, buf, sizeof(buf))) + return string(buf) + ":" + to_string(ntohs(speer->addr.sin_port)); + return "<invalid address>"; + } + return "<server closed>"; +} + +optional<Identity> Peer::identity() const +{ + if (auto speer = p->speer.lock()) + if (holds_alternative<Identity>(speer->identity)) + return std::get<Identity>(speer->identity); + return nullopt; +} + +void Peer::Priv::notifyWatchers() +{ + if (auto slist = list.lock()) { + Peer p(shared_from_this()); + for (const auto & w : slist->watchers) + w(listIndex, &p); + } +} + + +PeerList::PeerList(): p(new Priv) {} +PeerList::PeerList(const shared_ptr<PeerList::Priv> & p): p(p) {} +PeerList::~PeerList() = default; + +void PeerList::Priv::push(const shared_ptr<Server::Peer> & speer) +{ + scoped_lock lock(dataMutex); + size_t s = peers.size(); + + speer->lpeer.reset(new Peer::Priv); + speer->lpeer->speer = speer; + speer->lpeer->list = shared_from_this(); + speer->lpeer->listIndex = s; + + Peer p(speer->lpeer); + + peers.push_back(speer->lpeer); + for (const auto & w : watchers) + w(s, &p); +} + +size_t PeerList::size() const +{ + return p->peers.size(); +} + +Peer PeerList::at(size_t i) const +{ + return Peer(p->peers.at(i)); +} + +void PeerList::onUpdate(function<void(size_t, const Peer *)> w) +{ + p->watchers.push_back(w); +} + + Server::Priv::Priv(const Identity & self): self(self) { @@ -174,15 +257,16 @@ Server::Peer & Server::Priv::getPeer(const sockaddr_in & paddr) return *peer; auto st = self.ref()->storage().deriveEphemeralStorage(); - Peer * peer = new Peer { + shared_ptr<Peer> peer(new Peer { .server = *this, .addr = paddr, .identity = monostate(), .channel = monostate(), .tempStorage = st, .partStorage = st.derivePartialStorage(), - }; - peers.emplace_back(peer); + }); + peers.push_back(peer); + plist.p->push(peer); return *peer; } @@ -328,8 +412,11 @@ void Server::Peer::updateIdentity(ReplyBuilder & reply) { if (holds_alternative<shared_ptr<WaitingRef>>(identity)) if (auto ref = std::get<shared_ptr<WaitingRef>>(identity)->check(&reply.header)) - if (auto id = Identity::load(*ref)) + if (auto id = Identity::load(*ref)) { identity.emplace<Identity>(*id); + if (lpeer) + lpeer->notifyWatchers(); + } } void Server::Peer::updateChannel(ReplyBuilder & reply) diff --git a/src/network.h b/src/network.h index e07e020..07b5363 100644 --- a/src/network.h +++ b/src/network.h @@ -23,6 +23,8 @@ using std::variant; using std::vector; using std::weak_ptr; +using std::enable_shared_from_this; + namespace chrono = std::chrono; using chrono::steady_clock; @@ -49,11 +51,31 @@ struct Server::Peer Storage tempStorage; PartialStorage partStorage; + shared_ptr<erebos::Peer::Priv> lpeer = nullptr; + void send(const struct TransportHeader &, const vector<Object> &) const; void updateIdentity(struct ReplyBuilder &); void updateChannel(struct ReplyBuilder &); }; +struct Peer::Priv : enable_shared_from_this<Peer::Priv> +{ + weak_ptr<Server::Peer> speer; + weak_ptr<PeerList::Priv> list; + size_t listIndex; + + void notifyWatchers(); +}; + +struct PeerList::Priv : enable_shared_from_this<PeerList::Priv> +{ + mutex dataMutex; + vector<shared_ptr<Peer::Priv>> peers; + vector<function<void(size_t, const Peer *)>> watchers; + + void push(const shared_ptr<Server::Peer> &); +}; + struct TransportHeader { enum class Type { @@ -118,7 +140,9 @@ struct Server::Priv thread threadListen; thread threadAnnounce; - vector<unique_ptr<Peer>> peers; + vector<shared_ptr<Peer>> peers; + PeerList plist; + vector<struct TransportHeader> outgoing; vector<weak_ptr<WaitingRef>> waiting; |