diff --git a/core/src/Transport.cpp b/core/src/Transport.cpp index b8f1ed7..e31ae31 100644 --- a/core/src/Transport.cpp +++ b/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(messageId) << 32) | static_cast(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(index); + piece.fragmentCount = static_cast(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& 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(index); + piece.fragmentCount = static_cast(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); +