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