#1 - quicr module
This commit is contained in:
Martin Slachta
@@ -0,0 +1,176 @@
|
||||
#include "message_protocol/MessageConnection.hpp"
|
||||
|
||||
#include "message_protocol/MessageDispatcher.hpp"
|
||||
#include "quicr/QuicrConnection.hpp"
|
||||
|
||||
#include <spdlog/spdlog.h>
|
||||
|
||||
#include <cstring>
|
||||
|
||||
namespace tw::msg {
|
||||
|
||||
namespace {
|
||||
constexpr size_t INITIAL_SEND_BUFFER_SIZE = 64 * 1024;
|
||||
}
|
||||
|
||||
MessageConnection::MessageConnection(PeerId peer_id,
|
||||
net::quicr::QuicrConnection* connection,
|
||||
MessageDispatcher* dispatcher) :
|
||||
m_connection(connection),
|
||||
m_dispatcher(dispatcher),
|
||||
m_peer_id(peer_id),
|
||||
m_send_buffer(INITIAL_SEND_BUFFER_SIZE) {
|
||||
}
|
||||
|
||||
bool MessageConnection::is_established() const {
|
||||
return m_connection->state() == net::quicr::QuicrConnectionState::Established;
|
||||
}
|
||||
|
||||
uint32_t MessageConnection::next_seq() {
|
||||
uint32_t seq = m_next_seq++;
|
||||
if(m_next_seq == MessageHeader::SEQ_NONE) {
|
||||
m_next_seq = 1;
|
||||
}
|
||||
|
||||
return seq;
|
||||
}
|
||||
|
||||
tl::expected<void, MessageError> MessageConnection::send_impl(MessageType type,
|
||||
std::span<const std::byte> body,
|
||||
uint32_t seq,
|
||||
bool reliable) {
|
||||
const size_t size = MessageHeader::SIZE + body.size();
|
||||
|
||||
if(m_send_buffer.size() < size) {
|
||||
m_send_buffer.resize(size);
|
||||
}
|
||||
|
||||
MessageHeader{ type, seq }.encode(m_send_buffer);
|
||||
std::memcpy(m_send_buffer.data() + MessageHeader::SIZE, body.data(), body.size());
|
||||
|
||||
auto send_r = m_connection->send_message(std::span(m_send_buffer).subspan(0, size), reliable);
|
||||
if(!send_r) {
|
||||
return tl::make_unexpected(MessageError(MessageErrorType::SendFailed, send_r.error().message()));
|
||||
}
|
||||
|
||||
m_bytes_sent += size;
|
||||
m_messages_sent++;
|
||||
|
||||
return {};
|
||||
}
|
||||
|
||||
tl::expected<void, MessageError> MessageConnection::send(MessageType type,
|
||||
std::span<const std::byte> body,
|
||||
bool reliable) {
|
||||
return send_impl(type, body, MessageHeader::SEQ_NONE, reliable);
|
||||
}
|
||||
|
||||
tl::expected<void, MessageError> MessageConnection::send_framed(std::span<const std::byte> message,
|
||||
bool reliable) {
|
||||
if(message.size() < MessageHeader::SIZE) {
|
||||
return tl::make_unexpected(
|
||||
MessageError(MessageErrorType::SendFailed, "the message is too short to hold a header"));
|
||||
}
|
||||
|
||||
// send_message takes a writable span, so the bytes are staged in the send
|
||||
// buffer rather than sent straight from the caller's buffer.
|
||||
if(m_send_buffer.size() < message.size()) {
|
||||
m_send_buffer.resize(message.size());
|
||||
}
|
||||
|
||||
std::memcpy(m_send_buffer.data(), message.data(), message.size());
|
||||
|
||||
auto send_r = m_connection->send_message(std::span(m_send_buffer).subspan(0, message.size()), reliable);
|
||||
if(!send_r) {
|
||||
return tl::make_unexpected(MessageError(MessageErrorType::SendFailed, send_r.error().message()));
|
||||
}
|
||||
|
||||
m_bytes_sent += message.size();
|
||||
m_messages_sent++;
|
||||
|
||||
return {};
|
||||
}
|
||||
|
||||
tl::expected<void, MessageError> MessageConnection::request(MessageType type,
|
||||
std::span<const std::byte> body,
|
||||
ReplyHandler on_reply,
|
||||
std::chrono::milliseconds timeout,
|
||||
std::function<void()> on_timeout,
|
||||
bool reliable) {
|
||||
const uint32_t seq = next_seq();
|
||||
|
||||
auto send_r = send_impl(type, body, seq, reliable);
|
||||
if(!send_r) {
|
||||
return send_r;
|
||||
}
|
||||
|
||||
m_pending.emplace(seq,
|
||||
PendingRequest{ std::move(on_reply),
|
||||
std::move(on_timeout),
|
||||
std::chrono::steady_clock::now() + timeout });
|
||||
|
||||
return {};
|
||||
}
|
||||
|
||||
size_t MessageConnection::receive(std::span<std::byte> scratch) {
|
||||
size_t total = 0;
|
||||
|
||||
while(true) {
|
||||
auto read_r = m_connection->read_into(scratch);
|
||||
if(!read_r) {
|
||||
spdlog::error("Failed to read from peer {}: {}", m_peer_id, read_r.error().message());
|
||||
break;
|
||||
}
|
||||
|
||||
if(*read_r == 0) {
|
||||
break;
|
||||
}
|
||||
|
||||
total += *read_r;
|
||||
m_messages_received++;
|
||||
on_message(scratch.subspan(0, *read_r));
|
||||
}
|
||||
|
||||
return total;
|
||||
}
|
||||
|
||||
void MessageConnection::on_message(std::span<const std::byte> message) {
|
||||
auto header = MessageHeader::decode(message);
|
||||
if(!header) {
|
||||
spdlog::warn("Dropped a message of {} bytes, too short to hold a header", message.size());
|
||||
return;
|
||||
}
|
||||
|
||||
auto body = message.subspan(MessageHeader::SIZE);
|
||||
|
||||
if(header->seq != MessageHeader::SEQ_NONE) {
|
||||
auto pending = m_pending.find(header->seq);
|
||||
if(pending != m_pending.end()) {
|
||||
auto on_reply = std::move(pending->second.on_reply);
|
||||
m_pending.erase(pending);
|
||||
on_reply(body);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
if(!m_dispatcher->dispatch(m_peer_id, header->type, body)) {
|
||||
spdlog::warn("No handler for message type {}", header->type);
|
||||
}
|
||||
}
|
||||
|
||||
void MessageConnection::expire_requests(std::chrono::steady_clock::time_point now) {
|
||||
std::erase_if(m_pending, [&](auto& entry) {
|
||||
if(entry.second.expires_at > now) {
|
||||
return false;
|
||||
}
|
||||
|
||||
spdlog::warn("Request {} timed out", entry.first);
|
||||
if(entry.second.on_timeout) {
|
||||
entry.second.on_timeout();
|
||||
}
|
||||
|
||||
return true;
|
||||
});
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,162 @@
|
||||
#include "message_protocol/MessageEndpoint.hpp"
|
||||
|
||||
#include "quicr/QuicrAddress.hpp"
|
||||
#include "quicr/QuicrConnection.hpp"
|
||||
#include "quicr/QuicrConnectionListener.hpp"
|
||||
#include "quicr/QuicrEndpoint.hpp"
|
||||
|
||||
#include <spdlog/spdlog.h>
|
||||
|
||||
namespace tw::msg {
|
||||
|
||||
namespace {
|
||||
constexpr size_t RECEIVE_BUFFER_SIZE = 64 * 1024;
|
||||
}
|
||||
|
||||
MessageEndpoint::MessageEndpoint(std::unique_ptr<net::quicr::QuicrEndpoint> endpoint) :
|
||||
m_endpoint(std::move(endpoint)),
|
||||
m_receive_buffer(RECEIVE_BUFFER_SIZE) {
|
||||
}
|
||||
|
||||
MessageEndpoint::~MessageEndpoint() = default;
|
||||
|
||||
tl::expected<std::unique_ptr<MessageEndpoint>, MessageError> MessageEndpoint::create() {
|
||||
auto endpoint_r = net::quicr::QuicrEndpoint::create();
|
||||
if(!endpoint_r) {
|
||||
return tl::make_unexpected(
|
||||
MessageError(MessageErrorType::BindFailed, endpoint_r.error().message()));
|
||||
}
|
||||
|
||||
return std::unique_ptr<MessageEndpoint>(new MessageEndpoint(std::move(endpoint_r.value())));
|
||||
}
|
||||
|
||||
tl::expected<std::unique_ptr<MessageEndpoint>, MessageError> MessageEndpoint::bind(int port) {
|
||||
auto endpoint_r = create();
|
||||
if(!endpoint_r) {
|
||||
return endpoint_r;
|
||||
}
|
||||
|
||||
auto& endpoint = endpoint_r.value();
|
||||
|
||||
auto bind_r = endpoint->m_endpoint->bind(port);
|
||||
if(!bind_r) {
|
||||
return tl::make_unexpected(MessageError(MessageErrorType::BindFailed, bind_r.error().message()));
|
||||
}
|
||||
|
||||
auto listener_r = net::quicr::QuicrConnectionListener::listen(endpoint->m_endpoint.get());
|
||||
if(!listener_r) {
|
||||
return tl::make_unexpected(
|
||||
MessageError(MessageErrorType::BindFailed, listener_r.error().message()));
|
||||
}
|
||||
|
||||
endpoint->m_listener = std::move(listener_r.value());
|
||||
|
||||
return endpoint_r;
|
||||
}
|
||||
|
||||
MessageConnection* MessageEndpoint::add_peer(net::quicr::QuicrConnection* connection) {
|
||||
const PeerId id = m_next_peer_id++;
|
||||
|
||||
auto peer = std::make_unique<MessageConnection>(id, connection, &m_dispatcher);
|
||||
auto* raw = peer.get();
|
||||
|
||||
m_peers.emplace(id, std::move(peer));
|
||||
|
||||
if(m_on_peer_connected) {
|
||||
m_on_peer_connected(id);
|
||||
}
|
||||
|
||||
return raw;
|
||||
}
|
||||
|
||||
tl::expected<MessageConnection*, MessageError> MessageEndpoint::connect(const std::string& host, int port) {
|
||||
auto connection_r = m_endpoint->connect(net::quicr::QuicrAddress(host, port));
|
||||
if(!connection_r) {
|
||||
return tl::make_unexpected(
|
||||
MessageError(MessageErrorType::ConnectFailed, connection_r.error().message()));
|
||||
}
|
||||
|
||||
return add_peer(connection_r.value());
|
||||
}
|
||||
|
||||
void MessageEndpoint::accept_peers() {
|
||||
if(!m_listener) {
|
||||
return;
|
||||
}
|
||||
|
||||
while(net::quicr::QuicrConnection* connection = m_listener->listen()) {
|
||||
add_peer(connection);
|
||||
}
|
||||
}
|
||||
|
||||
void MessageEndpoint::receive() {
|
||||
for(auto& [id, peer] : m_peers) {
|
||||
m_bytes_received += peer->receive(m_receive_buffer);
|
||||
}
|
||||
}
|
||||
|
||||
void MessageEndpoint::update() {
|
||||
m_endpoint->poll();
|
||||
|
||||
accept_peers();
|
||||
receive();
|
||||
|
||||
const auto now = std::chrono::steady_clock::now();
|
||||
for(auto& [id, peer] : m_peers) {
|
||||
peer->expire_requests(now);
|
||||
}
|
||||
}
|
||||
|
||||
MessageConnection* MessageEndpoint::peer(PeerId id) {
|
||||
auto peer = m_peers.find(id);
|
||||
return peer != m_peers.end() ? peer->second.get() : nullptr;
|
||||
}
|
||||
|
||||
std::vector<MessageConnection*> MessageEndpoint::peers() const {
|
||||
std::vector<MessageConnection*> result;
|
||||
result.reserve(m_peers.size());
|
||||
|
||||
for(const auto& [id, peer] : m_peers) {
|
||||
result.push_back(peer.get());
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
void MessageEndpoint::broadcast(MessageType type, std::span<const std::byte> body, bool reliable) {
|
||||
for(auto& [id, peer] : m_peers) {
|
||||
auto send_r = peer->send(type, body, reliable);
|
||||
if(!send_r) {
|
||||
spdlog::error("Failed to send to peer {}: {}", id, send_r.error().message());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
uint64_t MessageEndpoint::bytes_sent() const {
|
||||
uint64_t total = 0;
|
||||
for(const auto& [id, peer] : m_peers) {
|
||||
total += peer->bytes_sent();
|
||||
}
|
||||
|
||||
return total;
|
||||
}
|
||||
|
||||
uint64_t MessageEndpoint::messages_sent() const {
|
||||
uint64_t total = 0;
|
||||
for(const auto& [id, peer] : m_peers) {
|
||||
total += peer->messages_sent();
|
||||
}
|
||||
|
||||
return total;
|
||||
}
|
||||
|
||||
uint64_t MessageEndpoint::messages_received() const {
|
||||
uint64_t total = 0;
|
||||
for(const auto& [id, peer] : m_peers) {
|
||||
total += peer->messages_received();
|
||||
}
|
||||
|
||||
return total;
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user