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