11 #ifndef RTI_ROUTING_PROCESSOR_DETAIL_PROCESSOR_FORWARDER_HPP_
12 #define RTI_ROUTING_PROCESSOR_DETAIL_PROCESSOR_FORWARDER_HPP_
14 #include <rti/core/Exception.hpp>
16 #include "routingservice/routingservice_processor_impl.h"
17 #include <rti/routing/detail/ForwarderUtils.hpp>
18 #include <rti/routing/detail/UpdatableEntityForwarder.hpp>
19 #include <rti/routing/processor/Route.hpp>
20 #include <rti/routing/processor/Input.hpp>
21 #include <rti/routing/processor/Output.hpp>
23 namespace rti {
namespace routing {
namespace processor {
namespace detail {
25 class ProcessorForwarder {
29 template <
typename PORT,
typename NATIVE>
33 ScopedPort(Route& route, int32_t index, NATIVE *native)
35 port_(new PORT(native, index, route.native_route_, route.native_env_))
64 static RTI_RoutingServiceProcessor * create_native(
65 ProcessorPlugin *plugin,
66 RTI_RoutingServiceRoute *native_route,
67 const struct RTI_RoutingServiceProperties *native_properties,
68 RTI_RoutingServiceEnvironment *environment)
71 using rti::routing::detail::ScopedForwarder;
73 std::map<std::string, std::string> properties;
74 rti::routing::PropertyAdapter::add_properties_from_native(
78 ProcessorForwarder *forwarder =
new ProcessorForwarder(
81 ScopedForwarder<ProcessorPlugin, ProcessorForwarder> scoped(
85 forwarder->processor_ = plugin->create_processor(
88 RTI_ROUTING_THROW_ON_NULL(forwarder->processor_);
91 return forwarder->native();
92 }
catch(
const std::exception& ex) {
93 RTI_RoutingServiceEnvironment_set_error(
99 RTI_RoutingServiceEnvironment_set_error(
101 "unexpected exception");
108 static void delete_native(
109 ProcessorPlugin *plugin,
110 RTI_RoutingServiceProcessor *native_processor,
111 RTI_RoutingServiceEnvironment *environment)
113 ProcessorForwarder *processor_forwarder =
114 static_cast<ProcessorForwarder*
>(native_processor->processor_data);
116 if (processor_forwarder->processor_ != NULL) {
117 plugin->delete_processor(
118 processor_forwarder->route_,
119 processor_forwarder->processor_);
120 processor_forwarder->processor_ = NULL;
122 }
catch(
const std::exception& ex) {
123 RTI_RoutingServiceEnvironment_set_error(
128 RTI_RoutingServiceEnvironment_set_error(
130 "unexpected exception");
133 delete processor_forwarder;
137 static void forward_on_route_event(
138 void *native_processor_data,
139 RTI_RoutingServiceRouteEvent *native_route_event,
140 RTI_RoutingServiceEnvironment *environment)
143 ProcessorForwarder *forwarder =
144 static_cast<ProcessorForwarder*
>(native_processor_data);
149 switch (RTI_RoutingServiceRouteEvent_get_kind(native_route_event)) {
151 case RTI_ROUTING_SERVICE_ROUTE_EVENT_DATA_ON_INPUTS:
152 forwarder->processor_->on_data_available(forwarder->route());
155 case RTI_ROUTING_SERVICE_ROUTE_EVENT_PERIODIC_ACTION:
157 forwarder->processor_->on_periodic_action(forwarder->route());
160 case RTI_ROUTING_SERVICE_ROUTE_EVENT_INPUT_ENABLED:
162 void *affected_entity =
163 RTI_RoutingServiceRouteEvent_get_affected_entity(native_route_event);
165 RTI_RoutingServiceRouteEvent_get_event_data(native_route_event);
166 ScopedPort<Input, RTI_RoutingServiceStreamReaderExt> port(
168 *(static_cast<int32_t*>(index)),
169 static_cast<RTI_RoutingServiceStreamReaderExt *>(affected_entity));
170 forwarder->processor_->on_input_enabled(
173 RTI_RoutingServiceRoute_set_stream_port_user_data(
174 forwarder->route_.native_route_,
175 static_cast<RTI_RoutingServiceStreamReaderExt *>(affected_entity)->stream_reader_data,
182 case RTI_ROUTING_SERVICE_ROUTE_EVENT_INPUT_DISABLED:
185 void *affected_entity =
186 RTI_RoutingServiceRouteEvent_get_affected_entity(native_route_event);
187 RTI_RoutingServiceStreamReaderExt *native_input =
188 static_cast<RTI_RoutingServiceStreamReaderExt *
>(affected_entity);
189 Input *input =
reinterpret_cast<Input*
>(
190 RTI_RoutingServiceRoute_get_stream_port_user_data(
191 forwarder->route_.native_route_,
192 native_input->stream_reader_data));
193 forwarder->processor_->on_input_disabled(
196 RTI_RoutingServiceRoute_set_stream_port_user_data(
197 forwarder->route_.native_route_,
198 native_input->stream_reader_data,
204 case RTI_ROUTING_SERVICE_ROUTE_EVENT_OUTPUT_ENABLED:
206 void *affected_entity =
207 RTI_RoutingServiceRouteEvent_get_affected_entity(native_route_event);
209 RTI_RoutingServiceRouteEvent_get_event_data(native_route_event);
210 ScopedPort<Output, RTI_RoutingServiceStreamWriterExt> port(
212 *(static_cast<int32_t*>(index)),
213 static_cast<RTI_RoutingServiceStreamWriterExt *>(affected_entity));
214 forwarder->processor_->on_output_enabled(
217 RTI_RoutingServiceRoute_set_stream_port_user_data(
218 forwarder->route_.native_route_,
219 static_cast<RTI_RoutingServiceStreamWriterExt *>(affected_entity)->stream_writer_data,
226 case RTI_ROUTING_SERVICE_ROUTE_EVENT_OUTPUT_DISABLED:
228 void *affected_entity =
229 RTI_RoutingServiceRouteEvent_get_affected_entity(native_route_event);
230 RTI_RoutingServiceStreamWriterExt *native_output =
231 static_cast<RTI_RoutingServiceStreamWriterExt *
>(affected_entity);
232 Output *output =
reinterpret_cast<Output*
>(
233 RTI_RoutingServiceRoute_get_stream_port_user_data(
234 forwarder->route_.native_route_,
235 native_output->stream_writer_data));
236 forwarder->processor_->on_output_disabled(
239 RTI_RoutingServiceRoute_set_stream_port_user_data(
240 forwarder->route_.native_route_,
241 native_output->stream_writer_data,
247 case RTI_ROUTING_SERVICE_ROUTE_EVENT_ROUTE_STARTED:
249 forwarder->processor_->on_start(forwarder->route());
252 case RTI_ROUTING_SERVICE_ROUTE_EVENT_ROUTE_STOPPED:
254 forwarder->processor_->on_stop(forwarder->route());
257 case RTI_ROUTING_SERVICE_ROUTE_EVENT_ROUTE_RUNNING:
259 forwarder->processor_->on_run(forwarder->route());
262 case RTI_ROUTING_SERVICE_ROUTE_EVENT_ROUTE_PAUSED:
264 forwarder->processor_->on_pause(forwarder->route());
271 }
catch (
const std::exception& ex) {
272 RTI_RoutingServiceEnvironment_set_error(
277 RTI_RoutingServiceEnvironment_set_error(
280 "unexpected exception");
284 static void forward_update(
285 void *native_processor_data,
286 const struct RTI_RoutingServiceProperties *native_properties,
287 RTI_RoutingServiceEnvironment *environment)
290 ProcessorForwarder *processorForwarder =
291 static_cast<ProcessorForwarder*
> (native_processor_data);
293 rti::routing::detail::UpdatableEntityForwarder::update(
294 processorForwarder->processor_,
299 RTI_RoutingServiceProcessor* native()
307 RTI_RoutingServiceRoute *native_route,
308 RTI_RoutingServiceEnvironment *native_env) :
310 route_(native_route, native_env)
312 RTIOsapiMemory_zero(&native_,
sizeof(native_));
314 native_.processor_data =
315 static_cast<void*
>(
this);
316 native_.on_route_event =
317 ProcessorForwarder::forward_on_route_event;
319 ProcessorForwarder::forward_update;
327 ~ProcessorForwarder()
332 RTI_RoutingServiceProcessor native_;
333 Processor *processor_;
340 #endif // RTI_ROUTING_PROCESSOR_DETAIL_PROCESSOR_FORWARDER_HPP_