feat: extend Transport

core/src/Transport.cpp

@@ -165,3 +165,161 @@ bool Datagram::decode(const uint8_t* data, std::size_t size) {
     return reader.ok();
 }
 
+uint64_t ReliableChannel::fragmentKey(uint32_t messageId, uint32_t fragmentIndex) {
+    return (static_cast<uint64_t>(messageId) << 32) | static_cast<uint64_t>(fragmentIndex);
+}
+
+void ReliableChannel::queueMessage(const uint8_t* payload, std::size_t size,
+                                   std::size_t fragmentPayload) {
+    uint32_t messageId = nextOutgoingMessageId++;
+    std::size_t total = size == 0 ? 1 : (size + fragmentPayload - 1) / fragmentPayload;
+    for (std::size_t index = 0; index < total; ++index) {
+        OutgoingFragment piece;
+        piece.messageId = messageId;
+        piece.fragmentIndex = static_cast<uint32_t>(index);
+        piece.fragmentCount = static_cast<uint32_t>(total);
+        std::size_t begin = index * fragmentPayload;
+        std::size_t end = std::min(size, begin + fragmentPayload);
+        piece.bytes.assign(payload + begin, payload + end);
+        outgoing.emplace(fragmentKey(messageId, piece.fragmentIndex), std::move(piece));
+    }
+}
+
+void ReliableChannel::collectReady(std::vector<uint64_t>& keys) const {
+    keys.clear();
+    for (const auto& entry : outgoing)
+        if (!entry.second.inFlight) keys.push_back(entry.first);
+}
+
+std::size_t ReliableChannel::readyFragmentCount() const {
+    std::size_t count = 0;
+    for (const auto& entry : outgoing)
+        if (!entry.second.inFlight) ++count;
+    return count;
+}
+
+const ReliableChannel::OutgoingFragment* ReliableChannel::fragment(uint64_t key) const {
+    auto it = outgoing.find(key);
+    return it == outgoing.end() ? nullptr : &it->second;
+}
+
+void ReliableChannel::markInFlight(uint64_t key) {
+    auto it = outgoing.find(key);
+    if (it != outgoing.end()) it->second.inFlight = true;
+}
+
+void ReliableChannel::markAcked(uint64_t key) { outgoing.erase(key); }
+
+void ReliableChannel::markLost(uint64_t key) {
+    auto it = outgoing.find(key);
+    if (it != outgoing.end()) it->second.inFlight = false;
+}
+
+bool ReliableChannel::accept(const Datagram::Fragment& piece) {
+    if (piece.messageId < nextDeliverMessageId) return false;
+    if (completed.count(piece.messageId) != 0) return false;
+
+    auto it = incoming.find(piece.messageId);
+    if (it == incoming.end()) {
+        Assembly assembly;
+        assembly.fragmentCount = piece.fragmentCount;
+        assembly.parts.resize(piece.fragmentCount);
+        assembly.have.assign(piece.fragmentCount, 0);
+        it = incoming.emplace(piece.messageId, std::move(assembly)).first;
+    }
+    Assembly& assembly = it->second;
+    if (piece.fragmentIndex >= assembly.fragmentCount) return false;
+    if (assembly.have[piece.fragmentIndex]) return false;
+
+    assembly.parts[piece.fragmentIndex] = piece.bytes;
+    assembly.have[piece.fragmentIndex] = 1;
+    ++assembly.received;
+
+    if (assembly.received == assembly.fragmentCount) {
+        ByteBuffer whole;
+        std::size_t total = 0;
+        for (const ByteBuffer& part : assembly.parts) total += part.size();
+        whole.reserve(total);
+        for (const ByteBuffer& part : assembly.parts)
+            whole.insert(whole.end(), part.begin(), part.end());
+        completed.emplace(piece.messageId, std::move(whole));
+        incoming.erase(it);
+        drainCompleted();
+    }
+    return true;
+}
+
+void ReliableChannel::drainCompleted() {
+    for (;;) {
+        auto it = completed.find(nextDeliverMessageId);
+        if (it == completed.end()) return;
+        delivered.push_back(std::move(it->second));
+        completed.erase(it);
+        ++nextDeliverMessageId;
+    }
+}
+
+bool ReliableChannel::popDelivered(ByteBuffer& out) {
+    if (delivered.empty()) return false;
+    out = std::move(delivered.front());
+    delivered.pop_front();
+    return true;
+}
+
+void UnreliableChannel::queueMessage(const uint8_t* payload, std::size_t size,
+                                     std::size_t fragmentPayload) {
+    uint32_t messageId = nextOutgoingMessageId++;
+    std::size_t total = size == 0 ? 1 : (size + fragmentPayload - 1) / fragmentPayload;
+    for (std::size_t index = 0; index < total; ++index) {
+        Datagram::Fragment piece;
+        piece.reliable = false;
+        piece.messageId = messageId;
+        piece.fragmentIndex = static_cast<uint32_t>(index);
+        piece.fragmentCount = static_cast<uint32_t>(total);
+        std::size_t begin = index * fragmentPayload;
+        std::size_t end = std::min(size, begin + fragmentPayload);
+        piece.bytes.assign(payload + begin, payload + end);
+        outgoing.push_back(std::move(piece));
+    }
+}
+
+const Datagram::Fragment* UnreliableChannel::peekOutgoing() const {
+    return outgoing.empty() ? nullptr : &outgoing.front();
+}
+
+void UnreliableChannel::dropOutgoing() { outgoing.pop_front(); }
+
+bool UnreliableChannel::accept(const Datagram::Fragment& piece) {
+    if (anyDelivered && piece.messageId <= highestDelivered) {
+        ++stale;
+        return false;
+    }
+
+    auto it = incoming.find(piece.messageId);
+    if (it == incoming.end()) {
+        Assembly assembly;
+        assembly.fragmentCount = piece.fragmentCount;
+        assembly.parts.resize(piece.fragmentCount);
+        assembly.have.assign(piece.fragmentCount, 0);
+        it = incoming.emplace(piece.messageId, std::move(assembly)).first;
+    }
+    Assembly& assembly = it->second;
+    if (piece.fragmentIndex >= assembly.fragmentCount) return false;
+    if (assembly.have[piece.fragmentIndex]) return false;
+
+    assembly.parts[piece.fragmentIndex] = piece.bytes;
+    assembly.have[piece.fragmentIndex] = 1;
+    ++assembly.received;
+    if (assembly.received != assembly.fragmentCount) return true;
+
+    ByteBuffer whole;
+    std::size_t total = 0;
+    for (const ByteBuffer& part : assembly.parts) total += part.size();
+    whole.reserve(total);
+    for (const ByteBuffer& part : assembly.parts)
+        whole.insert(whole.end(), part.begin(), part.end());
+    delivered.push_back(std::move(whole));
+    highestDelivered = piece.messageId;
+    anyDelivered = true;
+    incoming.erase(it);
+