PubSub.pm

package Mojo::Pg::PubSub;
use Mojo::Base -base;

use Scalar::Util 'weaken';

has 'pg';

sub emit {
  my ($self, $name, $payload) = @_;
  $self->_db->notify($name, $payload);
  return $self;
}

sub has_subscribers { !!@{shift->{events}{shift()} || []} }

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

sub subscribers { shift->{events}{shift()} || [] }

sub unsubscribe {
  my ($self, $name, $cb) = @_;

  # One
  if ($cb) {
    $self->{events}{$name} = [grep { $cb ne $_ } @{$self->{events}{$name}}];
    delete $self->{events}{$name} unless @{$self->{events}{$name}};
  }

  # All
  else { delete $self->{events}{$name} }

  $self->_db->unlisten($name) unless @{$self->{events}{$name}};

  return $self;
}

sub _db {
  my $self = shift;

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

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

  return $db;
}

1;