tcp_server.cpp

#include "tcp_server.h"
#include "api_config.h"
#include "consumer_queue.h"
#include "sample.h"
#include "send_buffer.h"
#include "socket_utils.h"
#include "stream_info_impl.h"
#include "sync_serialization.h"
#include "util/cast.hpp"
#include "util/endian.hpp"
#include "util/strfuns.hpp"
#include <asio/io_context.hpp>
#include <asio/ip/host_name.hpp>
#include <asio/ip/tcp.hpp>
#include <asio/read_until.hpp>
#include <asio/streambuf.hpp>
#include <asio/write.hpp>
#include <algorithm>
#include <atomic>
#include <condition_variable>
#include <cstdint>
#include <cstring>
#include <exception>
#include <istream>
#include <loguru.hpp>
#include <memory>
#include <thread>
#include <utility>
#include <vector>
#ifndef _WIN32
#include <poll.h>
#endif

#ifdef _MSC_VER
// (inefficiently converting int to bool in portable_oarchive instantiation...)
#pragma warning(disable : 4800)
#endif

// a convention that applies when including portable_oarchive.h in multiple .cpp files.
// otherwise, the templates are instantiated in this file and sample.cpp which leads
// to errors like "multiple definition of `typeinfo name"
#define NO_EXPLICIT_TEMPLATE_INSTANTIATION
#include "portable_archive/portable_oarchive.hpp"

using std::size_t;

namespace lsl {
/**
 * Active session with a TCP client.
 *
 * A note on memory ownership:
 * - Generally, the stream_outlet maintains shared ownership of the `tcp_server`, `io_context`s,
 * and `stream_info`.
 * - At any point in time there are likely multiple request/handler chains in flight somewhere
 * between the operating system, asio, and the various handlers below.
 * The handlers are set up such that any memory that may be referred to by them in the future is
 * owned (shared) by the handler/callback function objects (this is what is encapsulated by the
 * client_session instance).
 * Their lifetime is managed by asio and ends when the handler chain ends (e.g., is aborted).
 * Since the TCP server is referred to (occasionally) by handler code, the `client_session`s store
 * a std::weak_ptr to the tcp_server that's upgraded to a std::shared_ptr as needed
 * - There is a per-session transfer thread (client_session::transfer_samples_thread()) that owns
 * the respective `client_session` which goes out of scope once the server is being shut down.
 * - The TCP server and client session also have shared ownership of the io_context (since in
 * some cases some transfer threads can outlive the stream outlet, and so the io_context is still
 * kept around until all sockets have been properly released).
 * - So memory is generally owned by the code (functors and stack frames) that needs to refer to
 * it for the duration of the execution.
 */
/**
 * Handler for synchronous (blocking) writes to all connected clients.
 * This class manages sockets that have been handed off from client_session
 * for zero-copy synchronous data transfer.
 *
 * Supports clients with different endianness by maintaining two socket groups:
 * - sockets_native_: clients with same endianness (zero-copy writes)
 * - sockets_swapped_: clients needing byte-swapped data (one copy for all)
 */
class sync_write_handler {
public:
	/**
	 * @param sample_bytes Total bytes per sample (channel_count * value_size)
	 * @param value_size Bytes per channel value (for byte-swapping)
	 * @param num_channels Number of channels in the stream
	 * @param send_timeout Per-consumer blocking-send timeout in seconds; 0 means block forever.
	 *        A consumer that can't accept a sample within this time is disconnected.
	 */
	sync_write_handler(std::size_t sample_bytes, std::size_t value_size, uint32_t num_channels,
		double send_timeout)
		: io_ctx_(1), sample_bytes_(sample_bytes), value_size_(value_size),
		  num_channels_(num_channels), send_timeout_(send_timeout) {}

	~sync_write_handler() {
		// Close all sockets
		auto close_sockets = [](std::vector<tcp_socket_p> &sockets) {
			for (auto &sock : sockets) {
				if (sock && sock->is_open()) {
					asio::error_code ec;
					sock->close(ec);
				}
			}
		};
		close_sockets(sockets_native_);
		close_sockets(sockets_swapped_);
	}

	/// Send the handshake and register a socket atomically with respect to sample writes.
	void add_socket(tcp_socket::native_handle_type handle, tcp_socket::protocol_type protocol,
		bool reverse_byte_order, const asio::streambuf::const_buffers_type &header) {
		std::lock_guard<std::mutex> lock(mutex_);
		auto sock = std::make_unique<tcp_socket>(io_ctx_, protocol, handle);
		apply_send_timeout(*sock);
		// An inlet may return from open_stream() as soon as these bytes arrive.
		// Keep sample writes locked out until the socket is registered below.
		// This synchronous-mode handshake runs on the server IO thread while
		// holding the sample-write lock. It uses the configured sync send timeout;
		// a timeout of zero permits unbounded blocking of both IO and sample writes.
		asio::write(*sock, header);
		if (reverse_byte_order) {
			sockets_swapped_.push_back(std::move(sock));
			LOG_F(INFO, "Added sync socket (swapped endian), now have %zu native + %zu swapped",
				sockets_native_.size(), sockets_swapped_.size());
		} else {
			sockets_native_.push_back(std::move(sock));
			LOG_F(INFO, "Added sync socket (native endian), now have %zu native + %zu swapped",
				sockets_native_.size(), sockets_swapped_.size());
		}
	}

	/// Write buffers to all connected sockets (blocking gather-write)
	void write_all_blocking(const std::vector<asio::const_buffer> &bufs) {
		std::lock_guard<std::mutex> lock(mutex_);
		if (sockets_native_.empty() && sockets_swapped_.empty()) return;

		bool any_broken = false;

		// Write to same-endianness clients (zero-copy)
		if (!sockets_native_.empty()) {
			any_broken |= write_to_group(sockets_native_, bufs);
		}

		// Write to reverse-endianness clients (one copy, byte-swapped)
		if (!sockets_swapped_.empty()) {
			auto swapped_bufs = swap_buffers(bufs);
			any_broken |= write_to_group(sockets_swapped_, swapped_bufs);
		}

		// Remove closed sockets from both groups
		if (any_broken) {
			auto remove_closed = [](std::vector<tcp_socket_p> &sockets) {
				sockets.erase(std::remove_if(sockets.begin(), sockets.end(),
								  [](const tcp_socket_p &s) { return !s || !s->is_open(); }),
					sockets.end());
			};
			remove_closed(sockets_native_);
			remove_closed(sockets_swapped_);
			LOG_F(INFO, "After cleanup, have %zu native + %zu swapped sync consumers",
				sockets_native_.size(), sockets_swapped_.size());
		}
	}

	/// Check if there are any connected consumers
	bool have_consumers() {
		std::lock_guard<std::mutex> lock(mutex_);
		// Sync sockets have no running IO context. Check on demand, without changing
		// their native blocking mode (which SO_SNDTIMEO relies on).
		prune_disconnected(sockets_native_);
		prune_disconnected(sockets_swapped_);
		return !sockets_native_.empty() || !sockets_swapped_.empty();
	}

private:
	/// Called under mutex_, which also serializes socket writes and registration.
	static void prune_disconnected(std::vector<tcp_socket_p> &sockets) {
		sockets.erase(std::remove_if(sockets.begin(), sockets.end(), [](const tcp_socket_p &sock) {
			if (!sock || !sock->is_open()) return true;
			// After the handshake the feed is one-way: readability means EOF,
			// a reset, or unexpected client data. All end the feed, as in the
			// asynchronous peer-close watcher. Never issue a blocking receive.
#ifdef _WIN32
			fd_set readable;
			FD_ZERO(&readable);
			FD_SET(sock->native_handle(), &readable);
			timeval timeout{};
			return ::select(0, &readable, nullptr, nullptr, &timeout) > 0;
#else
			// poll also handles descriptors beyond select's FD_SETSIZE limit.
			pollfd fd{sock->native_handle(), POLLIN, 0};
			return ::poll(&fd, 1, 0) > 0 && (fd.revents & (POLLIN | POLLHUP | POLLERR));
#endif
		}), sockets.end());
	}

	/// Apply the configured blocking-send timeout to a socket (no-op if disabled). SO_SNDTIMEO
	/// only takes effect on a blocking socket, so we force blocking mode first. On expiry a
	/// synchronous write returns try_again/would_block/timed_out, which write_to_group treats
	/// as a disconnect.
	void apply_send_timeout(tcp_socket &sock) const {
		if (send_timeout_ <= 0) return;
		asio::error_code ec;
		sock.native_non_blocking(false, ec);
#if defined(_WIN32)
		DWORD ms = static_cast<DWORD>(send_timeout_ * 1000.0);
		::setsockopt(sock.native_handle(), SOL_SOCKET, SO_SNDTIMEO,
			reinterpret_cast<const char *>(&ms), sizeof(ms));
#else
		struct timeval tv;
		tv.tv_sec = static_cast<time_t>(send_timeout_);
		tv.tv_usec = static_cast<suseconds_t>((send_timeout_ - static_cast<double>(tv.tv_sec)) * 1e6);
		::setsockopt(sock.native_handle(), SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
#endif
	}

	/// Write buffers to a group of sockets, returns true if any socket was broken
	bool write_to_group(std::vector<tcp_socket_p> &sockets,
		const std::vector<asio::const_buffer> &bufs) {
		bool any_broken = false;
		for (auto &sock : sockets) {
			if (!sock || !sock->is_open()) continue;

			asio::error_code ec;
			asio::write(*sock, bufs, ec);

			if (ec) {
				switch (ec.value()) {
				case asio::error::broken_pipe:
				case asio::error::connection_reset:
				case asio::error::not_connected:
					LOG_F(WARNING, "Sync socket disconnected: %s", ec.message().c_str());
					break;
				// SO_SNDTIMEO expiry surfaces as would_block (== try_again on POSIX) or, on
				// Windows, timed_out.
				case asio::error::would_block:
				case asio::error::timed_out:
					LOG_F(WARNING, "Sync consumer too slow (send timed out after %.1fs), dropping it",
						send_timeout_);
					break;
				default: LOG_F(ERROR, "Sync write error: %s", ec.message().c_str());
				}
				sock->close(ec);
				any_broken = true;
			}
		}
		return any_broken;
	}

	/// Byte-swap buffers for reverse-endianness clients (see sync_swap_buffers).
	std::vector<asio::const_buffer> swap_buffers(const std::vector<asio::const_buffer> &bufs) {
		return sync_swap_buffers(bufs, swapped_data_, sample_bytes_, value_size_, num_channels_);
	}

	asio::io_context io_ctx_;
	std::vector<tcp_socket_p> sockets_native_;  // same endianness as server
	std::vector<tcp_socket_p> sockets_swapped_; // need byte-swapped data
	mutable std::mutex mutex_;

	// For byte-swapping
	std::size_t sample_bytes_;	// total bytes per sample
	std::size_t value_size_;	// bytes per channel value
	uint32_t num_channels_;		// number of channels
	std::vector<char> swapped_data_; // backing storage for byte-swapped buffers
	double send_timeout_;		// per-consumer blocking-send timeout in seconds (0 = forever)
};

class client_session : public std::enable_shared_from_this<client_session> {

public:
	/// Instantiate a new session & its socket.
	client_session(const tcp_server_p &serv, tcp_socket &&sock)
		: io_(serv->io_), serv_(serv), sock_(std::move(sock)), requeststream_(&requestbuf_) {}

	/// Destructor.
	~client_session();

	/// Get the socket of this session.
	tcp_socket &socket() { return sock_; }

	/// Ask the transfer thread to exit at its next wakeup.
	void abort_transfer() { abort_transfer_.store(true, std::memory_order_release); }

	/// Begin processing this session (i.e., data transmission over the socket).
	void begin_processing();

private:
	/// Handler that gets called when the reading of the 1st line (command line) of the inbound
	/// message finished.
	void handle_read_command_outcome(err_t err);

	/// Handler that gets called after finishing reading of the query line.
	void handle_read_query_outcome(err_t err);

	/// Helper function to send a status message to the connected party.
	void send_status_message(const std::string &msg);

	/// Handler that gets called after finishing the reading of feedparameters.
	void handle_read_feedparams(
		int request_protocol_version, const std::string &request_uid, err_t err);

	/// Handler that gets called sending the feedheader has completed.
	void handle_send_feedheader_outcome(
		err_t err, std::size_t n, std::shared_ptr<consumer_queue> queue);

	/// Detect peer close on the feed socket so idle consumers get unregistered (see #267).
	void abort_on_peer_close(const std::shared_ptr<consumer_queue> &queue);

	/// Transfers samples from the server's send buffer into the async send queues of IO threads
	void transfer_samples_thread(std::shared_ptr<client_session> keepalive,
		std::shared_ptr<consumer_queue> &&queue, int max_samples_per_chunk);

	/// Handler that gets called when a sample transfer has been completed.
	void handle_chunk_transfer_outcome(err_t err, std::size_t len);

	/// the IO context (not owned, see tcp_server::io_)
	asio::io_context &io_;
	/// the server that is associated with this connection
	std::weak_ptr<tcp_server> serv_;
	/// connection socket
	tcp_socket sock_;

	// data used by the transfer thread (and some other handlers)
	/// this buffer holds the data feed generated by us
	asio::streambuf feedbuf_;
	/// this buffer holds the request as received from the client (incrementally filled)
	asio::streambuf requestbuf_;
	/// output archive (wrapped around the feed buffer)
	std::unique_ptr<class eos::portable_oarchive> outarch_;
	/// this is a stream on top of the request buffer for convenient parsing
	std::istream requeststream_;
	/// scratchpad memory (e.g., for endianness conversion)
	char *scratch_{nullptr};
	/// protocol version to use for transmission
	int data_protocol_version_{100};
	/// is the client's endianness reversed (big<->little endian)
	bool reverse_byte_order_{false};
	/// our chunk granularity
	int chunk_granularity_{0};
	/// maximum number of samples buffered
	int max_buffered_{0};
	/// set when the consumer closes the feed connection (or sends unexpected data)
	std::atomic<bool> abort_transfer_{false};
	/// dummy byte for the peer-close detector's pending read
	char peer_close_byte_{0};

	// data exchanged between the transfer completion handler and the transfer thread
	/// whether the current transfer has finished (possibly with an error)
	bool transfer_completed_;
	/// the outcome of the last chunk transfer
	asio::error_code transfer_error_;
	/// the amount of bytes transferred
	std::size_t transfer_amount_;
	/// a mutex that protects the completion data
	std::mutex completion_mut_;
	/// a condition variable that signals completion
	std::condition_variable completion_cond_;
};

tcp_server::tcp_server(stream_info_impl_p info, asio::io_context &io, send_buffer_p sendbuf,
	factory_p factory, int chunk_size, bool allow_v4, bool allow_v6, bool do_sync)
	: chunk_size_(chunk_size), info_(std::move(info)), io_(io), factory_(std::move(factory)),
	  send_buffer_(std::move(sendbuf)) {
	// Create sync handler if sync mode is requested
	if (do_sync) {
		std::size_t sample_bytes = factory_->datasize();
		std::size_t value_size = format_sizes[info_->channel_format()];
		uint32_t num_channels = info_->channel_count();
		double send_timeout = api_config::get_instance()->sync_send_timeout();
		sync_handler_ = std::make_unique<sync_write_handler>(
			sample_bytes, value_size, num_channels, send_timeout);
		LOG_F(INFO, "TCP server initialized in synchronous (zero-copy) mode (send timeout %.1fs)",
			send_timeout);
	}
	// assign connection-dependent fields
	info_->session_id(api_config::get_instance()->session_id());
	info_->reset_uid();
	info_->created_at(lsl_clock());
	info_->hostname(asio::ip::host_name());

	if (allow_v4) {
		try {
			acceptor_v4_ = std::make_unique<tcp_acceptor>(io_, asio::ip::tcp::v4());
			auto port = bind_and_listen_to_port_in_range(*acceptor_v4_, asio::ip::tcp::v4(), 10);
			info_->v4data_port(port);
			LOG_F(1, "Created IPv%d TCP acceptor for %s @ port %d", 4, info_->name().c_str(), port);
		} catch (std::exception &e) {
			LOG_F(WARNING, "Failed to create IPv%d acceptor: %s", 4, e.what());
			acceptor_v4_.reset();
		}
	}
	if (allow_v6) {
		try {
			acceptor_v6_ = std::make_unique<tcp_acceptor>(io_, asio::ip::tcp::v6());
			auto port = bind_and_listen_to_port_in_range(*acceptor_v6_, asio::ip::tcp::v6(), 10);
			info_->v6data_port(port);
			LOG_F(1, "Created IPv%d TCP acceptor for %s @ port %d", 6, info_->name().c_str(), port);
		} catch (std::exception &e) {
			LOG_F(WARNING, "Failed to create IPv%d acceptor: %s", 6, e.what());
			acceptor_v6_.reset();
		}
	}
	if (!acceptor_v4_ && !acceptor_v6_)
		throw std::runtime_error("Failed to instantiate socket acceptors for the TCP server");
}

tcp_server::~tcp_server() {
	// sync_handler_ destructor will close all sync sockets
}

void tcp_server::write_all_blocking(const std::vector<asio::const_buffer> &bufs) {
	if (sync_handler_) {
		sync_handler_->write_all_blocking(bufs);
	}
}

bool tcp_server::wait_for_transfer_threads(double timeout) {
	std::unique_lock<std::mutex> lock(transfer_mut_);
	return transfer_cv_.wait_for(
		lock, std::chrono::duration<double>(timeout), [this]() { return transfer_threads_ == 0; });
}

bool tcp_server::have_sync_consumers() const {
	return sync_handler_ && sync_handler_->have_consumers();
}

// === externally issued asynchronous commands ===

void tcp_server::begin_serving() {
	// pre-generate the info's messages
	shortinfo_msg_ = info_->to_shortinfo_message();
	fullinfo_msg_ = info_->to_fullinfo_message();
	// start accepting connections
	if (acceptor_v4_) accept_next_connection(acceptor_v4_);
	if (acceptor_v6_) accept_next_connection(acceptor_v6_);
}

void tcp_server::end_serving() {
	// issue closure of the server socket; this will result in a cancellation of the associated IO
	// operations
	post(io_, [this, shared_this = shared_from_this()]() {
		if (acceptor_v4_) acceptor_v4_->close();
		if (acceptor_v6_) acceptor_v6_->close();
	});
	// issue closure of all active client session sockets; cancels the related outstanding IO jobs
	close_inflight_sessions();
	// also notify any transfer threads that are blocked waiting for a sample by pushing an empty
	// sample pointer (the threads check for !samp and skip it, see transfer_samples_thread)
	send_buffer_->push_sample(sample_p());
}

// === accept loop ===


void tcp_server::accept_next_connection(tcp_acceptor_p &acceptor) {
	try {
		if (!acceptor || !acceptor->is_open()) return;

		// Select the IO context for handling the socket
		auto &sock_io_ctx = io_;

		// accept a connection on the session's socket
		acceptor->async_accept(sock_io_ctx, [shared_this = shared_from_this(), &acceptor](
												err_t err, tcp_socket sock) {
			if (err == asio::error::operation_aborted || err == asio::error::shut_down) return;

			// no error: create a new session and start processing
			if (!err)
				std::make_shared<client_session>(shared_this, std::move(sock))->begin_processing();
			else
				LOG_F(WARNING, "Unhandled accept error: %s", err.message().c_str());

			// move on to the next connection if the acceptor is still open
			if (acceptor && acceptor->is_open())
				shared_this->accept_next_connection(acceptor);
		});
	} catch (std::exception &e) {
		LOG_F(ERROR, "Error during tcp_server::accept_next_connection: %s", e.what());
	}
}


// === graceful cancellation of in-flight sockets ===

void tcp_server::register_inflight_session(const std::shared_ptr<client_session> &session) {
	std::lock_guard<std::recursive_mutex> lock(inflight_mut_);
	inflight_.insert(std::make_pair(session.get(), session));
}

void tcp_server::unregister_inflight_session(client_session *session) {
	std::lock_guard<std::recursive_mutex> lock(inflight_mut_);
	auto pos = inflight_.find(session);
	if (pos != inflight_.end()) inflight_.erase(pos);
}

void tcp_server::close_inflight_sessions() {
	std::lock_guard<std::recursive_mutex> lock(inflight_mut_);
	for (auto &pair : inflight_) {
		auto session = pair.second.lock();
		// session has already expired on its own
		if (!session) continue;
		session->abort_transfer();
		post(session->socket().get_executor(), [session]() {
			asio::error_code ec;
			auto &sock = session->socket();
			if (sock.is_open()) {
				sock.shutdown(tcp_socket::shutdown_both, ec);
				sock.close(ec);
				if (ec) LOG_F(WARNING, "Error during shutdown_and_close: %s", ec.message().c_str());
			}
		});
	}
	inflight_.clear();
}

// === implementation of the client_session class ===

client_session::~client_session() {
	LOG_F(1, "Destructing session %p", this);
	delete[] scratch_;
	if (auto serv = serv_.lock()) serv->unregister_inflight_session(this);
}

void client_session::begin_processing() {
	try {
		sock_.set_option(asio::ip::tcp::no_delay(true));
		if (api_config::get_instance()->socket_send_buffer_size() > 0)
			sock_.set_option(asio::socket_base::send_buffer_size(
				api_config::get_instance()->socket_send_buffer_size()));
		if (api_config::get_instance()->socket_receive_buffer_size() > 0)
			sock_.set_option(asio::socket_base::receive_buffer_size(
				api_config::get_instance()->socket_receive_buffer_size()));
		// register this socket as "in-flight" with the server (so that any subsequent ops on it can
		// be aborted if necessary)
		if (auto serv = serv_.lock()) {
			serv->register_inflight_session(shared_from_this());
			// read the request line
			async_read_until(sock_, requestbuf_, "\r\n",
				[shared_this = shared_from_this()](err_t err, std::size_t /*unused*/) {
					shared_this->handle_read_command_outcome(err);
				});
		} else {
			throw std::runtime_error("server disappeared before start client session");
		}
	} catch (std::exception &e) {
		LOG_F(ERROR, "Error during client_session::begin_processing: %s", e.what());
	}
}

void client_session::handle_read_command_outcome(err_t read_err) {
	try {
		if (read_err) return;
		// parse request method
		std::string method;
		getline(requeststream_, method);
		method = trim(method);
		if (method == "LSL:shortinfo")
			// shortinfo request: read the content query string
			async_read_until(sock_, requestbuf_, "\r\n",
				[shared_this = shared_from_this()](err_t err, std::size_t /*unused*/) {
					shared_this->handle_read_query_outcome(err);
				});
		else if (method == "LSL:fullinfo") {
			// fullinfo request: reply right away
			auto serv = serv_.lock();
			if (serv)
				async_write(sock_, asio::buffer(serv->fullinfo_msg_),
					[shared_this = shared_from_this(), serv](
						err_t /*unused*/, std::size_t /*unused*/) {});
		} else if (method == "LSL:streamfeed")
			// streamfeed request (1.00): read feed parameters
			async_read_until(sock_, requestbuf_, "\r\n",
				[shared_this = shared_from_this()](err_t err, std::size_t /*unused*/) {
					shared_this->handle_read_feedparams(100, "", err);
				});
		else if (method.compare(0, 15, "LSL:streamfeed/") == 0) {
			// streamfeed request with version: read feed parameters
			std::vector<std::string> parts = splitandtrim(method, ' ', true);
			async_read_until(sock_, requestbuf_, "\r\n\r\n",
				[shared_this = shared_from_this(),
					request_protocol_version = std::stoi(parts[0].substr(15)),
					request_uid = (parts.size() > 1) ? parts[1] : ""](
					err_t err, std::size_t /*unused*/) {
					shared_this->handle_read_feedparams(request_protocol_version, request_uid, err);
				});
		}
	} catch (std::exception &e) {
		LOG_F(WARNING, "Unexpected error while parsing a client command: %s", e.what());
	}
}

void client_session::handle_read_query_outcome(err_t err) {
	try {
		if (err) return;
		// read the query line
		std::string query;
		getline(requeststream_, query);
		query = trim(query);
		auto serv = serv_.lock();
		if (!serv) return;
		if (serv->info_->matches_query(query)) {
			// matches: reply (otherwise just close the stream). Keep the session (and thus the
			// socket) alive until the shortinfo has been sent; the session is then destroyed,
			// closing the socket so the client sees EOF and knows the reply is complete.
			async_write(sock_, asio::buffer(serv->shortinfo_msg_),
				[shared_this = shared_from_this(), serv](err_t /*unused*/, std::size_t /*unused*/) {
				});
		} else {
			DLOG_F(INFO, "%p got a shortinfo query response for the wrong query", this);
		}
	} catch (std::exception &e) {
		LOG_F(WARNING, "Unexpected error while parsing a client request: %s", e.what());
	}
}

void client_session::send_status_message(const std::string &msg) {
	auto buf(std::make_shared<std::string>(msg));
	async_write(sock_, asio::buffer(*buf),
		[buf, shared_this = shared_from_this()](err_t /*unused*/,
			std::size_t /*unused*/) { /* keep objects alive until the message is sent */ });
}

void client_session::handle_read_feedparams(
	int request_protocol_version, const std::string &request_uid, err_t err) {
	try {
		if (err) return;
		DLOG_F(2, "%p got a streamfeed request", this);
		// --- protocol negotiation ---

		// check request validity
		auto cfg_proto_version = api_config::get_instance()->use_protocol_version();
		if (request_protocol_version / 100 > cfg_proto_version / 100) {
			send_status_message(
				"LSL/" + std::to_string(cfg_proto_version) + " 505 Version not supported");
			DLOG_F(WARNING, "%p Got a request for a too new protocol version", this);
			return;
		}
		auto serv = serv_.lock();
		if (!serv) return;
		auto &info = serv->info_;
		if (!request_uid.empty() && request_uid != info->uid()) {
			send_status_message("LSL/" + to_string(cfg_proto_version) + " 404 Not found");
			return;
		}

		if (request_protocol_version >= 110) {
			int client_byte_order = 1234;		  // assume little endian
			double client_endian_performance = 0; // the other party's endian conversion performance
			bool client_has_ieee754_floats =
				true; // the client has IEEE-754 compliant floating point formats
			bool client_supports_subnormals = true; // the client supports subnormal numbers
			int client_protocol_version =
				request_protocol_version; // assume that the client wants to use the same
										  // version for data transmission
			int client_value_size = info->channel_bytes(); // assume that the client has a standard
														   // size for the relevant data type
			lsl_channel_format_t format = info->channel_format();

			// read feed parameters
			char buf[16384] = {0};
			while (requeststream_.getline(buf, sizeof(buf)) && (buf[0] != '\r')) {
				std::string hdrline(buf);
				std::size_t colon = hdrline.find_first_of(':');
				if (colon != std::string::npos) {
					// strip off comments
					auto semicolon = hdrline.find_first_of(';');
					if (semicolon != std::string::npos) hdrline.erase(semicolon);
					// convert to lowercase
					for (auto &c : hdrline) c = ::tolower(c);
					// extract key & value
					std::string type = trim(hdrline.substr(0, colon)),
								rest = trim(hdrline.substr(colon + 1));
					// get the header information
					if (type == "native-byte-order") client_byte_order = std::stoi(rest);
					if (type == "endian-performance")
						client_endian_performance = from_string<double>(rest);
					if (type == "has-ieee754-floats")
						client_has_ieee754_floats = from_string<bool>(rest);
					if (type == "supports-subnormals")
						client_supports_subnormals = from_string<bool>(rest);
					if (type == "value-size") client_value_size = std::stoi(rest);
					if (type == "max-buffer-length") max_buffered_ = std::stoi(rest);
					if (type == "max-chunk-length") chunk_granularity_ = std::stoi(rest);
					if (type == "protocol-version") client_protocol_version = std::stoi(rest);
				} else {
					DLOG_F(WARNING, "%p Request line '%s' contained no key-value pair", this,
						hdrline.c_str());
				}
			}

			// determine the parameters for data transmission
			bool client_suppress_subnormals = false;

			lsl::Endianness use_byte_order = LSL_BYTE_ORDER;

			// use least common denominator data protocol version
			data_protocol_version_ = std::min(cfg_proto_version, client_protocol_version);
			// downgrade to 1.00 (portable binary format) if an unsupported binary conversion is
			// involved
			if (info->channel_format() != cft_string && info->channel_bytes() != client_value_size)
				data_protocol_version_ = 100;
			if (!format_ieee754[cft_double64] ||
				(format == cft_float32 && !format_ieee754[cft_float32]) ||
				!client_has_ieee754_floats)
				data_protocol_version_ = 100;
			if (data_protocol_version_ >= 110) {

				// enable endian conversion when
				// 1. our byte ordering is different from the client's *and*
				// 2. we can actually perform the conversion *and*
				// 3. the sample format is wide enough for endianness to matter *and*
				// 4. we're faster at converting than the client
				if (LSL_BYTE_ORDER != client_byte_order &&						  // (1)
					lsl::can_convert_endian(client_byte_order, client_value_size) // (2)
					&& client_value_size > 1 &&									  // (3)
					(measure_endian_performance() > client_endian_performance))	  // (4)
				{
					use_byte_order = static_cast<lsl::Endianness>(client_byte_order);
					reverse_byte_order_ = true;
				}

				// determine if subnormal suppression needs to be enabled
				client_suppress_subnormals =
					(format_subnormal[format] && !client_supports_subnormals);
			}

			// send the response
			std::ostream response_stream(&feedbuf_);
			response_stream << "LSL/" << cfg_proto_version << " 200 OK\r\n";
			response_stream << "UID: " << info->uid() << "\r\n";
			response_stream << "Byte-Order: " << use_byte_order << "\r\n";
			response_stream << "Suppress-Subnormals: " << client_suppress_subnormals << "\r\n";
			response_stream << "Data-Protocol-Version: " << data_protocol_version_ << "\r\n";
			response_stream << "\r\n" << std::flush;
		} else {
			// read feed parameters
			requeststream_ >> max_buffered_ >> chunk_granularity_;
		}

		// The synchronous writer only emits data protocol 1.10 samples.
		// Reject the request as unsupported.
		if (max_buffered_ > 0 && serv->is_sync_mode() && data_protocol_version_ < 110) {
			// Debug builds only: a rejected inlet retries every 500 ms. Protocol 1.10 inlets
			// get the reason in the 505 response below.
			DLOG_F(WARNING,
				"Rejected a streamfeed request: synchronous outlets require data protocol 1.10");
			if (request_protocol_version >= 110)
				send_status_message("LSL/" + to_string(cfg_proto_version) +
									" 505 Synchronous outlet requires data protocol 1.10");
			return;
		}

		// --- validation ---
		if (data_protocol_version_ == 100) {
			// create a portable output archive to write to
			outarch_ = std::make_unique<eos::portable_oarchive>(feedbuf_);
			// serialize the shortinfo message into an archive
			*outarch_ << serv->shortinfo_msg_;
		} else {
			// allocate scratchpad memory for endian conversion, etc.
			scratch_ = new char[format_sizes[info->channel_format()] * info->channel_count()];
		}

		// send test pattern samples
		lsl::factory fac(info->channel_format(), info->channel_count(), 4);

		for (int test_pattern : {4, 2}) {
			lsl::sample_p temp(fac.new_sample(0.0, false));
			temp->assign_test_pattern(test_pattern);
			if (data_protocol_version_ >= 110)
				temp->save_streambuf(
					feedbuf_, data_protocol_version_, reverse_byte_order_, scratch_);
			else
				*outarch_ << *temp;
		}

		if (max_buffered_ > 0 && serv->is_sync_mode()) {
			// The synchronous writer sends the header under its sample-write lock,
			// then registers the socket before an immediate push can acquire it.
			auto protocol = sock_.local_endpoint().protocol();
			try {
				serv->sync_handler_->add_socket(
					sock_.release(), protocol, reverse_byte_order_, feedbuf_.data());
			} catch (const asio::system_error &e) {
				// add_socket owns and closes the released handle on failure. Peer
				// resets here are connection failures, not serialization errors.
				LOG_F(INFO, "Synchronous stream handshake failed: %s", e.what());
			}
			serv->unregister_inflight_session(this);
			return;
		}

		// Subscribe before sending the handshake: receiving its test patterns is
		// what lets the inlet return from open_stream(). Queue samples until the
		// header write completes so data cannot overtake the handshake.
		// have_consumers()/wait_for_consumers() can now observe this subscription
		// before the handshake completes; open_stream() remains the inlet's gate.
		std::shared_ptr<consumer_queue> queue;
		if (max_buffered_ > 0) queue = serv->send_buffer_->new_consumer(max_buffered_);
		async_write(sock_, feedbuf_.data(),
			[shared_this = shared_from_this(), queue = std::move(queue)](
				err_t err, std::size_t len) mutable {
				shared_this->handle_send_feedheader_outcome(err, len, std::move(queue));
			});
		DLOG_F(2, "%p sent test pattern samples", this);
	} catch (std::exception &e) {
		LOG_F(WARNING, "Unexpected error while serializing the feed header: %s", e.what());
	}
}

void client_session::handle_send_feedheader_outcome(
	err_t err, std::size_t n, std::shared_ptr<consumer_queue> queue) {
	try {
		if (err) return;

		feedbuf_.consume(n);

		auto serv = serv_.lock();
		if (!serv) return;

		// quit if max_buffered_ is 0. This is a bit unexpected, but backwards compatible and quite
		// convenient for unit tests
		if (max_buffered_ <= 0) return;
		// teardown already began (handler ran during the outlet's final drain)
		if (abort_transfer_.load(std::memory_order_acquire)) return;

		abort_on_peer_close(queue);

		// determine the maximum chunk size
		int max_samples_per_chunk = std::numeric_limits<int>::max();
		if (chunk_granularity_)
			max_samples_per_chunk = chunk_granularity_;
		else if (serv->chunk_size_)
			max_samples_per_chunk = serv->chunk_size_;

		// spawn a sample transfer thread.
		{
			std::lock_guard<std::mutex> lock(serv->transfer_mut_);
			serv->transfer_threads_++;
		}
		std::thread(&client_session::transfer_samples_thread, this, shared_from_this(),
			std::move(queue), max_samples_per_chunk)
			.detach();
	} catch (std::exception &e) {
		LOG_F(WARNING, "Unexpected error while handling the feedheader send outcome: %s", e.what());
	}
}

void client_session::abort_on_peer_close(const std::shared_ptr<consumer_queue> &queue) {
	// The feed is one-way after the handshake; a completed read means EOF or unexpected data.
	// Either way the consumer is gone. A weak_ptr keeps the queue from staying registered if the
	// transfer thread has already exited (e.g. after a failed write).
	std::weak_ptr<consumer_queue> weak_queue = queue;
	sock_.async_read_some(asio::buffer(&peer_close_byte_, 1),
		[shared_this = shared_from_this(), weak_queue](err_t err, std::size_t /*n*/) {
			if (err == asio::error::operation_aborted) return;
			shared_this->abort_transfer_.store(true, std::memory_order_release);
			// Wake through the send buffer rather than pushing here: this runs on an IO thread
			// and would otherwise be a second producer on the outlet thread's queue.
			auto q = weak_queue.lock();
			auto serv = shared_this->serv_.lock();
			if (q && serv) serv->send_buffer_->wake_consumer(q);
		});
}

void client_session::transfer_samples_thread(std::shared_ptr<client_session> keepalive,
	std::shared_ptr<consumer_queue> &&queue, int max_samples_per_chunk) {
	int samples_in_current_chunk = 0;
	while (!serv_.expired() && !abort_transfer_.load(std::memory_order_relaxed)) {
		try {
			// get next sample from the sample queue (blocking)
			sample_p samp(queue->pop_sample());
			if (abort_transfer_.load(std::memory_order_acquire)) break;

			// ignore blank samples (they are basically wakeup notifiers from someone's
			// end_serving())
			if (!samp) continue;
			// serialize the sample into the stream
			if (data_protocol_version_ >= 110)
				samp->save_streambuf(
					feedbuf_, data_protocol_version_, reverse_byte_order_, scratch_);
			else
				*outarch_ << *samp;
			// if the sample is marked as force-push or the configured chunk size is reached
			if (samp->pushthrough || ++samples_in_current_chunk >= max_samples_per_chunk) {
				// send off the chunk that we aggregated so far
				std::unique_lock<std::mutex> lock(completion_mut_);
				transfer_completed_ = false;
				async_write(sock_, feedbuf_.data(),
					[shared_this = shared_from_this()](err_t err, std::size_t len) {
						shared_this->handle_chunk_transfer_outcome(err, len);
					});
				// wait for the completion condition
				completion_cond_.wait(lock, [this]() { return transfer_completed_; });
				// handle transfer outcome
				if (!transfer_error_) {
					feedbuf_.consume(transfer_amount_);
				} else
					break;
				samples_in_current_chunk = 0;
			}
		} catch (std::exception &e) {
			LOG_F(WARNING, "Unexpected glitch in transfer_samples_thread: %s", e.what());
		}
	}
	// Unregister immediately; cancel the peer-close read so the session can be destroyed.
	queue.reset();
	post(io_, [keepalive]() {
		asio::error_code ec;
		keepalive->sock_.cancel(ec);
	});
	// only after posting the cancel, so the outlet's final drain runs it
	if (auto serv = serv_.lock()) {
		std::lock_guard<std::mutex> lock(serv->transfer_mut_);
		if (--serv->transfer_threads_ == 0) serv->transfer_cv_.notify_all();
	}
}

void client_session::handle_chunk_transfer_outcome(err_t err, std::size_t len) {
	try {
		{
			std::lock_guard<std::mutex> lock(completion_mut_);
			// assign the transfer outcome
			transfer_error_ = err;
			transfer_amount_ = len;
			transfer_completed_ = true;
		}
		// notify the server thread
		completion_cond_.notify_all();
	} catch (std::exception &e) {
		LOG_F(WARNING,
			"Catastrophic error in handling the chunk transfer outcome (in tcp_server): %s",
			e.what());
	}
}
} // namespace lsl