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