feat: add support for direct connections (#350)

* feat: add support for direct connections * refactor: simplify a bit, change comments, extend tests * fix: compiler warning about unused variable * docs: add section on direct connections to the tutorial --------- Co-authored-by: Maksim Fedyarov <m.fedyarov@omp.ru> Co-authored-by: Stanislav Angelovič <stanislav.angelovic@protonmail.com>
2023-09-25 21:12:34 +03:00
parent c437b4d508
commit cfb71bd6cf
13 changed files with 364 additions and 44 deletions
--- a/docs/using-sdbus-c++.md
+++ b/docs/using-sdbus-c++.md
@@ -21,7 +21,8 @@ Using sdbus-c++ library
 16. [Standard D-Bus interfaces](#standard-d-bus-interfaces)
 17. [Representing D-Bus Types in sdbus-c++](#representing-d-bus-types-in-sdbus-c)
 18. [Support for match rules](#support-for-match-rules)
-19. [Conclusion](#conclusion)
+19. [Using direct (peer-to-peer) D-Bus connections](#using-direct-peer-to-peer-d-bus-connections)
+20. [Conclusion](#conclusion)

 Introduction
 ------------
@@ -1612,6 +1613,62 @@ Support for match rules

 `IConnection` class provides `addMatch` method that you can use to install match rules. An associated callback handler will be called upon an incoming message matching given match rule. There is support for both client-owned and floating (library-owned) match rules. Consult `IConnection` header or sdbus-c++ doxygen documentation for more information.

+Using direct (peer-to-peer) D-Bus connections
+---------------------------------------------
+
+sdbus-c++ provides an API to establish a direct connection between two peers -- between a client and a server, without going via the D-Bus daemon. The methods of interest, which will create a D-Bus server bus, and a client connection to it, respectively, are:
+
+* `sdbus::createServerBus()` creates and returns a new, custom bus object in server mode, out of provided file descriptor parameter.
+* `sdbus::createDirectBusConnection()` opens and returns direct D-Bus connection at the provided custom address(es), or at the provided file descriptor.
+
+Here is an example, extracted from the analogous test case in sdbus-c++ integration tests suite:
+
+```c++
+#include "Concatenator.h"
+#include "ConcatenatorProxy.h"
+#include <sdbus-c++/sdbus-c++.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+
+int main(int argc, char *argv[])
+{
+    int fds[2];
+
+    socketpair(AF_UNIX, SOCK_STREAM, 0, fds);
+
+    std::unique_ptr<sdbus::IConnection> serverConnection;
+    std::unique_ptr<sdbus::IConnection> clientConnection;
+
+    std::thread t([&]()
+    {
+        serverConnection = sdbus::createServerBus(fds[0]);
+        // This is necessary so that createDirectBusConnection() below does not block
+        serverConnection->enterEventLoopAsync();
+    });
+
+    clientConnection = sdbus::createDirectBusConnection(fds[1]);
+    clientConnection->enterEventLoopAsync();
+
+    t.join();
+
+    // We can now use connection objects in a familiar way, e.g. create adaptor and proxy objects on them, and exchange messages.
+    // Here, using Concatenator IDL-generated bindings example from chapters above:
+    const char* objectPath = "/org/sdbuscpp/concatenator";
+    Concatenator concatenator(*serverConnection, objectPath);
+    const char* emptyDestinationName = ""; // Destination may be empty in case of direct connections
+    ConcatenatorProxy concatenatorProxy(*clientConnection, emptyDestinationName, objectPath);
+
+    // Perform call of concatenate D-Bus method
+    std::vector<int> numbers = {1, 2, 3};
+    std::string separator = ":";
+    auto concatenatedString = concatenatorProxy.concatenate(numbers, separator);
+    assert(concatenatedString == "1:2:3");
+
+    clientConnection->leaveEventLoop();
+    serverConnection->leaveEventLoop();
+}
+```
+
 Conclusion
 ----------

--- a/include/sdbus-c++/IConnection.h
+++ b/include/sdbus-c++/IConnection.h
@@ -458,6 +458,45 @@ namespace sdbus {
     * @throws sdbus::Error in case of failure
     */
    [[nodiscard]] std::unique_ptr<sdbus::IConnection> createRemoteSystemBusConnection(const std::string& host);
+
+    /*!
+     * @brief Opens direct D-Bus connection at a custom address
+     *
+     * @param[in] address ";"-separated list of addresses of bus brokers to try to connect to
+     * @return Connection instance
+     *
+     * @throws sdbus::Error in case of failure
+     */
+    [[nodiscard]] std::unique_ptr<sdbus::IConnection> createDirectBusConnection(const std::string& address);
+
+    /*!
+     * @brief Opens direct D-Bus connection at the given file descriptor
+     *
+     * @param[in] fd File descriptor used to communicate directly from/to a D-Bus server
+     * @return Connection instance
+     *
+     * The underlying sdbus-c++ connection instance takes over ownership of fd, so the caller can let it go.
+     * If, however, the call throws an exception, the ownership of fd remains with the caller.
+     *
+     * @throws sdbus::Error in case of failure
+     */
+    [[nodiscard]] std::unique_ptr<sdbus::IConnection> createDirectBusConnection(int fd);
+
+    /*!
+     * @brief Opens direct D-Bus connection at fd as a server
+     *
+     * @param[in] fd File descriptor to use for server DBus connection
+     * @return Server connection instance
+     *
+     * This creates a new, custom bus object in server mode. One can then call createDirectBusConnection()
+     * on client side to connect to this bus.
+     *
+     * The underlying sdbus-c++ connection instance takes over ownership of fd, so the caller can let it go.
+     * If, however, the call throws an exception, the ownership of fd remains with the caller.
+     *
+     * @throws sdbus::Error in case of failure
+     */
+    [[nodiscard]] std::unique_ptr<sdbus::IConnection> createServerBus(int fd);
 }

 #endif /* SDBUS_CXX_ICONNECTION_H_ */
--- a/include/sdbus-c++/IProxy.h
+++ b/include/sdbus-c++/IProxy.h
@@ -562,6 +562,9 @@ namespace sdbus {
     * should make sure that an I/O event loop is running on that connection, so the proxy
     * may receive incoming signals and asynchronous method replies.
     *
+     * The destination parameter may be an empty string (useful e.g. in case of direct
+     * D-Bus connections to a custom server bus).
+     *
     * Code example:
     * @code
     * auto proxy = sdbus::createProxy(connection, "com.kistler.foo", "/com/kistler/foo");
@@ -585,6 +588,9 @@ namespace sdbus {
     * upon that connection in a separate internal thread. Handlers for incoming signals and
     * asynchronous method replies will be executed in the context of that thread.
     *
+     * The destination parameter may be an empty string (useful e.g. in case of direct
+     * D-Bus connections to a custom server bus).
+     *
     * Code example:
     * @code
     * auto proxy = sdbus::createProxy(std::move(connection), "com.kistler.foo", "/com/kistler/foo");
@@ -603,10 +609,14 @@ namespace sdbus {
     * @return Pointer to the object proxy instance
     *
     * The provided connection will be used by the proxy to issue calls against the object.
-     * The Object proxy becomes an exclusive owner of this connection, but will not start an event loop
-     * thread on this connection. This is cheap construction and is suitable for short-lived proxies
-     * created just to execute simple synchronous D-Bus calls and then destroyed. Such blocking request-reply
-     * calls will work without an event loop (but signals, async calls, etc. won't).
+     * The Object proxy becomes an exclusive owner of this connection, but will not start
+     * an event loop thread on this connection. This is cheap construction and is suitable
+     * for short-lived proxies created just to execute simple synchronous D-Bus calls and
+     * then destroyed. Such blocking request-reply calls will work without an event loop
+     * (but signals, async calls, etc. won't).
+     *
+     * The destination parameter may be an empty string (useful e.g. in case of direct
+     * D-Bus connections to a custom server bus).
     *
     * Code example:
     * @code
--- a/src/Connection.cpp
+++ b/src/Connection.cpp
@@ -70,6 +70,21 @@ Connection::Connection(std::unique_ptr<ISdBus>&& interface, remote_system_bus_t,
 {
 }

+Connection::Connection(std::unique_ptr<ISdBus>&& interface, private_bus_t, const std::string& address)
+    : Connection(std::move(interface), [&](sd_bus** bus) { return iface_->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 iface_->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 iface_->sd_bus_open_server(bus, fd); })
+{
+}
+
 Connection::Connection(std::unique_ptr<ISdBus>&& interface, pseudo_bus_t)
    : iface_(std::move(interface))
    , bus_(openPseudoBus())
@@ -262,7 +277,7 @@ MethodCall Connection::createMethodCall( const std::string& destination

    auto r = iface_->sd_bus_message_new_method_call( bus_.get()
                                                   , &sdbusMsg
-                                                   , destination.c_str()
+                                                   , destination.empty() ? nullptr : destination.c_str()
                                                   , objectPath.c_str()
                                                   , interfaceName.c_str()
                                                   , methodName.c_str() );
@@ -527,7 +542,8 @@ std::string Connection::composeSignalMatchFilter( const std::string &sender
    std::string filter;

    filter += "type='signal',";
-    filter += "sender='" + sender + "',";
+    if (!sender.empty())
+        filter += "sender='" + sender + "',";
    filter += "interface='" + interfaceName + "',";
    filter += "member='" + signalName + "',";
    filter += "path='" + objectPath + "'";
@@ -661,7 +677,6 @@ std::unique_ptr<sdbus::IConnection> createSessionBusConnection(const std::string
 std::unique_ptr<sdbus::IConnection> createSessionBusConnectionWithAddress(const std::string &address)
 {
    auto interface = std::make_unique<sdbus::internal::SdBus>();
-    assert(interface != nullptr);
    return std::make_unique<sdbus::internal::Connection>(std::move(interface), Connection::custom_session_bus, address);
 }

@@ -671,4 +686,22 @@ std::unique_ptr<sdbus::IConnection> createRemoteSystemBusConnection(const std::s
    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);
+}
+
 } // namespace sdbus
--- a/src/Connection.h
+++ b/src/Connection.h
@@ -57,6 +57,10 @@ namespace sdbus::internal {
        inline static constexpr custom_session_bus_t custom_session_bus{};
        struct remote_system_bus_t{};
        inline static constexpr remote_system_bus_t remote_system_bus{};
+        struct private_bus_t{};
+        inline static constexpr private_bus_t private_bus{};
+        struct server_bus_t{};
+        inline static constexpr server_bus_t server_bus{};
        struct pseudo_bus_t{}; // A bus connection that is not really established with D-Bus daemon
        inline static constexpr pseudo_bus_t pseudo_bus{};

@@ -65,6 +69,9 @@ namespace sdbus::internal {
        Connection(std::unique_ptr<ISdBus>&& interface, session_bus_t);
        Connection(std::unique_ptr<ISdBus>&& interface, custom_session_bus_t, const std::string& address);
        Connection(std::unique_ptr<ISdBus>&& interface, remote_system_bus_t, const std::string& host);
+        Connection(std::unique_ptr<ISdBus>&& interface, private_bus_t, const std::string& address);
+        Connection(std::unique_ptr<ISdBus>&& interface, private_bus_t, int fd);
+        Connection(std::unique_ptr<ISdBus>&& interface, server_bus_t, int fd);
        Connection(std::unique_ptr<ISdBus>&& interface, pseudo_bus_t);
        ~Connection() override;

--- a/src/ISdBus.h
+++ b/src/ISdBus.h
@@ -71,6 +71,9 @@ namespace sdbus::internal {
        virtual int sd_bus_open_user(sd_bus **ret) = 0;
        virtual int sd_bus_open_user_with_address(sd_bus **ret, const char* address) = 0;
        virtual int sd_bus_open_system_remote(sd_bus **ret, const char* host) = 0;
+        virtual int sd_bus_open_direct(sd_bus **ret, const char* address) = 0;
+        virtual int sd_bus_open_direct(sd_bus **ret, int fd) = 0;
+        virtual int sd_bus_open_server(sd_bus **ret, int fd) = 0;
        virtual int sd_bus_request_name(sd_bus *bus, const char *name, uint64_t flags) = 0;
        virtual int sd_bus_release_name(sd_bus *bus, const char *name) = 0;
        virtual int sd_bus_get_unique_name(sd_bus *bus, const char **name) = 0;
--- a/src/SdBus.cpp
+++ b/src/SdBus.cpp
@@ -194,26 +194,98 @@ int SdBus::sd_bus_open_user_with_address(sd_bus **ret, const char* address)
 {
    sd_bus* bus = nullptr;

-    int r = sd_bus_new(&bus);
+    int r = ::sd_bus_new(&bus);
    if (r < 0)
        return r;

-    r = sd_bus_set_address(bus, address);
+    r = ::sd_bus_set_address(bus, address);
    if (r < 0)
        return r;

-    r = sd_bus_set_bus_client(bus, true);
+    r = ::sd_bus_set_bus_client(bus, true);
    if (r < 0)
        return r;

    // Copying behavior from
    // https://github.com/systemd/systemd/blob/fee6441601c979165ebcbb35472036439f8dad5f/src/libsystemd/sd-bus/sd-bus.c#L1381
    // Here, we make the bus as trusted
-    r = sd_bus_set_trusted(bus, true);
+    r = ::sd_bus_set_trusted(bus, true);
    if (r < 0)
        return r;

-    r = sd_bus_start(bus);
+    r = ::sd_bus_start(bus);
+    if (r < 0)
+        return r;
+
+    *ret = bus;
+
+    return 0;
+}
+
+int SdBus::sd_bus_open_direct(sd_bus **ret, const char* address)
+{
+    sd_bus* bus = nullptr;
+
+    int r = ::sd_bus_new(&bus);
+    if (r < 0)
+        return r;
+
+    r = ::sd_bus_set_address(bus, address);
+    if (r < 0)
+        return r;
+
+    r = ::sd_bus_start(bus);
+    if (r < 0)
+        return r;
+
+    *ret = bus;
+
+    return 0;
+}
+
+int SdBus::sd_bus_open_direct(sd_bus **ret, int fd)
+{
+    sd_bus* bus = nullptr;
+
+    int r = ::sd_bus_new(&bus);
+    if (r < 0)
+        return r;
+
+    r = ::sd_bus_set_fd(bus, fd, fd);
+    if (r < 0)
+        return r;
+
+    r = ::sd_bus_start(bus);
+    if (r < 0)
+        return r;
+
+    *ret = bus;
+
+    return 0;
+}
+
+int SdBus::sd_bus_open_server(sd_bus **ret, int fd)
+{
+    sd_bus* bus = nullptr;
+
+    int r = ::sd_bus_new(&bus);
+    if (r < 0)
+        return r;
+
+    r = ::sd_bus_set_fd(bus, fd, fd);
+    if (r < 0)
+        return r;
+
+    sd_id128_t id;
+    r = ::sd_id128_randomize(&id);
+    if (r < 0)
+        return r;
+
+    r = ::sd_bus_set_server(bus, true, id);
+    if (r < 0)
+        return r;
+
+    r = ::sd_bus_start(bus);
    if (r < 0)
        return r;

@@ -270,7 +342,7 @@ int SdBus::sd_bus_add_match(sd_bus *bus, sd_bus_slot **slot, const char *match,
 {
    std::lock_guard lock(sdbusMutex_);

-    return :: sd_bus_add_match(bus, slot, match, callback, userdata);
+    return ::sd_bus_add_match(bus, slot, match, callback, userdata);
 }

 sd_bus_slot* SdBus::sd_bus_slot_unref(sd_bus_slot *slot)
--- a/src/SdBus.h
+++ b/src/SdBus.h
@@ -63,6 +63,9 @@ public:
    virtual int sd_bus_open_user(sd_bus **ret) override;
    virtual int sd_bus_open_user_with_address(sd_bus **ret, const char* address) override;
    virtual int sd_bus_open_system_remote(sd_bus **ret, const char* hsot) override;
+    virtual int sd_bus_open_direct(sd_bus **ret, const char* address) override;
+    virtual int sd_bus_open_direct(sd_bus **ret, int fd) override;
+    virtual int sd_bus_open_server(sd_bus **ret, int fd) override;
    virtual int sd_bus_request_name(sd_bus *bus, const char *name, uint64_t flags) override;
    virtual int sd_bus_release_name(sd_bus *bus, const char *name) override;
    virtual int sd_bus_get_unique_name(sd_bus *bus, const char **name) override;
--- a/src/Utils.h
+++ b/src/Utils.h
@@ -31,17 +31,17 @@
 #include SDBUS_HEADER

 #if LIBSYSTEMD_VERSION>=246
-#define SDBUS_CHECK_OBJECT_PATH(_PATH)                                                                                              \
-    SDBUS_THROW_ERROR_IF(!sd_bus_object_path_is_valid(_PATH.c_str()), "Invalid object path '" + _PATH + "' provided", EINVAL)       \
+#define SDBUS_CHECK_OBJECT_PATH(_PATH)                                                                                                            \
+    SDBUS_THROW_ERROR_IF(!sd_bus_object_path_is_valid(_PATH.c_str()), "Invalid object path '" + _PATH + "' provided", EINVAL)                     \
    /**/
-#define SDBUS_CHECK_INTERFACE_NAME(_NAME)                                                                                           \
-    SDBUS_THROW_ERROR_IF(!sd_bus_interface_name_is_valid(_NAME.c_str()), "Invalid interface name '" + _NAME + "' provided", EINVAL) \
+#define SDBUS_CHECK_INTERFACE_NAME(_NAME)                                                                                                         \
+    SDBUS_THROW_ERROR_IF(!sd_bus_interface_name_is_valid(_NAME.c_str()), "Invalid interface name '" + _NAME + "' provided", EINVAL)               \
    /**/
-#define SDBUS_CHECK_SERVICE_NAME(_NAME)                                                                                             \
-    SDBUS_THROW_ERROR_IF(!sd_bus_service_name_is_valid(_NAME.c_str()), "Invalid service name '" + _NAME + "' provided", EINVAL)     \
+#define SDBUS_CHECK_SERVICE_NAME(_NAME)                                                                                                           \
+    SDBUS_THROW_ERROR_IF(!_NAME.empty() && !sd_bus_service_name_is_valid(_NAME.c_str()), "Invalid service name '" + _NAME + "' provided", EINVAL) \
    /**/
-#define SDBUS_CHECK_MEMBER_NAME(_NAME)                                                                                              \
-    SDBUS_THROW_ERROR_IF(!sd_bus_member_name_is_valid(_NAME.c_str()), "Invalid member name '" + _NAME + "' provided", EINVAL)       \
+#define SDBUS_CHECK_MEMBER_NAME(_NAME)                                                                                                            \
+    SDBUS_THROW_ERROR_IF(!sd_bus_member_name_is_valid(_NAME.c_str()), "Invalid member name '" + _NAME + "' provided", EINVAL)                     \
    /**/
 #else
 #define SDBUS_CHECK_OBJECT_PATH(_PATH)
--- a/tests/integrationtests/DBusGeneralTests.cpp
+++ b/tests/integrationtests/DBusGeneralTests.cpp
@@ -46,6 +46,7 @@ using namespace std::chrono_literals;
 using namespace sdbus::test;

 using AConnection = TestFixture;
+using ADirectConnection = TestFixtureWithDirectConnection;

 /*-------------------------------------*/
 /* --          TEST CASES           -- */
@@ -144,3 +145,14 @@ TEST_F(AConnection, WillNotPassToMatchCallbackMessagesThatDoNotMatchTheRule)
    ASSERT_TRUE(waitUntil([&](){ return numberOfMatchingMessages == 2; }));
    ASSERT_FALSE(waitUntil([&](){ return numberOfMatchingMessages > 2; }, 1s));
 }
+
+// A simple direct connection test similar in nature to https://github.com/systemd/systemd/blob/main/src/libsystemd/sd-bus/test-bus-server.c
+TEST_F(ADirectConnection, CanBeUsedBetweenClientAndServer)
+{
+    auto val = m_proxy->sumArrayItems({1, 7}, {2, 3, 4});
+    m_adaptor->emitSimpleSignal();
+
+    // Make sure method call passes and emitted signal is received
+    ASSERT_THAT(val, Eq(1 + 7 + 2 + 3 + 4));
+    ASSERT_TRUE(waitUntil(m_proxy->m_gotSimpleSignal));
+}
--- a/tests/integrationtests/Defs.h
+++ b/tests/integrationtests/Defs.h
@@ -30,14 +30,17 @@
 #include "sdbus-c++/Types.h"
 #include <chrono>
 #include <ostream>
+#include <filesystem>

 namespace sdbus { namespace test {

 const std::string INTERFACE_NAME{"org.sdbuscpp.integrationtests"};
 const std::string BUS_NAME = INTERFACE_NAME;
+const std::string EMPTY_DESTINATION;
 const std::string MANAGER_PATH {"/org/sdbuscpp/integrationtests"};
 const std::string OBJECT_PATH  {"/org/sdbuscpp/integrationtests/ObjectA1"};
 const std::string OBJECT_PATH_2{"/org/sdbuscpp/integrationtests/ObjectB1"};
+const std::string DIRECT_CONNECTION_SOCKET_PATH{std::filesystem::temp_directory_path() / "sdbus-cpp-direct-connection-test"};

 constexpr const uint8_t UINT8_VALUE{1};
 constexpr const int16_t INT16_VALUE{21};
--- a/tests/integrationtests/TestFixture.h
+++ b/tests/integrationtests/TestFixture.h
@@ -39,6 +39,11 @@
 #include <atomic>
 #include <chrono>

+#include <sys/types.h>
+#include <sys/socket.h>
+#include <sys/stat.h>
+#include <sys/un.h>
+
 namespace sdbus { namespace test {

 class TestFixture : public ::testing::Test
@@ -59,28 +64,6 @@ public:
        s_proxyConnection->leaveEventLoop();
    }

-    template <typename _Fnc>
-    static bool waitUntil(_Fnc&& fnc, std::chrono::milliseconds timeout = std::chrono::seconds(5))
-    {
-        using namespace std::chrono_literals;
-
-        std::chrono::milliseconds elapsed{};
-        std::chrono::milliseconds step{5ms};
-        do {
-            std::this_thread::sleep_for(step);
-            elapsed += step;
-            if (elapsed > timeout)
-                return false;
-        } while (!fnc());
-
-        return true;
-    }
-
-    static bool waitUntil(std::atomic<bool>& flag, std::chrono::milliseconds timeout = std::chrono::seconds(5))
-    {
-        return waitUntil([&flag]() -> bool { return flag; }, timeout);
-    }
-
 private:
    void SetUp() override
    {
@@ -106,6 +89,101 @@ public:
    std::unique_ptr<TestProxy> m_proxy;
 };

+class TestFixtureWithDirectConnection : public ::testing::Test
+{
+private:
+    void SetUp() override
+    {
+        int sock = openUnixSocket();
+        createClientAndServerConnections(sock);
+        createAdaptorAndProxyObjects();
+    }
+
+    void TearDown() override
+    {
+        m_proxy.reset();
+        m_adaptor.reset();
+        m_proxyConnection->leaveEventLoop();
+        m_adaptorConnection->leaveEventLoop();
+    }
+
+    static int openUnixSocket()
+    {
+        int sock = socket(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC, 0);
+        assert(sock >= 0);
+
+        sockaddr_un sa;
+        memset(&sa, 0, sizeof(sa));
+        sa.sun_family = AF_UNIX;
+        snprintf(sa.sun_path, sizeof(sa.sun_path), "%s", DIRECT_CONNECTION_SOCKET_PATH.c_str());
+
+        unlink(DIRECT_CONNECTION_SOCKET_PATH.c_str());
+
+        umask(0000);
+        [[maybe_unused]] int r = bind(sock, (const sockaddr*) &sa, sizeof(sa.sun_path));
+        assert(r >= 0);
+
+        r = listen(sock, 5);
+        assert(r >= 0);
+
+        return sock;
+    }
+
+    void createClientAndServerConnections(int sock)
+    {
+        std::thread t([&]()
+        {
+            auto fd = accept4(sock, NULL, NULL, /*SOCK_NONBLOCK|*/SOCK_CLOEXEC);
+            m_adaptorConnection = sdbus::createServerBus(fd);
+            // This is necessary so that createDirectBusConnection() below does not block
+            m_adaptorConnection->enterEventLoopAsync();
+        });
+
+        m_proxyConnection = sdbus::createDirectBusConnection("unix:path=" + DIRECT_CONNECTION_SOCKET_PATH);
+        m_proxyConnection->enterEventLoopAsync();
+
+        t.join();
+    }
+
+    void createAdaptorAndProxyObjects()
+    {
+        assert(m_adaptorConnection != nullptr);
+        assert(m_proxyConnection != nullptr);
+
+        m_adaptor = std::make_unique<TestAdaptor>(*m_adaptorConnection, OBJECT_PATH);
+        // Destination parameter can be empty in case of direct connections
+        m_proxy = std::make_unique<TestProxy>(*m_proxyConnection, EMPTY_DESTINATION, OBJECT_PATH);
+    }
+
+public:
+    std::unique_ptr<sdbus::IConnection> m_adaptorConnection;
+    std::unique_ptr<sdbus::IConnection> m_proxyConnection;
+    std::unique_ptr<TestAdaptor> m_adaptor;
+    std::unique_ptr<TestProxy> m_proxy;
+};
+
+template <typename _Fnc>
+inline bool waitUntil(_Fnc&& fnc, std::chrono::milliseconds timeout = std::chrono::seconds(5))
+{
+    using namespace std::chrono_literals;
+
+    std::chrono::milliseconds elapsed{};
+    std::chrono::milliseconds step{5ms};
+    do {
+        std::this_thread::sleep_for(step);
+        elapsed += step;
+        if (elapsed > timeout)
+            return false;
+    } while (!fnc());
+
+    return true;
+}
+
+inline bool waitUntil(std::atomic<bool>& flag, std::chrono::milliseconds timeout = std::chrono::seconds(5))
+{
+    return waitUntil([&flag]() -> bool { return flag; }, timeout);
+}
+
 }}

 #endif /* SDBUS_CPP_INTEGRATIONTESTS_TESTFIXTURE_H_ */
--- a/tests/unittests/mocks/SdBusMock.h
+++ b/tests/unittests/mocks/SdBusMock.h
@@ -62,6 +62,9 @@ public:
    MOCK_METHOD1(sd_bus_open_user, int(sd_bus **ret));
    MOCK_METHOD2(sd_bus_open_user_with_address, int(sd_bus **ret, const char* address));
    MOCK_METHOD2(sd_bus_open_system_remote, int(sd_bus **ret, const char *host));
+    MOCK_METHOD2(sd_bus_open_direct, int(sd_bus **ret, const char* address));
+    MOCK_METHOD2(sd_bus_open_direct, int(sd_bus **ret, int fd));
+    MOCK_METHOD2(sd_bus_open_server, int(sd_bus **ret, int fd));
    MOCK_METHOD3(sd_bus_request_name, int(sd_bus *bus, const char *name, uint64_t flags));
    MOCK_METHOD2(sd_bus_release_name, int(sd_bus *bus, const char *name));
    MOCK_METHOD2(sd_bus_get_unique_name, int(sd_bus *bus, const char **name));