test

#include <sys/socket.h>

#include <sys/stat.h>

#include <sys/un.h>

#include <time.h>

#include <unistd.h>

#include "sysbus/client.h"

namespace syslite {

namespace {

long nowMs() {

    timespec ts{};

    clock_gettime(CLOCK_MONOTONIC, &ts);

    return static_cast<long>(ts.tv_sec) * 1000 + ts.tv_nsec / 1000000;

}

bool validServiceName(const std::string &n) {

    if (n.empty() || n.size() > 64 || n == "syslite") return false;

    for (char c : n)

        if (!((c >= 'a' && c <= 'z') || (c >= '0' && c <= '9') || c == '.' || c == '-' || c == '_')) return false;

    return true;

}

} // namespace

BusServer::~BusServer() { close(); }

bool BusServer::open(const std::string &path, Handler handler) {

void BusServer::close() {

    for (auto &c : _clients) ::close(c.first);

    _clients.clear();

    _services.clear();

    _pending.clear();

    if (_listenFd >= 0) {

        ::close(_listenFd);

        ::unlink(_path.c_str());

    for (auto &c : _clients) fds.push_back({c.first, POLLIN, 0});

    int r = ::poll(fds.data(), fds.size(), timeoutMs);

    expirePending();

    if (r <= 0) return;  // timeout, or EINTR from SIGCHLD & co.

    if (fds[0].revents & POLLIN) accept();

    sysbus::Message msg;

    while (c.decoder.next(msg)) {

        if (msg.type != sysbus::MessageType::MethodCall) continue;  // clients only call

        sysbus::Message reply = _handler(msg, c.peer);

        if (!sysbus::sendMessage(fd, reply)) {

            drop(fd);

            return;

        if (msg.type == sysbus::MessageType::MethodReturn || msg.type == sysbus::MessageType::Error) {

            answer(msg, fd);  // a service replying to a forwarded call

            if (!_clients.count(fd)) return;

            continue;

        }

        if (msg.type == sysbus::MessageType::Signal) {

            if (!c.peer.service.empty()) {

                msg.add("sysbus.sender", c.peer.service);

                broadcast(msg);

                if (!_clients.count(fd)) return;

            }

            continue;

        }

        std::string dest = msg.get("sysbus.destination");

        if (!dest.empty() && dest != "syslite") {

            route(msg, fd, c.peer, dest);

            if (!_clients.count(fd)) return;

            continue;

        }

        bool handled = false;

        sysbus::Message reply = busMethod(msg, fd, c.peer, handled);

        if (!handled) reply = _handler(msg, c.peer);

        if (!send(fd, reply)) return;

    }

    if (c.decoder.failed()) {

        std::cerr << "[sysbus] malformed message from pid " << c.peer.pid << ", disconnecting" << std::endl;

    }

}

bool BusServer::send(int fd, const sysbus::Message &msg) {

    if (sysbus::sendMessage(fd, msg)) return true;

    drop(fd);

    return false;

}

// The bus's own methods: RegisterService, ListServices.

sysbus::Message BusServer::busMethod(const sysbus::Message &call, int fd, BusPeer &peer, bool &handled) {

    namespace m = sysbus::method;

    namespace err = sysbus::error;

    handled = call.member == m::RegisterService || call.member == m::ListServices;

    if (call.member == m::ListServices) {

        sysbus::Message r = sysbus::Message::reply(call);

        for (auto &s : _services) {

            r.add("service", s.first);

            r.add("service_pid", static_cast<long long>(_clients[s.second].peer.pid));

        }

        return r;

    }

    if (call.member != m::RegisterService) return {};

    std::string name = call.get("name");

    if (peer.uid != 0) return sysbus::Message::error(call, err::AccessDenied, "RegisterService needs root");

    if (!validServiceName(name))

        return sysbus::Message::error(call, err::InvalidArgs, "invalid service name '" + name + "'");

    auto it = _services.find(name);

    if (it != _services.end() && it->second != fd)

        return sysbus::Message::error(call, err::Failed, "service " + name + " is already registered");

    if (!peer.service.empty() && peer.service != name)

        return sysbus::Message::error(call, err::Failed, "this connection is already service " + peer.service);

    _services[name] = fd;

    peer.service = name;

    std::cout << "[sysbus] service " << name << " registered (pid " << peer.pid << ")" << std::endl;

    return sysbus::Message::reply(call);

}

void BusServer::route(const sysbus::Message &call, int fd, const BusPeer &peer, const std::string &dest) {

    auto it = _services.find(dest);

    if (it == _services.end()) {

        send(fd, sysbus::Message::error(call, sysbus::error::ServiceUnknown, "no service " + dest + " on sysbus"));

        return;

    }

    sysbus::Message fwd(sysbus::MessageType::MethodCall, call.member);

    if (++_forwardSerial == 0) ++_forwardSerial;

    fwd.serial = _forwardSerial;

    fwd.add("sysbus.sender.uid", static_cast<long long>(peer.uid));

    fwd.add("sysbus.sender.pid", static_cast<long long>(peer.pid));

    for (auto &f : call.fields)

        if (f.first.compare(0, 7, "sysbus.") != 0) fwd.fields.push_back(f);

    int serviceFd = it->second;

    _pending[fwd.serial] = Pending{fd, call.serial, serviceFd, nowMs()};

    send(serviceFd, fwd);  // on failure drop() answers the caller

}

void BusServer::answer(const sysbus::Message &reply, int fd) {

    auto it = _pending.find(reply.replySerial);

    if (it == _pending.end() || it->second.serviceFd != fd) return;  // late, or not its call

    Pending p = it->second;

    _pending.erase(it);

    if (!_clients.count(p.callerFd)) return;

    sysbus::Message back = reply;

    back.replySerial = p.callerSerial;

    send(p.callerFd, back);

}

void BusServer::expirePending() {

    long now = nowMs();

    std::vector<uint32_t> late;

    for (auto &e : _pending)

        if (now - e.second.sentMs > ServiceTimeoutMs) late.push_back(e.first);

    for (uint32_t serial : late) {

        Pending p = _pending[serial];

        _pending.erase(serial);

        if (!_clients.count(p.callerFd)) continue;

        sysbus::Message call;

        call.serial = p.callerSerial;

        send(p.callerFd, sysbus::Message::error(call, sysbus::error::Timeout, "the service didn't reply in time"));

    }

}

void BusServer::drop(int fd) {

    ::close(fd);

    auto it = _clients.find(fd);

    std::string service = it != _clients.end() ? it->second.peer.service : "";

    _clients.erase(fd);

    if (!service.empty()) {

        _services.erase(service);

        std::cout << "[sysbus] service " << service << " gone" << std::endl;

    }

    // calls waiting on it fail; replies to calls it made are dropped

    std::vector<std::pair<int, uint32_t>> failed;

    for (auto e = _pending.begin(); e != _pending.end();) {

        if (e->second.serviceFd == fd) failed.emplace_back(e->second.callerFd, e->second.callerSerial);

        if (e->second.serviceFd == fd || e->second.callerFd == fd) e = _pending.erase(e);

        else ++e;

    }

    for (auto &f : failed) {

        if (!_clients.count(f.first)) continue;

        sysbus::Message call;

        call.serial = f.second;

        send(f.first, sysbus::Message::error(call, sysbus::error::ServiceGone, "service " + service + " went away"));

    }

}

void BusServer::broadcast(sysbus::Message signal) {

    signal.type = sysbus::MessageType::Signal;

// SYSBUS_SOCKET, hands each complete method call to a handler and sends its

// reply back, and broadcasts signals to subscribed clients. Single-threaded:

// poll() is called from syslite's main loop.

//

// It also routes for other daemons (see include/sysbus/service.h): a root

// client registers a name; a call carrying "sysbus.destination" with that

// name goes to it under a serial of the bus's own, with the caller's

// credentials added as "sysbus.sender.uid"/"sysbus.sender.pid" (any such

// fields from the caller are dropped first), and its reply goes back to

// the caller under the caller's serial. Signals a service sends reach the

// subscribed clients with "sysbus.sender" set to its name.

#include "sysbus/message.h"

    pid_t pid = -1;

    uid_t uid = static_cast<uid_t>(-1);

    bool subscribed = false;

    std::string service;  // the name this client registered, if any

};

class BusServer {