Allow users to disable the "DT in search" warning introduced in e1038213
[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
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
266 insert_bulk
267 update
268 delete
269 dbh
2ce6e9a6 270 txn_begin
64cdad22 271 txn_do
272 txn_commit
273 txn_rollback
2ce6e9a6 274 txn_scope_guard
90d7422f 275 _exec_txn_rollback
276 _exec_txn_begin
277 _exec_txn_commit
64cdad22 278 deploy
0180bef9 279 with_deferred_fk_checks
6f7344b8 280 dbh_do
2ce6e9a6 281 _prep_for_execute
6f7344b8 282 is_datatype_numeric
283 _count_select
6f7344b8 284 _subq_update_delete
285 svp_rollback
286 svp_begin
287 svp_release
e398f77e 288 relname_to_table_alias
3244fdcc 289 _dbh_last_insert_id
3244fdcc 290 _default_dbi_connect_attributes
291 _dbi_connect_info
b9ca4ff1 292 _dbic_connect_attributes
3244fdcc 293 auto_savepoint
0e773352 294 _query_start
3244fdcc 295 _query_end
0e773352 296 _format_for_trace
297 _dbi_attrs_for_bind
3244fdcc 298 bind_attribute_by_data_type
299 transaction_depth
300 _dbh
301 _select_args
302 _dbh_execute_array
3244fdcc 303 _sql_maker
3244fdcc 304 _per_row_update_delete
3244fdcc 305 _dbh_execute_inserts_with_no_binds
306 _select_args_to_query
5e782048 307 _gen_sql_bind
3244fdcc 308 _svp_generate_name
309 _multipk_update_delete
3244fdcc 310 _normalize_connect_info
311 _parse_connect_do
3244fdcc 312 _execute_array
3244fdcc 313 savepoints
3244fdcc 314 _sql_maker_opts
315 _conn_pid
3244fdcc 316 _dbh_autocommit
317 _native_data_type
318 _get_dbh
319 sql_maker_class
3244fdcc 320 _execute
321 _do_query
402ac1c9 322 _sth
3244fdcc 323 _dbh_sth
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
336
0e773352 337 source_bind_attributes
338
4bea1fe7 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
4bea1fe7 350 _prefetch_autovalues
351
352 _max_column_bytesize
353 _is_lob_type
354 _is_binary_lob_type
355 _is_text_lob_type
402ac1c9 356
357 sth
4bea1fe7 358 /,(
359 # the capability framework
360 # not sure if CMOP->initialize does evil things to DBIC::S::DBI, fix if a problem
361 grep
362 { $_ =~ /^ _ (?: use | supports | determine_supports ) _ /x }
363 ( Class::MOP::Class->initialize('DBIx::Class::Storage::DBI')->get_all_method_names )
364 )],
365};
366
367if (DBIx::Class::_ENV_::DBICTEST) {
368
369 my $seen;
370 for my $type (keys %$method_dispatch) {
371 for (@{$method_dispatch->{$type}}) {
372 push @{$seen->{$_}}, $type;
373 }
374 }
cb6ec758 375
4bea1fe7 376 if (my @dupes = grep { @{$seen->{$_}} > 1 } keys %$seen) {
377 die(join "\n", '',
378 'The following methods show up multiple times in ::Storage::DBI::Replicated handlers:',
379 (map { "$_: " . (join ', ', @{$seen->{$_}}) } sort @dupes),
380 '',
381 );
382 }
bbdda281 383
4bea1fe7 384 if (my @cant = grep { ! DBIx::Class::Storage::DBI->can($_) } keys %$seen) {
385 die(join "\n", '',
386 '::Storage::DBI::Replicated specifies handling of the following *NON EXISTING* ::Storage::DBI methods:',
387 @cant,
388 '',
389 );
390 }
391}
bbdda281 392
4bea1fe7 393for my $method (@{$method_dispatch->{unimplemented}}) {
394 __PACKAGE__->meta->add_method($method, sub {
70c28808 395 my $self = shift;
396 $self->throw_exception("$method must not be called on ".(blessed $self).' objects');
4bea1fe7 397 });
398}
31a8aaaf 399
4bea1fe7 400=head2 read_handler
584ea6e4 401
4bea1fe7 402Defines an object that implements the read side of L<BIx::Class::Storage::DBI>.
584ea6e4 403
4bea1fe7 404=cut
405
406has 'read_handler' => (
407 is=>'rw',
408 isa=>Object,
409 lazy_build=>1,
410 handles=>$method_dispatch->{reader},
e471ab87 411);
412
4bea1fe7 413=head2 write_handler
414
415Defines an object that implements the write side of L<BIx::Class::Storage::DBI>,
416as well as methods that don't write or read that can be called on only one
417storage, methods that return a C<$dbh>, and any methods that don't make sense to
418run on a replicant.
419
420=cut
421
422has 'write_handler' => (
423 is=>'ro',
424 isa=>Object,
425 lazy_build=>1,
426 handles=>$method_dispatch->{writer},
7f4433eb 427);
428
4bea1fe7 429
6d766626 430
b2e4d522 431has _master_connect_info_opts =>
432 (is => 'rw', isa => HashRef, default => sub { {} });
433
434=head2 around: connect_info
435
48580715 436Preserves master's C<connect_info> options (for merging with replicants.)
437Also sets any Replicated-related options from connect_info, such as
dcdf7b2c 438C<pool_type>, C<pool_args>, C<balancer_type> and C<balancer_args>.
b2e4d522 439
440=cut
441
442around connect_info => sub {
443 my ($next, $self, $info, @extra) = @_;
444
282a9a4f 445 my $merge = Hash::Merge->new('LEFT_PRECEDENT');
e666c5fd 446
b2e4d522 447 my %opts;
448 for my $arg (@$info) {
449 next unless (reftype($arg)||'') eq 'HASH';
e666c5fd 450 %opts = %{ $merge->merge($arg, \%opts) };
b2e4d522 451 }
b2e4d522 452 delete $opts{dsn};
453
dcdf7b2c 454 if (@opts{qw/pool_type pool_args/}) {
455 $self->pool_type(delete $opts{pool_type})
456 if $opts{pool_type};
457
b88b85e7 458 $self->pool_args(
e666c5fd 459 $merge->merge((delete $opts{pool_args} || {}), $self->pool_args)
b88b85e7 460 );
dcdf7b2c 461
64ae1667 462 ## Since we possibly changed the pool_args, we need to clear the current
463 ## pool object so that next time it is used it will be rebuilt.
464 $self->clear_pool;
dcdf7b2c 465 }
466
467 if (@opts{qw/balancer_type balancer_args/}) {
468 $self->balancer_type(delete $opts{balancer_type})
469 if $opts{balancer_type};
470
b88b85e7 471 $self->balancer_args(
e666c5fd 472 $merge->merge((delete $opts{balancer_args} || {}), $self->balancer_args)
b88b85e7 473 );
dcdf7b2c 474
67c43863 475 $self->balancer($self->_build_balancer)
6f7344b8 476 if $self->balancer;
dcdf7b2c 477 }
478
b2e4d522 479 $self->_master_connect_info_opts(\%opts);
480
cca282b6 481 my @res;
482 if (wantarray) {
0ce2d0d5 483 @res = $self->$next($info, @extra);
484 } else {
cca282b6 485 $res[0] = $self->$next($info, @extra);
0ce2d0d5 486 }
487
fd4eb9c2 488 # Make sure master is blessed into the correct class and apply role to it.
489 my $master = $self->master;
490 $master->_determine_driver;
491 Moose::Meta::Class->initialize(ref $master);
cea43436 492
ec0946db 493 DBIx::Class::Storage::DBI::Replicated::WithDSN->meta->apply($master);
cea43436 494
495 # link pool back to master
496 $self->pool->master($master);
0ce2d0d5 497
cca282b6 498 wantarray ? @res : $res[0];
b2e4d522 499};
500
26ab719a 501=head1 METHODS
2bf79155 502
26ab719a 503This class defines the following methods.
2bf79155 504
c354902c 505=head2 BUILDARGS
2bf79155 506
faaba25f 507L<DBIx::Class::Schema> when instantiating its storage passed itself as the
2ce6e9a6 508first argument. So we need to massage the arguments a bit so that all the
509bits get put into the correct places.
2bf79155 510
511=cut
512
c354902c 513sub BUILDARGS {
fd323bf1 514 my ($class, $schema, $storage_type_args, @args) = @_;
d4daee7b 515
c354902c 516 return {
6f7344b8 517 schema=>$schema,
518 %$storage_type_args,
519 @args
c354902c 520 }
521}
2bf79155 522
cb6ec758 523=head2 _build_master
2bf79155 524
cb6ec758 525Lazy builder for the L</master> attribute.
2bf79155 526
527=cut
528
cb6ec758 529sub _build_master {
2ce6e9a6 530 my $self = shift @_;
ee356d00 531 my $master = DBIx::Class::Storage::DBI->new($self->schema);
ee356d00 532 $master
106d5f3b 533}
534
26ab719a 535=head2 _build_pool
2bf79155 536
26ab719a 537Lazy builder for the L</pool> attribute.
2bf79155 538
539=cut
540
26ab719a 541sub _build_pool {
64cdad22 542 my $self = shift @_;
543 $self->create_pool(%{$self->pool_args});
2bf79155 544}
545
26ab719a 546=head2 _build_balancer
2bf79155 547
cb6ec758 548Lazy builder for the L</balancer> attribute. This takes a Pool object so that
549the balancer knows which pool it's balancing.
2bf79155 550
551=cut
552
26ab719a 553sub _build_balancer {
64cdad22 554 my $self = shift @_;
555 $self->create_balancer(
6f7344b8 556 pool=>$self->pool,
64cdad22 557 master=>$self->master,
558 %{$self->balancer_args},
559 );
2bf79155 560}
561
cb6ec758 562=head2 _build_write_handler
2bf79155 563
cb6ec758 564Lazy builder for the L</write_handler> attribute. The default is to set this to
565the L</master>.
50336325 566
567=cut
568
cb6ec758 569sub _build_write_handler {
64cdad22 570 return shift->master;
cb6ec758 571}
50336325 572
cb6ec758 573=head2 _build_read_handler
2bf79155 574
cb6ec758 575Lazy builder for the L</read_handler> attribute. The default is to set this to
576the L</balancer>.
2bf79155 577
578=cut
579
cb6ec758 580sub _build_read_handler {
64cdad22 581 return shift->balancer;
cb6ec758 582}
50336325 583
cb6ec758 584=head2 around: connect_replicants
2bf79155 585
cb6ec758 586All calls to connect_replicants needs to have an existing $schema tacked onto
b2e4d522 587top of the args, since L<DBIx::Storage::DBI> needs it, and any C<connect_info>
588options merged with the master, with replicant opts having higher priority.
955a6df6 589
cb6ec758 590=cut
955a6df6 591
b2e4d522 592around connect_replicants => sub {
593 my ($next, $self, @args) = @_;
594
595 for my $r (@args) {
596 $r = [ $r ] unless reftype $r eq 'ARRAY';
597
1a58752c 598 $self->throw_exception('coderef replicant connect_info not supported')
b2e4d522 599 if ref $r->[0] && reftype $r->[0] eq 'CODE';
600
601# any connect_info options?
602 my $i = 0;
603 $i++ while $i < @$r && (reftype($r->[$i])||'') ne 'HASH';
604
6f7344b8 605# make one if none
b2e4d522 606 $r->[$i] = {} unless $r->[$i];
607
608# merge if two hashes
b88b85e7 609 my @hashes = @$r[$i .. $#{$r}];
610
1a58752c 611 $self->throw_exception('invalid connect_info options')
b88b85e7 612 if (grep { reftype($_) eq 'HASH' } @hashes) != @hashes;
613
1a58752c 614 $self->throw_exception('too many hashrefs in connect_info')
b88b85e7 615 if @hashes > 2;
616
282a9a4f 617 my $merge = Hash::Merge->new('LEFT_PRECEDENT');
e666c5fd 618 my %opts = %{ $merge->merge(reverse @hashes) };
b88b85e7 619
620# delete them
b2e4d522 621 splice @$r, $i+1, ($#{$r} - $i), ();
622
0bd8e058 623# make sure master/replicants opts don't clash
624 my %master_opts = %{ $self->_master_connect_info_opts };
625 if (exists $opts{dbh_maker}) {
626 delete @master_opts{qw/dsn user password/};
627 }
628 delete $master_opts{dbh_maker};
629
b2e4d522 630# merge with master
e666c5fd 631 %opts = %{ $merge->merge(\%opts, \%master_opts) };
b2e4d522 632
633# update
634 $r->[$i] = \%opts;
635 }
636
637 $self->$next($self->schema, @args);
955a6df6 638};
2bf79155 639
2bf79155 640=head2 all_storages
641
642Returns an array of of all the connected storage backends. The first element
643in the returned array is the master, and the remainings are each of the
644replicants.
645
646=cut
647
648sub all_storages {
64cdad22 649 my $self = shift @_;
650 return grep {defined $_ && blessed $_} (
651 $self->master,
6412a592 652 values %{ $self->replicants },
64cdad22 653 );
2bf79155 654}
655
c4d3fae2 656=head2 execute_reliably ($coderef, ?@args)
657
658Given a coderef, saves the current state of the L</read_handler>, forces it to
48580715 659use reliable storage (e.g. sets it to the master), executes a coderef and then
c4d3fae2 660restores the original state.
661
662Example:
663
64cdad22 664 my $reliably = sub {
665 my $name = shift @_;
666 $schema->resultset('User')->create({name=>$name});
fd323bf1 667 my $user_rs = $schema->resultset('User')->find({name=>$name});
64cdad22 668 return $user_rs;
669 };
c4d3fae2 670
64cdad22 671 my $user_rs = $schema->storage->execute_reliably($reliably, 'John');
c4d3fae2 672
673Use this when you must be certain of your database state, such as when you just
674inserted something and need to get a resultset including it, etc.
675
676=cut
677
678sub execute_reliably {
64cdad22 679 my ($self, $coderef, @args) = @_;
d4daee7b 680
64cdad22 681 unless( ref $coderef eq 'CODE') {
682 $self->throw_exception('Second argument must be a coderef');
683 }
d4daee7b 684
64cdad22 685 ##Get copy of master storage
686 my $master = $self->master;
d4daee7b 687
64cdad22 688 ##Get whatever the current read hander is
689 my $current = $self->read_handler;
d4daee7b 690
64cdad22 691 ##Set the read handler to master
692 $self->read_handler($master);
d4daee7b 693
64cdad22 694 ## do whatever the caller needs
695 my @result;
696 my $want_array = wantarray;
d4daee7b 697
ed7ab0f4 698 try {
64cdad22 699 if($want_array) {
700 @result = $coderef->(@args);
701 } elsif(defined $want_array) {
702 ($result[0]) = ($coderef->(@args));
ed213e85 703 } else {
64cdad22 704 $coderef->(@args);
6f7344b8 705 }
ed7ab0f4 706 } catch {
707 $self->throw_exception("coderef returned an error: $_");
708 } finally {
709 ##Reset to the original state
710 $self->read_handler($current);
64cdad22 711 };
d4daee7b 712
cca282b6 713 return wantarray ? @result : $result[0];
c4d3fae2 714}
715
cb6ec758 716=head2 set_reliable_storage
717
718Sets the current $schema to be 'reliable', that is all queries, both read and
719write are sent to the master
d4daee7b 720
cb6ec758 721=cut
722
723sub set_reliable_storage {
64cdad22 724 my $self = shift @_;
725 my $schema = $self->schema;
726 my $write_handler = $self->schema->storage->write_handler;
d4daee7b 727
64cdad22 728 $schema->storage->read_handler($write_handler);
cb6ec758 729}
730
731=head2 set_balanced_storage
732
733Sets the current $schema to be use the </balancer> for all reads, while all
48580715 734writes are sent to the master only
d4daee7b 735
cb6ec758 736=cut
737
738sub set_balanced_storage {
64cdad22 739 my $self = shift @_;
740 my $schema = $self->schema;
bd5da369 741 my $balanced_handler = $self->schema->storage->balancer;
d4daee7b 742
bd5da369 743 $schema->storage->read_handler($balanced_handler);
cb6ec758 744}
2bf79155 745
746=head2 connected
747
748Check that the master and at least one of the replicants is connected.
749
750=cut
751
752sub connected {
64cdad22 753 my $self = shift @_;
754 return
755 $self->master->connected &&
756 $self->pool->connected_replicants;
2bf79155 757}
758
2bf79155 759=head2 ensure_connected
760
761Make sure all the storages are connected.
762
763=cut
764
765sub ensure_connected {
64cdad22 766 my $self = shift @_;
767 foreach my $source ($self->all_storages) {
768 $source->ensure_connected(@_);
769 }
2bf79155 770}
771
2bf79155 772=head2 limit_dialect
773
774Set the limit_dialect for all existing storages
775
776=cut
777
778sub limit_dialect {
64cdad22 779 my $self = shift @_;
780 foreach my $source ($self->all_storages) {
781 $source->limit_dialect(@_);
782 }
f3e9f010 783 return $self->master->limit_dialect;
2bf79155 784}
785
2bf79155 786=head2 quote_char
787
788Set the quote_char for all existing storages
789
790=cut
791
792sub quote_char {
64cdad22 793 my $self = shift @_;
794 foreach my $source ($self->all_storages) {
795 $source->quote_char(@_);
796 }
3fbe08e3 797 return $self->master->quote_char;
2bf79155 798}
799
2bf79155 800=head2 name_sep
801
802Set the name_sep for all existing storages
803
804=cut
805
806sub name_sep {
64cdad22 807 my $self = shift @_;
808 foreach my $source ($self->all_storages) {
809 $source->name_sep(@_);
810 }
3fbe08e3 811 return $self->master->name_sep;
2bf79155 812}
813
2bf79155 814=head2 set_schema
815
816Set the schema object for all existing storages
817
818=cut
819
820sub set_schema {
64cdad22 821 my $self = shift @_;
822 foreach my $source ($self->all_storages) {
823 $source->set_schema(@_);
824 }
2bf79155 825}
826
2bf79155 827=head2 debug
828
829set a debug flag across all storages
830
831=cut
832
833sub debug {
64cdad22 834 my $self = shift @_;
3fbe08e3 835 if(@_) {
836 foreach my $source ($self->all_storages) {
837 $source->debug(@_);
6f7344b8 838 }
64cdad22 839 }
3fbe08e3 840 return $self->master->debug;
2bf79155 841}
842
2bf79155 843=head2 debugobj
844
cea43436 845set a debug object
2bf79155 846
847=cut
848
849sub debugobj {
64cdad22 850 my $self = shift @_;
cea43436 851 return $self->master->debugobj(@_);
2bf79155 852}
853
2bf79155 854=head2 debugfh
855
cea43436 856set a debugfh object
2bf79155 857
858=cut
859
860sub debugfh {
64cdad22 861 my $self = shift @_;
cea43436 862 return $self->master->debugfh(@_);
2bf79155 863}
864
2bf79155 865=head2 debugcb
866
cea43436 867set a debug callback
2bf79155 868
869=cut
870
871sub debugcb {
64cdad22 872 my $self = shift @_;
cea43436 873 return $self->master->debugcb(@_);
2bf79155 874}
875
2bf79155 876=head2 disconnect
877
878disconnect everything
879
880=cut
881
882sub disconnect {
64cdad22 883 my $self = shift @_;
884 foreach my $source ($self->all_storages) {
885 $source->disconnect(@_);
886 }
2bf79155 887}
888
b2e4d522 889=head2 cursor_class
890
891set cursor class on all storages, or return master's
892
893=cut
894
895sub cursor_class {
896 my ($self, $cursor_class) = @_;
897
898 if ($cursor_class) {
899 $_->cursor_class($cursor_class) for $self->all_storages;
900 }
901 $self->master->cursor_class;
902}
d4daee7b 903
3244fdcc 904=head2 cursor
905
906set cursor class on all storages, or return master's, alias for L</cursor_class>
907above.
908
909=cut
910
911sub cursor {
912 my ($self, $cursor_class) = @_;
913
914 if ($cursor_class) {
915 $_->cursor($cursor_class) for $self->all_storages;
916 }
917 $self->master->cursor;
918}
919
920=head2 unsafe
921
922sets the L<DBIx::Class::Storage::DBI/unsafe> option on all storages or returns
923master's current setting
924
925=cut
926
927sub unsafe {
928 my $self = shift;
929
930 if (@_) {
931 $_->unsafe(@_) for $self->all_storages;
932 }
933
934 return $self->master->unsafe;
935}
936
937=head2 disable_sth_caching
938
939sets the L<DBIx::Class::Storage::DBI/disable_sth_caching> option on all storages
940or returns master's current setting
941
942=cut
943
944sub disable_sth_caching {
945 my $self = shift;
946
947 if (@_) {
948 $_->disable_sth_caching(@_) for $self->all_storages;
949 }
950
951 return $self->master->disable_sth_caching;
952}
953
954=head2 lag_behind_master
955
956returns the highest Replicant L<DBIx::Class::Storage::DBI/lag_behind_master>
957setting
958
959=cut
960
961sub lag_behind_master {
962 my $self = shift;
963
964 return max map $_->lag_behind_master, $self->replicants;
fd323bf1 965}
3244fdcc 966
967=head2 is_replicating
968
969returns true if all replicants return true for
970L<DBIx::Class::Storage::DBI/is_replicating>
971
972=cut
973
974sub is_replicating {
975 my $self = shift;
976
977 return (grep $_->is_replicating, $self->replicants) == ($self->replicants);
978}
979
980=head2 connect_call_datetime_setup
981
982calls L<DBIx::Class::Storage::DBI/connect_call_datetime_setup> for all storages
983
984=cut
985
986sub connect_call_datetime_setup {
987 my $self = shift;
988 $_->connect_call_datetime_setup for $self->all_storages;
989}
990
991sub _populate_dbh {
992 my $self = shift;
993 $_->_populate_dbh for $self->all_storages;
994}
995
996sub _connect {
997 my $self = shift;
998 $_->_connect for $self->all_storages;
999}
1000
1001sub _rebless {
1002 my $self = shift;
1003 $_->_rebless for $self->all_storages;
1004}
1005
1006sub _determine_driver {
1007 my $self = shift;
1008 $_->_determine_driver for $self->all_storages;
1009}
1010
1011sub _driver_determined {
1012 my $self = shift;
fd323bf1 1013
3244fdcc 1014 if (@_) {
1015 $_->_driver_determined(@_) for $self->all_storages;
1016 }
1017
1018 return $self->master->_driver_determined;
1019}
1020
1021sub _init {
1022 my $self = shift;
fd323bf1 1023
3244fdcc 1024 $_->_init for $self->all_storages;
1025}
1026
1027sub _run_connection_actions {
1028 my $self = shift;
fd323bf1 1029
3244fdcc 1030 $_->_run_connection_actions for $self->all_storages;
1031}
1032
1033sub _do_connection_actions {
1034 my $self = shift;
fd323bf1 1035
3244fdcc 1036 if (@_) {
1037 $_->_do_connection_actions(@_) for $self->all_storages;
1038 }
1039}
1040
1041sub connect_call_do_sql {
1042 my $self = shift;
1043 $_->connect_call_do_sql(@_) for $self->all_storages;
1044}
1045
1046sub disconnect_call_do_sql {
1047 my $self = shift;
1048 $_->disconnect_call_do_sql(@_) for $self->all_storages;
1049}
1050
1051sub _seems_connected {
1052 my $self = shift;
1053
1054 return min map $_->_seems_connected, $self->all_storages;
1055}
1056
1057sub _ping {
1058 my $self = shift;
1059
1060 return min map $_->_ping, $self->all_storages;
1061}
1062
bbdda281 1063# not using the normalized_version, because we want to preserve
1064# version numbers much longer than the conventional xxx.yyyzzz
7da56142 1065my $numify_ver = sub {
1066 my $ver = shift;
1067 my @numparts = split /\D+/, $ver;
bbdda281 1068 my $format = '%d.' . (join '', ('%06d') x (@numparts - 1));
7da56142 1069
1070 return sprintf $format, @numparts;
1071};
fecb38cb 1072sub _server_info {
1073 my $self = shift;
1074
bbdda281 1075 if (not $self->_dbh_details->{info}) {
1076 $self->_dbh_details->{info} = (
fd323bf1 1077 reduce { $a->[0] < $b->[0] ? $a : $b }
7da56142 1078 map [ $numify_ver->($_->{dbms_version}), $_ ],
1079 map $_->_server_info, $self->all_storages
1080 )->[1];
fecb38cb 1081 }
1082
bbdda281 1083 return $self->next::method;
fecb38cb 1084}
1085
1086sub _get_server_version {
1087 my $self = shift;
1088
1089 return $self->_server_info->{dbms_version};
1090}
1091
7e38d850 1092=head1 GOTCHAS
1093
1094Due to the fact that replicants can lag behind a master, you must take care to
1095make sure you use one of the methods to force read queries to a master should
1096you need realtime data integrity. For example, if you insert a row, and then
1097immediately re-read it from the database (say, by doing $row->discard_changes)
1098or you insert a row and then immediately build a query that expects that row
1099to be an item, you should force the master to handle reads. Otherwise, due to
1100the lag, there is no certainty your data will be in the expected state.
1101
1102For data integrity, all transactions automatically use the master storage for
1103all read and write queries. Using a transaction is the preferred and recommended
1104method to force the master to handle all read queries.
1105
1106Otherwise, you can force a single query to use the master with the 'force_pool'
1107attribute:
1108
1109 my $row = $resultset->search(undef, {force_pool=>'master'})->find($pk);
1110
1111This attribute will safely be ignore by non replicated storages, so you can use
1112the same code for both types of systems.
1113
1114Lastly, you can use the L</execute_reliably> method, which works very much like
1115a transaction.
1116
1117For debugging, you can turn replication on/off with the methods L</set_reliable_storage>
1118and L</set_balanced_storage>, however this operates at a global level and is not
1119suitable if you have a shared Schema object being used by multiple processes,
1120such as on a web application server. You can get around this limitation by
1121using the Schema clone method.
1122
1123 my $new_schema = $schema->clone;
1124 $new_schema->set_reliable_storage;
d4daee7b 1125
7e38d850 1126 ## $new_schema will use only the Master storage for all reads/writes while
1127 ## the $schema object will use replicated storage.
1128
f5d3a5de 1129=head1 AUTHOR
1130
64cdad22 1131 John Napiorkowski <john.napiorkowski@takkle.com>
f5d3a5de 1132
c4d3fae2 1133Based on code originated by:
f5d3a5de 1134
64cdad22 1135 Norbert Csongrádi <bert@cpan.org>
1136 Peter Siklósi <einon@einon.hu>
2156bbdd 1137
f5d3a5de 1138=head1 LICENSE
1139
1140You may distribute this code under the same terms as Perl itself.
1141
1142=cut
1143
c354902c 1144__PACKAGE__->meta->make_immutable;
1145
f5d3a5de 11461;