Caller.h

Go to the documentation of this file.

  /* Copyright 2006-2013 Savarese Software Research Corporation.
   *
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   *
   * http://www.savarese.com/software/ApacheLicense-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  #ifndef __SSRC_WISP_PROTOCOL_CALLER_H
  #define __SSRC_WISP_PROTOCOL_CALLER_H
  
  #include <ssrc/spread.h>
  #include <ssrc/wisp/protocol/ServiceProtocol.h>
  
  #include <queue>
  #include <sstream>
  #include <stdexcept>
  
  #include <boost/shared_ptr.hpp>
  #include <boost/multi_index_container.hpp>
  #include <boost/multi_index/mem_fun.hpp>
  #include <boost/multi_index/hashed_index.hpp>
  #include <boost/multi_index/composite_key.hpp>
  
  __BEGIN_NS_SSRC_WISP_PROTOCOL
  
  using std::string;
  using NS_SSRC_SPREAD::GroupList;
  using NS_SSRC_SPREAD::Message;
  using NS_SSRC_SPREAD::Mailbox;
  
  const Message::Service DefaultMessageServiceType = Message::SafeSelfDiscard;
  
  enum CallType {
   OneWay,
  
   TwoWay
  
   /*, MultiWay */
  };
  
  namespace detail {
   struct VoidReturnType { };
  }
  
  template<typename CallerType, typename ParameterType,
   typename ReturnType = detail::VoidReturnType>
  struct CallTraits {
   static const CallType call_type = TwoWay;
  
   typedef ParameterType parameter_type;
  
   typedef ReturnType return_type;
  
   typedef CallerType caller_type;
  
   typedef typename caller_type::template Future<return_type> future_type;
  
   typedef typename future_type::shared_ptr future_ptr;
  };
  
  template<typename CallerType, typename ParameterType>
  struct CallTraits<CallerType, ParameterType, detail::VoidReturnType> {
   static const CallType call_type = OneWay;
  
   typedef ParameterType parameter_type;
  
   typedef detail::VoidReturnType return_type;
  
   typedef CallerType caller_type;
  };
  
  #define WISP_ONE_WAY_CALL(caller, method) \
   typedef NS_SSRC_WISP_PROTOCOL::CallTraits<caller, Message ## method> \
   Call ## method
  
  #define WISP_TWO_WAY_CALL(caller, method, result) \
   typedef NS_SSRC_WISP_PROTOCOL::CallTraits<caller, Message ## method, Message ## result> \
   Call ## method
  
  class CallException : public std::runtime_error {
  public:
   explicit CallException(const string & message) :
   std::runtime_error(message)
   { }
  };
  
  typedef std::uint32_t wisp_call_token;
  
  typedef std::uint8_t wisp_call_role;
  
  
  enum CallRole {
   OneWayRequest,
  
   TwoWayRequest,
  
   TwoWayResponse
  
   /*, MultiWayResponse */
  };
  
  
  struct CallHeader {
   wisp_message_id id;
   wisp_call_token token;
   wisp_call_role role;
   unsigned int message_size;
  
   CallHeader() : id(0), token(0), role(0), message_size(0) { }
  
   CallHeader(const wisp_message_id id,
   const wisp_call_token token,
   const CallRole role,
   const unsigned int size = 0) :
   id(id), token(token), role(role), message_size(size) { }
  
   void clear() {
   id = 0, token = 0, role = 0, message_size = 0;
   }
  
   void init(const wisp_message_id _id,
   const wisp_call_token _token,
   const CallRole _role,
   const unsigned int _size = 0)
   {
   id = _id, token = _token, role = _role, message_size = _size;
   }
  
   bool is_jumbo_fragment() const {
   return (message_size > 0);
   }
  
   template<class Archive>
   void serialize(Archive & ar, const unsigned int) {
   ar & id & token & role & message_size;
   }
  };
  
  struct MessageInfo {
   CallHeader header;
   Message message;
   GroupList groups;
  
   explicit
   MessageInfo(const unsigned int message_capacity = Message::DefaultCapacity) :
   header(), message(message_capacity), groups()
   { }
  
   wisp_message_protocol protocol() const {
   return message.type();
   }
  
   wisp_message_id id() const {
   return header.id;
   }
  
   wisp_call_token token() const {
   return header.token;
   }
  
   wisp_call_role role() const {
   return header.role;
   }
  
   const string & sender() const {
   return message.sender();
   }
  
   void init_header(const wisp_message_id id,
   const wisp_call_token token,
   const CallRole role)
   {
   header.init(id, token, role);
   }
  
   void clear_header() {
   header.clear();
   }
  };
  
  typedef boost::shared_ptr<MessageInfo> message_info_ptr;
  
  const bool GroupMembershipEnable = true;
  const bool GroupMembershipDisable = false;
  
  template<typename PT = BinaryPackingTraits>
  class Caller {
   template<typename ReturnType> friend class Future;
  
  public:
   typedef PT packing_traits;
   typedef typename packing_traits::packer_type packer_type;
   typedef typename packing_traits::unpacker_type unpacker_type;
  
   // Don't remove this definition. It is necessary to prevent an
   // unresolved MaxUnfragmentedMessageSize symbol in debug builds.
  #define __WISP_CALLER_MAX_UNFRAGMENTED_MESSAGE_SIZE 131072U
   static const unsigned int MaxUnfragmentedMessageSize =
   __WISP_CALLER_MAX_UNFRAGMENTED_MESSAGE_SIZE;
  
  private:
   typedef std::deque<message_info_ptr> request_queue;
  
   // For now, use hashed_unique, but for multiple responses will need
   // hashed_non_unique.
   typedef boost::multi_index_container<
   message_info_ptr,
   boost::multi_index::indexed_by<
   boost::multi_index::hashed_unique<
   boost::multi_index::const_mem_fun<MessageInfo, wisp_call_token,
   &MessageInfo::token> >
   > > response_map;
  
   typedef boost::multi_index_container<
   message_info_ptr,
   boost::multi_index::indexed_by<
   boost::multi_index::hashed_unique<
   boost::multi_index::composite_key<
   MessageInfo,
   boost::multi_index::const_mem_fun<MessageInfo, const string &,
   &MessageInfo::sender>,
   boost::multi_index::const_mem_fun<MessageInfo, wisp_call_role,
   &MessageInfo::role>,
   boost::multi_index::const_mem_fun<MessageInfo, wisp_call_token,
   &MessageInfo::token> > >
   > > jumbo_message_map;
  
   Mailbox _mbox;
   wisp_call_token _call_token;
   packer_type _packer;
   unpacker_type _unpacker;
  
   request_queue _requests;
   response_map _responses;
   jumbo_message_map _jumbo_messages;
   jumbo_message_map::key_from_value _jumbo_key_from_value;
  
   const unsigned int _initial_message_capacity;
   // scratch variable
   message_info_ptr _info;
   MessageInfo _send_info;
  
   static void throwCallException(const wisp_message_protocol expected_proto,
   const wisp_message_protocol actual_proto,
   const wisp_message_id expected_id,
   const wisp_message_id actual_id)
   SSRC_DECL_THROW(CallException)
   {
   std::ostringstream message;
   message << "Mismatched message id or protocol.\nExpected protocol: "
   << expected_proto << " Actual protocol: " << actual_proto
   << "\nExpected id: " << expected_id
   << " Actual id: " << actual_id << std::endl;
   throw CallException(message.str());
   }
  
   static void throwCallException(const wisp_call_token expected_token,
   const wisp_call_token actual_token)
   SSRC_DECL_THROW(CallException)
   {
   std::ostringstream message;
   message << "Mismatched call token.\nExpected token: "
   << expected_token << " Actual token: " << actual_token
   << std::endl;
   throw CallException(message.str());
   }
  
   void reset_receive_info() {
   _info.reset(new MessageInfo(std::min(_info->message.capacity(), __WISP_CALLER_MAX_UNFRAGMENTED_MESSAGE_SIZE)));
   }
  
   void push_request() {
   _requests.push_back(_info);
   reset_receive_info();
   }
  
   void insert_response() {
   _responses.insert(_info);
   reset_receive_info();
   }
  
   enum JumboFragmentResult {
   JumboFragmentFirstFragment,
   JumboFragmentNextFragment,
   JumboFragmentCompleteRequest,
   JumboFragmentCompleteResponse
   };
  
   JumboFragmentResult insert_jumbo_fragment(const message_info_ptr & msginfo,
   const unsigned int header_size)
   {
   JumboFragmentResult result = JumboFragmentNextFragment;
   Message & message = msginfo->message;
   jumbo_message_map::iterator it =
   _jumbo_messages.find(_jumbo_key_from_value(msginfo));
  
   if(it == _jumbo_messages.end()) {
   // Pre-allocate memory for remaining fragments.
   message_info_ptr info(new MessageInfo(msginfo->header.message_size));
   message.seek(message.size());
   *info = *msginfo;
   _jumbo_messages.insert(info);
   result = JumboFragmentFirstFragment;
   } else {
   const message_info_ptr & info = *it;
  
   info->message.write(&message[header_size], message.size()-header_size);
  
   if(info->message.offset() >= info->header.message_size) {
   info->message.seek(header_size);
   if(info->role() == TwoWayResponse) {
   _responses.insert(info);
   result = JumboFragmentCompleteResponse;
   } else {
   _requests.push_back(info);
   result = JumboFragmentCompleteRequest;
   }
   _jumbo_messages.erase(it);
   }
   }
   return result;
   }
  
   // Receives call responses, but not requests and membership messages
   template<typename MessageType>
   void receive_response(MessageType & msg, const wisp_call_token token)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure,
   CallException)
   {
   begin_receive_response:
   typename response_map::iterator it(_responses.find(token));
  
   if(it == _responses.end()) {
   if(!_info.unique()) {
   reset_receive_info();
   }
   do {
   Message & message = _info->message;
  
   _mbox.receive(message);
   _mbox.copy_groups(_info->groups);
  
   if(message.is_regular()) {
   CallHeader & header = _info->header;
  
   message.rewind();
  
   const unsigned int header_size = _unpacker.unpack(header, message);
  
   if(header.is_jumbo_fragment()) {
   const JumboFragmentResult jumbo_result =
   insert_jumbo_fragment(_info, header_size);
   if(jumbo_result == JumboFragmentCompleteResponse) {
   goto begin_receive_response;
   }
   } else {
   if(header.role == TwoWayResponse) {
   if(header.token == token)
   break;
   else
   insert_response();
   } else
   push_request();
   }
   } else {
   _info->clear_header();
   push_request();
   }
   } while(true);
   unpack(msg, *_info);
   } else {
   unpack(msg, *(*it));
   _responses.erase(it);
   }
   }
  
   // We assume integers wrap around to min at max.
   wisp_call_token next_token() {
   return _call_token++;
   }
  
   template<typename DestinationType, typename MessageType>
   void send(const DestinationType & dest,
   const MessageType & msg,
   const wisp_call_token token,
   const CallRole role,
   const Message::Service service)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
   {
   bool done_packing(false);
  
   _send_info.init_header(msg.id, token, role);
  
   // We increase the message buffer size by a factor of two and try
   // again when the write area is exhausted. We don't try to
   // prevent an infinite loop because the message we are writing is
   // finite in size; therefore the loop must terminate. See WISP-11
   // for more information.
  
  #define __WISP_WRITE_AREA_EXHAUSTED "write area exhausted"
  
   do {
   try {
   _send_info.message.rewind();
   _packer.pack(_send_info.header, _send_info.message);
   _packer.pack(msg, _send_info.message);
   done_packing = true;
   } catch(const std::ios_base::failure & e) {
   constexpr std::size_t len = sizeof(__WISP_WRITE_AREA_EXHAUSTED) - 1;
   if(std::strncmp(e.what(), __WISP_WRITE_AREA_EXHAUSTED, len) == 0) {
   _send_info.message.resize(2*_send_info.message.capacity());
   } else {
   throw;
   }
   }
   } while(!done_packing);
  
  #undef __WISP_WRITE_AREA_EXHAUSTED
  
   _send_info.message.set_type(msg.protocol);
   _send_info.message.set_service(service);
  
   if(_send_info.message.size() <= MaxUnfragmentedMessageSize) {
   _mbox.send(_send_info.message, dest);
   } else {
   const unsigned int message_size =
   _send_info.header.message_size = _send_info.message.size();
   _send_info.message.rewind();
  
   const unsigned int header_size =
   _packer.pack(_send_info.header, _send_info.message);
   const unsigned int max_fragment_size =
   MaxUnfragmentedMessageSize - header_size;
  
   // We never fragment a header, so there's no risk of
   // header_size equaling message_size..
   for(unsigned int offset = header_size, size = 0;
   offset < message_size; offset+=size)
   {
   size = std::min(max_fragment_size, message_size - offset);
   _mbox.clear_message_parts();
   _mbox.add_message_part(&_send_info.message[0], header_size);
   _mbox.add_message_part(&_send_info.message[offset], size);
   _mbox.send(dest, msg.protocol, service);
   }
   _mbox.clear_message_parts();
   }
   }
  
  public:
   // Using jumbo_message_map::key_type doesn't work because it's a composite
   // key that stores a reference, so we have to use a compatible key.
   typedef
   boost::tuple<string, wisp_call_role, wisp_call_token> jumbo_message_key_type;
  
   template<typename ReturnType>
   class Future {
   friend class Caller;
  
   bool _valid;
   wisp_call_token _token;
   Caller & _caller;
  
   Future(Caller & caller, const wisp_call_token token) :
   _valid(true), _token(token), _caller(caller)
   { }
  
   public:
   typedef ReturnType return_type;
   typedef boost::shared_ptr<Future> shared_ptr;
  
   // Note, you can't sit in a while(!future.ready()) loop because
   // there's no receiver thread.
   bool ready() const {
   return _caller.returned(_token);
   }
  
   bool valid() const {
   return _valid;
   }
  
   wisp_call_token token() const {
   return _token;
   }
  
   void receive(return_type & result)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure,
   CallException)
   {
   if(!valid())
   throw CallException("Invalid future.");
  
   _caller.receive_response(result, _token);
  
   _valid = false;
   }
  
   // Exclusively for supporting continuations in Service
   void unpack(return_type & result, MessageInfo & info)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure,
   CallException)
   {
   if(!valid())
   throw CallException("Invalid future.");
  
   if(info.token() == _token)
   _caller.unpack(result, info);
   else
   throwCallException(_token, info.token());
  
   _valid = false;
   }
   };
  
   explicit
   Caller(const string & connection = "",
   const string & name = "",
   const unsigned int message_capacity = Message::DefaultCapacity,
   const bool group_membership = GroupMembershipDisable) :
   _mbox(connection, name, group_membership), _call_token(0),
   _packer(), _unpacker(), _requests(), _responses(),
   _jumbo_messages(),
   _jumbo_key_from_value(_jumbo_messages.key_extractor()),
   _initial_message_capacity(message_capacity),
   _info(new MessageInfo(std::min(message_capacity, __WISP_CALLER_MAX_UNFRAGMENTED_MESSAGE_SIZE))),
   _send_info(message_capacity)
   { }
  
  #undef __WISP_CALLER_MAX_UNFRAGMENTED_MESSAGE_SIZE
  
   // Not intended to be subclassed outside of Wisp.
   //virtual ~Caller() = default;
  
   // We expose this for allowing retrieval of group information.
   const Mailbox & mbox() {
   return _mbox;
   }
  
   packer_type & packer() {
   return _packer;
   }
  
   unpacker_type & unpacker() {
   return _unpacker;
   }
  
   wisp_call_token call_token() {
   return _call_token;
   }
  
   const string & name() const {
   return _mbox.private_group();
   }
  
   bool group_membership() const {
   return _mbox.group_membership();
   }
  
   void join(const string & group) SSRC_DECL_THROW(NS_SSRC_SPREAD::Error) {
   _mbox.join(group);
   }
  
   void leave(const string & group) SSRC_DECL_THROW(NS_SSRC_SPREAD::Error) {
   _mbox.leave(group);
   }
  
   unsigned int message_capacity_initial() const {
   return _initial_message_capacity;
   }
  
   unsigned int message_capacity_receive() const {
   return _info->message.capacity();
   }
  
   unsigned int message_capacity_send() const {
   return _send_info.message.capacity();
   }
  
   unsigned int request_queue_size() const {
   return _requests.size();
   }
  
   unsigned int response_map_size() const {
   return _responses.size();
   }
  
   bool returned(const wisp_call_token token) const {
   return (_responses.find(token) != _responses.end());
   }
  
   unsigned int count_jumbo_messages() const {
   return _jumbo_messages.size();
   }
  
   template<typename key_container>
   void collect_jumbo_message_keys(key_container & container) {
   for(jumbo_message_map::const_iterator it = _jumbo_messages.begin(),
   end = _jumbo_messages.end(); it != end; ++it)
   {
   const MessageInfo *info = it->get();
   container.insert(container.end(), jumbo_message_key_type(info->sender(), info->role(), info->token()));
   }
   }
  
   template<typename iterator_type>
   void erase_jumbo_messages(const iterator_type & begin,
   const iterator_type & end)
   {
   // You can't erase a compatible key, so we have to perform a lookup first
   for(iterator_type it = begin; it != end; ++it) {
   jumbo_message_map::iterator msg = _jumbo_messages.find(*it);
   if(msg != _jumbo_messages.end())
   _jumbo_messages.erase(msg);
   }
   }
  
   // Call only after unpacking CallHeader so message offset is correct.
   template<typename MessageType>
   void unpack(MessageType & msg, MessageInfo & info)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure,
   CallException)
   {
   if(info.id() == MessageType::id &&
   info.protocol() == MessageType::protocol)
   _unpacker.unpack(msg, info.message);
   else
   throwCallException(MessageType::protocol, info.protocol(),
   MessageType::id, info.id());
   }
  
   void receive(message_info_ptr & info)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
   {
   // Drain responses first, then requests.
   if(!_responses.empty()) {
   response_map::iterator it = _responses.begin();
   info = *it;
   _responses.erase(it);
   } else if(!_requests.empty()) {
   info = _requests.front();
   _requests.pop_front();
   } else {
   do {
   Message & message = _info->message;
  
   _mbox.receive(message);
   _mbox.copy_groups(_info->groups);
  
   if(message.is_regular()) {
   CallHeader & header = _info->header;
  
   message.rewind();
  
   const unsigned int header_size = _unpacker.unpack(header, message);
  
   if(header.is_jumbo_fragment()) {
   const JumboFragmentResult jumbo_result =
   insert_jumbo_fragment(_info, header_size);
   if(jumbo_result == JumboFragmentCompleteResponse) {
   response_map::iterator it = _responses.begin();
   info = *it;
   _responses.erase(it);
   return;
   } else if(jumbo_result == JumboFragmentCompleteRequest) {
   info = _requests.front();
   _requests.pop_front();
   return;
   }
   } else
   break;
   } else {
   _info->clear_header();
   break;
   }
   } while(true);
   info = _info;
   }
   }
  
   // Also implement timeout.
   // Receives requests and membership messages, but not call responses
   void receive_request(message_info_ptr & info)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
   {
   if(!_requests.empty()) {
   info = _requests.front();
   _requests.pop_front();
   } else {
   if(!_info.unique()) {
   reset_receive_info();
   }
   do {
   Message & message = _info->message;
  
   _mbox.receive(message);
   _mbox.copy_groups(_info->groups);
  
   if(message.is_regular()) {
   CallHeader & header = _info->header;
  
   message.rewind();
  
   const unsigned int header_size = _unpacker.unpack(header, message);
  
   if(header.is_jumbo_fragment()) {
   const JumboFragmentResult jumbo_result =
   insert_jumbo_fragment(_info, header_size);
   if(jumbo_result == JumboFragmentCompleteRequest) {
   info = _requests.front();
   _requests.pop_front();
   return;
   }
   } else {
   if(header.role == TwoWayResponse)
   insert_response();
   else
   break;
   }
   } else {
   _info->clear_header();
   break;
   }
   } while(true);
   info = _info;
   }
   }
  
   template<typename Traits, typename DestinationType>
   void send(const DestinationType & dest,
   const typename Traits::parameter_type & param,
   const Message::Service service = DefaultMessageServiceType)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
   {
   static_assert(Traits::call_type == OneWay, "expected call_type OneWay");
   send(dest, param, next_token(), OneWayRequest, service);
   }
  
   template<typename Traits, typename DestinationType>
   void reply(const DestinationType & dest,
   const wisp_call_token token,
   const typename Traits::parameter_type & param,
   const Message::Service service = DefaultMessageServiceType)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
   {
   static_assert(Traits::call_type == OneWay, "expected call_type OneWay");
   send(dest, param, token, TwoWayResponse, service);
   }
  
   // Synchronous call.
   template<typename Traits, typename DestinationType>
   void call(const DestinationType & dest,
   typename Traits::return_type *ret,
   const typename Traits::parameter_type & param,
   const Message::Service service = DefaultMessageServiceType)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure,
   CallException)
   {
   static_assert(Traits::call_type == TwoWay, "expected call_type TwoWay");
   wisp_call_token token = next_token();
   send(dest, param, token, TwoWayRequest, service);
   receive_response(*ret, token);
   }
  
   // Asynchronous call.
   template<typename Traits, typename DestinationType>
   typename Future<typename Traits::return_type>::shared_ptr
   call(const DestinationType & dest,
   const typename Traits::parameter_type & param,
   const Message::Service service = DefaultMessageServiceType)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
   {
   static_assert(Traits::call_type == TwoWay, "expected call_type TwoWay");
   typedef Future<typename Traits::return_type> future_type;
   wisp_call_token token = next_token();
  
   send(dest, param, token, TwoWayRequest, service);
  
   return typename future_type::shared_ptr(new future_type(*this, token));
   }
  
  private:
   template<CallType call_type> struct CallTag { };
  
   template<typename Traits, typename DestinationType>
   typename Future<typename Traits::return_type>::shared_ptr
   _call(const CallTag<OneWay> &,
   const DestinationType & dest,
   const typename Traits::parameter_type & param,
   const Message::Service service = DefaultMessageServiceType)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
   {
   typedef Future<typename Traits::return_type> future_type;
   send<Traits>(dest, param, service);
   return typename future_type::shared_ptr();
   }
  
   template<typename Traits, typename DestinationType>
   typename Future<typename Traits::return_type>::shared_ptr
   _call(const CallTag<TwoWay> &,
   const DestinationType & dest,
   const typename Traits::parameter_type & param,
   const Message::Service service = DefaultMessageServiceType)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
   {
   return call<Traits>(dest, param, service);
   }
  
  public:
  
   template<typename Traits, typename DestinationType>
   void operator()(const Traits &,
   const DestinationType & dest,
   const wisp_call_token token,
   const typename Traits::parameter_type & param,
   const Message::Service service =
   DefaultMessageServiceType)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
   {
   reply<Traits>(dest, token, param, service);
   }
  
   template<typename Traits, typename DestinationType>
   void operator()(const Traits &,
   const DestinationType & dest,
   typename Traits::return_type *ret,
   const typename Traits::parameter_type & param,
   const Message::Service service =
   DefaultMessageServiceType)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure,
   CallException)
   {
   call<Traits>(dest, ret, param, service);
   }
  
   template<typename Traits, typename DestinationType>
   typename Future<typename Traits::return_type>::shared_ptr
   operator()(const Traits &,
   const DestinationType & dest,
   const typename Traits::parameter_type & param,
   const Message::Service service = DefaultMessageServiceType)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
   {
   return _call<Traits>(CallTag<Traits::call_type>(), dest, param, service);
   }
  
  
   // Convenience functions that invoke parameter_type constructor for
   // you. We permit up to WISP_CALLER_MAX_PARAMETERS - 1
   // parameter_type arguments. Any more, and you should probably use
   // a different approach.
  
   // Send.
  
   template<typename Traits, typename DestinationType>
   void send(const DestinationType & dest,
   const Message::Service service = DefaultMessageServiceType)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
   {
   send<Traits>(dest, typename Traits::parameter_type(), service);
   }
  
   template<typename Traits, typename DestinationType, typename... P>
   void send(const Message::Service service,
   const DestinationType & dest,
   P && ...p)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
   {
   send<Traits>(dest, typename Traits::parameter_type(std::forward<P>(p)...),
   service);
   }
  
   template<typename Traits, typename DestinationType, typename... P>
   void sendp(const DestinationType & dest, P && ...p)
  
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
   {
   send<Traits>(dest, typename Traits::parameter_type(std::forward<P>(p)...));
   }
  
   // Reply.
  
   template<typename Traits, typename DestinationType>
   void reply(const DestinationType & dest,
   const wisp_call_token token,
   const Message::Service service = DefaultMessageServiceType)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
   {
   reply<Traits>(dest, token, typename Traits::parameter_type(), service);
   }
  
   template<typename Traits, typename DestinationType>
   void operator()(const Traits &,
   const DestinationType & dest,
   const wisp_call_token token,
   const Message::Service service =
   DefaultMessageServiceType)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
   {
   reply<Traits>(dest, token, typename Traits::parameter_type(), service);
   }
  
   template<typename Traits, typename DestinationType, typename... P>
   void reply(const Message::Service service,
   const DestinationType & dest,
   const wisp_call_token token,
   P && ...p)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
   {
   reply<Traits>(dest, token,
   typename Traits::parameter_type(std::forward<P>(p)...),
   service);
   }
  
   template<typename Traits, typename DestinationType, typename... P>
   void replyp(const DestinationType & dest,
   const wisp_call_token token,
   P && ...p)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
   {
   reply<Traits>(dest, token,
   typename Traits::parameter_type(std::forward<P>(p)...));
   }
  
   // Synchronous call.
   template<typename Traits, typename DestinationType>
   void call(const DestinationType & dest,
   typename Traits::return_type *ret,
   const Message::Service service = DefaultMessageServiceType)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure,
   CallException)
   {
   call<Traits>(dest, ret, typename Traits::parameter_type(), service);
   }
  
   template<typename Traits, typename DestinationType>
   void operator()(const Traits &,
   const DestinationType & dest,
   typename Traits::return_type *ret,
   const Message::Service service = DefaultMessageServiceType)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure,
   CallException)
   {
   call<Traits>(dest, ret, typename Traits::parameter_type(), service);
   }
  
   template<typename Traits, typename DestinationType, typename... P>
   void call(const Message::Service service,
   const DestinationType & dest,
   typename Traits::return_type *ret,
   P && ...p)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
   {
   call<Traits>(dest, ret,
   typename Traits::parameter_type(std::forward<P>(p)...),
   service);
   }
  
   template<typename Traits, typename DestinationType, typename... P>
   void callp(const DestinationType & dest,
   typename Traits::return_type *ret,
   P && ...p)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
   {
   call<Traits>(dest, ret,
   typename Traits::parameter_type(std::forward<P>(p)...));
   }
  
   // Asynchronous call.
   template<typename Traits, typename DestinationType>
   typename Future<typename Traits::return_type>::shared_ptr
   call(const DestinationType & dest,
   const Message::Service service = DefaultMessageServiceType)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
   {
   return call<Traits>(dest, typename Traits::parameter_type(), service);
   }
  
   template<typename Traits, typename DestinationType, typename... P>
   typename Future<typename Traits::return_type>::shared_ptr
   call(const Message::Service service, const DestinationType & dest,
   P && ...p)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
   {
   return call<Traits>(dest,
   typename Traits::parameter_type(std::forward<P>(p)...),
   service);
   }
  
   template<typename Traits, typename DestinationType, typename... P>
   typename Future<typename Traits::return_type>::shared_ptr
   callp(const DestinationType & dest, P && ...p)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
   {
   return call<Traits>(dest,
   typename Traits::parameter_type(std::forward<P>(p)...));
   }
  
   // Dual-purpose operator
  
   template<typename Traits, typename DestinationType>
   typename Future<typename Traits::return_type>::shared_ptr
   operator()(const Traits &,
   const DestinationType & dest,
   const Message::Service service = DefaultMessageServiceType)
   SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
   {
   return
   _call<Traits>(CallTag<Traits::call_type>(),
   dest, typename Traits::parameter_type(), service);
   }
  };
  
  __END_NS_SSRC_WISP_PROTOCOL
  
  #endif