NAME

VERSION

SYNOPSIS

    #### send/recv ####

    use v5.10;
    use ZMQ::FFI qw(ZMQ_REQ ZMQ_REP);

    my $endpoint = "ipc://zmq-ffi-$$";
    my $ctx      = ZMQ::FFI->new();

    my $s1 = $ctx->socket(ZMQ_REQ);
    $s1->connect($endpoint);

    my $s2 = $ctx->socket(ZMQ_REP);
    $s2->bind($endpoint);

    $s1->send('ohhai');

    say $s2->recv();
    # ohhai


    #### pub/sub ####

    use v5.10;
    use ZMQ::FFI qw(ZMQ_PUB ZMQ_SUB);
    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('');

        until ($s->has_pollin) {
            # compensate for slow subscriber
            usleep 100_000;
            $p->send('ohhai');
        }

        say $s->recv();
        # ohhai

        $s->unsubscribe('');
    }

    # specific topics
    {
        $s->subscribe('topic1');
        $s->subscribe('topic2');

        until ($s->has_pollin) {
            usleep 100_000;
            $p->send('topic1 ohhai');
            $p->send('topic2 ohhai');
        }

        while ($s->has_pollin) {
            say join ' ', $s->recv();
            # topic1 ohhai
            # topic2 ohhai
        }
    }


    #### multipart ####

    use v5.10;
    use ZMQ::FFI qw(ZMQ_DEALER ZMQ_ROUTER);

    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)]);

    say join ' ', $r->recv_multipart;
    # dealer ABC DEF GHI


    #### nonblocking ####

    use v5.10;
    use ZMQ::FFI 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::break();
            }
        }
    };


    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();


    #### specifying versions ####

    use ZMQ::FFI;

    # 2.x context
    my $ctx = ZMQ::FFI->new( soname => 'libzmq.so.1' );
    my ($major, $minor, $patch) = $ctx->version;

    # 3.x context
    my $ctx = ZMQ::FFI->new( soname => 'libzmq.so.3' );
    my ($major, $minor, $patch) = $ctx->version;

DESCRIPTION

As of 1.00 ZMQ::FFI is implemented using FFI::Platypus. This version has substantial performance improvements and you are encouraged to use 1.00 or newer.

CONTEXT API

    my $ctx = ZMQ::FFI->new(%options);

    my ($major, $minor, $patch) = $ctx->version();

    my $threads = $ctx->get(ZMQ_IO_THREADS)

    $ctx->set(ZMQ_MAX_SOCKETS, 42)

    my $socket = $ctx->socket(ZMQ_REQ)

    $ctx->proxy($frontend, $backend);

    $ctx->proxy($frontend, $backend, $capture);

    $ctx->device($type, $frontend, $backend);

SOCKET API

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.

    my ($major, $minor, $patch) = $socket->version();

    $socket->connect($endpoint);

    $socket->disconnect($endpoint);

    $socket->bind($endpoint);

    $socket->unbind($endpoint);

    my $linger = $socket->get_linger();

    $socket->set_linger($millis);

    my $ident = $socket->get_identity();

    $socket->set_identity($ident);

    my $fd = $socket->get_fd();

    my $option_value = $socket->get($option_name, $option_type);

    my $linger = $socket->get(ZMQ_LINGER, 'int');

    $socket->set($option_name, $option_type, $option_value);

    $socket->set(ZMQ_IDENTITY, 'binary', 'foo');

    $socket->subscribe($topic);

    $socket->unsubscribe($topic);

    $socket->send($msg);

    $socket->send($msg, $flags);

    $socket->send($parts_aref);

    $socket->send($parts_aref, $flags);

    my $msg = $socket->recv();

    my $msg = $socket->recv($flags);

    my @parts = $socket->recv_multipart();

    my @parts = $socket->recv_multipart($flags);

    while ( $socket->has_pollin ) { ... }

    $socket->die_on_error(0);

    $socket->die_on_error(1);

CLEANUP

• it uses 0 linger by default (this is the default used by czmq <https://github.com/zeromq/czmq> and jzmq <https://github.com/zeromq/jzmq>)
• during object destruction it will call close/destroy for you
• it arranges the reference hierarchy such that sockets will be properly cleaned up before their associated contexts
• it detects fork/thread situations and ensures sockets/contexts are only cleaned up in their originating process/thread
• it guards against double closes/destroys

Given the above you're probably better off letting "ZMQ::FFI" handle cleanup for you. But if for some reason you want to do explicit cleanup yourself you can. All the below will accomplish the same thing:

    # implicit cleanup
    {
        my $context = ZMQ::FFI->new();
        my $socket  = $ctx->socket($type);
        ...
        # close/destroy called in destructors at end of scope
    }

    # explicit cleanup
    $socket->close();
    $context->destroy();

    # ditto
    undef $socket;
    undef $context;

Regarding "linger", you can always set this to a value you prefer if you don't like the default. Once set the new value will be used when the socket is subsequently closed (either implicitly or explicitly):

    $socket->set_linger(-1); # infinite linger
                             # $context->destroy will block forever
                             # (or until all pending messages have been sent)

ERROR HANDLING

    $ctx->socket(-1);
    # dies with 'zmq_socket: Invalid argument'

Usually this is what you want, but not always. Some zmq operations can return errors that are not fatal and should be handled. For example using "ZMQ_DONTWAIT" with send/recv can return "EAGAIN" and simply means try again, not die.

For situations such as this you can turn off exceptional error handling by setting "die_on_error" to 0. It is then for you to check and manage any zmq errors by checking "last_errno":

    use Errno qw(EAGAIN);

    my $ctx = ZMQ::FFI->new();
    my $s   = $ctx->socket(ZMQ_DEALER);
    $s->bind('tcp://*:7200');

    $s->die_on_error(0); # turn off exceptional error handling

    while (1) {
        my $msg = $s->recv(ZMQ_DONTWAIT);

        if ($s->last_errno == EAGAIN) {
            sleep 1;
        }
        elsif ($s->last_errno) {
            die $s->last_strerror;
        }
        else {
            warn "recvd: $msg";
            last;
        }
    }

    $s->die_on_error(1); # turn back on exceptional error handling

FFI VS XS PERFORMANCE

Testing indicates FFI::Platypus xsubs are around 30% slower than "real" XS xsubs. That may sound like a lot, but to put it in perspective that means, for zeromq, the XS bindings can send 10 million messages 1-2 seconds faster than the ffi ones.

If you really care about 1-2 seconds over 10 million messages you should be writing your solution in C anyways. An equivalent C implementation will be several hundred percent faster or more.

Keep in mind also that the small speed bump you get using XS can easily be wiped out by crappy and poorly optimized Perl code.

Now that Perl finally has a great ffi interface, it is hard to make the case to continue using XS. The slight speed bump just isn't worth giving up the convenience, flexibility, and portability of ffi.

You can find the detailed performance results that informed this section at: <https://gist.github.com/calid/17df5bcfb81c83786d6f>

BUGS

SEE ALSO

• ZMQ::FFI::Constants
• ZMQ::FFI::Util
• FFI::Platypus
• FFI::Raw
• ZMQ::LibZMQ3

AUTHOR

COPYRIGHT AND LICENSE

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