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