RTI Routing Service Version 7.7.0
ProcessorForwarder.hpp
1/*
2 * (c) Copyright, Real-Time Innovations, 2017.
3 * All rights reserved.
4 *
5 * No duplications, whole or partial, manual or electronic, may be made
6 * without express written permission. Any such copies, or
7 * revisions thereof, must display this notice unaltered.
8 * This code contains trade secrets of Real-Time Innovations, Inc.
9 */
10
11#ifndef RTI_ROUTING_PROCESSOR_DETAIL_PROCESSOR_FORWARDER_HPP_
12#define RTI_ROUTING_PROCESSOR_DETAIL_PROCESSOR_FORWARDER_HPP_
13
14#include <rti/core/Exception.hpp>
15
16#include "rti/apputils/ServiceException.hpp"
17
18#include "routingservice/routingservice_processor_impl.h"
19#include <rti/routing/detail/ForwarderUtils.hpp>
20#include <rti/routing/detail/UpdatableEntityForwarder.hpp>
21#include <rti/routing/processor/Route.hpp>
22#include <rti/routing/processor/Input.hpp>
23#include <rti/routing/processor/Output.hpp>
24
25namespace rti { namespace routing { namespace processor { namespace detail {
26
27class ProcessorForwarder {
28
29private:
30
31 template <typename PORT, typename NATIVE>
32 class ScopedPort {
33
34 public:
35 ScopedPort(Route& route, int32_t index, NATIVE *native)
36 : route_(route),
37 port_(new PORT(native, index, route.native_route_))
38 {
39 }
40
41 PORT* get()
42 {
43 return port_;
44 }
45
46 void clear()
47 {
48 port_ = NULL;
49 }
50
51 ~ScopedPort()
52 {
53 if (port_ != NULL) {
54 delete port_;
55 port_ = NULL;
56 }
57 }
58
59 private:
60 Route& route_;
61 PORT *port_;
62 };
63
64public:
65
66 static RTI_RoutingServiceProcessor * create_native(
67 ProcessorPlugin *plugin,
68 RTI_RoutingServiceRoute *native_route,
69 const struct RTI_RoutingServiceProperties *native_properties,
70 RTI_RoutingServiceEnvironment *environment)
71 {
72 try {
73 using rti::routing::detail::ScopedForwarder;
74
75 std::map<std::string, std::string> properties;
76 rti::routing::PropertyAdapter::add_properties_from_native(
77 properties,
78 native_properties);
79
80 ProcessorForwarder *forwarder = new ProcessorForwarder(
81 native_route,
82 environment);
83 ScopedForwarder<ProcessorPlugin, ProcessorForwarder> scoped(
84 plugin,
85 forwarder,
86 environment);
87 forwarder->processor_ = plugin->create_processor(
88 forwarder->route(),
89 properties);
90 RTI_ROUTING_THROW_ON_NULL(forwarder->processor_);
91
92 scoped.release();
93
94 // TRUST-134: the forwarder is not leaked here. It might be leaked
95 // if delete_native() were not to be called, but Routing Service
96 // will make sure that is not the case
97
98 /* coverity[leaked_storage : FALSE] */
99 return forwarder->native();
100 } catch(const rti::apputils::FatalServiceException& ex) {
101 RTI_RoutingServiceEnvironment_fatal_error(
102 environment,
103 ex.what());
104 return NULL;
105 } catch(const std::exception& ex) {
106 RTI_RoutingServiceEnvironment_set_error(
107 environment,
108 ex.what());
109 return NULL;
110 } catch (...) {
111 RTI_RoutingServiceEnvironment_set_error(
112 environment,
113 "unexpected exception");
114 return NULL;
115 }
116
117
118 }
119
120 static void delete_native(
121 ProcessorPlugin *plugin,
122 RTI_RoutingServiceProcessor *native_processor,
123 RTI_RoutingServiceEnvironment *environment)
124 {
125 ProcessorForwarder *processor_forwarder =
126 static_cast<ProcessorForwarder*>(native_processor->processor_data);
127 try {
128 if (processor_forwarder->processor_ != NULL) {
129 plugin->delete_processor(
130 processor_forwarder->route_,
131 processor_forwarder->processor_);
132 processor_forwarder->processor_ = NULL;
133 }
134 } catch(const rti::apputils::FatalServiceException& ex) {
135 RTI_RoutingServiceEnvironment_fatal_error(
136 environment,
137 ex.what());
138 } catch(const std::exception& ex) {
139 RTI_RoutingServiceEnvironment_set_error(
140 environment,
141 ex.what());
142 } catch (...) {
143 RTI_RoutingServiceEnvironment_set_error(
144 environment,
145 "unexpected exception");
146 }
147
148 delete processor_forwarder;
149 }
150
151
152 static void forward_on_route_event(
153 void *native_processor_data,
154 RTI_RoutingServiceRouteEvent *native_route_event,
155 RTI_RoutingServiceEnvironment *environment)
156 {
157
158 ProcessorForwarder *forwarder =
159 static_cast<ProcessorForwarder*>(native_processor_data);
160
161 try {
162
163 // build up wrapper objects based on the event
164 switch (RTI_RoutingServiceRouteEvent_get_kind(native_route_event)) {
165
166 case RTI_ROUTING_SERVICE_ROUTE_EVENT_DATA_ON_INPUTS:
167 forwarder->processor_->on_data_available(forwarder->route());
168 break;
169
170 case RTI_ROUTING_SERVICE_ROUTE_EVENT_PERIODIC_ACTION:
171
172 forwarder->processor_->on_periodic_action(forwarder->route());
173 break;
174
175 case RTI_ROUTING_SERVICE_ROUTE_EVENT_INPUT_ENABLED:
176 {
177 void *affected_entity =
178 RTI_RoutingServiceRouteEvent_get_affected_entity(native_route_event);
179 void *index =
180 RTI_RoutingServiceRouteEvent_get_event_data(native_route_event);
181 /* ScopedPort helps to delete the Input if an exception occurs */
182 ScopedPort<Input, RTI_RoutingServiceInput> port(
183 forwarder->route_,
184 *(static_cast<int32_t*>(index)),
185 static_cast<RTI_RoutingServiceInput *>(affected_entity));
186 forwarder->processor_->on_input_enabled(
187 forwarder->route(),
188 *port.get());
189 RTI_RoutingServiceInput_set_user_data(
190 static_cast<RTI_RoutingServiceInput *>(affected_entity),
191 port.get());
192 /*
193 * Input successfully enabled, so clear the ScopedPort so it
194 * doesn't delete it
195 */
196 port.clear();
197
198 }
199 break;
200
201 case RTI_ROUTING_SERVICE_ROUTE_EVENT_INPUT_DISABLED:
202 {
203
204 void *affected_entity =
205 RTI_RoutingServiceRouteEvent_get_affected_entity(native_route_event);
206 RTI_RoutingServiceInput *native_input =
207 static_cast<RTI_RoutingServiceInput *>(affected_entity);
208 Input *input = reinterpret_cast<Input*>(
209 RTI_RoutingServiceInput_get_user_data(native_input));
210 forwarder->processor_->on_input_disabled(
211 forwarder->route(),
212 *input);
213 RTI_RoutingServiceInput_set_user_data(native_input, NULL);
214 delete input;
215 }
216 break;
217
218 case RTI_ROUTING_SERVICE_ROUTE_EVENT_OUTPUT_ENABLED:
219 {
220 void *affected_entity =
221 RTI_RoutingServiceRouteEvent_get_affected_entity(native_route_event);
222 void *index =
223 RTI_RoutingServiceRouteEvent_get_event_data(native_route_event);
224 /* ScopedPort helps to delete the Output if an exception occurs */
225 ScopedPort<Output, RTI_RoutingServiceOutput> port(
226 forwarder->route_,
227 *(static_cast<int32_t*>(index)),
228 static_cast<RTI_RoutingServiceOutput *>(affected_entity));
229 forwarder->processor_->on_output_enabled(
230 forwarder->route(),
231 *port.get());
232 RTI_RoutingServiceOutput_set_user_data(
233 static_cast<RTI_RoutingServiceOutput *>(affected_entity),
234 port.get());
235 /*
236 * Output successfully enabled, so clear the ScopedPort so it
237 * doesn't delete it
238 */
239 port.clear();
240
241 }
242 break;
243
244 case RTI_ROUTING_SERVICE_ROUTE_EVENT_OUTPUT_DISABLED:
245 {
246 void *affected_entity =
247 RTI_RoutingServiceRouteEvent_get_affected_entity(native_route_event);
248 RTI_RoutingServiceOutput *native_output =
249 static_cast<RTI_RoutingServiceOutput *>(affected_entity);
250 Output *output = reinterpret_cast<Output*>(
251 RTI_RoutingServiceOutput_get_user_data(native_output));
252 forwarder->processor_->on_output_disabled(
253 forwarder->route(),
254 *output);
255 RTI_RoutingServiceOutput_set_user_data(native_output, NULL);
256 delete output;
257 }
258 break;
259
260 case RTI_ROUTING_SERVICE_ROUTE_EVENT_ROUTE_STARTED:
261
262 forwarder->processor_->on_start(forwarder->route());
263 break;
264
265 case RTI_ROUTING_SERVICE_ROUTE_EVENT_ROUTE_STOPPED:
266
267 forwarder->processor_->on_stop(forwarder->route());
268 break;
269
270 case RTI_ROUTING_SERVICE_ROUTE_EVENT_ROUTE_RUNNING:
271
272 forwarder->processor_->on_run(forwarder->route());
273 break;
274
275 case RTI_ROUTING_SERVICE_ROUTE_EVENT_ROUTE_PAUSED:
276
277 forwarder->processor_->on_pause(forwarder->route());
278 break;
279
280 default:
281 break;
282 };
283
284 } catch (const rti::apputils::FatalServiceException& ex) {
285 RTI_RoutingServiceEnvironment_fatal_error(
286 environment,
287 ex.what());
288 } catch (const std::exception& ex) {
289 RTI_RoutingServiceEnvironment_set_error(
290 environment,
291 ex.what());
292 } catch (...) {
293 RTI_RoutingServiceEnvironment_set_error(
294 environment,
295 "%s",
296 "unexpected exception");
297 }
298 }
299
300 static void forward_update(
301 void *native_processor_data,
302 const struct RTI_RoutingServiceProperties *native_properties,
303 RTI_RoutingServiceEnvironment *environment)
304 {
305
306 ProcessorForwarder *processorForwarder =
307 static_cast<ProcessorForwarder*> (native_processor_data);
308
309 rti::routing::detail::UpdatableEntityForwarder::update(
310 processorForwarder->processor_,
311 native_properties,
312 environment);
313 }
314
315 RTI_RoutingServiceProcessor* native()
316 {
317 return &native_;
318 }
319
320private:
321
322 ProcessorForwarder(
323 RTI_RoutingServiceRoute *native_route,
324 RTI_RoutingServiceEnvironment *native_env) :
325 processor_(NULL),
326 route_(native_route, native_env)
327 {
328 RTIOsapiMemory_zero(&native_, sizeof(native_));
329 /* initialize native implementation */
330 native_.processor_data = static_cast<void *>(this);
331 native_.on_route_event = ProcessorForwarder::forward_on_route_event;
332 native_.update = ProcessorForwarder::forward_update;
333 }
334
335 rti::routing::processor::Route& route()
336 {
337 return route_;
338 }
339
340 ~ProcessorForwarder()
341 {
342 }
343
344private:
345 RTI_RoutingServiceProcessor native_;
346 Processor *processor_;
347 rti::routing::processor::Route route_;
348
349};
350
351}}}}
352
353#endif // RTI_ROUTING_PROCESSOR_DETAIL_PROCESSOR_FORWARDER_HPP_
rti::routing::processor::Route
Representation of the Route object that owns a Processor.
Definition: Route.hpp:84