File Coverage

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

Criterion	Covered	Total	%
statement	61	69	88.4
branch	6	14	42.8
condition	3	5	60.0
subroutine	17	19	89.4
pod	7	10	70.0
total	94	117	80.3

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

our $VERSION = '0.000005';

use File::Spec;
use Carp qw/croak/;
use POSIX qw/mkfifo/;
use IO::Socket::UNIX qw/SOCK_DGRAM/;
use IO::Select;

use parent 'IPC::Manager::Base::FS::Handle';
use Object::HashBase qw{
    +buffer
    +socket
    +socket_cache
};

sub check_path { -S $_[1] }
sub path_type  { 'UNIX Socket' }

sub handles_for_select { $_[0]->{+SOCKET} }

sub suspend { croak "suspend is not supported by the UnixSocket driver" }

sub make_path {
    my $self = shift;
    my $path = $self->path;

    my $s = IO::Socket::UNIX->new(
        Type     => SOCK_DGRAM,
        Local    => $path,
        Blocking => 0,
    ) or die "Cannot create reader socket: $!";

    $self->{+SOCKET} = $s;
}

sub pre_disconnect_hook {
    my $self = shift;
    unlink($self->{+PATH}) or warn "Could not unlink socket: $!";
}

sub init {
    my $self = shift;

    $self->{+BUFFER} //= [];

    $self->SUPER::init();
}

sub fill_buffer {
    my $self = shift;

    my $s = $self->{+SOCKET};
    while (my $msg = <$s>) {
        push @{$self->{+BUFFER}} => $msg;
    }

    return @{$self->{+BUFFER}} ? 1 : 0;
}

sub get_messages {
    my $self = shift;

    my @out;

    push @out => $self->read_resume_file;
    push @out => @{$self->{+BUFFER}};

    my $s = $self->{+SOCKET};
    while (my $msg = <$s>) {
        $msg = IPC::Manager::Message->new($self->{+SERIALIZER}->deserialize($msg));
        push @out => $msg;
        $self->{+STATS}->{read}->{$msg->{from}}++;
    }

    @{$self->{+BUFFER}} = ();

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

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

    my $peer_id = $msg->to or croak "No peer specified";

    $self->pid_check;
    my $sock = $self->peer_exists($peer_id) or die "'$peer_id' is not a valid message recipient";

    my $s = $self->{+SOCKET_CACHE}->{$sock} //= IO::Socket::UNIX->new(
        Type => SOCK_DGRAM,
        Peer => $sock,
    ) or die "Cannot connect to socket: $!";

    $s->send($self->{+SERIALIZER}->serialize($msg) . "\n") or die "Cannot send message: $!";

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

1;

__END__