File Coverage

blib/lib/Net/Clacks/PostgreSQL2Clacks.pm

Criterion	Covered	Total	%
statement	47	151	31.1
branch	0	52	0.0
condition			n/a
subroutine	16	22	72.7
pod	5	5	100.0
total	68	230	29.5

package Net::Clacks::PostgreSQL2Clacks;
#---AUTOPRAGMASTART---
use v5.36;
use strict;
use diagnostics;
use mro 'c3';
use English qw(-no_match_vars);
use Carp qw[carp croak confess cluck longmess shortmess];
our $VERSION = 35;
use autodie qw( close );
use Array::Contains;
use utf8;
use Encode qw(is_utf8 encode_utf8 decode_utf8);
use Data::Dumper;
use builtin qw[true false is_bool];
no warnings qw(experimental::builtin); ## no critic (TestingAndDebugging::ProhibitNoWarnings)
#---AUTOPRAGMAEND---

use Net::Clacks::Client;
use DBI;
use Time::HiRes qw(sleep);

sub new($class, %config) {
    my $self = bless \%config, $class;

    if(!defined($self->{clacks})) {
        croak("clacks config not defined");
    }

    my @requireds = qw(username password clientname);
    if(!defined($self->{clacks}->{socket})) {
        push @requireds, 'server';
        push @requireds, 'port';
    }

    foreach my $required (@requireds) {
        if(!defined($self->{clacks}->{$required})) {
            croak("clacks setting $required not defined");
        }
    }

    if(!defined($self->{clacks}->{caching})) {
        $self->{clacks}->{caching} = 0;
    }

    if(!defined($self->{postgres})) {
        croak("postgres config not defined");
    }

    foreach my $required (qw(url username password)) {
        if(!defined($self->{postgres}->{$required})) {
            croak("postgres setting $required not defined");
        }
    }

    my $clacks;

    if(defined($self->{clacks}->{socket})) {
        # Unix domain sockets
        $clacks = Net::Clacks::Client->newSocket($self->{clacks}->{socket}, $self->{clacks}->{username}, $self->{clacks}->{password}, $self->{clacks}->{clientname}, $self->{clacks}->{caching})
            or croak("Failed to connect to Clacks");
    } else {
        # TCP/IP connection
        $clacks = Net::Clacks::Client->new($self->{clacks}->{server}, $self->{clacks}->{port}, $self->{clacks}->{username}, $self->{clacks}->{password}, $self->{clacks}->{clientname}, $self->{clacks}->{caching})
            or croak("Failed to connect to Clacks");
    }

    my $dbh = DBI->connect($self->{postgres}->{url}, $self->{postgres}->{username}, $self->{postgres}->{password},
                               {
                                   AutoCommit => 0,
                                   RaiseError => 0,
                                   AutoInactiveDestroy => 1,
                               }) or croak($EVAL_ERROR);

    $self->{clacks} = $clacks;
    $self->{dbh} = $dbh;

    $self->{nextping} = time + 10;

    $self->{dbh}->do('LISTEN clacksmessage');
    $self->{dbh}->commit;

    return $self;
}

sub initFunctions($self) {

    my @stmts = $self->getStatements();

    # Create/update function on PostgreSQL side
    foreach my $stmt (@stmts) {
        if(!$self->{dbh}->do($stmt)) {
            croak($self->{dbh}->errstr);
        }
    }
    $self->{dbh}->commit;

    return;
}

sub run($self) {
    while(1) {
        my @lines = $self->runOnce();
        foreach my $line (@lines) {
            print STDERR $line, "\n";
        }
        sleep(0.1);
    }

    return;
}

sub runOnce($self) {
    if(!$self->{dbh}->ping) {
        croak("Database connection failed!");
    }
    my @lines;

    $self->{clacks}->doNetwork();

    if(time < $self->{nextping}) {
        $self->{clacks}->ping();
        $self->{nextping} = time + 10;
        $self->{clacks}->doNetwork();
    }


    while((my $message = $self->{clacks}->getNext())) {
        # We mostly do this to keep the inbuffer nice and empty.
        # Since we neither LISTEN nor expect any other message, we should be fine
        if($message->{type} eq 'serverinfo') {
            push @lines, "Connected to " . $message->{data};
        } elsif($message->{type} eq 'disconnect') {
            if($message->{data} eq 'timeout') {
                push @lines, "Connection timeout! If you use runOnce() instead of run(), make sure to call it at least every 10 seconds!";
            }
            push @lines, "Connection to server lost.";
        } elsif($message->{type} eq 'reconnected') {
            push @lines, "Reconnected to server.";
        } else {
            push @lines, "Clacks-Message: " . $message->{type} . " ignored.\n";
        }
    }


    while((my $notify = $self->{dbh}->pg_notifies)) {
        my ($nname, $npid, $npayload) = @{$notify};

        my ($command, $name, $value) = split/\§\§\§CLACKSDELIMETER\§\§\§/, $npayload;

        if(!defined($name)) {
            $name = '';
        }

        if(!defined($value)) {
            $value = '';
        }

        my $line = join(' ', $command, $name, $value);
        if($command eq 'NOTIFY') {
            $self->{clacks}->notify($name);
        } elsif($command eq 'SET') {
            $self->{clacks}->set($name, $value);
        } elsif($command eq 'STORE') {
            $self->{clacks}->store($name, $value);
        } elsif($command eq 'SETANDSTORE') {
            $self->{clacks}->setAndStore($name, $value);
        } elsif($command eq 'INCREMENT') {
            $self->{clacks}->increment($name, $value);
        } elsif($command eq 'DECREMENT') {
            $self->{clacks}->decrement($name, $value);
        } elsif($command eq 'REMOVE') {
            $self->{clacks}->remove($name);
        } else {
            $line = 'UNKNOWN COMMAND: ' . $line;
        }

        push @lines, $line;
        $self->{dbh}->commit;
        $self->{clacks}->doNetwork();
    }
    $self->{dbh}->commit;
    $self->{clacks}->doNetwork();

    return @lines;
}


sub DESTROY($self) {
    # Try to disconnect cleanly, but socket might already be DESTROYed, so catch any errors
    eval { ## no critic (ErrorHandling::RequireCheckingReturnValueOfEval)
        $self->{clacks}->disconnect();
        $self->{dbh}->disconnect();
    };

    return;
}

sub getStatements($self) {
    return (
        "CREATE OR REPLACE FUNCTION clacks_notify(clacksname text) RETURNS void AS \$\$
        BEGIN
           PERFORM pg_notify('clacksmessage', 'NOTIFY' || '§§§CLACKSDELIMETER§§§' || clacksname);
           RETURN;
        END
        \$\$ LANGUAGE plpgsql;",

        "CREATE OR REPLACE FUNCTION clacks_set(clacksname text, clacksvalue text) RETURNS void AS \$\$
        BEGIN
           PERFORM pg_notify('clacksmessage', 'SET' || '§§§CLACKSDELIMETER§§§' || clacksname || '§§§CLACKSDELIMETER§§§' || clacksvalue);
           RETURN;
        END
        \$\$ LANGUAGE plpgsql;",

        "CREATE OR REPLACE FUNCTION clacks_store(clacksname text, clacksvalue text) RETURNS void AS \$\$
        BEGIN
           PERFORM pg_notify('clacksmessage', 'STORE' || '§§§CLACKSDELIMETER§§§' || clacksname || '§§§CLACKSDELIMETER§§§' || clacksvalue);
           RETURN;
        END
        \$\$ LANGUAGE plpgsql;",

        "CREATE OR REPLACE FUNCTION clacks_setandstore(clacksname text, clacksvalue text) RETURNS void AS \$\$
        BEGIN
           PERFORM pg_notify('clacksmessage', 'SETANDSTORE' || '§§§CLACKSDELIMETER§§§' || clacksname || '§§§CLACKSDELIMETER§§§' || clacksvalue);
           RETURN;
        END
        \$\$ LANGUAGE plpgsql;",

        "CREATE OR REPLACE FUNCTION clacks_increment(clacksname text, clacksvalue text) RETURNS void AS \$\$
        BEGIN
           PERFORM pg_notify('clacksmessage', 'INCREMENT' || '§§§CLACKSDELIMETER§§§' || clacksname || '§§§CLACKSDELIMETER§§§' || clacksvalue);
           RETURN;
        END
        \$\$ LANGUAGE plpgsql;",

        "CREATE OR REPLACE FUNCTION clacks_decrement(clacksname text, clacksvalue text) RETURNS void AS \$\$
        BEGIN
           PERFORM pg_notify('clacksmessage', 'DECREMENT' || '§§§CLACKSDELIMETER§§§' || clacksname || '§§§CLACKSDELIMETER§§§' || clacksvalue);
           RETURN;
        END
        \$\$ LANGUAGE plpgsql;",

        "CREATE OR REPLACE FUNCTION clacks_remove(clacksname text) RETURNS void AS \$\$
        BEGIN
           PERFORM pg_notify('clacksmessage', 'REMOVE' || '§§§CLACKSDELIMETER§§§' || clacksname);
           RETURN;
        END
        \$\$ LANGUAGE plpgsql;",
    );
};


1;
__END__