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