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