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