session/service.cc

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.
 */

#include <ssrc/wispers/session/service.h>
#include <ssrc/wispers/utility/checkpoint.h>


__BEGIN_NS_SSRC_WSPR_SESSION

void Session::process_request(const MessageInsert & msg, const MessageInfo &) {
  unsigned int i = msg.values.size();

  while(i-- > 0) {
    session_map::iterator && it = find<BySID>(msg.values[i].sid);

    if(it != end<BySID>()) {
      const_cast<SessionData &>(*it).attributes = msg.values[i].attributes;
      touch_session(it);
    }
  }
}

void Session::process_request(const MessageCreateSession & msg,
                              const MessageInfo & msginfo)
{
  session_map::iterator it(end<BySID>());

  if(!msg.sid.empty())
    it = find<BySID>(msg.sid);

  if(it == end<BySID>())
    it = create_session().first;

  _caller.reply<CallSingleQueryResult>(Message::FIFOSelfDiscard,
                                       msginfo.sender(), msginfo.token(), *it);
}

void Session::process_request(const MessageGetSession & msg,
                              const MessageInfo & msginfo)
{
  MessageSingleQueryResult result;
  session_map::iterator && it = find<BySID>(msg.sid);

  if(it != end<BySID>()) {
    result.found  = true;
    result.result = *it;
    if(msg.update_access_time)
      touch_session(it);
  }

  _caller.reply<CallSingleQueryResult>(msginfo.sender(), msginfo.token(),
                                       result, Message::FIFOSelfDiscard);
}

void Session::process_request(MessageSetAttributes & msg,
                              const MessageInfo &)
{
  session_map::iterator && it = find<BySID>(msg.sid);

  if(it != end<BySID>()) {
    const_cast<SessionData &>(*it).attributes->merge(std::move(msg.attributes));
    touch_session(it);
  }
}

void Session::update_session(const SessionData & update) {
  session_map::iterator && it = find<BySID>(update.sid);

  if(it != end<BySID>()) {
    SessionData session = *it;
    get_index<BySID>().erase(it);
    session.uid = update.uid;
    session.attributes = update.attributes;
    _touch_session(session);

    index_by_uid & index = get_index<ByUID>();
    index_by_uid::iterator && uit = find<ByUID>(session.uid);

    if(uit != end<ByUID>()) {
      // We don't need to send an expiration message because the next
      // client action requiring a session will clear the session.
      // Everything else (e.g., EventQueue) indexes on uid.
      index.erase(uit);
    }

    insert<BySID>(session);
  }
}

void Session::process_request(const MessageLogoutSession & msg,
                              const MessageInfo &)
{
  index_by_uid & index = get_index<ByUID>();
  index_by_uid::iterator && it = find<ByUID>(msg.session.uid);

  if(it != end<ByUID>()) {
    index.erase(it);
    MessageExpireSession expire;
    expire.sessions.push_back(msg.session.uid);
    _caller.send<CallExpireSession>(protocol_traits::event_group_expire(),
                                    expire);
  }
}

void Session::load_checkpoint(const std::string & filename)
  SSRC_DECL_THROW(std::runtime_error, boost::archive::archive_exception)
{
  NS_SSRC_WSPR_UTILITY::load_checkpoint(_checkpoint_path, super::_index);

  for(index_by_sid::iterator it = begin<BySID>(); it != end<BySID>(); ++it)
    touch_session(it);
}

void Session::transition(State state) {
  switch(state) {
  case Starting:
    _caller.join(protocol_traits::service_group());
    _caller.join(protocol_traits::event_group_login());
    _caller.join(protocol_traits::event_group_logout());
    _poll_timeout = 
      schedule_timeout(boost::bind(&Session::check_for_expirations, this),
                       TimeValue(_poll_interval, 0));
    state = Started;
    break;
  case Stopping:
    if(_poll_timeout)
      cancel_timeout(_poll_timeout);
    _caller.leave(protocol_traits::event_group_logout());
    _caller.leave(protocol_traits::event_group_login());
    _caller.leave(protocol_traits::service_group());
    state = Stopped;
    //break;
  case Stopped:
    // TODO: remove this after we implement graceful restarts/upgrades.
    //expire_sessions(boost::integer_traits<idle_count_type>::const_max);
    NS_SSRC_WSPR_UTILITY::save_checkpoint(_checkpoint_path, super::_index);
    break;
  default:
    break;
  }

  super::transition(state);
}

Session::Session(super::caller_type & caller,
                 const SessionInitializer & initializer)
  SSRC_DECL_THROW(std::runtime_error, boost::archive::archive_exception) :
  super(caller), _random_id(),
  _expirations_per_message(initializer.expirations_per_message),
  _poll_interval(initializer.poll_interval),
  _max_idle_time(initializer.max_idle_time),
  _low_count(0), _touch_session(_max_idle_time / _poll_interval),
  _poll_timeout(), _checkpoint_path(initializer.checkpoint_path)
{
  add_service_type(protocol_traits::service_type());

  WISP_SERVICE_REQUEST(MessageCreateSession);
  WISP_SERVICE_REQUEST(MessageSetAttributes);
  WISP_SERVICE_REQUEST(MessageGetSession);
  WISP_SERVICE_REQUEST(MessageUpdateSession);
  WISP_SERVICE_REQUEST(MessageLoginSession);
  WISP_SERVICE_REQUEST(MessageLogoutSession);

  load_checkpoint(_checkpoint_path);
}

void Session::expire_sessions(const idle_count_type low_count) {
  index_by_access & index = get_index<ByLastAccess>();
  index_by_access::iterator upper = index.upper_bound(low_count);
  index_by_access::iterator it = index.begin();
  MessageExpireSession msg;
  unsigned int count = 0;

  do {
    while(it != upper && count < _expirations_per_message) {
      msg.sessions.push_back(it->uid);
      it = index.erase(it);
      ++count;
    }

    if(msg.sessions.size() > 0) {
      _caller.send<CallExpireSession>(protocol_traits::event_group_expire(),
                                      msg);
      msg.sessions.clear();
    }
  } while(it != upper);
}

string Session::create_sid() {
  sid_type sid(SessionIdNumChars, ' ');
  session_map::iterator && end_ = end<BySID>();

  do {
    _random_id(sid);
    // TODO: Protect against accidental infinite loop (e.g., if we fill up)!
  } while(find<BySID>(sid) != end_);

  return sid;
}



__END_NS_SSRC_WSPR_SESSION