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