00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016
00017 #include <oasys/debug/Log.h>
00018
00019 #include "BlockProcessor.h"
00020 #include "BlockInfo.h"
00021 #include "Bundle.h"
00022 #include "SDNV.h"
00023
00024 namespace dtn {
00025
00026
00027 BlockProcessor::BlockProcessor(int block_type)
00028 : block_type_(block_type)
00029 {
00030 }
00031
00032
00033 BlockProcessor::~BlockProcessor()
00034 {
00035 }
00036
00037
00038 int
00039 BlockProcessor::consume_preamble(BlockInfo* block,
00040 u_char* buf,
00041 size_t len,
00042 size_t preamble_size)
00043 {
00044 static const char* log = "/dtn/bundle/protocol";
00045 ASSERT(! block->complete());
00046 ASSERT(block->data_offset() == 0);
00047
00048 if (preamble_size == 0) {
00049 preamble_size = sizeof(BundleProtocol::BlockPreamble);
00050 }
00051
00052
00053
00054
00055
00056 size_t max_preamble = preamble_size + SDNV::MAX_LENGTH;
00057 size_t prev_consumed = block->contents().len();
00058 size_t tocopy = std::min(len, max_preamble - prev_consumed);
00059
00060 ASSERT(max_preamble > prev_consumed);
00061
00062
00063
00064 BlockInfo::DataBuffer* contents = block->writable_contents();
00065 ASSERT(contents->nfree() >= tocopy);
00066 memcpy(contents->end(), buf, tocopy);
00067 contents->set_len(contents->len() + tocopy);
00068
00069
00070
00071 if (contents->len() < preamble_size + 1) {
00072 ASSERT(tocopy == len);
00073 return len;
00074 }
00075
00076
00077
00078
00079 u_int64_t block_len;
00080 int sdnv_len =
00081 SDNV::decode(contents->buf() + preamble_size,
00082 contents->len() - preamble_size,
00083 &block_len);
00084 if (sdnv_len == -1) {
00085 ASSERT(tocopy == len);
00086 return len;
00087 }
00088
00089 if (block_len > 0xFFFFFFFFLL) {
00090
00091 log_err_p(log, "overflow in SDNV value for block type 0x%x",
00092 *contents->buf());
00093 return -1;
00094 }
00095
00096
00097
00098
00099
00100 block->set_data_length(static_cast<u_int32_t>(block_len));
00101 block->set_data_offset(preamble_size + sdnv_len);
00102 contents->set_len(preamble_size + sdnv_len);
00103
00104 log_debug_p(log, "BlockProcessor type 0x%x "
00105 "consumed preamble %zu/%u for block type (0x%x): "
00106 "data_offset %u data_length %u",
00107 block_type(), preamble_size + sdnv_len - prev_consumed,
00108 block->full_length(), block->type(),
00109 block->data_offset(), block->data_length());
00110
00111
00112
00113 ASSERT(preamble_size + sdnv_len > prev_consumed);
00114 return preamble_size + sdnv_len - prev_consumed;
00115 }
00116
00117
00118 void
00119 BlockProcessor::generate_preamble(BlockInfo* block,
00120 u_int8_t type,
00121 u_int8_t flags,
00122 size_t data_length)
00123 {
00124 static const char* log = "/dtn/bundle/protocol";
00125 (void)log;
00126
00127 size_t sdnv_len = SDNV::encoding_len(data_length);
00128 ASSERT(block->contents().len() == 0);
00129 ASSERT(block->contents().buf_len() >=
00130 sizeof(BundleProtocol::BlockPreamble) + sdnv_len);
00131
00132 BundleProtocol::BlockPreamble* bp =
00133 (BundleProtocol::BlockPreamble*)block->writable_contents()->buf();
00134
00135 bp->type = type;
00136 bp->flags = flags;
00137 SDNV::encode(data_length, &bp->length[0], sdnv_len);
00138
00139 block->set_data_length(data_length);
00140 block->set_data_offset(sizeof(*bp) + sdnv_len);
00141 block->writable_contents()->set_len(sizeof(*bp) + sdnv_len);
00142
00143 log_debug_p(log, "BlockProcessor type 0x%x "
00144 "generated preamble for block type 0x%x flags 0x%x: "
00145 "data_offset %u data_length %u",
00146 block_type(), block->type(), block->flags(),
00147 block->data_offset(), block->data_length());
00148 }
00149
00150
00151 int
00152 BlockProcessor::consume(Bundle* bundle, BlockInfo* block,
00153 u_char* buf, size_t len)
00154 {
00155 (void)bundle;
00156
00157 static const char* log = "/dtn/bundle/protocol";
00158 (void)log;
00159
00160 size_t consumed = 0;
00161
00162 ASSERT(! block->complete());
00163
00164
00165
00166
00167 if (block->data_offset() == 0) {
00168 int cc = consume_preamble(block, buf, len);
00169 if (cc == -1) {
00170 return -1;
00171 }
00172
00173 buf += cc;
00174 len -= cc;
00175
00176 consumed += cc;
00177 }
00178
00179
00180
00181 if (block->data_offset() == 0) {
00182 ASSERT(len == 0);
00183 }
00184
00185
00186
00187 if (block->data_offset() != 0 && block->data_length() == 0) {
00188 block->set_complete(true);
00189 }
00190
00191
00192 if (len == 0)
00193 return consumed;
00194
00195
00196
00197 ASSERT(block->data_length() == 0 ||
00198 block->full_length() > block->contents().len());
00199
00200
00201 block->writable_contents()->reserve(block->full_length());
00202
00203 size_t rcvd = block->contents().len();
00204 size_t remainder = block->full_length() - rcvd;
00205 size_t tocopy;
00206
00207 if (len >= remainder) {
00208 block->set_complete(true);
00209 tocopy = remainder;
00210 } else {
00211 tocopy = len;
00212 }
00213
00214
00215 memcpy(block->writable_contents()->end(), buf, tocopy);
00216 block->writable_contents()->set_len(rcvd + tocopy);
00217 len -= tocopy;
00218 consumed += tocopy;
00219
00220 log_debug_p(log, "BlockProcessor type 0x%x "
00221 "consumed %zu/%u for block type 0x%x (%s)",
00222 block_type(), consumed, block->full_length(), block->type(),
00223 block->complete() ? "complete" : "not complete");
00224
00225 return consumed;
00226 }
00227
00228
00229 bool
00230 BlockProcessor::validate(const Bundle* bundle, BlockInfo* block,
00231 BundleProtocol::status_report_reason_t* reception_reason,
00232 BundleProtocol::status_report_reason_t* deletion_reason)
00233 {
00234 static const char * log = "/dtn/bundle/protocol";
00235 (void)reception_reason;
00236
00237
00238
00239
00240 if (bundle->is_admin_ &&
00241 !block->primary_block() &&
00242 block->flags() & BundleProtocol::BLOCK_FLAG_REPORT_ONERROR) {
00243 log_err_p(log, "invalid block flag 0x%x for received admin bundle",
00244 BundleProtocol::BLOCK_FLAG_REPORT_ONERROR);
00245 *deletion_reason = BundleProtocol::REASON_BLOCK_UNINTELLIGIBLE;
00246 return false;
00247 }
00248
00249 return true;
00250 }
00251
00252
00253 void
00254 BlockProcessor::prepare(const Bundle* bundle,
00255 Link* link,
00256 BlockInfoVec* blocks,
00257 const BlockInfo* source)
00258 {
00259 (void)bundle;
00260 (void)link;
00261 blocks->push_back(BlockInfo(this, source));
00262 }
00263
00264
00265 void
00266 BlockProcessor::finalize(const Bundle* bundle, Link* link, BlockInfo* block)
00267 {
00268 (void)link;
00269
00270 if (bundle->is_admin_ && !block->primary_block()) {
00271 ASSERT((block->flags() &
00272 BundleProtocol::BLOCK_FLAG_REPORT_ONERROR) == 0);
00273 }
00274 }
00275
00276
00277 void
00278 BlockProcessor::produce(const Bundle* bundle, const BlockInfo* block,
00279 u_char* buf, size_t offset, size_t len)
00280 {
00281 (void)bundle;
00282 ASSERT(offset < block->contents().len());
00283 ASSERT(block->contents().len() >= offset + len);
00284 memcpy(buf, block->contents().buf() + offset, len);
00285 }
00286
00287
00288 void
00289 BlockProcessor::init_block(BlockInfo* block, u_int8_t type, u_int8_t flags,
00290 u_char* bp, size_t len)
00291 {
00292 ASSERT(block->owner() != NULL);
00293 generate_preamble(block, type, flags, len);
00294 ASSERT(block->data_offset() != 0);
00295 block->writable_contents()->reserve(block->full_length());
00296 block->writable_contents()->set_len(block->full_length());
00297 memcpy(block->writable_contents()->buf() + block->data_offset(),
00298 bp, len);
00299 }
00300
00301 }