IBR-DTNSuite  0.10
BundleMerger.cpp
Go to the documentation of this file.
1 /*
2  * BundleMerger.cpp
3  *
4  * Copyright (C) 2011 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 "ibrdtn/config.h"
23 #include "ibrdtn/data/BundleMerger.h"
24 #include "ibrdtn/data/Block.h"
25 #include "ibrdtn/data/PayloadBlock.h"
26 #include "ibrdtn/data/Exceptions.h"
27 #include <ibrcommon/thread/MutexLock.h>
28 #include <ibrcommon/Logger.h>
29 
30 namespace dtn
31 {
32  namespace data
33  {
34  BundleMerger::Container::Container(ibrcommon::BLOB::Reference &ref)
35  : _bundle(), _blob(ref), _initialized(false), _hasFirstFragBlocksAdded(false), _hasLastFragBlocksAdded(false), _appdatalength(0)
36  {
37  }
38 
39  BundleMerger::Container::~Container()
40  {
41 
42  }
43 
44  bool BundleMerger::Container::isComplete()
45  {
46  return Chunk::isComplete(_appdatalength, _chunks);
47  }
48 
49  Bundle& BundleMerger::Container::getBundle()
50  {
51  return _bundle;
52  }
53 
54  bool BundleMerger::Container::contains(Length offset, Length length) const
55  {
56  // check if the offered payload is already in the chunk list
57  for (std::set<Chunk>::const_iterator iter = _chunks.begin(); iter != _chunks.end(); ++iter)
58  {
59  const Chunk &chunk = (*iter);
60 
61  if (offset < chunk.offset) break;
62 
63  if ( (offset >= chunk.offset) && (offset < (chunk.offset + chunk.length)) )
64  {
65  if ((offset + length) <= (chunk.offset + chunk.length)) return true;
66  }
67  }
68 
69  return false;
70  }
71 
72  void BundleMerger::Container::add(Length offset, Length length)
73  {
74  BundleMerger::Chunk chunk(offset, length);
75  _chunks.insert(chunk);
76  }
77 
78  BundleMerger::Container &operator<<(BundleMerger::Container &c, dtn::data::Bundle &obj)
79  {
80  // check if the given bundle is a fragment
81  if (!obj.get(dtn::data::Bundle::FRAGMENT))
82  {
83  throw ibrcommon::Exception("This bundle is not a fragment!");
84  }
85 
86  if (c._initialized)
87  {
88  if ( (c._bundle.timestamp != obj.timestamp) ||
89  (c._bundle.sequencenumber != obj.sequencenumber) ||
90  (c._bundle.source != obj.source) )
91  throw ibrcommon::Exception("This fragment does not belongs to the others.");
92  }
93  else
94  {
95  // copy the bundle
96  c._bundle = obj;
97 
98  // store the app data length
99  c._appdatalength = obj.appdatalength.get<dtn::data::Length>();
100 
101  // remove all block of the copy
102  c._bundle.clear();
103 
104  // mark the copy as non-fragment
105  c._bundle.set(dtn::data::Bundle::FRAGMENT, false);
106 
107  // add a new payloadblock
108  c._bundle.push_back(c._blob);
109 
110  c._hasFirstFragBlocksAdded = false;
111  c._hasLastFragBlocksAdded = false;
112 
113  c._initialized = true;
114  }
115 
116  ibrcommon::BLOB::iostream stream = c._blob.iostream();
117  (*stream).seekp(obj.fragmentoffset.get<std::streampos>());
118 
119  dtn::data::PayloadBlock &p = obj.find<dtn::data::PayloadBlock>();
120  const Length plength = p.getLength();
121 
122  // skip write operation if chunk is already in the merged bundle
123  if (c.contains(obj.fragmentoffset.get<dtn::data::Length>(), plength)) return c;
124 
125  // copy payload of the fragment into the new blob
126  {
127  ibrcommon::BLOB::Reference ref = p.getBLOB();
128  ibrcommon::BLOB::iostream s = ref.iostream();
129  (*stream) << (*s).rdbuf() << std::flush;
130  }
131 
132  // add the chunk to the list of chunks
133  c.add(obj.fragmentoffset.get<dtn::data::Length>(), plength);
134 
135  // check if fragment is the first one
136  // add blocks only once
137  if (!c._hasFirstFragBlocksAdded && obj.fragmentoffset == 0)
138  {
139  c._hasFirstFragBlocksAdded = true;
140 
141  dtn::data::Bundle::iterator payload_it = obj.find(dtn::data::PayloadBlock::BLOCK_TYPE);
142 
143  // abort if the bundle do not contains a payload block
144  if (payload_it == obj.end()) throw ibrcommon::Exception("Payload block missing.");
145 
146  // iterate from begin to the payload block
147  for (dtn::data::Bundle::iterator block_it = obj.begin(); block_it != payload_it; ++block_it)
148  {
149  // get the current block and type
150  Block &current_block = (**block_it);
151  block_t block_type = current_block.getType();
152 
153  // search the position of the payload block
154  dtn::data::Bundle::iterator p = c._bundle.find(dtn::data::PayloadBlock::BLOCK_TYPE);
155  if (p == c._bundle.end()) throw ibrcommon::Exception("Payload block missing.");
156 
157  try
158  {
159  ExtensionBlock::Factory &f = dtn::data::ExtensionBlock::Factory::get(block_type);
160 
161  // insert new Block before payload block and copy block
162  dtn::data::Block &block = c._bundle.insert(p, f);
163  block = current_block;
164 
165  IBRCOMMON_LOGGER_DEBUG_TAG("BundleMerger", 5) << "Reassemble: Added Block before Payload: " << obj.toString() << " " << block_type << IBRCOMMON_LOGGER_ENDL;
166  }
167  catch(const ibrcommon::Exception &ex)
168  {
169  // insert new Block before payload block and copy block
170  dtn::data::Block &block = c._bundle.insert<dtn::data::ExtensionBlock>(p);
171  block = current_block;
172 
173  IBRCOMMON_LOGGER_DEBUG_TAG("BundleMerger", 5) << "Reassemble: Added Block before Payload: " << obj.toString() << " " << block_type << IBRCOMMON_LOGGER_ENDL;
174  }
175  }
176 
177  }
178 
179  //check if fragment is the last one
180  //add blocks only once
181  if(!c._hasLastFragBlocksAdded && obj.fragmentoffset + plength == obj.appdatalength)
182  {
183  c._hasLastFragBlocksAdded = true;
184 
185  // start to iterate after the payload block
186  dtn::data::Bundle::iterator payload_it = obj.find(dtn::data::PayloadBlock::BLOCK_TYPE);
187 
188  // abort if the bundle do not contains a payload block
189  if (payload_it == obj.end()) throw ibrcommon::Exception("Payload block missing.");
190 
191  // start with the block after the payload block
192  for (payload_it++; payload_it != obj.end(); ++payload_it)
193  {
194  //get the current block and type
195  Block &current_block = (**payload_it);
196  block_t block_type = current_block.getType();
197 
198  try
199  {
200  ExtensionBlock::Factory &f = dtn::data::ExtensionBlock::Factory::get(block_type);
201 
202  //push back new Block after payload block and copy block
203  dtn::data::Block &block = c._bundle.push_back(f);
204  block = current_block;
205 
206  IBRCOMMON_LOGGER_DEBUG_TAG("BundleMerger", 5) << "Reassemble: Added Block after Payload: " << obj.toString()<< " " << block_type << IBRCOMMON_LOGGER_ENDL;
207  }
208  catch(const ibrcommon::Exception &ex)
209  {
210  //push back new Block after payload block and copy block
211  dtn::data::Block &block = c._bundle.push_back<dtn::data::ExtensionBlock>();
212  block = current_block;
213 
214  IBRCOMMON_LOGGER_DEBUG_TAG("BundleMerger", 5) << "Reassemble: Added Block after Payload: " << obj.toString()<< " " << block_type << IBRCOMMON_LOGGER_ENDL;
215  }
216  }
217  }
218 
219  return c;
220  }
221 
222  BundleMerger::Container BundleMerger::getContainer()
223  {
224  ibrcommon::BLOB::Reference ref = ibrcommon::BLOB::create();
225  return Container(ref);
226  }
227 
228  BundleMerger::Container BundleMerger::getContainer(ibrcommon::BLOB::Reference &ref)
229  {
230  return Container(ref);
231  }
232 
233  BundleMerger::Chunk::Chunk(Length o, Length l)
234  : offset(o), length(l)
235  {
236  }
237 
238  BundleMerger::Chunk::~Chunk()
239  {
240  }
241 
242  bool BundleMerger::Chunk::operator<(const Chunk &other) const
243  {
244  if (offset < other.offset) return true;
245  if (offset != other.offset) return false;
246 
247  return (length < other.length);
248  }
249 
250  bool BundleMerger::Chunk::isComplete(Length length, const std::set<Chunk> &chunks)
251  {
252  // check if the bundle payload is complete
253  Length position = 0;
254 
255  for (std::set<Chunk>::const_iterator iter = chunks.begin(); iter != chunks.end(); ++iter)
256  {
257  const Chunk &chunk = (*iter);
258 
259  // if the next offset is too small, we do not got all fragments
260  if (chunk.offset > position) return false;
261 
262  position = chunk.offset + chunk.length;
263  }
264 
265  // return true, if we reached the application data length
266  return (position >= length);
267  }
268  }
269 }