forked from Kistler-Group/sdbus-cpp
Compare commits
2 Commits
feat/clang
...
feat/suppo
Author | SHA1 | Date | |
---|---|---|---|
690f0b1387 | |||
fe6650be0f |
@ -42,6 +42,7 @@ namespace sdbus {
|
||||
class IProxy;
|
||||
class Variant;
|
||||
class Error;
|
||||
class PendingAsyncCall;
|
||||
}
|
||||
|
||||
namespace sdbus {
|
||||
@ -193,7 +194,7 @@ namespace sdbus {
|
||||
template <typename _Rep, typename _Period>
|
||||
AsyncMethodInvoker& withTimeout(const std::chrono::duration<_Rep, _Period>& timeout);
|
||||
template <typename... _Args> AsyncMethodInvoker& withArguments(_Args&&... args);
|
||||
template <typename _Function> void uponReplyInvoke(_Function&& callback);
|
||||
template <typename _Function> PendingAsyncCall uponReplyInvoke(_Function&& callback);
|
||||
|
||||
private:
|
||||
IProxy& proxy_;
|
||||
|
@ -576,7 +576,7 @@ namespace sdbus {
|
||||
}
|
||||
|
||||
template <typename _Function>
|
||||
void AsyncMethodInvoker::uponReplyInvoke(_Function&& callback)
|
||||
PendingAsyncCall AsyncMethodInvoker::uponReplyInvoke(_Function&& callback)
|
||||
{
|
||||
assert(method_.isValid()); // onInterface() must be placed/called prior to this function
|
||||
|
||||
@ -594,7 +594,7 @@ namespace sdbus {
|
||||
sdbus::apply(callback, error, args);
|
||||
};
|
||||
|
||||
proxy_.callMethod(method_, std::move(asyncReplyHandler), timeout_);
|
||||
return proxy_.callMethod(method_, std::move(asyncReplyHandler), timeout_);
|
||||
}
|
||||
|
||||
/*** ---------------- ***/
|
||||
|
@ -38,6 +38,10 @@ namespace sdbus {
|
||||
class MethodCall;
|
||||
class MethodReply;
|
||||
class IConnection;
|
||||
class PendingAsyncCall;
|
||||
namespace internal {
|
||||
class Proxy;
|
||||
}
|
||||
}
|
||||
|
||||
namespace sdbus {
|
||||
@ -108,6 +112,7 @@ namespace sdbus {
|
||||
* @param[in] message Message representing an async method call
|
||||
* @param[in] asyncReplyCallback Handler for the async reply
|
||||
* @param[in] timeout Timeout for dbus call in microseconds
|
||||
* @return Cookie for the the pending asynchronous call
|
||||
*
|
||||
* The call is non-blocking. It doesn't wait for the reply. Once the reply arrives,
|
||||
* the provided async reply handler will get invoked from the context of the connection
|
||||
@ -117,7 +122,7 @@ namespace sdbus {
|
||||
*
|
||||
* @throws sdbus::Error in case of failure
|
||||
*/
|
||||
virtual void callMethod(const MethodCall& message, async_reply_handler asyncReplyCallback, uint64_t timeout = 0) = 0;
|
||||
virtual PendingAsyncCall callMethod(const MethodCall& message, async_reply_handler asyncReplyCallback, uint64_t timeout = 0) = 0;
|
||||
|
||||
/*!
|
||||
* @copydoc IProxy::callMethod(const MethodCall&,async_reply_handler,uint64_t)
|
||||
@ -263,6 +268,44 @@ namespace sdbus {
|
||||
[[nodiscard]] PropertySetter setProperty(const std::string& propertyName);
|
||||
};
|
||||
|
||||
/********************************************//**
|
||||
* @class PendingAsyncCall
|
||||
*
|
||||
* PendingAsyncCall represents a simple handle type to cancel the delivery
|
||||
* of the asynchronous D-Bus call result to the application.
|
||||
*
|
||||
***********************************************/
|
||||
class PendingAsyncCall
|
||||
{
|
||||
public:
|
||||
/*!
|
||||
* @brief Cancels the pending asynchronous call
|
||||
*
|
||||
* Removes the callback handler registered to the call result delivery.
|
||||
* Does nothing if the callback handler is in progress already.
|
||||
*/
|
||||
void cancel();
|
||||
|
||||
/*!
|
||||
* @brief Answers whether the asynchronous call is still pending
|
||||
*
|
||||
* @return True if the call is pending, false if the call has been fully completed
|
||||
*
|
||||
* Pending call in this context means a call whose results have not arrived,
|
||||
* or have arrived and are currently being handled by the async callback handler.
|
||||
* After the callback handler has completed, the call is deemed completed.
|
||||
*/
|
||||
bool isPending();
|
||||
|
||||
private:
|
||||
friend internal::Proxy;
|
||||
PendingAsyncCall(IProxy& proxy, void* slot);
|
||||
|
||||
private:
|
||||
IProxy& proxy_;
|
||||
void* slot_;
|
||||
};
|
||||
|
||||
// Out-of-line member definitions
|
||||
|
||||
template <typename _Rep, typename _Period>
|
||||
@ -304,6 +347,11 @@ namespace sdbus {
|
||||
return PropertySetter(*this, propertyName);
|
||||
}
|
||||
|
||||
inline PendingAsyncCall::PendingAsyncCall(IProxy& proxy, void* slot)
|
||||
: proxy_(proxy), slot_(slot)
|
||||
{
|
||||
}
|
||||
|
||||
/*!
|
||||
* @brief Creates a proxy object for a specific remote D-Bus object
|
||||
*
|
||||
|
@ -93,7 +93,7 @@ MethodReply Proxy::callMethod(const MethodCall& message, uint64_t timeout)
|
||||
return sendMethodCallMessageAndWaitForReply(message, timeout);
|
||||
}
|
||||
|
||||
void Proxy::callMethod(const MethodCall& message, async_reply_handler asyncReplyCallback, uint64_t timeout)
|
||||
PendingAsyncCall Proxy::callMethod(const MethodCall& message, async_reply_handler asyncReplyCallback, uint64_t timeout)
|
||||
{
|
||||
SDBUS_THROW_ERROR_IF(!message.isValid(), "Invalid async method call message provided", EINVAL);
|
||||
|
||||
@ -102,7 +102,10 @@ void Proxy::callMethod(const MethodCall& message, async_reply_handler asyncReply
|
||||
|
||||
callData->slot = message.send(callback, callData.get(), timeout);
|
||||
|
||||
pendingAsyncCalls_.addCall(callData->slot.get(), std::move(callData));
|
||||
auto slotPtr = callData->slot.get();
|
||||
pendingAsyncCalls_.addCall(slotPtr, std::move(callData));
|
||||
|
||||
return {*this, slotPtr};
|
||||
}
|
||||
|
||||
MethodReply Proxy::sendMethodCallMessageAndWaitForReply(const MethodCall& message, uint64_t timeout)
|
||||
@ -124,7 +127,7 @@ MethodReply Proxy::sendMethodCallMessageAndWaitForReply(const MethodCall& messag
|
||||
void Proxy::SyncCallReplyData::sendMethodReplyToWaitingThread(MethodReply& reply, const Error* error)
|
||||
{
|
||||
SCOPE_EXIT{ cond_.notify_one(); };
|
||||
std::unique_lock<std::mutex> lock{mutex_};
|
||||
std::unique_lock lock{mutex_};
|
||||
SCOPE_EXIT{ arrived_ = true; };
|
||||
|
||||
//error_ = nullptr; // Necessary if SyncCallReplyData instance is thread_local
|
||||
@ -137,7 +140,7 @@ void Proxy::SyncCallReplyData::sendMethodReplyToWaitingThread(MethodReply& reply
|
||||
|
||||
MethodReply Proxy::SyncCallReplyData::waitForMethodReply()
|
||||
{
|
||||
std::unique_lock<std::mutex> lock{mutex_};
|
||||
std::unique_lock lock{mutex_};
|
||||
cond_.wait(lock, [this](){ return arrived_; });
|
||||
|
||||
//arrived_ = false; // Necessary if SyncCallReplyData instance is thread_local
|
||||
@ -245,6 +248,26 @@ int Proxy::sdbus_signal_handler(sd_bus_message *sdbusMessage, void *userData, sd
|
||||
|
||||
namespace sdbus {
|
||||
|
||||
void PendingAsyncCall::cancel()
|
||||
{
|
||||
assert(dynamic_cast<internal::Proxy*>(&proxy_) != nullptr);
|
||||
assert(slot_ != nullptr);
|
||||
|
||||
static_cast<internal::Proxy&>(proxy_).pendingAsyncCalls_.removeCall(slot_);
|
||||
}
|
||||
|
||||
bool PendingAsyncCall::isPending()
|
||||
{
|
||||
assert(dynamic_cast<internal::Proxy*>(&proxy_) != nullptr);
|
||||
assert(slot_ != nullptr);
|
||||
|
||||
return static_cast<internal::Proxy&>(proxy_).pendingAsyncCalls_.existsCall(slot_);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
namespace sdbus {
|
||||
|
||||
std::unique_ptr<sdbus::IProxy> createProxy( IConnection& connection
|
||||
, std::string destination
|
||||
, std::string objectPath )
|
||||
|
33
src/Proxy.h
33
src/Proxy.h
@ -52,7 +52,7 @@ namespace sdbus::internal {
|
||||
|
||||
MethodCall createMethodCall(const std::string& interfaceName, const std::string& methodName) override;
|
||||
MethodReply callMethod(const MethodCall& message, uint64_t timeout) override;
|
||||
void callMethod(const MethodCall& message, async_reply_handler asyncReplyCallback, uint64_t timeout) override;
|
||||
PendingAsyncCall callMethod(const MethodCall& message, async_reply_handler asyncReplyCallback, uint64_t timeout) override;
|
||||
|
||||
void registerSignalHandler( const std::string& interfaceName
|
||||
, const std::string& signalName
|
||||
@ -81,6 +81,8 @@ namespace sdbus::internal {
|
||||
static int sdbus_signal_handler(sd_bus_message *sdbusMessage, void *userData, sd_bus_error *retError);
|
||||
|
||||
private:
|
||||
friend PendingAsyncCall;
|
||||
|
||||
std::unique_ptr< sdbus::internal::IConnection
|
||||
, std::function<void(sdbus::internal::IConnection*)>
|
||||
> connection_;
|
||||
@ -125,19 +127,38 @@ namespace sdbus::internal {
|
||||
return calls_.emplace(slot, std::move(asyncCallData)).second;
|
||||
}
|
||||
|
||||
bool removeCall(void* slot)
|
||||
bool existsCall(void* slot)
|
||||
{
|
||||
std::lock_guard lock(mutex_);
|
||||
return calls_.erase(slot) > 0;
|
||||
return calls_.count(slot) > 0;
|
||||
}
|
||||
|
||||
void removeCall(void* slot)
|
||||
{
|
||||
std::unique_lock lock(mutex_);
|
||||
if (auto it = calls_.find(slot); it != calls_.end())
|
||||
{
|
||||
auto callData = std::move(it->second);
|
||||
calls_.erase(it);
|
||||
lock.unlock();
|
||||
|
||||
// Releasing call slot pointer acquires global sd-bus mutex. We have to perform the release
|
||||
// out of the `mutex_' critical section here, because if the `removeCall` is called by some
|
||||
// thread and at the same time Proxy's async reply handler (which already holds global sd-bus
|
||||
// mutex) is in progress in a different thread, we get double-mutex deadlock.
|
||||
}
|
||||
}
|
||||
|
||||
void clear()
|
||||
{
|
||||
std::unique_lock<std::mutex> lock(mutex_);
|
||||
std::unique_lock lock(mutex_);
|
||||
auto asyncCallSlots = std::move(calls_);
|
||||
// Perform releasing of sd-bus slots outside of the calls_ critical section which avoids
|
||||
// double mutex dead lock when the async reply handler is invoked at the same time.
|
||||
lock.unlock();
|
||||
|
||||
// Releasing call slot pointer acquires global sd-bus mutex. We have to perform the release
|
||||
// out of the `mutex_' critical section here, because if the `clear` is called by some thread
|
||||
// and at the same time Proxy's async reply handler (which already holds global sd-bus
|
||||
// mutex) is in progress in a different thread, we get double-mutex deadlock.
|
||||
}
|
||||
|
||||
private:
|
||||
|
@ -385,6 +385,53 @@ TEST_F(SdbusTestObject, InvokesMethodAsynchronouslyOnClientSide)
|
||||
ASSERT_THAT(future.get(), Eq(100));
|
||||
}
|
||||
|
||||
TEST_F(SdbusTestObject, AnswersThatAsyncCallIsPendingIfItIsInProgress)
|
||||
{
|
||||
std::promise<uint32_t> promise;
|
||||
auto future = promise.get_future();
|
||||
m_proxy->installDoOperationClientSideAsyncReplyHandler([&](uint32_t /*res*/, const sdbus::Error* /*err*/){ promise.set_value(1); });
|
||||
|
||||
auto call = m_proxy->doOperationClientSideAsync(100);
|
||||
|
||||
ASSERT_TRUE(call.isPending());
|
||||
}
|
||||
|
||||
TEST_F(SdbusTestObject, CancelsPendingAsyncCallOnClientSide)
|
||||
{
|
||||
std::promise<uint32_t> promise;
|
||||
auto future = promise.get_future();
|
||||
m_proxy->installDoOperationClientSideAsyncReplyHandler([&](uint32_t /*res*/, const sdbus::Error* /*err*/){ promise.set_value(1); });
|
||||
auto call = m_proxy->doOperationClientSideAsync(100);
|
||||
|
||||
call.cancel();
|
||||
|
||||
ASSERT_THAT(future.wait_for(300ms), Eq(std::future_status::timeout));
|
||||
}
|
||||
|
||||
TEST_F(SdbusTestObject, AnswersThatAsyncCallIsNotPendingAfterItHasBeenCancelled)
|
||||
{
|
||||
std::promise<uint32_t> promise;
|
||||
auto future = promise.get_future();
|
||||
m_proxy->installDoOperationClientSideAsyncReplyHandler([&](uint32_t /*res*/, const sdbus::Error* /*err*/){ promise.set_value(1); });
|
||||
auto call = m_proxy->doOperationClientSideAsync(100);
|
||||
|
||||
call.cancel();
|
||||
|
||||
ASSERT_FALSE(call.isPending());
|
||||
}
|
||||
|
||||
TEST_F(SdbusTestObject, AnswersThatAsyncCallIsNotPendingAfterItHasBeenCompleted)
|
||||
{
|
||||
std::promise<uint32_t> promise;
|
||||
auto future = promise.get_future();
|
||||
m_proxy->installDoOperationClientSideAsyncReplyHandler([&](uint32_t /*res*/, const sdbus::Error* /*err*/){ promise.set_value(1); });
|
||||
|
||||
auto call = m_proxy->doOperationClientSideAsync(0);
|
||||
(void) future.get(); // Wait for the call to finish
|
||||
|
||||
ASSERT_FALSE(call.isPending());
|
||||
}
|
||||
|
||||
TEST_F(SdbusTestObject, InvokesErroneousMethodAsynchronouslyOnClientSide)
|
||||
{
|
||||
std::promise<uint32_t> promise;
|
||||
|
@ -156,15 +156,15 @@ public:
|
||||
return result;
|
||||
}
|
||||
|
||||
void doOperationClientSideAsync(uint32_t param)
|
||||
sdbus::PendingAsyncCall doOperationClientSideAsync(uint32_t param)
|
||||
{
|
||||
object_.callMethodAsync("doOperation")
|
||||
.onInterface(INTERFACE_NAME)
|
||||
.withArguments(param)
|
||||
.uponReplyInvoke([this](const sdbus::Error* error, uint32_t returnValue)
|
||||
{
|
||||
this->onDoOperationReply(returnValue, error);
|
||||
});
|
||||
return object_.callMethodAsync("doOperation")
|
||||
.onInterface(INTERFACE_NAME)
|
||||
.withArguments(param)
|
||||
.uponReplyInvoke([this](const sdbus::Error* error, uint32_t returnValue)
|
||||
{
|
||||
this->onDoOperationReply(returnValue, error);
|
||||
});
|
||||
}
|
||||
|
||||
void doErroneousOperationClientSideAsync()
|
||||
|
@ -33,9 +33,9 @@ protected:
|
||||
virtual void onConcatenateReply(const std::string& result, const sdbus::Error* error) = 0;
|
||||
|
||||
public:
|
||||
void concatenate(const std::map<std::string, sdbus::Variant>& params)
|
||||
sdbus::PendingAsyncCall concatenate(const std::map<std::string, sdbus::Variant>& params)
|
||||
{
|
||||
proxy_.callMethodAsync("concatenate").onInterface(INTERFACE_NAME).withArguments(params).uponReplyInvoke([this](const sdbus::Error* error, const std::string& result){ this->onConcatenateReply(result, error); });
|
||||
return proxy_.callMethodAsync("concatenate").onInterface(INTERFACE_NAME).withArguments(params).uponReplyInvoke([this](const sdbus::Error* error, const std::string& result){ this->onConcatenateReply(result, error); });
|
||||
}
|
||||
|
||||
private:
|
||||
|
@ -173,7 +173,8 @@ std::tuple<std::string, std::string> ProxyGenerator::processMethods(const Nodes&
|
||||
std::string outArgStr, outArgTypeStr;
|
||||
std::tie(outArgStr, outArgTypeStr, std::ignore, std::ignore) = argsToNamesAndTypes(outArgs);
|
||||
|
||||
definitionSS << tab << (async ? "void" : retType) << " " << name << "(" << inArgTypeStr << ")" << endl
|
||||
const std::string realRetType = (async && !dontExpectReply ? "sdbus::PendingAsyncCall" : async ? "void" : retType);
|
||||
definitionSS << tab << realRetType << " " << name << "(" << inArgTypeStr << ")" << endl
|
||||
<< tab << "{" << endl;
|
||||
|
||||
if (!timeoutValue.empty())
|
||||
@ -186,8 +187,8 @@ std::tuple<std::string, std::string> ProxyGenerator::processMethods(const Nodes&
|
||||
definitionSS << tab << tab << retType << " result;" << endl;
|
||||
}
|
||||
|
||||
definitionSS << tab << tab << "proxy_.callMethod" << (async ? "Async" : "") << "(\"" << name << "\")"
|
||||
".onInterface(INTERFACE_NAME)";
|
||||
definitionSS << tab << tab << (async && !dontExpectReply ? "return " : "")
|
||||
<< "proxy_.callMethod" << (async ? "Async" : "") << "(\"" << name << "\").onInterface(INTERFACE_NAME)";
|
||||
|
||||
if (!timeoutValue.empty())
|
||||
{
|
||||
|
Reference in New Issue
Block a user