/**
 * (C) 2016 - 2021 KISTLER INSTRUMENTE AG, Winterthur, Switzerland
 * (C) 2016 - 2024 Stanislav Angelovic <stanislav.angelovic@protonmail.com>
 *
 * @file Connection.cpp
 *
 * Created on: Nov 8, 2016
 * Project: sdbus-c++
 * Description: High-level D-Bus IPC C++ library based on sd-bus
 *
 * This file is part of sdbus-c++.
 *
 * sdbus-c++ is free software; you can redistribute it and/or modify it
 * under the terms of the GNU Lesser General Public License as published by
 * the Free Software Foundation, either version 2.1 of the License, or
 * (at your option) any later version.
 *
 * sdbus-c++ is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with sdbus-c++. If not, see <http://www.gnu.org/licenses/>.
 */

#include "Connection.h"

#include "sdbus-c++/Error.h"
#include "sdbus-c++/Message.h"
#include "sdbus-c++/Types.h"

#include "MessageUtils.h"
#include "ScopeGuard.h"
#include "SdBus.h"
#include "Utils.h"

#include <poll.h>
#include <sys/eventfd.h>
#include SDBUS_HEADER
#ifndef SDBUS_basu // sd_event integration is not supported in basu-based sdbus-c++
#include <systemd/sd-event.h>
#endif
#include <unistd.h>

namespace sdbus::internal {

Connection::Connection(std::unique_ptr<ISdBus>&& interface, const BusFactory& busFactory)
    : sdbus_(std::move(interface))
    , bus_(openBus(busFactory))
{
    assert(sdbus_ != nullptr);
}

Connection::Connection(std::unique_ptr<ISdBus>&& interface, default_bus_t)
    : Connection(std::move(interface), [this](sd_bus** bus){ return sdbus_->sd_bus_open(bus); })
{
}

Connection::Connection(std::unique_ptr<ISdBus>&& interface, system_bus_t)
    : Connection(std::move(interface), [this](sd_bus** bus){ return sdbus_->sd_bus_open_system(bus); })
{
}

Connection::Connection(std::unique_ptr<ISdBus>&& interface, session_bus_t)
    : Connection(std::move(interface), [this](sd_bus** bus){ return sdbus_->sd_bus_open_user(bus); })
{
}

Connection::Connection(std::unique_ptr<ISdBus>&& interface, custom_session_bus_t, const std::string& address)
        : Connection(std::move(interface), [&](sd_bus** bus) { return sdbus_->sd_bus_open_user_with_address(bus, address.c_str()); })
{
}

Connection::Connection(std::unique_ptr<ISdBus>&& interface, remote_system_bus_t, const std::string& host)
    : Connection(std::move(interface), [this, &host](sd_bus** bus){ return sdbus_->sd_bus_open_system_remote(bus, host.c_str()); })
{
}

Connection::Connection(std::unique_ptr<ISdBus>&& interface, private_bus_t, const std::string& address)
    : Connection(std::move(interface), [&](sd_bus** bus) { return sdbus_->sd_bus_open_direct(bus, address.c_str()); })
{
}

Connection::Connection(std::unique_ptr<ISdBus>&& interface, private_bus_t, int fd)
        : Connection(std::move(interface), [&](sd_bus** bus) { return sdbus_->sd_bus_open_direct(bus, fd); })
{
}

Connection::Connection(std::unique_ptr<ISdBus>&& interface, server_bus_t, int fd)
    : Connection(std::move(interface), [&](sd_bus** bus) { return sdbus_->sd_bus_open_server(bus, fd); })
{
}

Connection::Connection(std::unique_ptr<ISdBus>&& interface, sdbus_bus_t, sd_bus *bus)
        : Connection(std::move(interface), [&](sd_bus** b) { *b = bus; return 0; })
{
}

Connection::Connection(std::unique_ptr<ISdBus>&& interface, pseudo_bus_t)
    : sdbus_(std::move(interface))
    , bus_(openPseudoBus())
{
    assert(sdbus_ != nullptr);
}

Connection::~Connection()
try
{
    Connection::leaveEventLoop();
}
catch (...)
{
    // Theoretically, we can fail to notify the event fd or join with the joinable thread...
    // What to do now here in the destructor? That is the question:
    //   1. Report the problem... but how, where?
    //   2. Terminate immediately... too harsh?
    //   3. Ignore and go on... even when some resources may be lingering?
    // Since the failure here is expected to be very unlikely, we choose the defensive approach of (3).
}

void Connection::requestName(const ServiceName& name)
{
    SDBUS_CHECK_SERVICE_NAME(name.c_str());

    auto r = sdbus_->sd_bus_request_name(bus_.get(), name.c_str(), 0);
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to request bus name", -r);

    // In some cases we need to explicitly notify the event loop
    // to process messages that may have arrived while executing the call
    wakeUpEventLoopIfMessagesInQueue();
}

void Connection::releaseName(const ServiceName& name)
{
    auto r = sdbus_->sd_bus_release_name(bus_.get(), name.c_str());
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to release bus name", -r);

    // In some cases we need to explicitly notify the event loop
    // to process messages that may have arrived while executing the call
    wakeUpEventLoopIfMessagesInQueue();
}

BusName Connection::getUniqueName() const
{
    const char* name{};
    auto r = sdbus_->sd_bus_get_unique_name(bus_.get(), &name);
    SDBUS_THROW_ERROR_IF(r < 0 || name == nullptr, "Failed to get unique bus name", -r);
    return BusName{name};
}

void Connection::enterEventLoop()
{
    while (true)
    {
        // Process one pending event
        (void)processPendingEvent();

        // And go to poll(), which wakes us up right away
        // if there's another pending event, or sleeps otherwise.
        auto success = waitForNextEvent();
        if (!success)
            break; // Exit I/O event loop
    }
}

void Connection::enterEventLoopAsync()
{
    if (!asyncLoopThread_.joinable())
        asyncLoopThread_ = std::thread([this](){ enterEventLoop(); });
}

void Connection::leaveEventLoop()
{
    notifyEventLoopToExit();
    joinWithEventLoop();
}

Connection::PollData Connection::getEventLoopPollData() const
{
    ISdBus::PollData pollData{};
    auto r = sdbus_->sd_bus_get_poll_data(bus_.get(), &pollData);
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to get bus poll data", -r);

    assert(eventFd_.fd >= 0);

    auto timeout = pollData.timeout_usec == UINT64_MAX ? std::chrono::microseconds::max() : std::chrono::microseconds(pollData.timeout_usec);

    return {pollData.fd, pollData.events, timeout, eventFd_.fd};
}

void Connection::addObjectManager(const ObjectPath& objectPath)
{
    auto r = sdbus_->sd_bus_add_object_manager(bus_.get(), nullptr, objectPath.c_str());

    SDBUS_THROW_ERROR_IF(r < 0, "Failed to add object manager", -r);
}

Slot Connection::addObjectManager(const ObjectPath& objectPath, return_slot_t)
{
    sd_bus_slot *slot{};

    auto r = sdbus_->sd_bus_add_object_manager(bus_.get(), &slot, objectPath.c_str());

    SDBUS_THROW_ERROR_IF(r < 0, "Failed to add object manager", -r);

    return {slot, [this](void *slot){ sdbus_->sd_bus_slot_unref(static_cast<sd_bus_slot*>(slot)); }};
}

void Connection::setMethodCallTimeout(uint64_t timeout)
{
    auto r = sdbus_->sd_bus_set_method_call_timeout(bus_.get(), timeout);

    SDBUS_THROW_ERROR_IF(r < 0, "Failed to set method call timeout", -r);
}

uint64_t Connection::getMethodCallTimeout() const
{
    uint64_t timeout{};

    auto r = sdbus_->sd_bus_get_method_call_timeout(bus_.get(), &timeout);

    SDBUS_THROW_ERROR_IF(r < 0, "Failed to get method call timeout", -r);

    return timeout;
}

void Connection::addMatch(const std::string& match, message_handler callback)
{
    floatingMatchRules_.push_back(addMatch(match, std::move(callback), return_slot));
}

Slot Connection::addMatch(const std::string& match, message_handler callback, return_slot_t)
{
    SDBUS_THROW_ERROR_IF(!callback, "Invalid match callback handler provided", EINVAL);

    auto matchInfo = std::make_unique<MatchInfo>(MatchInfo{std::move(callback), {}, *this, {}});

    sd_bus_slot *slot{};
    auto r = sdbus_->sd_bus_add_match(bus_.get(), &slot, match.c_str(), &Connection::sdbus_match_callback, matchInfo.get());
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to add match", -r);

    matchInfo->slot = {slot, [this](void *slot){ sdbus_->sd_bus_slot_unref(static_cast<sd_bus_slot*>(slot)); }};

    return {matchInfo.release(), [](void *ptr){ delete static_cast<MatchInfo*>(ptr); }}; // NOLINT(cppcoreguidelines-owning-memory)
}

void Connection::addMatchAsync(const std::string& match, message_handler callback, message_handler installCallback)
{
    floatingMatchRules_.push_back(addMatchAsync(match, std::move(callback), std::move(installCallback), return_slot));
}

Slot Connection::addMatchAsync( const std::string& match
                              , message_handler callback
                              , message_handler installCallback
                              , return_slot_t )
{
    SDBUS_THROW_ERROR_IF(!callback, "Invalid match callback handler provided", EINVAL);

    sd_bus_message_handler_t sdbusInstallCallback = installCallback ? &Connection::sdbus_match_install_callback : nullptr;
    auto matchInfo = std::make_unique<MatchInfo>(MatchInfo{std::move(callback), std::move(installCallback), *this, {}});

    sd_bus_slot *slot{};
    auto r = sdbus_->sd_bus_add_match_async( bus_.get()
                                           , &slot
                                           , match.c_str()
                                           , &Connection::sdbus_match_callback
                                           , sdbusInstallCallback
                                           , matchInfo.get());
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to add match", -r);

    matchInfo->slot = {slot, [this](void *slot){ sdbus_->sd_bus_slot_unref(static_cast<sd_bus_slot*>(slot)); }};

    return {matchInfo.release(), [](void *ptr){ delete static_cast<MatchInfo*>(ptr); }}; // NOLINT(cppcoreguidelines-owning-memory)
}

void Connection::attachSdEventLoop(sd_event *event, int priority)
{
#ifndef SDBUS_basu
    auto pollData = getEventLoopPollData();

    auto sdEvent = createSdEventSlot(event);
    auto sdTimeEventSource = createSdTimeEventSourceSlot(event, priority);
    auto sdIoEventSource = createSdIoEventSourceSlot(event, pollData.fd, priority);
    auto sdInternalEventSource = createSdInternalEventSourceSlot(event, pollData.eventFd, priority);

    sdEvent_ = std::make_unique<SdEvent>(SdEvent{ std::move(sdEvent)
                                                , std::move(sdTimeEventSource)
                                                , std::move(sdIoEventSource)
                                                , std::move(sdInternalEventSource) });
#else
    (void)event;
    (void)priority;
    SDBUS_THROW_ERROR("sd_event integration is not supported on this platform", EOPNOTSUPP);
#endif
}

void Connection::detachSdEventLoop()
{
    sdEvent_.reset();
}

sd_event *Connection::getSdEventLoop()
{
    return sdEvent_ ? static_cast<sd_event*>(sdEvent_->sdEvent.get()) : nullptr;
}

#ifndef SDBUS_basu

Slot Connection::createSdEventSlot(sd_event *event)
{
    // Get default event if no event is provided by the caller
    if (event != nullptr)
        event = sd_event_ref(event);
    else
        (void)sd_event_default(&event);
    SDBUS_THROW_ERROR_IF(!event, "Invalid sd_event handle", EINVAL);

    return Slot{event, [](void* event){ sd_event_unref(static_cast<sd_event*>(event)); }};
}

Slot Connection::createSdTimeEventSourceSlot(sd_event *event, int priority)
{
    sd_event_source *timeEventSource{};
    auto r = sd_event_add_time(event, &timeEventSource, CLOCK_MONOTONIC, 0, 0, onSdTimerEvent, this);
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to add timer event", -r);
    Slot sdTimeEventSource{timeEventSource, [](void* source){ deleteSdEventSource(static_cast<sd_event_source*>(source)); }};

    r = sd_event_source_set_priority(timeEventSource, priority);
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to set time event priority", -r);

    r = sd_event_source_set_description(timeEventSource, "bus-time");
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to set time event description", -r);

    return sdTimeEventSource;
}

Slot Connection::createSdIoEventSourceSlot(sd_event *event, int fd, int priority) // NOLINT(bugprone-easily-swappable-parameters)
{
    sd_event_source *ioEventSource{};
    auto r = sd_event_add_io(event, &ioEventSource, fd, 0, onSdIoEvent, this);
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to add io event", -r);
    Slot sdIoEventSource{ioEventSource, [](void* source){ deleteSdEventSource(static_cast<sd_event_source*>(source)); }};

    r = sd_event_source_set_prepare(ioEventSource, onSdEventPrepare);
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to set prepare callback for IO event", -r);

    r = sd_event_source_set_priority(ioEventSource, priority);
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to set priority for IO event", -r);

    r = sd_event_source_set_description(ioEventSource, "bus-input");
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to set priority for IO event", -r);

    return sdIoEventSource;
}

Slot Connection::createSdInternalEventSourceSlot(sd_event *event, int fd, int priority) // NOLINT(bugprone-easily-swappable-parameters)
{
    sd_event_source *internalEventSource{};
    auto r = sd_event_add_io(event, &internalEventSource, fd, 0, onSdInternalEvent, this);
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to add internal event", -r);
    Slot sdInternalEventSource{internalEventSource, [](void* source){ deleteSdEventSource(static_cast<sd_event_source*>(source)); }};

    // sd-event loop calls prepare callbacks for all event sources, not just for the one that fired now.
    // So since onSdEventPrepare is already registered on ioEventSource, we don't need to duplicate it here.
    //r = sd_event_source_set_prepare(internalEventSource, onSdEventPrepare);
    //SDBUS_THROW_ERROR_IF(r < 0, "Failed to set prepare callback for internal event", -r);

    r = sd_event_source_set_priority(internalEventSource, priority);
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to set priority for internal event", -r);

    r = sd_event_source_set_description(internalEventSource, "internal-event");
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to set priority for IO event", -r);

    return sdInternalEventSource;
}

int Connection::onSdTimerEvent(sd_event_source */*s*/, uint64_t /*usec*/, void *userdata)
{
    auto *connection = static_cast<Connection*>(userdata);
    assert(connection != nullptr);

    (void)connection->processPendingEvent();

    return 1;
}

int Connection::onSdIoEvent(sd_event_source */*s*/, int /*fd*/, uint32_t /*revents*/, void *userdata)
{
    auto *connection = static_cast<Connection*>(userdata);
    assert(connection != nullptr);

    (void)connection->processPendingEvent();

    return 1;
}

int Connection::onSdInternalEvent(sd_event_source */*s*/, int /*fd*/, uint32_t /*revents*/, void *userdata)
{
    auto *connection = static_cast<Connection*>(userdata);
    assert(connection != nullptr);

    // It's not really necessary to processPendingEvent() here. We just clear the event fd.
    // The sd-event loop will before the next poll call prepare callbacks for all event sources,
    // including I/O bus fd. This will get up-to-date poll timeout, which will be zero if there
    // are pending D-Bus messages in the read queue, which will immediately wake up next poll
    // and go to onSdIoEvent() handler, which calls processPendingEvent(). Viola.
    // For external event loops that only have access to public sdbus-c++ API, processPendingEvent()
    // is the only option to clear event fd (it comes at a little extra cost but on the other hand
    // the solution is simpler for clients -- we don't provide an extra method for just clearing
    // the event fd. There is one method for both fd's -- and that's processPendingEvent().

    // Kept here so that potential readers know what to do in their custom external event loops.
    //(void)connection->processPendingEvent();

    connection->eventFd_.clear();

    return 1;
}

int Connection::onSdEventPrepare(sd_event_source */*s*/, void *userdata)
{
    auto *connection = static_cast<Connection*>(userdata);
    assert(connection != nullptr);

    auto sdbusPollData = connection->getEventLoopPollData();

    // Set poll events to watch out for on I/O fd
    auto* sdIoEventSource = static_cast<sd_event_source*>(connection->sdEvent_->sdIoEventSource.get());
    auto r = sd_event_source_set_io_events(sdIoEventSource, sdbusPollData.events);
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to set poll events for IO event source", -r);

    // Set poll events to watch out for on internal event fd
    auto* sdInternalEventSource = static_cast<sd_event_source*>(connection->sdEvent_->sdInternalEventSource.get());
    r = sd_event_source_set_io_events(sdInternalEventSource, POLLIN);
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to set poll events for internal event source", -r);

    // Set current timeout to the time event source (it may be zero if there are messages in the sd-bus queues to be processed)
    auto* sdTimeEventSource = static_cast<sd_event_source*>(connection->sdEvent_->sdTimeEventSource.get());
    r = sd_event_source_set_time(sdTimeEventSource, static_cast<uint64_t>(sdbusPollData.timeout.count()));
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to set timeout for time event source", -r);
    // In case the timeout is infinite, we disable the timer in the sd_event loop.
    // This prevents a syscall error, where `timerfd_settime` returns `EINVAL`,
    // because the value is too big. See #324 for details
    r = sd_event_source_set_enabled(sdTimeEventSource, sdbusPollData.timeout != std::chrono::microseconds::max() ? SD_EVENT_ONESHOT : SD_EVENT_OFF);
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to enable time event source", -r);

    return 1;
}

void Connection::deleteSdEventSource(sd_event_source *source)
{
#if LIBSYSTEMD_VERSION>=243
    sd_event_source_disable_unref(source);
#else
    sd_event_source_set_enabled(s, SD_EVENT_OFF);
    sd_event_source_unref(s);
#endif
}

#endif // SDBUS_basu

Slot Connection::addObjectVTable( const ObjectPath& objectPath
                                , const InterfaceName& interfaceName
                                , const sd_bus_vtable* vtable
                                , void* userData
                                , return_slot_t )
{
    sd_bus_slot *slot{};

    auto r = sdbus_->sd_bus_add_object_vtable( bus_.get()
                                             , &slot
                                             , objectPath.c_str()
                                             , interfaceName.c_str()
                                             , vtable
                                             , userData );

    SDBUS_THROW_ERROR_IF(r < 0, "Failed to register object vtable", -r);

    return {slot, [this](void *slot){ sdbus_->sd_bus_slot_unref(static_cast<sd_bus_slot*>(slot)); }};
}

PlainMessage Connection::createPlainMessage() const
{
    sd_bus_message* sdbusMsg{};

    auto r = sdbus_->sd_bus_message_new(bus_.get(), &sdbusMsg, _SD_BUS_MESSAGE_TYPE_INVALID);

    SDBUS_THROW_ERROR_IF(r < 0, "Failed to create a plain message", -r);

    // TODO: const_cast..? Finish the const correctness design
    // NOLINTNEXTLINE(cppcoreguidelines-pro-type-const-cast)
    return Message::Factory::create<PlainMessage>(sdbusMsg, const_cast<Connection*>(this), adopt_message);
}

MethodCall Connection::createMethodCall( const ServiceName& destination
                                       , const ObjectPath& objectPath
                                       , const InterfaceName& interfaceName
                                       , const MethodName& methodName ) const
{
    return Connection::createMethodCall(destination.c_str(), objectPath.c_str(), interfaceName.c_str(), methodName.c_str());
}

MethodCall Connection::createMethodCall( const char* destination
                                       , const char* objectPath
                                       , const char* interfaceName
                                       , const char* methodName ) const
{
    sd_bus_message *sdbusMsg{};

    auto r = sdbus_->sd_bus_message_new_method_call( bus_.get()
                                                   , &sdbusMsg
                                                   , *destination == '\0' ? nullptr : destination
                                                   , objectPath
                                                   , interfaceName
                                                   , methodName);

    SDBUS_THROW_ERROR_IF(r < 0, "Failed to create method call", -r);

    // TODO: const_cast..? Finish the const correctness design
    // NOLINTNEXTLINE(cppcoreguidelines-pro-type-const-cast)
    return Message::Factory::create<MethodCall>(sdbusMsg, const_cast<Connection*>(this), adopt_message);
}

Signal Connection::createSignal( const ObjectPath& objectPath
                               , const InterfaceName& interfaceName
                               , const SignalName& signalName ) const
{
    return Connection::createSignal(objectPath.c_str(), interfaceName.c_str(), signalName.c_str());
}

Signal Connection::createSignal( const char* objectPath
                               , const char* interfaceName
                               , const char* signalName ) const
{
    sd_bus_message *sdbusMsg{};

    auto r = sdbus_->sd_bus_message_new_signal(bus_.get(), &sdbusMsg, objectPath, interfaceName, signalName);

    SDBUS_THROW_ERROR_IF(r < 0, "Failed to create signal", -r);

    // TODO: const_cast..? Finish the const correctness design
    // NOLINTNEXTLINE(cppcoreguidelines-pro-type-const-cast)
    return Message::Factory::create<Signal>(sdbusMsg, const_cast<Connection*>(this), adopt_message);
}

void Connection::emitPropertiesChangedSignal( const ObjectPath& objectPath
                                            , const InterfaceName& interfaceName
                                            , const std::vector<PropertyName>& propNames )
{
    Connection::emitPropertiesChangedSignal(objectPath.c_str(), interfaceName.c_str(), propNames);
}

void Connection::emitPropertiesChangedSignal( const char* objectPath
                                            , const char* interfaceName
                                            , const std::vector<PropertyName>& propNames )
{
    auto names = to_strv(propNames);

    auto r = sdbus_->sd_bus_emit_properties_changed_strv( bus_.get()
                                                        , objectPath
                                                        , interfaceName
                                                        , propNames.empty() ? nullptr : names.data() );

    SDBUS_THROW_ERROR_IF(r < 0, "Failed to emit PropertiesChanged signal", -r);
}

void Connection::emitInterfacesAddedSignal(const ObjectPath& objectPath)
{
    auto r = sdbus_->sd_bus_emit_object_added(bus_.get(), objectPath.c_str());

    SDBUS_THROW_ERROR_IF(r < 0, "Failed to emit InterfacesAdded signal for all registered interfaces", -r);
}

void Connection::emitInterfacesAddedSignal( const ObjectPath& objectPath
                                          , const std::vector<InterfaceName>& interfaces )
{
    auto names = to_strv(interfaces);

    auto r = sdbus_->sd_bus_emit_interfaces_added_strv( bus_.get()
                                                      , objectPath.c_str()
                                                      , interfaces.empty() ? nullptr : names.data() );

    SDBUS_THROW_ERROR_IF(r < 0, "Failed to emit InterfacesAdded signal", -r);
}

void Connection::emitInterfacesRemovedSignal(const ObjectPath& objectPath)
{
    auto r = sdbus_->sd_bus_emit_object_removed(bus_.get(), objectPath.c_str());

    SDBUS_THROW_ERROR_IF(r < 0, "Failed to emit InterfacesRemoved signal for all registered interfaces", -r);
}

void Connection::emitInterfacesRemovedSignal( const ObjectPath& objectPath
                                            , const std::vector<InterfaceName>& interfaces )
{
    auto names = to_strv(interfaces);

    auto r = sdbus_->sd_bus_emit_interfaces_removed_strv( bus_.get()
                                                        , objectPath.c_str()
                                                        , interfaces.empty() ? nullptr : names.data() );

    SDBUS_THROW_ERROR_IF(r < 0, "Failed to emit InterfacesRemoved signal", -r);
}

Slot Connection::registerSignalHandler( const char* sender
                                      , const char* objectPath
                                      , const char* interfaceName
                                      , const char* signalName
                                      , sd_bus_message_handler_t callback
                                      , void* userData
                                      , return_slot_t )
{
    sd_bus_slot *slot{};

    auto r = sdbus_->sd_bus_match_signal( bus_.get()
                                        , &slot
                                        , *sender == '\0' ? nullptr : sender
                                        , *objectPath == '\0' ? nullptr : objectPath
                                        , *interfaceName == '\0' ? nullptr : interfaceName
                                        , *signalName == '\0' ? nullptr : signalName
                                        , callback
                                        , userData );

    SDBUS_THROW_ERROR_IF(r < 0, "Failed to register signal handler", -r);

    return {slot, [this](void *slot){ sdbus_->sd_bus_slot_unref(static_cast<sd_bus_slot*>(slot)); }};
}

sd_bus_message* Connection::incrementMessageRefCount(sd_bus_message* sdbusMsg)
{
    return sdbus_->sd_bus_message_ref(sdbusMsg);
}

sd_bus_message* Connection::decrementMessageRefCount(sd_bus_message* sdbusMsg)
{
    return sdbus_->sd_bus_message_unref(sdbusMsg);
}

int Connection::querySenderCredentials(sd_bus_message* sdbusMsg, uint64_t mask, sd_bus_creds **creds)
{
    return sdbus_->sd_bus_query_sender_creds(sdbusMsg, mask, creds);
}

sd_bus_creds* Connection::incrementCredsRefCount(sd_bus_creds* creds)
{
    return sdbus_->sd_bus_creds_ref(creds);
}

sd_bus_creds* Connection::decrementCredsRefCount(sd_bus_creds* creds)
{
    return sdbus_->sd_bus_creds_unref(creds);
}

sd_bus_message* Connection::callMethod(sd_bus_message* sdbusMsg, uint64_t timeout)
{
    sd_bus_error sdbusError = SD_BUS_ERROR_NULL;
    SCOPE_EXIT{ sd_bus_error_free(&sdbusError); };

    // This call will block the bus connection from serving other messages
    // until the reply arrives or the call times out.
    sd_bus_message* sdbusReply{};
    auto r = sdbus_->sd_bus_call(nullptr, sdbusMsg, timeout, &sdbusError, &sdbusReply);

    if (sd_bus_error_is_set(&sdbusError)) // NOLINT(readability-implicit-bool-conversion)
        throw Error(Error::Name{sdbusError.name}, sdbusError.message);

    SDBUS_THROW_ERROR_IF(r < 0, "Failed to call method", -r);

    // Wake up event loop to process messages that may have arrived in the meantime,
    // or to dispatch the outbound message that hasn't yet been fully sent out.
    wakeUpEventLoopIfMessagesInQueue();

    return sdbusReply;
}

Slot Connection::callMethodAsync(sd_bus_message* sdbusMsg, sd_bus_message_handler_t callback, void* userData, uint64_t timeout, return_slot_t)
{
    sd_bus_slot *slot{};

    // TODO: Think of ways of optimizing these three locking/unlocking of sdbus mutex (merge into one call?)
    auto timeoutBefore = getEventLoopPollData().timeout;
    auto r = sdbus_->sd_bus_call_async(nullptr, &slot, sdbusMsg, callback, userData, timeout);
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to call method asynchronously", -r);
    auto timeoutAfter = getEventLoopPollData().timeout;

    // An event loop may wait in poll with timeout `t1', while in another thread an async call is made with
    // timeout `t2'. If `t2' < `t1', then we have to wake up the event loop thread to update its poll timeout.
    // We also have to wake up the event loop to process the messages that may be in the read/write queues.
    if (timeoutAfter < timeoutBefore || arePendingMessagesInQueues())
        notifyEventLoopToWakeUpFromPoll();

    return {slot, [this](void *slot){ sdbus_->sd_bus_slot_unref(static_cast<sd_bus_slot*>(slot)); }};
}

void Connection::sendMessage(sd_bus_message* sdbusMsg)
{
    auto r = sdbus_->sd_bus_send(nullptr, sdbusMsg, nullptr);

    // Wake up event loop to continue dispatching the (fairly large) outbound message that hasn't yet been fully sent
    wakeUpEventLoopIfMessagesInQueue();

    SDBUS_THROW_ERROR_IF(r < 0, "Failed to send D-Bus message", -r);
}

sd_bus_message* Connection::createMethodReply(sd_bus_message* sdbusMsg)
{
    sd_bus_message* sdbusReply{};

    auto r = sdbus_->sd_bus_message_new_method_return(sdbusMsg, &sdbusReply);
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to create method reply", -r);

    return sdbusReply;
}

sd_bus_message* Connection::createErrorReplyMessage(sd_bus_message* sdbusMsg, const Error& error)
{
    sd_bus_error sdbusError = SD_BUS_ERROR_NULL;
    SCOPE_EXIT{ sd_bus_error_free(&sdbusError); };
    sd_bus_error_set(&sdbusError, error.getName().c_str(), error.getMessage().c_str());

    sd_bus_message* sdbusErrorReply{};
    auto r = sdbus_->sd_bus_message_new_method_error(sdbusMsg, &sdbusErrorReply, &sdbusError);
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to create method error reply", -r);

    return sdbusErrorReply;
}

Connection::BusPtr Connection::openBus(const BusFactory& busFactory)
{
    sd_bus* bus{};
    int r = busFactory(&bus);
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to open bus", -r);

    BusPtr busPtr{bus, [this](sd_bus* bus){ return sdbus_->sd_bus_flush_close_unref(bus); }};
    finishHandshake(busPtr.get());
    return busPtr;
}

Connection::BusPtr Connection::openPseudoBus()
{
    sd_bus* bus{};

    int r = sdbus_->sd_bus_new(&bus);
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to open pseudo bus", -r);

    (void)sdbus_->sd_bus_start(bus);
    // It is expected that sd_bus_start has failed here, returning -EINVAL, due to having
    // not set a bus address, but it will leave the bus in an OPENING state, which enables
    // us to create plain D-Bus messages as a local data storage (for Variant, for example),
    // without dependency on real IPC communication with the D-Bus broker daemon.
    SDBUS_THROW_ERROR_IF(r < 0 && r != -EINVAL, "Failed to start pseudo bus", -r);

    return {bus, [this](sd_bus* bus){ return sdbus_->sd_bus_close_unref(bus); }};
}

void Connection::finishHandshake(sd_bus* bus)
{
    // Process all requests that are part of the initial handshake,
    // like processing the Hello message response, authentication etc.,
    // to avoid connection authentication timeout in dbus daemon.

    assert(bus != nullptr);

    auto r = sdbus_->sd_bus_flush(bus);

    SDBUS_THROW_ERROR_IF(r < 0, "Failed to flush bus on opening", -r);
}

void Connection::notifyEventLoopToExit()
{
    loopExitFd_.notify();
}

void Connection::notifyEventLoopToWakeUpFromPoll()
{
    eventFd_.notify();
}

void Connection::wakeUpEventLoopIfMessagesInQueue()
{
    // We need this in two cases:
    // 1. When doing a sync call, other D-Bus messages may have arrived, waiting in the read queue.
    // In case an event loop is inside a poll in another thread, or an external event loop polls in the
    // same thread but as an unrelated event source, then we need to wake up the poll explicitly so the
    // event loop 1. processes all messages in the read queue, 2. updates poll timeout before next poll.
    // 2. Additionally, when sending out messages, these may be too long to be sent out entirely within
    // the single sd_bus_send() or sd_bus_call_async() call, in which case they are queued in the write
    // queue. We need to wake up the event loop to continue sending the message until it's fully sent.
    if (arePendingMessagesInQueues())
        notifyEventLoopToWakeUpFromPoll();
}

void Connection::joinWithEventLoop()
{
    if (asyncLoopThread_.joinable())
        asyncLoopThread_.join();
}

bool Connection::processPendingEvent()
{
    auto *bus = bus_.get();
    assert(bus != nullptr);

    int r = sdbus_->sd_bus_process(bus, nullptr);
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to process bus requests", -r);

    // In correct use of sdbus-c++ API, r can be 0 only when processPendingEvent()
    // is called from an external event loop as a reaction to event fd being signalled.
    // If there are no more D-Bus messages to process, we know we have to clear event fd.
    if (r == 0)
        eventFd_.clear();

    return r > 0;
}

bool Connection::waitForNextEvent() // NOLINT(misc-no-recursion)
{
    assert(bus_ != nullptr);
    assert(loopExitFd_.fd >= 0);
    assert(eventFd_.fd >= 0);

    auto sdbusPollData = getEventLoopPollData();
    struct pollfd fds[] = { {sdbusPollData.fd, sdbusPollData.events, 0}
                          , {eventFd_.fd, POLLIN, 0}
                          , {loopExitFd_.fd, POLLIN, 0} };
    constexpr auto fdsCount = sizeof(fds)/sizeof(fds[0]);

    // Are there pending messages in the inbound queue? Then sd-bus will set timeout to 0, so poll() will wake up right away.
    // Are there pending messages in the outbound queue? Then sd-bus will add POLLOUT to events, so poll() will wake up right away.
    auto timeout = sdbusPollData.getPollTimeout();
    auto r = poll(fds, fdsCount, timeout);

    if (r < 0 && errno == EINTR)
        return true; // Try again

    SDBUS_THROW_ERROR_IF(r < 0, "Failed to wait on the bus", -errno);

    // Wake up notification, in order that we re-enter poll with freshly read PollData (namely, new poll timeout thereof)
    if (fds[1].revents & POLLIN) // NOLINT(readability-implicit-bool-conversion)
    {
        auto cleared = eventFd_.clear();
        SDBUS_THROW_ERROR_IF(!cleared, "Failed to read from the event descriptor", -errno);
        // Go poll() again, but with freshly calculated, up-to-date timeout and with up-to-date events to watch
        return waitForNextEvent();
    }
    // Loop exit notification
    if (fds[2].revents & POLLIN) // NOLINT(readability-implicit-bool-conversion)
    {
        auto cleared = loopExitFd_.clear();
        SDBUS_THROW_ERROR_IF(!cleared, "Failed to read from the loop exit descriptor", -errno);
        return false;
    }

    return true;
}

bool Connection::arePendingMessagesInQueues() const
{
    uint64_t readQueueSize{};
    uint64_t writeQueueSize{};

    auto r = sdbus_->sd_bus_get_n_queued(bus_.get(), &readQueueSize, &writeQueueSize);
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to get number of pending messages in sd-bus queues", -r);

    return readQueueSize > 0 || writeQueueSize > 0;
}

Message Connection::getCurrentlyProcessedMessage() const
{
    auto* sdbusMsg = sdbus_->sd_bus_get_current_message(bus_.get());

    // TODO: const_cast..? Finish the const correctness design
    // NOLINTNEXTLINE(cppcoreguidelines-pro-type-const-cast)
    return Message::Factory::create<Message>(sdbusMsg, const_cast<Connection*>(this));
}

template <typename StringBasedType>
std::vector</*const */char*> Connection::to_strv(const std::vector<StringBasedType>& strings)
{
    std::vector</*const */char*> strv;
    strv.reserve(strings.size());
    for (auto& str : strings)
        strv.push_back(const_cast<char*>(str.c_str())); // NOLINT(cppcoreguidelines-pro-type-const-cast)
    strv.push_back(nullptr);
    return strv;
}

int Connection::sdbus_match_callback(sd_bus_message *sdbusMessage, void *userData, sd_bus_error *retError)
{
    auto* matchInfo = static_cast<MatchInfo*>(userData);
    assert(matchInfo != nullptr);
    assert(matchInfo->callback);

    auto message = Message::Factory::create<PlainMessage>(sdbusMessage, &matchInfo->connection);

    auto ok = invokeHandlerAndCatchErrors([&](){ matchInfo->callback(std::move(message)); }, retError);

    return ok ? 0 : -1;
}

int Connection::sdbus_match_install_callback(sd_bus_message *sdbusMessage, void *userData, sd_bus_error *retError)
{
    auto* matchInfo = static_cast<MatchInfo*>(userData);
    assert(matchInfo != nullptr);
    assert(matchInfo->installCallback);

    auto message = Message::Factory::create<PlainMessage>(sdbusMessage, &matchInfo->connection);

    auto ok = invokeHandlerAndCatchErrors([&](){ matchInfo->installCallback(std::move(message)); }, retError);

    return ok ? 0 : -1;
}

Connection::EventFd::EventFd()
    : fd(eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK))
{
    SDBUS_THROW_ERROR_IF(fd < 0, "Failed to create event object", -errno);
}

Connection::EventFd::~EventFd()
{
    assert(fd >= 0);
    close(fd);
}

void Connection::EventFd::notify() const
{
    assert(fd >= 0);
    auto r = eventfd_write(fd, 1);
    SDBUS_THROW_ERROR_IF(r < 0, "Failed to notify event descriptor", -errno);
}

bool Connection::EventFd::clear() const
{
    assert(fd >= 0);

    uint64_t value{};
    auto r = eventfd_read(fd, &value);
    return r >= 0;
}

} // namespace sdbus::internal

namespace sdbus {

std::chrono::microseconds IConnection::PollData::getRelativeTimeout() const
{
    constexpr auto zero = std::chrono::microseconds::zero();
    constexpr auto max = std::chrono::microseconds::max();
    using internal::now;

    if (timeout == zero)
        return zero;
    if (timeout == max)
        return max;

    return std::max(std::chrono::duration_cast<std::chrono::microseconds>(timeout - now()), zero);
}

int IConnection::PollData::getPollTimeout() const
{
    const auto relativeTimeout = getRelativeTimeout();

    if (relativeTimeout == decltype(relativeTimeout)::max())
        return -1;

    return static_cast<int>(std::chrono::ceil<std::chrono::milliseconds>(relativeTimeout).count());
}

} // namespace sdbus

namespace sdbus::internal {

std::unique_ptr<sdbus::internal::IConnection> createPseudoConnection()
{
    auto interface = std::make_unique<sdbus::internal::SdBus>();
    return std::make_unique<sdbus::internal::Connection>(std::move(interface), Connection::pseudo_bus);
}

} // namespace sdbus::internal

namespace sdbus {

using internal::Connection;

std::unique_ptr<sdbus::IConnection> createBusConnection()
{
    auto interface = std::make_unique<sdbus::internal::SdBus>();
    return std::make_unique<sdbus::internal::Connection>(std::move(interface), Connection::default_bus);
}

std::unique_ptr<sdbus::IConnection> createBusConnection(const ServiceName& name)
{
    auto conn = createBusConnection();
    conn->requestName(name);
    return conn;
}

std::unique_ptr<sdbus::IConnection> createSystemBusConnection()
{
    auto interface = std::make_unique<sdbus::internal::SdBus>();
    return std::make_unique<sdbus::internal::Connection>(std::move(interface), Connection::system_bus);
}

std::unique_ptr<sdbus::IConnection> createSystemBusConnection(const ServiceName& name)
{
    auto conn = createSystemBusConnection();
    conn->requestName(name);
    return conn;
}

std::unique_ptr<sdbus::IConnection> createSessionBusConnection()
{
    auto interface = std::make_unique<sdbus::internal::SdBus>();
    return std::make_unique<sdbus::internal::Connection>(std::move(interface), Connection::session_bus);
}

std::unique_ptr<sdbus::IConnection> createSessionBusConnection(const ServiceName& name)
{
    auto conn = createSessionBusConnection();
    conn->requestName(name);
    return conn;
}

std::unique_ptr<sdbus::IConnection> createSessionBusConnectionWithAddress(const std::string &address)
{
    auto interface = std::make_unique<sdbus::internal::SdBus>();
    return std::make_unique<sdbus::internal::Connection>(std::move(interface), Connection::custom_session_bus, address);
}

std::unique_ptr<sdbus::IConnection> createRemoteSystemBusConnection(const std::string& host)
{
    auto interface = std::make_unique<sdbus::internal::SdBus>();
    return std::make_unique<sdbus::internal::Connection>(std::move(interface), Connection::remote_system_bus, host);
}

std::unique_ptr<sdbus::IConnection> createDirectBusConnection(const std::string& address)
{
    auto interface = std::make_unique<sdbus::internal::SdBus>();
    return std::make_unique<sdbus::internal::Connection>(std::move(interface), Connection::private_bus, address);
}

std::unique_ptr<sdbus::IConnection> createDirectBusConnection(int fd)
{
    auto interface = std::make_unique<sdbus::internal::SdBus>();
    return std::make_unique<sdbus::internal::Connection>(std::move(interface), Connection::private_bus, fd);
}

std::unique_ptr<sdbus::IConnection> createServerBus(int fd)
{
    auto interface = std::make_unique<sdbus::internal::SdBus>();
    return std::make_unique<sdbus::internal::Connection>(std::move(interface), Connection::server_bus, fd);
}

std::unique_ptr<sdbus::IConnection> createBusConnection(sd_bus *bus)
{
    SDBUS_THROW_ERROR_IF(bus == nullptr, "Invalid bus argument", EINVAL);

    auto interface = std::make_unique<sdbus::internal::SdBus>();
    return std::make_unique<sdbus::internal::Connection>(std::move(interface), Connection::sdbus_bus, bus);
}

} // namespace sdbus