BEGIN {
die('The following modules are required for Replication ' . DBIx::Class::Optional::Dependencies->req_missing_for ('replicated') . "\n" )
unless DBIx::Class::Optional::Dependencies->req_ok_for ('replicated');
}
has 'schema' => (
is=>'rw',
isa=>DBICSchema,
weak_ref=>1,
required=>1,
);
has 'pool_type' => (
is=>'rw',
isa=>ClassName,
default=>'DBIx::Class::Storage::DBI::Replicated::Pool',
handles=>{
'create_pool' => 'new',
},
);
has 'pool_args' => (
is=>'rw',
isa=>HashRef,
lazy=>1,
default=>sub { {} },
);
has 'balancer_type' => (
is=>'rw',
isa=>BalancerClassNamePart,
coerce=>1,
required=>1,
default=> 'DBIx::Class::Storage::DBI::Replicated::Balancer::First',
handles=>{
'create_balancer' => 'new',
},
);
has 'balancer_args' => (
is=>'rw',
isa=>HashRef,
lazy=>1,
required=>1,
default=>sub { {} },
);
has 'pool' => (
is=>'ro',
isa=>'DBIx::Class::Storage::DBI::Replicated::Pool',
lazy_build=>1,
handles=>[qw/
connect_replicants
replicants
has_replicants
/],
);
has 'balancer' => (
is=>'rw',
isa=>'DBIx::Class::Storage::DBI::Replicated::Balancer',
lazy_build=>1,
handles=>[qw/auto_validate_every/],
);
has 'master' => (
is=> 'ro',
isa=>DBICStorageDBI,
lazy_build=>1,
);
my $method_dispatch = {
writer => [qw/
on_connect_do
on_disconnect_do
on_connect_call
on_disconnect_call
connect_info
_connect_info
throw_exception
sql_maker
sqlt_type
create_ddl_dir
deployment_statements
datetime_parser
datetime_parser_type
build_datetime_parser
last_insert_id
insert
insert_bulk
update
delete
dbh
txn_begin
txn_do
txn_commit
txn_rollback
txn_scope_guard
_exec_txn_rollback
_exec_txn_begin
_exec_txn_commit
deploy
with_deferred_fk_checks
dbh_do
_prep_for_execute
is_datatype_numeric
_count_select
svp_rollback
svp_begin
svp_release
relname_to_table_alias
_dbh_last_insert_id
_default_dbi_connect_attributes
_dbi_connect_info
_dbic_connect_attributes
auto_savepoint
_query_start
_query_end
_format_for_trace
_dbi_attrs_for_bind
bind_attribute_by_data_type
transaction_depth
_dbh
_select_args
_dbh_execute_for_fetch
_sql_maker
_dbh_execute_inserts_with_no_binds
_select_args_to_query
_gen_sql_bind
_svp_generate_name
_normalize_connect_info
_parse_connect_do
savepoints
_sql_maker_opts
_conn_pid
_dbh_autocommit
_native_data_type
_get_dbh
sql_maker_class
_execute
_do_query
_dbh_execute
/, Class::MOP::Class->initialize('DBIx::Class::Storage::DBIHacks')->get_method_list ],
reader => [qw/
select
select_single
columns_info_for
_dbh_columns_info_for
_select
/],
unimplemented => [qw/
_arm_global_destructor
_verify_pid
get_use_dbms_capability
set_use_dbms_capability
get_dbms_capability
set_dbms_capability
_dbh_details
_dbh_get_info
_determine_connector_driver
_describe_connection
_warn_undetermined_driver
sql_limit_dialect
sql_quote_char
sql_name_sep
_prefetch_autovalues
_perform_autoinc_retrieval
_autoinc_supplied_for_op
_resolve_bindattrs
_max_column_bytesize
_is_lob_type
_is_binary_lob_type
_is_binary_type
_is_text_lob_type
_prepare_sth
_bind_sth_params
/,(
grep
{ $_ =~ /^ _ (?: use | supports | determine_supports ) _ /x }
( Class::MOP::Class->initialize('DBIx::Class::Storage::DBI')->get_all_method_names )
)],
};
if (DBIx::Class::_ENV_::DBICTEST) {
my $seen;
for my $type (keys %$method_dispatch) {
for (@{$method_dispatch->{$type}}) {
push @{$seen->{$_}}, $type;
}
}
if (my @dupes = grep { @{$seen->{$_}} > 1 } keys %$seen) {
die(join "\n", '',
'The following methods show up multiple times in ::Storage::DBI::Replicated handlers:',
(map { "$_: " . (join ', ', @{$seen->{$_}}) } sort @dupes),
'',
);
}
if (my @cant = grep { ! DBIx::Class::Storage::DBI->can($_) } keys %$seen) {
die(join "\n", '',
'::Storage::DBI::Replicated specifies handling of the following *NON EXISTING* ::Storage::DBI methods:',
@cant,
'',
);
}
}
for my $method (@{$method_dispatch->{unimplemented}}) {
__PACKAGE__->meta->add_method($method, sub {
my $self = shift;
$self->throw_exception("$method() must not be called on ".(blessed $self).' objects');
});
}
has 'read_handler' => (
is=>'rw',
isa=>Object,
lazy_build=>1,
handles=>$method_dispatch->{reader},
);
has 'write_handler' => (
is=>'ro',
isa=>Object,
lazy_build=>1,
handles=>$method_dispatch->{writer},
);
has _master_connect_info_opts =>
(is => 'rw', isa => HashRef, default => sub { {} });
around connect_info => sub {
my ($next, $self, $info, @extra) = @_;
$self->throw_exception(
'connect_info can not be retrieved from a replicated storage - '
. 'accessor must be called on a specific pool instance'
) unless defined $info;
my $merge = Hash::Merge->new('LEFT_PRECEDENT');
my %opts;
for my $arg (@$info) {
next unless (reftype($arg)||'') eq 'HASH';
%opts = %{ $merge->merge($arg, \%opts) };
}
delete $opts{dsn};
if (@opts{qw/pool_type pool_args/}) {
$self->pool_type(delete $opts{pool_type})
if $opts{pool_type};
$self->pool_args(
$merge->merge((delete $opts{pool_args} || {}), $self->pool_args)
);
$self->clear_pool;
}
if (@opts{qw/balancer_type balancer_args/}) {
$self->balancer_type(delete $opts{balancer_type})
if $opts{balancer_type};
$self->balancer_args(
$merge->merge((delete $opts{balancer_args} || {}), $self->balancer_args)
);
$self->balancer($self->_build_balancer)
if $self->balancer;
}
$self->_master_connect_info_opts(\%opts);
return preserve_context {
$self->$next($info, @extra);
} after => sub {
my $master = $self->master;
$master->_determine_driver;
Moose::Meta::Class->initialize(ref $master);
DBIx::Class::Storage::DBI::Replicated::WithDSN->meta->apply($master);
$self->pool->master($master);
};
};
sub BUILDARGS {
my ($class, $schema, $storage_type_args, @args) = @_;
return {
schema=>$schema,
%$storage_type_args,
@args
}
}
sub _build_master {
my $self = shift @_;
my $master = DBIx::Class::Storage::DBI->new($self->schema);
$master
}
sub _build_pool {
my $self = shift @_;
$self->create_pool(%{$self->pool_args});
}
sub _build_balancer {
my $self = shift @_;
$self->create_balancer(
pool=>$self->pool,
master=>$self->master,
%{$self->balancer_args},
);
}
sub _build_write_handler {
return shift->master;
}
sub _build_read_handler {
return shift->balancer;
}
around connect_replicants => sub {
my ($next, $self, @args) = @_;
for my $r (@args) {
$r = [ $r ] unless reftype $r eq 'ARRAY';
$self->throw_exception('coderef replicant connect_info not supported')
if ref $r->[0] && reftype $r->[0] eq 'CODE';
my $i = 0;
$i++ while $i < @$r && (reftype($r->[$i])||'') ne 'HASH';
$r->[$i] = {} unless $r->[$i];
my @hashes = @$r[$i .. $
$self->throw_exception('invalid connect_info options')
if (grep { reftype($_) eq 'HASH' } @hashes) != @hashes;
$self->throw_exception('too many hashrefs in connect_info')
if @hashes > 2;
my $merge = Hash::Merge->new('LEFT_PRECEDENT');
my %opts = %{ $merge->merge(reverse @hashes) };
splice @$r, $i+1, ($
my %master_opts = %{ $self->_master_connect_info_opts };
if (exists $opts{dbh_maker}) {
delete @master_opts{qw/dsn user password/};
}
delete $master_opts{dbh_maker};
%opts = %{ $merge->merge(\%opts, \%master_opts) };
$r->[$i] = \%opts;
}
$self->$next($self->schema, @args);
};
sub all_storages {
my $self = shift @_;
return grep {defined $_ && blessed $_} (
$self->master,
values %{ $self->replicants },
);
}
sub execute_reliably {
my $self = shift;
my $coderef = shift;
unless( ref $coderef eq 'CODE') {
$self->throw_exception('Second argument must be a coderef');
}
local $self->{read_handler} = $self->master;
my $args = \@_;
return try {
$coderef->(@$args);
} catch {
$self->throw_exception("coderef returned an error: $_");
};
}
sub set_reliable_storage {
my $self = shift @_;
my $schema = $self->schema;
my $write_handler = $self->schema->storage->write_handler;
$schema->storage->read_handler($write_handler);
}
sub set_balanced_storage {
my $self = shift @_;
my $schema = $self->schema;
my $balanced_handler = $self->schema->storage->balancer;
$schema->storage->read_handler($balanced_handler);
}
sub connected {
my $self = shift @_;
return
$self->master->connected &&
$self->pool->connected_replicants;
}
sub ensure_connected {
my $self = shift @_;
foreach my $source ($self->all_storages) {
$source->ensure_connected(@_);
}
}
sub limit_dialect {
my $self = shift @_;
foreach my $source ($self->all_storages) {
$source->limit_dialect(@_);
}
return $self->master->limit_dialect;
}
sub quote_char {
my $self = shift @_;
foreach my $source ($self->all_storages) {
$source->quote_char(@_);
}
return $self->master->quote_char;
}
sub name_sep {
my $self = shift @_;
foreach my $source ($self->all_storages) {
$source->name_sep(@_);
}
return $self->master->name_sep;
}
sub set_schema {
my $self = shift @_;
foreach my $source ($self->all_storages) {
$source->set_schema(@_);
}
}
sub debug {
my $self = shift @_;
if(@_) {
foreach my $source ($self->all_storages) {
$source->debug(@_);
}
}
return $self->master->debug;
}
sub debugobj {
my $self = shift @_;
return $self->master->debugobj(@_);
}
sub debugfh {
my $self = shift @_;
return $self->master->debugfh(@_);
}
sub debugcb {
my $self = shift @_;
return $self->master->debugcb(@_);
}
sub disconnect {
my $self = shift @_;
foreach my $source ($self->all_storages) {
$source->disconnect(@_);
}
}
sub cursor_class {
my ($self, $cursor_class) = @_;
if ($cursor_class) {
$_->cursor_class($cursor_class) for $self->all_storages;
}
$self->master->cursor_class;
}
sub cursor {
my ($self, $cursor_class) = @_;
if ($cursor_class) {
$_->cursor($cursor_class) for $self->all_storages;
}
$self->master->cursor;
}
sub unsafe {
my $self = shift;
if (@_) {
$_->unsafe(@_) for $self->all_storages;
}
return $self->master->unsafe;
}
sub disable_sth_caching {
my $self = shift;
if (@_) {
$_->disable_sth_caching(@_) for $self->all_storages;
}
return $self->master->disable_sth_caching;
}
sub lag_behind_master {
my $self = shift;
return max map $_->lag_behind_master, $self->replicants;
}
sub is_replicating {
my $self = shift;
return (grep $_->is_replicating, $self->replicants) == ($self->replicants);
}
sub connect_call_datetime_setup {
my $self = shift;
$_->connect_call_datetime_setup for $self->all_storages;
}
sub _populate_dbh {
my $self = shift;
$_->_populate_dbh for $self->all_storages;
}
sub _connect {
my $self = shift;
$_->_connect for $self->all_storages;
}
sub _rebless {
my $self = shift;
$_->_rebless for $self->all_storages;
}
sub _determine_driver {
my $self = shift;
$_->_determine_driver for $self->all_storages;
}
sub _driver_determined {
my $self = shift;
if (@_) {
$_->_driver_determined(@_) for $self->all_storages;
}
return $self->master->_driver_determined;
}
sub _init {
my $self = shift;
$_->_init for $self->all_storages;
}
sub _run_connection_actions {
my $self = shift;
$_->_run_connection_actions for $self->all_storages;
}
sub _do_connection_actions {
my $self = shift;
if (@_) {
$_->_do_connection_actions(@_) for $self->all_storages;
}
}
sub connect_call_do_sql {
my $self = shift;
$_->connect_call_do_sql(@_) for $self->all_storages;
}
sub disconnect_call_do_sql {
my $self = shift;
$_->disconnect_call_do_sql(@_) for $self->all_storages;
}
sub _seems_connected {
my $self = shift;
return min map $_->_seems_connected, $self->all_storages;
}
sub _ping {
my $self = shift;
return min map $_->_ping, $self->all_storages;
}
my $numify_ver = sub {
my $ver = shift;
my @numparts = split /\D+/, $ver;
my $format = '%d.' . (join '', ('%06d') x (@numparts - 1));
return sprintf $format, @numparts;
};
sub _server_info {
my $self = shift;
if (not $self->_dbh_details->{info}) {
$self->_dbh_details->{info} = (
reduce { $a->[0] < $b->[0] ? $a : $b }
map [ $numify_ver->($_->{dbms_version}), $_ ],
map $_->_server_info, $self->all_storages
)->[1];
}
return $self->next::method;
}
sub _get_server_version {
my $self = shift;
return $self->_server_info->{dbms_version};
}
__PACKAGE__->meta->make_immutable;
1;