service/service.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_SERVICE_SERVICE_H
  #define __SSRC_WISP_SERVICE_SERVICE_H
  
  #include <ssrc/wisp/protocol/ContinuationCaller.h>
  #include <ssrc/wisp/service/EventLoop.h>
  
  #include <functional>
  
  #ifdef WISP_DEBUG
  #include <iostream>
  #endif
  
  __BEGIN_NS_SSRC_WISP_SERVICE
  
  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;
  using NS_SSRC_WISP_UTILITY::make_smart_ptr;
  
  #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 std::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 std::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() = default;
  
   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 std::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:
  
   explicit 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(make_smart_ptr<timeout_handler_ptr>(*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) {
   constexpr void (ServiceProtocolProcessor::*handler)(Impl &, MessageInfo &) =
   &ServiceProtocolProcessor::template request<MessageType, Impl>;
   return set_request_handler(message_handler_entry(MessageType::protocol,
   MessageType::id,
   std::bind(handler, this, std::ref(impl), std::placeholders::_1)));
   }
  
   template<typename MessageType, typename Impl>
   bool set_request_handler(Impl & impl, MessageType & buffer) {
   constexpr void (ServiceProtocolProcessor::*handler)
   (Impl &, MessageType &, MessageInfo &) =
   &ServiceProtocolProcessor::template request<MessageType, Impl>;
   return set_request_handler(message_handler_entry(MessageType::protocol,
   MessageType::id,
   std::bind(handler, this, std::ref(impl), std::ref(buffer), std::placeholders::_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) {
   constexpr void (ServiceProtocolProcessor::*handler)(Impl &, MessageInfo &) =
   &ServiceProtocolProcessor::template respond<MessageType, Impl>;
   return set_response_handler(message_handler_entry(MessageType::protocol,
   MessageType::id,
   std::bind(handler, this, std::ref(impl), std::placeholders::_1))); }
  
   template<typename MessageType, typename Impl>
   bool set_response_handler(Impl & impl, MessageType & buffer) {
   constexpr void (ServiceProtocolProcessor::*handler)
   (Impl &, MessageType &, MessageInfo &) =
   &ServiceProtocolProcessor::template respond<MessageType, Impl>;
   return set_response_handler(message_handler_entry(MessageType::protocol,
   MessageType::id,
   std::bind(handler, this, std::ref(impl), std::ref(buffer), std::placeholders::_1)));
   }
  
   virtual void transition(State state) {
   _state = state;
   }
  
  public:
  
   explicit ServiceProtocolProcessor(caller_type & caller) :
   _state(Stopped),
   _request_handlers(), _response_handlers(), _caller(caller)
   { }
  
   virtual ~ServiceProtocolProcessor() = default;
  
   std::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(std::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(std::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(make_smart_ptr<message_info_ptr>(caller.message_capacity_receive())),
   _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() = default;
  
   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 std::string & connection = "",
   const std::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 std::string & connection = "",
   const std::string & name = "",
   const unsigned int message_capacity =
   Message::DefaultCapacity) :
   _caller(connection, name, message_capacity,
   protocol_processor::GroupMembership),
   _handler(_caller, initializer)
   { }
  
   const std::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