service/service.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_SERVICE_SERVICE_H
#define __SSRC_WISP_SERVICE_SERVICE_H

#include <ssrc/wisp/protocol/ContinuationCaller.h>
#include <ssrc/wisp/service/EventLoop.h>

#include <boost/bind.hpp>

#ifdef WISP_DEBUG
#include <iostream>
#endif

__BEGIN_NS_SSRC_WISP_SERVICE

using std::string;
using NS_SSRC_SPREAD::MembershipInfo;
using NS_SSRC_SPREAD::Message;
using NS_SSRC_SPREAD::GroupList;
using NS_SSRC_WISP_PROTOCOL::MessageInfo;
using NS_SSRC_WISP_PROTOCOL::message_info_ptr;
using NS_SSRC_WISP_PROTOCOL::CallException;
using NS_SSRC_WISP_PROTOCOL::GroupMembershipDisable;
using NS_SSRC_WISP_PROTOCOL::wisp_message_protocol;
using NS_SSRC_WISP_PROTOCOL::wisp_message_id;

#define WISP_SERVICE_REQUEST(MessageType) \
  set_request_handler<MessageType>(*this)
#define WISP_SERVICE_RESPONSE(MessageType) \
  set_response_handler<MessageType>(*this)
#define WISP_SERVICE_REQUEST_T(MessageType) \
  this->template set_request_handler<MessageType>(*this)
#define WISP_SERVICE_RESPONSE_T(MessageType) \
  this->template set_response_handler<MessageType>(*this)

#define WISP_SERVICE_REQUEST_BUFFERED(MessageType, msg) \
  set_request_handler<MessageType>(*this, msg)
#define WISP_SERVICE_RESPONSE_BUFFERED(MessageType, msg) \
  set_response_handler<MessageType>(*this, msg)
#define WISP_SERVICE_REQUEST_BUFFERED_T(MessageType, msg) \
  this->template set_request_handler<MessageType>(*this, msg)
#define WISP_SERVICE_RESPONSE_BUFFERED_T(MessageType, msg) \
  this->template set_response_handler<MessageType>(*this, msg)


typedef boost::function<void (const EventInfo &)> timeout_handler;
// We expose message_handler_type, message_handler_entry, and
// message_handler_map specifically to support dynamically loaded handlers.
typedef boost::function<void (MessageInfo &)> message_handler_type;

struct message_handler_entry {
  wisp_message_protocol protocol;
  wisp_message_id id;
  message_handler_type handle_message;

  message_handler_entry(const wisp_message_protocol protocol,
                        const wisp_message_id id,
                        const message_handler_type & message_handler) :
    protocol(protocol), id(id), handle_message(message_handler)
  { }
};

typedef boost::multi_index_container<
  message_handler_entry,
  boost::multi_index::indexed_by<
    boost::multi_index::hashed_unique<
      boost::multi_index::composite_key<
        message_handler_entry,
        boost::multi_index::member<message_handler_entry,
                                   wisp_message_protocol,
                                   &message_handler_entry::protocol>,
        boost::multi_index::member<message_handler_entry, wisp_message_id,
                                   &message_handler_entry::id> >
      >
    > > message_handler_map;

/* BEGIN EXPERIMENTAL CODE */
// TODO: This is all one big kluge and should be re-implemented, but it works..
class ServiceContext {
public:
  class TimeoutHandler : public EventHandler {
    ServiceContext & _context;
    bool _once;
    timeout_handler _handler;

  public:

    TimeoutHandler(ServiceContext & context,
                   const timeout_handler & handler, bool once) :
      _context(context), _once(once), _handler(handler)
    { }

    virtual ~TimeoutHandler() { }

    virtual void handle_timeout(const EventInfo & info) {
      _handler(info);
      // TODO: WARNING: This could blow up because the current object
      // may be deleted on the call to remove_timeout.  This works only
      // because the removal happens at the end of the function and we
      // don't access 'this' again.  Seek an alternative solution.
      if(_once)
        _context.remove_timeout(this);
    }

    void execute() {
      handle_timeout(EventInfo(_context.event_loop(), EventLoop::None, false));
    }

    bool once() {
      return _once;
    }

    const TimeoutHandler *address() const {
      return this;
    }
  };

  typedef boost::shared_ptr<TimeoutHandler> timeout_handler_ptr;

  typedef boost::multi_index_container<
    timeout_handler_ptr,
    boost::multi_index::indexed_by<
      boost::multi_index::hashed_unique<
        boost::multi_index::const_mem_fun<TimeoutHandler,
                                          const TimeoutHandler *,
                                          &TimeoutHandler::address> >
  > > timeout_map;

private:
  EventLoop *loop;
  timeout_map _timeouts;

public:

  ServiceContext(EventLoop * loop = 0) : loop(loop) { }

  timeout_map::size_type count_timeouts() const {
    return _timeouts.size();
  }

  void clear_timeouts() {
    for(timeout_map::iterator it = _timeouts.begin(), end = _timeouts.end();
        it != end; ++it)
      remove_timeout(it->get());
  }

  timeout_handler_ptr add_timeout(const timeout_handler & handler_fun,
                                  const TimeValue & timeout,
                                  const bool once)
  {
    timeout_handler_ptr handler(new TimeoutHandler(*this, handler_fun, once));
    _timeouts.insert(handler);
    loop->add_handler(*handler, EventLoop::None, timeout);
    return handler;
  }

  void remove_timeout(TimeoutHandler *timeout) {
    // must remove from loop first or we lose pointer!!!
    loop->remove_handler(*timeout);
    _timeouts.erase(timeout);
  }

  EventLoop & event_loop() { return *loop; }
};


typedef ServiceContext::timeout_handler_ptr timeout_ptr; 

/* END EXPERIMENTAL CODE */

template<typename PackingTraits = BinaryPackingTraits>
class ServiceProtocolProcessor {
public:
  typedef PackingTraits packing_traits;
  typedef typename protocol::ContinuationCaller<packing_traits> caller_type;

  enum State { Starting, Started, Stopping, Stopped };

  static const bool GroupMembership = GroupMembershipDisable;

private:
  State _state;

private:

  message_handler_map _request_handlers;
  message_handler_map _response_handlers;

protected:
  caller_type & _caller;

  // WARNING!! This is highly experimental and only made available so
  // a subclass may add handlers to the event loop for I/O on other
  // descriptors.  The context is only valid in the Starting and
  // Started states.
  ServiceContext & context() { return _context; };

  virtual void process_membership_message(const MessageInfo & msginfo,
                                          const MembershipInfo & meminfo)
  { }

  // So you can reuse message types that are expensive to create on each call.
  template<typename MessageType, typename Impl>
  void request(Impl & impl, MessageType & msg, MessageInfo & msginfo)
    SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure,
                    CallException)
  {
    _caller.unpack(msg, msginfo);
    // Cast forces msginfo to be const arg in process_request
    impl.process_request(msg, static_cast<const MessageInfo &>(msginfo));
  }

  template<typename MessageType, typename Impl>
  void request(Impl & impl, MessageInfo & msginfo)
    SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
  {
    MessageType msg;
    request(impl, msg, msginfo);
  }

  template<typename MessageType, typename Impl>
  void respond(Impl & impl, MessageType & msg, MessageInfo & msginfo)
    SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure,
                    CallException)
  {
    _caller.unpack(msg, msginfo);
    // Cast forces msginfo to be const arg in process_response
    impl.process_response(msg, static_cast<const MessageInfo &>(msginfo));
  }

  template<typename MessageType, typename Impl>
  void respond(Impl & impl, MessageInfo & msginfo)
    SSRC_DECL_THROW(boost::archive::archive_exception, std::ios_base::failure)
  {
    MessageType msg;
    respond(impl, msg, msginfo);
  }

  bool set_request_handler(const message_handler_entry & handler) {
    return _request_handlers.insert(handler).second;
  }

  message_handler_map::size_type
  remove_request_handler(const message_handler_entry & handler) {
    return
      _request_handlers.erase(_request_handlers.key_extractor()((handler)));
  }

  void clear_request_handlers() {
    _request_handlers.clear();
  }

  template<typename MessageType, typename Impl>
  bool set_request_handler(Impl & impl) {
    return set_request_handler(message_handler_entry(MessageType::protocol,
                                                     MessageType::id,
                                                     boost::bind(&ServiceProtocolProcessor::template request<MessageType, Impl>, this, std::ref(impl), _1)));
  }

  template<typename MessageType, typename Impl>
  bool set_request_handler(Impl & impl, MessageType & buffer)  {
    return set_request_handler(message_handler_entry(MessageType::protocol,
                                                     MessageType::id,
                                                     boost::bind(&ServiceProtocolProcessor::template request<MessageType, Impl>, this, std::ref(impl), std::ref(buffer), _1)));
  }

  bool set_response_handler(const message_handler_entry & handler) {
    return _response_handlers.insert(handler).second;
  }

  message_handler_map::size_type
  remove_response_handler(const message_handler_entry & handler) {
    return
      _response_handlers.erase(_response_handlers.key_extractor()((handler)));
  }

  void clear_response_handlers() {
    _response_handlers.clear();
  }

  template<typename MessageType, typename Impl>
  bool set_response_handler(Impl & impl) {
    return set_response_handler(message_handler_entry(MessageType::protocol,
                                                      MessageType::id,
                                                      boost::bind(&ServiceProtocolProcessor::template respond<MessageType, Impl>, this, std::ref(impl), _1)));  }

  template<typename MessageType, typename Impl>
  bool set_response_handler(Impl & impl, MessageType & buffer) {
    return set_response_handler(message_handler_entry(MessageType::protocol,
                                                      MessageType::id,
                                                      boost::bind(&ServiceProtocolProcessor::template respond<MessageType, Impl>, this, std::ref(impl), std::ref(buffer), _1)));
  }

  virtual void transition(State state) {
    _state = state;
  }

public:

  explicit ServiceProtocolProcessor(caller_type & caller) :
    _state(Stopped),
    _request_handlers(), _response_handlers(), _caller(caller)
  { }

  virtual ~ServiceProtocolProcessor() { }

  string name() const {
    return _caller.name();
  }

  State state() const {
    return _state;
  }

  void start() {
    transition(Starting);
  }

  // Services listening for membership messages should wait for self-leave
  // before transitioning to Stopped.
  void stop() {
    transition(Stopping);
  }

  void membership(const MessageInfo & msginfo,
                  const MembershipInfo & meminfo)
  {
    process_membership_message(msginfo, meminfo);
  }

  void request(MessageInfo & msginfo) {
    message_handler_map::iterator it = 
      _request_handlers.find(boost::make_tuple(msginfo.protocol(),
                                               msginfo.id()));
    if(it != _request_handlers.end())
      it->handle_message(msginfo);
    // TODO: add an else log unhandled message
  }

  void response(MessageInfo & msginfo) {
    if(!_caller.resume(msginfo)) {
      message_handler_map::iterator it = 
        _response_handlers.find(boost::make_tuple(msginfo.protocol(),
                                                  msginfo.id()));
      if(it != _response_handlers.end())
        it->handle_message(msginfo);
      // TODO: add an else log unhandled message
    }
  }

  /* BEGIN EXPERIMENTAL CODE */
  // TODO: This is all one big kluge and should be re-implemented (which
  // is why this is stuffed at the end of the class), but it works.

  timeout_ptr schedule_timeout(const timeout_handler & handler,
                               const TimeValue & timeout,
                               const bool once = EventLoop::Persist)
  {
    return _context.add_timeout(handler, timeout, once);
  }

  void cancel_timeout(const timeout_ptr & timeout) {
    _context.remove_timeout(timeout.get());
  }

  void clear_timeouts() {
    _context.clear_timeouts();
  }

  ServiceContext::timeout_map::size_type count_timeouts() const {
    return _context.count_timeouts();
  }

private:
  template<typename PP> friend class ServiceEventHandler;

  ServiceContext _context;

  void set_service_context(const ServiceContext & context) {
    _context = context;
  }

  /* END EXPERIMENTAL CODE */
};

template<typename PP>
class ServiceEventHandler : public EventHandler {
public:
  typedef PP protocol_processor;
  WISP_IMPORT_T(protocol_processor, caller_type);

private:
  caller_type & _caller;
  protocol_processor _protocol;
  // Scratch variables for handle_event().
  message_info_ptr _message_info;
  MembershipInfo _membership_info;
  protocol::wisp_call_token _min_token, _max_token;
  std::vector<typename caller_type::jumbo_message_key_type> _jumbo_message_keys;

protected:

  virtual void remove_handler(EventLoop & loop) {
    /* BEGIN EXPERIMENTAL CODE */
    // TODO: This is a kluge and should be re-implemented, but it works.
    _protocol.clear_timeouts();
    /* END EXPERIMENTAL CODE */
    loop.remove_handler(*this);
    _caller.cancel_all();
  }

public:

  explicit ServiceEventHandler(caller_type & caller) :
    _caller(caller), _protocol(caller),
    _message_info(new MessageInfo(caller.message_capacity())),
    _membership_info(),
    _min_token(0), _max_token(0),
    _jumbo_message_keys(4)
  { }

  template<typename Initializer>
  explicit ServiceEventHandler(caller_type & caller,
                               const Initializer & initializer) :
    _caller(caller), _protocol(caller, initializer),
    _message_info(),
    _membership_info(),
    _min_token(0), _max_token(0),
    _jumbo_message_keys(4)
  { }

  virtual ~ServiceEventHandler() { }

  typename protocol_processor::State state() const {
    return _protocol.state();
  }

  virtual int event_descriptor() const {
    return _caller.mbox().descriptor();
  }

  virtual void handle_read(const EventInfo & info) {
    try {
      if(state() != protocol_processor::Stopped) {
        _caller.receive(_message_info);

        if(_message_info->message.is_regular() &&
           _protocol.state() < protocol_processor::Stopping)
        {
          if(_message_info->role() == protocol::TwoWayResponse)
            _protocol.response(*_message_info);
          else
            _protocol.request(*_message_info);
        } else if(_message_info->message.is_membership()) {
          _message_info->message.get_membership_info(_membership_info);
          _protocol.membership(*_message_info, _membership_info);
        }

        if(state() == protocol_processor::Stopped)
          remove_handler(info.event_loop());
      }
    } catch(const boost::archive::archive_exception & ae) {
      // TODO: log
#ifdef WISP_DEBUG
      std::cerr << _protocol.name()
                << ": Caught boost::archive::archive_exception: " 
                << ae.what() << "\nContinuing.";
#endif
    } catch(const std::ios_base::failure & iof) {
      // TODO: log. probably don't want to try again
#ifdef WISP_DEBUG
      std::cerr << _protocol.name() << ": Caught std::ios_base::failure: "
                << iof.what() << "\nContinuing.";
#endif
    }
  }

  virtual void handle_timeout(const EventInfo & info) {
    // Cancel continuations
    _caller.cancel_range(_min_token, _max_token);
    _min_token = _max_token;
    _max_token = _caller.call_token();

    // Clear out incomplete jumbo messages
    if(!_jumbo_message_keys.empty()) {
      if(_caller.count_jumbo_messages() > 0) {
        _caller.erase_jumbo_messages(_jumbo_message_keys.begin(),
                                     _jumbo_message_keys.end());
      }
      _jumbo_message_keys.clear();
    }

    if(_caller.count_jumbo_messages() > 0) {
      _caller.collect_jumbo_message_keys(_jumbo_message_keys);
    }
  }

  protocol_processor & protocol() {
    return _protocol;
  }

  void start(EventLoop & loop, const unsigned int call_timeout) {
    _min_token = _max_token = _caller.call_token();
    loop.add_handler(*this, EventLoop::Read, TimeValue(call_timeout, 0));
    /* BEGIN EXPERIMENTAL CODE */
    // TODO: This is a kluge and should be re-implemented, but it works.
    _protocol.set_service_context(ServiceContext(&loop));
    /* END EXPERIMENTAL CODE */
    _protocol.start();
  }

  void stop() {
    _protocol.stop();
  }
};


template<typename EH>
class Service {
public:
  typedef EH event_handler;
  WISP_IMPORT_T(event_handler, caller_type);
  WISP_IMPORT_T(event_handler, protocol_processor);

private:
  caller_type _caller;
  event_handler _handler;

protected:

  protocol_processor & protocol() {
    return _handler.protocol();
  }

public:

  explicit Service(const string & connection = "",
                   const string & name = "", 
                   const unsigned int message_capacity =
                   Message::DefaultCapacity) :
    _caller(connection, name, message_capacity,
            protocol_processor::GroupMembership),
    _handler(_caller)
  { }

  template<typename Initializer>
  explicit Service(const Initializer & initializer,
                   const string & connection = "",
                   const string & name = "", 
                   const unsigned int message_capacity =
                   Message::DefaultCapacity) :
    _caller(connection, name, message_capacity,
            protocol_processor::GroupMembership),
    _handler(_caller, initializer)
  { }

  const string & name() const {
    return _caller.name();
  }

  typename protocol_processor::State state() {
    return protocol().state();
  }

  void start(EventLoop & loop, unsigned int call_timeout) {
    _handler.start(loop, call_timeout);
  }

  void stop() {
    _handler.stop();
  }
};

__END_NS_SSRC_WISP_SERVICE

#endif