~drizzle-trunk/drizzle/development

/* -*- mode: c++; c-basic-offset: 2; indent-tabs-mode: nil; -*-

 *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:

 *

 *  Copyright (C) 2008-2009 Sun Microsystems

 *

 *  Authors:

 *

 *    Jay Pipes <joinfu@sun.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

 */

/**

 * @file Server-side utility which is responsible for managing the 

 * communication between the kernel, replicator plugins, and applier plugins.

 *

 * TransactionServices is a bridge between modules and the kernel, and its

 * primary function is to take internal events (for instance the start of a 

 * transaction, the changing of a record, or the rollback of a transaction) 

 * and construct GPB Messages that are passed to the registered replicator and

 * applier plugins.

 *

 * The reason for this functionality is to encapsulate all communication

 * between the kernel and the replicator/applier plugins into GPB Messages.

 * Instead of the plugin having to understand the (often fluidly changing)

 * mechanics of the kernel, all the plugin needs to understand is the message

 * format, and GPB messages provide a nice, clear, and versioned format for 

 * these messages.

 *

 * @see /drizzled/message/transaction.proto

 */

#include "drizzled/server_includes.h"

#include "drizzled/transaction_services.h"

#include "drizzled/plugin/replicator.h"

#include "drizzled/plugin/applier.h"

#include "drizzled/message/transaction.pb.h"

#include "drizzled/message/table.pb.h"

#include "drizzled/gettext.h"

#include "drizzled/session.h"

#include "drizzled/plugin_registry.h"

#include <vector>

drizzled::TransactionServices transaction_services;

void add_replicator(drizzled::plugin::Replicator *repl)

{

  transaction_services.attachReplicator(repl);

}

void remove_replicator(drizzled::plugin::Replicator *repl)

{

  transaction_services.detachReplicator(repl);

}

namespace drizzled

{

void TransactionServices::attachReplicator(drizzled::plugin::Replicator *in_replicator)

{

  replicators.push_back(in_replicator);

}

void TransactionServices::detachReplicator(drizzled::plugin::Replicator *in_replicator)

{

  replicators.erase(std::find(replicators.begin(), replicators.end(), in_replicator));

}

void TransactionServices::attachApplier(drizzled::plugin::Applier *in_applier)

{

  appliers.push_back(in_applier);

}

void TransactionServices::detachApplier(drizzled::plugin::Applier *in_applier)

{

  appliers.erase(std::find(appliers.begin(), appliers.end(), in_applier));

}

void TransactionServices::setCommandTransactionContext(drizzled::message::Command *in_command

                                                     , Session *in_session) const

{

  using namespace drizzled::message;

  TransactionContext *trx= in_command->mutable_transaction_context();

  trx->set_server_id(in_session->getServerId());

  trx->set_transaction_id(in_session->getTransactionId());

}

void TransactionServices::startTransaction(Session *in_session)

{

  using namespace drizzled::message;

  if (replicators.size() == 0 || appliers.size() == 0)

    return;

  Command command;

  command.set_type(Command::START_TRANSACTION);

  command.set_timestamp(in_session->getCurrentTimestamp());

  setCommandTransactionContext(&command, in_session);

  push(&command);

}

void TransactionServices::commitTransaction(Session *in_session)

{

  using namespace drizzled::message;

  if (replicators.size() == 0 || appliers.size() == 0)

    return;

  Command command;

  command.set_type(Command::COMMIT);

  command.set_timestamp(in_session->getCurrentTimestamp());

  setCommandTransactionContext(&command, in_session);

  push(&command);

}

void TransactionServices::rollbackTransaction(Session *in_session)

{

  using namespace drizzled::message;

  if (replicators.size() == 0 || appliers.size() == 0)

    return;

  Command command;

  command.set_type(Command::ROLLBACK);

  command.set_timestamp(in_session->getCurrentTimestamp());

  setCommandTransactionContext(&command, in_session);

  push(&command);

}

void TransactionServices::insertRecord(Session *in_session, Table *in_table)

{

  using namespace drizzled::message;

  if (replicators.size() == 0 || appliers.size() == 0)

    return;

  Command command;

  command.set_type(Command::INSERT);

  command.set_timestamp(in_session->getCurrentTimestamp());

  setCommandTransactionContext(&command, in_session);

  const char *schema_name= in_table->getShare()->db.str;

  const char *table_name= in_table->getShare()->table_name.str;

  command.set_schema(schema_name);

  command.set_table(table_name);

  /* 

   * Now we construct the specialized InsertRecord command inside

   * the Command container...

   */

  InsertRecord *change_record= command.mutable_insert_record();

  Field *current_field;

  Field **table_fields= in_table->field;

  String *string_value= new (in_session->mem_root) String(100); /* 100 initially. field->val_str() is responsible for re-adjusting */

  string_value->set_charset(system_charset_info);

  Table::Field *cur_field;

  while ((current_field= *table_fields++) != NULL) 

  {

    cur_field= change_record->add_insert_field();

    cur_field->set_name(std::string(current_field->field_name));

    cur_field->set_type(Table::Field::VARCHAR); /* @TODO real types! */

    string_value= current_field->val_str(string_value);

    change_record->add_insert_value(std::string(string_value->c_ptr()));

    string_value->free(); /* I wish there was a clear() method... */

  }

  if (string_value)

    delete string_value; /* Is this needed with memroot allocation? */

  push(&command);

}

void TransactionServices::updateRecord(Session *in_session, Table *in_table, const unsigned char *, const unsigned char *)

{

  using namespace drizzled::message;

  if (replicators.size() == 0 || appliers.size() == 0)

    return;

  Command command;

  command.set_type(Command::UPDATE);

  command.set_timestamp(in_session->getCurrentTimestamp());

  setCommandTransactionContext(&command, in_session);

  const char *schema_name= in_table->getShare()->db.str;

  const char *table_name= in_table->getShare()->table_name.str;

  command.set_schema(schema_name);

  command.set_table(table_name);

  /* 

   * Now we construct the specialized UpdateRecord command inside

   * the Command container...

   */

  //UpdateRecord *change_record= command.mutable_update_record();

  push(&command);

}

void TransactionServices::deleteRecord(Session *in_session, Table *in_table)

{

  using namespace drizzled::message;

  if (replicators.size() == 0 || appliers.size() == 0)

    return;

  Command command;

  command.set_type(Command::DELETE);

  command.set_timestamp(in_session->getCurrentTimestamp());

  setCommandTransactionContext(&command, in_session);

  const char *schema_name= in_table->getShare()->db.str;

  const char *table_name= in_table->getShare()->table_name.str;

  command.set_schema(schema_name);

  command.set_table(table_name);

  /* 

   * Now we construct the specialized DeleteRecord command inside

   * the Command container...

   */

  //DeleteRecord *change_record= command.mutable_delete_record();

  push(&command);

}

void TransactionServices::rawStatement(Session *in_session, const char *in_query, size_t in_query_len)

{

  using namespace drizzled::message;

  if (replicators.size() == 0 || appliers.size() == 0)

    return;

  Command command;

  command.set_type(Command::RAW_SQL);

  command.set_timestamp(in_session->getCurrentTimestamp());

  setCommandTransactionContext(&command, in_session);

  std::string query(in_query, in_query_len);

  command.set_sql(query);

  push(&command);

}

void TransactionServices::push(drizzled::message::Command *to_push)

{

  std::vector<drizzled::plugin::Replicator *>::iterator repl_iter= replicators.begin();

  std::vector<drizzled::plugin::Applier *>::iterator appl_start_iter, appl_iter;

  appl_start_iter= appliers.begin();

  drizzled::plugin::Replicator *cur_repl;

  drizzled::plugin::Applier *cur_appl;

  while (repl_iter != replicators.end())

  {

    cur_repl= *repl_iter;

    if (! cur_repl->isActive())

    {

      ++repl_iter;

      continue;

    }

    appl_iter= appl_start_iter;

    while (appl_iter != appliers.end())

    {

      cur_appl= *appl_iter;

      if (! cur_appl->isActive())

      {

        ++appl_iter;

        continue;

      }

      cur_repl->replicate(cur_appl, to_push);

      ++appl_iter;

    }

    ++repl_iter;

  }

}

} /* end namespace drizzled */