basic_rtmp_connection.cpp

#include "pch.h"
#include "basic_rtmp_connection.h"
#include "rtmp_application.h"
#include "rtmp_channel.h"
#include "rtmp_message.h"

namespace fms
{
	basic_rtmp_connection::basic_rtmp_connection(std::uint32_t id, boost::asio::io_context &io_context, app_host *app_manager)
		: client_session(id, app_manager)
		, m_io_context(io_context)
	{}

	void basic_rtmp_connection::post_close()
	{
		// run close() on this connection's own io_context — its socket/timers are
		// not safe to touch from another thread (e.g. the admin thread).
		boost::asio::post(m_io_context, [self = shared_from_this()]() { self->close(); });
	}

	void basic_rtmp_connection::handle_bytes_read(std::size_t bytes_transferred)
	{
		client_session::handle_bytes_read(bytes_transferred);

		if (ack_due(m_bytes_read, m_bytes_read_notify))
		{
			// No app until connect() is routed; leave the threshold so the ack
			// still goes out once there is one.
			if (m_app == nullptr)
				return;

			m_bytes_read_notify = next_ack_threshold(m_bytes_read, m_win_ack);
			rtmp_message_bytes_read_ptr const msg = std::make_shared<rtmp_message_bytes_read>(m_bytes_read);
			get_app()->enqueue_async_message(m_id, msg);
			notify();
		}
	}

	void basic_rtmp_connection::handle_message(rtmp_channel_ptr channel, rtmp_message_ptr msg)
	{
		rtmp_message_ptr result;
		boost::tribool ret;

		++m_messages_read;
		if (m_app != nullptr) // do we have an rtmp app assigned to us?
		{
			ret = get_app()->handle_message(msg, m_id, channel->received_header(), result);
			get_app()->update_stats(true, false, 1);
		}
		else
		{
			ret = m_app_manager->handle_message(msg, m_id, channel->received_header(), result);
			if (m_app != nullptr) // if app has been selected, update stats
				get_app()->update_stats(true, false, 1);
		}

		if (ret && result.get() != nullptr)
			handle_app_result(channel, result);
	}

	void basic_rtmp_connection::handle_internal_message(rtmp_message_ptr msg)
	{
		if (msg->type() == rtmp_message::eMessageChunkSize)
		{
			rtmp_message_chunk_size_ptr const cs_msg = std::static_pointer_cast<rtmp_message_chunk_size>(msg);
			(void)m_parser.set_chunk_size(cs_msg->chunk_size());   // out of range: keep the current size
		}
		else if (msg->type() == rtmp_message::eMessageWindowAcknowledgementSize)
		{
			rtmp_message_window_acknowledgement_size_ptr const ack = std::static_pointer_cast<rtmp_message_window_acknowledgement_size>(msg);
			if (std::uint32_t const n = ack->size(); n != 0)
			{
				m_win_ack = clamp_window(n);
				m_bytes_read_notify = next_ack_threshold(m_bytes_read, m_win_ack);
			}
		}
	}

}