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

   1 package DBIx::Class::Storage::DBI::Replicated;
   2
   3 BEGIN {
   4   use Carp::Clan qw/^DBIx::Class/;
   5   use DBIx::Class;
   6   croak('The following modules are required for Replication ' . DBIx::Class::Optional::Dependencies->req_missing_for ('replicated') )
   7     unless DBIx::Class::Optional::Dependencies->req_ok_for ('replicated');
   8 }
   9
  10 use Moo;
  11 use Role::Tiny ();
  12 use DBIx::Class::Storage::DBI;
  13 use Scalar::Util qw(reftype blessed);
  14 use List::Util qw(min max reduce);
  15 use Try::Tiny;
  16 use Sub::Name 'subname';
  17 use Class::Inspector;
  18 use DBIx::Class::Storage::DBI::Replicated::Types
  19   qw(DBICSchema DBICStorageDBI ClassName HashRef Object
  20   DoesDBICStorageReplicatedBalancer DBICStorageDBIReplicatedPool Defined);
  21
  22 =head1 NAME
  23
  24 DBIx::Class::Storage::DBI::Replicated - BETA Replicated database support
  25
  26 =head1 SYNOPSIS
  27
  28 The Following example shows how to change an existing $schema to a replicated
  29 storage type, add some replicated (read-only) databases, and perform reporting
  30 tasks.
  31
  32 You should set the 'storage_type attribute to a replicated type.  You should
  33 also define your arguments, such as which balancer you want and any arguments
  34 that the Pool object should get.
  35
  36   my $schema = Schema::Class->clone;
  37   $schema->storage_type( ['::DBI::Replicated', {balancer=>'::Random'}] );
  38   $schema->connection(...);
  39
  40 Next, you need to add in the Replicants.  Basically this is an array of
  41 arrayrefs, where each arrayref is database connect information.  Think of these
  42 arguments as what you'd pass to the 'normal' $schema->connect method.
  43
  44   $schema->storage->connect_replicants(
  45     [$dsn1, $user, $pass, \%opts],
  46     [$dsn2, $user, $pass, \%opts],
  47     [$dsn3, $user, $pass, \%opts],
  48   );
  49
  50 Now, just use the $schema as you normally would.  Automatically all reads will
  51 be delegated to the replicants, while writes to the master.
  52
  53   $schema->resultset('Source')->search({name=>'etc'});
  54
  55 You can force a given query to use a particular storage using the search
  56 attribute 'force_pool'.  For example:
  57
  58   my $RS = $schema->resultset('Source')->search(undef, {force_pool=>'master'});
  59
  60 Now $RS will force everything (both reads and writes) to use whatever was setup
  61 as the master storage.  'master' is hardcoded to always point to the Master,
  62 but you can also use any Replicant name.  Please see:
  63 L<DBIx::Class::Storage::DBI::Replicated::Pool> and the replicants attribute for more.
  64
  65 Also see transactions and L</execute_reliably> for alternative ways to
  66 force read traffic to the master.  In general, you should wrap your statements
  67 in a transaction when you are reading and writing to the same tables at the
  68 same time, since your replicants will often lag a bit behind the master.
  69
  70 See L<DBIx::Class::Storage::DBI::Replicated::Instructions> for more help and
  71 walkthroughs.
  72
  73 =head1 DESCRIPTION
  74
  75 Warning: This class is marked BETA.  This has been running a production
  76 website using MySQL native replication as its backend and we have some decent
  77 test coverage but the code hasn't yet been stressed by a variety of databases.
  78 Individual DBs may have quirks we are not aware of.  Please use this in first
  79 development and pass along your experiences/bug fixes.
  80
  81 This class implements replicated data store for DBI. Currently you can define
  82 one master and numerous slave database connections. All write-type queries
  83 (INSERT, UPDATE, DELETE and even LAST_INSERT_ID) are routed to master
  84 database, all read-type queries (SELECTs) go to the slave database.
  85
  86 Basically, any method request that L<DBIx::Class::Storage::DBI> would normally
  87 handle gets delegated to one of the two attributes: L</read_handler> or to
  88 L</write_handler>.  Additionally, some methods need to be distributed
  89 to all existing storages.  This way our storage class is a drop in replacement
  90 for L<DBIx::Class::Storage::DBI>.
  91
  92 Read traffic is spread across the replicants (slaves) occurring to a user
  93 selected algorithm.  The default algorithm is random weighted.
  94
  95 =head1 NOTES
  96
  97 The consistency between master and replicants is database specific.  The Pool
  98 gives you a method to validate its replicants, removing and replacing them
  99 when they fail/pass predefined criteria.  Please make careful use of the ways
 100 to force a query to run against Master when needed.
 101
 102 =head1 REQUIREMENTS
 103
 104 Replicated Storage has additional requirements not currently part of
 105 L<DBIx::Class>. See L<DBIx::Class::Optional::Dependencies> for more details.
 106
 107 =head1 ATTRIBUTES
 108
 109 This class defines the following attributes.
 110
 111 =head2 schema
 112
 113 The underlying L<DBIx::Class::Schema> object this storage is attaching
 114
 115 =cut
 116
 117 has 'schema' => (
 118   is=>'rw',
 119   isa=>DBICSchema,
 120   weak_ref=>1,
 121   required=>1,
 122 );
 123
 124 =head2 pool_type
 125
 126 Contains the classname which will instantiate the L</pool> object.  Defaults
 127 to: L<DBIx::Class::Storage::DBI::Replicated::Pool>.
 128
 129 =cut
 130
 131 has 'pool_type' => (
 132   is=>'rw',
 133   isa=>ClassName,
 134   default=> sub { 'DBIx::Class::Storage::DBI::Replicated::Pool'},
 135   handles=>{
 136     'create_pool' => 'new',
 137   },
 138 );
 139
 140 =head2 pool_args
 141
 142 Contains a hashref of initialized information to pass to the Balancer object.
 143 See L<DBIx::Class::Storage::DBI::Replicated::Pool> for available arguments.
 144
 145 =cut
 146
 147 has 'pool_args' => (
 148   is=>'rw',
 149   isa =>HashRef,
 150   lazy=>1,
 151   default=>sub { {} },
 152 );
 153
 154
 155 =head2 balancer_type
 156
 157 The replication pool requires a balance class to provider the methods for
 158 choose how to spread the query load across each replicant in the pool.
 159
 160 =cut
 161
 162 has 'balancer_type' => (
 163   is=>'rw',
 164   isa=>Defined,
 165   default=>sub { 'DBIx::Class::Storage::DBI::Replicated::Balancer::First' },
 166 );
 167
 168 sub create_balancer {
 169     my ($self, @args) = @_;
 170     my $type = $self->balancer_type;
 171     $type = 'DBIx::Class::Storage::DBI::Replicated::Balancer'.$type
 172       if ($type=~m/^::/);
 173     $self->schema->ensure_class_loaded($type);
 174     return $type->new(@args);
 175 }
 176
 177 =head2 balancer_args
 178
 179 Contains a hashref of initialized information to pass to the Balancer object.
 180 See L<DBIx::Class::Storage::DBI::Replicated::Balancer> for available arguments.
 181
 182 =cut
 183
 184 has 'balancer_args' => (
 185   is=>'rw',
 186   isa =>HashRef,
 187   lazy=>1,
 188   default=>sub { +{} },
 189 );
 190
 191 =head2 pool
 192
 193 Is a L<DBIx::Class::Storage::DBI::Replicated::Pool> or derived class.  This is a
 194 container class for one or more replicated databases.
 195
 196 =cut
 197
 198 has 'pool' => (
 199   is=>'ro',
 200   isa =>DBICStorageDBIReplicatedPool,
 201   lazy=>1,
 202   builder=>'_build_pool',
 203   clearer=>'clear_pool',
 204   handles=>[qw/
 205     connect_replicants
 206     replicants
 207   /],
 208 );
 209
 210 =head2 balancer
 211
 212 Is a L<DBIx::Class::Storage::DBI::Replicated::Balancer> or derived class.  This
 213 is a class that takes a pool (L<DBIx::Class::Storage::DBI::Replicated::Pool>)
 214
 215 =cut
 216
 217 has 'balancer' => (
 218   is=>'rw',
 219   isa => DoesDBICStorageReplicatedBalancer,
 220   lazy=>1,
 221   builder=>'_build_balancer',
 222   handles=>[qw/auto_validate_every/],
 223 );
 224
 225 =head2 master
 226
 227 The master defines the canonical state for a pool of connected databases.  All
 228 the replicants are expected to match this databases state.  Thus, in a classic
 229 Master / Slaves distributed system, all the slaves are expected to replicate
 230 the Master's state as quick as possible.  This is the only database in the
 231 pool of databases that is allowed to handle write traffic.
 232
 233 =cut
 234
 235 has 'master' => (
 236   is=> 'ro',
 237   isa => DBICStorageDBI,
 238   lazy=>1,
 239   builder=>'_build_master',
 240 );
 241
 242 =head1 ATTRIBUTES IMPLEMENTING THE DBIx::Storage::DBI INTERFACE
 243
 244 The following methods are delegated all the methods required for the
 245 L<DBIx::Class::Storage::DBI> interface.
 246
 247 =head2 read_handler
 248
 249 Defines an object that implements the read side of L<BIx::Class::Storage::DBI>.
 250
 251 =cut
 252
 253 has 'read_handler' => (
 254   is=>'rw',
 255   isa=>Object,
 256   lazy=>1,
 257   builder=>'_build_read_handler',
 258   handles=>[qw/
 259     select
 260     select_single
 261     columns_info_for
 262     _dbh_columns_info_for
 263     _select
 264   /],
 265 );
 266
 267 =head2 write_handler
 268
 269 Defines an object that implements the write side of L<BIx::Class::Storage::DBI>,
 270 as well as methods that don't write or read that can be called on only one
 271 storage, methods that return a C<$dbh>, and any methods that don't make sense to
 272 run on a replicant.
 273
 274 =cut
 275
 276 has 'write_handler' => (
 277   is=>'ro',
 278   isa=>Object,
 279   lazy=>1,
 280   builder=>'_build_write_handler',
 281   handles=>[qw/
 282     on_connect_do
 283     on_disconnect_do
 284     on_connect_call
 285     on_disconnect_call
 286     connect_info
 287     _connect_info
 288     throw_exception
 289     sql_maker
 290     sqlt_type
 291     create_ddl_dir
 292     deployment_statements
 293     datetime_parser
 294     datetime_parser_type
 295     build_datetime_parser
 296     last_insert_id
 297     insert
 298     insert_bulk
 299     update
 300     delete
 301     dbh
 302     txn_begin
 303     txn_do
 304     txn_commit
 305     txn_rollback
 306     txn_scope_guard
 307     sth
 308     deploy
 309     with_deferred_fk_checks
 310     dbh_do
 311     reload_row
 312     _prep_for_execute
 313
 314     backup
 315     is_datatype_numeric
 316     _count_select
 317     _subq_update_delete
 318     svp_rollback
 319     svp_begin
 320     svp_release
 321     relname_to_table_alias
 322     _dbh_last_insert_id
 323     _fix_bind_params
 324     _default_dbi_connect_attributes
 325     _dbi_connect_info
 326     _dbic_connect_attributes
 327     auto_savepoint
 328     _sqlt_version_ok
 329     _query_end
 330     bind_attribute_by_data_type
 331     transaction_depth
 332     _dbh
 333     _select_args
 334     _dbh_execute_array
 335     _sql_maker
 336     _query_start
 337     _sqlt_version_error
 338     _per_row_update_delete
 339     _dbh_begin_work
 340     _dbh_execute_inserts_with_no_binds
 341     _select_args_to_query
 342     _svp_generate_name
 343     _multipk_update_delete
 344     source_bind_attributes
 345     _normalize_connect_info
 346     _parse_connect_do
 347     _dbh_commit
 348     _execute_array
 349     savepoints
 350     _sqlt_minimum_version
 351     _sql_maker_opts
 352     _conn_pid
 353     _dbh_autocommit
 354     _native_data_type
 355     _get_dbh
 356     sql_maker_class
 357     _dbh_rollback
 358     _adjust_select_args_for_complex_prefetch
 359     _resolve_ident_sources
 360     _resolve_column_info
 361     _prune_unused_joins
 362     _strip_cond_qualifiers
 363     _resolve_aliastypes_from_select_args
 364     _execute
 365     _do_query
 366     _dbh_sth
 367     _dbh_execute
 368   /],
 369 );
 370
 371 my @unimplemented = qw(
 372   _arm_global_destructor
 373   _preserve_foreign_dbh
 374   _verify_pid
 375   _verify_tid
 376
 377   get_use_dbms_capability
 378   set_use_dbms_capability
 379   get_dbms_capability
 380   set_dbms_capability
 381   _dbh_details
 382
 383   sql_limit_dialect
 384
 385   _inner_join_to_node
 386   _group_over_selection
 387   _extract_order_criteria
 388   _is_lob_type
 389   _max_column_bytesize
 390   _prefetch_autovalues
 391 );
 392
 393 # the capability framework
 394 push @unimplemented, ( grep
 395   { $_ =~ /^ _ (?: use | supports | determine_supports ) _ /x }
 396   @{Class::Inspector->methods('DBIx::Class::Storage::DBI')||[]}
 397 );
 398
 399 for my $method (@unimplemented) {
 400   {
 401     no strict qw/refs/;
 402     *{__PACKAGE__ ."::$method"} = subname $method => sub {
 403       croak "$method must not be called on ".(blessed shift).' objects';
 404     };
 405   }
 406 }
 407
 408 has _master_connect_info_opts => (
 409   is => 'rw',
 410   isa =>HashRef ,
 411   default => sub { +{} },
 412 );
 413
 414 =head2 around: connect_info
 415
 416 Preserves master's C<connect_info> options (for merging with replicants.)
 417 Also sets any Replicated-related options from connect_info, such as
 418 C<pool_type>, C<pool_args>, C<balancer_type> and C<balancer_args>.
 419
 420 =cut
 421
 422 around connect_info => sub {
 423   my ($next, $self, $info, @extra) = @_;
 424
 425   my $merge = Hash::Merge->new('LEFT_PRECEDENT');
 426
 427   my %opts;
 428   for my $arg (@$info) {
 429     next unless (reftype($arg)||'') eq 'HASH';
 430     %opts = %{ $merge->merge($arg, \%opts) };
 431   }
 432   delete $opts{dsn};
 433
 434   if (@opts{qw/pool_type pool_args/}) {
 435     $self->pool_type(delete $opts{pool_type})
 436       if $opts{pool_type};
 437
 438     $self->pool_args(
 439       $merge->merge((delete $opts{pool_args} || {}), $self->pool_args)
 440     );
 441
 442     ## Since we possibly changed the pool_args, we need to clear the current
 443     ## pool object so that next time it is used it will be rebuilt.
 444     $self->clear_pool;
 445   }
 446
 447   if (@opts{qw/balancer_type balancer_args/}) {
 448     $self->balancer_type(delete $opts{balancer_type})
 449       if $opts{balancer_type};
 450
 451     $self->balancer_args(
 452       $merge->merge((delete $opts{balancer_args} || {}), $self->balancer_args)
 453     );
 454
 455     $self->balancer($self->_build_balancer)
 456       if $self->balancer;
 457   }
 458
 459   $self->_master_connect_info_opts(\%opts);
 460
 461   my @res;
 462   if (wantarray) {
 463     @res = $self->$next($info, @extra);
 464   } else {
 465     $res[0] = $self->$next($info, @extra);
 466   }
 467
 468   # Make sure master is blessed into the correct class and apply role to it.
 469   my $master = $self->master;
 470   $master->_determine_driver;
 471
 472   ## Moose::Meta::Class->initialize(ref $master);
 473   Role::Tiny->apply_roles_to_object($master, 'DBIx::Class::Storage::DBI::Replicated::WithDSN');
 474   ## DBIx::Class::Storage::DBI::Replicated::WithDSN->meta->apply($master);
 475
 476   # link pool back to master
 477   $self->pool->master($master);
 478
 479   wantarray ? @res : $res[0];
 480 };
 481
 482 =head1 METHODS
 483
 484 This class defines the following methods.
 485
 486 =head2 new
 487
 488 L<DBIx::Class::Schema> when instantiating its storage passed itself as the
 489 first argument.  So we need to massage the arguments a bit so that all the
 490 bits get put into the correct places.
 491
 492 =cut
 493
 494 around 'new', sub {
 495   my ($orig, $class, $schema, $storage_type_args, @args) = @_;
 496   return $orig->(
 497     $class,
 498     schema => $schema,
 499     %$storage_type_args,
 500     @args,
 501   );
 502 };
 503
 504 =head2 _build_master
 505
 506 Lazy builder for the L</master> attribute.
 507
 508 =cut
 509
 510 sub _build_master {
 511   my $self = shift @_;
 512   my $master = DBIx::Class::Storage::DBI->new($self->schema);
 513   return $master;
 514 }
 515
 516 =head2 _build_pool
 517
 518 Lazy builder for the L</pool> attribute.
 519
 520 =cut
 521
 522 sub _build_pool {
 523   my $self = shift @_;
 524   $self->create_pool(%{$self->pool_args});
 525 }
 526
 527 =head2 _build_balancer
 528
 529 Lazy builder for the L</balancer> attribute.  This takes a Pool object so that
 530 the balancer knows which pool it's balancing.
 531
 532 =cut
 533
 534 sub _build_balancer {
 535   my $self = shift @_;
 536   $self->create_balancer(
 537     pool=>$self->pool,
 538     master=>$self->master,
 539     %{$self->balancer_args},
 540   );
 541 }
 542
 543 =head2 _build_write_handler
 544
 545 Lazy builder for the L</write_handler> attribute.  The default is to set this to
 546 the L</master>.
 547
 548 =cut
 549
 550 sub _build_write_handler {
 551   return shift->master;
 552 }
 553
 554 =head2 _build_read_handler
 555
 556 Lazy builder for the L</read_handler> attribute.  The default is to set this to
 557 the L</balancer>.
 558
 559 =cut
 560
 561 sub _build_read_handler {
 562   return shift->balancer;
 563 }
 564
 565 =head2 around: connect_replicants
 566
 567 All calls to connect_replicants needs to have an existing $schema tacked onto
 568 top of the args, since L<DBIx::Storage::DBI> needs it, and any C<connect_info>
 569 options merged with the master, with replicant opts having higher priority.
 570
 571 =cut
 572
 573 around connect_replicants => sub {
 574   my ($next, $self, @args) = @_;
 575
 576   for my $r (@args) {
 577     $r = [ $r ] unless reftype $r eq 'ARRAY';
 578
 579     $self->throw_exception('coderef replicant connect_info not supported')
 580       if ref $r->[0] && reftype $r->[0] eq 'CODE';
 581
 582 # any connect_info options?
 583     my $i = 0;
 584     $i++ while $i < @$r && (reftype($r->[$i])||'') ne 'HASH';
 585
 586 # make one if none
 587     $r->[$i] = {} unless $r->[$i];
 588
 589 # merge if two hashes
 590     my @hashes = @$r[$i .. $#{$r}];
 591
 592     $self->throw_exception('invalid connect_info options')
 593       if (grep { reftype($_) eq 'HASH' } @hashes) != @hashes;
 594
 595     $self->throw_exception('too many hashrefs in connect_info')
 596       if @hashes > 2;
 597
 598     my $merge = Hash::Merge->new('LEFT_PRECEDENT');
 599     my %opts = %{ $merge->merge(reverse @hashes) };
 600
 601 # delete them
 602     splice @$r, $i+1, ($#{$r} - $i), ();
 603
 604 # make sure master/replicants opts don't clash
 605     my %master_opts = %{ $self->_master_connect_info_opts };
 606     if (exists $opts{dbh_maker}) {
 607         delete @master_opts{qw/dsn user password/};
 608     }
 609     delete $master_opts{dbh_maker};
 610
 611 # merge with master
 612     %opts = %{ $merge->merge(\%opts, \%master_opts) };
 613
 614 # update
 615     $r->[$i] = \%opts;
 616   }
 617
 618   $self->$next($self->schema, @args);
 619 };
 620
 621 =head2 all_storages
 622
 623 Returns an array of of all the connected storage backends.  The first element
 624 in the returned array is the master, and the remainings are each of the
 625 replicants.
 626
 627 =cut
 628
 629 sub all_storages {
 630   my $self = shift @_;
 631   return grep {defined $_ && blessed $_} (
 632      $self->master,
 633      values %{ $self->replicants },
 634   );
 635 }
 636
 637 =head2 execute_reliably ($coderef, ?@args)
 638
 639 Given a coderef, saves the current state of the L</read_handler>, forces it to
 640 use reliable storage (e.g. sets it to the master), executes a coderef and then
 641 restores the original state.
 642
 643 Example:
 644
 645   my $reliably = sub {
 646     my $name = shift @_;
 647     $schema->resultset('User')->create({name=>$name});
 648     my $user_rs = $schema->resultset('User')->find({name=>$name});
 649     return $user_rs;
 650   };
 651
 652   my $user_rs = $schema->storage->execute_reliably($reliably, 'John');
 653
 654 Use this when you must be certain of your database state, such as when you just
 655 inserted something and need to get a resultset including it, etc.
 656
 657 =cut
 658
 659 sub execute_reliably {
 660   my ($self, $coderef, @args) = @_;
 661
 662   unless( ref $coderef eq 'CODE') {
 663     $self->throw_exception('Second argument must be a coderef');
 664   }
 665
 666   ##Get copy of master storage
 667   my $master = $self->master;
 668
 669   ##Get whatever the current read hander is
 670   my $current = $self->read_handler;
 671
 672   ##Set the read handler to master
 673   $self->read_handler($master);
 674
 675   ## do whatever the caller needs
 676   my @result;
 677   my $want_array = wantarray;
 678
 679   try {
 680     if($want_array) {
 681       @result = $coderef->(@args);
 682     } elsif(defined $want_array) {
 683       ($result[0]) = ($coderef->(@args));
 684     } else {
 685       $coderef->(@args);
 686     }
 687   } catch {
 688     $self->throw_exception("coderef returned an error: $_");
 689   } finally {
 690     ##Reset to the original state
 691     $self->read_handler($current);
 692   };
 693
 694   return wantarray ? @result : $result[0];
 695 }
 696
 697 =head2 set_reliable_storage
 698
 699 Sets the current $schema to be 'reliable', that is all queries, both read and
 700 write are sent to the master
 701
 702 =cut
 703
 704 sub set_reliable_storage {
 705   my $self = shift @_;
 706   my $schema = $self->schema;
 707   my $write_handler = $self->schema->storage->write_handler;
 708
 709   $schema->storage->read_handler($write_handler);
 710 }
 711
 712 =head2 set_balanced_storage
 713
 714 Sets the current $schema to be use the </balancer> for all reads, while all
 715 writes are sent to the master only
 716
 717 =cut
 718
 719 sub set_balanced_storage {
 720   my $self = shift @_;
 721   my $schema = $self->schema;
 722   my $balanced_handler = $self->schema->storage->balancer;
 723
 724   $schema->storage->read_handler($balanced_handler);
 725 }
 726
 727 =head2 connected
 728
 729 Check that the master and at least one of the replicants is connected.
 730
 731 =cut
 732
 733 sub connected {
 734   my $self = shift @_;
 735   return
 736     $self->master->connected &&
 737     $self->pool->connected_replicants;
 738 }
 739
 740 =head2 ensure_connected
 741
 742 Make sure all the storages are connected.
 743
 744 =cut
 745
 746 sub ensure_connected {
 747   my $self = shift @_;
 748   foreach my $source ($self->all_storages) {
 749     $source->ensure_connected(@_);
 750   }
 751 }
 752
 753 =head2 limit_dialect
 754
 755 Set the limit_dialect for all existing storages
 756
 757 =cut
 758
 759 sub limit_dialect {
 760   my $self = shift @_;
 761   foreach my $source ($self->all_storages) {
 762     $source->limit_dialect(@_);
 763   }
 764   return $self->master->limit_dialect;
 765 }
 766
 767 =head2 quote_char
 768
 769 Set the quote_char for all existing storages
 770
 771 =cut
 772
 773 sub quote_char {
 774   my $self = shift @_;
 775   foreach my $source ($self->all_storages) {
 776     $source->quote_char(@_);
 777   }
 778   return $self->master->quote_char;
 779 }
 780
 781 =head2 name_sep
 782
 783 Set the name_sep for all existing storages
 784
 785 =cut
 786
 787 sub name_sep {
 788   my $self = shift @_;
 789   foreach my $source ($self->all_storages) {
 790     $source->name_sep(@_);
 791   }
 792   return $self->master->name_sep;
 793 }
 794
 795 =head2 set_schema
 796
 797 Set the schema object for all existing storages
 798
 799 =cut
 800
 801 sub set_schema {
 802   my $self = shift @_;
 803   foreach my $source ($self->all_storages) {
 804     $source->set_schema(@_);
 805   }
 806 }
 807
 808 =head2 debug
 809
 810 set a debug flag across all storages
 811
 812 =cut
 813
 814 sub debug {
 815   my $self = shift @_;
 816   if(@_) {
 817     foreach my $source ($self->all_storages) {
 818       $source->debug(@_);
 819     }
 820   }
 821   return $self->master->debug;
 822 }
 823
 824 =head2 debugobj
 825
 826 set a debug object
 827
 828 =cut
 829
 830 sub debugobj {
 831   my $self = shift @_;
 832   return $self->master->debugobj(@_);
 833 }
 834
 835 =head2 debugfh
 836
 837 set a debugfh object
 838
 839 =cut
 840
 841 sub debugfh {
 842   my $self = shift @_;
 843   return $self->master->debugfh(@_);
 844 }
 845
 846 =head2 debugcb
 847
 848 set a debug callback
 849
 850 =cut
 851
 852 sub debugcb {
 853   my $self = shift @_;
 854   return $self->master->debugcb(@_);
 855 }
 856
 857 =head2 disconnect
 858
 859 disconnect everything
 860
 861 =cut
 862
 863 sub disconnect {
 864   my $self = shift @_;
 865   foreach my $source ($self->all_storages) {
 866     $source->disconnect(@_);
 867   }
 868 }
 869
 870 =head2 cursor_class
 871
 872 set cursor class on all storages, or return master's
 873
 874 =cut
 875
 876 sub cursor_class {
 877   my ($self, $cursor_class) = @_;
 878
 879   if ($cursor_class) {
 880     $_->cursor_class($cursor_class) for $self->all_storages;
 881   }
 882   $self->master->cursor_class;
 883 }
 884
 885 =head2 cursor
 886
 887 set cursor class on all storages, or return master's, alias for L</cursor_class>
 888 above.
 889
 890 =cut
 891
 892 sub cursor {
 893   my ($self, $cursor_class) = @_;
 894
 895   if ($cursor_class) {
 896     $_->cursor($cursor_class) for $self->all_storages;
 897   }
 898   $self->master->cursor;
 899 }
 900
 901 =head2 unsafe
 902
 903 sets the L<DBIx::Class::Storage::DBI/unsafe> option on all storages or returns
 904 master's current setting
 905
 906 =cut
 907
 908 sub unsafe {
 909   my $self = shift;
 910
 911   if (@_) {
 912     $_->unsafe(@_) for $self->all_storages;
 913   }
 914
 915   return $self->master->unsafe;
 916 }
 917
 918 =head2 disable_sth_caching
 919
 920 sets the L<DBIx::Class::Storage::DBI/disable_sth_caching> option on all storages
 921 or returns master's current setting
 922
 923 =cut
 924
 925 sub disable_sth_caching {
 926   my $self = shift;
 927
 928   if (@_) {
 929     $_->disable_sth_caching(@_) for $self->all_storages;
 930   }
 931
 932   return $self->master->disable_sth_caching;
 933 }
 934
 935 =head2 lag_behind_master
 936
 937 returns the highest Replicant L<DBIx::Class::Storage::DBI/lag_behind_master>
 938 setting
 939
 940 =cut
 941
 942 sub lag_behind_master {
 943   my $self = shift;
 944
 945   return max map $_->lag_behind_master, $self->replicants;
 946 }
 947
 948 =head2 is_replicating
 949
 950 returns true if all replicants return true for
 951 L<DBIx::Class::Storage::DBI/is_replicating>
 952
 953 =cut
 954
 955 sub is_replicating {
 956   my $self = shift;
 957
 958   return (grep $_->is_replicating, $self->replicants) == ($self->replicants);
 959 }
 960
 961 =head2 connect_call_datetime_setup
 962
 963 calls L<DBIx::Class::Storage::DBI/connect_call_datetime_setup> for all storages
 964
 965 =cut
 966
 967 sub connect_call_datetime_setup {
 968   my $self = shift;
 969   $_->connect_call_datetime_setup for $self->all_storages;
 970 }
 971
 972 sub _populate_dbh {
 973   my $self = shift;
 974   $_->_populate_dbh for $self->all_storages;
 975 }
 976
 977 sub _connect {
 978   my $self = shift;
 979   $_->_connect for $self->all_storages;
 980 }
 981
 982 sub _rebless {
 983   my $self = shift;
 984   $_->_rebless for $self->all_storages;
 985 }
 986
 987 sub _determine_driver {
 988   my $self = shift;
 989   $_->_determine_driver for $self->all_storages;
 990 }
 991
 992 sub _driver_determined {
 993   my $self = shift;
 994
 995   if (@_) {
 996     $_->_driver_determined(@_) for $self->all_storages;
 997   }
 998
 999   return $self->master->_driver_determined;
1000 }
1001
1002 sub _init {
1003   my $self = shift;
1004
1005   $_->_init for $self->all_storages;
1006 }
1007
1008 sub _run_connection_actions {
1009   my $self = shift;
1010
1011   $_->_run_connection_actions for $self->all_storages;
1012 }
1013
1014 sub _do_connection_actions {
1015   my $self = shift;
1016
1017   if (@_) {
1018     $_->_do_connection_actions(@_) for $self->all_storages;
1019   }
1020 }
1021
1022 sub connect_call_do_sql {
1023   my $self = shift;
1024   $_->connect_call_do_sql(@_) for $self->all_storages;
1025 }
1026
1027 sub disconnect_call_do_sql {
1028   my $self = shift;
1029   $_->disconnect_call_do_sql(@_) for $self->all_storages;
1030 }
1031
1032 sub _seems_connected {
1033   my $self = shift;
1034
1035   return min map $_->_seems_connected, $self->all_storages;
1036 }
1037
1038 sub _ping {
1039   my $self = shift;
1040
1041   return min map $_->_ping, $self->all_storages;
1042 }
1043
1044 # not using the normalized_version, because we want to preserve
1045 # version numbers much longer than the conventional xxx.yyyzzz
1046 my $numify_ver = sub {
1047   my $ver = shift;
1048   my @numparts = split /\D+/, $ver;
1049   my $format = '%d.' . (join '', ('%06d') x (@numparts - 1));
1050
1051   return sprintf $format, @numparts;
1052 };
1053 sub _server_info {
1054   my $self = shift;
1055
1056   if (not $self->_dbh_details->{info}) {
1057     $self->_dbh_details->{info} = (
1058       reduce { $a->[0] < $b->[0] ? $a : $b }
1059       map [ $numify_ver->($_->{dbms_version}), $_ ],
1060       map $_->_server_info, $self->all_storages
1061     )->[1];
1062   }
1063
1064   return $self->next::method;
1065 }
1066
1067 sub _get_server_version {
1068   my $self = shift;
1069
1070   return $self->_server_info->{dbms_version};
1071 }
1072
1073 =head1 GOTCHAS
1074
1075 Due to the fact that replicants can lag behind a master, you must take care to
1076 make sure you use one of the methods to force read queries to a master should
1077 you need realtime data integrity.  For example, if you insert a row, and then
1078 immediately re-read it from the database (say, by doing $row->discard_changes)
1079 or you insert a row and then immediately build a query that expects that row
1080 to be an item, you should force the master to handle reads.  Otherwise, due to
1081 the lag, there is no certainty your data will be in the expected state.
1082
1083 For data integrity, all transactions automatically use the master storage for
1084 all read and write queries.  Using a transaction is the preferred and recommended
1085 method to force the master to handle all read queries.
1086
1087 Otherwise, you can force a single query to use the master with the 'force_pool'
1088 attribute:
1089
1090   my $row = $resultset->search(undef, {force_pool=>'master'})->find($pk);
1091
1092 This attribute will safely be ignore by non replicated storages, so you can use
1093 the same code for both types of systems.
1094
1095 Lastly, you can use the L</execute_reliably> method, which works very much like
1096 a transaction.
1097
1098 For debugging, you can turn replication on/off with the methods L</set_reliable_storage>
1099 and L</set_balanced_storage>, however this operates at a global level and is not
1100 suitable if you have a shared Schema object being used by multiple processes,
1101 such as on a web application server.  You can get around this limitation by
1102 using the Schema clone method.
1103
1104   my $new_schema = $schema->clone;
1105   $new_schema->set_reliable_storage;
1106
1107   ## $new_schema will use only the Master storage for all reads/writes while
1108   ## the $schema object will use replicated storage.
1109
1110 =head1 AUTHOR
1111
1112   John Napiorkowski <jjnapiork@cpan.org>
1113
1114 Based on code originated by:
1115
1116   Norbert Csongrádi <bert@cpan.org>
1117   Peter Siklósi <einon@einon.hu>
1118
1119 =head1 LICENSE
1120
1121 You may distribute this code under the same terms as Perl itself.
1122
1123 =cut
1124
1125 1;