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