~drizzle-trunk/drizzle/development

/* -*- mode: c++; c-basic-offset: 2; indent-tabs-mode: nil; -*-
 *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
 *
 *  Copyright (C) 2008 Mark Atwood
 *
 *  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 <drizzled/server_includes.h>
#include <drizzled/replicator.h>
#include <drizzled/gettext.h>
#include <drizzled/session.h>

int replicator_initializer(st_plugin_int *plugin)
{
  replicator_t *p;

  p= new replicator_t;
  if (p == NULL) return 1;
  memset(p, 0, sizeof(replicator_t));

  plugin->data= (void *)p;

  if (plugin->plugin->init)
  {
    if (plugin->plugin->init((void *)p))
    {
      /* TRANSLATORS: The leading word "replicator" is the name
        of the plugin api, and so should not be translated. */
      errmsg_printf(ERRMSG_LVL_ERROR, _("replicator plugin '%s' init() failed"),
                      plugin->name.str);
      goto err;
    }
  }
  plugin->state= PLUGIN_IS_READY;

  return 0;

 err:
  delete p;
  return 1;
}

int replicator_finalizer(st_plugin_int *plugin)
{
  replicator_t *p= (replicator_t *) plugin->data;

  if (plugin->plugin->deinit)
    {
      if (plugin->plugin->deinit((void *)p))
        {
          /* TRANSLATORS: The leading word "replicator" is the name
             of the plugin api, and so should not be translated. */
          errmsg_printf(ERRMSG_LVL_ERROR, _("replicator plugin '%s' deinit() failed"),
                          plugin->name.str);
        }
    }

  if (p) delete p;

  return 0;
}

/* This gets called by plugin_foreach once for each loaded replicator plugin */
static bool replicator_session_iterate(Session *session, plugin_ref plugin, void *)
{
  replicator_t *repl= plugin_data(plugin, replicator_t *);
  bool error;

  /* call this loaded replicator plugin's replicator_func1 function pointer */
  if (repl && repl->session_init)
  {
    error= repl->session_init(session);
    if (error)
    {
      /* TRANSLATORS: The leading word "replicator" is the name
        of the plugin api, and so should not be translated. */
      errmsg_printf(ERRMSG_LVL_ERROR, _("replicator plugin '%s' session_init() failed"),
                      (char *)plugin_name(plugin));
      return true;
    }
  }

  return false;
}

/*
  This call is called once at the begining of each transaction.
*/
extern handlerton *binlog_hton;
bool replicator_session_init(Session *session)
{
  bool foreach_rv;

  if (session->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN))
    trans_register_ha(session, true, binlog_hton);
  trans_register_ha(session, false, binlog_hton);

  if (session->getReplicationData())
    return false;

  /* 
    call replicator_session_iterate
    once for each loaded replicator plugin
  */
  foreach_rv= plugin_foreach(session, replicator_session_iterate,
                             DRIZZLE_REPLICATOR_PLUGIN, NULL);

  return foreach_rv;
}

/* The plugin_foreach() iterator requires that we
   convert all the parameters of a plugin api entry point
   into just one single void ptr, plus the session.
   So we will take all the additional paramters of replicator_do2,
   and marshall them into a struct of this type, and
   then just pass in a pointer to it.
*/
enum repl_row_exec_t{
  repl_insert,
  repl_update,
  repl_delete
};

typedef struct replicator_row_parms_st
{
  repl_row_exec_t type;
  Table *table;
  const unsigned char *before;
  const unsigned char *after;
} replicator_row_parms_st;


/* This gets called by plugin_foreach once for each loaded replicator plugin */
static bool replicator_do_row_iterate (Session *session, plugin_ref plugin, void *p)
{
  replicator_t *repl= plugin_data(plugin, replicator_t *);
  replicator_row_parms_st *params= (replicator_row_parms_st *) p;

  switch (params->type) {
  case repl_insert:
    if (repl && repl->row_insert)
    {
      if (repl->row_insert(session, params->table))
      {
        /* TRANSLATORS: The leading word "replicator" is the name
          of the plugin api, and so should not be translated. */
        errmsg_printf(ERRMSG_LVL_ERROR, _("replicator plugin '%s' row_insert() failed"),
                        (char *)plugin_name(plugin));

        return true;
      }
    }
    break;
  case repl_update:
    if (repl && repl->row_update)
    {
      if (repl->row_update(session, params->table, params->before, params->after))
      {
        /* TRANSLATORS: The leading word "replicator" is the name
          of the plugin api, and so should not be translated. */
        errmsg_printf(ERRMSG_LVL_ERROR, _("replicator plugin '%s' row_update() failed"),
                        (char *)plugin_name(plugin));

        return true;
      }
    }
    break;
  case repl_delete:
    if (repl && repl->row_delete)
    {
      if (repl->row_delete(session, params->table))
      {
        /* TRANSLATORS: The leading word "replicator" is the name
          of the plugin api, and so should not be translated. */
        errmsg_printf(ERRMSG_LVL_ERROR, _("replicator plugin '%s' row_delete() failed"),
                        (char *)plugin_name(plugin));

        return true;
      }
    }
    break;
  }
  return false;
}

/* This is the replicator_do2 entry point.
   This gets called by the rest of the Drizzle server code */
static bool replicator_do_row (Session *session, replicator_row_parms_st *params)
{
  bool foreach_rv;

  foreach_rv= plugin_foreach(session, replicator_do_row_iterate,
                             DRIZZLE_REPLICATOR_PLUGIN, params);
  return foreach_rv;
}

bool replicator_write_row(Session *session, Table *table)
{
  replicator_row_parms_st param;

  param.type= repl_insert;
  param.table= table;
  param.after= NULL;
  param.before= NULL;

  return replicator_do_row(session, &param);
}

bool replicator_update_row(Session *session, Table *table,
                           const unsigned char *before,
                           const unsigned char *after)
{
  replicator_row_parms_st param;

  param.type= repl_update;
  param.table= table;
  param.after= after;
  param.before= before;

  return replicator_do_row(session, &param);
}

bool replicator_delete_row(Session *session, Table *table)
{
  replicator_row_parms_st param;

  param.type= repl_delete;
  param.table= table;
  param.after= NULL;
  param.before= NULL;

  return replicator_do_row(session, &param);
}

/*
  Here be Dragons!

  Ok, not so much dragons, but this is where we handle either commits or rollbacks of
  statements.
*/
typedef struct replicator_row_end_st
{
  bool autocommit;
  bool commit;
} replicator_row_end_st;

/* We call this to end a statement (on each registered plugin) */
static bool replicator_end_transaction_iterate (Session *session, plugin_ref plugin, void *p)
{
  replicator_t *repl= plugin_data(plugin, replicator_t *);
  replicator_row_end_st *params= (replicator_row_end_st *)p;

  /* call this loaded replicator plugin's replicator_func1 function pointer */
  if (repl && repl->end_transaction)
  {
    if (repl->end_transaction(session, params->autocommit, params->commit))
    {
      /* TRANSLATORS: The leading word "replicator" is the name
        of the plugin api, and so should not be translated. */
      errmsg_printf(ERRMSG_LVL_ERROR, _("replicator plugin '%s' end_transaction() failed"),
                      (char *)plugin_name(plugin));
      return true;
    }
  }

  return false;
}

bool replicator_end_transaction(Session *session, bool autocommit, bool commit)
{
  bool foreach_rv;
  replicator_row_end_st params;

  params.autocommit= autocommit;
  params.commit= commit;

  /* We need to free any data we did an init of for the session */
  foreach_rv= plugin_foreach(session, replicator_end_transaction_iterate,
                             DRIZZLE_REPLICATOR_PLUGIN, (void *) &params);

  return foreach_rv;
}

/*
  If you can do real 2PC this is your hook poing to know that the event is coming.

  Always true for the moment.

*/
bool replicator_prepare(Session *)
{
  return false;
}

/*
  Replicate statement.
*/
typedef struct replicator_statement_st
{
  const char *query;
  size_t query_length;
} replicator_statement_st;

/* We call this to end a statement (on each registered plugin) */
static bool replicator_statement_iterate(Session *session, plugin_ref plugin, void *p)
{
  replicator_t *repl= plugin_data(plugin, replicator_t *);
  replicator_statement_st *params= (replicator_statement_st *)p;

  /* call this loaded replicator plugin's replicator_func1 function pointer */
  if (repl && repl->statement)
  {
    if (repl->statement(session, params->query, params->query_length))
    {
      /* TRANSLATORS: The leading word "replicator" is the name
        of the plugin api, and so should not be translated. */
      errmsg_printf(ERRMSG_LVL_ERROR, _("replicator plugin '%s' statement() failed"),
                      (char *)plugin_name(plugin));
      return true;
    }
  }

  return false;
}

bool replicator_statement(Session *session, const char *query, size_t query_length)
{
  bool foreach_rv;
  replicator_statement_st params;
  
  params.query= query;
  params.query_length= query_length;

  /* We need to free any data we did an init of for the session */
  foreach_rv= plugin_foreach(session, replicator_statement_iterate,
                             DRIZZLE_REPLICATOR_PLUGIN, (void *) &params);

  return foreach_rv;
}