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
|
// Copyright (c) 2020 - 2021 by Apex.AI Inc. All rights reserved.
//
// 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.
//
// SPDX-License-Identifier: Apache-2.0
#include "iceoryx_hoofs/posix_wrapper/signal_handler.hpp"
#include "iceoryx_posh/popo/untyped_subscriber.hpp"
#include "iceoryx_posh/popo/user_trigger.hpp"
#include "iceoryx_posh/popo/wait_set.hpp"
#include "iceoryx_posh/runtime/posh_runtime.hpp"
#include "topic_data.hpp"
#include <chrono>
#include <iostream>
std::atomic_bool keepRunning{true};
iox::popo::UserTrigger shutdownTrigger;
static void sigHandler(int f_sig IOX_MAYBE_UNUSED)
{
shutdownTrigger.trigger();
}
//! [shutdown callback]
void shutdownCallback(iox::popo::UserTrigger*)
{
std::cout << "CTRL+C pressed - exiting now" << std::endl;
}
//! [shutdown callback]
// The callback of the event. Every callback must have an argument which is
// a pointer to the origin of the Trigger. In our case the event origin is
// the untyped subscriber.
//! [subscriber callback]
void subscriberCallback(iox::popo::UntypedSubscriber* const subscriber, uint64_t* const sumOfAllSamples)
{
while (subscriber->hasData())
{
subscriber->take().and_then([&](auto& userPayload) {
auto chunkHeader = iox::mepoo::ChunkHeader::fromUserPayload(userPayload);
auto flags = std::cout.flags();
std::cout << "subscriber: " << std::hex << subscriber << " length: " << std::dec
<< chunkHeader->userPayloadSize() << " ptr: " << std::hex << chunkHeader->userPayload()
<< std::dec << std::endl;
std::cout.setf(flags);
});
// no nullptr check required since it is guaranteed != nullptr
++(*sumOfAllSamples);
}
}
//! [subscriber callback]
int main()
{
constexpr uint64_t NUMBER_OF_SUBSCRIBERS = 2U;
constexpr uint64_t ONE_SHUTDOWN_TRIGGER = 1U;
// register sigHandler
auto signalIntGuard = iox::posix::registerSignalHandler(iox::posix::Signal::INT, sigHandler);
auto signalTermGuard = iox::posix::registerSignalHandler(iox::posix::Signal::TERM, sigHandler);
iox::runtime::PoshRuntime::initRuntime("iox-cpp-waitset-gateway");
//! [create waitset]
iox::popo::WaitSet<NUMBER_OF_SUBSCRIBERS + ONE_SHUTDOWN_TRIGGER> waitset;
// attach shutdownTrigger to handle CTRL+C
waitset.attachEvent(shutdownTrigger, iox::popo::createNotificationCallback(shutdownCallback)).or_else([](auto) {
std::cerr << "failed to attach shutdown trigger" << std::endl;
std::exit(EXIT_FAILURE);
});
//! [create waitset]
//! [configure]
uint64_t sumOfAllSamples = 0U;
// create subscribers and subscribe them to our service
iox::cxx::vector<iox::popo::UntypedSubscriber, NUMBER_OF_SUBSCRIBERS> subscriberVector;
for (auto i = 0U; i < NUMBER_OF_SUBSCRIBERS; ++i)
{
subscriberVector.emplace_back(iox::capro::ServiceDescription{"Radar", "FrontLeft", "Counter"});
auto& subscriber = subscriberVector.back();
/// important: the user has to ensure that the contextData (sumOfAllSamples) lives as long as
/// the subscriber with its callback is attached to the listener
waitset
.attachEvent(subscriber,
iox::popo::SubscriberEvent::DATA_RECEIVED,
0,
createNotificationCallback(subscriberCallback, sumOfAllSamples))
.or_else([&](auto) {
std::cerr << "failed to attach subscriber" << i << std::endl;
std::exit(EXIT_FAILURE);
});
}
//! [configure]
//! [event loop]
while (keepRunning)
{
auto notificationVector = waitset.wait();
for (auto& notification : notificationVector)
{
if (notification->doesOriginateFrom(&shutdownTrigger))
{
(*notification)();
keepRunning = false;
}
else
{
// call the callback which was assigned to the notification
(*notification)();
}
}
auto flags = std::cout.flags();
std::cout << "sum of all samples: " << std::dec << sumOfAllSamples << std::endl;
std::cout.setf(flags);
}
//! [event loop]
return (EXIT_SUCCESS);
}
|
