L:/channel/cvm/examples/cvm_instance/tasks/ctrl/CtrlTask.cpp

Go to the documentation of this file.

// Copyright (c) 2005, 2006 Yigong Liu
// Permission to use, copy, modify, distribute and sell this software for any 
//     purpose is hereby granted without fee, provided that the above copyright 
//     notice appear in all copies and that both that copyright notice and this 
//     permission notice appear in supporting documentation.
// The author makes no representations about the 
//     suitability of this software for any purpose. It is provided "as is" 
//     without express or implied warranty.
 
#include <CtrlTask.h>
#include <CtrlTask_export.h>
#include <Task_Msg.h>

//----------------------------------------------
// the following 3 methods should be overwritten
// to implement application logic
//----------------------------------------------

//app hooks for resource reservation & cleanup
Status Ctrl_Task::prepare(void)
{  
  // Register ourselves to receive signals so we can shut down
  // gracefully.
  ACE_Sig_Set sig_set;
  sig_set.sig_add (SIGINT);
  sig_set.sig_add (SIGQUIT);
  sig_set.sig_add (SIGHUP);

  if (ACE_Reactor::instance ()->register_handler (sig_set,
                                                  this) == -1)
    ACE_ERROR_RETURN ((LM_ERROR,
                       "(%t) %p\n",
                       "register_handler"),
                      FAILURE);

  //publish msgs
  my_port()->publish_msg(TEST_STRING_MSG, SCOPE_GLOBAL);

  return SUCCESS;
}

Status Ctrl_Task::cleanup(void)
{
  //unsubscribe msg 
  my_port()->unpublish_msg(TEST_STRING_MSG);
  return SUCCESS;
}

void Ctrl_Task::showTestMenu()
{
  ACE_DEBUG((LM_DEBUG, "                  Channel Test Menu      \n"));
  ACE_DEBUG((LM_DEBUG, "     1. connect to remote node.\n"));
  ACE_DEBUG((LM_DEBUG, "     2. publish text message. \n"));
  ACE_DEBUG((LM_DEBUG, "     3. dump local routing tables. \n"));
  ACE_DEBUG((LM_DEBUG, "     4. quit \n\n"));
  ACE_DEBUG((LM_DEBUG, " Please enter :\n"));
  //ACE_OS::fflush(stdout);
}

int Ctrl_Task::work() {
  ACE_DEBUG ((LM_DEBUG,
              "(%t) %s ctrl_task coming up ...\n", my_name().c_str()));
  ACE_Reactor *reactor = ACE_Reactor::instance();

  UnixSockConnector *unix_conn = NULL;
  TcpSockConnector *tcp_conn = NULL;

  for(;;) {
    std::string conn_name;
    std::string user_input;
    showTestMenu();
    std::getline (std::cin, user_input, '\n');
    if (user_input == "4") {
      ::Quit_Handler *quit_handler = 0;
      ACE_NEW_RETURN (quit_handler, ::Quit_Handler (reactor), 0);
      reactor->notify (quit_handler);
      break;
    } else if(user_input == "3") {
      my_chan()->dump_routing_tables();
    } else if(user_input == "1") {
      ACE_DEBUG((LM_DEBUG, "--- Please enter Connector name: "));
      std::getline (std::cin, conn_name, '\n');
      ACE_DEBUG((LM_DEBUG, "--- Please enter destination (ip:port or unix_addr): "));
      std::getline (std::cin, user_input, '\n');
      ConnInfo ci(user_input);
      if(ci.valid()) {
        switch(ci.type()) {
        case INET_SOCK:
          tcp_conn = CvmTcpConnector::find_connector(conn_name.c_str());
          if (tcp_conn != NULL) 
            tcp_conn->connect(ci);
          else
            ACE_DEBUG((LM_DEBUG,  "Failed to find tcp connector [%s]", conn_name.c_str()));
          break;
        case UNIX_SOCK:
          unix_conn = CvmUnixConnector::find_connector(conn_name.c_str());
          if (unix_conn != NULL) 
            unix_conn->connect(ci);
          else
            ACE_DEBUG((LM_DEBUG,  "Failed to find unix connector [%s]", conn_name.c_str()));
          break;
        }
      }
    } else if(user_input == "2") {
      ACE_DEBUG((LM_DEBUG,  "--- Please enter one line msg: "));
      Test_String_Msg *tm =  new Test_String_Msg();
      if (!gets (tm->data))
        break;
      tm->len = strlen(tm->data)+1;
      ACE_DEBUG((LM_DEBUG,  "You entered: %s",tm->data));
      Msg *m = new Msg(TEST_STRING_MSG, tm);
      my_port()->send_msg (m);
    } 
  }

  ACE_DEBUG ((LM_DEBUG,
              "(%t) %s ctrl_task exits...\n", my_name().c_str()));

  return 0;
}

int
Ctrl_Task::handle_signal (int signum, siginfo_t *, ucontext_t *)
{
  switch(signum) {
    //SIGINT|SIGQUIT|SIGTERM cause proc/channel to exit, no need to broadcast to app tasks
  case SIGINT:
  case SIGQUIT:
    {
      ACE_Reactor *reactor = ACE_Reactor::instance ();
      // Shut down
      ::Quit_Handler *quit_handler = 0;
      ACE_NEW_RETURN (quit_handler, ::Quit_Handler (reactor), 0);
      reactor->notify (quit_handler);
    }
    break;
  default:  //other signal will be dropped
    {
      ACE_DEBUG((LM_DEBUG,  "--- recv signal[%d] , drpped ---\n",signum));
    }
    break;
  }
  return 0;
}


int Ctrl_Task::fini ()
{ 
  ACE_DEBUG((LM_DEBUG, "(%t) Task [%s] fini...\n", my_name().c_str()));

  //we still need wait some time for OS cleanup Task threads
  ACE_OS::sleep(1);

  //clean up port
  delete my_port_;

  return 0;
}


ACE_FACTORY_DEFINE (CtrlTask, Ctrl_Task)

---