lib/DBIx/Class/Storage/DBI/Replication.pm

   1 package DBIx::Class::Storage::DBI::Replication;
   2
   3 use strict;
   4 use warnings;
   5
   6 use DBIx::Class::Storage::DBI;
   7 use DBD::Multi;
   8 use base qw/Class::Accessor::Fast/;
   9
  10 __PACKAGE__->mk_accessors( qw/read_source write_source/ );
  11
  12 =head1 NAME
  13
  14 DBIx::Class::Storage::DBI::Replication - EXPERIMENTAL Replicated database support
  15
  16 =head1 SYNOPSIS
  17
  18   # change storage_type in your schema class
  19     $schema->storage_type( '::DBI::Replication' );
  20     $schema->connect_info( [
  21                      [ "dbi:mysql:database=test;hostname=master", "username", "password", { AutoCommit => 1 } ], # master
  22                      [ "dbi:mysql:database=test;hostname=slave1", "username", "password", { priority => 10 } ],  # slave1
  23                      [ "dbi:mysql:database=test;hostname=slave2", "username", "password", { priority => 10 } ],  # slave2
  24                      <...>,
  25                      { limit_dialect => 'LimitXY' } # If needed, see below
  26                     ] );
  27
  28 =head1 DESCRIPTION
  29
  30 Warning: This class is marked EXPERIMENTAL. It works for the authors but does
  31 not currently have automated tests so your mileage may vary.
  32
  33 This class implements replicated data store for DBI. Currently you can define
  34 one master and numerous slave database connections. All write-type queries
  35 (INSERT, UPDATE, DELETE and even LAST_INSERT_ID) are routed to master
  36 database, all read-type queries (SELECTs) go to the slave database.
  37
  38 For every slave database you can define a priority value, which controls data
  39 source usage pattern. It uses L<DBD::Multi>, so first the lower priority data
  40 sources used (if they have the same priority, the are used randomized), than
  41 if all low priority data sources fail, higher ones tried in order.
  42
  43 =head1 CONFIGURATION
  44
  45 =head2 Limit dialect
  46
  47 If you use LIMIT in your queries (effectively, if you use
  48 SQL::Abstract::Limit), do not forget to set up limit_dialect (perldoc
  49 SQL::Abstract::Limit) by passing it as an option in the (optional) hash
  50 reference to connect_info.  DBIC can not set it up automatically, since it can
  51 not guess DBD::Multi connection types.
  52
  53 =cut
  54
  55 sub new {
  56     my $proto = shift;
  57     my $class = ref( $proto ) || $proto;
  58     my $self = {};
  59
  60     bless( $self, $class );
  61
  62     $self->write_source( DBIx::Class::Storage::DBI->new );
  63     $self->read_source( DBIx::Class::Storage::DBI->new );
  64
  65     return $self;
  66 }
  67
  68 sub all_sources {
  69     my $self = shift;
  70
  71     my @sources = ($self->read_source, $self->write_source);
  72
  73     return wantarray ? @sources : \@sources;
  74 }
  75
  76 sub connect_info {
  77     my( $self, $source_info ) = @_;
  78
  79     my( $info, $global_options, $options, @dsns );
  80
  81     $info = [ @$source_info ];
  82
  83     $global_options = ref $info->[-1] eq 'HASH' ? pop( @$info ) : {};
  84     if( ref( $options = $info->[0]->[-1] ) eq 'HASH' ) {
  85         # Local options present in dsn, merge them with global options
  86         map { $global_options->{$_} = $options->{$_} } keys %$options;
  87         pop @{$info->[0]};
  88     }
  89
  90     # We need to copy-pass $global_options, since connect_info clears it while
  91     # processing options
  92     $self->write_source->connect_info( [ @{$info->[0]}, { %$global_options } ] );
  93
  94     @dsns = map { ($_->[3]->{priority} || 10) => $_ } @{$info}[1..@$info-1];
  95     $global_options->{dsns} = \@dsns;
  96
  97     $self->read_source->connect_info( [ 'dbi:Multi:', undef, undef, { %$global_options } ] );
  98 }
  99
 100 sub select {
 101     shift->read_source->select( @_ );
 102 }
 103 sub select_single {
 104     shift->read_source->select_single( @_ );
 105 }
 106 sub throw_exception {
 107     shift->read_source->throw_exception( @_ );
 108 }
 109 sub sql_maker {
 110     shift->read_source->sql_maker( @_ );
 111 }
 112 sub columns_info_for {
 113     shift->read_source->columns_info_for( @_ );
 114 }
 115 sub sqlt_type {
 116     shift->read_source->sqlt_type( @_ );
 117 }
 118 sub create_ddl_dir {
 119     shift->read_source->create_ddl_dir( @_ );
 120 }
 121 sub deployment_statements {
 122     shift->read_source->deployment_statements( @_ );
 123 }
 124 sub datetime_parser {
 125     shift->read_source->datetime_parser( @_ );
 126 }
 127 sub datetime_parser_type {
 128     shift->read_source->datetime_parser_type( @_ );
 129 }
 130 sub build_datetime_parser {
 131     shift->read_source->build_datetime_parser( @_ );
 132 }
 133
 134 sub limit_dialect {
 135     my $self = shift;
 136     $self->$_->limit_dialect( @_ ) for( $self->all_sources );
 137 }
 138 sub quote_char {
 139     my $self = shift;
 140     $self->$_->quote_char( @_ ) for( $self->all_sources );
 141 }
 142 sub name_sep {
 143     my $self = shift;
 144     $self->$_->quote_char( @_ ) for( $self->all_sources );
 145 }
 146 sub disconnect {
 147     my $self = shift;
 148     $self->$_->disconnect( @_ ) for( $self->all_sources );
 149 }
 150 sub DESTROY {
 151     my $self = shift;
 152
 153     undef $self->{write_source};
 154     undef $self->{read_sources};
 155 }
 156
 157 sub last_insert_id {
 158     shift->write_source->last_insert_id( @_ );
 159 }
 160 sub insert {
 161     shift->write_source->insert( @_ );
 162 }
 163 sub update {
 164     shift->write_source->update( @_ );
 165 }
 166 sub update_all {
 167     shift->write_source->update_all( @_ );
 168 }
 169 sub delete {
 170     shift->write_source->delete( @_ );
 171 }
 172 sub delete_all {
 173     shift->write_source->delete_all( @_ );
 174 }
 175 sub create {
 176     shift->write_source->create( @_ );
 177 }
 178 sub find_or_create {
 179     shift->write_source->find_or_create( @_ );
 180 }
 181 sub update_or_create {
 182     shift->write_source->update_or_create( @_ );
 183 }
 184 sub connected {
 185     shift->write_source->connected( @_ );
 186 }
 187 sub ensure_connected {
 188     shift->write_source->ensure_connected( @_ );
 189 }
 190 sub dbh {
 191     shift->write_source->dbh( @_ );
 192 }
 193 sub txn_begin {
 194     shift->write_source->txn_begin( @_ );
 195 }
 196 sub txn_commit {
 197     shift->write_source->txn_commit( @_ );
 198 }
 199 sub txn_rollback {
 200     shift->write_source->txn_rollback( @_ );
 201 }
 202 sub sth {
 203     shift->write_source->sth( @_ );
 204 }
 205 sub deploy {
 206     shift->write_source->deploy( @_ );
 207 }
 208
 209
 210 sub debugfh { shift->_not_supported( 'debugfh' ) };
 211 sub debugcb { shift->_not_supported( 'debugcb' ) };
 212
 213 sub _not_supported {
 214     my( $self, $method ) = @_;
 215
 216     die "This Storage does not support $method method.";
 217 }
 218
 219 =head1 SEE ALSO
 220
 221 L<DBI::Class::Storage::DBI>, L<DBD::Multi>, L<DBI>
 222
 223 =head1 AUTHOR
 224
 225 Norbert Csongrádi <bert@cpan.org>
 226
 227 Peter Siklósi <einon@einon.hu>
 228
 229 =head1 LICENSE
 230
 231 You may distribute this code under the same terms as Perl itself.
 232
 233 =cut
 234
 235 1;