blob: df8a28536b6bc4c2d32a6b0939fc77e25a081041 [file] [log] [blame]
Junxiao Shid6dcd2c2014-02-16 14:49:54 -07001/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
2/**
3 * Copyright (C) 2014 Named Data Networking Project
4 * See COPYING for copyright and distribution information.
5 */
6
7#include "ndnlp-partial-message-store.hpp"
8
9namespace nfd {
10namespace ndnlp {
11
12PartialMessage::PartialMessage()
13 : m_fragCount(0)
14 , m_received(0)
15 , m_totalLength(0)
16{
17}
18
19bool
20PartialMessage::add(uint16_t fragIndex, uint16_t fragCount, const Block& payload)
21{
22 if (m_received == 0) { // first packet
23 m_fragCount = fragCount;
24 m_payloads.resize(fragCount);
25 }
26
27 if (m_fragCount != fragCount || fragIndex >= m_fragCount) {
28 return false;
29 }
30
31 if (!m_payloads[fragIndex].empty()) { // duplicate
32 return false;
33 }
34
35 m_payloads[fragIndex] = payload;
36 ++m_received;
37 m_totalLength += payload.value_size();
Junxiao Shic0d05912014-02-17 19:06:21 -070038 return true;
Junxiao Shid6dcd2c2014-02-16 14:49:54 -070039}
40
41bool
42PartialMessage::isComplete() const
43{
44 return m_received == m_fragCount;
45}
46
47Block
48PartialMessage::reassemble()
49{
50 BOOST_ASSERT(this->isComplete());
51
52 ndn::BufferPtr buffer = make_shared<ndn::Buffer>(m_totalLength);
53 uint8_t* buf = buffer->get();
54 for (std::vector<Block>::const_iterator it = m_payloads.begin();
55 it != m_payloads.end(); ++it) {
56 const Block& payload = *it;
57 memcpy(buf, payload.value(), payload.value_size());
58 buf += payload.value_size();
59 }
60
61 return Block(buffer);
62}
63
64PartialMessageStore::PartialMessageStore(Scheduler& scheduler, time::Duration idleDuration)
65 : m_scheduler(scheduler)
66 , m_idleDuration(idleDuration)
67{
68}
69
70PartialMessageStore::~PartialMessageStore()
71{
72}
73
74void
75PartialMessageStore::receiveNdnlpData(const Block& pkt)
76{
77 NdnlpData parsed;
78 parsed.wireDecode(pkt);
79 if (parsed.m_fragCount == 1) { // single fragment
80 this->onReceive(parsed.m_payload.blockFromValue());
81 return;
82 }
83
84 uint64_t messageIdentifier = parsed.m_seq - parsed.m_fragIndex;
85 shared_ptr<PartialMessage> pm = m_partialMessages[messageIdentifier];
86 if (!static_cast<bool>(pm)) {
87 m_partialMessages[messageIdentifier] = pm = make_shared<PartialMessage>();
88 }
89 this->scheduleCleanup(messageIdentifier, pm);
90
91 pm->add(parsed.m_fragIndex, parsed.m_fragCount, parsed.m_payload);
92 if (pm->isComplete()) {
93 this->onReceive(pm->reassemble());
94 this->cleanup(messageIdentifier);
95 }
96}
97
98void
99PartialMessageStore::scheduleCleanup(uint64_t messageIdentifier,
100 shared_ptr<PartialMessage> partialMessage)
101{
102 partialMessage->m_expiry = m_scheduler.scheduleEvent(m_idleDuration,
103 bind(&PartialMessageStore::cleanup, this, messageIdentifier));
104}
105
106void
107PartialMessageStore::cleanup(uint64_t messageIdentifier)
108{
109 std::map<uint64_t, shared_ptr<PartialMessage> >::iterator it =
110 m_partialMessages.find(messageIdentifier);
111 if (it == m_partialMessages.end()) {
112 return;
113 }
114
115 m_scheduler.cancelEvent(it->second->m_expiry);
116 m_partialMessages.erase(it);
117}
118
119} // namespace ndnlp
120} // namespace nfd