/*
   Тест производительности по мотивам сравнения Java-фреймворков.
   http://sujitpal.blogspot.com/2009/01/more-java-actor-frameworks-compared.html
*/

#include <iostream>

#include <cstdio>

#include <ace/ACE.h>
#include <ace/OS.h>
#include <ace/OS_main.h>
#include <ace/Time_Value.h>

#include <so_4/api/h/api.hpp>
#include <so_4/rt/h/rt.hpp>

#include <so_4/timer_thread/simple/h/pub.hpp>
#include <so_4/disp/one_thread/h/pub.hpp>
#include <so_4/disp/active_obj/h/pub.hpp>

#include <test/bench/time_value_msec_helper.hpp>

inline void
search_and_replace(
   const std::string & what,
   const std::string & to,
   std::string & where )
   {
      std::string::size_type p = where.find( what );
      if( std::string::npos != p )
         where.replace( p, to.size(), to );
   }

// Класс агента, который выполняет основную обработку.
class a_actor_t
   :  public so_4::rt::agent_t
   {
      typedef so_4::rt::agent_t base_type_t;

   public :
      a_actor_t(
         const std::string & self_name,
         const std::string & next_actor )
         :  base_type_t( self_name )
         ,  m_next_actor( next_actor )
         {
            so_4::disp::active_obj::make_active( *this );
         }

      virtual const char *
      so_query_type() const;

      // Сообщение, которое используется для передачи
      // информации между агентами.
      struct msg_process
         {
            std::string m_value;

            msg_process()
               {}
            msg_process(
               const char * value )
               :  m_value( value )
               {}
         };

      // Сообщение, которые используется для завершения работы.
      struct msg_stop {};

      virtual void
      so_on_subscription()
         {
            so_subscribe( "evt_process", "msg_process" );
            so_subscribe( "evt_stop", "msg_stop" );
         }

      void
      evt_process( const msg_process & cmd )
         {
            std::auto_ptr< msg_process > next( new msg_process() );
            act( cmd.m_value, next->m_value );
            if( !m_next_actor.empty() )
               so_4::api::send_msg_safely( m_next_actor, "msg_process", next );
         }

      void
      evt_stop()
         {
            if( !m_next_actor.empty() )
               so_4::api::send_msg( m_next_actor, "msg_stop" );
            else
               so_4::api::send_msg(
                     so_4::rt::sobjectizer_agent_name(),
                     "msg_normal_shutdown" );
         }

   protected :
      virtual void
      act(
         const std::string & source,
         std::string & target ) const = 0;

   private :
      const std::string m_next_actor;
   };

SOL4_CLASS_START( a_actor_t )
   SOL4_MSG_START( msg_process, a_actor_t::msg_process )
   SOL4_MSG_FINISH()

   SOL4_MSG_START( msg_stop, a_actor_t::msg_stop )
   SOL4_MSG_FINISH()

   SOL4_EVENT_STC( evt_process, a_actor_t::msg_process )
   SOL4_EVENT( evt_stop )

   SOL4_STATE_START( st_normal )
      SOL4_STATE_EVENT( evt_process )
      SOL4_STATE_EVENT( evt_stop )
   SOL4_STATE_FINISH()
SOL4_CLASS_FINISH()

class a_download_actor_t
   :  public a_actor_t
   {
   public :
      a_download_actor_t(
         const std::string & self_name,
         const std::string & next_actor )
         :  a_actor_t( self_name, next_actor )
         ,  m_what( "Requested " )
         ,  m_to( "Downloaded " )
         {}

   protected :
      virtual void
      act(
         const std::string & source,
         std::string & target ) const
         {
            target = source;
            search_and_replace( m_what, m_to, target );
         }

   private :
      const std::string m_what;
      const std::string m_to;
   };

class a_index_actor_t
   :  public a_actor_t
   {
   public :
      a_index_actor_t(
         const std::string & self_name,
         const std::string & next_actor )
         :  a_actor_t( self_name, next_actor )
         ,  m_what( "Downloaded " )
         ,  m_to( "Indexed " )
         {}

   protected :
      virtual void
      act(
         const std::string & source,
         std::string & target ) const
         {
            target = source;
            search_and_replace( m_what, m_to, target );
         }

   private :
      const std::string m_what;
      const std::string m_to;
   };

class a_write_actor_t
   :  public a_actor_t
   {
   public :
      a_write_actor_t(
         const std::string & self_name,
         const std::string & next_actor )
         :  a_actor_t( self_name, next_actor )
         ,  m_what( "Indexed " )
         ,  m_to( "Wrote " )
         {}

   protected :
      virtual void
      act(
         const std::string & source,
         std::string & target ) const
         {
            target = source;
            search_and_replace( m_what, m_to, target );
         }

   private :
      const std::string m_what;
      const std::string m_to;
   };

// Класс агента, который генерирует набор сообщений msg_process.
class a_actor_manager_t
   :  public so_4::rt::agent_t
   {
      typedef so_4::rt::agent_t base_type_t;
  
   public :
      a_actor_manager_t(
         const std::string & self_name,
         const std::string & first_actor,
         int message_count )
         :  base_type_t( self_name )
         ,  m_first_actor( first_actor )
         ,  m_message_count( message_count )
         {
            so_4::disp::active_obj::make_active( *this );
         }

      virtual const char *
      so_query_type() const;

      virtual void
      so_on_subscription()
         {
            so_subscribe( "evt_start",
                  so_4::rt::sobjectizer_agent_name(),
                  "msg_start" );
         }

      void
      evt_start()
         {
            for( int i = 0; i != m_message_count; ++i )
               {
                  char tmp[ 64 ];
                  std::sprintf( tmp, "Requested %d", i );

                  std::auto_ptr< a_actor_t::msg_process > msg(
                        new a_actor_t::msg_process( tmp ) );

                  so_4::api::send_msg_safely(
                        m_first_actor,
                        "msg_process",
                        msg );
               }

            so_4::api::send_msg( m_first_actor, "msg_stop" );
         }

   private :
      const std::string m_first_actor;
      const int m_message_count;
   };

SOL4_CLASS_START( a_actor_manager_t )
   SOL4_EVENT( evt_start )

   SOL4_STATE_START( st_normal )
      SOL4_STATE_EVENT( evt_start )
   SOL4_STATE_FINISH()
SOL4_CLASS_FINISH()

void
show_result(
   const ACE_Time_Value & start_time,
   const ACE_Time_Value & finish_time )
{
   double total_msec = milliseconds( finish_time ) -
         milliseconds( start_time );

   std::cout << "elapsed: " << total_msec / 1000.0 << std::endl;
}

int
main( int argc, char ** argv )
{
   a_actor_manager_t a_manager( "a_manager", "a_downloader", 1000000 );
   a_download_actor_t a_downloader( "a_downloader", "a_indexer" );
   a_index_actor_t a_indexer( "a_indexer", "a_writer" );
   a_write_actor_t a_writer( "a_writer", std::string() );

   so_4::rt::agent_t * agents[] =
   {
      &a_manager, &a_downloader, &a_indexer, &a_writer
   };

   so_4::rt::agent_coop_t coop( "test",
      agents, sizeof( agents ) / sizeof( agents[ 0 ] ) );

   ACE_Time_Value start_time = ACE_OS::gettimeofday();

   so_4::ret_code_t rc = so_4::api::start(
      // Диспетчер будет уничтожен при выходе из start().
      so_4::disp::active_obj::create_disp(
         // Таймер будет уничтожен диспетчером.
         so_4::timer_thread::simple::create_timer_thread(),
         so_4::auto_destroy_timer ),
      so_4::auto_destroy_disp,
      &coop );

   ACE_Time_Value finish_time = ACE_OS::gettimeofday();

   if( rc )
      std::cerr << "start: " << rc << std::endl;
   else
      show_result( start_time, finish_time );

   return int( rc );
}