File Coverage

blib/lib/IPC/Manager/Client/MessageFiles.pm

Criterion	Covered	Total	%
statement	95	107	88.7
branch	28	50	56.0
condition	4	9	44.4
subroutine	19	21	90.4
pod	6	13	46.1
total	152	200	76.0

package IPC::Manager::Client::MessageFiles;
use strict;
use warnings;

our $VERSION = '0.000005';

use Carp qw/croak confess/;
use File::Spec;

BEGIN {
    if (eval { require Linux::Inotify2; require IO::Select; Linux::Inotify2->can('fh') ? 1 : 0 }) {
        *USE_INOTIFY = sub() { 1 };
    }
    else {
        *USE_INOTIFY = sub() { 0 };
    }
}

use parent 'IPC::Manager::Base::FS';
use Object::HashBase qw{
    +dir_handle
    +inotify

    +pend_count
    +ready_count
};

sub check_path { -d $_[1] }
sub make_path  { mkdir($_[1]) or die "Could not make dir '$_[1]': $!" }
sub path_type  { 'subdir' }

sub init {
    my $self = shift;

    $self->SUPER::init();

    $self->{+PEND_COUNT}  = 0;
    $self->{+READY_COUNT} = 0;

    return unless USE_INOTIFY;
}

sub handles_for_select { $_[0]->inotify->fh }

sub inotify {
    my $self = shift;
    croak "Not Implemented (Or you are missing one of: Linux::Inotify2, IO::Select)" unless USE_INOTIFY();

    return $self->{+INOTIFY} if $self->{+INOTIFY};

    my $i = Linux::Inotify2->new;
    $i->watch($self->path, Linux::Inotify2::IN_CREATE());

    return $self->{+INOTIFY} = $i;
}

sub pre_disconnect_hook {
    my $self = shift;

    my $new_path = File::Spec->catfile($self->{+ROUTE}, "_" . $self->{+ID});
    rename($self->path, $new_path) or die "Cannot rename directory: $!";
    $self->{+PATH} = $new_path;
}

sub dir_handle {
    my $self = shift;
    $self->pid_check;
    my $out = $self->{+DIR_HANDLE} //= do {
        opendir(my $dh, $self->path) or die "Could not open dir: $!";
        $dh;
    };

    rewinddir($out);

    return $out;
}

sub pending_messages {
    my $self = shift;
    return 1 if $self->{+PEND_COUNT};
    return $self->message_files('pend') ? 1 : 0;
}

sub ready_messages {
    my $self = shift;
    return 1 if $self->{+READY_COUNT};
    return 1 if $self->have_resume_file;
    return $self->message_files('ready') ? 1 : 0;
}

sub message_files {
    my $self = shift;
    $self->pid_check;
    my ($ext) = @_;

    return undef if USE_INOTIFY && !$self->select->can_read(0);

    my (@pend, @ready);
    for my $file (readdir($self->dir_handle)) {
        if ($file =~ m/\.ready$/) {
            push @ready => $file;
        }
        elsif ($file =~ m/\.pend$/) {
            push @pend => $file;
        }
    }

    $self->{+READY_COUNT} = @ready;
    $self->{+PEND_COUNT}  = @pend;

    return @ready ? \@ready : undef if $ext eq 'ready';
    return @pend  ? \@pend  : undef if $ext eq 'pend';

    return undef;
}

sub get_messages {
    my $self = shift;
    my ($ext) = @_;

    my @out;

    my $ready = $self->message_files('ready') or return;

    for my $msg (@$ready) {
        my $full = File::Spec->catfile($self->path, $msg);
        open(my $fh, '<', $full) or die "Could not open file '$full': $!";
        my $content = do { local $/; <$fh> };
        close($full);
        unlink($full) or die "Could not unlink file '$full': $!";

        my $msg = IPC::Manager::Message->new($self->{+SERIALIZER}->deserialize($content));
        push @out => $msg;

        $self->{+STATS}->{read}->{$msg->{from}}++;
    }

    push @out => $self->read_resume_file;

    return sort { $a->stamp <=> $b->stamp } @out;
}

sub _write_message_file {
    my $self = shift;
    my ($msg, $peer) = @_;

    $peer //= $msg->to or croak "Message has no peer";

    my $msg_dir  = $self->peer_exists($peer) or croak "Client does not exist";
    my $msg_file = File::Spec->catfile($msg_dir, $msg->id);

    my $pend  = "$msg_file.pend";
    my $ready = "$msg_file.ready";

    confess "Message file '$msg_file' already exists" if -e $pend || -e $ready;

    open(my $fh, '>', $pend) or die "Could not open '$pend': $!";

    print $fh $self->{+SERIALIZER}->serialize($msg);

    close($fh);

    rename($pend, $ready) or die "Could not rename file: $!";

    $self->{+STATS}->{sent}->{$msg->{to}}++;
    return $ready;
}

sub send_message {
    my $self = shift;
    my $msg  = $self->build_message(@_);
    $self->pid_check;
    $self->_write_message_file($msg);
}

1;

__END__