10 #include <fmt/format.h>
11 #include <fmt/ostream.h>
12 #include <log4cplus/loggingmacros.h>
27 bool AwaitUnwrapReplies(boost::future<std::vector<boost::future<bool>>>&& futures) {
28 std::vector<boost::future<bool>> values = futures.get();
33 std::vector<std::exception_ptr> exceptions;
35 size_t num_exceptions = 0;
36 for (
auto& f : values) {
41 }
catch (std::exception
const& e) {
45 fmt::format(
"daq::op::AwaitUnwrapReplies: Source replied with exception: {}",
51 fmt::format(
"daq::op::AwaitUnwrapReplies: {}/{} replies contained exceptions. {}/{} "
52 "replies report success",
58 return num_ok == values.size();
61 class RecWaitSpec :
public recif::RecWaitSpec {
63 RecWaitSpec(
float timeout) : m_timeout{timeout}, m_info{} {
65 std::string getInfo()
const override {
68 void setInfo(
const std::string& info)
override {
72 float getTimeout()
const override {
75 void setTimeout(
float timeout)
override {
78 bool hasKey()
const override {
81 std::unique_ptr<recif::RecWaitSpec> cloneKey()
const override {
82 throw std::runtime_error(
"not clonable");
84 std::unique_ptr<recif::RecWaitSpec> clone()
const override {
85 throw std::runtime_error(
"not clonable");
87 bool keyEquals(
const recif::RecWaitSpec& other)
const override {
100 : m_params(m_params), m_error(
false), m_parts(), m_aborted(
false) {
105 fmt::format(
"AwaitPrimAsync::Initiate: Operation initiating"));
107 return m_promise.get_future();
110 void AwaitPrimAsync::InitiateAwait() {
114 .then(m_params.
common.
executor, [
this](
auto) { return AwaitOnceAsync(); })
116 .then(m_params.
common.
executor, [
this](boost::future<bool> fut) ->
void {
119 LOG4CPLUS_DEBUG(m_params.common.logger,
120 fmt::format(
"AwaitPrimAsync::InitiateAwait: Operation aborted"));
124 auto is_ok = fut.get();
127 this->m_promise.set_value({m_error, std::move(m_parts)});
130 this->InitiateAwait();
134 this->m_promise.set_exception(boost::current_exception());
139 boost::future<void> AwaitPrimAsync::MakeInterval() {
140 using std::chrono::duration_cast;
141 using std::chrono::milliseconds;
142 using std::chrono::steady_clock;
146 return boost::make_ready_future();
149 auto now = steady_clock::now();
150 auto next_start = *m_last_start + duration_cast<steady_clock::duration>(m_params.wait_interval);
151 if (now >= next_start) {
153 return boost::make_ready_future();
155 LOG4CPLUS_DEBUG(m_params.common.logger,
156 fmt::format(
"AwaitPrimAsync::MakeInterval: Waiting {}ms until sending RecWait",
157 duration_cast<milliseconds>(next_start - now).count()));
160 m_interval.emplace(m_params.common.executor.get_io_context(), next_start);
161 m_interval->timer.async_wait([
this](boost::system::error_code
const& ec) {
164 m_interval->promise.set_value();
166 return m_interval->promise.get_future();
169 void AwaitPrimAsync::Abort() noexcept {
170 LOG4CPLUS_INFO(m_params.common.logger,
171 fmt::format(
"AwaitPrimAsync::Abort: Requested to abort! "
172 "Number of files received so far: {}",
179 m_promise.set_value({m_error, std::move(m_parts)});
182 boost::future<bool> AwaitPrimAsync::AwaitOnceAsync() {
183 using Reply = std::shared_ptr<recif::RecWaitStatus>;
184 using Seconds = std::chrono::duration<float>;
185 using std::chrono::duration_cast;
187 LOG4CPLUS_DEBUG(m_params.common.logger,
188 fmt::format(
"AwaitPrimAsync::AwaitOnceAsync: Operation aborted"));
189 return boost::make_ready_future<bool>(
true);
191 LOG4CPLUS_DEBUG(m_params.common.logger,
192 fmt::format(
"AwaitPrimAsync::AwaitOnceAsync: Sending requests..."));
194 m_last_start = std::chrono::steady_clock::now();
196 return SendRequestAndCollectReplies<bool>(
197 m_params.common.prim_sources.begin(),
198 m_params.common.prim_sources.end(),
201 return IsSubsequentState(State::Stopped, source.GetState());
206 auto& client = s.GetSource().GetRrClient();
207 auto seconds = duration_cast<Seconds>(this->m_params.wait_interval).count();
208 #if defined(UNIT_TEST)
213 auto spec = std::make_shared<RecWaitSpec>(seconds);
215 auto mal = client.getMal();
216 BOOST_ASSERT_MSG(mal,
"MAL RR client returned invalid MAL pointer");
217 auto spec = mal->createDataEntity<recif::RecWaitSpec>();
219 *spec = RecWaitSpec(seconds);
221 return client.RecWait(spec);
225 ->
bool { return HandleRecWaitReply(source, std::move(fut)); },
226 std::string_view(
"AwaitPrimAsync: await primary data acquisition"))
227 .then(m_params.common.executor, AwaitUnwrapReplies);
230 bool AwaitPrimAsync::HandleRecWaitReply(
231 Source<PrimSource>& source, boost::future<std::shared_ptr<recif::RecWaitStatus>>&& fut) {
237 auto reply = fut.get();
239 m_params.common.status.ClearAlert(alert_id);
241 auto status = reply->getStatus();
242 if (status == recif::Success) {
246 auto rec_status = reply->getRecStatus();
247 LOG4CPLUS_INFO(m_params.common.logger,
248 fmt::format(
"Data source '{}' replied successfully and provides {} "
251 rec_status->getDpFiles().size()));
252 if (rec_status->getDpFiles().empty()) {
253 LOG4CPLUS_WARN(m_params.common.logger,
254 fmt::format(
"Data source '{}' replied successfully for "
255 "RecWait but did not produce any files!",
256 source.GetSource()));
258 for (
auto const& file : rec_status->getDpFiles()) {
259 m_parts.emplace_back(std::string(source.GetSource().GetName()), file);
267 }
catch (recif::ExceptionErr
const& e) {
268 m_params.common.status.SetAlert(
270 fmt::format(
"Primary source '{}' request 'RecWaitStatus' replied "
271 "with ICD error: ({}) {}",
272 source.GetSource().GetName(),
276 throw boost::enable_current_exception(
277 DaqSourceError(
"RecWait", std::string(source.GetSource().GetName()), e.what()));
280 m_params.common.status.SetAlert(
282 fmt::format(
"Primary source '{}' request 'RecWaitStatus' replied "
283 "with non-ICD error: {}",
284 source.GetSource().GetName(),
287 throw boost::enable_current_exception(DaqSourceError(
288 "RecWait", std::string(source.GetSource().GetName()),
"unknown exception"));
Contains declaration for the AwaitPrimAsync operation.
Declares daq::State and related functions.
constexpr std::string_view REQUEST
Request.
void FormatException(std::ostream &os, std::exception_ptr ptr)
Report without nesting.
void Reply(AsyncOpParams ¶ms, SourceType const &source, std::string description, bool error)
AlertId MakeAlertId(std::string_view category, std::string key)
@ Stopped
All data sources have reported they have stopped acquiring data.
Alert MakeAlert(std::string_view category, std::string key, std::string description)
Construct alert.
Simple class that holds the source and associated state.
rad::IoExecutor & executor
log4cplus::Logger & logger
Await specific parameters that is not provided with AsyncOpParams.
boost::future< Result< DpParts > > Initiate()
Initiates operation that await acquisition completion.
AwaitPrimAsync(AwaitOpParams params) noexcept
Constructs operation with the privided parameters.
Contains declaration for the async op utilities.