Caller.h

Go to the documentation of this file.

/*
 * Copyright 2006-2008 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)
  {
    unsigned int message_size;

    _send_info.init_header(msg.id, token, role);
    _send_info.message.rewind();
    _packer.pack(_send_info.header, _send_info.message);
    _packer.pack(msg, _send_info.message);
    _send_info.message.set_type(msg.protocol);
    _send_info.message.set_service(service);

    if((message_size = _send_info.message.size())
       <= MaxUnfragmentedMessageSize)
      _mbox.send(_send_info.message, dest);
    else {
      _send_info.header.message_size = 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() { }

  // 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 initial_message_capacity() const {
    return _initial_message_capacity;
  }

  unsigned int message_capacity() const {
    return _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