diff --git a/core/src/NetSession.cpp b/core/src/NetSession.cpp index 2b7a9ad..e2ed26d 100644 --- a/core/src/NetSession.cpp +++ b/core/src/NetSession.cpp @@ -482,3 +482,70 @@ void NetSession::flushBlockChanges() { dirtyLookup.clear(); } +void NetSession::serviceKeepAlive(uint64_t tick) { + if (currentPhase != SessionPhase::Play) return; + + if (tick - lastKeepAliveHeard >= config.keepAliveTimeoutTicks) { + keepAliveExpired = true; + disconnect("keep alive timeout"); + return; + } + if (role != Role::Server) return; + if (tick - lastKeepAliveSent < config.keepAliveIntervalTicks) return; + + KeepAlive beat; + beat.token = static_cast(tick); + PacketWriter body; + beat.encode(body); + enqueue(PacketId::KeepAlive, body, true); + pendingKeepAliveToken = beat.token; + lastKeepAliveSent = tick; + ++statistics.keepAlivesSent; +} + +void NetSession::flushOutbound() { + std::size_t budget = config.bandwidthBytesPerTick; + std::size_t used = 0; + while (!outbound.empty()) { + const QueuedPacket& head = outbound.front(); + // A packet bigger than the whole budget still goes out once nothing + // else has, otherwise it would block the queue forever. + if (budget != 0 && used != 0 && used + head.frame.size() > budget) break; + if (head.reliable) + transport->sendReliable(head.frame); + else + transport->sendUnreliable(head.frame); + used += head.frame.size(); + statistics.bytesFlushed += head.frame.size(); + ++statistics.packetsFlushed; + outbound.pop_front(); + if (budget != 0 && used >= budget) break; + } +} + +void NetSession::receive(uint64_t tick) { + transport->pump(tick); + ByteBuffer message; + while (transport->popReliable(message)) handleMessage(message, tick); + while (transport->popUnreliable(message)) handleMessage(message, tick); +} + +void NetSession::send(uint64_t tick) { + if (currentPhase == SessionPhase::Disconnected) return; + serviceKeepAlive(tick); + if (currentPhase == SessionPhase::Disconnected) return; + + if (role == Role::Server && currentPhase == SessionPhase::Play) { + streamChunks(); + flushBlockChanges(); + } + flushOutbound(); + transport->flush(tick); +} + +void NetSession::update(uint64_t tick) { + receive(tick); + send(tick); +} + +} // namespace vf