sysconf/main.cpp

/*
  Generic MONitoring Tools 1
*/

/*
  Пример использования SO SysConf.
*/

#include <iostream>

#include <ace/OS_main.h>

#include <cpp_util_2/h/defs.hpp>
#include <cpp_util_2/h/lexcast.hpp>

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

#include <so_4/transport_layer/socket/h/pub.hpp>
#include <so_4/rt/comm/h/a_sop_incoming_channel_processor.hpp>

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

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

#include <so_sysconf_2/h/pub.hpp>
#include <so_sysconf_2/h/breakflag.hpp>

#include <gemont_1/h/pub.hpp>
#include <gemont_1/h/temporary_sources.hpp>
#include <gemont_1/h/value_holder.hpp>

/*
  Агент, который с переодически изменяет значение
  своих источников мониторинговых данных.
*/
class a_data_source_t :
  public virtual so_4::rt::agent_t
{
  typedef so_4::rt::agent_t agent_type_t;
  public :
    a_data_source_t();
    virtual ~a_data_source_t();

    virtual const char *
    so_query_type() const;

    virtual void
    so_on_subscription();

    struct  msg_tick {};

    void
    evt_start();

    void
    evt_tick();

  private :
    // Текущее значение.
    unsigned int  m_ticker;
    // Еще один счетчик.
    gemont_1::value_holder_t< unsigned int >  m_ticker_2;
    // И еще одно изменяемое значение.
    gemont_1::value_holder_t< std::string > m_appender;
    // И еще одно изменяемое значение, которое будет создано
    // автоматически и при этом сразу в запущеном состоянии.
    std::auto_ptr<
      gemont_1::value_holder_t< std::string > > m_dynamic_appender;

    // Этот value_holder так же является свойством агента.
    gemont_1::value_holder_as_trait_t< unsigned int > m_cycles;

    // Источник данных, который будет информировать о том,
    // является ли текущее значение четным.
    gemont_1::str_data_source_t m_parity;

    // Текущее значение счетчика.
    gemont_1::uint_data_source_t  m_ticker_monitor;
    // Удвоенное текущее значение счетчика.
    gemont_1::uint_data_source_t  m_2x_ticker_monitor;
    // Утроенное текущее значение счетчика.
    gemont_1::uint_data_source_t  m_3x_ticker_monitor;

    // Хранилище временных источников данных.
    // Элементы помещаются сюда, когда m_3x_ticker
    // превосходит очередной десяток.
    gemont_1::temporary_sources_t<
        gemont_1::uint_data_source_t,
        unsigned int >
      m_3x_tens;

    // Счетчик, значение которого увеличивается на 2.
    gemont_1::scalar_data_source_t< unsigned int >  m_2plus_ticker;
    // Счетчик, значение которого увеличивается на 3 и отображается
    // в виде строки. Источник данных создается после начала работы
    // агента.
    std::auto_ptr<
      gemont_1::scalar_data_source_t< std::string > > m_3plus_ticker;

    // Интенсивно изменяемый источник данных.
    gemont_1::uint_data_source_t  m_frequently_changed;
};

SOL4_CLASS_START( a_data_source_t )

  SOL4_MSG_START( msg_tick, a_data_source_t::msg_tick )
  SOL4_MSG_FINISH()

  SOL4_EVENT( evt_start )
  SOL4_EVENT( evt_tick )

  SOL4_STATE_START( st_normal )
    SOL4_STATE_EVENT( evt_start )
    SOL4_STATE_EVENT( evt_tick )
  SOL4_STATE_FINISH()

SOL4_CLASS_FINISH()

a_data_source_t::a_data_source_t()
:
  agent_type_t( "a_data_source" )
, m_ticker( 0 )
, m_ticker_2( "Тикер N2", "ticker", 0 )
, m_appender( "Appender", "appender", "a" )
, m_cycles( "Циклы", "cycles", 0 )
, m_parity( "Четность", "parity", "НЕ ИЗВЕСТНО" )
, m_ticker_monitor( "Тикер", "ticker", m_ticker )
, m_2x_ticker_monitor( "Тикер * 2", "2x_ticker", 2 * m_ticker )
, m_3x_ticker_monitor( "Тикер * 3", "3x_ticker", 3 * m_ticker )
, m_2plus_ticker( "Тикер + 2", "2plus_ticker", 2 + m_ticker )
, m_frequently_changed( "Frequently changed", "frequently_changed", 0 )
{
  so_add_traits( m_cycles );
  so_add_traits( m_parity );
  so_add_traits( m_ticker_monitor );
  so_add_traits( m_2x_ticker_monitor );
  so_add_traits( m_3x_ticker_monitor );
  so_add_traits( m_3x_tens );
  so_add_traits( m_frequently_changed );
}

a_data_source_t::~a_data_source_t()
{
}

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

  so_subscribe( "evt_tick", "msg_tick" );

}

void
a_data_source_t::evt_start()
{
  so_4::api::send_msg( so_query_name(), "msg_tick", 0,
    so_query_name(), 1500, 1500 );

  m_2plus_ticker.start();
  m_3plus_ticker.reset(
    new gemont_1::scalar_data_source_t< std::string >(
      "Тикер + 3", "3plus_ticker",
      cpp_util_2::slexcast( m_ticker ) + " + 3",
      &gemont_1::auto_start ) );

  m_ticker_2.start();
  m_appender.start();
  m_dynamic_appender.reset(
    new gemont_1::value_holder_t< std::string >(
      "Динамический appender", "appender", "d",
      &gemont_1::auto_start ) );
}

void
a_data_source_t::evt_tick()
{
  std::cout << "m_ticker: " << m_ticker << std::endl;

  m_parity.set( ( m_ticker & 1 ? "Не четное" : "Четное" ) );
  m_ticker_monitor.set( m_ticker );
  m_2x_ticker_monitor.set( 2 * m_ticker );

  unsigned int ticker_3x = 3 * m_ticker;
  m_3x_ticker_monitor.set( ticker_3x );
  // Если превзошли очередной десяток, то создадим новый
  // источник данных. В противном случае увеличим уже существующий.
  unsigned int t = ticker_3x / 10;
  gemont_1::uint_data_source_t * tens = m_3x_tens.find( t );
  if( tens )
    tens->set( tens->current() + 1 );
  else
  {
    m_3x_tens.insert( t,
      new gemont_1::uint_data_source_t(
        std::string( "tens " ) + cpp_util_2::slexcast( t ),
        "3x_ticker_tens", 1 ) );
  }

  m_2plus_ticker.set( 2 + m_ticker );
  m_3plus_ticker->set( cpp_util_2::slexcast( m_ticker ) + " + 3" );

  // Инициируем массовый поток мониторинговой инфорамации.
  for( unsigned int i = 0, i_max = m_3x_ticker_monitor.current();
      i <= i_max; ++i )
  {
    m_frequently_changed.set( i );
  }

  ++m_ticker;
  ++(**m_ticker_2);
  if( m_ticker > 45 )
  {
    m_ticker = 0;
    m_3x_tens.clear();
    (**m_ticker_2) = 0;
    ++(**m_cycles);
        // Дерегистрируем кооперацию, чтобы проверить реакцию Qt_widget на
        // дерегистрацию источников данных
//    so_4::api::deregister_coop( "a_data_source" );
  }

  (***m_dynamic_appender).append( "d" );
  (**m_appender).append( "a" );
  if( 20 < (**m_appender).length() )
    {
      (**m_appender) = "a";
      (***m_dynamic_appender) = "d";
    }
}

// Имя коммуникационного агента.
const std::string comm_channel_agent( "a_srv_channel" );

//
// a_main_t
//

// Главный стартовый агент.
/*
  Класс агента a_main, отвечающего за старт примера.
*/
class a_main_t
  : public so_4::rt::agent_t
{
  public :
    a_main_t(
      /* Имя стартового конфигурационного файла для
        подсистемы sysconf */
      const std::string & cfg_file_name );
    virtual ~a_main_t();

    virtual const char *
    so_query_type() const;

    virtual void
    so_on_subscription();

    // По этому сообщению создается тестовый источник данных.
    struct  msg_make_data_source {};

    // Реакция на успешное создание коммуникационного сокета.
    /*
      \par Приоритет:
      0.
    */
    void
    evt_comm_socket_creation_ok();

    // Реакция на неудачное создание коммуникационного сокета.
    /*
      \par Приоритет:
      0
    */
    void
    evt_comm_socket_creation_fail(
      const so_4::rt::comm::msg_fail & cmd );

    // Нужно создать источник данных.
    void
    evt_make_data_source();

  private :
    // Псевдоним для базового типа.
    typedef so_4::rt::agent_t base_type_t;

    /* Имя стартового конфигурационного файла для
      подсистемы sysconf */
    std::string m_cfg_file_name;
};// class a_main_t


// Описание класса для SObjectizer-а
SOL4_CLASS_START( a_main_t )

  SOL4_MSG_START( msg_make_data_source,
    a_main_t::msg_make_data_source )
  SOL4_MSG_FINISH()

  SOL4_EVENT(
    evt_comm_socket_creation_ok )
  SOL4_EVENT_STC(
    evt_comm_socket_creation_fail,
    so_4::rt::comm::msg_fail )
  SOL4_EVENT( evt_make_data_source )

  SOL4_STATE_START( st_initial )
    SOL4_STATE_EVENT( evt_comm_socket_creation_ok )
    SOL4_STATE_EVENT( evt_comm_socket_creation_fail )
    SOL4_STATE_EVENT( evt_make_data_source )
  SOL4_STATE_FINISH()

SOL4_CLASS_FINISH()

// Реализация класса
a_main_t::a_main_t( const std::string & cfg_file_name )
  : base_type_t( "a_main" )
  , m_cfg_file_name( cfg_file_name )
{
}

a_main_t::~a_main_t()
{
}

void
a_main_t::so_on_subscription()
{
  so_subscribe( "evt_comm_socket_creation_ok",
    comm_channel_agent, "msg_success" );

  so_subscribe( "evt_comm_socket_creation_fail",
    comm_channel_agent, "msg_fail" );

  so_subscribe( "evt_make_data_source", "msg_make_data_source" );
}

void
a_main_t::evt_comm_socket_creation_ok()
{
  // Создаем подсистему sysconf.
  so_sysconf_2::register_coop();
  // Обрабатываем стартовый конфигурационный файл.
  if( !so_sysconf_2::run_script( m_cfg_file_name ) )
  {
    // Ошибка разбора стартового конфигурационного файла.
    // Система не запущена и должна быть остановлена.
    so_4::api::send_msg(
      so_4::rt::sobjectizer_agent_name(),
      "msg_normal_shutdown" );
  }

  // Источник данных нужно зарегистрировать чуть позже,
  // иначе не успеют загеристрироваться агенты из
  // gemont_1.
  so_4::api::send_msg( so_query_name(), "msg_make_data_source", 0,
    so_query_name(), 250 );
}

void
a_main_t::evt_comm_socket_creation_fail(
  const so_4::rt::comm::msg_fail & cmd )
{
  std::cerr << "Creation of communication socket failed: "
    << cmd.m_reason << std::endl;

  // Завершаем работу системы
  so_4::api::send_msg(
    so_4::rt::sobjectizer_agent_name(),
    "msg_alarm_shutdown" );
}

void
a_main_t::evt_make_data_source()
{
  // Регистрируем агента, который будет производить информацию.
  so_4::rt::dyn_agent_coop_helper_t helper(
    new so_4::rt::dyn_agent_coop_t(
      new a_data_source_t() ) );
}

// Создание стартовой кооперации.
so_4::rt::agent_coop_t *
create_main_coop(
  const std::string & ip_address,
  const std::string & cfg_file_name)
{
  a_main_t *  a_main = new a_main_t( cfg_file_name );

  // Коммуникационный сокет для управления модулем.
  using namespace so_4::rt::comm;
  using namespace so_4::transport_layer;
  using namespace so_4::transport_layer::socket;
  std::auto_ptr< a_sop_incoming_channel_processor_t > a_socksrv(
    new a_sop_incoming_channel_processor_t(
      comm_channel_agent,
      create_acceptor_controller(
        acceptor_params( ip_address ) ) )
  );
  // Должен быть активным объектом.
  so_4::disp::active_obj::make_active( *a_socksrv );

  so_4::rt::agent_t * main_coop_agents[] =
  {
    a_main, a_socksrv.release()
  };

  // Регистрируем кооперацию.
  so_4::rt::dyn_agent_coop_t * main_coop =
    new so_4::rt::dyn_agent_coop_t(
      "a_main", main_coop_agents,
      CPP_UTIL_2_ASIZE( main_coop_agents ) );

  return main_coop;
}

int
main( int argc, char ** argv )
{
  if( 2 == argc )
  {
    so_sysconf_2::setup_signal_handlers();

    so_4::ret_code_t rc = so_4::api::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,
      create_main_coop( argv[ 1 ], "etc/sysconf.cfg" ) );

    if( rc )
      std::cerr << rc << std::endl;

    return rc.m_code;
  }
  else {
    std::cerr << "sample_sysconf <ip_address>\n\n"
      "Configuration file sysconf.cfg must exist in "
      "directory ./etc" << std::endl;
    return 1;
  }
}