feedc0de/sdbus-cpp
1
0
Fork 0
forked from Kistler-Group/sdbus-cpp
Code Pull Requests Activity

refactor: improve Proxy signal subscription (#389)

Browse Source 
This makes D-Bus proxy signal registration more flexible, more dynamic, and less error-prone since no `finishRegistration()` call is needed. A proxy can register to a signal at any time during its lifetime, and can unregister freely by simply destroying the associated slot.
This commit is contained in:
Stanislav Angelovič
2023-12-30 21:11:02 +01:00
committed by Stanislav Angelovic
parent e3040c0998
commit bdd0bc2c27
17 changed files with 227 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Connection.cpp
+20 -14
View File
@@ -233,15 +233,13 @@ Slot Connection::addMatch(const std::string& match, message_handler callback)
auto matchInfo = std::make_unique<MatchInfo>(MatchInfo{std::move(callback), {}, *this, {}});
auto r = sdbus_->sd_bus_add_match(bus_.get(), &matchInfo->slot, match.c_str(), &Connection::sdbus_match_callback, matchInfo.get());
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);
return {matchInfo.release(), [this](void *ptr)
{
auto* matchInfo = static_cast<MatchInfo*>(ptr);
sdbus_->sd_bus_slot_unref(matchInfo->slot);
std::default_delete<MatchInfo>{}(matchInfo);
}};
matchInfo->slot = {slot, [this](void *slot){ sdbus_->sd_bus_slot_unref((sd_bus_slot*)slot); }};
return {matchInfo.release(), [](void *ptr){ delete static_cast<MatchInfo*>(ptr); }};
}
void Connection::addMatch(const std::string& match, message_handler callback, floating_slot_t)
@@ -256,20 +254,18 @@ Slot Connection::addMatchAsync(const std::string& match, message_handler callbac
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()
, &matchInfo->slot
, &slot
, match.c_str()
, &Connection::sdbus_match_callback
, sdbusInstallCallback
, matchInfo.get());
SDBUS_THROW_ERROR_IF(r < 0, "Failed to add match", -r);
return {matchInfo.release(), [this](void *ptr)
{
auto* matchInfo = static_cast<MatchInfo*>(ptr);
sdbus_->sd_bus_slot_unref(matchInfo->slot);
std::default_delete<MatchInfo>{}(matchInfo);
}};
matchInfo->slot = {slot, [this](void *slot){ sdbus_->sd_bus_slot_unref((sd_bus_slot*)slot); }};
return {matchInfo.release(), [](void *ptr){ delete static_cast<MatchInfo*>(ptr); }};
}
void Connection::addMatchAsync(const std::string& match, message_handler callback, message_handler installCallback, floating_slot_t)
@@ -796,16 +792,26 @@ std::vector</*const */char*> Connection::to_strv(const std::vector<std::string>&
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.getSdBusInterface());
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.getSdBusInterface());
auto ok = invokeHandlerAndCatchErrors([&](){ matchInfo->installCallback(std::move(message)); }, retError);
return ok ? 0 : -1;
}
src/Connection.h
+1 -1
View File
@@ -193,7 +193,7 @@ namespace sdbus::internal {
message_handler callback;
message_handler installCallback;
Connection& connection;
sd_bus_slot *slot;
Slot slot;
};
// sd-event integration
src/Proxy.cpp
+25 -46
View File
@@ -135,59 +135,37 @@ std::future<MethodReply> Proxy::callMethod(const MethodCall& message, uint64_t t
void Proxy::registerSignalHandler( const std::string& interfaceName
, const std::string& signalName
, signal_handler signalHandler )
{
auto slot = Proxy::registerSignalHandler(interfaceName, signalName, std::move(signalHandler), request_slot);
floatingSignalSlots_.push_back(std::move(slot));
}
Slot Proxy::registerSignalHandler( const std::string& interfaceName
, const std::string& signalName
, signal_handler signalHandler
, request_slot_t )
{
SDBUS_CHECK_INTERFACE_NAME(interfaceName);
SDBUS_CHECK_MEMBER_NAME(signalName);
SDBUS_THROW_ERROR_IF(!signalHandler, "Invalid signal handler provided", EINVAL);
auto& interface = interfaces_[interfaceName];
auto signalInfo = std::make_unique<SignalInfo>(SignalInfo{std::move(signalHandler), *this, {}});
auto signalData = std::make_unique<InterfaceData::SignalData>(*this, std::move(signalHandler), nullptr);
auto insertionResult = interface.signals_.emplace(signalName, std::move(signalData));
signalInfo->slot = connection_->registerSignalHandler( destination_
, objectPath_
, interfaceName
, signalName
, &Proxy::sdbus_signal_handler
, signalInfo.get() );
auto inserted = insertionResult.second;
SDBUS_THROW_ERROR_IF(!inserted, "Failed to register signal handler: handler already exists", EINVAL);
}
void Proxy::unregisterSignalHandler( const std::string& interfaceName
, const std::string& signalName )
{
auto it = interfaces_.find(interfaceName);
if (it != interfaces_.end())
it->second.signals_.erase(signalName);
}
void Proxy::finishRegistration()
{
registerSignalHandlers(*connection_);
}
void Proxy::registerSignalHandlers(sdbus::internal::IConnection& connection)
{
for (auto& interfaceItem : interfaces_)
{
const auto& interfaceName = interfaceItem.first;
auto& signalsOnInterface = interfaceItem.second.signals_;
for (auto& signalItem : signalsOnInterface)
{
const auto& signalName = signalItem.first;
auto* signalData = signalItem.second.get();
signalData->slot = connection.registerSignalHandler( destination_
, objectPath_
, interfaceName
, signalName
, &Proxy::sdbus_signal_handler
, signalData);
}
}
return {signalInfo.release(), [](void *ptr){ delete static_cast<SignalInfo*>(ptr); }};
}
void Proxy::unregister()
{
pendingAsyncCalls_.clear();
interfaces_.clear();
floatingSignalSlots_.clear();
}
sdbus::IConnection& Proxy::getConnection() const
@@ -241,13 +219,14 @@ int Proxy::sdbus_async_reply_handler(sd_bus_message *sdbusMessage, void *userDat
int Proxy::sdbus_signal_handler(sd_bus_message *sdbusMessage, void *userData, sd_bus_error *retError)
{
auto* signalData = static_cast<InterfaceData::SignalData*>(userData);
assert(signalData != nullptr);
assert(signalData->callback);
auto* signalInfo = static_cast<SignalInfo*>(userData);
assert(signalInfo != nullptr);
assert(signalInfo->callback);
auto message = Message::Factory::create<Signal>(sdbusMessage, &signalData->proxy.connection_->getSdBusInterface());
// TODO: Hide Message factory invocation under Connection API (tell, don't ask principle), then we can remove getSdBusInterface()
auto message = Message::Factory::create<Signal>(sdbusMessage, &signalInfo->proxy.connection_->getSdBusInterface());
auto ok = invokeHandlerAndCatchErrors([&](){ signalData->callback(std::move(message)); }, retError);
auto ok = invokeHandlerAndCatchErrors([&](){ signalInfo->callback(std::move(message)); }, retError);
return ok ? 0 : -1;
}
src/Proxy.h
+35 -28
View File
@@ -32,10 +32,9 @@
#include SDBUS_HEADER
#include <string>
#include <memory>
#include <map>
#include <deque>
#include <vector>
#include <mutex>
#include <condition_variable>
namespace sdbus::internal {
@@ -63,10 +62,10 @@ namespace sdbus::internal {
void registerSignalHandler( const std::string& interfaceName
, const std::string& signalName
, signal_handler signalHandler ) override;
void unregisterSignalHandler( const std::string& interfaceName
, const std::string& signalName ) override;
void finishRegistration() override;
Slot registerSignalHandler( const std::string& interfaceName
, const std::string& signalName
, signal_handler signalHandler
, request_slot_t ) override;
void unregister() override;
sdbus::IConnection& getConnection() const override;
@@ -74,9 +73,8 @@ namespace sdbus::internal {
Message getCurrentlyProcessedMessage() const override;
private:
void registerSignalHandlers(sdbus::internal::IConnection& connection);
static int sdbus_async_reply_handler(sd_bus_message *sdbusMessage, void *userData, sd_bus_error *retError);
static int sdbus_signal_handler(sd_bus_message *sdbusMessage, void *userData, sd_bus_error *retError);
static int sdbus_async_reply_handler(sd_bus_message *sdbusMessage, void *userData, sd_bus_error *retError);
private:
friend PendingAsyncCall;
@@ -87,28 +85,37 @@ namespace sdbus::internal {
std::string destination_;
std::string objectPath_;
using InterfaceName = std::string;
struct InterfaceData
std::vector<Slot> floatingSignalSlots_;
struct SignalInfo
{
using SignalName = std::string;
struct SignalData
{
SignalData(Proxy& proxy, signal_handler callback, Slot slot)
: proxy(proxy)
, callback(std::move(callback))
, slot(std::move(slot))
{}
Proxy& proxy;
signal_handler callback;
// slot must be listed after callback to ensure that slot is destructed first.
// Destructing the slot will sd_bus_slot_unref() the callback.
// Only after sd_bus_slot_unref(), we can safely delete the callback. The bus mutex (SdBus::sdbusMutex_)
// ensures that sd_bus_slot_unref() and the callback execute sequentially.
Slot slot;
};
std::map<SignalName, std::unique_ptr<SignalData>> signals_;
signal_handler callback;
Proxy& proxy;
Slot slot;
};
std::map<InterfaceName, InterfaceData> interfaces_;
// using InterfaceName = std::string;
// struct InterfaceData
// {
// using SignalName = std::string;
// struct SignalData
// {
// SignalData(Proxy& proxy, signal_handler callback, Slot slot)
// : proxy(proxy)
// , callback(std::move(callback))
// , slot(std::move(slot))
// {}
// Proxy& proxy;
// signal_handler callback;
// // slot must be listed after callback to ensure that slot is destructed first.
// // Destructing the slot will sd_bus_slot_unref() the callback.
// // Only after sd_bus_slot_unref(), we can safely delete the callback. The bus mutex (SdBus::sdbusMutex_)
// // ensures that sd_bus_slot_unref() and the callback execute sequentially.
// Slot slot;
// };
// std::map<SignalName, std::unique_ptr<SignalData>> signals_;
// };
// std::map<InterfaceName, InterfaceData> interfaces_;
// We need to keep track of pending async calls. When the proxy is being destructed, we must
// remove all slots of these pending calls, otherwise in case when the connection outlives