RTI Routing Service Version 7.7.0
ConnectionForwarder.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_ADAPTER_DETAIL_CONNECTION_FORWARDER_HPP_
12#define RTI_ROUTING_ADAPTER_DETAIL_CONNECTION_FORWARDER_HPP_
13
14#include <rti/core/Exception.hpp>
15
16#include <rti/apputils/ServiceException.hpp>
17
18#include <osapi/osapi_heap.h>
19#include <routingservice/routingservice_adapter_new.h>
20
21#include <rti/routing/adapter/detail/SessionForwarder.hpp>
22#include <rti/routing/adapter/detail/StreamReaderForwarder.hpp>
23#include <rti/routing/adapter/detail/StreamWriterForwarder.hpp>
24#include <rti/routing/adapter/detail/DiscoveryStreamReaderForwarder.hpp>
25#include <rti/routing/detail/UpdatableEntityForwarder.hpp>
26#include <rti/routing/detail/ForwarderUtils.hpp>
27
28namespace rti { namespace routing { namespace adapter { namespace detail {
29
30class ConnectionForwarder {
31public:
32 static RTI_RoutingServiceConnectionExt * create_native(
33 AdapterPlugin *adapter,
34 const char *,
35 const char *,
36 const RTI_RoutingServiceStreamReaderListenerExt *native_output_stream_listener,
37 const RTI_RoutingServiceStreamReaderListenerExt *native_input_stream_listener,
38 const RTI_RoutingServiceTypeInfo **,
39 int,
40 const RTI_RoutingServiceProperties *native_properties,
41 RTI_RoutingServiceEnvironment *environment)
42 {
43 try {
44 using rti::routing::detail::ScopedForwarder;
45
46 // Set properties
47 std::map<std::string, std::string> properties;
48 rti::routing::PropertyAdapter::add_properties_from_native(
49 properties,
50 native_properties);
51
52 ConnectionForwarder *forwarder = new ConnectionForwarder(
53 native_output_stream_listener,
54 native_input_stream_listener);
55 ScopedForwarder<AdapterPlugin, ConnectionForwarder> scoped(
56 adapter,
57 forwarder,
58 environment);
59 // Create connection
60 forwarder->connection_ = adapter->create_connection(
61 &forwarder->input_discovery_listener_,
62 &forwarder->output_discovery_listener_,
63 properties);
64 RTI_ROUTING_THROW_ON_NULL(forwarder->connection_);
65
66 // Create built-in SRs
67 forwarder->input_stream_discovery_forwarder_ =
68 new DiscoveryStreamReaderForwarder(
69 forwarder->connection_
70 ->input_stream_discovery_reader());
71
72 forwarder->output_stream_discovery_forwarder_ =
73 new DiscoveryStreamReaderForwarder(
74 forwarder->connection_
75 ->output_stream_discovery_reader());
76
77 scoped.release();
78
79 // TRUST-134: the forwarder is not leaked here. It might be leaked
80 // if delete_native() were not to be called, but Routing Service
81 // will make sure that is not the case
82
83 /* coverity[leaked_storage : FALSE] */
84 return forwarder->native();
85 } catch (const rti::apputils::FatalServiceException& ex) {
86 RTI_RoutingServiceEnvironment_fatal_error(
87 environment,
88 ex.what());
89 return NULL;
90 } catch (const std::exception& ex) {
91 RTI_RoutingServiceEnvironment_set_error(
92 environment,
93 ex.what());
94 return NULL;
95 } catch (...) {
96 RTI_RoutingServiceEnvironment_set_error(
97 environment,
98 "unexpected exception");
99 return NULL;
100 }
101 }
102
103 static void delete_native(
104 AdapterPlugin *adapter,
105 RTI_RoutingServiceConnectionExt *native_connection,
106 RTI_RoutingServiceEnvironment *environment)
107 {
108 ConnectionForwarder *forwarder = static_cast<ConnectionForwarder*>(
109 native_connection->connection_data);
110 try {
111 // delete built-in SRs
112 delete forwarder->input_stream_discovery_forwarder_;
113 delete forwarder->output_stream_discovery_forwarder_;
114
115 // delete connection
116 if (forwarder->connection_ != NULL) {
117 adapter->delete_connection(forwarder->connection_);
118 forwarder->connection_ = NULL;
119 }
120 } catch(const rti::apputils::FatalServiceException& ex) {
121 RTI_RoutingServiceEnvironment_fatal_error(
122 environment,
123 ex.what());
124 } catch(const std::exception& ex) {
125 RTI_RoutingServiceEnvironment_set_error(
126 environment,
127 ex.what());
128 } catch (...) {
129 }
130
131 delete forwarder;
132 }
133
134
135 static RTI_RoutingServiceStreamReaderExt *
136 get_input_stream_discovery_reader(
137 void *native_connection_data,
138 RTI_RoutingServiceEnvironment *)
139 {
140 ConnectionForwarder *forwarder = static_cast<ConnectionForwarder*>(
141 native_connection_data);
142 return forwarder->input_stream_discovery_forwarder_->native();
143 }
144
145 static RTI_RoutingServiceStreamReaderExt *
146 get_output_stream_discovery_reader(
147 void *native_connection_data,
148 RTI_RoutingServiceEnvironment *)
149 {
150 ConnectionForwarder *forwarder = static_cast<ConnectionForwarder*>(
151 native_connection_data);
152 return forwarder->output_stream_discovery_forwarder_->native();
153 }
154
155 static RTI_RoutingServiceSessionExt * create_session(
156 void *native_connection_data,
157 const struct RTI_RoutingServiceProperties * native_properties,
158 RTI_RoutingServiceEnvironment *environment)
159 {
160
161 ConnectionForwarder *forwarder = static_cast<ConnectionForwarder*>(
162 native_connection_data);
163
164 return SessionForwarder::create_native(
165 forwarder->connection_,
166 native_properties,
167 environment);
168 }
169
170 static void delete_session(
171 void *native_connection_data,
172 RTI_RoutingServiceSessionExt *native_session,
173 RTI_RoutingServiceEnvironment *environment)
174 {
175 ConnectionForwarder *forwarder = static_cast<ConnectionForwarder*> (
176 native_connection_data);
177
178 SessionForwarder::delete_native(
179 forwarder->connection_,
180 native_session,
181 environment);
182 }
183
184 static RTI_RoutingServiceStreamReaderExt* create_stream_reader(
185 void *native_connection_data,
186 void *native_session_data,
187 const struct RTI_RoutingServiceStreamInfo *native_stream_info,
188 const struct RTI_RoutingServiceProperties *native_properties,
189 const struct RTI_RoutingServiceStreamReaderListenerExt *native_listener,
190 RTI_RoutingServiceEnvironment *environment)
191 {
192 ConnectionForwarder *forwarder = static_cast<ConnectionForwarder*>(
193 native_connection_data);
194 SessionForwarder *session_forwarder = static_cast<SessionForwarder*>(
195 native_session_data);
196
197 return StreamReaderForwarder::create_native(
198 forwarder->connection_,
199 session_forwarder->session(),
200 native_stream_info,
201 native_properties,
202 native_listener,
203 environment);
204 }
205
206 static void delete_stream_reader(
207 void *native_connection_data,
208 RTI_RoutingServiceStreamReaderExt *native_reader,
209 RTI_RoutingServiceEnvironment *environment)
210 {
211 ConnectionForwarder *forwarder = static_cast<ConnectionForwarder*>(
212 native_connection_data);
213 StreamReaderForwarder::delete_native(
214 forwarder->connection_,
215 native_reader,
216 environment);
217 }
218
219 static RTI_RoutingServiceStreamWriterExt* create_stream_writer(
220 void *native_connection_data,
221 void *native_session_data,
222 const struct RTI_RoutingServiceStreamInfo *native_stream_info,
223 const struct RTI_RoutingServiceProperties *native_properties,
224 const struct RTI_RoutingServiceStreamWriterListenerExt *,
225 RTI_RoutingServiceEnvironment *environment)
226 {
227 ConnectionForwarder *forwarder = static_cast<ConnectionForwarder*>(
228 native_connection_data);
229 SessionForwarder *session_forwarder = static_cast<SessionForwarder*>(
230 native_session_data);
231
232 return StreamWriterForwarder::create_native(
233 forwarder->connection_,
234 session_forwarder->session(),
235 native_stream_info,
236 native_properties,
237 environment);
238 }
239
240 static void delete_stream_writer(
241 void *native_connection_data,
242 RTI_RoutingServiceStreamWriterExt *native_writer,
243 RTI_RoutingServiceEnvironment *environment)
244 {
245 ConnectionForwarder *forwarder = static_cast<ConnectionForwarder*>(
246 native_connection_data);
247 StreamWriterForwarder::delete_native(
248 forwarder->connection_,
249 native_writer,
250 environment);
251 }
252
253 static void update(
254 void *native_connection_data,
255 const struct RTI_RoutingServiceProperties *native_properties,
256 RTI_RoutingServiceEnvironment *environment)
257 {
258 ConnectionForwarder *forwarder =
259 static_cast<ConnectionForwarder*> (native_connection_data);
260
261 rti::routing::detail::UpdatableEntityForwarder::update(
262 forwarder->connection_,
263 native_properties,
264 environment);
265 }
266
267
268 Connection * connection()
269 {
270 return connection_;
271 }
272
273 RTI_RoutingServiceConnectionExt* native()
274 {
275 return &this->native_;
276 }
277
278 RTI_RoutingServiceStreamReaderExt* native_input_discovery_reader()
279 {
280 return input_stream_discovery_forwarder_->native();
281 }
282
283 RTI_RoutingServiceStreamReaderExt* native_output_discovery_reader()
284 {
285 return output_stream_discovery_forwarder_->native();
286 }
287
288private:
289
290 ConnectionForwarder(
291 const RTI_RoutingServiceStreamReaderListenerExt *native_output_stream_listener,
292 const RTI_RoutingServiceStreamReaderListenerExt *native_input_stream_listener) :
293 connection_(NULL),
294 output_stream_discovery_forwarder_(NULL),
295 output_discovery_listener_(native_output_stream_listener),
296 input_stream_discovery_forwarder_(NULL),
297 input_discovery_listener_(native_input_stream_listener)
298 {
299 RTIOsapiMemory_zero(&native_, sizeof(native_));
300
301 native_.connection_data = static_cast<void*>(this);
302 native_.create_session = ConnectionForwarder::create_session;
303 native_.delete_session = ConnectionForwarder::delete_session;
304 native_.create_stream_reader =
305 ConnectionForwarder::create_stream_reader;
306 native_.delete_stream_reader =
307 ConnectionForwarder::delete_stream_reader;
308 native_.create_stream_writer =
309 ConnectionForwarder::create_stream_writer;
310 native_.delete_stream_writer =
311 ConnectionForwarder::delete_stream_writer;
312 native_.get_input_stream_discovery_reader =
313 ConnectionForwarder::get_input_stream_discovery_reader;
314 native_.get_output_stream_discovery_reader =
315 ConnectionForwarder::get_output_stream_discovery_reader;
316 native_.update = ConnectionForwarder::update;
317 }
318
319 ~ConnectionForwarder()
320 {
321 }
322
323private:
324 RTI_RoutingServiceConnectionExt native_;
325 Connection *connection_;
326 DiscoveryStreamReaderForwarder *output_stream_discovery_forwarder_;
327 StreamReaderListener output_discovery_listener_;
328 DiscoveryStreamReaderForwarder *input_stream_discovery_forwarder_;
329 StreamReaderListener input_discovery_listener_;
330
331};
332
333}}}}
334
335#endif // RTI_ROUTING_ADAPTER_DETAIL_CONNECTION_FORWARDER_HPP_
StreamReaderListener
Listener representation used by StreamReader to notify RTI Routing Service when new data is available...