File Coverage

blib/lib/Mojo/MySQL5/PubSub.pm

Criterion	Covered	Total	%
statement	9	86	10.4
branch	0	26	0.0
condition	1	13	7.6
subroutine	3	11	27.2
pod	3	3	100.0
total	16	139	11.5

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

use Scalar::Util 'weaken';

use constant DEBUG => $ENV{MOJO_PUBSUB_DEBUG} || 0;

has 'mysql';

sub DESTROY {
  my $self = shift;
  return unless $self->{db} and $self->mysql;
  $self->mysql->db->query('delete from mojo_pubsub_subscribe where pid = ?', $self->_subscriber_pid);
}

sub listen {
  my ($self, $channel, $cb) = @_;
  my $pid = $self->_subscriber_pid;
  warn "listen channel:$channel subscriber:$pid\n" if DEBUG;
  $self->mysql->db->query('replace mojo_pubsub_subscribe(pid, channel, ts) values (?, ?, current_timestamp)',
    $pid, $channel);
  push @{$self->{chans}{$channel}}, $cb;
  return $cb;
}

sub notify {
  my ($self, $channel, $payload) = @_;
  my $db = $self->mysql->db;
  $payload //= '';
  warn "notify channel:$channel $payload\n" if DEBUG;
  $self->_init($db) unless $self->{init}++;
  $db->query('insert into mojo_pubsub_notify(channel, payload) values (?, ?)', $channel, $payload);
  return $self;
}

sub unlisten {
  my ($self, $channel, $cb) = @_;
  my $pid = $self->_subscriber_pid;
  warn "unlisten channel:$channel subscriber:$pid\n" if DEBUG;
  my $chan = $self->{chans}{$channel};
  @$chan = grep { $cb ne $_ } @$chan;
  return $self if @$chan;
  $self->mysql->db->query('delete from mojo_pubsub_subscribe where pid = ? and channel = ?', $pid, $channel);
  delete $self->{chans}{$channel};
  return $self;
}

sub _notifications {
  my $self = shift;
  my $result = $self->{db}->query('select id, channel, payload from mojo_pubsub_notify where id > ?', $self->{last_id});

  while (my $row = $result->array) {
    my ($id, $channel, $payload) = @$row;
    $self->{last_id} = $id;
    next unless exists $self->{chans}{$channel};
    warn "received $id on $channel: $payload\n" if DEBUG;
    for my $cb (@{$self->{chans}{$channel}}) { $self->$cb($payload) }
  }
}

sub _init {
  my ($self, $db) = @_;

  $self->mysql->migrations->name('pubsub')->from_data->migrate;

  # cleanup old subscriptions and notifications
  $db->query('delete from mojo_pubsub_notify where ts < date_add(current_timestamp, interval -10 minute)');
  $db->query('delete from mojo_pubsub_subscribe where ts < date_add(current_timestamp, interval -1 hour)');
}

sub _subscriber_pid {
  my $self = shift;

  # Fork-safety
  if (($self->{pid} //= $$) ne $$) {
    my $pid = $self->{db}->pid if $self->{db};
    warn 'forked subscriber pid:' . ($pid || 'N/A') . "\n" if DEBUG;
    $self->{db}->disconnect if $pid;
    delete @$self{qw(chans init pid db)};
  }

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

  $self->{db} = $self->mysql->db;
  my $pid = $self->{db}->pid;
  $self->{db}->connection->url->options->{query_timeout} = 610;

  $self->_init($self->{db}) unless $self->{init}++;

  if (defined $self->{last_id}) {

    # read unread notifications
    $self->_notifications;
  }
  else {
    # get id of the last message
    my $array = $self->{db}->query('select id from mojo_pubsub_notify order by id desc limit 1')->array;
    $self->{last_id} = defined $array ? $array->[0] : 0;
  }

  # re-subscribe
  $self->{db}->query('replace mojo_pubsub_subscribe(pid, channel) values (?, ?)', $pid, $_)
    for keys %{$self->{chans}};

  weaken $self->{db}->{mysql};
  weaken $self;

  my $cb;
  $cb = sub {
    my ($db, $err, $result) = @_;
    if ($err) {
      warn "wake up error: $err" if DEBUG;
      eval { $db->disconnect };
      delete $self->{db};
      eval { $self->_subscriber_pid };
    }
    elsif ($self and $self->{db}) {
      $self->_notifications;
      $db->query('update mojo_pubsub_subscribe set ts = current_timestamp where pid = ?', $pid);
      $db->query('select sleep(600)', $cb);
    }
  };
  $self->{db}->query('select sleep(600)', $cb);

  warn "reconnect subscriber pid: $pid\n" if DEBUG;
  $self->emit(reconnect => $self->{db});

  return $pid;
}

1;

=encoding utf8

=head1 NAME

Mojo::MySQL5::PubSub - Publish/Subscribe

=head1 SYNOPSIS

  use Mojo::MySQL5::PubSub;

  my $pubsub = Mojo::MySQL5::PubSub->new(mysql => $mysql);
  my $cb = $pubsub->listen(foo => sub {
    my ($pubsub, $payload) = @_;
    say "Received: $payload";
  });
  $pubsub->notify(foo => 'bar');
  $pubsub->unlisten(foo => $cb);

=head1 DESCRIPTION

L is implementation of the publish/subscribe
pattern used by L.

Although MySQL does not have C like PostgreSQL and other RDBMs,
this module implements similar feature.

Single Database connection waits for notification by executing C on server.
C and subscribed channels in stored in C table.
Inserting new row in C table triggers C for
all connections waiting for notification.

C privilege is needed for MySQL user to see other users processes.
C privilege is needed to be able to execute C for statements
started by other users.
C privilege may be needed to be able to define trigger.

If your applications use this module using different MySQL users it is important
the migration script to be executed by user having C privilege on the database.


=head1 EVENTS

L inherits all events from L and can
emit the following new ones.

=head2 reconnect

  $pubsub->on(reconnect => sub {
    my ($pubsub, $db) = @_;
    ...
  });

Emitted after switching to a new database connection for sending and receiving
notifications.

=head1 ATTRIBUTES

L implements the following attributes.

=head2 mysql

  my $mysql = $pubsub->mysql;
  $pubsub   = $pubsub->mysql(Mojo::MySQL5->new);

L object this publish/subscribe container belongs to.

=head1 METHODS

L inherits all methods from L and
implements the following new ones.

=head2 listen

  my $cb = $pubsub->listen(foo => sub {...});

Subscribe to a channel, there is no limit on how many subscribers a channel can
have.

  # Subscribe to the same channel twice
  $pubsub->listen(foo => sub {
    my ($pubsub, $payload) = @_;
    say "One: $payload";
  });
  $pubsub->listen(foo => sub {
    my ($pubsub, $payload) = @_;
    say "Two: $payload";
  });

=head2 notify

  $pubsub = $pubsub->notify('foo');
  $pubsub = $pubsub->notify(foo => 'bar');

Notify a channel.

=head2 unlisten

  $pubsub = $pubsub->unlisten(foo => $cb);

Unsubscribe from a channel.

=head1 DEBUGGING

You can set the C environment variable to get some
advanced diagnostics information printed to C.

  MOJO_PUBSUB_DEBUG=1

=head1 SEE ALSO

L, L, L.

=cut

__DATA__