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