view release on metacpan or  search on metacpan

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

  };
}

sub _conninfo_string {
  my ($self, $ci) = @_;
  $ci = $self->_current_async_connect_info($self->access_broker_mode)->[0]
    if ! defined $ci;
  return conninfo_string($ci);
}

# --- SQL Generation ---


sub sql_maker {
  my $self = shift;
  $self->{sql_maker} ||= do {
    my $class = $self->{sql_maker_class};
    $class->new(
      quote_char     => '"',
      name_sep       => '.',
      limit_dialect  => 'LimitOffset',
    );
  };
}

sub _generate_sql {
  my ($self, $op, @args) = @_;
  my $sm = $self->sql_maker;
  my $method = {
    select        => 'select',
    select_single => 'select',
    insert        => 'insert',
    update        => 'update',
    delete        => 'delete',
  }->{$op} or croak "Unknown operation: $op";

  return $sm->$method(@args);
}

# --- Async Query Execution ---


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.
# All the PG-specific shaping (RETURNING on insert, first-row for
# select_single) lives here exactly once, so the pooled path
# (_pool_runner) and the pinned txn path (a pinned runner supplied by
# TransactionContext) share identical behavior.
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') {
    my ($sql, @bind) = $sm->insert(@args);
    # PostgreSQL RETURNING for auto-generated columns
    $sql .= ' RETURNING *' unless $sql =~ /RETURNING/i;
    return $runner->($sql, \@bind);
  }
  elsif ($op eq 'update') {
    my ($sql, @bind) = $sm->update(@args);
    return $runner->($sql, \@bind);
  }
  elsif ($op eq 'delete') {
    my ($sql, @bind) = $sm->delete(@args);
    return $runner->($sql, \@bind);
  }
  croak "Unknown CRUD operation: $op";
}

# Runner that executes on a freshly-acquired pooled connection and
# releases it when done. This is the normal (non-transactional) path.
sub _pool_runner {
  my $self = shift;
  return sub {
    my ($sql, $bind) = @_;
    return $self->_query_async($sql, $bind);
  };
}

# Runner that executes on a pinned connection without releasing it.
# Used by TransactionContext so CRUD inside a txn hits the same
# connection that BEGIN/COMMIT ran on.
sub _pinned_runner {

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

sub copy_in {
  my ($self, $table, $columns, $coderef) = @_;

  my $col_list = join(', ', map { $self->sql_maker->_quote($_) } @$columns);
  my $quoted_table = $self->sql_maker->_quote($table);
  my $sql = "COPY $quoted_table ($col_list) FROM STDIN";

  return $self->pool->acquire->then(sub {
    my $pg = shift;
    my $f = Future->new;

    $pg->query($sql, sub {
      my ($status, $err) = @_;
      if ($err) {
        $self->pool->release($pg);
        $f->fail($err);
        return;
      }

      my $put = sub {
        my $row = shift;
        my $line = join("\t", map { defined $_ ? $_ : '\N' } @$row) . "\n";
        $pg->put_copy_data($line);
      };

      eval { $coderef->($put) };
      if ($@) {
        $pg->put_copy_end($@);
        $self->pool->release($pg);
        $f->fail($@);
      } else {
        $pg->put_copy_end;
        $self->pool->release($pg);
        $f->done(1);
      }
    });

    return $f;
  });
}

# --- Sync Fallbacks ---
# These allow sync methods (->all, ->first etc.) to work
# by blocking the event loop. 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;
}

# --- Schema Integration ---

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

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

sub disconnect {
  my $self = shift;
  if ($self->{pool}) {
    $self->{pool}->shutdown;
    $self->{pool} = undef;
  }
  if ($self->{_listen_pg}) {
    $self->{_listen_pg}->finish;
    $self->{_listen_pg} = undef;
  }
}

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

1;

__END__

=pod

=encoding UTF-8

=head1 NAME

DBIO::PostgreSQL::Async::Storage - Async PostgreSQL storage driver using EV::Pg

=head1 VERSION

version 0.900000

=head1 DESCRIPTION

Implements L<DBIO::Storage::Async> using L<EV::Pg> — a non-blocking
PostgreSQL client that speaks libpq's async protocol directly.
No DBI, no DBD::Pg, just raw libpq performance.