so_5/default_message_limits/main.cpp

/*
 * An example of using default message limits.
 */

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

#include <so_5/all.hpp>

using namespace std::chrono_literals;

// A base part for every request.
struct request_base
{
  // Request ID.
  int m_id;

  request_base( int id ) : m_id{ id } {}
};

// Messages for different request types.
struct req_A final : request_base {};
struct req_B final : request_base {};
struct req_C final : request_base {};
struct req_D final : request_base {};
struct req_E final : request_base {};
struct req_F final : request_base {};
struct req_G final : request_base {};
struct req_I final : request_base {};

// 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();
  }
};

// An agent that plays a role of "trash can" for rejected requests.
class a_trash_can_t final : public so_5::agent_t
{
  const so_5::mbox_t m_logger;

  template< typename Request >
  [[nodiscard]]
  auto
  make_event_handler( std::string name )
  {
    return [this, req_name=std::move(name)]( mhood_t<Request> cmd ) {
      so_5::send< log_message >( m_logger,
          req_name + ": redirected to trash can, id=" +
          std::to_string(cmd->m_id) );
    };
  }

public :
  a_trash_can_t(
    context_t ctx,
    so_5::mbox_t logger )
    : so_5::agent_t{ ctx
        // Hold no more than 10 instances of any message type.
        + limit_then_drop< any_unspecified_message >( 10u )
      }
    , m_logger{ std::move(logger) }
  {}

  void so_define_agent() override
  {
    so_default_state()
      .event( make_event_handler< req_A >( "req_A" ) )
      .event( make_event_handler< req_B >( "req_B" ) )
      .event( make_event_handler< req_C >( "req_C" ) )
      .event( make_event_handler< req_D >( "req_D" ) )
      .event( make_event_handler< req_E >( "req_E" ) )
      .event( make_event_handler< req_F >( "req_F" ) )
      .event( make_event_handler< req_G >( "req_G" ) )
      .event( make_event_handler< req_I >( "req_I" ) )
      ;
  }

};

// An agent that does "normal" processing of incoming messages.
class a_processor_t final : public so_5::agent_t
{
  const so_5::mbox_t m_logger;

  template< typename Request >
  [[nodiscard]]
  auto make_request_handler(
    std::string name,
    std::chrono::milliseconds duration )
  {
    return [this, req_name=std::move(name), duration]
      ( mhood_t< Request > cmd )
      {
        so_5::send< log_message >( m_logger,
            "processing request " + req_name + " (" +
            std::to_string( cmd->m_id ) + ") for " +
            std::to_string( duration.count() ) + "ms" );

        // Imitation of some intensive processing.
        std::this_thread::sleep_for(
            std::chrono::milliseconds( duration ) );
      };
  }

public :
  struct shutdown final : public so_5::signal_t {};

  a_processor_t(
    context_t ctx,
    so_5::mbox_t logger,
    so_5::mbox_t trash_can )
    : so_5::agent_t{ ctx
        // Set the personal limits for several of messages.
        + limit_then_redirect< req_A >( 10u, trash_can )
        + limit_then_redirect< req_C >( 8u, trash_can )
        + limit_then_abort< shutdown >( 1u )
        // All other messages will have the same limit and reaction.
        + limit_then_redirect< any_unspecified_message >( 4u, trash_can )
      }
    , m_logger( std::move( logger ) )
  {}

  void so_define_agent() override
  {
    so_default_state()
      .event( make_request_handler< req_A >( "req_A", 1ms ) )
      .event( make_request_handler< req_B >( "req_B", 15ms ) )
      .event( make_request_handler< req_C >( "req_C", 2ms ) )
      .event( make_request_handler< req_D >( "req_D", 18ms ) )
      .event( make_request_handler< req_E >( "req_E", 22ms ) )
      .event( make_request_handler< req_F >( "req_F", 19ms ) )
      .event( make_request_handler< req_G >( "req_G", 24ms ) )
      .event( make_request_handler< req_I >( "req_I", 23ms ) )
      .event( [this]( mhood_t< shutdown > ) {
          so_deregister_agent_coop_normally();
        } )
      ;
  }
};

using sender_t = void(*)( const so_5::mbox_t & to, int id );

template< typename Request >
[[nodiscard]]
sender_t make_sender()
{
  return []( const so_5::mbox_t & to, int id ) {
    so_5::send< Request >( to, id );
  };
}


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

      // Trash can will work on its own dispatcher.
      auto trash_can = coop.make_agent_with_binder< a_trash_can_t >(
          so_5::disp::one_thread::make_dispatcher( env ).binder(),
          logger->so_direct_mbox() );

      // Processor will work on its own dispatcher.
      auto processor = coop.make_agent_with_binder< a_processor_t >(
          so_5::disp::one_thread::make_dispatcher( env ).binder(),
          logger->so_direct_mbox(),
          trash_can->so_direct_mbox() );

      return processor->so_direct_mbox();
    });

  // Senders for messages of various types.
  std::array< sender_t, 10 > senders{
    make_sender< req_A >(),
    make_sender< req_B >(),
    make_sender< req_C >(),
    make_sender< req_D >(),
    make_sender< req_E >(),
    make_sender< req_F >(),
    make_sender< req_G >(),
    make_sender< req_I >(),
    make_sender< req_A >(),
    make_sender< req_E >()
  };

  // Shuffle senders.
  {
    std::random_device rd;
    std::mt19937 generator{ rd() };

    std::shuffle( std::begin(senders), std::end(senders), generator );
  }

  // Do several interations of sending messages.
  int id = 0;
  for( int i = 0; i < 15; ++i )
  {
    for( const auto & s : senders )
      s( processor_mbox, id++ );

    std::this_thread::sleep_for( 13ms );
  }

  // Finish the example.
  so_5::send< a_processor_t::shutdown >( processor_mbox );
}

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

  return 0;
}