S3FC project page

S3FC home page

---

               

---

s3_rpc.tcc

Go to the documentation of this file.

#include <s3fc/s3_rpc.h>
#include <s3fc/s3_conversion.h>

//
template<typename T_obj>
void s3_rpc_type_common<T_obj>::clear_callmap(callmap_t& cmap)
{
   // clear the callmap
   while (! cmap.empty())
   {
      // deallocate storage 
      delete cmap.begin()->second;
      cmap.erase(cmap.begin());
   }
}

//
template<typename T_obj>
s3_rpc_exception s3_rpc_caller_m<T_obj>::err_invalid_args(
   const std::string& where,
   const std::vector<std::string>& args)
{
   std::stringstream ss;
   ss << "Invalid arguments: '";
   if (args.size() > 0)
   {
      ss << args[0];
      for (unsigned int i = 1; i < args.size(); ++i)
      {
    ss << "|" << args[1];
      }
   }
   ss << "'";
   return s3_rpc_exception(where, ss.str());
}

//
template<typename T_obj>
s3_rpc_caller_m_v_v<T_obj>::s3_rpc_caller_m_v_v(
   methodptr_t n_method) :
   method(n_method) 
{
}

//
template<typename T_obj>
void s3_rpc_caller_m_v_v<T_obj>::call(
   T_obj* obj,
   const std::vector<std::string>& args)
{
   S3FC_DBG(
      std::cerr << "s3_rpc_caller_m_v<T_obj>::call()"
      << " obj: " << obj
      << " method: " << method
      << std::endl << std::flush;
      );

   if (args.size() != 0)
   {
      throw err_invalid_args(
    "s3_rpc_caller_m_v_t1::call()",
    args);
   }
   (obj->*method)();

   S3FC_DBG(
   std::cerr << "s3_rpc_caller_m_v<T_obj>::call() done"
        << std::endl << std::flush;
      );
}

//
template<typename T_obj>
unsigned int s3_rpc_caller_m_v_v<T_obj>::get_num_params()
{
   return 0;
}

//
template<typename T_obj, typename T_arg1>
s3_rpc_caller_m_v_t1<T_obj, T_arg1>::s3_rpc_caller_m_v_t1(
   methodptr_t n_method) :
   method(n_method) 
{
}

//
template<typename T_obj, typename T_arg1>
void s3_rpc_caller_m_v_t1<T_obj, T_arg1>::call(
   T_obj* obj,
   const std::vector<std::string>& args)
{
   if (args.size() != 1)
   {
      throw err_invalid_args(
    "s3_rpc_caller_m_v_t1::call()",
    args);
   }
   (obj->*method)(s3_conversion::from_string<T_arg1>(args[0]));
}

//
template<typename T_obj, typename T_arg1>
unsigned int s3_rpc_caller_m_v_t1<T_obj, T_arg1>::get_num_params()
{
   return 1;
}

//
template<typename T_obj, typename T_arg1, typename T_arg2>
s3_rpc_caller_m_v_t2<T_obj, T_arg1, T_arg2>::s3_rpc_caller_m_v_t2(
   methodptr_t n_method) :
   method(n_method) 
{
}

//
template<typename T_obj, typename T_arg1, typename T_arg2>
void s3_rpc_caller_m_v_t2<T_obj, T_arg1, T_arg2>::call(
   T_obj* obj,
   const std::vector<std::string>& args)
{
   if (args.size() != 2)
   {
      throw err_invalid_args(
    "s3_rpc_caller_m_v_t2::call()",
    args);
   }
   (obj->*method)(s3_conversion::from_string<T_arg1>(args[0]),
        s3_conversion::from_string<T_arg2>(args[1]));
}

//
template<typename T_obj, typename T_arg1, typename T_arg2>
unsigned int s3_rpc_caller_m_v_t2<T_obj, T_arg1, T_arg2>::get_num_params()
{
   return 2;
}

//
template<typename T_obj>
s3_rpc_server<T_obj>::s3_rpc_server(const std::string& n_name,
                T_obj& n_obj) :
   obj(&n_obj),
   cmap(s3_rpc_callmap<T_obj>::value())
{
   std::string lname = s3_rpc_common::rpc_server_name(n_name);
   try
   {
      // init message box
      mb.set_name(lname);
      mb.connect();
      // start myself
      s3_thread_base::start();
      started.wait();
   }
   catch(const std::exception& e)
   {
      std::stringstream ss;
      ss << "s3_message_box::connect() threw exception (name: "
    << lname << "): [" << e.what() << "]";
      // rethrow as RPC exception
      throw s3_rpc_exception("s3_rpc_server::s3_rpc_server",
              ss.str());
   }
}

//
template<typename T_obj>
s3_rpc_server<T_obj>::~s3_rpc_server()
{
   // stop the main loop
   s3_thread_base::request_terminate();
   s3_thread_base::wait_on_exit();
   mb.disconnect();
   // deallocate the callmap storage avoid leaks
   s3_rpc_type_common<T_obj>::clear_callmap(cmap);
}

//
template<typename T_obj>
void s3_rpc_server<T_obj>::main_loop()
{
   started.post();
   // notify to state_changed when message received
   mb.subscribe_arrival(state_changed);
   while(! test_terminate())
   {
      wait_on_state_changed();
      S3FC_DBG(
    std::cerr << "s3_rpc_server<T_obj>::main_loop(): state_changed"
    << std::endl << std::flush;
    );
      handle_message(mb.get_msg());
   }
}


//
template<typename T_obj>
void s3_rpc_server<T_obj>::handle_message(const s3_message& msg)
{
   if (! msg.is_empty()) 
   {
      std::string body = msg.get_body();
      // extract and handle command
      if (body.length() > 0)
      {
    std::string cmd = remove_first(body);
    // RPC call command
    if (cmd == s3_rpc_common::cmd_tag_call)
    {
       exec_cmd_call(msg, body);
    }
    // invalid command - send reply
    else
    {
       std::stringstream ss;
       ss << "s3_rpc_server::handle_message: RPC message error: "
          << "[Invalid RPC command: '" << cmd << "']";
       reply_client(mb, msg, ss.str(), true);
    }
      }
      // no command
      else
      {
    reply_client(mb, msg,
            "s3_rpc_server::handle_message: RPC message error: "
            "[No RPC command]",
            true);
      }
   }
}

//
template<typename T_obj>
void s3_rpc_server<T_obj>::exec_cmd_call(
   const s3_message& msg,
   std::string body)
{
   // extract and handle method
   if (body.length() > 0)
   {
      std::string method = remove_first(body);
      // 
      if (cmap.find(method) != cmap.end())
      {
    try
    {
       std::vector<std::string> body_parts =
          s3_conversion::split_string(body, "|");
       cmap[method]->call(obj, body_parts);
       reply_client(mb, msg, "", false);
    }
    // specialised rpc caller rejected call - probably due to incorrect
    // number of arguments
    catch(const std::exception& e)
    {
       std::stringstream ss;
       ss << "s3_rpc_server::handle_message: RPC message error: ["
          << e.what() << "]";
       reply_client(mb, msg, ss.str(), true);
    }
      }
      // unknown method (i.e. not in callmap)
      else
      {
    std::stringstream ss;
    ss << "s3_rpc_server::handle_message: RPC message error: ["
       << "Invalid RPC method: '" << method << "']";
    reply_client(mb, msg, ss.str(), true);
      }
   }
}

//
template<typename T_obj>
std::string s3_rpc_server<T_obj>::remove_first(std::string& str)
{
   std::string rv;
   
   const char delim = '|';
   std::string::size_type di = str.find_first_of(delim);
   // found delimeter - return first string
   if (di != std::string::npos)
   {
      // extract first part (excluding delimeter) as return value
      rv.assign(str, 0, di);
      // remove extracted part + delimiter from str
      str.erase(0, di+1);
   }
   else
   {
      // extract entire string as return value
      rv = str;
      // clear input string
      str = "";
   }
   return rv;
}

//
template<typename T_obj>
void s3_rpc_server<T_obj>::reply_client(s3_message_box& mb,
               const s3_message& msg,
               const std::string& reply,
               bool err)
{
   std::stringstream ss;
   if (err)
   {
      ss << "ERROR|" << reply;
   }
   else
   {
      ss << "OK|" << reply;
   }
   mb.reply_msg(msg, ss.str());

   S3FC_DBG(
      std::cerr << "s3_rpc_server<T_obj>::reply_client: " << std::endl
      << "Original: " << std::endl
      << " From: " << msg.get_from() << std::endl
      << " Body: " << msg.get_body() << std::endl
      << "Reply: " << std::endl
      << " Body: " << ss.str() << std::endl;
      );
}

//
template<typename T_obj>
s3_rpc_client<T_obj>::s3_rpc_client(const std::string& n_name) :
   server_name(s3_rpc_common::rpc_server_name(n_name)),
   cmap(s3_rpc_callmap<T_obj>::value())
{
   std::string lname = s3_rpc_common::rpc_random_client_name(n_name);
   try
   {
      mb.set_name(lname);
      mb.connect();
   }
   catch(const std::exception& e)
   {
      std::stringstream ss;
      ss << "s3_message_box::connect() threw exception (name: "
    << lname << "): [" << e.what() << "]";
      // rethrow as RPC exception
      throw s3_rpc_exception("s3_rpc_client::s3_rpc_client",
              ss.str());
   }
}

//
template<typename T_obj>
s3_rpc_client<T_obj>::~s3_rpc_client()
{
   mb.disconnect();
   s3_rpc_type_common<T_obj>::clear_callmap(cmap);
}

//
template<typename T_obj>
std::string s3_rpc_client<T_obj>::call(const std::string& method)
{
   assert_method_args("s3_rpc_client<T_obj>::call()",
            method, 0);
   return call_common(method, "");
}

//
template<typename T_obj> template<typename T_arg1>
std::string s3_rpc_client<T_obj>::call(const std::string& method,
                   const T_arg1& arg1)
{
   assert_method_args("s3_rpc_client<T_obj>::call(T1)",
            method, 1);

   std::stringstream ss;
   ss << s3_conversion::to_string<T_arg1>(arg1) << "|";
   return call_common(method, ss.str());
}

//
template<typename T_obj>  template<typename T_arg1, typename T_arg2>
std::string s3_rpc_client<T_obj>::call(const std::string& method,
                   const T_arg1& arg1,
                   const T_arg2& arg2)
{
   assert_method_args("s3_rpc_client<T_obj>::call(T1,T2)",
            method, 2);

   std::stringstream ss;
   ss << s3_conversion::to_string<T_arg1>(arg1) << "|";
   ss << s3_conversion::to_string<T_arg2>(arg2) << "|";
   return call_common(method, ss.str());
}

//
template<typename T_obj>
std::string s3_rpc_client<T_obj>::call_common(const std::string& method,
                   const std::string& args)
{
   std::stringstream ss;
   ss << s3_rpc_common::cmd_tag_call << "|" << method << "|" << args;
   mb.send_msg(server_name, ss.str());
   return "";
}

template<typename T_obj>
void s3_rpc_client<T_obj>::assert_method_args(const std::string& where,
                     const std::string& method,
                     unsigned int num_params)
{
   // invalid method ?
   if (cmap.find(method) == cmap.end())
   {
      std::stringstream ss;
      ss << "Invalid method: '" << method << "'";
      throw s3_rpc_exception(where, ss.str());
   }
   if (cmap[method]->get_num_params() != num_params)
   {
      std::stringstream ss;
      ss << "Invalid number of parameters to method '" << method
    << "': " << num_params << " (" <<
    cmap[method]->get_num_params() << " expected)";
      throw s3_rpc_exception(where, ss.str());
   }
}

---

Send comments to: s3fc@stonethree.com