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