use v6;

use NativeCall;
use NativeHelpers::CStruct;
need DBDish::Pg::Connection;



head

Pg::Notify - Reactive interface to PostgreSQL notifications


code

=begin code

use Pg::Notify;
use DBIish;

my $db = DBIish.connect('Pg', database => "dbdishtest");
my $channel = "test";

my $notify = Pg::Notify.new(:$db, :$channel );

react {
    whenever $notify -> $notification {
        say $notification.extra;
    }
    whenever Supply.interval(1) -> $v {
        say $db.do("NOTIFY $channel, '$v'");
    }
}

=end code


This provides a simple mechanism to get a supply of the PostgresQL
notifications for a single or group of channels. The supply will emit a stream
of pg-notify objects corresponding to a NOTIFY executed on the connected
Postgres database.


Typically the NOTIFY will be invoked in a trigger or other server side
code but could just as easily be in some other user code (as in the
Synopsis above,)


The objects of type Pg::Notify have a Supply method that allows coercion
in places that expect a Supply (such as whenever in the Synopsis
above.) but you can this Supply directly if you want to tap it for
instance.


method new(DBDish::Pg::Connection :$db, Str :@channel)


Both parameters are required. C<$db> should be the result of a
C<DBIish.connect> to a Postgres database,  C<@channel> should be
a list of NOTIFY topics that you want to receive.


Using multiple Pg::Notify on the same DB connection may result in
lost notifications.


This returns the Supply onto which the notifications are emitted as
they arrive, it will act as a coercer in certain places such as a
C<whenever> which expect a Supply.


The objects emitted are C<DBDish::Pg::Native::pg-notify> with the
following members:


The name of the "topic" or "channel" the notification is for.


The Process ID of the process in which the NOTIFY was executed.


The string "payload" of the notification, it is optional on the
NOTIFY command so may be undefined.  There is a limit of about
8192 Bytes on the size of the oayload. The encoding of the payload
is a function of the server configuration.


This will call UNLISTEN on the database connection and terminate
the thread that polls for new notifications, after calling this
if you want to resume listening for notifications you will need
to create a new object.



class Pg::Notify {
    has DBDish::Pg::Connection  $.db      is required;
    has Str                     @.channel is required;

    has $!thread;

    has Supplier $!supplier;

    has Promise $!run-promise;

    class PollFD is repr('CStruct') {
        has int32 $.fd;
        has int16 $.events;
        has int16 $.revents;
    }

    sub poll(PollFD $fds, int64 $nfds, int32 $timeout --> int32 ) is native { * }

    method supplier( --> Supplier ) handles <Supply> {
        $!supplier //= do {
            my $supplier = Supplier.new;
            self.listen;
            $!run-promise = Promise.new;
            $!thread = Thread.start: :app_lifetime, {
                loop {
                    $!db.pg-consume-input;

                    # Retrieve all pending notifications.
                    while $!db.pg-notifies -> $not {
                        if @!channel (cont) $not.relname {
                            $supplier.emit: $not;
                        }
                    }
					self.poll-once;
                }

                # Exceptions inside this block typically indicate some type
                # of communications issue with the database.
                CATCH {
                    default {
                        $supplier.quit($_);
                    }
                }
            }
            $supplier;
        }
    }

    method poll-once( --> Int ) {
        my $fds = LinearArray[PollFD].new(1);
        $fds[0] = PollFD.new(fd => $!db.pg-socket,  events => 1, revents => 0);
        poll($fds.base, 1, -1);
        my $rc = $fds[0].revents;
        $fds.dispose;
        $rc;
    }


    method listen() {
        for @!channel -> $channel {
            $!db.do("LISTEN " ~ $channel);
        }
    }

    method unlisten() {
        for @!channel -> $channel {
            $!db.do("UNLISTEN " ~ $channel);
        }
        if $!run-promise {
            $!run-promise.keep: True;
        }
    }
}


# vim: ft=raku ts=4 sw=4 expandtab


root

README

React

=begin React :component<HeaderCol> :id<menu> 
=para L<Documentation|/doc/introduction>
=para L<Modules| /mods>
=para L<Examples| file:../examples/index.podlite>
=para L<Download |file:../download/index.podlite>
=para L<About|file:../about/index.podlite>
=para 🔍 K<⌘K>/K<ctrl-K>
=end React


=begin React :component<HeaderCol> :id<footer>  
=begin nested :!nested
=item1 B<Language>
    =item2 L<Get started|file:../getting-started/getting-started.podlite>
    =item2 L<Why Raku?|/doc/language/faq#Why-should-I-learn-Raku-What's-so-great-about-it>
    =item2 L<Try Raku|https://glot.io/new/raku>
    =item2 L<Raku cheat sheet|https://github.com/Raku/mu/blob/master/docs/Perl6/Cheatsheet/cheatsheet.txt>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<Raku on Exercism|https://exercism.org/tracks/raku>
    =item2 L<Wikipedia|https://en.wikipedia.org/wiki/Raku_(programming_language)>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>

=end nested
=begin nested :!nested

=item1 B<L<Documentation|/doc>>
    =item2 L<Getting started, Migration guides from other languages, & Tutorials | file:../doc/introduction.podlite>
    =item2 L<Language References|file:../doc/reference.podlite>
    =item2 L<Type Reference| file:../doc/types.podlite>
    =item2 L<Miscellaneous| file:../doc/miscellaneous.podlite>
    =item2 L<FAQs (Frequently Asked Questions)|/doc/language/faq>
    =item2 L<Community|/doc/language/community>
    =item2 L<The list of all documents|file:../doc/index.podlite>

=item1 B<Resources>
    =item2 L<The Raku Guide|https://raku.guide/>
    =item2 L<Books |https://perl6book.com/>
    =item2 L<Rosetta Code | https://www.raku.org/community/rosettacode>

=end nested
=begin nested :!nested

=item1 B<Resoures>
    =item2 L<Download | file:../download/index.podlite>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>

=item1 B<Learning>
    =item2 L<The Raku Guide|https://raku.guide/>
    =item2 L<Wikibook |https://en.wikibooks.org/wiki/Raku_Programming>
    =item2 L<Books |https://perl6book.com/>
    =item2 L<Rosetta Code | https://www.raku.org/community/rosettacode>
    =item2 L<Learn Raku in Y minutes|https://learnxinyminutes.com/docs/raku/>
    =item2 L<Golfing |https://github.com/AlexDaniel/raku-golf-cheatsheet>
=end nested
=begin nested :!nested

=item1 B<Explore>
    =item2 L<Raku Blog Aggregator|https://planet.raku.org/>
    =item2 L<Rakudo Weekly|https://rakudoweekly.blog/>
    =item2 L<The Weekly Challenge |https://perlweeklychallenge.org/>
    =item2 L<Raku Advent Calendar|https://raku-advent.blog/>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>

=end nested
=end React


=begin React :component<CookieConsent> :id<CookieConsent> :buttonCaption("Got it!")
=para
This website uses cookies for analytics.
=end React


Notify

NAME

SYNOPSIS

DESCRIPTION

METHODS

method new

method Supply

relname

be_pid

extra

method unlisten()

Pg::Notify v0.0.7

Authors

License

Dependencies

Test Dependencies

Provides

Documentation