#include <chrono>
#include <cstdlib>
#include <ctime>
#include <deque>
#include <iostream>
#include <sstream>
#include <string>
#include <random>

#include <so_5/all.hpp>

// A signal to worker agent to do something.
struct msg_start_thinking : public so_5::signal_t {};

// Message for logger.
struct log_message
{
	// Text to be logged.
	std::string m_what;
};

// Logger agent.
class a_logger_t final : public so_5::agent_t
{
public :
	a_logger_t( context_t ctx )
		:	so_5::agent_t( ctx
				// Limit the count of messages.
				// Because we can't lost log messages the overlimit
				// must lead to application crash.
				+ limit_then_abort< log_message >( 100 ) )
		,	m_started_at( std::chrono::steady_clock::now() )
	{}

	void so_define_agent() override
	{
		so_default_state().event(
			[this]( const log_message & evt ) {
				std::cout << "[+" << time_delta()
						<< "] -- " << evt.m_what << std::endl;
			} );
	}

private :
	const std::chrono::steady_clock::time_point m_started_at;

	std::string time_delta() const
	{
		auto now = std::chrono::steady_clock::now();

		std::ostringstream ss;
		ss << double(
				std::chrono::duration_cast< std::chrono::milliseconds >(
						now - m_started_at ).count()
				) / 1000.0 << "ms";

		return ss.str();
	}
};

// Agent for receiving run-time monitoring information.
class a_stats_listener_t final : public so_5::agent_t
{
public :
	a_stats_listener_t(
		// Environment to work in.
		context_t ctx,
		// Address of logger.
		so_5::mbox_t logger )
		:	so_5::agent_t( ctx )
		,	m_logger( std::move( logger ) )
	{}

	void so_define_agent() override
	{
		using namespace so_5::stats;

		auto & controller = so_environment().stats_controller();

		// Set up a filter for messages with run-time monitoring information.
		so_set_delivery_filter(
			// Message box to which delivery filter must be set.
			controller.mbox(),
			// Delivery predicate.
			[]( const messages::quantity< std::size_t > & msg ) {
				// Process only messages related to dispatcher's queue sizes.
				return suffixes::work_thread_queue_size() == msg.m_suffix;
			} );

		// We must receive messages from run-time monitor.
		so_default_state()
			.event(
				// This is mbox to that run-time statistic will be sent.
				controller.mbox(),
				&a_stats_listener_t::evt_quantity )
			.event( controller.mbox(),
				[this]( const messages::distribution_started & ) {
					so_5::send< log_message >( m_logger, "--- DISTRIBUTION STARTED ---" );
				} )
			.event( controller.mbox(),
				[this]( const messages::distribution_finished & ) {
					so_5::send< log_message >( m_logger, "--- DISTRIBUTION FINISHED ---" );
				} );
	}

	void so_evt_start() override
	{
		// Change the speed of run-time monitor updates.
		so_environment().stats_controller().set_distribution_period(
				std::chrono::milliseconds( 330 ) );
		// Turn the run-timer monitoring on.
		so_environment().stats_controller().turn_on();
	}

private :
	const so_5::mbox_t m_logger;

	void evt_quantity(
		const so_5::stats::messages::quantity< std::size_t > & evt )
	{
		std::ostringstream ss;

		ss << "stats: '" << evt.m_prefix << evt.m_suffix << "': " << evt.m_value;

		so_5::send< log_message >( m_logger, ss.str() );
	}
};

// Load generation agent.
class a_generator_t final : public so_5::agent_t
{
public :
	a_generator_t(
		// Environment to work in.
		context_t ctx,
		// Address of logger.
		so_5::mbox_t logger,
		// Addresses of worker agents.
		std::vector< so_5::mbox_t > workers )
		:	so_5::agent_t( ctx )
		,	m_logger( std::move( logger ) )
		,	m_workers( std::move( workers ) )
		,	m_turn_pause( 600 )
	{}

	void so_define_agent() override
	{
		so_default_state()
			.event( &a_generator_t::evt_next_turn );
	}

	void so_evt_start() override
	{
		// Start work cycle.
		so_5::send< msg_next_turn >( *this );
	}

private :
	// Signal about start of the next turn.
	struct msg_next_turn : public so_5::signal_t {};

	// Logger.
	const so_5::mbox_t m_logger;

	// Workers.
	const std::vector< so_5::mbox_t > m_workers;

	// Pause between working turns.
	const std::chrono::milliseconds m_turn_pause;

	void evt_next_turn(mhood_t< msg_next_turn >)
	{
		// Create and send new requests.
		generate_new_requests( random( 100, 200 ) );

		// Wait for next turn and process replies.
		so_5::send_delayed< msg_next_turn >( *this, m_turn_pause );
	}

	void generate_new_requests( unsigned int requests )
	{
		const auto size = m_workers.size();

		for( unsigned int i = 0; i != requests; ++i )
			so_5::send< msg_start_thinking >( m_workers[ i % size ] );

		so_5::send< log_message >( m_logger,
				std::to_string( requests ) + " requests are sent" );
	}

	static unsigned int
	random( unsigned int left, unsigned int right )
	{
		std::random_device rd;
		std::mt19937 gen{ rd() };
		return std::uniform_int_distribution< unsigned int >{left, right}(gen);
	}
};

// Worker agent.
class a_worker_t final : public so_5::agent_t
{
public :
	a_worker_t( context_t ctx )
		:	so_5::agent_t( ctx
				// Limit the maximum count of messages.
				+ limit_then_drop< msg_start_thinking >( 50 ) )
	{}

	void so_define_agent() override
	{
		so_default_state().event( [](mhood_t< msg_start_thinking >) {
				std::this_thread::sleep_for(
					std::chrono::milliseconds( 10 ) );
			} );
	}
};

void init( so_5::environment_t & env )
{
	env.introduce_coop( [&env]( so_5::coop_t & coop ) {
		// Logger will work on the default dispatcher.
		auto logger = coop.make_agent< a_logger_t >();

		// Run-time stats listener will work on a dedicated
		// one-thread dispatcher.
		coop.make_agent_with_binder< a_stats_listener_t >(
				so_5::disp::one_thread::make_dispatcher(
						env, "stats_listener" ).binder(),
				logger->so_direct_mbox() );

		// Bunch of workers.
		// Must work on dedicated thread_pool dispatcher.
		auto worker_disp = so_5::disp::thread_pool::make_dispatcher(
				env,
				"workers", // Name of dispatcher (for convience of monitoring).
				3 ); // Count of working threads.
		const auto worker_binding_params = so_5::disp::thread_pool::bind_params_t{}
				.fifo( so_5::disp::thread_pool::fifo_t::individual );

		std::vector< so_5::mbox_t > workers;
		for( int i = 0; i != 5; ++i )
		{
			auto w = coop.make_agent_with_binder< a_worker_t >(
					worker_disp.binder( worker_binding_params ) );
			workers.push_back( w->so_direct_mbox() );
		}

		// Generators will work on dedicated active_obj dispatcher.
		auto generator_disp = so_5::disp::active_obj::make_dispatcher(
				env, "generator" );

		coop.make_agent_with_binder< a_generator_t >(
				generator_disp.binder(),
				logger->so_direct_mbox(),
				std::move( workers ) );
	});

	// Take some time to work.
	std::this_thread::sleep_for( std::chrono::seconds(50) );

	env.stop();
}

int main()
{
	try
	{
		so_5::launch( &init );
	}
	catch( const std::exception & ex )
	{
		std::cerr << "Error: " << ex.what() << std::endl;
		return 1;
	}

	return 0;
}