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('select', $self->_pool_runner, @_);
}


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


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


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


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

# Build SQL for a CRUD op and execute it through $runner, a coderef that
# takes ($sql, \@bind) and returns a Future of result rows. 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. Placeholder-dialect shaping (?->$N) is NOT applied here:
# the query seams receive raw sql_maker '?' SQL and shape it internally (karr
# #70 / ADR 0032), so the runner is handed the raw $sql.
sub _run_crud {
  my ($self, $op, $runner, @args) = @_;

  my $sm = $self->sql_maker;
  if ($op eq 'select' || $op eq 'select_single') {
    my ($sql, @bind) = $sm->select(@args);
    my $f = $runner->($sql, \@bind);
    return $f unless $op eq 'select_single';
    return $f->then(sub {
      my @rows = @_;
      return @rows ? $rows[0] : undef;
    });
  }
  elsif ($op eq 'insert') {
    # The CRUD contract is insert_async($source, \%rowdata) (ADR 0031 §3),
    # where $source is a result source. SQLMaker wants a plain table name,
    # so unwrap a blessed source to its ->name; a bare string passes through.
    my ($source, @rest) = @args;
    my $to_insert = $rest[0];
    my $table = (blessed($source) && $source->can('name'))
      ? $source->name
      : $source;

    my ($sql, @bind) = $sm->insert($table, @rest);
    my $post = $self->_post_insert_sql;
    $sql .= $post if length $post && $sql !~ /RETURNING/i;

    # ADR 0031 §3: insert_async must resolve the returned-columns HASHREF --
    # the supplied insert data overlaid with the DB-populated columns (autoinc
    # PK + retrieve-on-insert), exactly what sync $storage->insert returns --
    # so create_async / Row::insert_async can fold it back via
    # _store_inserted_columns. The runner yields the raw RETURNING row(s); map
    # the first onto the insert data. (select_async / select_single_async keep
    # resolving raw arrayrefs, matching the sync cursor shape.)
    return $runner->($sql, \@bind)->then(sub {
      my @rows = @_;
      return $self->_insert_returned_columns($source, $to_insert, $rows[0]);
    });
  }
  elsif ($op eq 'update') {
    my ($sql, @bind) = $sm->update(@args);
    return $runner->($sql, \@bind);

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

}


sub _use_transactional_ddl {
  my ($self) = @_;
  my $owner = $self->_owner_storage or return 0;
  return 0 unless $owner->can('_use_transactional_ddl');
  return $owner->_use_transactional_ddl ? 1 : 0;
}


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

  my @stmts = grep { !/^\s*--/ } DBIO::SQL::Util::_split_statements($ddl);
  my $fc = $self->future_class;

  my $recur;
  $recur = sub {
    my $stmt = shift @stmts;
    return $fc->done unless defined $stmt;
    return $self->_query_async_pinned($conn, $stmt, [])->then($recur);
  };
  return $recur->();
}


sub _drop_statements_for {
  my ($self, $schema) = @_;
  my @out;
  for my $name ($schema->sources) {
    my $source = $schema->source($name);
    next if $source->isa('DBIO::ResultSource::View');
    my $table = $source->name;
    next if ref $table;
    next if $table =~ /\s|\(/;
    push @out, sprintf 'DROP TABLE IF EXISTS %s CASCADE;', DBIO::SQL::Util::_quote_ident($table);
  }
  return join("\n\n", @out);
}

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

# --- 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;
  if ($self->{pool}) {
    $self->{pool}->shutdown;
    $self->{pool} = undef;
  }
}

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

1;

__END__

=pod

=encoding UTF-8

=head1 NAME

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

=head1 VERSION

version 0.900001

=head1 SYNOPSIS

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

  # Async queries return Futures
  $schema->resultset('Artist')->all_async->then(sub {
      my @artists = @_;
      say $_->name for @artists;
  });

See F<t/test/09_async.t> for a runnable example.

=head1 DESCRIPTION