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