view release on metacpan or  search on metacpan

lib/DBIO/Storage/Async.pm  view on Meta::CPAN

package DBIO::Storage::Async;
# ABSTRACT: Base class for async storage implementations

use strict;
use warnings;
use base 'DBIO::Storage';

use Carp 'croak';
use Scalar::Util qw(blessed weaken);
use DBIO::Storage::Async::TransactionContext;
use DBIO::SQL::Util ();
use namespace::clean;


sub new {
  my ($class, $schema, $args) = @_;
  my $self = bless {
    schema             => $schema,
    pool               => undef,
    connect_info       => undef,
    sql_maker          => undef,
    _conninfo_provider => undef,
    debug              => $ENV{DBIO_TRACE} || 0,
    debugobj           => undef,
  }, $class;
  weaken($self->{schema}) if ref $self->{schema};
  return $self;
}


sub future_class {
  croak 'Subclass must override future_class';
}


sub pool {
  croak 'Subclass must override pool';
}


sub _is_access_broker_connect_info {
  my ($self, $info) = @_;

  return 0 unless ref $info eq 'ARRAY' && @$info == 1;
  return 0 unless blessed($info->[0]);

  return $info->[0]->isa('DBIO::AccessBroker');
}


sub _setup_access_broker {
  my ($self, $broker) = @_;

  $self->set_access_broker($broker, 'write');
  $self->{_conninfo_provider} = sub {
    return $self->_async_broker_conninfo($self->access_broker_mode);
  };

  return $self;
}


sub _clear_access_broker {
  my $self = shift;

  $self->clear_access_broker;
  $self->{_conninfo_provider} = undef;

  return $self;
}


sub _conninfo_provider { $_[0]->{_conninfo_provider} }


sub _async_broker_conninfo {
  croak 'Subclass must override _async_broker_conninfo to consume an AccessBroker';
}


sub connect_info {
  my ($self, $info) = @_;
  if ($info) {
    # karr #66: strip DBIO-private attrs before the DB-specific reshape seam,
    # so _normalize_conninfo (and DBI->connect) only see real DBD attributes.
    $info = $self->_strip_private_connect_attrs($info);

lib/DBIO/Storage/Async.pm  view on Meta::CPAN

sub _strip_private_connect_attrs {
  my ($self, $info) = @_;

  return $info
    if ref $info ne 'ARRAY'
    or $self->_is_access_broker_connect_info($info);

  my @args = @$info;   # shallow copy; the attr hashes below are copied before mutation

  my @attr_slots;
  if (ref $args[0] eq 'CODE') {         # coderef + optional \%extra_attributes
    push @attr_slots, \$args[1] if ref $args[1] eq 'HASH';
  }
  elsif (ref $args[0] eq 'HASH') {      # single \%hash (Catalyst-style config)
    push @attr_slots, \$args[0];
  }
  else {                                # dsn/user/pass + \%attrs [+ \%extra_attrs]
    push @attr_slots, \$args[$_] for grep ref $args[$_] eq 'HASH', 3, 4;
  }

  my @storage_opt_names = ($self->_dbio_storage_option_names, 'cursor_class');
  my %storage_options;

  for my $slot (@attr_slots) {
    my %attrs = %{ $$slot };   # copy before mutating

    # storage_options + cursor_class -> keep for the storage, out of
    # DBI->connect. Same sink the sync path uses (DBIO::Storage::DBI stashes
    # these as storage_options). Held for karr #68 (pool on_connect seam:
    # on_connect_do / on_connect_call) to consume; no further processing here.
    $storage_options{$_} = delete $attrs{$_}
      for grep exists $attrs{$_}, @storage_opt_names;

    # quote_char/name_sep/quote_names -> sql_maker config, not DBD attrs. The
    # async sql_maker path (_sql_maker_args) does not source quoting from
    # connect_info yet (driver adapters set quote_char statically), so dropping
    # them is correct for now. Their proper future sink is the backend's
    # sql_maker -- see the TODO block in t/storage/async_connect_info_leak.t so
    # a later async-quoting feature does not pull them back into raw DBD attrs.
    delete @attrs{ $self->_dbio_sql_maker_option_names };

    # async -> mode selector; the backend already knows its own mode, so it must
    # never reach DBI->connect. ignore_version -> consumed by the
    # DeploymentHandler version-storage component, never by the DBD.
    delete @attrs{qw( async ignore_version )};

    $$slot = \%attrs;
  }

  # Slot the storage options aside for karr #68; nothing consumes them yet.
  $self->{_storage_options} = \%storage_options;

  return \@args;
}


sub _owner_storage {
  my $self = shift;
  if (@_) {
    $self->{_owner_storage} = $_[0];
    weaken($self->{_owner_storage}) if ref $self->{_owner_storage};
  }
  return $self->{_owner_storage};
}


sub _setup_pool_connection {
  my ($self, $conn) = @_;

  my $owner = $self->_owner_storage or return;
  return unless $owner->can('_run_pool_connect_actions');

  $owner->_run_pool_connect_actions(sub {
    $self->_run_pool_connect_statement($conn, @_);
  });

  return;
}


sub _teardown_pool_connection {
  my ($self, $conn) = @_;

  my $owner = $self->_owner_storage or return;
  return unless $owner->can('_run_pool_disconnect_actions');

  $owner->_run_pool_disconnect_actions(sub {
    $self->_run_pool_connect_statement($conn, @_);
  });

  return;
}


sub _run_pool_connect_statement {
  my ($self, $conn, $sql, @args) = @_;
  my $dbh = $self->_pool_connect_dbh($conn);
  $dbh->do($sql, @args);
  return;
}

# Extract a blocking-do-capable DBI handle from a pool connection value. The two
# shapes the DBD-based Model-B backends use are a bare $dbh and the future_io
# { dbh => $dbh } wrapper (DBIO::PostgreSQL::Storage::Async::_create_pool_connection
# and friends). Anything else is a native backend that must override
# _run_pool_connect_statement.
sub _pool_connect_dbh {
  my ($self, $conn) = @_;

  return $conn
    if blessed($conn) && $conn->can('do');

  return $conn->{dbh}
    if ref $conn eq 'HASH' && blessed($conn->{dbh}) && $conn->{dbh}->can('do');

  croak 'Cannot run pool connect action: connection is neither a do-capable '
      . 'DBI handle nor a { dbh => $dbh } wrapper -- override '
      . '_run_pool_connect_statement for this backend';
}