b3247abff02dceb070b95f2233898965ae9ef583
[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_type => '::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     update
269     delete
270     dbh
271     txn_begin
272     txn_do
273     txn_commit
274     txn_rollback
275     txn_scope_guard
276     _exec_txn_rollback
277     _exec_txn_begin
278     _exec_txn_commit
279     deploy
280     with_deferred_fk_checks
281     dbh_do
282     _prep_for_execute
283     is_datatype_numeric
284     _count_select
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     _dbh_execute_inserts_with_no_binds
305     _select_args_to_query
306     _gen_sql_bind
307     _svp_generate_name
308     _normalize_connect_info
309     _parse_connect_do
310     savepoints
311     _sql_maker_opts
312     _conn_pid
313     _dbh_autocommit
314     _native_data_type
315     _get_dbh
316     sql_maker_class
317     insert_bulk
318     _insert_bulk
319     _execute
320     _do_query
321     _dbh_execute
322   /, Class::MOP::Class->initialize('DBIx::Class::Storage::DBIHacks')->get_method_list ],
323   reader => [qw/
324     select
325     select_single
326     columns_info_for
327     _dbh_columns_info_for
328     _select
329   /],
330   unimplemented => [qw/
331     _arm_global_destructor
332     _verify_pid
333
334     get_use_dbms_capability
335     set_use_dbms_capability
336     get_dbms_capability
337     set_dbms_capability
338     _dbh_details
339     _dbh_get_info
340
341     _determine_connector_driver
342     _extract_driver_from_connect_info
343     _describe_connection
344     _warn_undetermined_driver
345
346     sql_limit_dialect
347     sql_quote_char
348     sql_name_sep
349
350     _prefetch_autovalues
351     _perform_autoinc_retrieval
352     _autoinc_supplied_for_op
353
354     _resolve_bindattrs
355
356     _max_column_bytesize
357     _is_lob_type
358     _is_binary_lob_type
359     _is_binary_type
360     _is_text_lob_type
361
362     _prepare_sth
363     _bind_sth_params
364   /,(
365     # the capability framework
366     # not sure if CMOP->initialize does evil things to DBIC::S::DBI, fix if a problem
367     grep
368       { $_ =~ /^ _ (?: use | supports | determine_supports ) _ /x }
369       ( Class::MOP::Class->initialize('DBIx::Class::Storage::DBI')->get_all_method_names )
370   )],
371 };
372
373 if (DBIx::Class::_ENV_::DBICTEST) {
374
375   my $seen;
376   for my $type (keys %$method_dispatch) {
377     for (@{$method_dispatch->{$type}}) {
378       push @{$seen->{$_}}, $type;
379     }
380   }
381
382   if (my @dupes = grep { @{$seen->{$_}} > 1 } keys %$seen) {
383     die(join "\n", '',
384       'The following methods show up multiple times in ::Storage::DBI::Replicated handlers:',
385       (map { "$_: " . (join ', ', @{$seen->{$_}}) } sort @dupes),
386       '',
387     );
388   }
389
390   if (my @cant = grep { ! DBIx::Class::Storage::DBI->can($_) } keys %$seen) {
391     die(join "\n", '',
392       '::Storage::DBI::Replicated specifies handling of the following *NON EXISTING* ::Storage::DBI methods:',
393       @cant,
394       '',
395     );
396   }
397 }
398
399 for my $method (@{$method_dispatch->{unimplemented}}) {
400   __PACKAGE__->meta->add_method($method, sub {
401     my $self = shift;
402     $self->throw_exception("$method() must not be called on ".(blessed $self).' objects');
403   });
404 }
405
406 =head2 read_handler
407
408 Defines an object that implements the read side of L<DBIx::Class::Storage::DBI>.
409
410 =cut
411
412 has 'read_handler' => (
413   is=>'rw',
414   isa=>Object,
415   lazy_build=>1,
416   handles=>$method_dispatch->{reader},
417 );
418
419 =head2 write_handler
420
421 Defines an object that implements the write side of L<DBIx::Class::Storage::DBI>,
422 as well as methods that don't write or read that can be called on only one
423 storage, methods that return a C<$dbh>, and any methods that don't make sense to
424 run on a replicant.
425
426 =cut
427
428 has 'write_handler' => (
429   is=>'ro',
430   isa=>Object,
431   lazy_build=>1,
432   handles=>$method_dispatch->{writer},
433 );
434
435
436
437 has _master_connect_info_opts =>
438   (is => 'rw', isa => HashRef, default => sub { {} });
439
440 =head2 around: connect_info
441
442 Preserves master's C<connect_info> options (for merging with replicants.)
443 Also sets any Replicated-related options from connect_info, such as
444 C<pool_type>, C<pool_args>, C<balancer_type> and C<balancer_args>.
445
446 =cut
447
448 around connect_info => sub {
449   my ($next, $self, $info, @extra) = @_;
450
451   $self->throw_exception(
452     'connect_info can not be retrieved from a replicated storage - '
453   . 'accessor must be called on a specific pool instance'
454   ) unless defined $info;
455
456   my $merge = Hash::Merge->new('LEFT_PRECEDENT');
457
458   my %opts;
459   for my $arg (@$info) {
460     next unless (reftype($arg)||'') eq 'HASH';
461     %opts = %{ $merge->merge($arg, \%opts) };
462   }
463   delete $opts{dsn};
464
465   if (@opts{qw/pool_type pool_args/}) {
466     $self->pool_type(delete $opts{pool_type})
467       if $opts{pool_type};
468
469     $self->pool_args(
470       $merge->merge((delete $opts{pool_args} || {}), $self->pool_args)
471     );
472
473     ## Since we possibly changed the pool_args, we need to clear the current
474     ## pool object so that next time it is used it will be rebuilt.
475     $self->clear_pool;
476   }
477
478   if (@opts{qw/balancer_type balancer_args/}) {
479     $self->balancer_type(delete $opts{balancer_type})
480       if $opts{balancer_type};
481
482     $self->balancer_args(
483       $merge->merge((delete $opts{balancer_args} || {}), $self->balancer_args)
484     );
485
486     $self->balancer($self->_build_balancer)
487       if $self->balancer;
488   }
489
490   $self->_master_connect_info_opts(\%opts);
491
492   return preserve_context {
493     $self->$next($info, @extra);
494   } after => sub {
495     # Make sure master is blessed into the correct class and apply role to it.
496     my $master = $self->master;
497     $master->_determine_driver;
498     Moose::Meta::Class->initialize(ref $master);
499
500     DBIx::Class::Storage::DBI::Replicated::WithDSN->meta->apply($master);
501
502     # link pool back to master
503     $self->pool->master($master);
504   };
505 };
506
507 =head1 METHODS
508
509 This class defines the following methods.
510
511 =head2 BUILDARGS
512
513 L<DBIx::Class::Schema> when instantiating its storage passed itself as the
514 first argument.  So we need to massage the arguments a bit so that all the
515 bits get put into the correct places.
516
517 =cut
518
519 sub BUILDARGS {
520   my ($class, $schema, $storage_type_args, @args) = @_;
521
522   return {
523     schema=>$schema,
524     %$storage_type_args,
525     @args
526   }
527 }
528
529 =head2 _build_master
530
531 Lazy builder for the L</master> attribute.
532
533 =cut
534
535 sub _build_master {
536   my $self = shift @_;
537   my $master = DBIx::Class::Storage::DBI->new($self->schema);
538   $master
539 }
540
541 =head2 _build_pool
542
543 Lazy builder for the L</pool> attribute.
544
545 =cut
546
547 sub _build_pool {
548   my $self = shift @_;
549   $self->create_pool(%{$self->pool_args});
550 }
551
552 =head2 _build_balancer
553
554 Lazy builder for the L</balancer> attribute.  This takes a Pool object so that
555 the balancer knows which pool it's balancing.
556
557 =cut
558
559 sub _build_balancer {
560   my $self = shift @_;
561   $self->create_balancer(
562     pool=>$self->pool,
563     master=>$self->master,
564     %{$self->balancer_args},
565   );
566 }
567
568 =head2 _build_write_handler
569
570 Lazy builder for the L</write_handler> attribute.  The default is to set this to
571 the L</master>.
572
573 =cut
574
575 sub _build_write_handler {
576   return shift->master;
577 }
578
579 =head2 _build_read_handler
580
581 Lazy builder for the L</read_handler> attribute.  The default is to set this to
582 the L</balancer>.
583
584 =cut
585
586 sub _build_read_handler {
587   return shift->balancer;
588 }
589
590 =head2 around: connect_replicants
591
592 All calls to connect_replicants needs to have an existing $schema tacked onto
593 top of the args, since L<DBIx::Class::Storage::DBI> needs it, and any
594 L<connect_info|DBIx::Class::Storage::DBI/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 all the connected storage backends.  The first element
650 in the returned array is the master, and the rest 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
1086 L<< $result->discard_changes|DBIx::Class::Row/discard_changes >>)
1087 or you insert a row and then immediately build a query that expects that row
1088 to be an item, you should force the master to handle reads.  Otherwise, due to
1089 the lag, there is no certainty your data will be in the expected state.
1090
1091 For data integrity, all transactions automatically use the master storage for
1092 all read and write queries.  Using a transaction is the preferred and recommended
1093 method to force the master to handle all read queries.
1094
1095 Otherwise, you can force a single query to use the master with the 'force_pool'
1096 attribute:
1097
1098   my $result = $resultset->search(undef, {force_pool=>'master'})->find($pk);
1099
1100 This attribute will safely be ignored by non replicated storages, so you can use
1101 the same code for both types of systems.
1102
1103 Lastly, you can use the L</execute_reliably> method, which works very much like
1104 a transaction.
1105
1106 For debugging, you can turn replication on/off with the methods L</set_reliable_storage>
1107 and L</set_balanced_storage>, however this operates at a global level and is not
1108 suitable if you have a shared Schema object being used by multiple processes,
1109 such as on a web application server.  You can get around this limitation by
1110 using the Schema clone method.
1111
1112   my $new_schema = $schema->clone;
1113   $new_schema->set_reliable_storage;
1114
1115   ## $new_schema will use only the Master storage for all reads/writes while
1116   ## the $schema object will use replicated storage.
1117
1118 =head1 AUTHOR
1119
1120   John Napiorkowski <john.napiorkowski@takkle.com>
1121
1122 Based on code originated by:
1123
1124   Norbert Csongrádi <bert@cpan.org>
1125   Peter Siklósi <einon@einon.hu>
1126
1127 =head1 LICENSE
1128
1129 You may distribute this code under the same terms as Perl itself.
1130
1131 =cut
1132
1133 __PACKAGE__->meta->make_immutable;
1134
1135 1;