replication_services.cc

/* -*- mode: c++; c-basic-offset: 2; indent-tabs-mode: nil; -*-
 *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
 *
 *  Copyright (C) 2008-2009 Sun Microsystems, Inc.
 *  Copyright (C) 2009-2010 Jay Pipes <jaypipes@gmail.com>
 *
 *  Authors:
 *
 *    Jay Pipes <jaypipes@gmail.com>
 *
 *  This program is free software; you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation; version 2 of the License.
 *
 *  This program is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with this program; if not, write to the Free Software
 *  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
 */

#include <config.h>
#include <drizzled/replication_services.h>
#include <drizzled/plugin/transaction_replicator.h>
#include <drizzled/plugin/transaction_applier.h>
#include <drizzled/message/transaction.pb.h>
#include <drizzled/gettext.h>
#include <drizzled/session.h>
#include <drizzled/error.h>
#include <drizzled/errmsg_print.h>

#include <string>
#include <vector>
#include <algorithm>

using namespace std;

namespace drizzled {

typedef std::vector<plugin::TransactionReplicator*> Replicators;
typedef std::vector<std::pair<std::string, plugin::TransactionApplier*> > Appliers;

static bool is_active= false;
static atomic<uint64_t> last_applied_timestamp;
static Replicators replicators;
static Appliers appliers;
static ReplicationServices::ReplicationStreams replication_streams;

static void normalizeReplicatorName(string &name)
{
  boost::to_lower(name);
  if (name.find("replicator") == string::npos)
    name.append("replicator");
  {
    size_t found_underscore= name.find('_');
    while (found_underscore != string::npos)
    {
      name.erase(found_underscore, 1);
      found_underscore= name.find('_');
    }
  }
}

bool ReplicationServices::evaluateRegisteredPlugins()
{
  /* 
   * We loop through appliers that have registered with us
   * and attempts to pair the applier with its requested
   * replicator.  If an applier has requested a replicator
   * that has either not been built or has not registered
   * with the replication services, we print an error and
   * return false
   */
  if (appliers.empty())
    return true;

  if (replicators.empty() && not appliers.empty())
  {
    errmsg_printf(error::ERROR,
                  N_("You registered a TransactionApplier plugin but no "
                     "TransactionReplicator plugins were registered.\n"));
    return false;
  }

  BOOST_FOREACH(Appliers::reference appl_iter, appliers)
  {
    plugin::TransactionApplier *applier= appl_iter.second;
    string requested_replicator_name= appl_iter.first;
    normalizeReplicatorName(requested_replicator_name);

    bool found= false;
    Replicators::iterator repl_iter;
    for (repl_iter= replicators.begin(); repl_iter != replicators.end(); ++repl_iter)
    {
      string replicator_name= (*repl_iter)->getName();
      normalizeReplicatorName(replicator_name);

      if (requested_replicator_name.compare(replicator_name) == 0)
      {
        found= true;
        break;
      }
    }
    if (not found)
    {
      errmsg_printf(error::ERROR,
                    N_("You registered a TransactionApplier plugin but no "
                       "TransactionReplicator plugins were registered that match the "
                       "requested replicator name of '%s'.\n"
                       "We have deactivated the TransactionApplier '%s'.\n"),
                       requested_replicator_name.c_str(),
                       applier->getName().c_str());
      applier->deactivate();
      return false;
    }
    else
    {
      replication_streams.push_back(make_pair(*repl_iter, applier));
    }
  }
  is_active= true;
  return true;
}

void ReplicationServices::attachReplicator(plugin::TransactionReplicator *in_replicator)
{
  replicators.push_back(in_replicator);
}

void ReplicationServices::detachReplicator(plugin::TransactionReplicator *in_replicator)
{
  replicators.erase(std::find(replicators.begin(), replicators.end(), in_replicator));
}

void ReplicationServices::attachApplier(plugin::TransactionApplier *in_applier, const string &requested_replicator_name)
{
  appliers.push_back(make_pair(requested_replicator_name, in_applier));
}

void ReplicationServices::detachApplier(plugin::TransactionApplier *)
{
}

bool ReplicationServices::isActive()
{
  return is_active;
}

plugin::ReplicationReturnCode ReplicationServices::pushTransactionMessage(Session &in_session,
                                                                          message::Transaction &to_push)
{
  plugin::ReplicationReturnCode result= plugin::SUCCESS;

  BOOST_FOREACH(ReplicationStreams::reference iter, replication_streams)
  {
    plugin::TransactionReplicator *cur_repl= iter.first;
    plugin::TransactionApplier *cur_appl= iter.second;

    result= cur_repl->replicate(cur_appl, in_session, to_push);

    if (result == plugin::SUCCESS)
    {
      /* 
       * We update the timestamp for the last applied Transaction so that
       * publisher plugins can ask the replication services when the
       * last known applied Transaction was using the getLastAppliedTimestamp()
       * method.
       */
      last_applied_timestamp.fetch_and_store(to_push.transaction_context().end_timestamp());
    }
    else
      return result;
  }
  return result;
}

ReplicationServices::ReplicationStreams &ReplicationServices::getReplicationStreams()
{
  return replication_streams;
}

} /* namespace drizzled */