PubSub.pm

package Mojo::Pg::PubSub;
use Mojo::Base 'Mojo::EventEmitter';

use Scalar::Util 'weaken';

has 'pg';

sub listen {
  my ($self, $name, $cb) = @_;
  push @{$self->{chans}{$name}}, $cb;
  $self->_db->listen($name) unless @{$self->{chans}{$name}} > 1;
  return $cb;
}

sub notify { $_[0]->_db->notify(@_[1, 2]) and return $_[0] }

sub unlisten {
  my ($self, $name, $cb) = @_;
  my $chans = $self->{chans}{$name};
  @$chans = grep { $cb ne $_ } @$chans;
  $self->_db->unlisten($name) and delete $self->{chans}{$name} unless @$chans;
  return $self;
}

sub _db {
  my $self = shift;

  return $self->{db} if $self->{db};

  $self->emit(reconnect => my $db = $self->{db} = $self->pg->db);
  weaken $self;
  $db->on(
    notification => sub {
      my ($db, $name, $pid, $payload) = @_;
      for my $cb (@{$self->{chans}{$name}}) { $self->$cb($payload) }
    }
  );
  $db->once(
    close => sub {
      delete $self->{db};
      my $db = $self->_db;
      $db->listen($_) for keys %{$self->{chans}};
    }
  );

  return $db;
}

1;