IBR-DTNSuite  0.10
NativeSession.cpp
Go to the documentation of this file.
1 /*
2  * NativeSession.cpp
3  *
4  * Copyright (C) 2013 IBR, TU Braunschweig
5  *
6  * Written-by: Johannes Morgenroth <morgenroth@ibr.cs.tu-bs.de>
7  *
8  * Licensed under the Apache License, Version 2.0 (the "License");
9  * you may not use this file except in compliance with the License.
10  * You may obtain a copy of the License at
11  *
12  * http://www.apache.org/licenses/LICENSE-2.0
13  *
14  * Unless required by applicable law or agreed to in writing, software
15  * distributed under the License is distributed on an "AS IS" BASIS,
16  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17  * See the License for the specific language governing permissions and
18  * limitations under the License.
19  *
20  */
21 
22 #include "api/NativeSession.h"
23 #include "api/NativeSerializer.h"
24 #include "core/BundleCore.h"
25 #include "core/EventDispatcher.h"
26 #include "routing/QueueBundleEvent.h"
27 
28 #include <ibrdtn/data/PayloadBlock.h>
29 #include <ibrcommon/data/BLOB.h>
30 #include <ibrcommon/Logger.h>
31 #include <ibrcommon/thread/RWLock.h>
32 #include <ibrcommon/thread/MutexLock.h>
33 
34 namespace dtn
35 {
36  namespace api
37  {
38  NativeSessionCallback::~NativeSessionCallback()
39  {
40  }
41 
42  const std::string NativeSession::TAG = "NativeSession";
43 
44  NativeSession::NativeSession(NativeSessionCallback *session_cb, NativeSerializerCallback *serializer_cb)
45  : _receiver(*this), _session_cb(session_cb), _serializer_cb(serializer_cb)
46  {
47  // set the local endpoint to the default
48  _endpoint = _registration.getDefaultEID();
49 
50  // listen to QueueBundleEvents
51  dtn::core::EventDispatcher<dtn::routing::QueueBundleEvent>::add(&_receiver);
52 
53  IBRCOMMON_LOGGER_DEBUG_TAG(NativeSession::TAG, 15) << "Session created" << IBRCOMMON_LOGGER_ENDL;
54  }
55 
56  NativeSession::~NativeSession()
57  {
58  // invalidate the callback pointer
59  {
60  ibrcommon::RWLock l(_cb_mutex, ibrcommon::RWMutex::LOCK_READWRITE);
61  _session_cb = NULL;
62  _serializer_cb = NULL;
63  }
64 
65  IBRCOMMON_LOGGER_DEBUG_TAG(NativeSession::TAG, 15) << "Session destroyed" << IBRCOMMON_LOGGER_ENDL;
66  }
67 
68  void NativeSession::destroy() throw ()
69  {
70  // un-listen from QueueBundleEvents
71  dtn::core::EventDispatcher<dtn::routing::QueueBundleEvent>::remove(&_receiver);
72 
73  _registration.abort();
74  }
75 
76  const dtn::data::EID& NativeSession::getNodeEID() const throw ()
77  {
78  return dtn::core::BundleCore::local;
79  }
80 
81  void NativeSession::fireNotificationBundle(const dtn::data::BundleID &id) throw ()
82  {
83  ibrcommon::RWLock l(_cb_mutex, ibrcommon::RWMutex::LOCK_READONLY);
84  if (_session_cb == NULL) return;
85  _session_cb->notifyBundle(id);
86  }
87 
88  void NativeSession::fireNotificationStatusReport(const dtn::data::EID &source, const dtn::data::StatusReportBlock &report) throw ()
89  {
90  ibrcommon::RWLock l(_cb_mutex, ibrcommon::RWMutex::LOCK_READONLY);
91  if (_session_cb == NULL) return;
92  _session_cb->notifyStatusReport(source, report);
93  }
94 
95  void NativeSession::fireNotificationCustodySignal(const dtn::data::EID &source, const dtn::data::CustodySignalBlock &custody) throw ()
96  {
97  ibrcommon::RWLock l(_cb_mutex, ibrcommon::RWMutex::LOCK_READONLY);
98  if (_session_cb == NULL) return;
99  _session_cb->notifyCustodySignal(source, custody);
100  }
101 
102  void NativeSession::setEndpoint(const std::string &suffix) throw (NativeSessionException)
103  {
104  const dtn::data::EID new_endpoint = dtn::core::BundleCore::local.add( dtn::core::BundleCore::local.getDelimiter() + suffix );
105 
106  // error checking
107  if (new_endpoint == dtn::data::EID())
108  {
109  throw NativeSessionException("given endpoint is not acceptable");
110  }
111  else
112  {
113  /* unsubscribe from the old endpoint and subscribe to the new one */
114  _registration.unsubscribe(_endpoint);
115  _registration.subscribe(new_endpoint);
116  _endpoint = new_endpoint;
117  }
118 
119  IBRCOMMON_LOGGER_DEBUG_TAG(NativeSession::TAG, 20) << "Endpoint set to " << _endpoint.getString() << IBRCOMMON_LOGGER_ENDL;
120  }
121 
122  void NativeSession::resetEndpoint() throw ()
123  {
124  _registration.unsubscribe(_endpoint);
125  _endpoint = _registration.getDefaultEID();
126  _registration.subscribe(_endpoint);
127 
128  IBRCOMMON_LOGGER_DEBUG_TAG(NativeSession::TAG, 20) << "Endpoint set to " << _endpoint.getString() << IBRCOMMON_LOGGER_ENDL;
129  }
130 
131  void NativeSession::addEndpoint(const std::string &suffix) throw (NativeSessionException)
132  {
133  const dtn::data::EID new_endpoint = dtn::core::BundleCore::local.add( dtn::core::BundleCore::local.getDelimiter() + suffix );
134 
135  // error checking
136  if (new_endpoint == dtn::data::EID())
137  {
138  throw NativeSessionException("given endpoint is not acceptable");
139  }
140  else
141  {
142  _registration.subscribe(new_endpoint);
143  }
144 
145  IBRCOMMON_LOGGER_DEBUG_TAG(NativeSession::TAG, 20) << "Endpoint " << suffix << " added" << IBRCOMMON_LOGGER_ENDL;
146  }
147 
148  void NativeSession::removeEndpoint(const std::string &suffix) throw (NativeSessionException)
149  {
150  const dtn::data::EID old_endpoint = dtn::core::BundleCore::local.add( dtn::core::BundleCore::local.getDelimiter() + suffix );
151 
152  // error checking
153  if (old_endpoint == dtn::data::EID())
154  {
155  throw NativeSessionException("given endpoint is not acceptable");
156  }
157  else
158  {
159  _registration.unsubscribe(old_endpoint);
160  }
161 
162  IBRCOMMON_LOGGER_DEBUG_TAG(NativeSession::TAG, 20) << "Endpoint " << suffix << " removed" << IBRCOMMON_LOGGER_ENDL;
163  }
164 
165  void NativeSession::addRegistration(const dtn::data::EID &eid) throw (NativeSessionException)
166  {
167  // error checking
168  if (eid == dtn::data::EID())
169  {
170  throw NativeSessionException("given endpoint is not acceptable");
171  }
172  else
173  {
174  _registration.subscribe(eid);
175  }
176 
177  IBRCOMMON_LOGGER_DEBUG_TAG(NativeSession::TAG, 20) << "Registration " << eid.getString() << " added" << IBRCOMMON_LOGGER_ENDL;
178  }
179 
180  void NativeSession::removeRegistration(const dtn::data::EID &eid) throw (NativeSessionException)
181  {
182  // error checking
183  if (eid == dtn::data::EID())
184  {
185  throw NativeSessionException("given endpoint is not acceptable");
186  }
187  else
188  {
189  _registration.unsubscribe(eid);
190  }
191 
192  IBRCOMMON_LOGGER_DEBUG_TAG(NativeSession::TAG, 20) << "Registration " << eid.getString() << " removed" << IBRCOMMON_LOGGER_ENDL;
193  }
194 
195  void NativeSession::clearRegistration() throw ()
196  {
197  resetEndpoint();
198 
199  const std::set<dtn::data::EID> subs = _registration.getSubscriptions();
200  for (std::set<dtn::data::EID>::const_iterator it = subs.begin(); it != subs.end(); ++it) {
201  const dtn::data::EID &e = (*it);
202  if (e != _endpoint) {
203  _registration.unsubscribe(e);
204  }
205  }
206 
207  IBRCOMMON_LOGGER_DEBUG_TAG(NativeSession::TAG, 20) << "Registrations cleared" << IBRCOMMON_LOGGER_ENDL;
208  }
209 
210  std::vector<std::string> NativeSession::getSubscriptions() throw ()
211  {
212  const std::set<dtn::data::EID> subs = _registration.getSubscriptions();
213  std::vector<std::string> ret;
214  for (std::set<dtn::data::EID>::const_iterator it = subs.begin(); it != subs.end(); ++it) {
215  ret.push_back((*it).getString());
216  }
217  return ret;
218  }
219 
220  void NativeSession::next(RegisterIndex ri) throw (BundleNotFoundException)
221  {
222  try {
223  const dtn::data::BundleID id = _bundle_queue.getnpop();
224 
225  IBRCOMMON_LOGGER_DEBUG_TAG(NativeSession::TAG, 20) << "Next bundle in queue is " << id.toString() << IBRCOMMON_LOGGER_ENDL;
226 
227  load(ri, id);
228  } catch (const ibrcommon::QueueUnblockedException &ex) {
229  IBRCOMMON_LOGGER_DEBUG_TAG(NativeSession::TAG, 15) << "No next bundle available" << IBRCOMMON_LOGGER_ENDL;
230  throw BundleNotFoundException();
231  }
232  }
233 
234  void NativeSession::load(RegisterIndex ri, const dtn::data::BundleID &id) throw (BundleNotFoundException)
235  {
236  // load the bundle
237  try {
238  _bundle[ri] = dtn::core::BundleCore::getInstance().getStorage().get(id);
239 
240  // process the bundle block (security, compression, ...)
241  dtn::core::BundleCore::processBlocks(_bundle[ri]);
242 
243  IBRCOMMON_LOGGER_DEBUG_TAG(NativeSession::TAG, 20) << "Bundle " << id.toString() << " loaded" << IBRCOMMON_LOGGER_ENDL;
244  } catch (const ibrcommon::Exception &ex) {
245  IBRCOMMON_LOGGER_DEBUG_TAG(NativeSession::TAG, 15) << "Failed to load bundle " << id.toString() << ", Exception: " << ex.what() << IBRCOMMON_LOGGER_ENDL;
246  throw BundleNotFoundException();
247  }
248  }
249 
250  void NativeSession::get(RegisterIndex ri) throw ()
251  {
252  ibrcommon::RWLock l(_cb_mutex, ibrcommon::RWMutex::LOCK_READONLY);
253  if (_serializer_cb == NULL) return;
254 
255  NativeSerializer serializer(*_serializer_cb, NativeSerializer::BUNDLE_FULL);
256  try {
257  serializer << _bundle[ri];
258  } catch (const ibrcommon::Exception &ex) {
259  IBRCOMMON_LOGGER_TAG(NativeSession::TAG, error) << "Get failed " << ex.what() << IBRCOMMON_LOGGER_ENDL;
260  }
261  }
262 
263  void NativeSession::getInfo(RegisterIndex ri) throw ()
264  {
265  ibrcommon::RWLock l(_cb_mutex, ibrcommon::RWMutex::LOCK_READONLY);
266  if (_serializer_cb == NULL) return;
267 
268  NativeSerializer serializer(*_serializer_cb, NativeSerializer::BUNDLE_INFO);
269  try {
270  serializer << _bundle[ri];
271  } catch (const ibrcommon::Exception &ex) {
272  IBRCOMMON_LOGGER_TAG(NativeSession::TAG, error) << "Get failed " << ex.what() << IBRCOMMON_LOGGER_ENDL;
273  }
274  }
275 
276  void NativeSession::free(RegisterIndex ri) throw (BundleNotFoundException)
277  {
278  try {
279  dtn::core::BundleCore::getInstance().getStorage().remove(_bundle[ri]);
280  _bundle[ri] = dtn::data::Bundle();
281  } catch (const ibrcommon::Exception&) {
282  throw BundleNotFoundException();
283  }
284  }
285 
286  void NativeSession::clear(RegisterIndex ri) throw ()
287  {
288  _bundle[ri] = dtn::data::Bundle();
289  }
290 
291  void NativeSession::delivered(const dtn::data::BundleID &id) const throw (BundleNotFoundException)
292  {
293  try {
294  // announce this bundle as delivered
295  dtn::data::MetaBundle meta = dtn::core::BundleCore::getInstance().getStorage().get(id);
296  _registration.delivered(meta);
297 
298  IBRCOMMON_LOGGER_DEBUG_TAG(NativeSession::TAG, 20) << "Bundle " << id.toString() << " marked as delivered" << IBRCOMMON_LOGGER_ENDL;
299  } catch (const ibrcommon::Exception&) {
300  throw BundleNotFoundException();
301  }
302  }
303 
304  dtn::data::BundleID NativeSession::send(RegisterIndex ri) throw ()
305  {
306  // create a new sequence number
307  _bundle[ri].relabel();
308 
309  // forward the bundle to the storage processing
310  dtn::api::Registration::processIncomingBundle(_endpoint, _bundle[ri]);
311 
312  IBRCOMMON_LOGGER_DEBUG_TAG(NativeSession::TAG, 20) << "Bundle " << _bundle[ri].toString() << " sent" << IBRCOMMON_LOGGER_ENDL;
313 
314  return _bundle[ri];
315  }
316 
317  void NativeSession::put(RegisterIndex ri, const dtn::data::Bundle &b) throw ()
318  {
319  // Copy the given bundle into the local register
320  _bundle[ri] = b;
321  }
322 
323  void NativeSession::put(RegisterIndex ri, const dtn::data::PrimaryBlock &p) throw ()
324  {
325  // clear all blocks in the register
326  _bundle[ri].clear();
327 
328  // Copy the given primary block into the local register
329  ((dtn::data::PrimaryBlock&)_bundle[ri]) = p;
330  }
331 
332  void NativeSession::write(RegisterIndex ri, const char *buf, const size_t len, const size_t offset) throw ()
333  {
334  try {
335  dtn::data::PayloadBlock &payload = _bundle[ri].find<dtn::data::PayloadBlock>();
336 
337  ibrcommon::BLOB::Reference ref = payload.getBLOB();
338  ibrcommon::BLOB::iostream stream = ref.iostream();
339 
340  std::streamsize stream_size = stream.size();
341 
342  if ((offset > 0) || (stream_size < static_cast<std::streamsize>(offset))) {
343  (*stream).seekp(0, std::ios_base::end);
344  } else {
345  (*stream).seekp(offset);
346  }
347 
348  (*stream).write(buf, len);
349  (*stream) << std::flush;
350  } catch (const dtn::data::Bundle::NoSuchBlockFoundException&) {
351  dtn::data::PayloadBlock &payload = _bundle[ri].push_back<dtn::data::PayloadBlock>();
352 
353  ibrcommon::BLOB::Reference ref = payload.getBLOB();
354  ibrcommon::BLOB::iostream stream = ref.iostream();
355 
356  std::streamsize stream_size = stream.size();
357 
358  if ((offset > 0) || (stream_size < static_cast<std::streamsize>(offset))) {
359  (*stream).seekp(0, std::ios_base::end);
360  } else {
361  (*stream).seekp(offset);
362  }
363  (*stream).write(buf, len);
364  (*stream) << std::flush;
365  }
366 
367  IBRCOMMON_LOGGER_DEBUG_TAG(NativeSession::TAG, 25) << len << " bytes added to the payload" << IBRCOMMON_LOGGER_ENDL;
368  }
369 
370  void NativeSession::read(RegisterIndex ri, char *buf, size_t &len, const size_t offset) throw ()
371  {
372  try {
373  dtn::data::PayloadBlock &payload = _bundle[ri].find<dtn::data::PayloadBlock>();
374 
375  ibrcommon::BLOB::Reference ref = payload.getBLOB();
376  ibrcommon::BLOB::iostream stream = ref.iostream();
377 
378  (*stream).seekg(offset);
379  (*stream).read(buf, len);
380 
381  len = (*stream).gcount();
382  } catch (const dtn::data::Bundle::NoSuchBlockFoundException&) {
383  IBRCOMMON_LOGGER_DEBUG_TAG(NativeSession::TAG, 25) << "no payload block available" << IBRCOMMON_LOGGER_ENDL;
384  len = 0;
385  }
386  }
387 
388  NativeSession::BundleReceiver::BundleReceiver(NativeSession &session)
389  : _session(session)
390  {
391  }
392 
393  NativeSession::BundleReceiver::~BundleReceiver()
394  {
395  }
396 
397  void NativeSession::BundleReceiver::raiseEvent(const dtn::core::Event *evt) throw ()
398  {
399  try {
400  const dtn::routing::QueueBundleEvent &queued = dynamic_cast<const dtn::routing::QueueBundleEvent&>(*evt);
401 
402  // ignore fragments - we can not deliver them directly to the client
403  if (queued.bundle.fragment) return;
404 
405  if (_session._registration.hasSubscribed(queued.bundle.destination))
406  {
407  _session._registration.notify(Registration::NOTIFY_BUNDLE_AVAILABLE);
408  }
409  } catch (const std::bad_cast&) { };
410  }
411 
412  void NativeSession::receive() throw (NativeSessionException)
413  {
414  Registration &reg = _registration;
415  try {
416  try {
417  const dtn::data::MetaBundle id = reg.receiveMetaBundle();
418 
419  if (id.procflags & dtn::data::PrimaryBlock::APPDATA_IS_ADMRECORD) {
420  // transform custody signals & status reports into notifies
421  fireNotificationAdministrativeRecord(id);
422 
423  // announce the delivery of this bundle
424  reg.delivered(id);
425  } else {
426  IBRCOMMON_LOGGER_DEBUG_TAG(NativeSession::TAG, 20) << "fire notification for new bundle " << id.toString() << IBRCOMMON_LOGGER_ENDL;
427 
428  // put the bundle into the API queue
429  _bundle_queue.push(id);
430 
431  // notify the client about the new bundle
432  fireNotificationBundle(id);
433  }
434  } catch (const dtn::storage::NoBundleFoundException&) {
435  IBRCOMMON_LOGGER_DEBUG_TAG(NativeSession::TAG, 25) << "no more bundles found - wait until we are notified" << IBRCOMMON_LOGGER_ENDL;
436  reg.wait_for_bundle();
437  }
438  } catch (const ibrcommon::QueueUnblockedException &ex) {
439  throw NativeSessionException(std::string("loop aborted - ") + ex.what());
440  } catch (const std::exception &ex) {
441  throw NativeSessionException(std::string("loop aborted - ") + ex.what());
442  }
443  }
444 
445  void NativeSession::fireNotificationAdministrativeRecord(const dtn::data::MetaBundle &bundle)
446  {
447  // load the whole bundle
448  const dtn::data::Bundle b = dtn::core::BundleCore::getInstance().getStorage().get(bundle);
449 
450  // get the payload block of the bundle
451  const dtn::data::PayloadBlock &payload = b.find<dtn::data::PayloadBlock>();
452 
453  try {
454  // try to decode as status report
455  dtn::data::StatusReportBlock report;
456  report.read(payload);
457 
458  IBRCOMMON_LOGGER_DEBUG_TAG(NativeSession::TAG, 20) << "fire notification for status report" << IBRCOMMON_LOGGER_ENDL;
459 
460  // fire the status report notification
461  fireNotificationStatusReport(b.source, report);
462  } catch (const dtn::data::StatusReportBlock::WrongRecordException&) {
463  // this is not a status report
464  }
465 
466  try {
467  // try to decode as custody signal
468  dtn::data::CustodySignalBlock custody;
469  custody.read(payload);
470 
471  IBRCOMMON_LOGGER_DEBUG_TAG(NativeSession::TAG, 20) << "fire notification for custody signal" << IBRCOMMON_LOGGER_ENDL;
472 
473  // fire the custody signal notification
474  fireNotificationCustodySignal(b.source, custody);
475  } catch (const dtn::data::CustodySignalBlock::WrongRecordException&) {
476  // this is not a custody report
477  }
478  }
479  } /* namespace net */
480 } /* namespace dtn */