registry/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 <set>

#include <ssrc/wispers/registry/service.h>

__BEGIN_NS_SSRC_WSPR_REGISTRY

using NS_SSRC_WSPR_SERVICE::ServiceProtocol;

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

  while(index-- > 0) {
    if(groups[index] == group)
      return index;
  }

  return index;
}

void Registry::request_registrations(const string & registry,
                                     const bool self_discard)
{
  Message::Service service_type =
    (self_discard ? Message::FIFO : Message::FIFOSelfDiscard);

  _caller.send<ServiceProtocol::CallReregister>
    (service_type, ServiceProtocol::service_group(), registry);
}

void Registry::process_membership_message(const MessageInfo & msginfo,
                                          const MembershipInfo & meminfo)
{
  string sender(msginfo.sender());

  if(sender == ServiceProtocol::service_group()) {
    if(meminfo.caused_by_join()) {
      string changed(meminfo.changed_member());

      if(name() == changed) {
        // TODO: Move to separate method.
        // Register all existing services (includes us) as unknown.
        const unsigned int count_groups = _caller.mbox().count_groups();
        for(unsigned int i = 0; i < count_groups; ++i)
          service_join(msginfo.groups[i]);

        request_registrations(name());
      } else
        service_join(changed);
    } else if(meminfo.caused_by_leave() || meminfo.caused_by_disconnect())
      service_leave(meminfo.changed_member());
    else if(meminfo.caused_by_network()) {
      // Remove all service entries that aren't in new group.
      typedef boost::multi_index_container<string> group_set;
      group_set group;
      int i = _caller.mbox().count_groups();

      while(i-- > 0)
        group.insert(msginfo.groups[i]);

      group_set::iterator group_end(group.end());
      index_by_name::iterator it(begin<ByName>());
      index_by_name::iterator end(end<ByName>());

      while(it != end) {
        if(group.find(it->service_name) == group_end)
          get_index<ByName>().erase(it);
        ++it;
      }
    }
  } else if(sender == service_group_registry()) {
    if(meminfo.is_regular_membership()) {
      if(_caller.mbox().count_groups() > 0)
        _leader_name = msginfo.groups[0];
      else
        _leader_name.clear();

      _is_leader = (_leader_name == name());

      // Request reregistrations to go to registry.
      if(meminfo.caused_by_network() && _is_leader)
        request_registrations(service_group_registry(), false);
    }
  }
}

void Registry::transition(State state) {
  switch(state) {
  case Starting:
    _caller.join(service_group_registry());
    state = Started;
    break;
  case Stopping:
    _caller.leave(service_group_registry());
    break;
  default:
    break;
  }

  // Transition at end so we've joined registry group before
  // super::transition registers us.
  super::transition(state);
}

__END_NS_SSRC_WSPR_REGISTRY