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