38 , m_linkService(linkService)
42 std::tuple<bool, Block, lp::Packet>
45 BOOST_ASSERT(packet.has<lp::FragmentField>());
48 uint64_t fragIndex = 0;
49 uint64_t fragCount = 1;
50 if (packet.has<lp::FragIndexField>()) {
51 fragIndex = packet.get<lp::FragIndexField>();
53 if (packet.has<lp::FragCountField>()) {
54 fragCount = packet.get<lp::FragCountField>();
57 if (fragIndex >= fragCount) {
59 return {
false, {}, {}};
64 return {
false, {}, {}};
68 if (fragIndex == 0 && fragCount == 1) {
69 auto frag = packet.get<lp::FragmentField>();
70 Block netPkt({frag.first, frag.second});
71 return {
true, netPkt, packet};
75 if (!packet.has<lp::SequenceField>()) {
77 return {
false, {}, {}};
80 lp::Sequence messageIdentifier = packet.get<lp::SequenceField>() - fragIndex;
81 Key key(remoteEndpoint, messageIdentifier);
84 PartialPacket& pp = m_partialPackets[key];
85 if (pp.fragCount == 0) {
86 pp.fragCount = fragCount;
87 pp.nReceivedFragments = 0;
88 pp.fragments.resize(fragCount);
91 if (fragCount != pp.fragCount) {
93 return {
false, {}, {}};
97 if (pp.fragments[fragIndex].has<lp::SequenceField>()) {
99 return {
false, {}, {}};
102 pp.fragments[fragIndex] = packet;
103 ++pp.nReceivedFragments;
106 if (pp.nReceivedFragments == pp.fragCount) {
107 Block reassembled = doReassembly(key);
108 lp::Packet firstFrag(std::move(pp.fragments[0]));
109 m_partialPackets.erase(key);
110 return {
true, reassembled, firstFrag};
116 return {
false, {}, {}};
120 LpReassembler::doReassembly(
const Key& key)
122 PartialPacket& pp = m_partialPackets[key];
124 size_t payloadSize = std::accumulate(pp.fragments.begin(), pp.fragments.end(), 0U,
125 [&] (
size_t sum,
const lp::Packet& pkt) ->
size_t {
126 auto [fragBegin, fragEnd] = pkt.get<lp::FragmentField>();
127 return sum + std::distance(fragBegin, fragEnd);
130 ndn::Buffer fragBuffer(payloadSize);
131 auto it = fragBuffer.begin();
132 for (
const lp::Packet& frag : pp.fragments) {
133 auto [fragBegin, fragEnd] = frag.get<lp::FragmentField>();
134 it = std::copy(fragBegin, fragEnd, it);
136 return Block(fragBuffer);
140 LpReassembler::timeoutPartialPacket(
const Key& key)
142 auto it = m_partialPackets.find(key);
143 if (it == m_partialPackets.end()) {
147 this->
beforeTimeout(std::get<0>(key), it->second.nReceivedFragments);
148 m_partialPackets.erase(it);
154 if (flh.
obj.getLinkService() ==
nullptr) {
155 os <<
"[id=0,local=unknown,remote=unknown] ";
158 os << FaceLogHelper<LinkService>(*flh.
obj.getLinkService());
For internal use by FaceLogging macros.
The upper half of a Face.
std::tuple< bool, Block, lp::Packet > receiveFragment(const EndpointId &remoteEndpoint, const lp::Packet &packet)
Adds received fragment to the buffer.
signal::Signal< LpReassembler, EndpointId, size_t > beforeTimeout
Notifies before a partial packet is dropped due to timeout.
LpReassembler(const Options &options, const LinkService *linkService=nullptr)
#define NFD_LOG_FACE_WARN(msg)
Log a message at WARN level.
#define NFD_LOG_FACE_TRACE(msg)
Log a message at TRACE level.
#define NFD_LOG_INIT(name)
std::ostream & operator<<(std::ostream &os, const FaceLogHelper< Face > &flh)
std::variant< std::monostate, ethernet::Address, udp::Endpoint > EndpointId
Identifies a remote endpoint on the link.
Scheduler & getScheduler()
Returns the global Scheduler instance for the calling thread.
Options that control the behavior of LpReassembler.
time::nanoseconds reassemblyTimeout
Timeout before a partially reassembled packet is dropped.
size_t nMaxFragments
Maximum number of fragments in a packet.