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