51bd137d295f7e81b69169d9664f3ac50ebdc639
[dbsrgits/DBIx-Class.git] / 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 Moose;
11 use DBIx::Class::Storage::DBI;
12 use DBIx::Class::Storage::DBI::Replicated::Pool;
13 use DBIx::Class::Storage::DBI::Replicated::Balancer;
14 use DBIx::Class::Storage::DBI::Replicated::Types qw/BalancerClassNamePart DBICSchema DBICStorageDBI/;
15 use MooseX::Types::Moose qw/ClassName HashRef Object/;
16 use Scalar::Util 'reftype';
17 use Hash::Merge;
18 use List::Util qw/min max reduce/;
19
20 use namespace::clean -except => 'meta';
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=>'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=>BalancerClassNamePart,
165   coerce=>1,
166   required=>1,
167   default=> 'DBIx::Class::Storage::DBI::Replicated::Balancer::First',
168   handles=>{
169     'create_balancer' => 'new',
170   },
171 );
172
173 =head2 balancer_args
174
175 Contains a hashref of initialized information to pass to the Balancer object.
176 See L<DBIx::Class::Storage::DBI::Replicated::Balancer> for available arguments.
177
178 =cut
179
180 has 'balancer_args' => (
181   is=>'rw',
182   isa=>HashRef,
183   lazy=>1,
184   required=>1,
185   default=>sub { {} },
186 );
187
188 =head2 pool
189
190 Is a <DBIx::Class::Storage::DBI::Replicated::Pool> or derived class.  This is a
191 container class for one or more replicated databases.
192
193 =cut
194
195 has 'pool' => (
196   is=>'ro',
197   isa=>'DBIx::Class::Storage::DBI::Replicated::Pool',
198   lazy_build=>1,
199   handles=>[qw/
200     connect_replicants
201     replicants
202     has_replicants
203   /],
204 );
205
206 =head2 balancer
207
208 Is a <DBIx::Class::Storage::DBI::Replicated::Balancer> or derived class.  This 
209 is a class that takes a pool (<DBIx::Class::Storage::DBI::Replicated::Pool>)
210
211 =cut
212
213 has 'balancer' => (
214   is=>'rw',
215   isa=>'DBIx::Class::Storage::DBI::Replicated::Balancer',
216   lazy_build=>1,
217   handles=>[qw/auto_validate_every/],
218 );
219
220 =head2 master
221
222 The master defines the canonical state for a pool of connected databases.  All
223 the replicants are expected to match this databases state.  Thus, in a classic
224 Master / Slaves distributed system, all the slaves are expected to replicate
225 the Master's state as quick as possible.  This is the only database in the
226 pool of databases that is allowed to handle write traffic.
227
228 =cut
229
230 has 'master' => (
231   is=> 'ro',
232   isa=>DBICStorageDBI,
233   lazy_build=>1,
234 );
235
236 =head1 ATTRIBUTES IMPLEMENTING THE DBIx::Storage::DBI INTERFACE
237
238 The following methods are delegated all the methods required for the 
239 L<DBIx::Class::Storage::DBI> interface.
240
241 =head2 read_handler
242
243 Defines an object that implements the read side of L<BIx::Class::Storage::DBI>.
244
245 =cut
246
247 has 'read_handler' => (
248   is=>'rw',
249   isa=>Object,
250   lazy_build=>1,
251   handles=>[qw/
252     select
253     select_single
254     columns_info_for
255     _dbh_columns_info_for 
256     _select
257   /],
258 );
259
260 =head2 write_handler
261
262 Defines an object that implements the write side of L<BIx::Class::Storage::DBI>,
263 as well as methods that don't write or read that can be called on only one
264 storage, methods that return a C<$dbh>, and any methods that don't make sense to
265 run on a replicant.
266
267 =cut
268
269 has 'write_handler' => (
270   is=>'ro',
271   isa=>Object,
272   lazy_build=>1,
273   handles=>[qw/
274     on_connect_do
275     on_disconnect_do
276     on_connect_call
277     on_disconnect_call
278     connect_info
279     _connect_info
280     throw_exception
281     sql_maker
282     sqlt_type
283     create_ddl_dir
284     deployment_statements
285     datetime_parser
286     datetime_parser_type
287     build_datetime_parser
288     last_insert_id
289     insert
290     insert_bulk
291     update
292     delete
293     dbh
294     txn_begin
295     txn_do
296     txn_commit
297     txn_rollback
298     txn_scope_guard
299     sth
300     deploy
301     with_deferred_fk_checks
302     dbh_do
303     reload_row
304     with_deferred_fk_checks
305     _prep_for_execute
306
307     backup
308     is_datatype_numeric
309     _supports_insert_returning
310     _count_select
311     _subq_count_select
312     _subq_update_delete
313     svp_rollback
314     svp_begin
315     svp_release
316     relname_to_table_alias
317     _straight_join_to_node
318     _dbh_last_insert_id
319     _fix_bind_params
320     _default_dbi_connect_attributes
321     _dbi_connect_info
322     auto_savepoint
323     _sqlt_version_ok
324     _query_end
325     bind_attribute_by_data_type
326     transaction_depth
327     _dbh
328     _select_args
329     _dbh_execute_array
330     _sql_maker_args
331     _sql_maker
332     _query_start
333     _sqlt_version_error
334     _per_row_update_delete
335     _dbh_begin_work
336     _dbh_execute_inserts_with_no_binds
337     _select_args_to_query
338     _svp_generate_name
339     _multipk_update_delete
340     source_bind_attributes
341     _normalize_connect_info
342     _parse_connect_do
343     _dbh_commit
344     _execute_array
345     _placeholders_supported
346     savepoints
347     _sqlt_minimum_version
348     _sql_maker_opts
349     _conn_pid
350     _typeless_placeholders_supported
351     _conn_tid
352     _dbh_autocommit
353     _native_data_type
354     _get_dbh
355     sql_maker_class
356     _dbh_rollback
357     _adjust_select_args_for_complex_prefetch
358     _resolve_ident_sources
359     _resolve_column_info
360     _prune_unused_joins
361     _strip_cond_qualifiers
362     _parse_order_by
363     _resolve_aliastypes_from_select_args
364     _execute
365     _do_query
366     _dbh_sth
367     _dbh_execute
368     _prefetch_insert_auto_nextvals
369     _server_info_hash
370   /],
371 );
372
373 my @unimplemented = qw(
374   _arm_global_destructor
375   _preserve_foreign_dbh
376 );
377
378 for my $method (@unimplemented) {
379   __PACKAGE__->meta->add_method($method, sub {
380     croak "$method must not be called on ".(blessed shift).' objects';
381   });
382 }
383
384 has _master_connect_info_opts =>
385   (is => 'rw', isa => HashRef, default => sub { {} });
386
387 =head2 around: connect_info
388
389 Preserves master's C<connect_info> options (for merging with replicants.)
390 Also sets any Replicated-related options from connect_info, such as
391 C<pool_type>, C<pool_args>, C<balancer_type> and C<balancer_args>.
392
393 =cut
394
395 around connect_info => sub {
396   my ($next, $self, $info, @extra) = @_;
397
398   my $wantarray = wantarray;
399
400   my $merge = Hash::Merge->new('LEFT_PRECEDENT');
401
402   my %opts;
403   for my $arg (@$info) {
404     next unless (reftype($arg)||'') eq 'HASH';
405     %opts = %{ $merge->merge($arg, \%opts) };
406   }
407   delete $opts{dsn};
408
409   if (@opts{qw/pool_type pool_args/}) {
410     $self->pool_type(delete $opts{pool_type})
411       if $opts{pool_type};
412
413     $self->pool_args(
414       $merge->merge((delete $opts{pool_args} || {}), $self->pool_args)
415     );
416
417     $self->pool($self->_build_pool)
418       if $self->pool;
419   }
420
421   if (@opts{qw/balancer_type balancer_args/}) {
422     $self->balancer_type(delete $opts{balancer_type})
423       if $opts{balancer_type};
424
425     $self->balancer_args(
426       $merge->merge((delete $opts{balancer_args} || {}), $self->balancer_args)
427     );
428
429     $self->balancer($self->_build_balancer)
430       if $self->balancer;
431   }
432
433   $self->_master_connect_info_opts(\%opts);
434
435   my (@res, $res);
436   if ($wantarray) {
437     @res = $self->$next($info, @extra);
438   } else {
439     $res = $self->$next($info, @extra);
440   }
441
442   # Make sure master is blessed into the correct class and apply role to it.
443   my $master = $self->master;
444   $master->_determine_driver;
445   Moose::Meta::Class->initialize(ref $master);
446
447   DBIx::Class::Storage::DBI::Replicated::WithDSN->meta->apply($master);
448
449   # link pool back to master
450   $self->pool->master($master);
451
452   $wantarray ? @res : $res;
453 };
454
455 =head1 METHODS
456
457 This class defines the following methods.
458
459 =head2 BUILDARGS
460
461 L<DBIx::Class::Schema> when instantiating its storage passed itself as the
462 first argument.  So we need to massage the arguments a bit so that all the
463 bits get put into the correct places.
464
465 =cut
466
467 sub BUILDARGS {
468   my ($class, $schema, $storage_type_args, @args) = @_;  
469
470   return {
471     schema=>$schema,
472     %$storage_type_args,
473     @args
474   }
475 }
476
477 =head2 _build_master
478
479 Lazy builder for the L</master> attribute.
480
481 =cut
482
483 sub _build_master {
484   my $self = shift @_;
485   my $master = DBIx::Class::Storage::DBI->new($self->schema);
486   $master
487 }
488
489 =head2 _build_pool
490
491 Lazy builder for the L</pool> attribute.
492
493 =cut
494
495 sub _build_pool {
496   my $self = shift @_;
497   $self->create_pool(%{$self->pool_args});
498 }
499
500 =head2 _build_balancer
501
502 Lazy builder for the L</balancer> attribute.  This takes a Pool object so that
503 the balancer knows which pool it's balancing.
504
505 =cut
506
507 sub _build_balancer {
508   my $self = shift @_;
509   $self->create_balancer(
510     pool=>$self->pool,
511     master=>$self->master,
512     %{$self->balancer_args},
513   );
514 }
515
516 =head2 _build_write_handler
517
518 Lazy builder for the L</write_handler> attribute.  The default is to set this to
519 the L</master>.
520
521 =cut
522
523 sub _build_write_handler {
524   return shift->master;
525 }
526
527 =head2 _build_read_handler
528
529 Lazy builder for the L</read_handler> attribute.  The default is to set this to
530 the L</balancer>.
531
532 =cut
533
534 sub _build_read_handler {
535   return shift->balancer;
536 }
537
538 =head2 around: connect_replicants
539
540 All calls to connect_replicants needs to have an existing $schema tacked onto
541 top of the args, since L<DBIx::Storage::DBI> needs it, and any C<connect_info>
542 options merged with the master, with replicant opts having higher priority.
543
544 =cut
545
546 around connect_replicants => sub {
547   my ($next, $self, @args) = @_;
548
549   for my $r (@args) {
550     $r = [ $r ] unless reftype $r eq 'ARRAY';
551
552     $self->throw_exception('coderef replicant connect_info not supported')
553       if ref $r->[0] && reftype $r->[0] eq 'CODE';
554
555 # any connect_info options?
556     my $i = 0;
557     $i++ while $i < @$r && (reftype($r->[$i])||'') ne 'HASH';
558
559 # make one if none
560     $r->[$i] = {} unless $r->[$i];
561
562 # merge if two hashes
563     my @hashes = @$r[$i .. $#{$r}];
564
565     $self->throw_exception('invalid connect_info options')
566       if (grep { reftype($_) eq 'HASH' } @hashes) != @hashes;
567
568     $self->throw_exception('too many hashrefs in connect_info')
569       if @hashes > 2;
570
571     my $merge = Hash::Merge->new('LEFT_PRECEDENT');
572     my %opts = %{ $merge->merge(reverse @hashes) };
573
574 # delete them
575     splice @$r, $i+1, ($#{$r} - $i), ();
576
577 # make sure master/replicants opts don't clash
578     my %master_opts = %{ $self->_master_connect_info_opts };
579     if (exists $opts{dbh_maker}) {
580         delete @master_opts{qw/dsn user password/};
581     }
582     delete $master_opts{dbh_maker};
583
584 # merge with master
585     %opts = %{ $merge->merge(\%opts, \%master_opts) };
586
587 # update
588     $r->[$i] = \%opts;
589   }
590
591   $self->$next($self->schema, @args);
592 };
593
594 =head2 all_storages
595
596 Returns an array of of all the connected storage backends.  The first element
597 in the returned array is the master, and the remainings are each of the
598 replicants.
599
600 =cut
601
602 sub all_storages {
603   my $self = shift @_;
604   return grep {defined $_ && blessed $_} (
605      $self->master,
606      values %{ $self->replicants },
607   );
608 }
609
610 =head2 execute_reliably ($coderef, ?@args)
611
612 Given a coderef, saves the current state of the L</read_handler>, forces it to
613 use reliable storage (e.g. sets it to the master), executes a coderef and then
614 restores the original state.
615
616 Example:
617
618   my $reliably = sub {
619     my $name = shift @_;
620     $schema->resultset('User')->create({name=>$name});
621     my $user_rs = $schema->resultset('User')->find({name=>$name}); 
622     return $user_rs;
623   };
624
625   my $user_rs = $schema->storage->execute_reliably($reliably, 'John');
626
627 Use this when you must be certain of your database state, such as when you just
628 inserted something and need to get a resultset including it, etc.
629
630 =cut
631
632 sub execute_reliably {
633   my ($self, $coderef, @args) = @_;
634
635   unless( ref $coderef eq 'CODE') {
636     $self->throw_exception('Second argument must be a coderef');
637   }
638
639   ##Get copy of master storage
640   my $master = $self->master;
641
642   ##Get whatever the current read hander is
643   my $current = $self->read_handler;
644
645   ##Set the read handler to master
646   $self->read_handler($master);
647
648   ## do whatever the caller needs
649   my @result;
650   my $want_array = wantarray;
651
652   eval {
653     if($want_array) {
654       @result = $coderef->(@args);
655     } elsif(defined $want_array) {
656       ($result[0]) = ($coderef->(@args));
657     } else {
658       $coderef->(@args);
659     }
660   };
661
662   ##Reset to the original state
663   $self->read_handler($current);
664
665   ##Exception testing has to come last, otherwise you might leave the 
666   ##read_handler set to master.
667
668   if($@) {
669     $self->throw_exception("coderef returned an error: $@");
670   } else {
671     return $want_array ? @result : $result[0];
672   }
673 }
674
675 =head2 set_reliable_storage
676
677 Sets the current $schema to be 'reliable', that is all queries, both read and
678 write are sent to the master
679
680 =cut
681
682 sub set_reliable_storage {
683   my $self = shift @_;
684   my $schema = $self->schema;
685   my $write_handler = $self->schema->storage->write_handler;
686
687   $schema->storage->read_handler($write_handler);
688 }
689
690 =head2 set_balanced_storage
691
692 Sets the current $schema to be use the </balancer> for all reads, while all
693 writes are sent to the master only
694
695 =cut
696
697 sub set_balanced_storage {
698   my $self = shift @_;
699   my $schema = $self->schema;
700   my $balanced_handler = $self->schema->storage->balancer;
701
702   $schema->storage->read_handler($balanced_handler);
703 }
704
705 =head2 connected
706
707 Check that the master and at least one of the replicants is connected.
708
709 =cut
710
711 sub connected {
712   my $self = shift @_;
713   return
714     $self->master->connected &&
715     $self->pool->connected_replicants;
716 }
717
718 =head2 ensure_connected
719
720 Make sure all the storages are connected.
721
722 =cut
723
724 sub ensure_connected {
725   my $self = shift @_;
726   foreach my $source ($self->all_storages) {
727     $source->ensure_connected(@_);
728   }
729 }
730
731 =head2 limit_dialect
732
733 Set the limit_dialect for all existing storages
734
735 =cut
736
737 sub limit_dialect {
738   my $self = shift @_;
739   foreach my $source ($self->all_storages) {
740     $source->limit_dialect(@_);
741   }
742   return $self->master->quote_char;
743 }
744
745 =head2 quote_char
746
747 Set the quote_char for all existing storages
748
749 =cut
750
751 sub quote_char {
752   my $self = shift @_;
753   foreach my $source ($self->all_storages) {
754     $source->quote_char(@_);
755   }
756   return $self->master->quote_char;
757 }
758
759 =head2 name_sep
760
761 Set the name_sep for all existing storages
762
763 =cut
764
765 sub name_sep {
766   my $self = shift @_;
767   foreach my $source ($self->all_storages) {
768     $source->name_sep(@_);
769   }
770   return $self->master->name_sep;
771 }
772
773 =head2 set_schema
774
775 Set the schema object for all existing storages
776
777 =cut
778
779 sub set_schema {
780   my $self = shift @_;
781   foreach my $source ($self->all_storages) {
782     $source->set_schema(@_);
783   }
784 }
785
786 =head2 debug
787
788 set a debug flag across all storages
789
790 =cut
791
792 sub debug {
793   my $self = shift @_;
794   if(@_) {
795     foreach my $source ($self->all_storages) {
796       $source->debug(@_);
797     }
798   }
799   return $self->master->debug;
800 }
801
802 =head2 debugobj
803
804 set a debug object
805
806 =cut
807
808 sub debugobj {
809   my $self = shift @_;
810   return $self->master->debugobj(@_);
811 }
812
813 =head2 debugfh
814
815 set a debugfh object
816
817 =cut
818
819 sub debugfh {
820   my $self = shift @_;
821   return $self->master->debugfh(@_);
822 }
823
824 =head2 debugcb
825
826 set a debug callback
827
828 =cut
829
830 sub debugcb {
831   my $self = shift @_;
832   return $self->master->debugcb(@_);
833 }
834
835 =head2 disconnect
836
837 disconnect everything
838
839 =cut
840
841 sub disconnect {
842   my $self = shift @_;
843   foreach my $source ($self->all_storages) {
844     $source->disconnect(@_);
845   }
846 }
847
848 =head2 cursor_class
849
850 set cursor class on all storages, or return master's
851
852 =cut
853
854 sub cursor_class {
855   my ($self, $cursor_class) = @_;
856
857   if ($cursor_class) {
858     $_->cursor_class($cursor_class) for $self->all_storages;
859   }
860   $self->master->cursor_class;
861 }
862
863 =head2 cursor
864
865 set cursor class on all storages, or return master's, alias for L</cursor_class>
866 above.
867
868 =cut
869
870 sub cursor {
871   my ($self, $cursor_class) = @_;
872
873   if ($cursor_class) {
874     $_->cursor($cursor_class) for $self->all_storages;
875   }
876   $self->master->cursor;
877 }
878
879 =head2 unsafe
880
881 sets the L<DBIx::Class::Storage::DBI/unsafe> option on all storages or returns
882 master's current setting
883
884 =cut
885
886 sub unsafe {
887   my $self = shift;
888
889   if (@_) {
890     $_->unsafe(@_) for $self->all_storages;
891   }
892
893   return $self->master->unsafe;
894 }
895
896 =head2 disable_sth_caching
897
898 sets the L<DBIx::Class::Storage::DBI/disable_sth_caching> option on all storages
899 or returns master's current setting
900
901 =cut
902
903 sub disable_sth_caching {
904   my $self = shift;
905
906   if (@_) {
907     $_->disable_sth_caching(@_) for $self->all_storages;
908   }
909
910   return $self->master->disable_sth_caching;
911 }
912
913 =head2 lag_behind_master
914
915 returns the highest Replicant L<DBIx::Class::Storage::DBI/lag_behind_master>
916 setting
917
918 =cut
919
920 sub lag_behind_master {
921   my $self = shift;
922
923   return max map $_->lag_behind_master, $self->replicants;
924
925
926 =head2 is_replicating
927
928 returns true if all replicants return true for
929 L<DBIx::Class::Storage::DBI/is_replicating>
930
931 =cut
932
933 sub is_replicating {
934   my $self = shift;
935
936   return (grep $_->is_replicating, $self->replicants) == ($self->replicants);
937 }
938
939 =head2 connect_call_datetime_setup
940
941 calls L<DBIx::Class::Storage::DBI/connect_call_datetime_setup> for all storages
942
943 =cut
944
945 sub connect_call_datetime_setup {
946   my $self = shift;
947   $_->connect_call_datetime_setup for $self->all_storages;
948 }
949
950 sub _populate_dbh {
951   my $self = shift;
952   $_->_populate_dbh for $self->all_storages;
953 }
954
955 sub _connect {
956   my $self = shift;
957   $_->_connect for $self->all_storages;
958 }
959
960 sub _rebless {
961   my $self = shift;
962   $_->_rebless for $self->all_storages;
963 }
964
965 sub _determine_driver {
966   my $self = shift;
967   $_->_determine_driver for $self->all_storages;
968 }
969
970 sub _driver_determined {
971   my $self = shift;
972   
973   if (@_) {
974     $_->_driver_determined(@_) for $self->all_storages;
975   }
976
977   return $self->master->_driver_determined;
978 }
979
980 sub _init {
981   my $self = shift;
982   
983   $_->_init for $self->all_storages;
984 }
985
986 sub _run_connection_actions {
987   my $self = shift;
988   
989   $_->_run_connection_actions for $self->all_storages;
990 }
991
992 sub _do_connection_actions {
993   my $self = shift;
994   
995   if (@_) {
996     $_->_do_connection_actions(@_) for $self->all_storages;
997   }
998 }
999
1000 sub connect_call_do_sql {
1001   my $self = shift;
1002   $_->connect_call_do_sql(@_) for $self->all_storages;
1003 }
1004
1005 sub disconnect_call_do_sql {
1006   my $self = shift;
1007   $_->disconnect_call_do_sql(@_) for $self->all_storages;
1008 }
1009
1010 sub _seems_connected {
1011   my $self = shift;
1012
1013   return min map $_->_seems_connected, $self->all_storages;
1014 }
1015
1016 sub _ping {
1017   my $self = shift;
1018
1019   return min map $_->_ping, $self->all_storages;
1020 }
1021
1022 my $numify_ver = sub {
1023   my $ver = shift;
1024   my @numparts = split /\D+/, $ver;
1025   my $format = '%d.' . (join '', ('%05d') x (@numparts - 1));
1026
1027   return sprintf $format, @numparts;
1028 };
1029
1030 sub _server_info {
1031   my $self = shift;
1032
1033   if (not $self->_server_info_hash) {
1034     my $min_version_info = (
1035       reduce { $a->[0] < $b->[0] ? $a : $b } 
1036       map [ $numify_ver->($_->{dbms_version}), $_ ],
1037       map $_->_server_info, $self->all_storages
1038     )->[1];
1039
1040     $self->_server_info_hash($min_version_info); # on master
1041   }
1042
1043   return $self->_server_info_hash;
1044 }
1045
1046 sub _get_server_version {
1047   my $self = shift;
1048
1049   return $self->_server_info->{dbms_version};
1050 }
1051
1052 sub _verify_pid {
1053   my $self = shift;
1054
1055   for my $storage ($self->all_storages) {
1056     $storage->_verify_pid;
1057   }
1058 }
1059
1060 sub _verify_tid {
1061   my $self = shift;
1062
1063   for my $storage ($self->all_storages) {
1064     $storage->_verify_tid;
1065   }
1066 }
1067
1068 =head1 GOTCHAS
1069
1070 Due to the fact that replicants can lag behind a master, you must take care to
1071 make sure you use one of the methods to force read queries to a master should
1072 you need realtime data integrity.  For example, if you insert a row, and then
1073 immediately re-read it from the database (say, by doing $row->discard_changes)
1074 or you insert a row and then immediately build a query that expects that row
1075 to be an item, you should force the master to handle reads.  Otherwise, due to
1076 the lag, there is no certainty your data will be in the expected state.
1077
1078 For data integrity, all transactions automatically use the master storage for
1079 all read and write queries.  Using a transaction is the preferred and recommended
1080 method to force the master to handle all read queries.
1081
1082 Otherwise, you can force a single query to use the master with the 'force_pool'
1083 attribute:
1084
1085   my $row = $resultset->search(undef, {force_pool=>'master'})->find($pk);
1086
1087 This attribute will safely be ignore by non replicated storages, so you can use
1088 the same code for both types of systems.
1089
1090 Lastly, you can use the L</execute_reliably> method, which works very much like
1091 a transaction.
1092
1093 For debugging, you can turn replication on/off with the methods L</set_reliable_storage>
1094 and L</set_balanced_storage>, however this operates at a global level and is not
1095 suitable if you have a shared Schema object being used by multiple processes,
1096 such as on a web application server.  You can get around this limitation by
1097 using the Schema clone method.
1098
1099   my $new_schema = $schema->clone;
1100   $new_schema->set_reliable_storage;
1101
1102   ## $new_schema will use only the Master storage for all reads/writes while
1103   ## the $schema object will use replicated storage.
1104
1105 =head1 AUTHOR
1106
1107   John Napiorkowski <john.napiorkowski@takkle.com>
1108
1109 Based on code originated by:
1110
1111   Norbert Csongrádi <bert@cpan.org>
1112   Peter Siklósi <einon@einon.hu>
1113
1114 =head1 LICENSE
1115
1116 You may distribute this code under the same terms as Perl itself.
1117
1118 =cut
1119
1120 __PACKAGE__->meta->make_immutable;
1121
1122 1;