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