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