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