EventQueue.h

Go to the documentation of this file.

/*
 * Copyright 2006-2009 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
 *
 *     https://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_WSPR_WS_EVENT_QUEUE_H
#define __SSRC_WSPR_WS_EVENT_QUEUE_H

#include <boost/ptr_container/ptr_unordered_map.hpp>

#include <ssrc/wispers/ws/service.h>
#include <ssrc/wispers/session/protocol.h>
#include <ssrc/wispers/group_session/protocol.h>

#ifdef WSPR_DEBUG
#include <ssrc/wispers/utility/PropertiesToString.h>
#endif

__BEGIN_NS_SSRC_WSPR_WS

using NS_SSRC_WSPR_UTILITY::property_vector;
using NS_SSRC_WSPR_UTILITY::StringTo;
using NS_SSRC_WSPR_SESSION::SessionProtocol;
using NS_SSRC_WSPR_GROUP_SESSION::GroupSessionProtocol;
using NS_SSRC_WSPR_GROUP_SESSION::uid_container;
using NS_SSRC_WSPR_SESSION::SessionProtocol;

struct EventQueueInitializer {
  // In seconds.
  unsigned int fetch_timeout;

  EventQueueInitializer() :
    fetch_timeout(20)
  { }
};

class EventQueue : public WebService {
  typedef WebService super;
  friend class NS_SSRC_WISP_SERVICE::ServiceProtocolProcessor<packing_traits>;

  WISP_IMPORT(SessionProtocol, MessageLoginSession);

  struct EventEntry {
    //uid_type recipient;
    unsigned int channel;
    timeout_ptr timeout;
    properties_ptr events;
    property_vector & evec;

    explicit EventEntry(const unsigned int channel = 1) :
      channel(channel), timeout(), events(new Properties),
      evec(events->create_property_vector("wspr", "event"))
    { }

    property_vector::size_type size() const {
      return evec.size();
    }

    // Destructively merge.
    void add(const Properties && event) {
      evec.push_back(new Properties(event));
    }

    // Non-destructively copy.
    void add(const Properties & event) {
      evec.push_back(new Properties(event));
    }

    void remove(unsigned int confirm) {
      const property_vector::size_type sz = size();

      if(sz > 0 && confirm > 0) {
        if(confirm >= sz) {
          evec.clear();
        } else {
          evec.erase(evec.begin(), evec.begin() + confirm);
        }
      }

#ifdef WSPR_DEBUG
      std::cerr << /*recipient <<*/ " Size: " << size() << std::endl;
#endif
    }
  };

  typedef boost::ptr_unordered_map<uid_type, EventEntry> event_map;

  TimeValue _fetch_timeout;
  event_map _events;
  StringTo _number_cast;
  
protected:
  WISP_IMPORT(SessionProtocol, MessageExpireSession);
  WISP_IMPORT(GroupSessionProtocol, MessageExpireGroupSession);

  void reply(const string & sender,
             const MessageResponse & response,
             const NS_SSRC_WISP_PROTOCOL::wisp_call_token token)
  {
    try {
#ifdef WSPR_DEBUG
      std::cerr << "MessageResponse:\n"
                << NS_SSRC_WSPR_UTILITY::to_string(*response.template_data)
                << std::endl;
#endif
      _caller.reply<CallResponse>(sender, token, response);
    } catch(std::ios_base::failure &) {
      event_map::iterator && it = _events.find(response.session->uid);

#ifdef WSPR_DEBUG
      std::cerr << "Caught std::ios_base::failure" << std::endl;
#endif

      if(it != _events.end()) {
        it->second->remove(it->second->size());
      }

      // TODO: prepare an error response.
      _caller.reply<CallResponse>(sender, token,
                                  MessageResponse(response.session, 0));
    }
  }

  void complete_fetch(const string & sender,
                      MessageResponse & response,
                      const NS_SSRC_WISP_PROTOCOL::wisp_call_token token,
                      const EventInfo &)
  {
#ifdef WSPR_DEBUG
    std::cerr << "COMPLETE FETCH " << response.session->uid << std::endl;
#endif
    event_map::iterator && it = _events.find(response.session->uid);

    if(it != _events.end()) {
      response.template_data = it->second->events;

      if(it->second->timeout)
        it->second->timeout.reset();
    }

    reply(sender, response, token);
  }

  void fetch(const WebServiceCall & call, const MessageInfo & msginfo) {
    string channel_str;

    MessageResponse response(call.session, 0);

#ifdef WSPR_DEBUG
    std::cerr << "START FETCH ";
    if(call.session)
      std::cerr << call.session->uid;
    std::cerr << std::endl;
#endif

    if(call.session) {
      unsigned int channel = 0;

      if(call.get_parameter("channel", channel_str)) {
        try {
          channel = _number_cast.cast<int>(channel_str);
        } catch(const std::bad_cast & blc) {
          goto missing_channel;
        }

        event_map::iterator && it = _events.find(call.session->uid);

        if(it != _events.end()) {
          EventEntry & entry = *it->second;
          unsigned int confirm(0);
          timeout_ptr timeout(entry.timeout);

          try {
            call.get_parameter("confirm", confirm);
          } catch(const std::bad_cast &) {
            confirm = 0;
          }

          if(confirm > 0) {
            entry.remove(confirm);
          }

          if(entry.size() > 0) {
            if(channel != entry.channel) {
              response.template_data.reset(new Properties);
              Properties & event = web_event_direct(*response.template_data,
                                                     "wspr.Event",
                                                     "ChangeChannel");
              event.set(entry.channel, "channel");
              event.set(1, "_no_confirmation");
            } else
              response.template_data = entry.events;
          }

          if(timeout) {
#ifdef WSPR_DEBUG
            std::cerr << "FETCH TIMEOUT " << call.session->uid << std::endl;
#endif
            timeout->execute();
            //entry.timeout.reset();
          }
        } else {
          // Even though we create a queue on login, keep this code
          // here because it's where we will add support for multiple
          // channels (TODO).
          std::pair<event_map::iterator,bool> && result =
            _events.insert(call.session->uid, new EventEntry);
          if(result.second)
            it = result.first;
          else
            response.set_error("could not create queue");
        }

        if(it != _events.end() && it->second->size() == 0) {
          EventEntry & entry = *it->second;
#ifdef WSPR_DEBUG
          std::cerr << "SCHEDULING FETCH COMPLETION "
                    << call.session->uid << std::endl;
#endif
          if(channel != entry.channel) {
            response.template_data.reset(new Properties);
            Properties & event = web_event_direct(*response.template_data,
                                                   "wspr.Event",
                                                   "ChangeChannel");
            if(channel == 0) {
              do {
                entry.channel++;
              } while(entry.channel == 0);
            }

            event.set(entry.channel, "channel");
            event.set(1, "_no_confirmation");
          } else {
            entry.timeout =
              schedule_timeout(boost::bind(&EventQueue::complete_fetch, this,
                                           msginfo.sender(), response,
                                           msginfo.token(), _1),
                               _fetch_timeout, EventLoop::Once);
            return;
          }
        }
      } else {
      missing_channel:
        response.set_error("missing channel");
      }
    }

#ifdef WSPR_DEBUG
    std::cerr << "END FETCH " << call.session->uid << std::endl;
#endif

    reply(msginfo.sender(), response, msginfo.token());
  }

  void process_request(const MessageDeliverEvent & msg, const MessageInfo &) {
    unsigned int size = msg.keys.size();
    event_map::iterator && end = _events.end();

    if(!msg.template_data) {
#ifdef WSPR_DEBUG
      std::cerr << "RECEIVED NULL MessageDeliverEvent::template_data!!\n";
#endif
      return;
    }

    while(size-- > 0) {
      event_map::iterator && it = _events.find(msg.keys[size]);

      if(it != end) {
        EventEntry & entry = *it->second;

#ifdef WSPR_DEBUG
        std::cerr << "DELIVER_EVENT TO: " << msg.keys[size] << std::endl
                  << "EVENT: "
                  << NS_SSRC_WSPR_UTILITY::to_string(*msg.template_data)
                  << std::endl;
#endif
        entry.add(std::move(*msg.template_data));

        if(entry.timeout) {
#ifdef WSPR_DEBUG
          std::cerr << "DELIVER_EVENT TIMEOUT " << msg.keys[size] << std::endl;
#endif
          entry.timeout->execute();
        }
      }
    }
  }

  void process_request(const MessageDeliverEvents & msg, const MessageInfo &) {
    unsigned int size = msg.keys.size();
    event_map::iterator && end = _events.end();

    if(msg.events.empty()) {
#ifdef WSPR_DEBUG
      std::cerr << "RECEIVED empty MessageDeliverEvents::events!!\n";
#endif
      return;
    }

    while(size-- > 0) {
      event_map::iterator && it = _events.find(msg.keys[size]);

      if(it != end) {
        EventEntry & entry = *it->second;
#ifdef WSPR_DEBUG
        std::cerr << "DELIVER_EVENTS TO: " << msg.keys[size] << std::endl;
#endif

        for(MessageDeliverEvents::event_container::const_iterator && event =
              msg.events.begin(), && event_end = msg.events.end();
            event != event_end; ++event)
        {
#ifdef WSPR_DEBUG
          std::cerr << "EVENT: "
                    << NS_SSRC_WSPR_UTILITY::to_string(*event)
                    << std::endl;
#endif
          entry.add(std::move(*event));
        }

        if(entry.timeout) {
#ifdef WSPR_DEBUG
          std::cerr << "DELIVER_EVENTS TIMEOUT " << msg.keys[size] << std::endl;
#endif
          entry.timeout->execute();
        }
      }
    }
  }

  void process_request(const MessageExpireSession & msg, const MessageInfo &) {
    if(msg.sessions.size() > 0) {
      MessageExpireSession::session_container::const_iterator && it =
        msg.sessions.begin();
      MessageExpireSession::session_container::const_iterator && end =
        msg.sessions.end();
      properties_unique_ptr && eventp =
        web_event("wspr.Event", "ExpireSession");
      Properties & event = *eventp;

      while(it != end) {
        event_map::iterator && queue = _events.find(*it);

        if(queue != _events.end()) {
          EventEntry & entry = *queue->second;

          if(entry.timeout) {
            entry.add(event);
            entry.timeout->execute();
          }

          _events.erase(queue);

#ifdef WSPR_DEBUG
          std::cerr <<  "EXPIRED QUEUE " << *it << std::endl;
#endif
        }

        ++it;
      }
    }
  }

  void process_request(const MessageExpireGroupSession & msg,
                       const MessageInfo &)
  {
    properties_unique_ptr && eventp =
      web_event("wspr.GroupSession", "ExpireGroupSession");
    Properties & event = *eventp;
    event_map::iterator && events_end = _events.end();

    for(GroupSessionProtocol::expiration_notification_container::const_iterator
          && it = msg.notifications.begin(), && end = msg.notifications.end();
        it != end; ++it)
    {
      event.set(it->gsid, "gsid");

      for(uid_container::const_iterator && uid = it->uids.begin(),
            uid_end = it->uids.end();
          uid != uid_end; ++uid)
      {
        event_map::iterator && queue = _events.find(*uid);

        if(queue != events_end)
          queue->second->add(event);
      }
    }
  }

  void process_request(const MessageLoginSession & msg, const MessageInfo &) {
    uid_type uid(msg.session.uid);
    event_map::iterator && it = _events.find(uid);
    if(it == _events.end()) {
#if WSPR_DEBUG
      std::cerr << "Created queue on login: " << uid << std::endl;
#endif
      _events.insert(uid, new EventEntry);
    }
  }

  virtual void transition(State state) {
    switch(state) {
    case Starting:
      _caller.join(SessionProtocol::event_group_expire());
      _caller.join(GroupSessionProtocol::event_group_expire());
      _caller.join(SessionProtocol::event_group_login());
      _caller.join(WebServiceProtocol::event_queue_ws_group());
      state = Started;
      break;
    case Stopping:
      _caller.leave(WebServiceProtocol::event_queue_ws_group());
      _caller.leave(SessionProtocol::event_group_login());
      _caller.leave(GroupSessionProtocol::event_group_expire());
      _caller.leave(SessionProtocol::event_group_expire());
      state = Stopped;
      // break;
    case Stopped:
      for(event_map::iterator && it = _events.begin(), && end = _events.end();
          it != end; ++it)
        {
          if(it->second->timeout) {
            it->second->timeout->execute();
            //it->second->timeout.reset();
          }
        }
      _events.clear();
      break;
    default:
      break;
    }

    super::transition(state);
  }

public:

  explicit EventQueue(super::caller_type & caller,
                      const EventQueueInitializer & initializer) :
    super(caller), _fetch_timeout(initializer.fetch_timeout, 0), _events(),
    _number_cast()
  {
    super::add_service_type(WebServiceProtocol::event_queue_ws_type());

    WISP_SERVICE_REQUEST(MessageDeliverEvent);
    WISP_SERVICE_REQUEST(MessageDeliverEvents);
    WISP_SERVICE_REQUEST(MessageExpireSession);
    WISP_SERVICE_REQUEST(MessageExpireGroupSession);
    WISP_SERVICE_REQUEST(MessageLoginSession);

    WS_TWO_WAY_CALL("fetch", &EventQueue::fetch);
  }

  virtual ~EventQueue() { }
};

__END_NS_SSRC_WSPR_WS

#endif