view release on metacpan or  search on metacpan

lib/AnyEvent/ZeroMQ/Handle.pm  view on Meta::CPAN

package AnyEvent::ZeroMQ::Handle;
BEGIN {
  $AnyEvent::ZeroMQ::Handle::VERSION = '0.01';
}
# ABSTRACT: AnyEvent::Handle-like interface for 0MQ sockets
use Moose;

use AnyEvent::ZeroMQ;
use AnyEvent::ZeroMQ::Types qw(IdentityStr);
use ZeroMQ::Raw::Constants qw(ZMQ_NOBLOCK ZMQ_IDENTITY);

use Params::Util qw(_CODELIKE);
use Scalar::Util qw(weaken);
use Try::Tiny;
use POSIX qw(EAGAIN EWOULDBLOCK);

use true;
use namespace::autoclean;

has 'socket' => (
    is       => 'ro',
    isa      => 'ZeroMQ::Raw::Socket',
    handles  => [qw/bind connect/],
    required => 1,
);

before qw/bind connect/ => sub {
    $_[0]->identity;
};

after qw/bind connect/ => sub {
    my $self = shift;
    # this can change readability/writability status, so do the checks
    # again
    $self->read;
    $self->write;
};

has 'identity' => (
    is         => 'rw',    # note: you can change this, but it has
                           # no effect until a new bind/connect.
    isa        => IdentityStr,
    lazy_build => 1,
    trigger    => sub { shift->_change_identity(@_) },
);

has 'on_read' => (
    is        => 'rw',
    isa       => 'CodeRef',
    predicate => 'has_on_read',
    clearer   => 'clear_on_read',
    trigger   => sub { $_[0]->read },
);

has 'on_error' => (
    is        => 'rw',
    isa       => 'CodeRef',
    predicate => 'has_on_error',
    clearer   => 'clear_on_error',
);

sub handle_error {
    my ($self, $str) = @_;
    return $self->on_error->($str)
        if $self->has_on_error;

    warn "AnyEvent::ZeroMQ::Handle: error in callback (ignoring): $str";
}

has 'on_drain' => (
    is        => 'rw',
    isa       => 'CodeRef',
    predicate => 'has_on_drain',
    clearer   => 'clear_on_drain',