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