group_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/group_session/service.h>
#include <ssrc/wispers/ws/protocol.h>

__BEGIN_NS_SSRC_WSPR_GROUP_SESSION

using NS_SSRC_WISP_PROTOCOL::GroupMembershipDisable;
using NS_SSRC_WSPR_WS::WebServiceProtocol;

WISP_IMPORT(WebServiceProtocol, MessageDeliverEvent);
WISP_IMPORT(WebServiceProtocol, MessageDeliverEvents);
WISP_IMPORT(WebServiceProtocol, CallDeliverEvent);
WISP_IMPORT(WebServiceProtocol, CallDeliverEvents);

const string GroupSessionService::EventGroupExpireGroupSession =
  GroupSessionProtocol::event_group_expire();

/*********************
 * Support Functions *
 *********************/

void GroupSessionService::transition(State state) {
  switch(state) {
  case Starting:
    _caller.join(SessionProtocol::event_group_expire());
    _caller.join(SessionProtocol::event_group_login());
    _caller.join(protocol_traits::service_group());
    _caller.join(gs_protocol_group(_protocol_id));
    _caller.join(gs_protocol_partition_group(_protocol_id, _partition_id));
    check_for_expirations(TimeValue::now_seconds());
    _gs_poll_timeout =
      schedule_timeout(boost::bind(&GroupSessionService::_check_for_expirations, this, _1), _gs_poll_interval);
    state = Started;
    break;
  case Stopping:
    if(_gs_poll_timeout)
      cancel_timeout(_gs_poll_timeout);
    _caller.leave(gs_protocol_partition_group(_protocol_id, _partition_id));
    _caller.leave(gs_protocol_group(_protocol_id));
    _caller.leave(protocol_traits::service_group());
    _caller.leave(SessionProtocol::event_group_login());
    _caller.leave(SessionProtocol::event_group_expire());
    if(_caller.group_membership() == GroupMembershipDisable)
      state = Stopped;
    break;
  default:
    break;
  }

  super::transition(state);
}

void GroupSessionService::send_expirations(const gsid_container & gsids) {
  unsigned int id_count(0);
  GroupSessionProtocol::ExpirationNotification notification;
  MessageExpireGroupSession msg;
  unsigned int index = gsids.size();

  while(index-- > 0) {
    const unsigned int uid_count =
      _database->get_member_uids(notification.uids, gsids[index]);

    if(uid_count > 0) {
      notification.gsid = gsids[index];
      // count gsid
      ++id_count;

      if(id_count + uid_count <= _ids_per_expiration_message) {
        msg.notifications.push_back(notification);
        id_count+=uid_count;
        notification.uids.clear();
      } else {
        // TODO: unit test this section to verify it does what we want.
        GroupSessionProtocol::ExpirationNotification n;
        unsigned int copy_size, uid_offset = 0;
        n.gsid = notification.gsid;

        do {
          copy_size = _ids_per_expiration_message - id_count;

          if(copy_size > 0) {
            copy_size = std::min(copy_size, uid_count - uid_offset);
            n.uids.insert(n.uids.begin(),
                          notification.uids.begin() + uid_offset,
                          notification.uids.begin() + (uid_offset + copy_size));
            msg.notifications.push_back(n);
            n.uids.clear();
            uid_offset+=copy_size;
            id_count+=copy_size;
          }

          if(id_count >= _ids_per_expiration_message) {
            _caller.send<CallExpireGroupSession>(EventGroupExpireGroupSession,
                                                 msg);
            msg.notifications.clear();
            id_count = (uid_offset < uid_count);
          }
        } while(uid_offset < uid_count);
      }
    }
  }

  if(msg.notifications.size() > 0)
    _caller.send<CallExpireGroupSession>(EventGroupExpireGroupSession, msg);
}

unsigned int GroupSessionService::check_for_expirations(const sec_type now) {
  unsigned int session_changes(0), reservation_changes(0);

  try {
    gsid_container gsids;

    if(_database->find_expired_gsids(gsids, now) > 0) {
      // We have to send the expirations first before the members get dropped.
      send_expirations(gsids);

      DatabaseTransaction transaction(*_database);
      session_changes     = _database->remove_expired_sessions(now);
      reservation_changes = _database->remove_expired_reservations(now);
      transaction.end();
    }
  } catch(const DatabaseException & de) {
    // TODO: log and recover
  }

  _session_count-=session_changes;
  _reservation_count-=reservation_changes;

  return (session_changes + reservation_changes);
}

unsigned int GroupSessionService::cancel_reservation(const gsid_type gsid,
                                                     const uid_type requestor,
                                                     const MessageInfo &)
{
  // When we cancel a reservation, we have to remove the metadata,
  // but not when we activate it.  When a group session is removed,
  // a trigger removes the metadata, but we can't do that for
  // reservations because activation removes the reservation.
  unsigned int changes = 0;

  try {
    DatabaseTransaction transaction(*_database);
    changes = _database->remove_reservation(gsid, requestor);
    transaction.end();
  } catch(const DatabaseException & de) {
    // TODO: log and recover.
  }

  _reservation_count-=changes;

  return changes;
}

/********************
 * Service Requests *
 ********************/

void
GroupSessionService::process_request(MessageCreateGroupSession & msg,
                                     const MessageInfo &)
{
  try {
    unsigned int count = 0;
    const sec_type created(TimeValue::now_seconds());
    const sec_type expires = created + msg.lifetime;
    GroupSession session(0, created, expires, msg.session_type,
                         msg.session_group, msg.session_name,
                         msg.max_observers);
    DatabaseTransaction transaction(*_database);

    do {
      // TODO: genericize this for similar services and make the try limit
      // configurable.
      if(++count > 10)
        throw DatabaseException("GroupSession add_session failure");
      session.gsid = generate_gsid();
    } while(_database->reservation_ops.find(session.gsid).first ||
            !_database->group_session_ops.insert(session));

    for(uid_container::const_iterator && it = msg.participants.begin(),
          && end = msg.participants.end(); it != end; ++it)
    {
      _database->participant_ops.insert_statement->execute(session.gsid, *it);
    }

    transaction.end();
    ++_session_count;
    start_group_session(session, msg.payload);
  } catch(const DatabaseException & de) {
    // TODO: log and recover.
  }
}

void GroupSessionService::process_request(MessageCreateReservation & msg,
                                          const MessageInfo &)
{
  try {
    unsigned int count = 0;
    const sec_type created(TimeValue::now_seconds());
    const sec_type reservation_expires = created + msg.reservation_lifetime;
    Reservation reservation(0, created, reservation_expires,
                            msg.session_type, msg.session_group,
                            msg.session_name, msg.max_observers,
                            msg.creator_uid,
                            msg.gs_lifetime);
    DatabaseTransaction transaction(*_database);

    do {
      // TODO: genericize this for similar services and make the try limit
      // configurable.
      if(++count > 100)
        throw DatabaseException("GroupSession add_reservation failure");
      reservation.gsid = generate_gsid();
    } while(_database->group_session_ops.find(reservation.gsid).first ||
            !insert_reservation(reservation, msg.payload));

    _database->reservation_participant_ops.insert_statement->execute(reservation.gsid, msg.creator_uid);

    transaction.end();
  } catch(const DatabaseException & de) {
    // TODO: log and recover.
    recount_sessions();
  }
}

void
GroupSessionService::process_request(const MessageFindGroupSessionsByType & msg,
                                     const MessageInfo & msginfo)
{
  WISP_IMPORT(protocol_traits, query_list);

  try {
    MessageReturnGroupSessionsByType result;

    // TODO: Is it more efficient to create a dynamic query that does
    // WHERE type = ? OR type = ? OR type = ? ...
    // Obviously, that is more efficient for an non-embedded database.
    for(query_list::const_iterator it = msg.keys.begin(), end = msg.keys.end();
        it != end; ++it)
      _database->get_sessions_by_type(result.sessions, *it);

    _caller.reply<CallReturnGroupSessionsByType>(msginfo.sender(),
                                                 msginfo.token(), result,
                                                 Message::FIFOSelfDiscard);
  } catch(const DatabaseException & de) {
    // TODO: log and recover.
  }
}

void GroupSessionService::process_request(const MessageFindGroupSessions & msg,
                                          const MessageInfo & msginfo)
{
  try {
    MessageFindGroupSessionsResult result;
    result.requestor = msg.requestor;
    result.total_sessions = _session_count;
    result.limit = msg.limit;
    result.offset = msg.offset;
    result.gs_name = _gs_name;

    _database->find_sessions(result.sessions, msg.limit, msg.offset);

    _caller.send<CallFindGroupSessionsResult>(msginfo.sender(), result,
                                              Message::CausalSelfDiscard);
  } catch(const DatabaseException & de) {
    // TODO: log and recover.
  }
}

void GroupSessionService::process_request(const MessageFindMembers & msg,
                                          const MessageInfo & msginfo)
{
  try {
    MessageFindMembersResult result;

    for(gsid_container::const_iterator it = msg.sessions.begin(),
          end = msg.sessions.end(); it != end; ++it)
    {
      _database->find_participants(result.participants, *it);
      _database->find_observers(result.observers, *it);
    }
    result.requestor = msg.requestor;
    _caller.send<CallFindMembersResult>(msginfo.sender(), result,
                                        Message::CausalSelfDiscard);
  } catch(const DatabaseException & de) {
    // TODO: log and recover.
  }
}

void GroupSessionService::process_request(const MessageAddMembers & msg,
                                          const MessageInfo &)
{
  if(!msg.participants.empty() || !msg.observers.empty()) {
    try {
      DatabaseTransaction transaction(*_database);
      if(!msg.participants.empty())
        _database->participant_ops.insert(msg.participants.begin(),
                                          msg.participants.end());
      if(!msg.observers.empty())
        _database->observer_ops.insert(msg.observers.begin(),
                                       msg.observers.end());
      transaction.end();
    } catch(const DatabaseException & de) {
      // TODO: log and recover.
    }
  }
}

void GroupSessionService::process_request(const MessageRemoveMembers & msg,
                                          const MessageInfo &)
{
  if(!msg.participants.empty() || !msg.observers.empty()) {
    try {
      DatabaseTransaction transaction(*_database);
      if(!msg.participants.empty())
        _database->participant_ops.erase_range(msg.participants.begin(),
                                               msg.participants.end());
      if(!msg.observers.empty())
        _database->observer_ops.erase_range(msg.observers.begin(),
                                            msg.observers.end());
      transaction.end();
    } catch(const DatabaseException & de) {
      // TODO: log and recover.
    }
  }
}

void GroupSessionService::process_request(const MessageAddMember & msg,
                                          const MessageInfo & msginfo)
{
  try {
    protocol_traits::AddMemberResultCode result_code =
      protocol_traits::AddMemberInternalError;
    GroupSessionDatabase::GroupSessionOperations::find_result_type
      find_result = _database->group_session_ops.find(msg.member.gsid);

    if(find_result.first) {
      if(!_database->is_member(msg.member.uid, msg.member.gsid)) {
        bool success(false);

        if(msg.member_type == protocol_traits::Observer) {
          const unsigned int max_observers = find_result.second.max_observers;

          if(max_observers > 0 &&
             _database->count_observers(msg.member.gsid) < max_observers)
          {
            success = _database->observer_ops.insert(msg.member);
          } else
            result_code = protocol_traits::AddMemberMaxObserversReached;
        } else {
          success = _database->participant_ops.insert(msg.member);
        }

        if(success)
          result_code = protocol_traits::AddMemberSuccess;
      } else
        result_code = protocol_traits::AddMemberDuplicateEntry;
    } else
      result_code = protocol_traits::AddMemberNonexistentGroupSession;

    _caller.sendp<CallAddMemberConfirm>(msginfo.sender(),
                                        result_code,
                                        msg.member_type,
                                        msg.member.uid,
                                        _gs_name,
                                        find_result.second);
  } catch(const DatabaseException & de) {
    // TODO: log and recover.
    //result_code = protocol_traits::AddMemberInternalError;
 }
}

void GroupSessionService::process_request(const MessageRemoveMember & msg,
                                          const MessageInfo & msginfo)
{
  try {
    unsigned int changes = 0;

    switch(msg.member_type) {
    case GroupSessionProtocol::Participant:
      changes = _database->participant_ops.erase(msg.member);
      break;
    case GroupSessionProtocol::Observer:
      changes = _database->observer_ops.erase(msg.member);
      break;
    }

    if(changes > 0) {
      GroupSessionDatabase::GroupSessionOperations::find_result_type
        find_result = _database->group_session_ops.find(msg.member.gsid);

      if(find_result.first) {
        _caller.
          sendp<CallRemoveMemberConfirm>(msginfo.sender(),
                                         msg.member_type, msg.member, _gs_name,
                                         find_result.second);
      }
    }
  } catch(const DatabaseException & de) {
    // TODO: log and recover.
  }
}

void GroupSessionService::process_request(const MessageFindGroupSessionsForMember & msg,
                                          const MessageInfo & msginfo)
{
  try {
    MessageReturnGroupSessionsForMember result;
    result.uid = msg.uid;
    result.gs_name = _gs_name;
    _database->get_participant_sessions(result.participating, msg.uid);
    _database->get_observer_sessions(result.observing, msg.uid);
    _database->get_participant_reservations(result.awaiting, msg.uid);
    _caller.reply<CallReturnGroupSessionsForMember>(msginfo.sender(),
                                                    msginfo.token(), result,
                                                    Message::FIFOSelfDiscard);
  } catch(const DatabaseException & de) {
    // TODO: log and recover
  }
}

void GroupSessionService::process_request(const MessageRelayEvent & msg,
                                          const MessageInfo &)
{
  try {
    MessageDeliverEvent mde(0);
    _database->get_member_uids(mde.keys, msg.gsid);

    if(mde.keys.size() > 0) {
      mde.template_data = msg.event;
      _caller.send<CallDeliverEvent>(msg.event_queue, mde);
    }
  } catch(const DatabaseException & de) {
    // TODO: log and recover
  }
}

void
GroupSessionService::process_request(const MessageRelayEventSelfDiscard & msg,
                                     const MessageInfo &)
{
  try {
    MessageDeliverEvent mde(0);
    _database->get_member_uids_discard(mde.keys, msg.gsid, msg.source);

    if(!mde.keys.empty()) {
      mde.template_data = msg.event;
      _caller.send<CallDeliverEvent>(msg.event_queue, mde);
    }
  } catch(const DatabaseException & de) {
    // TODO: log and recover
  }
}

void GroupSessionService::process_request(MessageRelayEvents & msg,
                                          const MessageInfo &)
{
  try {
    MessageDeliverEvents mde(std::move(msg.events));
    _database->get_member_uids(mde.keys, msg.gsid);

    if(mde.keys.size() > 0) {
      _caller.send<CallDeliverEvents>(msg.event_queue, mde);
    }
  } catch(const DatabaseException & de) {
    // TODO: log and recover
  }
}

void
GroupSessionService::process_request(MessageRelayEventsSelfDiscard & msg,
                                     const MessageInfo &)
{
  try {
    MessageDeliverEvents mde(std::move(msg.events));
    _database->get_member_uids_discard(mde.keys, msg.gsid, msg.source);

    if(!mde.keys.empty()) {
      _caller.send<CallDeliverEvents>(msg.event_queue, mde);
    }
  } catch(const DatabaseException & de) {
    // TODO: log and recover
  }
}

void GroupSessionService::process_request(const MessageExpireSession & msg,
                                          const MessageInfo &)
{
  using NS_SSRC_WSPR_WS::web_event_indirect;
  using NS_SSRC_WSPR_UTILITY::Properties;
  using NS_SSRC_WSPR_UTILITY::primitive_property_vector;

  if(!msg.sessions.empty()) {
    try {
      // TODO: optimize by reducing number of queries, messages, and
      //       memory operations.
      //       for example, if we add a DeliverEvents message to EventQueue
      //       we can batch the delivery of multiple events to different
      //       sets of uids.
      // Two things are wrong with this implementation:
      //   1. It's a brute force approach
      //   2. It has direct knowledge of the event queue, which we normally
      //      place at the Web service level.
      // Brute force approach:
      //   1. For each uid, get all members of sessions where uid is
      //      a participant or observer, using as keys for a DeliverEvent
      //      message.  If keys is non-zero in size, add a web_event called
      //      ExpireMember (with protocol type "wspr.GroupSession" (don't
      //      hard code that if avoidable)) containing the expired session uid.
      for(MessageExpireSession::session_container::const_iterator
            && it = msg.sessions.begin(), && end = msg.sessions.end();
          it != end; ++it)
      {
        MessageDeliverEvent mde;

        _database->find_members_for_uid(mde.keys, *it);

        if(!mde.keys.empty()) {
          Properties & event = web_event_indirect(*mde.template_data,
                                                  "wspr.GroupSession",
                                                  "ExpireMember");
          event.set(*it, "uid");
          _caller.send<CallDeliverEvent>(WebServiceProtocol::event_queue_ws_group(), mde);
        }
      }
      {
        DatabaseTransaction transaction(*_database);
        _database->remove_observers_by_uid(msg.sessions.begin(),
                                           msg.sessions.end());
        transaction.end();
      }
    } catch(const DatabaseException & de) {
      // TODO: log and recover
    }
  }
}

void GroupSessionService::process_request(const MessageLoginSession & msg,
                                          const MessageInfo &)
{
  using NS_SSRC_WSPR_WS::web_event_indirect;
  using NS_SSRC_WSPR_UTILITY::Properties;

  const uid_type uid(msg.session.uid);

  try {
    // TODO: optimize
    // Two things are wrong with this implementation:
    //   1. It's a brute force approach (in the sense that all group session
    //      services do a lookup on login)
    //   2. It has direct knowledge of the event queue, which we normally
    //      place at the Web service level.
    MessageDeliverEvent mde;
    _database->find_members_for_participant(mde.keys, uid);

    if(!mde.keys.empty()) {
      Properties & event = web_event_indirect(*mde.template_data,
                                              "wspr.GroupSession",
                                              "LoginParticipant");
      event.set(uid, "uid");
      _caller.send<CallDeliverEvent>(WebServiceProtocol::event_queue_ws_group(), mde);
    }
  } catch(const DatabaseException & de) {
    // TODO: log and recover and recover
  }
}

/****************************
 * Constructors/Destructors *
 ****************************/


GroupSessionService::GroupSessionService(super::caller_type & caller,
                                         const wisp_message_protocol protocol,
                                         GroupSessionDatabase *db, 
                                         const GroupSessionInitializer & initializer)
  SSRC_DECL_THROW(DatabaseException, std::invalid_argument) :
  super(caller),
  _protocol_id(protocol),
  _partition_id(initializer.partition_id),
  _num_partitions(initializer.num_partitions),
  _ids_per_expiration_message(initializer.ids_per_expiration_message),
  _database(db),
  _gsid_min(gsid_min(_protocol_id, _partition_id, _num_partitions)),
  _gsid_max(gsid_max(_protocol_id, _partition_id, _num_partitions)),
  _random(_gsid_min, _gsid_max),
  _gs_poll_interval(initializer.gs_poll_interval, 0),
  _gs_poll_timeout(),
  _session_count(_database->count_sessions()),
  _reservation_count(_database->count_reservations()),
  _gs_name(initializer.gs_name)
{
  if(_partition_id >= _num_partitions)
    SSRC_DECL_THROW(std::invalid_argument("invalid group session partition"));

  add_service_type(protocol_traits::service_type());

  WISP_SERVICE_REQUEST(MessageCreateGroupSession);
  WISP_SERVICE_REQUEST(MessageEndGroupSession);
  WISP_SERVICE_REQUEST(MessageRelayEvent);
  WISP_SERVICE_REQUEST(MessageRelayEventSelfDiscard);
  WISP_SERVICE_REQUEST(MessageRelayEvents);
  WISP_SERVICE_REQUEST(MessageRelayEventsSelfDiscard);
  WISP_SERVICE_REQUEST(MessageFindGroupSessions);
  WISP_SERVICE_REQUEST(MessageFindGroupSessionsForMember);
  WISP_SERVICE_REQUEST(MessageFindGroupSessionsByType);
  WISP_SERVICE_REQUEST(MessageCreateReservation);
  WISP_SERVICE_REQUEST(MessageCancelReservation);
  WISP_SERVICE_REQUEST(MessageFindMembers);
  WISP_SERVICE_REQUEST(MessageAddMembers);
  WISP_SERVICE_REQUEST(MessageRemoveMembers);
  WISP_SERVICE_REQUEST(MessageAddMember);
  WISP_SERVICE_REQUEST(MessageRemoveMember);
  WISP_SERVICE_REQUEST(MessageExpireSession);
  WISP_SERVICE_REQUEST(MessageLoginSession);
}

__END_NS_SSRC_WSPR_GROUP_SESSION