ContinuationCaller.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_CONTINUATION_CALLER_H
#define __SSRC_WISP_PROTOCOL_CONTINUATION_CALLER_H

#include <ssrc/wisp/protocol/Caller.h>

#include <boost/function.hpp>
#include <boost/multi_index/member.hpp>
#include <boost/multi_index/ordered_index.hpp>

__BEGIN_NS_SSRC_WISP_PROTOCOL

struct Continuation {
  // We save token as an optimization to avoid invoking a virtual function
  // in the hash index.
  wisp_call_token token;

protected:
  explicit Continuation(wisp_call_token token) : token(token) { }

public:
  virtual ~Continuation() { }

  virtual void resume(MessageInfo & msginfo) = 0;
};


typedef boost::shared_ptr<Continuation> continuation_ptr;
const continuation_ptr NullContinuation = continuation_ptr();

template<typename CallTraits_, typename Closure>
class FutureContinuation : public Continuation {
public:
  typedef CallTraits_ call_traits;
  typedef typename call_traits::future_ptr future_ptr;
  typedef typename call_traits::return_type return_type;
  // Closure should be void (const return_type &)
  typedef Closure continuation_function;

private:
  future_ptr _future;
  return_type _result;
  continuation_function _continue;

public:

  FutureContinuation(const future_ptr & future,
                     continuation_function && continue_) :
    Continuation(future->token()),
    _future(future), _result(),
    _continue(std::forward<continuation_function>(continue_))
  { }

  virtual ~FutureContinuation() { }

  virtual void resume(MessageInfo & msginfo) {
    _future->unpack(_result, msginfo);
    _continue(_result);
  }
};

// Specialization for use with boost::bind
template<typename CallTraits_>
class FutureContinuation<CallTraits_,
                         boost::function<void (const typename CallTraits_::return_type &)> >:
  public Continuation
{
public:
  typedef CallTraits_ call_traits;
  typedef typename call_traits::future_ptr future_ptr;
  typedef typename call_traits::return_type return_type;
  typedef boost::function<void (const return_type &)> continuation_function;

private:
  future_ptr _future;
  return_type _result;
  continuation_function _continue;

public:

  FutureContinuation(const future_ptr & future,
                     continuation_function && continue_) :
    Continuation(future->token()),
    _future(future), _result(),
    _continue(std::forward<continuation_function>(continue_))
  { }

  virtual ~FutureContinuation() { }

  virtual void resume(MessageInfo & msginfo) {
    _future->unpack(_result, msginfo);
    _continue(_result);
  }
};

template<typename PT = BinaryPackingTraits>
class ContinuationCaller : public protocol::Caller<PT> {
  typedef protocol::Caller<PT> super;

  enum { ByHash, ByOrder };

  typedef boost::multi_index_container<
    continuation_ptr,
    boost::multi_index::indexed_by<
      boost::multi_index::hashed_unique<
        boost::multi_index::member<Continuation, wisp_call_token,
                                   &Continuation::token> >,
      boost::multi_index::ordered_non_unique<
        boost::multi_index::member<Continuation, wisp_call_token,
                                   &Continuation::token> >
      > > continuation_map;

  typedef
  typename continuation_map::template nth_index<ByHash>::type index_by_hash;
  typedef
  typename continuation_map::template nth_index<ByOrder>::type index_by_order;

  continuation_map _continuations;

public:

  explicit
  ContinuationCaller(const string & connection = "",
                     const string & name = "", 
                     const unsigned int message_capacity = Message::DefaultCapacity,
                     const bool group_membership = GroupMembershipDisable) :
    super(connection, name, message_capacity, group_membership),
    _continuations()
  { }

  unsigned int continuations_map_size() const {
    return _continuations.size();
  }

  template<typename CallTraits_, typename Closure, typename DestinationType>
  continuation_ptr
  split_call(Closure && continue_,
             const DestinationType & dest,
             const typename CallTraits_::parameter_type & param,
             const Message::Service service = DefaultMessageServiceType)
    SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
  {
    return continuation_ptr(new FutureContinuation<CallTraits_, Closure>(super::template call<CallTraits_>(dest, param, service), std::forward<Closure>(continue_)));
  }

  void schedule(const continuation_ptr & continuation) {
    _continuations.insert(continuation);
  }

  template<typename CallTraits_, typename Closure, typename DestinationType>
  void future_call(Closure && continue_,
                   const DestinationType & dest,
                   const typename CallTraits_::parameter_type & param,
                   const Message::Service service = DefaultMessageServiceType)
    SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
  {
    schedule(split_call<CallTraits_>(std::forward<Closure>(continue_),
                                     dest, param, service));
  }

  // Cancels all pending requests with min_token <= token < max_token
  void cancel_range(wisp_call_token min_token, wisp_call_token max_token) {
    index_by_order & index = _continuations.template get<ByOrder>();

  start:
    if(min_token < max_token)
      index.erase(index.lower_bound(min_token), index.upper_bound(max_token));
    else if(min_token > max_token) {
      index.erase(index.lower_bound(min_token), index.end());
      min_token = 0;
      goto start;
    }
  }

  void cancel_all() {
    _continuations.clear();
  }

  continuation_ptr cancel(wisp_call_token token) {
    continuation_map::iterator it = _continuations.find(token);

    if(it != _continuations.end()) {
      continuation_ptr result = *it;
      _continuations.erase(it);
      return result;
    }

    return NullContinuation;
  }

  bool resume(MessageInfo & msginfo) {
    continuation_ptr result = cancel(msginfo.token());

    if(result != NullContinuation) {
      result->resume(msginfo);
      return true;
    }

    return false;
  }

  template<typename CallTraits_, typename Closure, typename DestinationType>
  void future_call(Closure && continue_,
                   const DestinationType & dest,
                   const Message::Service service = DefaultMessageServiceType)
    SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
  {
    future_call<CallTraits_>(std::forward<Closure>(continue_),
                             dest, typename CallTraits_::parameter_type(),
                             service);
  }

  template<typename CallTraits_, typename Closure, typename DestinationType,
           typename... P>
  void future_call(const Message::Service service,
                   Closure && continue_,
                   const DestinationType & dest, P && ...p)
    SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)    
  {
    future_call<CallTraits_>(std::forward<Closure>(continue_), dest,
                             typename CallTraits_::parameter_type(std::forward<P>(p)...),
                             service);
  }

  template<typename CallTraits_, typename Closure, typename DestinationType,
           typename... P>
  void future_callp(Closure && continue_,
                    const DestinationType & dest, P && ...p)
    SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)    
  {
    future_call<CallTraits_>(std::forward<Closure>(continue_), dest,
                             typename CallTraits_::parameter_type(std::forward<P>(p)...));
  }

};

__END_NS_SSRC_WISP_PROTOCOL

#endif