group_session/service.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_GROUP_SESSION_SERVICE_H
#define __SSRC_WSPR_GROUP_SESSION_SERVICE_H

#include <ssrc/wispers/service/service.h>
#include <ssrc/wispers/session/protocol.h>
#include <ssrc/wispers/group_session/GroupSessionDatabase.h>
#include <ssrc/wispers/group_session/protocol.h>
#include <ssrc/wispers/group_session/GroupSessionOptions.h>
#include <ssrc/wispers/utility/Random.h>

__BEGIN_NS_SSRC_WSPR_GROUP_SESSION

using NS_SSRC_SPREAD::Message;
using NS_SSRC_WISP_PROTOCOL::MessageInfo;
using NS_SSRC_WSPR_SERVICE::EventInfo;
using NS_SSRC_WSPR_SERVICE::timeout_ptr;
using NS_SSRC_WSPR_SESSION::SessionProtocol;
using NS_SSRC_WSPR_SERVICE::ServiceProtocolProcessor;
using NS_SSRC_WSPR_UTILITY::random_engine;
using NS_SSRC_WSPR_UTILITY::Random;

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

protected:
  typedef GroupSessionProtocol protocol_traits;

  WISP_IMPORT(protocol_traits, MessageCreateGroupSession);
  WISP_IMPORT(protocol_traits, MessageEndGroupSession);
  WISP_IMPORT(protocol_traits, MessageExpireGroupSession);
  WISP_IMPORT(protocol_traits, MessageRelayEvent);
  WISP_IMPORT(protocol_traits, MessageRelayEventSelfDiscard);
  WISP_IMPORT(protocol_traits, MessageRelayEvents);
  WISP_IMPORT(protocol_traits, MessageRelayEventsSelfDiscard);
  WISP_IMPORT(protocol_traits, MessageFindGroupSessions);
  WISP_IMPORT(protocol_traits, MessageFindGroupSessionsResult);
  WISP_IMPORT(protocol_traits, MessageFindGroupSessionsForMember);
  WISP_IMPORT(protocol_traits, MessageReturnGroupSessionsForMember);
  WISP_IMPORT(protocol_traits, MessageFindGroupSessionsByType);
  WISP_IMPORT(protocol_traits, MessageReturnGroupSessionsByType);
  WISP_IMPORT(protocol_traits, MessageCreateReservation);
  WISP_IMPORT(protocol_traits, MessageCancelReservation);

  WISP_IMPORT(protocol_traits, MessageFindMembers);
  WISP_IMPORT(protocol_traits, MessageFindMembersResult);
  WISP_IMPORT(protocol_traits, MessageAddMembers);
  WISP_IMPORT(protocol_traits, MessageRemoveMembers);
  WISP_IMPORT(protocol_traits, MessageAddMember);
  WISP_IMPORT(protocol_traits, MessageRemoveMember);

  WISP_IMPORT(SessionProtocol, MessageExpireSession);
  WISP_IMPORT(SessionProtocol, MessageLoginSession);

  WISP_IMPORT(protocol_traits, CallStartGroupSession);
  WISP_IMPORT(protocol_traits, CallExpireGroupSession);
  WISP_IMPORT(protocol_traits, CallReturnGroupSessionsForMember);
  WISP_IMPORT(protocol_traits, CallReturnGroupSessionsByType);
  WISP_IMPORT(protocol_traits, CallFindGroupSessionsResult);
  WISP_IMPORT(protocol_traits, CallFindMembersResult);
  WISP_IMPORT(protocol_traits, CallAddMemberConfirm);
  WISP_IMPORT(protocol_traits, CallRemoveMemberConfirm);

  static const string EventGroupExpireGroupSession;

  const unsigned int _protocol_id;
  const unsigned int _partition_id;
  const unsigned int _num_partitions;
  const unsigned int _ids_per_expiration_message;
  boost::scoped_ptr<GroupSessionDatabase> _database;
  const gsid_type _gsid_min, _gsid_max;
  Random<gsid_type> _random;
  TimeValue _gs_poll_interval;
  timeout_ptr _gs_poll_timeout;
  unsigned int _session_count, _reservation_count;
  const std::string _gs_name;

  gsid_type generate_gsid() {
    return _random();
  }

  void recount_sessions() SSRC_DECL_THROW(DatabaseException) {
    _session_count = _database->count_sessions();
    _reservation_count = _database->count_reservations();
  }

  virtual void start_group_session(const GroupSession & session,
                                   ByteBuffer & payload)
  {
    _caller.sendp<CallStartGroupSession>(session.group, _gs_name, session);
  }

  void process_request(MessageCreateGroupSession & msg,
                       const MessageInfo &);


  virtual void end_group_session(const gsid_type gsid) {
    DatabaseTransaction transaction(*_database);
    if(_database->group_session_ops.erase(gsid) == 1)
      --_session_count;
    transaction.end();
    
  }

  void process_request(const MessageEndGroupSession & msg,
                       const MessageInfo &)
  {
    end_group_session(msg.gsid);
  }

  virtual bool insert_reservation(const Reservation & reservation,
                                  ByteBuffer & payload)
    SSRC_DECL_THROW(DatabaseException)
  {
    if(_database->reservation_ops.insert(reservation)) {
      ++_reservation_count;
      return true;
    }
    return false;
  }

  void process_request(MessageCreateReservation & msg,
                       const MessageInfo &);

  virtual void confirm_reservation_activation(const GroupSession & session) {
    _caller.sendp<CallStartGroupSession>(session.group, _gs_name, session);
  }


  template<typename reservation_type>
  void activate_reservation(const reservation_type & reservation,
                            const string & session_name,
                            const member_container & additional_participants)
  {
    try {
      const sec_type created(TimeValue::now_seconds());
      const GroupSession session(reservation.gsid,
                                 created,
                                 created + reservation.gs_lifetime,
                                 reservation.type,
                                 reservation.group,
                                 (session_name.empty() ?
                                  reservation.name :
                                  session_name),
                                 reservation.max_observers);
      DatabaseTransaction transaction(*_database);

      if(_database->group_session_ops.insert(session)) {
        ++_session_count;
        _database->activate_reservation_participants(session.gsid);

        if(additional_participants.size() > 0)
          _database->participant_ops.insert(additional_participants.begin(),
                                            additional_participants.end());

        if(_database->remove_reservation(session.gsid,
                                         reservation.creator_uid) == 1)
        {
          --_reservation_count;
        }
      }

      transaction.end();

      confirm_reservation_activation(session);
    } catch(const DatabaseException & de) {
      // TODO: log.
      recount_sessions();
    }
  }

  void process_request(const MessageFindGroupSessionsByType & msg,
                       const MessageInfo & msginfo);


  void process_request(const MessageFindGroupSessions & msg,
                       const MessageInfo & msginfo);

  void process_request(const MessageFindMembers & msg,
                       const MessageInfo & msginfo);

  void process_request(const MessageAddMembers & msg,
                       const MessageInfo &);

  void process_request(const MessageRemoveMembers & msg,
                       const MessageInfo &);

  void process_request(const MessageAddMember & msg,
                       const MessageInfo & msginfo);

  void process_request(const MessageRemoveMember & msg,
                       const MessageInfo & msginfo);

  virtual unsigned int cancel_reservation(const gsid_type gsid,
                                          const uid_type requestor,
                                          const MessageInfo &);

  void process_request(const MessageCancelReservation & msg,
                       const MessageInfo & msginfo)
  {
    cancel_reservation(msg.gsid, msg.requestor, msginfo);
  }

  void process_request(const MessageFindGroupSessionsForMember & msg,
                       const MessageInfo & msginfo);

  void process_request(const MessageRelayEvent & msg, const MessageInfo &);

  void process_request(const MessageRelayEventSelfDiscard & msg,
                       const MessageInfo &);

  void process_request(MessageRelayEvents & msg, const MessageInfo &);

  void process_request(MessageRelayEventsSelfDiscard & msg,
                       const MessageInfo &);

  void process_request(const MessageExpireSession & msg,
                       const MessageInfo &);

  void process_request(const MessageLoginSession & msg,
                       const MessageInfo &);

  virtual void transition(State state);

public:

  GroupSessionService(super::caller_type & caller,
                      const wisp_message_protocol protocol,
                      GroupSessionDatabase *db, 
                      const GroupSessionInitializer & initializer)
    SSRC_DECL_THROW(DatabaseException, std::invalid_argument);

  virtual ~GroupSessionService() { }

  virtual unsigned int check_for_expirations(const sec_type now);

private:

  void send_expirations(const gsid_container & gsids);

  void _check_for_expirations(const EventInfo & info) {
    check_for_expirations(TimeValue::now_seconds());
  }
};

__END_NS_SSRC_WSPR_GROUP_SESSION

#endif