view release on metacpan or  search on metacpan

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

    });

    return $f;
  });
}

# --- Pipeline Mode ---


sub pipeline {
  my ($self, $coderef) = @_;

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

    my $result = eval { $coderef->($self) };
    my $err = $@;

    if ($err) {
      $pg->exit_pipeline;
      $self->pool->release($pg);
      return Future->fail($err);
    }

    # Sync the pipeline — callback fires when all results are in
    my $f = Future->new;
    $pg->pipeline_sync(sub {
      $pg->exit_pipeline;
      $self->pool->release($pg);
      if (ref $result && $result->can('then')) {
        $result->on_done(sub { $f->done(@_) })
               ->on_fail(sub { $f->fail(@_) });
      } else {
        $f->done($result);
      }
    });

    return $f;
  });
}

# --- LISTEN/NOTIFY ---


sub listen {
  my ($self, $channel, $cb) = @_;

  $self->{_listeners}{$channel} = $cb;

  # Use a dedicated connection for LISTEN (not from the pool).
  # EV::Pg->new returns before the socket is actually connected; query()
  # dispatched on a not-yet-connected handle throws "not connected".
  # We buffer LISTEN/UNLISTEN until on_connect fires and then flush.
  $self->{_listen_pg} ||= do {
    require EV::Pg;
    $self->{_listen_pending} = [];
    $self->{_listen_connected} = 0;
    my $pg = EV::Pg->new(
      conninfo   => $self->_conninfo_string,
      keep_alive => 1,
      on_connect => sub {
        $self->{_listen_connected} = 1;
        my $q = delete $self->{_listen_pending} || [];
        $self->{_listen_pending} = [];
        $self->{_listen_pg}->query($_, sub {}) for @$q;
      },
      on_error   => sub { warn "LISTEN connection error: $_[0]\n" },
      on_notify  => sub {
        my ($ch, $payload, $pid) = @_;
        if (my $handler = $self->{_listeners}{$ch}) {
          $handler->($ch, $payload, $pid);
        }
      },
    );
    $pg;
  };

  my $quoted = $self->sql_maker->_quote($channel);
  my $sql = "LISTEN $quoted";
  if ($self->{_listen_connected}) {
    $self->{_listen_pg}->query($sql, sub {});
  } else {
    push @{ $self->{_listen_pending} }, $sql;
  }
}


sub unlisten {
  my ($self, $channel) = @_;
  delete $self->{_listeners}{$channel};
  if ($self->{_listen_pg}) {
    my $quoted = $self->sql_maker->_quote($channel);
    my $sql = "UNLISTEN $quoted";
    if ($self->{_listen_connected}) {
      $self->{_listen_pg}->query($sql, sub {});
    } else {
      push @{ $self->{_listen_pending} }, $sql;
    }
  }
}


sub notify {
  my ($self, $channel, $payload) = @_;

  croak 'Channel name required' unless defined $channel && $channel ne '';

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

    # pg_notify() with bind params -- NOTIFY itself takes no placeholders,
    # and inlining the payload as a string literal invites quoting bugs.
    $pg->query_params('SELECT pg_notify($1, $2)', [ $channel, $payload // '' ], sub {
      my ($res, $err) = @_;
      $self->pool->release($pg);
      if ($err) {
        $f->fail($err);
      } else {
        $f->done;