FFI/ZMQ4_1/Socket.pm

#
# Module Generated by Template::Tiny on Thu Mar 14 04:05:17 UTC 2019
#

package ZMQ::FFI::ZMQ4_1::Socket;
$ZMQ::FFI::ZMQ4_1::Socket::VERSION = '1.17';
use FFI::Platypus;
use FFI::Platypus::Buffer;
use FFI::Platypus::Memory qw(malloc free memcpy);

use Carp qw(croak carp);
use Try::Tiny;

use ZMQ::FFI::ZMQ4_1::Raw;
use ZMQ::FFI::Custom::Raw;
use ZMQ::FFI::Constants qw(:all);
use ZMQ::FFI::Util qw(current_tid);

use Moo;
use namespace::clean;

no if $] >= 5.018, warnings => "experimental";
use feature 'switch';

with qw(
    ZMQ::FFI::SocketRole
    ZMQ::FFI::ErrorHelper
    ZMQ::FFI::Versioner
);

my $FFI_LOADED;

sub BUILD {
    my ($self) = @_;

    unless ($FFI_LOADED) {
        ZMQ::FFI::Custom::Raw::load($self->soname);
        ZMQ::FFI::ZMQ4_1::Raw::load($self->soname);
        $FFI_LOADED = 1;
    }

    # force init zmq_msg_t
    $self->_zmq_msg_t;

    # ensure clean edge state
    while ( $self->has_pollin ) {
        $self->recv();
    }

    # set default linger
    $self->set_linger(0);
}


sub connect {
    my ($self, $endpoint) = @_;

    if ($_[0]->socket_ptr == -1) {
        carp "Operation on closed socket";
        return;
    }

    unless ($endpoint) {
        croak 'usage: $socket->connect($endpoint)';
    }

    $self->check_error(
        'zmq_connect',
        zmq_connect($self->socket_ptr, $endpoint)
    );
}

sub disconnect {
    my ($self, $endpoint) = @_;

    if ($_[0]->socket_ptr == -1) {
        carp "Operation on closed socket";
        return;
    }

    unless ($endpoint) {
        croak 'usage: $socket->disconnect($endpoint)';
    }

    $self->check_error(
        'zmq_disconnect',
        zmq_disconnect($self->socket_ptr, $endpoint)
    );
}

sub bind {
    my ($self, $endpoint) = @_;

    if ($_[0]->socket_ptr == -1) {
        carp "Operation on closed socket";
        return;
    }

    unless ($endpoint) {
        croak 'usage: $socket->bind($endpoint)'
    }

    $self->check_error(
        'zmq_bind',
        zmq_bind($self->socket_ptr, $endpoint)
    );
}

sub unbind {
    my ($self, $endpoint) = @_;

    if ($_[0]->socket_ptr == -1) {
        carp "Operation on closed socket";
        return;
    }

    unless ($endpoint) {
        croak 'usage: $socket->unbind($endpoint)';
    }

    $self->check_error(
        'zmq_unbind',
        zmq_unbind($self->socket_ptr, $endpoint)
    );
}

sub send {
    # 0: self
    # 1: data
    # 2: flags

    if ($_[0]->socket_ptr == -1) {
        carp "Operation on closed socket";
        return;
    }

    $_[0]->{last_errno} = 0;

    use bytes;
    my $length = length($_[1]);
    no bytes;

    if ( -1 == zmq_send($_[0]->socket_ptr, $_[1], $length, ($_[2] // 0)) ) {
        $_[0]->{last_errno} = zmq_errno();

        if ($_[0]->die_on_error) {
            $_[0]->fatal('zmq_send');
        }

        return;
    }
}

sub send_multipart {
    # 0: self
    # 1: partsref
    # 2: flags

    if ($_[0]->socket_ptr == -1) {
        carp "Operation on closed socket";
        return;
    }

    my @parts = @{$_[1] // []};
    unless (@parts) {
        croak 'usage: send_multipart($parts, $flags)';
    }

    for my $i (0..$#parts-1) {
        $_[0]->send($parts[$i], ($_[2] // 0) | ZMQ_SNDMORE);

        # don't need to explicitly check die_on_error
        # since send would have exploded if it was true
        if ($_[0]->has_error) {
            return;
        }
    }

    $_[0]->send($parts[$#parts], $_[2] // 0);
}

sub recv {
    # 0: self
    # 1: flags

    if ($_[0]->socket_ptr == -1) {
        carp "Operation on closed socket";
        return;
    }

    $_[0]->{last_errno} = 0;

    # retval = msg size
    my $retval = zmq_msg_recv($_[0]->{"_zmq_msg_t"}, $_[0]->socket_ptr, $_[1] // 0);

    if ( $retval == -1 ) {
        $_[0]->{last_errno} = zmq_errno();

        if ($_[0]->die_on_error) {
            $_[0]->fatal('zmq_msg_recv');
        }


        return;
    }

    if ($retval) {
        return buffer_to_scalar(zmq_msg_data($_[0]->{"_zmq_msg_t"}), $retval);
    }

    return '';
}

sub recv_multipart {
    # 0: self
    # 1: flags

    if ($_[0]->socket_ptr == -1) {
        carp "Operation on closed socket";
        return;
    }

    my @parts = ( $_[0]->recv($_[1]) );

    if ($_[0]->has_error) {
        return;
    }

    my $type = ($_[0]->version)[0] == 2 ? 'int64_t' : 'int';

    while ( $_[0]->get(ZMQ_RCVMORE, $type) ){
        push @parts, $_[0]->recv($_[1] // 0);

        # don't need to explicitly check die_on_error
        # since recv would have exploded if it was true
        if ($_[0]->has_error) {
            return;
        }
    }

    return @parts;
}

sub get_fd {
    if ($_[0]->socket_ptr == -1) {
        carp "Operation on closed socket";
        return;
    }

    return $_[0]->get(ZMQ_FD, 'int');
}

sub get_linger {
    if ($_[0]->socket_ptr == -1) {
        carp "Operation on closed socket";
        return;
    }

    return $_[0]->get(ZMQ_LINGER, 'int');
}

sub set_linger {
    my ($self, $linger) = @_;

    if ($_[0]->socket_ptr == -1) {
        carp "Operation on closed socket";
        return;
    }

    $self->set(ZMQ_LINGER, 'int', $linger);
}

sub get_identity {
    if ($_[0]->socket_ptr == -1) {
        carp "Operation on closed socket";
        return;
    }

    return $_[0]->get(ZMQ_IDENTITY, 'binary');
}

sub set_identity {
    my ($self, $id) = @_;

    if ($_[0]->socket_ptr == -1) {
        carp "Operation on closed socket";
        return;
    }

    $self->set(ZMQ_IDENTITY, 'binary', $id);
}

sub subscribe {
    my ($self, $topic) = @_;

    if ($_[0]->socket_ptr == -1) {
        carp "Operation on closed socket";
        return;
    }

    $self->set(ZMQ_SUBSCRIBE, 'binary', $topic);
}

sub unsubscribe {
    my ($self, $topic) = @_;

    if ($_[0]->socket_ptr == -1) {
        carp "Operation on closed socket";
        return;
    }

    $self->set(ZMQ_UNSUBSCRIBE, 'binary', $topic);
}

sub has_pollin {
    if ($_[0]->socket_ptr == -1) {
        carp "Operation on closed socket";
        return;
    }

    return $_[0]->get(ZMQ_EVENTS, 'int') & ZMQ_POLLIN;
}

sub has_pollout {
    if ($_[0]->socket_ptr == -1) {
        carp "Operation on closed socket";
        return;
    }

    return $_[0]->get(ZMQ_EVENTS, 'int') & ZMQ_POLLOUT;
}

sub get {
    my ($self, $opt, $opt_type) = @_;

    if ($_[0]->socket_ptr == -1) {
        carp "Operation on closed socket";
        return;
    }

    my $optval;
    my $optval_len;

    for ($opt_type) {
        when (/^(binary|string)$/) {
            # ZMQ_IDENTITY uses binary type and can be at most 255 bytes long
            #
            # ZMQ_LAST_ENDPOINT uses string type and expects a buffer large
            # enough to hold an endpoint string
            #
            # So for these cases 256 should be sufficient (including \0).
            # Other binary/string opts are being added all the time, and
            # hopefully this value scales, but we can always increase it if
            # necessary
            my $optval_ptr = malloc(256);
            $optval_len    = 256;

            $self->check_error(
                'zmq_getsockopt',
                zmq_getsockopt_binary(
                    $self->socket_ptr,
                    $opt,
                    $optval_ptr,
                    \$optval_len
                )
            );

            if ($self->has_error) {
                free($optval_ptr);
                return;
            }

            if ($opt_type eq 'binary') {
                $optval = buffer_to_scalar($optval_ptr, $optval_len);
                free($optval_ptr);
            }
            else { # string
                # FFI::Platypus already appends a null terminating byte for
                # strings, so strip the one included by zeromq (otherwise test
                # comparisons fail due to the extra NUL)
                $optval = buffer_to_scalar($optval_ptr, $optval_len-1);
                free($optval_ptr);
            }
        }

        when ('int') {
            $optval_len = $self->sockopt_sizes->{'int'};
            $self->check_error(
                'zmq_getsockopt',
                zmq_getsockopt_int(
                    $self->socket_ptr,
                    $opt,
                    \$optval,
                    \$optval_len
                )
            );
        }

        when ('int64_t') {
            $optval_len = $self->sockopt_sizes->{'sint64'};
            $self->check_error(
                'zmq_getsockopt',
                zmq_getsockopt_int64(
                    $self->socket_ptr,
                    $opt,
                    \$optval,
                    \$optval_len
                )
            );
        }

        when ('uint64_t') {
            $optval_len = $self->sockopt_sizes->{'uint64'};
            $self->check_error(
                'zmq_getsockopt',
                zmq_getsockopt_uint64(
                    $self->socket_ptr,
                    $opt,
                    \$optval,
                    \$optval_len
                )
            );
        }

        default {
            croak "unknown type $opt_type";
        }
    }

    if ($optval ne '') {
        return $optval;
    }

    return;
}

sub set {
    my ($self, $opt, $opt_type, $optval) = @_;

    if ($_[0]->socket_ptr == -1) {
        carp "Operation on closed socket";
        return;
    }

    for ($opt_type) {
        when (/^(binary|string)$/) {
            my ($optval_ptr, $optval_len) = scalar_to_buffer($optval);
            $self->check_error(
                'zmq_setsockopt',
                zmq_setsockopt_binary(
                    $self->socket_ptr,
                    $opt,
                    $optval_ptr,
                    $optval_len
                )
            );
        }

        when ('int') {
            $self->check_error(
                'zmq_setsockopt',
                zmq_setsockopt_int(
                    $self->socket_ptr,
                    $opt,
                    \$optval,
                    $self->sockopt_sizes->{'int'}
                )
            );
        }

        when ('int64_t') {
            $self->check_error(
                'zmq_setsockopt',
                zmq_setsockopt_int64(
                    $self->socket_ptr,
                    $opt,
                    \$optval,
                    $self->sockopt_sizes->{'sint64'}
                )
            );
        }

        when ('uint64_t') {
            $self->check_error(
                'zmq_setsockopt',
                zmq_setsockopt_uint64(
                    $self->socket_ptr,
                    $opt,
                    \$optval,
                    $self->sockopt_sizes->{'uint64'}
                )
            );
        }

        default {
            croak "unknown type $opt_type";
        }
    }

    return;
}

sub close {
    my ($self) = @_;

    if ($_[0]->socket_ptr == -1) {
        carp "Operation on closed socket";
        return;
    }

    # don't try to cleanup socket cloned from another thread
    return unless $self->_tid == current_tid();

    # don't try to cleanup socket copied from another process (fork)
    return unless $self->_pid == $$;

    $self->check_error(
        'zmq_msg_close',
        zmq_msg_close($self->_zmq_msg_t)
    );

    $self->check_error(
        'zmq_close',
        zmq_close($self->socket_ptr)
    );

    $self->socket_ptr(-1);
}


sub DEMOLISH {
    my ($self) = @_;

    # remove ourselves from the context object so that we dont leak
    $self->context->_remove_socket($self) if (defined $self->context);

    return if $self->socket_ptr == -1;

    $self->close();
}

1;

# vim:ft=perl

__END__

=pod

=encoding UTF-8

=head1 NAME

ZMQ::FFI::ZMQ4_1::Socket

=head1 VERSION

version 1.17

=head1 AUTHOR

Dylan Cali <calid1984@gmail.com>

=head1 COPYRIGHT AND LICENSE

This software is copyright (c) 2019 by Dylan Cali.

This is free software; you can redistribute it and/or modify it under
the same terms as the Perl 5 programming language system itself.

=cut