Class::DBI::Replicated - Replication from single master to multiple slaves

  package My::DBI;
  use base qw(Class::DBI::Replicated::mysql);
  # use base qw(Class::DBI::Replicated::Pg::Slony1);

  My::DBI->replication(\%arg);

  $class->replication(\%arg);

  if ($class->repl_check) { ...

  unless ($class->repl_wait(\%arg)) {
    # not up to date
  }

  my $later = $class->repl_compare($my_pos, $master_pos);

package Class::DBI::Replicated;

use warnings;
use strict;
use Class::Trigger;
use Sub::Install qw(install_sub);
use base qw(Class::Accessor::Class Class::Data::Inheritable);
use Params::Validate qw(:all);
use Devel::Peek qw(CvGV);
use Carp qw(croak);

our $VERSION = '0.040';

# we also track the most recent slave db name
sub replication_db {
  my $class = shift;
  return $class->__replication_db unless @_;
  my ($db_name) = @_;
  if ($db_name =~ /^Slave_/) {
    $class->__slave_db($db_name);
  }
  $class->__replication_db($db_name);
}

__PACKAGE__->mk_classdata('__replication_std_triggers');
__PACKAGE__->mk_classdata('__force_master');

sub replication {
  my $class = shift;
  my $arg = { validate_with(
    params => [ %{ +shift } ],
    spec   => {
      master   => { type => ARRAYREF },
      slaves   => { type => ARRAYREF },
      user     => { type => SCALAR, optional => 1,
                    depends => ['password'],
                  },
      password => { type => SCALAR, optional => 1,
                    depends => ['user'],
                  },
      $class->replication_args,
    }
  ) };

  $class->mk_class_accessors(
    '__slave_names',
    '__slave_db',
    '__replication_db',
    '__repl_user',
    '__repl_pass',
    '__replication_setup',
    'repl_pos',
  );
  
  if ($arg->{user}) {
    $class->__repl_user($arg->{user});
    $class->__repl_pass($arg->{password});
  }

  $class->__add_std_triggers;
  $class->replication_setup($arg);

  my @slaves = @{$arg->{slaves}};

  if (!@slaves or @slaves % 2) {
    croak "list of slaves must be name => dsn pairs\n";
  }
  $class->Ima::DBI::set_db('Master' => @{$arg->{master}});
  if ($arg->{user}) {
    $class->Ima::DBI::set_db(
      'Master_Repl',
      $arg->{master}->[0],
      $arg->{user},
      $arg->{password},
    );
  }
  my @names;
  while (my ($name, $dsn) = splice @slaves, 0, 2) {
    push @names, $name;
    my $slave_arg = ref $dsn eq 'HASH' ? $dsn : { dsn => $dsn };
    $class->Ima::DBI::set_db("Slave_$name" => @{ $slave_arg->{dsn} });
    if ($arg->{user}) {
      $class->Ima::DBI::set_db(
        "Slave_$name\_Repl",
        $slave_arg->{dsn}->[0],
        $arg->{user},
        $arg->{password},
      );
    }
    unless ($class->replication_db) {
      $class->replication_db("Slave_$name");
    }
  }
  $class->__slave_names([ @names ]);
}

sub db_Main {
  my ($class) = @_;
  my $db_name = $class->__force_master ?
    'db_Master' : 'db_' . $class->replication_db;
  $class->call_trigger(
    'repl_db',
    $db_name,
  );
  return $class->$db_name;
}

sub db_Slave {
  my ($class) = @_;
  my $db_name = 'db_' . $class->__slave_db;
  return $class->$db_name;
}

sub db_Slave_Repl {
  my ($class) = @_;
  return $class->db_Slave unless $class->__repl_user;
  my $db_name = 'db_' . $class->__slave_db . '_Repl';
  return $class->$db_name;
}

# any auto-generated db_Master_Repl will override this

sub db_Master_Repl {
  my ($class) = @_;
  return $class->db_Master;
}

sub switch_to_master {
  my ($class) = @_;
  return if $class->replication_db eq 'Master';
  $class->replication_db('Master');
  $class->call_trigger('switch_to_master');
}

sub __default_slave {
  my ($class) = @_;
  return $class->__slave_names->[0];
}

sub switch_to_slave {
  my ($class, $name) = @_;
  $name ||= $class->__default_slave;
  $class->replication_db("Slave_$name");
  $class->call_trigger('switch_to_slave', $name);
}

sub wait_for_slave {
  my ($class, $name) = @_;
  my $ok = eval {
    $class->repl_wait({
      slave => $name
    });
  };
  my $err = $@;
  die $err if $err;
  return $ok;
}

sub _mk_unimplemented {
  my ($class, $meth) = @_;
  no strict 'refs';
  *{$class . "::" . $meth} = sub {
    croak "$class does not implement $meth"
  };
}

sub repl_mark {
  my ($class) = @_;
  $class->call_trigger(
    'repl_mark',
  );
  my $pos = $class->repl_get_master;
  $class->repl_pos($pos);
}

sub repl_wait {
  my ($class, $arg) = @_;
  $arg->{timeout} = 30 unless defined $arg->{timeout};
  $arg->{slave}   ||= $class->__default_slave;
  my $tries = 0;
  my $done  = $class->repl_check($arg);
  while ($tries < $arg->{timeout} and not $done) {
    sleep 1;
    $done = $class->repl_check($arg);
  }
  if ($tries >= $arg->{timeout} and not $done) {
    die "$arg->{slave} is not up to date after $arg->{timeout} seconds"
      if $arg->{fatal};
    return 0;
  }
  unless ($class->replication_db eq "Slave_$arg->{slave}") {
    $class->switch_to_slave($arg->{slave});
  }
  return 1;
}  

sub repl_check {
  my ($class, $arg) = @_;
  $arg->{slave} ||= $class->__default_slave;
  return 1 unless defined $class->repl_pos;

  $class->call_trigger(
    'repl_check',
  );

  my $get = sub { $class->repl_get_slave };
  my $slave_pos  = do {
    if ($arg->{slave} eq $class->__slave_db) {
      $get->();
    } else {
      my $old = $class->__slave_db;
      $class->__slave_db("Slave_" . $arg->{slave});
      my $return = $get->();
      $class->__slave_db($old);
      $return;
    }
  };
  if ($slave_pos && $class->repl_compare($slave_pos, $class->repl_pos)) {
    $class->repl_pos(undef);
    return 1;
  }

  return 0;
}

for (qw(
        repl_get_master
        repl_get_slave
        repl_compare
      )) {
  __PACKAGE__->_mk_unimplemented($_);
}

sub _mark {
  shift->repl_mark;
}

sub _check {
  shift->repl_wait({
    timeout => 0,
  });
}

# this exists only because you can't take a reference to a
# method and still let inheritance have a chance, unlike the
# two subs above which actually add functionality
sub __master {
  shift->switch_to_master;
}

sub __add_std_triggers {
  my ($class) = @_;
  return if $class->__replication_std_triggers;
  $class->__real_add_std_triggers;
  $class->__replication_std_triggers(1);
}

sub __real_add_std_triggers {
  my ($class) = @_;
  $class->add_trigger(
    before_create => \&__master,
    before_update => \&__master,
    before_delete => \&__master,
    after_create  => \&_mark,
    after_update  => \&_mark,
    after_delete  => \&_mark,
    select        => \&_check,
  );
}

sub mk_force_masters {
  my $class = shift;
  for my $meth (@_) {
    # XXX this is very disrespectful.
    my $oldcode = $class->can($meth);
    install_sub({
      code => sub {
        my $class = shift;
        my $old = $class->__force_master;
        $class->__force_master(1);
        my $r = $class->$oldcode(@_);
        $class->__force_master($old);
        return $r;
      },
      into => $class,
      as   => $meth,
    });
  }
}

# XXX fix duplication here

sub mk_markers {
  my $class = shift;
  for my $meth (@_) {
    # XXX this is very disrespectful.
    my $oldcode = $class->can($meth);
    install_sub({
      code => sub {
        my $class = shift;
        $class->switch_to_master;
        my $r = $class->$oldcode(@_);
        $class->repl_mark;
        $class->repl_wait({ timeout => 0 });
        return $r;
      },
      into => $class,
      as   => $meth,
    });
  }
}

sub replication_args { () }

sub replication_setup { () }

1; # End of Class::DBI::Replicated