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 { $_->limit_dialect( @_ ) for( shift->all_sources ) }
 135 sub quote_char { $_->quote_char( @_ ) for( shift->all_sources ) }
 136 sub name_sep { $_->quote_char( @_ ) for( shift->all_sources ) }
 137 sub disconnect { $_->disconnect( @_ ) for( shift->all_sources ) }
 138 sub set_schema { $_->set_schema( @_ ) for( shift->all_sources ) }
 139
 140 sub DESTROY {
 141     my $self = shift;
 142
 143     undef $self->{write_source};
 144     undef $self->{read_sources};
 145 }
 146
 147 sub last_insert_id {
 148     shift->write_source->last_insert_id( @_ );
 149 }
 150 sub insert {
 151     shift->write_source->insert( @_ );
 152 }
 153 sub update {
 154     shift->write_source->update( @_ );
 155 }
 156 sub update_all {
 157     shift->write_source->update_all( @_ );
 158 }
 159 sub delete {
 160     shift->write_source->delete( @_ );
 161 }
 162 sub delete_all {
 163     shift->write_source->delete_all( @_ );
 164 }
 165 sub create {
 166     shift->write_source->create( @_ );
 167 }
 168 sub find_or_create {
 169     shift->write_source->find_or_create( @_ );
 170 }
 171 sub update_or_create {
 172     shift->write_source->update_or_create( @_ );
 173 }
 174 sub connected {
 175     shift->write_source->connected( @_ );
 176 }
 177 sub ensure_connected {
 178     shift->write_source->ensure_connected( @_ );
 179 }
 180 sub dbh {
 181     shift->write_source->dbh( @_ );
 182 }
 183 sub txn_begin {
 184     shift->write_source->txn_begin( @_ );
 185 }
 186 sub txn_commit {
 187     shift->write_source->txn_commit( @_ );
 188 }
 189 sub txn_rollback {
 190     shift->write_source->txn_rollback( @_ );
 191 }
 192 sub sth {
 193     shift->write_source->sth( @_ );
 194 }
 195 sub deploy {
 196     shift->write_source->deploy( @_ );
 197 }
 198
 199
 200 sub debugfh { shift->_not_supported( 'debugfh' ) };
 201 sub debugcb { shift->_not_supported( 'debugcb' ) };
 202
 203 sub _not_supported {
 204     my( $self, $method ) = @_;
 205
 206     die "This Storage does not support $method method.";
 207 }
 208
 209 =head1 SEE ALSO
 210
 211 L<DBI::Class::Storage::DBI>, L<DBD::Multi>, L<DBI>
 212
 213 =head1 AUTHOR
 214
 215 Norbert Csongrádi <bert@cpan.org>
 216
 217 Peter Siklósi <einon@einon.hu>
 218
 219 =head1 LICENSE
 220
 221 You may distribute this code under the same terms as Perl itself.
 222
 223 =cut
 224
 225 1;