update ::DBI::Replicated
[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
311 _subq_count_select
312 _subq_update_delete
313 svp_rollback
314 svp_begin
315 svp_release
e398f77e 316 relname_to_table_alias
317 _straight_join_to_node
3244fdcc 318 _dbh_last_insert_id
319 _fix_bind_params
320 _default_dbi_connect_attributes
321 _dbi_connect_info
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
330 _sql_maker_args
331 _sql_maker
332 _query_start
333 _sqlt_version_error
334 _per_row_update_delete
335 _dbh_begin_work
336 _dbh_execute_inserts_with_no_binds
337 _select_args_to_query
338 _svp_generate_name
339 _multipk_update_delete
340 source_bind_attributes
341 _normalize_connect_info
342 _parse_connect_do
343 _dbh_commit
344 _execute_array
345 _placeholders_supported
3244fdcc 346 savepoints
347 _sqlt_minimum_version
348 _sql_maker_opts
349 _conn_pid
350 _typeless_placeholders_supported
351 _conn_tid
352 _dbh_autocommit
353 _native_data_type
354 _get_dbh
355 sql_maker_class
356 _dbh_rollback
357 _adjust_select_args_for_complex_prefetch
358 _resolve_ident_sources
359 _resolve_column_info
360 _prune_unused_joins
361 _strip_cond_qualifiers
362 _parse_order_by
363 _resolve_aliastypes_from_select_args
364 _execute
365 _do_query
366 _dbh_sth
367 _dbh_execute
6ed1cd2e 368 _prefetch_insert_auto_nextvals
6d766626 369 _server_info_hash
64cdad22 370 /],
cb6ec758 371);
372
e471ab87 373my @unimplemented = qw(
374 _arm_global_destructor
375 _preserve_foreign_dbh
376);
377
378for my $method (@unimplemented) {
379 __PACKAGE__->meta->add_method($method, sub {
380 croak "$method must not be called on ".(blessed shift).' objects';
381 });
382}
6d766626 383
b2e4d522 384has _master_connect_info_opts =>
385 (is => 'rw', isa => HashRef, default => sub { {} });
386
387=head2 around: connect_info
388
48580715 389Preserves master's C<connect_info> options (for merging with replicants.)
390Also sets any Replicated-related options from connect_info, such as
dcdf7b2c 391C<pool_type>, C<pool_args>, C<balancer_type> and C<balancer_args>.
b2e4d522 392
393=cut
394
395around connect_info => sub {
396 my ($next, $self, $info, @extra) = @_;
397
0ce2d0d5 398 my $wantarray = wantarray;
399
282a9a4f 400 my $merge = Hash::Merge->new('LEFT_PRECEDENT');
e666c5fd 401
b2e4d522 402 my %opts;
403 for my $arg (@$info) {
404 next unless (reftype($arg)||'') eq 'HASH';
e666c5fd 405 %opts = %{ $merge->merge($arg, \%opts) };
b2e4d522 406 }
b2e4d522 407 delete $opts{dsn};
408
dcdf7b2c 409 if (@opts{qw/pool_type pool_args/}) {
410 $self->pool_type(delete $opts{pool_type})
411 if $opts{pool_type};
412
b88b85e7 413 $self->pool_args(
e666c5fd 414 $merge->merge((delete $opts{pool_args} || {}), $self->pool_args)
b88b85e7 415 );
dcdf7b2c 416
67c43863 417 $self->pool($self->_build_pool)
6f7344b8 418 if $self->pool;
dcdf7b2c 419 }
420
421 if (@opts{qw/balancer_type balancer_args/}) {
422 $self->balancer_type(delete $opts{balancer_type})
423 if $opts{balancer_type};
424
b88b85e7 425 $self->balancer_args(
e666c5fd 426 $merge->merge((delete $opts{balancer_args} || {}), $self->balancer_args)
b88b85e7 427 );
dcdf7b2c 428
67c43863 429 $self->balancer($self->_build_balancer)
6f7344b8 430 if $self->balancer;
dcdf7b2c 431 }
432
b2e4d522 433 $self->_master_connect_info_opts(\%opts);
434
0ce2d0d5 435 my (@res, $res);
436 if ($wantarray) {
437 @res = $self->$next($info, @extra);
438 } else {
439 $res = $self->$next($info, @extra);
440 }
441
fd4eb9c2 442 # Make sure master is blessed into the correct class and apply role to it.
443 my $master = $self->master;
444 $master->_determine_driver;
445 Moose::Meta::Class->initialize(ref $master);
cea43436 446
ec0946db 447 DBIx::Class::Storage::DBI::Replicated::WithDSN->meta->apply($master);
cea43436 448
449 # link pool back to master
450 $self->pool->master($master);
0ce2d0d5 451
452 $wantarray ? @res : $res;
b2e4d522 453};
454
26ab719a 455=head1 METHODS
2bf79155 456
26ab719a 457This class defines the following methods.
2bf79155 458
c354902c 459=head2 BUILDARGS
2bf79155 460
faaba25f 461L<DBIx::Class::Schema> when instantiating its storage passed itself as the
2ce6e9a6 462first argument. So we need to massage the arguments a bit so that all the
463bits get put into the correct places.
2bf79155 464
465=cut
466
c354902c 467sub BUILDARGS {
d7a58a29 468 my ($class, $schema, $storage_type_args, @args) = @_;
d4daee7b 469
c354902c 470 return {
6f7344b8 471 schema=>$schema,
472 %$storage_type_args,
473 @args
c354902c 474 }
475}
2bf79155 476
cb6ec758 477=head2 _build_master
2bf79155 478
cb6ec758 479Lazy builder for the L</master> attribute.
2bf79155 480
481=cut
482
cb6ec758 483sub _build_master {
2ce6e9a6 484 my $self = shift @_;
ee356d00 485 my $master = DBIx::Class::Storage::DBI->new($self->schema);
ee356d00 486 $master
106d5f3b 487}
488
26ab719a 489=head2 _build_pool
2bf79155 490
26ab719a 491Lazy builder for the L</pool> attribute.
2bf79155 492
493=cut
494
26ab719a 495sub _build_pool {
64cdad22 496 my $self = shift @_;
497 $self->create_pool(%{$self->pool_args});
2bf79155 498}
499
26ab719a 500=head2 _build_balancer
2bf79155 501
cb6ec758 502Lazy builder for the L</balancer> attribute. This takes a Pool object so that
503the balancer knows which pool it's balancing.
2bf79155 504
505=cut
506
26ab719a 507sub _build_balancer {
64cdad22 508 my $self = shift @_;
509 $self->create_balancer(
6f7344b8 510 pool=>$self->pool,
64cdad22 511 master=>$self->master,
512 %{$self->balancer_args},
513 );
2bf79155 514}
515
cb6ec758 516=head2 _build_write_handler
2bf79155 517
cb6ec758 518Lazy builder for the L</write_handler> attribute. The default is to set this to
519the L</master>.
50336325 520
521=cut
522
cb6ec758 523sub _build_write_handler {
64cdad22 524 return shift->master;
cb6ec758 525}
50336325 526
cb6ec758 527=head2 _build_read_handler
2bf79155 528
cb6ec758 529Lazy builder for the L</read_handler> attribute. The default is to set this to
530the L</balancer>.
2bf79155 531
532=cut
533
cb6ec758 534sub _build_read_handler {
64cdad22 535 return shift->balancer;
cb6ec758 536}
50336325 537
cb6ec758 538=head2 around: connect_replicants
2bf79155 539
cb6ec758 540All calls to connect_replicants needs to have an existing $schema tacked onto
b2e4d522 541top of the args, since L<DBIx::Storage::DBI> needs it, and any C<connect_info>
542options merged with the master, with replicant opts having higher priority.
955a6df6 543
cb6ec758 544=cut
955a6df6 545
b2e4d522 546around connect_replicants => sub {
547 my ($next, $self, @args) = @_;
548
549 for my $r (@args) {
550 $r = [ $r ] unless reftype $r eq 'ARRAY';
551
1a58752c 552 $self->throw_exception('coderef replicant connect_info not supported')
b2e4d522 553 if ref $r->[0] && reftype $r->[0] eq 'CODE';
554
555# any connect_info options?
556 my $i = 0;
557 $i++ while $i < @$r && (reftype($r->[$i])||'') ne 'HASH';
558
6f7344b8 559# make one if none
b2e4d522 560 $r->[$i] = {} unless $r->[$i];
561
562# merge if two hashes
b88b85e7 563 my @hashes = @$r[$i .. $#{$r}];
564
1a58752c 565 $self->throw_exception('invalid connect_info options')
b88b85e7 566 if (grep { reftype($_) eq 'HASH' } @hashes) != @hashes;
567
1a58752c 568 $self->throw_exception('too many hashrefs in connect_info')
b88b85e7 569 if @hashes > 2;
570
282a9a4f 571 my $merge = Hash::Merge->new('LEFT_PRECEDENT');
e666c5fd 572 my %opts = %{ $merge->merge(reverse @hashes) };
b88b85e7 573
574# delete them
b2e4d522 575 splice @$r, $i+1, ($#{$r} - $i), ();
576
0bd8e058 577# make sure master/replicants opts don't clash
578 my %master_opts = %{ $self->_master_connect_info_opts };
579 if (exists $opts{dbh_maker}) {
580 delete @master_opts{qw/dsn user password/};
581 }
582 delete $master_opts{dbh_maker};
583
b2e4d522 584# merge with master
e666c5fd 585 %opts = %{ $merge->merge(\%opts, \%master_opts) };
b2e4d522 586
587# update
588 $r->[$i] = \%opts;
589 }
590
591 $self->$next($self->schema, @args);
955a6df6 592};
2bf79155 593
2bf79155 594=head2 all_storages
595
596Returns an array of of all the connected storage backends. The first element
597in the returned array is the master, and the remainings are each of the
598replicants.
599
600=cut
601
602sub all_storages {
64cdad22 603 my $self = shift @_;
604 return grep {defined $_ && blessed $_} (
605 $self->master,
6412a592 606 values %{ $self->replicants },
64cdad22 607 );
2bf79155 608}
609
c4d3fae2 610=head2 execute_reliably ($coderef, ?@args)
611
612Given a coderef, saves the current state of the L</read_handler>, forces it to
48580715 613use reliable storage (e.g. sets it to the master), executes a coderef and then
c4d3fae2 614restores the original state.
615
616Example:
617
64cdad22 618 my $reliably = sub {
619 my $name = shift @_;
620 $schema->resultset('User')->create({name=>$name});
621 my $user_rs = $schema->resultset('User')->find({name=>$name});
622 return $user_rs;
623 };
c4d3fae2 624
64cdad22 625 my $user_rs = $schema->storage->execute_reliably($reliably, 'John');
c4d3fae2 626
627Use this when you must be certain of your database state, such as when you just
628inserted something and need to get a resultset including it, etc.
629
630=cut
631
632sub execute_reliably {
64cdad22 633 my ($self, $coderef, @args) = @_;
d4daee7b 634
64cdad22 635 unless( ref $coderef eq 'CODE') {
636 $self->throw_exception('Second argument must be a coderef');
637 }
d4daee7b 638
64cdad22 639 ##Get copy of master storage
640 my $master = $self->master;
d4daee7b 641
64cdad22 642 ##Get whatever the current read hander is
643 my $current = $self->read_handler;
d4daee7b 644
64cdad22 645 ##Set the read handler to master
646 $self->read_handler($master);
d4daee7b 647
64cdad22 648 ## do whatever the caller needs
649 my @result;
650 my $want_array = wantarray;
d4daee7b 651
64cdad22 652 eval {
653 if($want_array) {
654 @result = $coderef->(@args);
655 } elsif(defined $want_array) {
656 ($result[0]) = ($coderef->(@args));
ed213e85 657 } else {
64cdad22 658 $coderef->(@args);
6f7344b8 659 }
64cdad22 660 };
d4daee7b 661
64cdad22 662 ##Reset to the original state
6f7344b8 663 $self->read_handler($current);
d4daee7b 664
64cdad22 665 ##Exception testing has to come last, otherwise you might leave the
666 ##read_handler set to master.
d4daee7b 667
64cdad22 668 if($@) {
669 $self->throw_exception("coderef returned an error: $@");
670 } else {
671 return $want_array ? @result : $result[0];
672 }
c4d3fae2 673}
674
cb6ec758 675=head2 set_reliable_storage
676
677Sets the current $schema to be 'reliable', that is all queries, both read and
678write are sent to the master
d4daee7b 679
cb6ec758 680=cut
681
682sub set_reliable_storage {
64cdad22 683 my $self = shift @_;
684 my $schema = $self->schema;
685 my $write_handler = $self->schema->storage->write_handler;
d4daee7b 686
64cdad22 687 $schema->storage->read_handler($write_handler);
cb6ec758 688}
689
690=head2 set_balanced_storage
691
692Sets the current $schema to be use the </balancer> for all reads, while all
48580715 693writes are sent to the master only
d4daee7b 694
cb6ec758 695=cut
696
697sub set_balanced_storage {
64cdad22 698 my $self = shift @_;
699 my $schema = $self->schema;
bd5da369 700 my $balanced_handler = $self->schema->storage->balancer;
d4daee7b 701
bd5da369 702 $schema->storage->read_handler($balanced_handler);
cb6ec758 703}
2bf79155 704
705=head2 connected
706
707Check that the master and at least one of the replicants is connected.
708
709=cut
710
711sub connected {
64cdad22 712 my $self = shift @_;
713 return
714 $self->master->connected &&
715 $self->pool->connected_replicants;
2bf79155 716}
717
2bf79155 718=head2 ensure_connected
719
720Make sure all the storages are connected.
721
722=cut
723
724sub ensure_connected {
64cdad22 725 my $self = shift @_;
726 foreach my $source ($self->all_storages) {
727 $source->ensure_connected(@_);
728 }
2bf79155 729}
730
2bf79155 731=head2 limit_dialect
732
733Set the limit_dialect for all existing storages
734
735=cut
736
737sub limit_dialect {
64cdad22 738 my $self = shift @_;
739 foreach my $source ($self->all_storages) {
740 $source->limit_dialect(@_);
741 }
3fbe08e3 742 return $self->master->quote_char;
2bf79155 743}
744
2bf79155 745=head2 quote_char
746
747Set the quote_char for all existing storages
748
749=cut
750
751sub quote_char {
64cdad22 752 my $self = shift @_;
753 foreach my $source ($self->all_storages) {
754 $source->quote_char(@_);
755 }
3fbe08e3 756 return $self->master->quote_char;
2bf79155 757}
758
2bf79155 759=head2 name_sep
760
761Set the name_sep for all existing storages
762
763=cut
764
765sub name_sep {
64cdad22 766 my $self = shift @_;
767 foreach my $source ($self->all_storages) {
768 $source->name_sep(@_);
769 }
3fbe08e3 770 return $self->master->name_sep;
2bf79155 771}
772
2bf79155 773=head2 set_schema
774
775Set the schema object for all existing storages
776
777=cut
778
779sub set_schema {
64cdad22 780 my $self = shift @_;
781 foreach my $source ($self->all_storages) {
782 $source->set_schema(@_);
783 }
2bf79155 784}
785
2bf79155 786=head2 debug
787
788set a debug flag across all storages
789
790=cut
791
792sub debug {
64cdad22 793 my $self = shift @_;
3fbe08e3 794 if(@_) {
795 foreach my $source ($self->all_storages) {
796 $source->debug(@_);
6f7344b8 797 }
64cdad22 798 }
3fbe08e3 799 return $self->master->debug;
2bf79155 800}
801
2bf79155 802=head2 debugobj
803
cea43436 804set a debug object
2bf79155 805
806=cut
807
808sub debugobj {
64cdad22 809 my $self = shift @_;
cea43436 810 return $self->master->debugobj(@_);
2bf79155 811}
812
2bf79155 813=head2 debugfh
814
cea43436 815set a debugfh object
2bf79155 816
817=cut
818
819sub debugfh {
64cdad22 820 my $self = shift @_;
cea43436 821 return $self->master->debugfh(@_);
2bf79155 822}
823
2bf79155 824=head2 debugcb
825
cea43436 826set a debug callback
2bf79155 827
828=cut
829
830sub debugcb {
64cdad22 831 my $self = shift @_;
cea43436 832 return $self->master->debugcb(@_);
2bf79155 833}
834
2bf79155 835=head2 disconnect
836
837disconnect everything
838
839=cut
840
841sub disconnect {
64cdad22 842 my $self = shift @_;
843 foreach my $source ($self->all_storages) {
844 $source->disconnect(@_);
845 }
2bf79155 846}
847
b2e4d522 848=head2 cursor_class
849
850set cursor class on all storages, or return master's
851
852=cut
853
854sub cursor_class {
855 my ($self, $cursor_class) = @_;
856
857 if ($cursor_class) {
858 $_->cursor_class($cursor_class) for $self->all_storages;
859 }
860 $self->master->cursor_class;
861}
d4daee7b 862
3244fdcc 863=head2 cursor
864
865set cursor class on all storages, or return master's, alias for L</cursor_class>
866above.
867
868=cut
869
870sub cursor {
871 my ($self, $cursor_class) = @_;
872
873 if ($cursor_class) {
874 $_->cursor($cursor_class) for $self->all_storages;
875 }
876 $self->master->cursor;
877}
878
879=head2 unsafe
880
881sets the L<DBIx::Class::Storage::DBI/unsafe> option on all storages or returns
882master's current setting
883
884=cut
885
886sub unsafe {
887 my $self = shift;
888
889 if (@_) {
890 $_->unsafe(@_) for $self->all_storages;
891 }
892
893 return $self->master->unsafe;
894}
895
896=head2 disable_sth_caching
897
898sets the L<DBIx::Class::Storage::DBI/disable_sth_caching> option on all storages
899or returns master's current setting
900
901=cut
902
903sub disable_sth_caching {
904 my $self = shift;
905
906 if (@_) {
907 $_->disable_sth_caching(@_) for $self->all_storages;
908 }
909
910 return $self->master->disable_sth_caching;
911}
912
913=head2 lag_behind_master
914
915returns the highest Replicant L<DBIx::Class::Storage::DBI/lag_behind_master>
916setting
917
918=cut
919
920sub lag_behind_master {
921 my $self = shift;
922
923 return max map $_->lag_behind_master, $self->replicants;
924}
925
926=head2 is_replicating
927
928returns true if all replicants return true for
929L<DBIx::Class::Storage::DBI/is_replicating>
930
931=cut
932
933sub is_replicating {
934 my $self = shift;
935
936 return (grep $_->is_replicating, $self->replicants) == ($self->replicants);
937}
938
939=head2 connect_call_datetime_setup
940
941calls L<DBIx::Class::Storage::DBI/connect_call_datetime_setup> for all storages
942
943=cut
944
945sub connect_call_datetime_setup {
946 my $self = shift;
947 $_->connect_call_datetime_setup for $self->all_storages;
948}
949
950sub _populate_dbh {
951 my $self = shift;
952 $_->_populate_dbh for $self->all_storages;
953}
954
955sub _connect {
956 my $self = shift;
957 $_->_connect for $self->all_storages;
958}
959
960sub _rebless {
961 my $self = shift;
962 $_->_rebless for $self->all_storages;
963}
964
965sub _determine_driver {
966 my $self = shift;
967 $_->_determine_driver for $self->all_storages;
968}
969
970sub _driver_determined {
971 my $self = shift;
972
973 if (@_) {
974 $_->_driver_determined(@_) for $self->all_storages;
975 }
976
977 return $self->master->_driver_determined;
978}
979
980sub _init {
981 my $self = shift;
982
983 $_->_init for $self->all_storages;
984}
985
986sub _run_connection_actions {
987 my $self = shift;
988
989 $_->_run_connection_actions for $self->all_storages;
990}
991
992sub _do_connection_actions {
993 my $self = shift;
994
995 if (@_) {
996 $_->_do_connection_actions(@_) for $self->all_storages;
997 }
998}
999
1000sub connect_call_do_sql {
1001 my $self = shift;
1002 $_->connect_call_do_sql(@_) for $self->all_storages;
1003}
1004
1005sub disconnect_call_do_sql {
1006 my $self = shift;
1007 $_->disconnect_call_do_sql(@_) for $self->all_storages;
1008}
1009
1010sub _seems_connected {
1011 my $self = shift;
1012
1013 return min map $_->_seems_connected, $self->all_storages;
1014}
1015
1016sub _ping {
1017 my $self = shift;
1018
1019 return min map $_->_ping, $self->all_storages;
1020}
1021
7da56142 1022my $numify_ver = sub {
1023 my $ver = shift;
1024 my @numparts = split /\D+/, $ver;
1025 my $format = '%d.' . (join '', ('%05d') x (@numparts - 1));
1026
1027 return sprintf $format, @numparts;
1028};
1029
fecb38cb 1030sub _server_info {
1031 my $self = shift;
1032
1033 if (not $self->_server_info_hash) {
7da56142 1034 my $min_version_info = (
1035 reduce { $a->[0] < $b->[0] ? $a : $b }
1036 map [ $numify_ver->($_->{dbms_version}), $_ ],
1037 map $_->_server_info, $self->all_storages
1038 )->[1];
fecb38cb 1039
1040 $self->_server_info_hash($min_version_info); # on master
1041 }
1042
1043 return $self->_server_info_hash;
1044}
1045
1046sub _get_server_version {
1047 my $self = shift;
1048
1049 return $self->_server_info->{dbms_version};
1050}
1051
e471ab87 1052sub _verify_pid {
1053 my $self = shift;
1054
1055 for my $storage ($self->all_storages) {
1056 $storage->_verify_pid;
1057 }
1058}
1059
1060sub _verify_tid {
1061 my $self = shift;
1062
1063 for my $storage ($self->all_storages) {
1064 $storage->_verify_tid;
1065 }
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;