registry/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_REGISTRY_SERVICE_H
#define __SSRC_WSPR_REGISTRY_SERVICE_H

#include <ssrc/wispers/index/IndexService.h>

__BEGIN_NS_SSRC_WSPR_REGISTRY

using NS_SSRC_SPREAD::MembershipInfo;
using namespace NS_SSRC_WSPR_INDEX;

// TODO: move this into utility package or into ssrcspread
int group_index(const NS_SSRC_SPREAD::GroupList & groups,
                const string & group);

class Registry :
  public IndexService<service_map, IndexScheme,
                      NS_SSRC_WISP_PROTOCOL::GroupMembershipEnable>
{
  friend class NS_SSRC_WISP_SERVICE::ServiceProtocolProcessor<packing_traits>;
  typedef IndexService<service_map, IndexScheme, GroupMembership> super;

  WISP_IMPORT(NS_SSRC_WSPR_REGISTRY, MessageReregister);
  WISP_IMPORT(NS_SSRC_WSPR_REGISTRY, CallReregister);

  bool _is_leader;
  string _leader_name;

  void request_registrations(const string & registry,
                             const bool self_discard = true);

  void process_request(const MessageRegister & msg, const MessageInfo &) {
    register_service(msg.service_name, msg.service_types.begin(),
                     msg.service_types.end());
  }
  void process_request(const MessageReregister & msg, const MessageInfo &) {
    reregister_service(msg.service_name, msg.service_types.begin(),
                       msg.service_types.end());
  }

  void process_request(const MessageUnregister & msg, const MessageInfo &){
    unregister_service(msg.service_name, msg.service_types.begin(),
                       msg.service_types.end());
  }

  void process_request(const MessageQueryByName & msg,
                       const MessageInfo & msginfo)
  {
    if(is_leader() || group_index(msginfo.groups, name()) >= 0)
      process_query<ByName, CallQueryResult>(msg, msginfo);
  }

  void process_request(const MessageQueryByType & msg,
                       const MessageInfo & msginfo)
  {
    if(is_leader() || group_index(msginfo.groups, name()) >= 0)
      process_query<ByType, CallQueryResult>(msg, msginfo);
  }

  void process_request(const MessageQueryAll & msg,
                       const MessageInfo & msginfo)
  {
    if(is_leader() || group_index(msginfo.groups, name()) >= 0) {
      _caller.reply<CallQueryResult>(Message::FIFOSelfDiscard, msginfo.sender(),
                                     msginfo.token(), _index);
    }
  }

  virtual void process_membership_message(const MessageInfo & msginfo,
                                          const MembershipInfo & meminfo);

  virtual void transition(State state);

public:

  explicit Registry(super::caller_type & caller) :
    super(caller), _is_leader(false), _leader_name()
  {
    add_service_type(protocol::service_type_registry());

    WISP_SERVICE_REQUEST(MessageRegister);
    WISP_SERVICE_REQUEST(MessageReregister);
    WISP_SERVICE_REQUEST(MessageUnregister);
    WISP_SERVICE_REQUEST(MessageQueryByName);
    WISP_SERVICE_REQUEST(MessageQueryByType);
    WISP_SERVICE_REQUEST(MessageQueryAll);
  }

  virtual ~Registry() { }

  bool is_leader() {
    return _is_leader;
  }

  string leader_name() {
    return _leader_name;
  }

  void service_join(const string & service_name) {
    if(count<ByName>(service_name) == 0)
      insert<ByName>(RegistryEntry(service_name,
                                   protocol::service_type_unknown()));
  }

  void service_leave(const string & service_name) {
    erase<ByName>(service_name);
  }

  template<typename InputIterator>
  void register_service(const string & service_name,
                        InputIterator service_type,
                        const InputIterator & end_of_range)
  {
    while(service_type != end_of_range) {
      insert<ByName>(RegistryEntry(service_name, *service_type));
      ++service_type;
    }

    // Remove unknown type.
    index_by_composite & index = get_index<ByComposite>();
    index_by_composite::iterator it =
      index.find(boost::make_tuple(service_name,
                                   protocol::service_type_unknown()));

    if(it != index.end())
      index.erase(it);
  }

  template<typename InputIterator>
  void reregister_service(const string & service_name,
                          InputIterator service_type,
                          const InputIterator & end_of_range)
  {
    get_index<ByName>().erase(service_name);
    register_service(service_name, service_type, end_of_range);
  }

  template<typename InputIterator>
  void unregister_service(const string & service_name,
                          InputIterator service_type,
                          const InputIterator & end_of_range)
  {
    index_by_composite & index = get_index<ByComposite>();
    index_by_composite::iterator it;

    while(service_type != end_of_range) {
      it = index.find(boost::make_tuple(service_name, *service_type));
      if(it != index.end())
        index.erase(it);
      ++service_type;
    }

    // Add unknown type entry if every service_name entry has been removed.
    if(count<ByName>(service_name) == 0)
      insert<ByName>(RegistryEntry(service_name,
                                   protocol::service_type_unknown()));
  }

};

__END_NS_SSRC_WSPR_REGISTRY

#endif