lib/DBIx/Class/Storage/DBI/Replicated.pm

   1 package DBIx::Class::Storage::DBI::Replicated;
   2
   3 use Moose;
   4 use DBIx::Class::Storage::DBI;
   5 use DBIx::Class::Storage::DBI::Replicated::Pool;
   6 use DBIx::Class::Storage::DBI::Replicated::Balancer;
   7 use Scalar::Util qw(blessed);
   8
   9 extends 'DBIx::Class::Storage::DBI', 'Moose::Object';
  10
  11 =head1 NAME
  12
  13 DBIx::Class::Storage::DBI::Replicated - ALPHA Replicated database support
  14
  15 =head1 SYNOPSIS
  16
  17 The Following example shows how to change an existing $schema to a replicated
  18 storage type, add some replicated (readonly) databases, and perform reporting
  19 tasks.
  20
  21     ## Change storage_type in your schema class
  22     $schema->storage_type( ['::DBI::Replicated', {balancer=>'::Random'}] );
  23
  24     ## Add some slaves.  Basically this is an array of arrayrefs, where each
  25     ## arrayref is database connect information
  26
  27     $schema->storage->connect_replicants(
  28         [$dsn1, $user, $pass, \%opts],
  29         [$dsn1, $user, $pass, \%opts],
  30         [$dsn1, $user, $pass, \%opts],
  31     );
  32
  33 =head1 DESCRIPTION
  34
  35 Warning: This class is marked ALPHA.  We are using this in development and have
  36 some basic test coverage but the code hasn't yet been stressed by a variety
  37 of databases.  Individual DB's may have quirks we are not aware of.  Please
  38 use this in development and pass along your experiences/bug fixes.
  39
  40 This class implements replicated data store for DBI. Currently you can define
  41 one master and numerous slave database connections. All write-type queries
  42 (INSERT, UPDATE, DELETE and even LAST_INSERT_ID) are routed to master
  43 database, all read-type queries (SELECTs) go to the slave database.
  44
  45 Basically, any method request that L<DBIx::Class::Storage::DBI> would normally
  46 handle gets delegated to one of the two attributes: L</read_handler> or to
  47 L</write_handler>.  Additionally, some methods need to be distributed
  48 to all existing storages.  This way our storage class is a drop in replacement
  49 for L<DBIx::Class::Storage::DBI>.
  50
  51 Read traffic is spread across the replicants (slaves) occuring to a user
  52 selected algorithm.  The default algorithm is random weighted.
  53
  54 =head1 NOTES
  55
  56 The consistancy betweeen master and replicants is database specific.  The Pool
  57 gives you a method to validate it's replicants, removing and replacing them
  58 when they fail/pass predefined criteria.  It is recommened that your application
  59 define two schemas, one using the replicated storage and another that just
  60 connects to the master.
  61
  62 =head1 ATTRIBUTES
  63
  64 This class defines the following attributes.
  65
  66 =head2 pool_type
  67
  68 Contains the classname which will instantiate the L</pool> object.  Defaults
  69 to: L<DBIx::Class::Storage::DBI::Replicated::Pool>.
  70
  71 =cut
  72
  73 has 'pool_type' => (
  74     is=>'ro',
  75     isa=>'ClassName',
  76     lazy_build=>1,
  77     handles=>{
  78         'create_pool' => 'new',
  79     },
  80 );
  81
  82 =head2 pool_args
  83
  84 Contains a hashref of initialized information to pass to the Balancer object.
  85 See L<DBIx::Class::Storage::Replicated::Pool> for available arguments.
  86
  87 =cut
  88
  89 has 'pool_args' => (
  90     is=>'ro',
  91     isa=>'HashRef',
  92     lazy=>1,
  93     required=>1,
  94     default=>sub { {} },
  95 );
  96
  97
  98 =head2 balancer_type
  99
 100 The replication pool requires a balance class to provider the methods for
 101 choose how to spread the query load across each replicant in the pool.
 102
 103 =cut
 104
 105 has 'balancer_type' => (
 106     is=>'ro',
 107     isa=>'ClassName',
 108     lazy_build=>1,
 109     handles=>{
 110         'create_balancer' => 'new',
 111     },
 112 );
 113
 114 =head2 balancer_args
 115
 116 Contains a hashref of initialized information to pass to the Balancer object.
 117 See L<DBIx::Class::Storage::Replicated::Balancer> for available arguments.
 118
 119 =cut
 120
 121 has 'balancer_args' => (
 122     is=>'ro',
 123     isa=>'HashRef',
 124     lazy=>1,
 125     required=>1,
 126     default=>sub { {} },
 127 );
 128
 129 =head2 pool
 130
 131 Is a <DBIx::Class::Storage::DBI::Replicated::Pool> or derived class.  This is a
 132 container class for one or more replicated databases.
 133
 134 =cut
 135
 136 has 'pool' => (
 137     is=>'ro',
 138     isa=>'DBIx::Class::Storage::DBI::Replicated::Pool',
 139     lazy_build=>1,
 140     handles=>[qw/
 141         connect_replicants
 142         replicants
 143         has_replicants
 144     /],
 145 );
 146
 147 =head2 balancer
 148
 149 Is a <DBIx::Class::Storage::DBI::Replicated::Balancer> or derived class.  This
 150 is a class that takes a pool (<DBIx::Class::Storage::DBI::Replicated::Pool>)
 151
 152 =cut
 153
 154 has 'balancer' => (
 155     is=>'ro',
 156     isa=>'DBIx::Class::Storage::DBI::Replicated::Balancer',
 157     lazy_build=>1,
 158     handles=>[qw/auto_validate_every/],
 159 );
 160
 161 =head2 master
 162
 163 The master defines the canonical state for a pool of connected databases.  All
 164 the replicants are expected to match this databases state.  Thus, in a classic
 165 Master / Slaves distributed system, all the slaves are expected to replicate
 166 the Master's state as quick as possible.  This is the only database in the
 167 pool of databases that is allowed to handle write traffic.
 168
 169 =cut
 170
 171 has 'master' => (
 172     is=> 'ro',
 173     isa=>'DBIx::Class::Storage::DBI',
 174     lazy_build=>1,
 175 );
 176
 177 =head1 ATTRIBUTES IMPLEMENTING THE DBIx::Storage::DBI INTERFACE
 178
 179 The following methods are delegated all the methods required for the
 180 L<DBIx::Class::Storage::DBI> interface.
 181
 182 =head2 read_handler
 183
 184 Defines an object that implements the read side of L<BIx::Class::Storage::DBI>.
 185
 186 =cut
 187
 188 has 'read_handler' => (
 189     is=>'rw',
 190     isa=>'Object',
 191     lazy_build=>1,
 192     handles=>[qw/
 193         select
 194         select_single
 195         columns_info_for
 196     /],
 197 );
 198
 199 =head2 write_handler
 200
 201 Defines an object that implements the write side of L<BIx::Class::Storage::DBI>.
 202
 203 =cut
 204
 205 has 'write_handler' => (
 206     is=>'ro',
 207     isa=>'Object',
 208     lazy_build=>1,
 209     lazy_build=>1,
 210     handles=>[qw/
 211         on_connect_do
 212         on_disconnect_do
 213         connect_info
 214         throw_exception
 215         sql_maker
 216         sqlt_type
 217         create_ddl_dir
 218         deployment_statements
 219         datetime_parser
 220         datetime_parser_type
 221         last_insert_id
 222         insert
 223         insert_bulk
 224         update
 225         delete
 226         dbh
 227         txn_do
 228         txn_commit
 229         txn_rollback
 230         sth
 231         deploy
 232         schema
 233     /],
 234 );
 235
 236 =head1 METHODS
 237
 238 This class defines the following methods.
 239
 240 =head2 new
 241
 242 L<DBIx::Class::Schema> when instantiating it's storage passed itself as the
 243 first argument.  We need to invoke L</new> on the underlying parent class, make
 244 sure we properly give it a L<Moose> meta class, and then correctly instantiate
 245 our attributes.  Basically we pass on whatever the schema has in it's class
 246 data for 'storage_type_args' to our replicated storage type.
 247
 248 =cut
 249
 250 sub new {
 251     my $class = shift @_;
 252     my $schema = shift @_;
 253     my $storage_type_args = shift @_;
 254     my $obj = $class->SUPER::new($schema, $storage_type_args, @_);
 255
 256     ## Hate to do it this way, but can't seem to get advice on the attribute working right
 257     ## maybe we can do a type and coercion for it.
 258     if( $storage_type_args->{balancer_type} && $storage_type_args->{balancer_type}=~m/^::/) {
 259         $storage_type_args->{balancer_type} = 'DBIx::Class::Storage::DBI::Replicated::Balancer'.$storage_type_args->{balancer_type};
 260         eval "require $storage_type_args->{balancer_type}";
 261     }
 262
 263     return $class->meta->new_object(
 264         __INSTANCE__ => $obj,
 265         %$storage_type_args,
 266         @_,
 267     );
 268 }
 269
 270 =head2 _build_master
 271
 272 Lazy builder for the L</master> attribute.
 273
 274 =cut
 275
 276 sub _build_master {
 277         DBIx::Class::Storage::DBI->new;
 278 }
 279
 280 =head2 _build_pool_type
 281
 282 Lazy builder for the L</pool_type> attribute.
 283
 284 =cut
 285
 286 sub _build_pool_type {
 287     return 'DBIx::Class::Storage::DBI::Replicated::Pool';
 288 }
 289
 290 =head2 _build_pool
 291
 292 Lazy builder for the L</pool> attribute.
 293
 294 =cut
 295
 296 sub _build_pool {
 297         my $self = shift @_;
 298     $self->create_pool(%{$self->pool_args});
 299 }
 300
 301 =head2 _build_balancer_type
 302
 303 Lazy builder for the L</balancer_type> attribute.
 304
 305 =cut
 306
 307 sub _build_balancer_type {
 308     return 'DBIx::Class::Storage::DBI::Replicated::Balancer::First';
 309 }
 310
 311 =head2 _build_balancer
 312
 313 Lazy builder for the L</balancer> attribute.  This takes a Pool object so that
 314 the balancer knows which pool it's balancing.
 315
 316 =cut
 317
 318 sub _build_balancer {
 319     my $self = shift @_;
 320     $self->create_balancer(
 321         pool=>$self->pool,
 322         master=>$self->master,
 323         %{$self->balancer_args},);
 324 }
 325
 326 =head2 _build_write_handler
 327
 328 Lazy builder for the L</write_handler> attribute.  The default is to set this to
 329 the L</master>.
 330
 331 =cut
 332
 333 sub _build_write_handler {
 334     return shift->master;
 335 }
 336
 337 =head2 _build_read_handler
 338
 339 Lazy builder for the L</read_handler> attribute.  The default is to set this to
 340 the L</balancer>.
 341
 342 =cut
 343
 344 sub _build_read_handler {
 345     return shift->balancer;
 346 }
 347
 348 =head2 around: connect_replicants
 349
 350 All calls to connect_replicants needs to have an existing $schema tacked onto
 351 top of the args, since L<DBIx::Storage::DBI> needs it.
 352
 353 =cut
 354
 355 around 'connect_replicants' => sub {
 356         my ($method, $self, @args) = @_;
 357         $self->$method($self->schema, @args);
 358 };
 359
 360 =head2 all_storages
 361
 362 Returns an array of of all the connected storage backends.  The first element
 363 in the returned array is the master, and the remainings are each of the
 364 replicants.
 365
 366 =cut
 367
 368 sub all_storages {
 369         my $self = shift @_;
 370
 371         return grep {defined $_ && blessed $_} (
 372            $self->master,
 373            $self->replicants,
 374         );
 375 }
 376
 377 =head2 set_reliable_storage
 378
 379 Sets the current $schema to be 'reliable', that is all queries, both read and
 380 write are sent to the master
 381
 382 =cut
 383
 384 sub set_reliable_storage {
 385         my $self = shift @_;
 386         my $schema = $self->schema;
 387         my $write_handler = $self->schema->storage->write_handler;
 388
 389         $schema->storage->read_handler($write_handler);
 390 }
 391
 392 =head2 set_balanced_storage
 393
 394 Sets the current $schema to be use the </balancer> for all reads, while all
 395 writea are sent to the master only
 396
 397 =cut
 398
 399 sub set_balanced_storage {
 400     my $self = shift @_;
 401     my $schema = $self->schema;
 402     my $write_handler = $self->schema->storage->balancer;
 403
 404     $schema->storage->read_handler($write_handler);
 405 }
 406
 407 =head2 connected
 408
 409 Check that the master and at least one of the replicants is connected.
 410
 411 =cut
 412
 413 sub connected {
 414         my $self = shift @_;
 415
 416         return
 417            $self->master->connected &&
 418            $self->pool->connected_replicants;
 419 }
 420
 421 =head2 ensure_connected
 422
 423 Make sure all the storages are connected.
 424
 425 =cut
 426
 427 sub ensure_connected {
 428     my $self = shift @_;
 429     foreach my $source ($self->all_storages) {
 430         $source->ensure_connected(@_);
 431     }
 432 }
 433
 434 =head2 limit_dialect
 435
 436 Set the limit_dialect for all existing storages
 437
 438 =cut
 439
 440 sub limit_dialect {
 441     my $self = shift @_;
 442     foreach my $source ($self->all_storages) {
 443         $source->limit_dialect(@_);
 444     }
 445 }
 446
 447 =head2 quote_char
 448
 449 Set the quote_char for all existing storages
 450
 451 =cut
 452
 453 sub quote_char {
 454     my $self = shift @_;
 455     foreach my $source ($self->all_storages) {
 456         $source->quote_char(@_);
 457     }
 458 }
 459
 460 =head2 name_sep
 461
 462 Set the name_sep for all existing storages
 463
 464 =cut
 465
 466 sub name_sep {
 467     my $self = shift @_;
 468     foreach my $source ($self->all_storages) {
 469         $source->name_sep(@_);
 470     }
 471 }
 472
 473 =head2 set_schema
 474
 475 Set the schema object for all existing storages
 476
 477 =cut
 478
 479 sub set_schema {
 480         my $self = shift @_;
 481         foreach my $source ($self->all_storages) {
 482                 $source->set_schema(@_);
 483         }
 484 }
 485
 486 =head2 debug
 487
 488 set a debug flag across all storages
 489
 490 =cut
 491
 492 sub debug {
 493     my $self = shift @_;
 494     foreach my $source ($self->all_storages) {
 495         $source->debug(@_);
 496     }
 497 }
 498
 499 =head2 debugobj
 500
 501 set a debug object across all storages
 502
 503 =cut
 504
 505 sub debugobj {
 506     my $self = shift @_;
 507     foreach my $source ($self->all_storages) {
 508         $source->debugobj(@_);
 509     }
 510 }
 511
 512 =head2 debugfh
 513
 514 set a debugfh object across all storages
 515
 516 =cut
 517
 518 sub debugfh {
 519     my $self = shift @_;
 520     foreach my $source ($self->all_storages) {
 521         $source->debugfh(@_);
 522     }
 523 }
 524
 525 =head2 debugcb
 526
 527 set a debug callback across all storages
 528
 529 =cut
 530
 531 sub debugcb {
 532     my $self = shift @_;
 533     foreach my $source ($self->all_storages) {
 534         $source->debugcb(@_);
 535     }
 536 }
 537
 538 =head2 disconnect
 539
 540 disconnect everything
 541
 542 =cut
 543
 544 sub disconnect {
 545     my $self = shift @_;
 546     foreach my $source ($self->all_storages) {
 547         $source->disconnect(@_);
 548     }
 549 }
 550
 551 =head1 AUTHOR
 552
 553 Norbert Csongrádi <bert@cpan.org>
 554
 555 Peter Siklósi <einon@einon.hu>
 556
 557 John Napiorkowski <john.napiorkowski@takkle.com>
 558
 559 =head1 LICENSE
 560
 561 You may distribute this code under the same terms as Perl itself.
 562
 563 =cut
 564
 565 1;