39 , m_linkService(linkService)
43 std::tuple<bool, Block, lp::Packet>
46 BOOST_ASSERT(packet.has<lp::FragmentField>());
48 static auto FALSE_RETURN = std::make_tuple(
false, Block(), lp::Packet());
51 uint64_t fragIndex = 0;
52 uint64_t fragCount = 1;
53 if (packet.has<lp::FragIndexField>()) {
54 fragIndex = packet.get<lp::FragIndexField>();
56 if (packet.has<lp::FragCountField>()) {
57 fragCount = packet.get<lp::FragCountField>();
60 if (fragIndex >= fragCount) {
71 if (fragIndex == 0 && fragCount == 1) {
72 ndn::Buffer::const_iterator fragBegin, fragEnd;
73 std::tie(fragBegin, fragEnd) = packet.get<lp::FragmentField>();
74 Block netPkt(&*fragBegin, std::distance(fragBegin, fragEnd));
75 return std::make_tuple(
true, netPkt, packet);
79 if (!packet.has<lp::SequenceField>()) {
83 lp::Sequence messageIdentifier = packet.get<lp::SequenceField>() - fragIndex;
84 Key key = std::make_tuple(remoteEndpoint, messageIdentifier);
87 PartialPacket& pp = m_partialPackets[key];
88 if (pp.fragCount == 0) {
89 pp.fragCount = fragCount;
90 pp.nReceivedFragments = 0;
91 pp.fragments.resize(fragCount);
94 if (fragCount != pp.fragCount) {
100 if (pp.fragments[fragIndex].has<lp::SequenceField>()) {
105 pp.fragments[fragIndex] = packet;
106 ++pp.nReceivedFragments;
109 if (pp.nReceivedFragments == pp.fragCount) {
110 Block reassembled = doReassembly(key);
111 lp::Packet firstFrag(std::move(pp.fragments[0]));
112 m_partialPackets.erase(key);
113 return std::make_tuple(
true, reassembled, firstFrag);
123 LpReassembler::doReassembly(
const Key& key)
125 PartialPacket& pp = m_partialPackets[key];
127 size_t payloadSize = std::accumulate(pp.fragments.begin(), pp.fragments.end(), 0U,
128 [&] (
size_t sum,
const lp::Packet& pkt) ->
size_t {
129 ndn::Buffer::const_iterator fragBegin, fragEnd;
130 std::tie(fragBegin, fragEnd) = pkt.get<lp::FragmentField>();
131 return sum + std::distance(fragBegin, fragEnd);
134 ndn::Buffer fragBuffer(payloadSize);
135 auto it = fragBuffer.begin();
137 for (
const lp::Packet& frag : pp.fragments) {
138 ndn::Buffer::const_iterator fragBegin, fragEnd;
139 std::tie(fragBegin, fragEnd) = frag.get<lp::FragmentField>();
140 it = std::copy(fragBegin, fragEnd, it);
143 return Block(&*(fragBuffer.cbegin()), std::distance(fragBuffer.cbegin(), fragBuffer.cend()));
147 LpReassembler::timeoutPartialPacket(
const Key& key)
149 auto it = m_partialPackets.find(key);
150 if (it == m_partialPackets.end()) {
154 this->
beforeTimeout(std::get<0>(key), it->second.nReceivedFragments);
155 m_partialPackets.erase(it);
159 operator<<(std::ostream& os, const FaceLogHelper<LpReassembler>& flh)
161 if (flh.obj.getLinkService() ==
nullptr) {
162 os <<
"[id=0,local=unknown,remote=unknown] ";
165 os << FaceLogHelper<LinkService>(*flh.obj.getLinkService());
#define NFD_LOG_FACE_TRACE(msg)
Log a message at TRACE level.
LpReassembler(const Options &options, const LinkService *linkService=nullptr)
std::tuple< bool, Block, lp::Packet > receiveFragment(EndpointId remoteEndpoint, const lp::Packet &packet)
adds received fragment to the buffer
uint64_t EndpointId
Identifies a remote endpoint on the link.
Scheduler & getScheduler()
Returns the global Scheduler instance for the calling thread.
#define NFD_LOG_FACE_WARN(msg)
Log a message at WARN level.
Copyright (c) 2014-2015, Regents of the University of California, Arizona Board of Regents...
Options that control the behavior of LpReassembler.
signal::Signal< LpReassembler, EndpointId, size_t > beforeTimeout
signals before a partial packet is dropped due to timeout
size_t nMaxFragments
maximum number of fragments in a packet
#define NFD_LOG_INIT(name)
time::nanoseconds reassemblyTimeout
timeout before a partially reassembled packet is dropped