view release on metacpan or  search on metacpan

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

sub _track_statement {
  my ($self, $future) = @_;
  return $future unless Scalar::Util::blessed($future)
    && $future->can('then') && $future->can('is_ready')
    && !$future->is_ready;
  my $gate = ref($future)->new;
  $self->{storage}->_hold_derived($future->then(
    sub { $gate->done(@_) unless $gate->is_ready; return },
    sub { $gate->fail(@_) unless $gate->is_ready; return },
  ));
  push @{ $self->{_statements} }, $gate;
  return $gate;
}

# Every kept statement Future, answered or not, in the order handed out.
sub _statements { return @{ $_[0]->{_statements} || [] } }

# Every statement still open fails with the error that ended the batch.
sub _fail_open_statements {
  my ($self, $error) = @_;
  for my $gate (grep { !$_->is_ready } $self->_statements) {
    $gate->fail($error);
  }
  return;
}

# CRUD inside the batch MUST run on the connection the scaffold put into
# pipeline mode, not a fresh pooled one -- that is the whole point of the
# bracket. Each method routes through the storage's shared CRUD builder with
# the bracketed connection, so SQL generation (insert post-processing,
# select_single first-row) is not duplicated. Forwarding to
# $storage->select_async etc. would acquire a separate pool connection and run
# the query OUTSIDE the pipeline.


sub select_async {
  my $self = shift;
  $self->_assert_context_active;
  return $self->_track_statement(
    $self->{storage}->_run_crud_pinned('select', $self->{pipeline_conn}, @_)
  );
}

sub select_single_async {
  my $self = shift;
  $self->_assert_context_active;
  return $self->_track_statement(
    $self->{storage}->_run_crud_pinned('select_single', $self->{pipeline_conn}, @_)
  );
}

sub insert_async {
  my $self = shift;
  $self->_assert_context_active;
  return $self->_track_statement(
    $self->{storage}->_run_crud_pinned('insert', $self->{pipeline_conn}, @_)
  );
}

sub update_async {
  my $self = shift;
  $self->_assert_context_active;
  return $self->_track_statement(
    $self->{storage}->_run_crud_pinned('update', $self->{pipeline_conn}, @_)
  );
}

sub delete_async {
  my $self = shift;
  $self->_assert_context_active;
  return $self->_track_statement(
    $self->{storage}->_run_crud_pinned('delete', $self->{pipeline_conn}, @_)
  );
}

# No blocking counterparts. The sync form of every other DBIO async surface is
# "call the *_async twin and ->get it" -- which cannot work here: a batched
# statement's result arrives at _pipeline_sync, and the sync only runs once the
# batch coderef has returned. ->get inside the batch would therefore wait for a
# result whose delivery is waiting on the caller it is blocking. Refusing with
# the reason beats hanging the process (and beats "Can't locate object method",
# which reads like the context is simply incomplete).


sub select        { $_[0]->_no_blocking('select')        }
sub select_single { $_[0]->_no_blocking('select_single') }
sub insert        { $_[0]->_no_blocking('insert')        }
sub update        { $_[0]->_no_blocking('update')        }
sub delete        { $_[0]->_no_blocking('delete')        }

sub _no_blocking {
  my ($self, $op) = @_;
  croak "Blocking ->$op is not available inside a pipeline batch: a batched "
    . "statement resolves at the pipeline sync, which runs after the batch "
    . "coderef returns -- blocking on it here would deadlock. Call "
    . "->${op}_async and return an aggregate Future from the batch.";
}

# Delegations


sub sql_maker { my $self = shift; $self->{storage}->sql_maker(@_) }
sub debug     { my $self = shift; $self->{storage}->debug(@_)     }

1;

__END__

=pod

=encoding UTF-8

=head1 NAME

DBIO::Storage::Async::PipelineContext - Generic pinned-connection context for async pipeline batches

=head1 VERSION

version 0.900005

=head1 DESCRIPTION

The batch context handed to the coderef of
L<DBIO::Storage::Async/pipeline>. Pipeline mode is B<per-connection> state:
a statement is only batched if it is sent on the very connection the
scaffold put into pipeline mode. This context is what makes that structural
instead of advisory -- it carries the bracketed connection and routes every
CRUD call through the storage's shared C<_run_crud_pinned> builder onto it,
exactly as L<DBIO::Storage::Async::TransactionContext> does for the pinned
transaction connection (dbio-async ADR 0002 / ADR 0003).

Without it, an C<*_async> call inside the batch coderef would go through the
pooled path and take a B<different> connection out of the pool -- running
outside the bracket, unbatched, and at C<< pool_size => 1 >> only after the
pipeline has released. The statements would still execute; the round-trip
economy the batch was written for would silently not happen.

The context is only valid for the synchronous run of the batch coderef: once
the coderef returns, the scaffold invalidates the context, syncs, exits
pipeline mode and releases the connection. A query issued on the context past
that point throws a L<DBIO::Exception> (C<pipeline context is no longer
active>) instead of reaching a connection that is back in the pool.

=head1 ATTRIBUTES

=head2 storage

The parent L<DBIO::Storage::Async> instance.