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