view release on metacpan or  search on metacpan

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

  # Load-order: this context class is loaded standalone by dbio-async's
  # DBIO::Async::TransactionContext (use base ...) WITHOUT pulling
  # DBIO::Storage::Async, and a bare mock backend (above) is not an Async
  # subclass, so Async.pm may not be loaded yet when this runs -- the helper
  # would then be an undefined subroutine (karr core k209). A compile-time
  # `use DBIO::Storage::Async` cannot fix it: Async.pm use-s this class back
  # (its line 13), so that closes a circular use. Break the cycle with a
  # runtime require, which is idempotent and safe here because the helper is
  # only ever needed at call time (dbio-perl-syntax: require to break a genuine
  # circular dependency).
  require DBIO::Storage::Async;
  my $owner = DBIO::Storage::Async::_effective_owner_of($backend);
  DBIO::Exception->throw('async context ownership cannot be proven')
    unless $owner;
  $owner->_assert_async_context_owner($self);
  return $self;
}


sub _query_async {
  my ($self, $sql, $bind) = @_;
  $self->_assert_context_active;
  return $self->_track_statement(
    $self->{storage}->_query_async_pinned($self->{txn_conn}, $sql, $bind)
  );
}

# CRUD inside a transaction MUST run on the pinned connection that
# BEGIN/COMMIT ran on, not a fresh pooled one. Each method routes
# through the storage's shared CRUD builder with the pinned 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
# transaction.


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

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

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

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

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

# Sync fallbacks -- block the event loop via ->get


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

# Delegations


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

sub pipeline {
  my ($self, $coderef) = @_;
  $self->_assert_context_active;
  if (my $refusal = $self->{storage}->_pipeline_refusal) {
    return $refusal;
  }
  # No $release: the transaction boundary hands the connection back. A $retire
  # that marks it instead, for that boundary to act on (karr k232).
  my $batch = $self->{storage}->_pipeline_on_conn(
    $self->{txn_conn}, $coderef, undef,
    sub { $self->_mark_txn_conn_unusable(@_) },
  );
  # Kept apart from the statements: the context lets go of those as it ends,
  # and the transaction's ROLLBACK waits for these (karr k331). Weakly -- what
  # holds the Future until it is ready is _track_statement and its retain
  # (karr k292); a context still in hand must not hold it any longer.
  if (Scalar::Util::blessed($batch) && $batch->can('catch')) {
    push @{ $self->{_txn_pipelines} }, $batch;
    Scalar::Util::weaken($self->{_txn_pipelines}[-1]);
  }
  return $self->_track_statement($batch);
}


sub _mark_txn_conn_unusable {
  my ($self, $error) = @_;
  $self->{_txn_conn_unusable} = 1;
  $self->{_txn_conn_unusable_error} = $error
    if defined $error && !defined $self->{_txn_conn_unusable_error};
  return;
}


sub _txn_conn_unusable_error { return $_[0]->{_txn_conn_unusable_error} }

sub _txn_conn_unusable { return $_[0]->{_txn_conn_unusable} ? 1 : 0 }


sub _set_txn_guard { $_[0]->{_txn_guard} = $_[1]; return }

sub _txn_guard { return $_[0]->{_txn_guard} }


sub _track_statement {
  my ($self, $future) = @_;
  push @{ $self->{_txn_statements} }, $future
    if Scalar::Util::blessed($future) && $future->can('catch');
  return $future;
}

sub _txn_statements { return @{ $_[0]->{_txn_statements} || [] } }