view release on metacpan or  search on metacpan

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

}

# 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';
}


sub sql_maker {
  my $self = shift;
  $self->{sql_maker} ||= do {
    my $class = $self->sql_maker_class;
    $class->new($self->_sql_maker_args);
  };
}


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


sub _sql_maker_args { () }


sub transport_capabilities { () }


sub select_async {
  my $self = shift;
  return $self->_run_crud_pooled('select', @_);
}


sub select_single_async {
  my $self = shift;
  return $self->_run_crud_pooled('select_single', @_);
}


sub insert_async {
  my $self = shift;
  return $self->_run_crud_pooled('insert', @_);
}


sub update_async {
  my $self = shift;
  return $self->_run_crud_pooled('update', @_);
}


sub delete_async {
  my $self = shift;
  return $self->_run_crud_pooled('delete', @_);
}

# Prepare a SELECT through the owning sync storage, preserving the exact
# validation, join pruning and complex-prefetch rewrite its cursor path uses.
# _select_args returns the operation name first; the async SQLMaker consumes
# only the prepared select arguments that follow it.
sub _prepare_select_args_async {
  my ($self, $ident, $select, $where, $attrs) = @_;
  $attrs = {} unless defined $attrs;

  my $owner = _effective_owner_of($self);
  $self->throw_exception('async SELECT requires its owning sync storage')
    unless $owner && $owner->can('_select_args');

  my @prepared = $owner->_select_args($ident, $select, $where, $attrs);
  shift @prepared;
  return @prepared;
}


# The DB-specific INSERT-column retrieval (_post_insert_sql) and first-row
# post-processing for select_single live here exactly once, so the pooled path
# (_pool_runner) and the pinned txn path (_pinned_runner, used by the
# TransactionContext) share identical behaviour.
sub _run_crud {
  my ($self, $op, $runner, @args) = @_;

  my $sm = $self->sql_maker;
  if ($op eq 'select' || $op eq 'select_single') {
    my $ident = $args[0];
    @args = $self->_prepare_select_args_async(@args);
    my $attrs = $args[3] || {};
    my ($sql, @bind) = $sm->select(@args);
    # as in _gen_sql_bind: the values of the bind attribute go first
    @bind = bare_binds(@{ $attrs->{bind} || [] }, @bind);
    my $bind_attrs = $self->_crud_bind_attrs(select => $ident, \@bind, @args);
    my $future = $runner->($sql, \@bind, $bind_attrs);

    if ($op eq 'select' && $attrs->{software_limit}) {
      my $rows   = $attrs->{rows};
      my $offset = $attrs->{offset} || 0;
      my $fc = $self->future_class;
      $future = $future->then(sub {
        my @raw_rows = @_;
        splice @raw_rows, 0, $offset if $offset;
        splice @raw_rows, $rows if $rows && @raw_rows > $rows;
        return $fc->done(@raw_rows);
      });
    }

    return $self->_hold_derived($future) unless $op eq 'select_single';
    return $self->_hold_derived($future->then(sub {

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

    # through without the callback, and nothing after it is sent).
    my $next = $from + 1;
    return $f->then(sub {
      return $self->_run_ddl_stmts_async($conn, $stmts, $next, $pool);
    });
  }

  return $self->future_class->done;
}


sub _ddl_quote_ident {
  my ($self, $name) = @_;
  return DBIO::SQL::Util::_quote_ident($name);
}


sub _drop_statements_for {
  my ($self, $schema) = @_;
  return DBIO::SQL::Util::_drop_statements(
    $schema, sub { $self->_ddl_quote_ident($_[0]) }, { cascade => 1 },
  );
}

# ADR 0026: one-shot informational warning when async DDL runs on a
# non-transactional engine, mirroring the sync DBIO::Deploy::Base helper. Kept
# as a local one-time closure (not DBIO::Carp) so nothing leaks into the
# package's method namespace (t/55namespaces_cleaned.t). Keyed on the full
# message so a different class gets its own one-shot.
my %_WARNED_NON_TXN_DDL;
sub _warn_non_txn_ddl_once {
  my ($self, $msg) = @_;
  return if $_WARNED_NON_TXN_DDL{$msg}++;
  warn "non-transactional DDL on $msg\n";
}

# ADR 0026 addendum (k132), async parity (k199): one-shot notice that a
# transactional engine's DDL batch was split into multiple runs because the
# per-statement hook declared some statement non-transactional, so the batch is
# no longer atomic as a whole. Keyed on the CLASS, not the message -- the text
# carries per-batch statement/run counts that vary, so keying on the message
# would re-arm the one-shot for every new upgrade shape (the same reasoning as
# the sync DBIO::Deploy::Base::_warn_split_ddl_once). Local closure, not
# DBIO::Carp, to keep the package method namespace clean (t/55namespaces_cleaned.t).
my %_WARNED_SPLIT_DDL;
sub _warn_split_ddl_once {
  my ($self, $msg) = @_;
  my $key = ref($self) || $self;
  return if $_WARNED_SPLIT_DDL{$key}++;
  warn "split DDL batch on $msg\n";
}

# --- Sync fallbacks ---
# Let sync methods work by blocking on the async result via ->get.
# Useful for scripts/migrations.


sub select        { my $self = shift; return $self->select_async(@_)->get        }
sub select_single { my $self = shift; return $self->select_single_async(@_)->get }
sub insert        { my $self = shift; return $self->insert_async(@_)->get        }
sub update        { my $self = shift; return $self->update_async(@_)->get        }
sub delete        { my $self = shift; return $self->delete_async(@_)->get        }
sub txn_do        { my $self = shift; return $self->txn_do_async(@_)->get        }
sub deploy        { my $self = shift; return $self->deploy_async(@_)->get        }

# --- Schema integration ---

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


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


sub in_txn { 0 }


sub connected { defined $_[0]->{pool} && $_[0]->pool->available > 0 }


sub disconnect {
  my $self = shift;
  # Rotate before shutdown so no context from the outgoing pool generation can
  # dispatch while or after that pool is being torn down. This also rotates when
  # the pool was never materialised.
  $self->_rotate_context_owner_token;
  $self->_listen_teardown(undef, 'the storage was disconnected')
    if $self->{_listen};
  if ($self->{pool}) {
    $self->{pool}->shutdown;
    $self->{pool} = undef;
  }
}

sub DESTROY {
  my $self = shift;
  $self->disconnect if $self->{pool} || $self->{_listen};
}

1;

__END__

=pod

=encoding UTF-8

=head1 NAME

DBIO::Storage::Async - Base class for async storage implementations

=head1 VERSION

version 0.900005

=head1 SYNOPSIS

  # Async is chosen per connection, not by naming a storage class:
  my $schema = MyApp::Schema->connect(
      $dsn, $user, $pass,
      { async => 'future_io' },
  );