-
Notifications
You must be signed in to change notification settings - Fork 91
Expand file tree
/
Copy pathconnection.hpp
More file actions
354 lines (340 loc) · 12.6 KB
/
Copy pathconnection.hpp
File metadata and controls
354 lines (340 loc) · 12.6 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
/*
// Copyright (c) 2018 Intel Corporation
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
*/
#pragma once
#ifndef BOOST_COROUTINES_NO_DEPRECATION_WARNING
// users should define this if they directly include boost/asio/spawn.hpp,
// but by defining it here, warnings won't cause problems with a compile
#define BOOST_COROUTINES_NO_DEPRECATION_WARNING
#endif
#include <boost/asio/async_result.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/posix/stream_descriptor.hpp>
#include <boost/asio/post.hpp>
#ifndef SDBUSPLUS_DISABLE_BOOST_COROUTINES
#include <boost/asio/spawn.hpp>
#endif
#include <boost/callable_traits.hpp>
#include <sdbusplus/asio/detail/async_send_handler.hpp>
#include <sdbusplus/message.hpp>
#include <sdbusplus/utility/read_into_tuple.hpp>
#include <sdbusplus/utility/type_traits.hpp>
#include <chrono>
#include <string>
#include <tuple>
namespace sdbusplus
{
namespace asio
{
/// Root D-Bus IO object
/**
* A connection to a bus, through which messages may be sent or received.
*/
class connection : public sdbusplus::bus_t
{
public:
// default to system bus
connection(boost::asio::io_context& io) :
sdbusplus::bus_t(sdbusplus::bus::new_default()), io_(io),
socket(io_.get_executor(), get_fd())
{
read_immediate();
}
connection(boost::asio::io_context& io, sd_bus* bus) :
sdbusplus::bus_t(bus), io_(io), socket(io_.get_executor(), get_fd())
{
read_immediate();
}
~connection()
{
// The FD will be closed by the socket object, so assign null to the
// sd_bus object to avoid a double close() Ignore return codes here,
// because there's nothing we can do about errors
socket.release();
}
/** @brief Perform an asynchronous send of a message, executing the handler
* upon return and return
*
* @param[in] m - A message ready to send
* @param[in] token - The completion token to execute upon completion;
* @param[in] timeout - The timeout in microseconds
*
*/
template <typename CompletionToken>
inline auto async_send(message_t& m, CompletionToken&& token,
uint64_t timeout = 0)
{
#ifdef SDBUSPLUS_DISABLE_BOOST_COROUTINES
constexpr bool is_yield = false;
#else
constexpr bool is_yield =
std::is_same_v<CompletionToken, boost::asio::yield_context>;
#endif
using return_t = std::conditional_t<is_yield, message_t, message_t&>;
using callback_t = void(boost::system::error_code, return_t);
return boost::asio::async_initiate<CompletionToken, callback_t>(
detail::async_send_handler(get(), m, timeout), token);
}
/** @brief Perform an asynchronous method call, with input parameter packing
* and return value unpacking.
*
* @param[in] handler - A function object that is to be called as a
* continuation for the async dbus method call. The
* arguments to parse on the return are deduced from
* the handler's signature and then passed in along
* with an error code and optional message_t
* @param[in] service - The service to call.
* @param[in] objpath - The object's path for the call.
* @param[in] interf - The object's interface to call.
* @param[in] method - The object's method to call.
* @param[in] timeout - The timeout for the method call in usec (0 results
* in using the default value).
* @param[in] a - Optional parameters for the method call.
*
*/
template <typename MessageHandler, typename... InputArgs>
void async_method_call_timed(MessageHandler&& handler,
const std::string& service,
const std::string& objpath,
const std::string& interf,
const std::string& method, uint64_t timeout,
const InputArgs&... a)
{
using FunctionTuple = boost::callable_traits::args_t<MessageHandler>;
using FunctionTupleType = utility::decay_tuple_t<FunctionTuple>;
constexpr bool returnWithMsg = []() {
if constexpr ((std::tuple_size_v<FunctionTupleType>) > 1)
{
return std::is_same_v<
std::tuple_element_t<1, FunctionTupleType>,
sdbusplus::message_t>;
}
return false;
}();
using UnpackType = utility::strip_first_n_args_t<returnWithMsg ? 2 : 1,
FunctionTupleType>;
auto applyHandler = [handler = std::forward<MessageHandler>(handler)](
boost::system::error_code ec,
message_t& r) mutable {
UnpackType responseData;
if (!ec)
{
try
{
utility::read_into_tuple(responseData, r);
}
catch (const std::exception&)
{
// Set error code if not already set
ec = boost::system::errc::make_error_code(
boost::system::errc::invalid_argument);
}
}
// Note. Callback is called whether or not the unpack was
// successful to allow the user to implement their own handling
if constexpr (returnWithMsg)
{
auto response = std::tuple_cat(std::make_tuple(ec),
std::forward_as_tuple(r),
std::move(responseData));
std::apply(handler, response);
}
else
{
auto response = std::tuple_cat(std::make_tuple(ec),
std::move(responseData));
std::apply(handler, response);
}
};
message_t m;
boost::system::error_code ec;
try
{
m = new_method_call(service.c_str(), objpath.c_str(),
interf.c_str(), method.c_str());
m.append(a...);
}
catch (const exception::SdBusError& e)
{
ec = boost::system::errc::make_error_code(
static_cast<boost::system::errc::errc_t>(e.get_errno()));
applyHandler(ec, m);
return;
}
async_send(m, std::forward<decltype(applyHandler)>(applyHandler),
timeout);
}
/** @brief Perform an asynchronous method call, with input parameter packing
* and return value unpacking. Uses the default timeout value.
*
* @param[in] handler - A function object that is to be called as a
* continuation for the async dbus method call. The
* arguments to parse on the return are deduced from
* the handler's signature and then passed in along
* with an error code and optional message_t
* @param[in] service - The service to call.
* @param[in] objpath - The object's path for the call.
* @param[in] interf - The object's interface to call.
* @param[in] method - The object's method to call.
* @param[in] a - Optional parameters for the method call.
*
*/
template <typename MessageHandler, typename... InputArgs>
void async_method_call(MessageHandler&& handler, const std::string& service,
const std::string& objpath,
const std::string& interf, const std::string& method,
const InputArgs&... a)
{
async_method_call_timed(std::forward<MessageHandler>(handler), service,
objpath, interf, method, 0, a...);
}
#ifndef SDBUSPLUS_DISABLE_BOOST_COROUTINES
/** @brief Perform a yielding asynchronous method call, with input
* parameter packing and return value unpacking
*
* @param[in] yield - A yield context to async block upon.
* @param[in] ec - an error code that will be set for any errors
* @param[in] service - The service to call.
* @param[in] objpath - The object's path for the call.
* @param[in] interf - The object's interface to call.
* @param[in] method - The object's method to call.
* @param[in] a - Optional parameters for the method call.
*
* @return Unpacked value of RetType
*/
template <typename... RetTypes, typename... InputArgs>
auto yield_method_call(boost::asio::yield_context yield,
boost::system::error_code& ec,
const std::string& service,
const std::string& objpath,
const std::string& interf, const std::string& method,
const InputArgs&... a)
{
message_t m;
try
{
m = new_method_call(service.c_str(), objpath.c_str(),
interf.c_str(), method.c_str());
m.append(a...);
}
catch (const exception::SdBusError& e)
{
ec = boost::system::errc::make_error_code(
static_cast<boost::system::errc::errc_t>(e.get_errno()));
}
message_t r;
if (!ec)
{
r = async_send(m, yield[ec]);
}
if constexpr (sizeof...(RetTypes) == 0)
{
// void return
return;
}
else if constexpr (sizeof...(RetTypes) == 1)
{
if constexpr (std::is_same_v<utility::first_type_t<RetTypes...>,
void>)
{
return;
}
else
{
// single item return
utility::first_type_t<RetTypes...> responseData{};
// before attempting to read, check ec and bail on error
if (ec)
{
return responseData;
}
try
{
r.read(responseData);
}
catch (const std::exception&)
{
ec = boost::system::errc::make_error_code(
boost::system::errc::invalid_argument);
// responseData will be default-constructed...
}
return responseData;
}
}
else
{
// tuple of things to return
std::tuple<RetTypes...> responseData{};
// before attempting to read, check ec and bail on error
if (ec)
{
return responseData;
}
try
{
r.read(responseData);
}
catch (const std::exception&)
{
ec = boost::system::errc::make_error_code(
boost::system::errc::invalid_argument);
// responseData will be default-constructed...
}
return responseData;
}
}
#endif
boost::asio::io_context& get_io_context()
{
return io_;
}
private:
boost::asio::io_context& io_;
boost::asio::posix::stream_descriptor socket;
void read_wait()
{
socket.async_read_some(
boost::asio::null_buffers(),
[&](const boost::system::error_code& ec, std::size_t) {
if (ec)
{
return;
}
if (process_discard())
{
read_immediate();
}
else
{
read_wait();
}
});
}
void read_immediate()
{
boost::asio::post(io_, [&] {
if (process_discard())
{
read_immediate();
}
else
{
read_wait();
}
});
}
};
} // namespace asio
} // namespace sdbusplus