ZMQ::FFI - zeromq bindings using libffi and FFI::Raw
version 0.03
#### send/recv #### use v5.10; use ZMQ::FFI; use ZMQ::FFI::Constants qw(ZMQ_REQ ZMQ_REP ZMQ_DONTWAIT); my $endpoint = "ipc://zmq-ffi-$$"; my $ctx = ZMQ::FFI->new( threads => 1 ); my $s1 = $ctx->socket(ZMQ_REQ); $s1->connect($endpoint); my $s2 = $ctx->socket(ZMQ_REP); $s2->bind($endpoint); $s1->send('ohhai', ZMQ_DONTWAIT); say $s2->recv(); # ohhai #### pub/sub #### use v5.10; use ZMQ::FFI; use ZMQ::FFI::Constants qw(ZMQ_PUB ZMQ_SUB ZMQ_DONTWAIT); use Time::HiRes q(usleep); my $endpoint = "ipc://zmq-ffi-$$"; my $ctx = ZMQ::FFI->new(); my $s = $ctx->socket(ZMQ_SUB); my $p = $ctx->socket(ZMQ_PUB); $s->connect($endpoint); $p->bind($endpoint); # all topics { $s->subscribe(''); $p->send('ohhai', ZMQ_DONTWAIT); until ($s->has_pollin) { # compensate for slow subscriber usleep 100_000; $p->send('ohhai', ZMQ_DONTWAIT); } say $s->recv(); # ohhai $s->unsubscribe(''); } # specific topics { $s->subscribe('topic1'); $s->subscribe('topic2'); $p->send('topic1 ohhai', ZMQ_DONTWAIT); $p->send('topic2 ohhai', ZMQ_DONTWAIT); until ($s->has_pollin) { usleep 100_000; $p->send('topic1 ohhai', ZMQ_DONTWAIT); $p->send('topic2 ohhai', ZMQ_DONTWAIT); } while ($s->has_pollin) { say join ' ', $s->recv(); # topic1 ohhai # topic2 ohhai } } #### multipart #### use v5.10; use ZMQ::FFI; use ZMQ::FFI::Constants qw(ZMQ_DEALER ZMQ_ROUTER ZMQ_DONTWAIT); my $endpoint = "ipc://zmq-ffi-$$"; my $ctx = ZMQ::FFI->new(); my $d = $ctx->socket(ZMQ_DEALER); $d->set_identity('dealer'); my $r = $ctx->socket(ZMQ_ROUTER); $d->connect($endpoint); $r->bind($endpoint); $d->send_multipart([qw(ABC DEF GHI)], ZMQ_DONTWAIT); say join ' ', $r->recv_multipart; # dealer ABC DEF GHI #### nonblocking #### use v5.10; use ZMQ::FFI; use ZMQ::FFI::Constants qw(ZMQ_PUSH ZMQ_PULL); use AnyEvent; use EV; my $endpoint = "ipc://zmq-ffi-$$"; my $ctx = ZMQ::FFI->new(); my @messages = qw(foo bar baz); my $pull = $ctx->socket(ZMQ_PULL); $pull->bind($endpoint); my $fd = $pull->get_fd(); my $recv = 0; my $w = AE::io $fd, 0, sub { while ( $pull->has_pollin ) { say $pull->recv(); # foo, bar, baz $recv++; if ($recv == 3) { EV::unloop(); } } }; my $push = $ctx->socket(ZMQ_PUSH); $push->connect($endpoint); my $sent = 0; my $t; $t = AE::timer 0, .1, sub { $push->send($messages[$sent]); $sent++; if ($sent == 3) { undef $t; } }; EV::run();
ZMQ::FFI exposes a high level, transparent, OO interface to zeromq independent of the underlying libzmq version. Where semantics differ, it will dispatch to the appropriate backend for you. As it uses ffi, there is no dependency on XS or compilation.
ZMQ::FFI->new( threads => 42, max_sockets => 42 )
returns a new context object, appropriate for the version of libzmq found on your system. It accepts the following optional attributes:
zeromq thread pool size. Default: 1
only for zeromq 3.x
max number of sockets allowed for context. Default: 1024
return the libzmq version as the string 'X.Y.Z'. You can also call ZMQ::FFI::Util::zmq_version directly if you want the raw list of ($major, $minor, $patch) without needing to split
split
$ctx->get(ZMQ_IO_THREADS)
get a context option value
$ctx->set(ZMQ_MAX_SOCKETS, 42)
set a context option value
$ctx->socket(ZMQ_REQ)
returns a socket of the specified type. See "SOCKET API" below
destroys the underlying zmq context. This is called automatically when the object goes out of scope
The following API is available on socket objects created by $ctx->socket.
$ctx->socket
For core attributes and functions, common across all versions of zeromq, convenience methods are provided. Otherwise, generic get/set methods are provided that will work independent of version.
As attributes are constantly being added/removed from zeromq, it is unlikely the 'static' accessors will grow much beyond the current set.
does socket connect on the specified endpoint
does socket bind on the specified endpoint
get or set the current socket linger period
get or set the socket identity for request/reply patterns
get the file descriptor associated with the socket
add $topic to the subscription list
$topic
remove $topic from the subscription list
$socket->send('ohhai', ZMQ_DONTWAIT)
sends a message using the optional flags
$socket->send([qw(foo bar baz)])
given an array ref of message parts, sends the multipart message using the optional flags. ZMQ_SNDMORE semantics are handled for you
receive a message
receives a multipart message, returning an array of parts. ZMQ_RCVMORE semantics are handled for you
checks ZMQ_EVENTS for ZMQ_POLLIN and ZMQ_POLLOUT respectively, and returns true/false depending on the state
$socket->get(ZMQ_LINGER, 'int')
return the value for the specified socket option. $option_type is the type associated with the option value in the zeromq API (zmq_getsockopt man page)
$option_type
zmq_getsockopt
$socket->set(ZMQ_IDENTITY, 'binary', 'foo')
set the socket option to the specified value. $option_type is the type associated with the option value in the zeromq API (zmq_setsockopt man page)
zmq_setsockopt
close the underlying zmq socket. This is called automatically when the object goes out of scope
ZMQ::FFI checks the return codes of underlying zmq functions for you, and in the case of an error it will die with the plain english system error message.
$ctx->socket(-1); # dies with 'zmq_socket: Invalid argument'
ZMQ::FFI::Constants
ZMQ::FFI::Util
Dylan Cali <calid1984@gmail.com
This software is copyright (c) 2013 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.
To install ZMQ::FFI, copy and paste the appropriate command in to your terminal.
cpanm
cpanm ZMQ::FFI
CPAN shell
perl -MCPAN -e shell install ZMQ::FFI
For more information on module installation, please visit the detailed CPAN module installation guide.