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