group_session/protocol.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_PROTOCOL_H
#define __SSRC_WSPR_GROUP_SESSION_PROTOCOL_H

#include <ssrc/wispers/protocol.h>
#include <ssrc/wispers/utility/Properties.h>
#include <ssrc/wispers/utility/ToString.h>
#include <ssrc/wispers/group_session/types.h>

// Includes serialize function for std::pair.
#include <boost/serialization/utility.hpp>

__BEGIN_NS_SSRC_WSPR_PROTOCOL

WSPR_DEFINE_PROTOCOL(GroupSession,group_session);

__END_NS_SSRC_WSPR_PROTOCOL

__BEGIN_NS_SSRC_WSPR_GROUP_SESSION

using NS_SSRC_WSPR::ByteBuffer;
using NS_SSRC_WSPR_UTILITY::Properties;
using NS_SSRC_WSPR_UTILITY::properties_ptr;
using NS_SSRC_WSPR_UTILITY::ToString;
using NS_SSRC_WISP_PROTOCOL::wisp_message_protocol;

typedef std::vector<gsid_type> gsid_container;
typedef std::vector<uid_type> uid_container;
typedef std::vector<GroupSession> group_session_container;
typedef std::vector<Reservation> reservation_container;
typedef std::vector<Member> member_container;

struct GroupSessionProtocol :
  public protocol::ServiceProtocol<protocol::GroupSession>
{
  typedef protocol::ServiceProtocol<protocol::GroupSession> super;

  WISP_IMPORT(super, caller_type);

  static const gsid_type GSIDMax = boost::integer_traits<gsid_type>::const_max;

  static const gsid_type GSIDMin = boost::integer_traits<gsid_type>::const_min;

  // Calculation is independent of whether gsid_type is signed or unsigned.
  // Protocol numbers start at 0, which is why we use
  // boost::integer_traits<wisp_message_protocol>::const_max without respect
  // to the distance to const_min.
  static const gsid_type GSIDSpace =
      (((boost::integer_traits<gsid_type>::const_max >> 2) + 1)
       - (boost::integer_traits<gsid_type>::const_min >> 2))  /
      ((boost::integer_traits<wisp_message_protocol>::const_max >> 2) + 1);

  static string event_group_expire() {
    return WSPR_EVENT_GROUP("group_session", "expire");
  }

  enum {
    RelayEvent,
    RelayEventSelfDiscard,
    RelayEvents,
    RelayEventsSelfDiscard,
    CreateGroupSession,
    EndGroupSession,
    ExpireGroupSession,
    FindGroupSessions,
    FindGroupSessionsResult,
    FindGroupSessionsForMember,
    ReturnGroupSessionsForMember,
    FindGroupSessionsByType,
    ReturnGroupSessionsByType,
    CreateReservation,
    CancelReservation,
    StartGroupSession,
    FindMembers,
    FindMembersResult,
    AddMembers,
    RemoveMembers,
    AddMember,
    AddMemberConfirm,
    RemoveMember,
    RemoveMemberConfirm
  };

  enum { DefaultPayloadCapacity = NS_SSRC_SPREAD::Message::DefaultCapacity };

  enum MemberType { Participant, Observer };

  struct MessageCreateGroupSession :
    public protocol::MessageGroupSession<CreateGroupSession>
  {
    string session_name;
    string session_group;
    string session_type;
    sec_type lifetime;
    uid_container participants;
    unsigned int max_observers;
    ByteBuffer payload;

    MessageCreateGroupSession() : payload(DefaultPayloadCapacity) { }

    MessageCreateGroupSession(const string & session_name,
                              const string & session_group,
                              const string & session_type,
                              const sec_type lifetime,
                              const uid_container & participants,
                              const unsigned int max_observers) :
      session_name(session_name),
      session_group(session_group),
      session_type(session_type),
      lifetime(lifetime),
      participants(participants),
      max_observers(max_observers),
      payload(DefaultPayloadCapacity)
    { }

    template<class Archive>
    void serialize(Archive & ar, const unsigned int) {
      ar & session_name & session_group & session_type & lifetime
         & participants & max_observers & payload;
    }
  };

  struct MessageCreateReservation :
    public protocol::MessageGroupSession<CreateReservation>
  {
    string session_name;
    string session_group;
    string session_type;
    sec_type gs_lifetime;
    uid_type creator_uid;
    sec_type reservation_lifetime;
    unsigned int max_observers;
    ByteBuffer payload;

    MessageCreateReservation() : payload(DefaultPayloadCapacity) { }

    MessageCreateReservation(const string & session_name,
                             const string & session_group,
                             const string & session_type,
                             const sec_type gs_lifetime,
                             const uid_type creator_uid,
                             const sec_type reservation_lifetime,
                             const unsigned int max_observers) :
      session_name(session_name),
      session_group(session_group),
      session_type(session_type),
      gs_lifetime(gs_lifetime),
      creator_uid(creator_uid),
      reservation_lifetime(reservation_lifetime),
      max_observers(max_observers),
      payload(DefaultPayloadCapacity)
    { }

    template<class Archive>
    void serialize(Archive & ar, const unsigned int) {
      ar & session_name & session_group & session_type & gs_lifetime
        & creator_uid & reservation_lifetime & max_observers & payload;
    }
  };

  WISP_PROTOCOL_MESSAGE(EndGroupSession, protocol::MessageGroupSession,
                        ((gsid_type, gsid)));

  WISP_PROTOCOL_MESSAGE(CancelReservation, protocol::MessageGroupSession,
                        ((gsid_type, gsid))((uid_type, requestor)));

  WISP_PROTOCOL_MESSAGE(StartGroupSession,
                        protocol::MessageGroupSession,
                        ((string, gs_name))
                        ((GroupSession, session)));

  WISP_STRUCT(ExpirationNotification,
              ((gsid_type, gsid))
              ((uid_container, uids)));
  typedef
  std::vector<ExpirationNotification> expiration_notification_container;
  WISP_PROTOCOL_MESSAGE(ExpireGroupSession, protocol::MessageGroupSession,
                        ((expiration_notification_container, notifications)));

  WISP_PROTOCOL_MESSAGE_WITH_INIT(RelayEvent, protocol::MessageGroupSession,
                                  ((gsid_type, gsid))
                                  ((string, event_queue))
                                  ((properties_ptr, event)),
                                  ((gsid_type()))
                                  ((string()))
                                  ((properties_ptr(new Properties))));

  WISP_PROTOCOL_MESSAGE_WITH_INIT(RelayEventSelfDiscard,
                                  protocol::MessageGroupSession,
                                  ((uid_type, source))
                                  ((gsid_type, gsid))
                                  ((string, event_queue))
                                  ((properties_ptr, event)),
                                  ((uid_type()))
                                  ((gsid_type()))
                                  ((string()))
                                  ((properties_ptr(new Properties))));

  WISP_PROTOCOL_MESSAGE_WITH_INIT(RelayEvents, protocol::MessageGroupSession,
                                  ((gsid_type, gsid))
                                  ((string, event_queue))
                                  ((std::vector<Properties>, events)),
                                  ((gsid_type()))
                                  ((string()))
                                  ((std::vector<Properties>())));

  WISP_PROTOCOL_MESSAGE_WITH_INIT(RelayEventsSelfDiscard,
                                  protocol::MessageGroupSession,
                                  ((uid_type, source))
                                  ((gsid_type, gsid))
                                  ((string, event_queue))
                                  ((std::vector<Properties>, events)),
                                  ((uid_type()))
                                  ((gsid_type()))
                                  ((string()))
                                  ((std::vector<Properties>())));

  WISP_PROTOCOL_MESSAGE(FindGroupSessions, protocol::MessageGroupSession,
                        ((uid_type, requestor))
                        ((db_limit_type, limit))
                        ((db_offset_type, offset)));

  WISP_PROTOCOL_MESSAGE(FindGroupSessionsResult, protocol::MessageGroupSession,
                        ((uid_type, requestor))
                        ((unsigned int, total_sessions))
                        ((db_limit_type, limit))
                        ((db_offset_type, offset))
                        ((string, gs_name))
                        ((group_session_container, sessions)));

  WISP_PROTOCOL_MESSAGE(FindGroupSessionsForMember,
                        protocol::MessageGroupSession,
                        ((uid_type, uid)));

  WISP_PROTOCOL_MESSAGE(ReturnGroupSessionsForMember,
                        protocol::MessageGroupSession,
                        ((uid_type, uid))
                        ((string, gs_name))
                        ((group_session_container, participating))
                        ((group_session_container, observing))
                        ((reservation_container, awaiting)));

  typedef std::vector<string> query_list;
  WISP_PROTOCOL_MESSAGE(FindGroupSessionsByType, protocol::MessageGroupSession,
                        ((query_list, keys)));
  WISP_PROTOCOL_MESSAGE(ReturnGroupSessionsByType,
                        protocol::MessageGroupSession,
                        ((group_session_container, sessions)));

  WISP_PROTOCOL_MESSAGE(FindMembers,
                        protocol::MessageGroupSession,
                        ((uid_type, requestor))
                        ((gsid_container, sessions)));

  WISP_PROTOCOL_MESSAGE(FindMembersResult,
                        protocol::MessageGroupSession,
                        ((uid_type, requestor))
                        ((member_container, participants))
                        ((member_container, observers)));

  WISP_PROTOCOL_MESSAGE(AddMembers,
                        protocol::MessageGroupSession,
                        ((member_container, participants))
                        ((member_container, observers)));

  WISP_PROTOCOL_MESSAGE(RemoveMembers,
                        protocol::MessageGroupSession,
                        ((member_container, participants))
                        ((member_container, observers)));

  WISP_PROTOCOL_MESSAGE(AddMember,
                        protocol::MessageGroupSession,
                        ((MemberType, member_type))
                        ((Member, member)));

  enum AddMemberResultCode {
    AddMemberInternalError,
    AddMemberNonexistentGroupSession,
    AddMemberDuplicateEntry,
    AddMemberMaxObserversReached,
    AddMemberSuccess
  };

  WISP_PROTOCOL_MESSAGE(AddMemberConfirm,
                        protocol::MessageGroupSession,
                        ((AddMemberResultCode, result))
                        ((MemberType, member_type))
                        ((uid_type, uid))
                        ((string, gs_name))
                        ((GroupSession, session)));

  WISP_PROTOCOL_MESSAGE(RemoveMember,
                        protocol::MessageGroupSession,
                        ((MemberType, member_type))
                        ((Member, member)));

  WISP_PROTOCOL_MESSAGE(RemoveMemberConfirm,
                        protocol::MessageGroupSession,
                        ((MemberType, member_type))
                        ((Member, member))
                        ((string, gs_name))
                        ((GroupSession, session)));

  WISP_ONE_WAY_CALL(caller_type, CreateGroupSession);
  WISP_ONE_WAY_CALL(caller_type, EndGroupSession);
  WISP_ONE_WAY_CALL(caller_type, ExpireGroupSession);
  WISP_ONE_WAY_CALL(caller_type, RelayEvent);
  WISP_ONE_WAY_CALL(caller_type, RelayEventSelfDiscard);
  WISP_ONE_WAY_CALL(caller_type, RelayEvents);
  WISP_ONE_WAY_CALL(caller_type, RelayEventsSelfDiscard);

  WISP_ONE_WAY_CALL(caller_type, CreateReservation);
  WISP_ONE_WAY_CALL(caller_type, CancelReservation);
  WISP_ONE_WAY_CALL(caller_type, StartGroupSession);

  WISP_TWO_WAY_CALL(caller_type, FindGroupSessionsForMember,
                    ReturnGroupSessionsForMember);
  WISP_ONE_WAY_CALL(caller_type, ReturnGroupSessionsForMember);

  WISP_TWO_WAY_CALL(caller_type, FindGroupSessionsByType,
                    ReturnGroupSessionsByType);
  WISP_ONE_WAY_CALL(caller_type, ReturnGroupSessionsByType);

  // Conceptually these are two-way calls, but we don't invoke them that way.
  WISP_ONE_WAY_CALL(caller_type, FindGroupSessions);
  WISP_ONE_WAY_CALL(caller_type, FindGroupSessionsResult);
  WISP_ONE_WAY_CALL(caller_type, FindMembers);
  WISP_ONE_WAY_CALL(caller_type, FindMembersResult);


  WISP_ONE_WAY_CALL(caller_type, AddMembers);
  WISP_ONE_WAY_CALL(caller_type, RemoveMembers);
  WISP_ONE_WAY_CALL(caller_type, AddMember);
  WISP_ONE_WAY_CALL(caller_type, AddMemberConfirm);
  WISP_ONE_WAY_CALL(caller_type, RemoveMember);
  WISP_ONE_WAY_CALL(caller_type, RemoveMemberConfirm);
};

template<typename session_type>
inline void bind_group_session_properties(const session_type & session,
                                          Properties & node)
{
  node.set(session.gsid, "gsid");
  node.set(session.created, "created");
  node.set(session.expires, "expires");
  node.set(session.type, "type");
  node.set(session.name, "name");
}

inline gsid_type gsid_min(const wisp_message_protocol protocol) {
  return
    (GroupSessionProtocol::GSIDMin +
     (static_cast<gsid_type>(protocol) * GroupSessionProtocol::GSIDSpace));
}

inline gsid_type gsid_max(const wisp_message_protocol protocol) {
  return
    ((GroupSessionProtocol::GSIDMin +
      (static_cast<gsid_type>(protocol) + 1) * GroupSessionProtocol::GSIDSpace) - 1);
}

inline gsid_type gsid_min(const wisp_message_protocol protocol,
                          const unsigned int partition_id,
                          const unsigned int num_partitions)
{
  const gsid_type min_gsid = gsid_min(protocol);
  const gsid_type space = GroupSessionProtocol::GSIDSpace / num_partitions;

  return (min_gsid + space * partition_id);
}

inline gsid_type gsid_max(const wisp_message_protocol protocol,
                          const unsigned int partition_id,
                          const unsigned int num_partitions)
{
  // Spill over space rounding error to last partition.
  if(partition_id >= (num_partitions - 1))
    return gsid_max(protocol);

  return gsid_min(protocol, partition_id + 1, num_partitions) - 1;
}

inline wisp_message_protocol gsid_to_protocol(const gsid_type gsid) {
  // Can't use basic algebra because of overflow.
  gsid_type result(gsid);

  if(gsid < 0)
    result-=GroupSessionProtocol::GSIDMin;

  result /= GroupSessionProtocol::GSIDSpace;

  if(gsid >= 0) {
    const gsid_type midpoint = 
      (boost::integer_traits<wisp_message_protocol>::const_min / -2);
    result+=midpoint;
  }

  return static_cast<wisp_message_protocol>(result);
}

inline
unsigned int gsid_to_protocol_partition(gsid_type gsid,
                                        const unsigned int num_partitions)
{
  const wisp_message_protocol protocol = gsid_to_protocol(gsid);
  const gsid_type space = GroupSessionProtocol::GSIDSpace / num_partitions;

  gsid-=gsid_min(protocol);
  gsid /= space;

  if(gsid >= num_partitions)
    gsid = num_partitions - 1;

  return static_cast<unsigned int>(gsid);
}

class GSToGroup {
  const string _gs_protocol_prefix;
  string _gs_group;
  ToString _string_cast;

public:

  GSToGroup() :
    _gs_protocol_prefix("wspr.gs.protocol."),
    _gs_group(_gs_protocol_prefix),
    _string_cast()
  { }

  const string & gs_protocol_group(const wisp_message_protocol protocol) {
    _gs_group.erase(_gs_protocol_prefix.size());
    return _gs_group.append(_string_cast(protocol));
  }

  const string &
  gs_protocol_partition_group(const wisp_message_protocol protocol,
                              const unsigned int partition_id)
  {
    _gs_group.erase(_gs_protocol_prefix.size());
    _gs_group.append(_string_cast(protocol)).append(".");
    // Can't string together all of the appends because of undefined
    // execution order of _string_cast, which returns a reference and
    // can cause the protocol value to be used in each case.
    return _gs_group.append(_string_cast(partition_id));
  }

  const string & gsid_protocol_group(const gsid_type gsid) {
    return gs_protocol_group(gsid_to_protocol(gsid));
  }
};

inline string gs_protocol_group(const wisp_message_protocol protocol) {
  GSToGroup gsid_to_group;
  return gsid_to_group.gs_protocol_group(protocol);
}

inline
string gs_protocol_partition_group(const wisp_message_protocol protocol,
                                   const unsigned int partition_id)
{
  GSToGroup gsid_to_group;
  return gsid_to_group.gs_protocol_partition_group(protocol, partition_id);
}

inline string gsid_protocol_group(const gsid_type gsid) {
  GSToGroup gsid_to_group;
  return gsid_to_group.gsid_protocol_group(gsid);
}

__END_NS_SSRC_WSPR_GROUP_SESSION

#endif