//  This is a stable class, and may not change except for emergencies. It
    //  is provided in stable builds.
    //  This class has draft methods, which may change over time. They are not
    //  in stable releases, by default. Use --enable-drafts to enable.
    // Actors get a pipe and arguments from caller
    typedef void (zactor_fn) (
        zsock_t *pipe, void *args);
    
    //  Create a new actor passing arbitrary arguments reference.
    CZMQ_EXPORT zactor_t *
        zactor_new (zactor_fn task, void *args);
    
    //  Destroy an actor.
    CZMQ_EXPORT void
        zactor_destroy (zactor_t **self_p);
    
    //  Send a zmsg message to the actor, take ownership of the message
    //  and destroy when it has been sent.
    CZMQ_EXPORT int
        zactor_send (zactor_t *self, zmsg_t **msg_p);
    
    //  Receive a zmsg message from the actor. Returns NULL if the actor
    //  was interrupted before the message could be received, or if there
    //  was a timeout on the actor.
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT zmsg_t *
        zactor_recv (zactor_t *self);
    
    //  Probe the supplied object, and report if it looks like a zactor_t.
    CZMQ_EXPORT bool
        zactor_is (void *self);
    
    //  Probe the supplied reference. If it looks like a zactor_t instance,
    //  return the underlying libzmq actor handle; else if it looks like
    //  a libzmq actor handle, return the supplied value.
    CZMQ_EXPORT void *
        zactor_resolve (void *self);
    
    //  Return the actor's zsock handle. Use this when you absolutely need
    //  to work with the zsock instance rather than the actor.
    CZMQ_EXPORT zsock_t *
        zactor_sock (zactor_t *self);
    
    //  Self test of this class.
    CZMQ_EXPORT void
        zactor_test (bool verbose);
    
    #ifdef CZMQ_BUILD_DRAFT_API
    // Function to be called on zactor_destroy. Default behavior is to send zmsg_t with string "$TERM" in a first frame.
    //
    // An example - to send $KTHXBAI string
    //
    //     if (zstr_send (self, "$KTHXBAI") == 0)
    //         zsock_wait (self);
    typedef void (zactor_destructor_fn) (
        zactor_t *self);
    
    //  *** Draft method, for development use, may change without warning ***
    //  Change default destructor by custom function. Actor MUST be able to handle new message instead of default $TERM.
    CZMQ_EXPORT void
        zactor_set_destructor (zactor_t *self, zactor_destructor_fn destructor);
    
    #endif // CZMQ_BUILD_DRAFT_API

    zactor_t *actor = zactor_new (echo_actor, "Hello, World");
    assert (actor);
    zstr_sendx (actor, "ECHO", "This is a string", NULL);
    char *string = zstr_recv (actor);
    assert (streq (string, "This is a string"));
    freen (string);
    zactor_destroy (&actor);
    
    // custom destructor
    // KTHXBAI_actor ends on "$KTHXBAI" string
    zactor_t *KTHXBAI = zactor_new (KTHXBAI_actor, NULL);
    assert (KTHXBAI);
    // which is the one sent by KTHXBAI_destructor
    zactor_set_destructor (KTHXBAI, KTHXBAI_destructor);
    zactor_destroy (&KTHXBAI);
    
    // custom destructor
    // destructor using bsend/brecv
    zactor_t *BSEND = zactor_new (BSEND_actor, NULL);
    assert (BSEND);
    zactor_set_destructor (BSEND, BSEND_destructor);
    zactor_destroy (&BSEND);
    #if defined (__WINDOWS__)
    zsys_shutdown();
    #endif

    #define CURVE_ALLOW_ANY "*"
    
    //  CZMQ v3 API (for use with zsock, not zsocket, which is deprecated).
    //
    //  Create new zauth actor instance. This installs authentication on all 
    //  zsock sockets. Until you add policies, all incoming NULL connections are
    //  allowed (classic ZeroMQ behaviour), and all PLAIN and CURVE connections
    //  are denied:
    //  
    //      zactor_t *auth = zactor_new (zauth, NULL);
    //
    //  Destroy zauth instance. This removes authentication and allows all
    //  connections to pass, without authentication:
    //  
    //      zactor_destroy (&auth);
    //
    //  Note that all zauth commands are synchronous, so your application always
    //  waits for a signal from the actor after each command.
    //
    //  Enable verbose logging of commands and activity. Verbose logging can help
    //  debug non-trivial authentication policies:
    //
    //      zstr_send (auth, "VERBOSE");
    //      zsock_wait (auth);
    //
    //  Allow a list of IP addresses. For NULL, all clients from
    //  these addresses will be accepted. For PLAIN and CURVE, they will be
    //  allowed to continue with authentication. You can call this method
    //  multiple times to allow more IP addresses. If you allow one
    //  or more addresses, any non-allowed addresses are treated as
    //  blocked:
    //  
    //      zstr_sendx (auth, "ALLOW", "127.0.0.1", "127.0.0.2", NULL);
    //      zsock_wait (auth);
    //  
    //  Deny (block) a list of IP addresses. For all security mechanisms,
    //  this rejects the connection without any further authentication. Use
    //  either an allowed list, or a blocked list, not not both. If you define both
    //  an allowed list and a block list, only the allowed list takes effect:
    //  
    //      zstr_sendx (auth, "DENY", "192.168.0.1", "192.168.0.2", NULL);
    //      zsock_wait (auth);
    //
    //  Configure PLAIN authentication using a plain-text password file. You can
    //  modify the password file at any time; zauth will reload it automatically
    //  if modified externally:
    //  
    //      zstr_sendx (auth, "PLAIN", filename, NULL);
    //      zsock_wait (auth);
    //
    //  Configure CURVE authentication, using a directory that holds all public
    //  client certificates, i.e. their public keys. The certificates must be in
    //  zcert_save format. You can add and remove certificates in that directory
    //  at any time. To allow all client keys without checking, specify
    //  CURVE_ALLOW_ANY for the directory name:
    //
    //      zstr_sendx (auth, "CURVE", directory, NULL);
    //      zsock_wait (auth);
    //
    //  Configure GSSAPI authentication, using an underlying mechanism (usually
    //  Kerberos) to establish a secure context and perform mutual authentication:
    //
    //      zstr_sendx (auth, "GSSAPI", NULL);
    //      zsock_wait (auth);
    //
    //  This is the zauth constructor as a zactor_fn:
    CZMQ_EXPORT void
        zauth (zsock_t *pipe, void *certstore);
    
    //  Selftest
    CZMQ_EXPORT void
        zauth_test (bool verbose);

    
    const char *SELFTEST_DIR_RW = "src/selftest-rw";
    
    const char *testbasedir  = ".test_zauth";
    const char *testpassfile = "password-file";
    const char *testcertfile = "mycert.txt";
    char *basedirpath = NULL;   // subdir in a test, under SELFTEST_DIR_RW
    char *passfilepath = NULL;  // pathname to testfile in a test, in dirpath
    char *certfilepath = NULL;  // pathname to testfile in a test, in dirpath
    
    basedirpath = zsys_sprintf ("%s/%s", SELFTEST_DIR_RW, testbasedir);
    assert (basedirpath);
    passfilepath = zsys_sprintf ("%s/%s", basedirpath, testpassfile);
    assert (passfilepath);
    certfilepath = zsys_sprintf ("%s/%s", basedirpath, testcertfile);
    assert (certfilepath);
    
    // Make sure old aborted tests do not hinder us
    zdir_t *dir = zdir_new (basedirpath, NULL);
    if (dir) {
        zdir_remove (dir, true);
        zdir_destroy (&dir);
    }
    zsys_file_delete (passfilepath);
    zsys_file_delete (certfilepath);
    zsys_dir_delete  (basedirpath);
    
    //  Create temporary directory for test files
    zsys_dir_create (basedirpath);
    
    //  Check there's no authentication
    zsock_t *server = zsock_new (ZMQ_PULL);
    assert (server);
    zsock_t *client = zsock_new (ZMQ_PUSH);
    assert (client);
    bool success = s_can_connect (&server, &client, true);
    assert (success);
    
    //  Install the authenticator
    zactor_t *auth = zactor_new (zauth, NULL);
    assert (auth);
    if (verbose) {
        zstr_sendx (auth, "VERBOSE", NULL);
        zsock_wait (auth);
    }
    //  Check there's no authentication on a default NULL server
    success = s_can_connect (&server, &client, true);
    assert (success);
    
    //  When we set a domain on the server, we switch on authentication
    //  for NULL sockets, but with no policies, the client connection
    //  will be allowed.
    zsock_set_zap_domain (server, "global");
    success = s_can_connect (&server, &client, true);
    assert (success);
    
    //  Block 127.0.0.1, connection should fail
    zsock_set_zap_domain (server, "global");
    zstr_sendx (auth, "DENY", "127.0.0.1", NULL);
    zsock_wait (auth);
    success = s_can_connect (&server, &client, true);
    assert (!success);
    
    //  Allow our address, which overrides the block list
    zsock_set_zap_domain (server, "global");
    zstr_sendx (auth, "ALLOW", "127.0.0.1", NULL);
    zsock_wait (auth);
    success = s_can_connect (&server, &client, true);
    assert (success);
    
    //  Try PLAIN authentication
    zsock_set_zap_domain (server, "global");
    zsock_set_plain_server (server, 1);
    zsock_set_plain_username (client, "admin");
    zsock_set_plain_password (client, "Password");
    success = s_can_connect (&server, &client, true);
    assert (!success);
    
    FILE *password = fopen (passfilepath, "w");
    assert (password);
    fprintf (password, "admin=Password\n");
    fclose (password);
    zsock_set_zap_domain (server, "global");
    zsock_set_plain_server (server, 1);
    zsock_set_plain_username (client, "admin");
    zsock_set_plain_password (client, "Password");
    zstr_sendx (auth, "PLAIN", passfilepath, NULL);
    zsock_wait (auth);
    success = s_can_connect (&server, &client, false);
    assert (success);
    
    #if (ZMQ_VERSION >= ZMQ_MAKE_VERSION (4, 1, 0))
    // Test that the User-Id metadata is present
    zframe_t *frame = zframe_recv (server);
    assert (frame != NULL);
    const char *user_id = zframe_meta (frame, "User-Id");
    assert (user_id != NULL);
    assert (streq (user_id, "admin"));
    zframe_destroy (&frame);
    #endif
    s_renew_sockets(&server, &client);
    
    zsock_set_zap_domain (server, "global");
    zsock_set_plain_server (server, 1);
    zsock_set_plain_username (client, "admin");
    zsock_set_plain_password (client, "Bogus");
    success = s_can_connect (&server, &client, true);
    assert (!success);
    
    if (zsys_has_curve ()) {
        //  Try CURVE authentication
        //  We'll create two new certificates and save the client public
        //  certificate on disk; in a real case we'd transfer this securely
        //  from the client machine to the server machine.
        zcert_t *server_cert = zcert_new ();
        assert (server_cert);
        zcert_t *client_cert = zcert_new ();
        assert (client_cert);
        const char *server_key = zcert_public_txt (server_cert);
    
        //  Test without setting-up any authentication
        zcert_apply (server_cert, server);
        zcert_apply (client_cert, client);
        zsock_set_curve_server (server, 1);
        zsock_set_curve_serverkey (client, server_key);
        zsock_set_zap_domain (server, "global");
        success = s_can_connect (&server, &client, true);
        assert (!success);
    
        //  Test CURVE_ALLOW_ANY
        zcert_apply (server_cert, server);
        zcert_apply (client_cert, client);
        zsock_set_curve_server (server, 1);
        zsock_set_curve_serverkey (client, server_key);
        zstr_sendx (auth, "CURVE", CURVE_ALLOW_ANY, NULL);
        zsock_wait (auth);
        success = s_can_connect (&server, &client, true);
        assert (success);
    
        //  Test full client authentication using certificates
        zcert_set_meta (client_cert, "Hello", "%s", "World!");
        zcert_apply (server_cert, server);
        zcert_apply (client_cert, client);
        zsock_set_curve_server (server, 1);
        zsock_set_curve_serverkey (client, server_key);
        zcert_save_public (client_cert, certfilepath);
        zstr_sendx (auth, "CURVE", basedirpath, NULL);
        zsock_wait (auth);
        zsock_set_zap_domain (server, "global");
        success = s_can_connect (&server, &client, false);
        assert (success);
    
    #if (ZMQ_VERSION >= ZMQ_MAKE_VERSION (4, 1, 0))
        // Test send/recv certificate metadata
        zframe_t *frame = zframe_recv (server);
        assert (frame != NULL);
        const char *meta = zframe_meta (frame, "Hello");
        assert (meta != NULL);
        assert (streq (meta, "World!"));
        const char *user_id = zframe_meta (frame, "User-Id");
        assert (user_id != NULL);
        assert (streq (user_id, zcert_public_txt(client_cert)));
        zframe_destroy (&frame);
        s_renew_sockets(&server, &client);
    #endif
    
        zcert_destroy (&server_cert);
        zcert_destroy (&client_cert);
    
        // Test custom zcertstore
        zcertstore_t *certstore = zcertstore_new (NULL);
        zcertstore_set_loader (certstore, s_test_loader, NULL, NULL);
        zactor_destroy(&auth);
        auth = zactor_new (zauth, certstore);
        assert (auth);
        if (verbose) {
            zstr_sendx (auth, "VERBOSE", NULL);
            zsock_wait (auth);
        }
    
        byte public_key [32] = { 105, 76, 150, 58, 214, 191, 218, 65, 50, 172,
                                 131, 188, 247, 211, 136, 170, 227, 26, 57, 170,
                                 185, 63, 246, 225, 177, 230, 12, 8, 134, 136,
                                 105, 106 };
        byte secret_key [32] = { 245, 217, 172, 73, 106, 28, 195, 17, 218, 132,
                                 135, 209, 99, 240, 98, 232, 7, 137, 244, 100,
                                 242, 23, 29, 114, 70, 223, 83, 1, 113, 207,
                                 132, 149 };
        zcert_t *shared_cert = zcert_new_from (public_key, secret_key);
        assert (shared_cert);
        zcert_apply (shared_cert, server);
        zcert_apply (shared_cert, client);
        zsock_set_curve_server (server, 1);
        zsock_set_curve_serverkey (client, "x?T*N/1Y{8goubv{Ts}#&#f}TXJ//DVe#D2HkoLU");
        success = s_can_connect (&server, &client, true);
        assert (success);
        zcert_destroy (&shared_cert);
    }
    //  Remove the authenticator and check a normal connection works
    zactor_destroy (&auth);
    success = s_can_connect (&server, &client, true);
    assert (success);
    
    zsock_destroy (&client);
    zsock_destroy (&server);
    
    //  Delete all test files
    dir = zdir_new (basedirpath, NULL);
    assert (dir);
    zdir_remove (dir, true);
    zdir_destroy (&dir);
    
    zstr_free (&passfilepath);
    zstr_free (&certfilepath);
    zstr_free (&basedirpath);
    
    #endif
    
    #if defined (__WINDOWS__)
    zsys_shutdown();
    #endif
    

    //  Create new zbeacon actor instance:
    //
    //      zactor_t *beacon = zactor_new (zbeacon, NULL);
    //
    //  Destroy zbeacon instance:
    //
    //      zactor_destroy (&beacon);
    //
    //  Enable verbose logging of commands and activity:
    //
    //      zstr_send (beacon, "VERBOSE");
    //
    //  Configure beacon to run on specified UDP port, and return the name of
    //  the host, which can be used as endpoint for incoming connections. To
    //  force the beacon to operate on a given interface, set the environment
    //  variable ZSYS_INTERFACE, or call zsys_set_interface() before creating
    //  the beacon. If the system does not support UDP broadcasts (lacking a
    //  workable interface), returns an empty hostname:
    //
    //      //  Pictures: 's' = C string, 'i' = int
    //      zsock_send (beacon, "si", "CONFIGURE", port_number);
    //      char *hostname = zstr_recv (beacon);
    //
    //  Start broadcasting a beacon at a specified interval in msec. The beacon
    //  data can be at most UDP_FRAME_MAX bytes; this constant is defined in
    //  zsys.h to be 255:
    //
    //      //  Pictures: 'b' = byte * data + size_t size
    //      zsock_send (beacon, "sbi", "PUBLISH", data, size, interval);
    //
    //  Stop broadcasting the beacon:
    //
    //      zstr_sendx (beacon, "SILENCE", NULL);
    //
    //  Start listening to beacons from peers. The filter is used to do a prefix
    //  match on received beacons, to remove junk. Note that any received data
    //  that is identical to our broadcast beacon_data is discarded in any case.
    //  If the filter size is zero, we get all peer beacons:
    //  
    //      zsock_send (beacon, "sb", "SUBSCRIBE", filter_data, filter_size);
    //
    //  Stop listening to other peers
    //
    //      zstr_sendx (beacon, "UNSUBSCRIBE", NULL);
    //
    //  Receive next beacon from a peer. Received beacons are always a 2-frame
    //  message containing the ipaddress of the sender, and then the binary
    //  beacon data as published by the sender:
    //
    //      zmsg_t *msg = zmsg_recv (beacon);
    //
    //  This is the zbeacon constructor as a zactor_fn:
    CZMQ_EXPORT void
        zbeacon (zsock_t *pipe, void *unused);
    
    //  Self test of this class
    CZMQ_EXPORT void
        zbeacon_test (bool verbose);

    //  Test 1 - two beacons, one speaking, one listening
    //  Create speaker beacon to broadcast our service
    zactor_t *speaker = zactor_new (zbeacon, NULL);
    assert (speaker);
    if (verbose)
        zstr_sendx (speaker, "VERBOSE", NULL);
    
    zsock_send (speaker, "si", "CONFIGURE", 9999);
    char *hostname = zstr_recv (speaker);
    if (!*hostname) {
        printf ("OK (skipping test, no UDP broadcasting)\n");
        zactor_destroy (&speaker);
        freen (hostname);
        return;
    }
    freen (hostname);
    
    //  Create listener beacon on port 9999 to lookup service
    zactor_t *listener = zactor_new (zbeacon, NULL);
    assert (listener);
    if (verbose)
        zstr_sendx (listener, "VERBOSE", NULL);
    zsock_send (listener, "si", "CONFIGURE", 9999);
    hostname = zstr_recv (listener);
    assert (*hostname);
    freen (hostname);
    
    //  We will broadcast the magic value 0xCAFE
    byte announcement [2] = { 0xCA, 0xFE };
    zsock_send (speaker, "sbi", "PUBLISH", announcement, 2, 100);
    //  We will listen to anything (empty subscription)
    zsock_send (listener, "sb", "SUBSCRIBE", "", 0);
    
    //  Wait for at most 1/2 second if there's no broadcasting
    zsock_set_rcvtimeo (listener, 500);
    char *ipaddress = zstr_recv (listener);
    if (ipaddress) {
        zframe_t *content = zframe_recv (listener);
        assert (zframe_size (content) == 2);
        assert (zframe_data (content) [0] == 0xCA);
        assert (zframe_data (content) [1] == 0xFE);
        zframe_destroy (&content);
        zstr_free (&ipaddress);
        zstr_sendx (speaker, "SILENCE", NULL);
    }
    zactor_destroy (&listener);
    zactor_destroy (&speaker);
    
    //  Test subscription filter using a 3-node setup
    zactor_t *node1 = zactor_new (zbeacon, NULL);
    assert (node1);
    zsock_send (node1, "si", "CONFIGURE", 5670);
    hostname = zstr_recv (node1);
    assert (*hostname);
    freen (hostname);
    
    zactor_t *node2 = zactor_new (zbeacon, NULL);
    assert (node2);
    zsock_send (node2, "si", "CONFIGURE", 5670);
    hostname = zstr_recv (node2);
    assert (*hostname);
    freen (hostname);
    
    zactor_t *node3 = zactor_new (zbeacon, NULL);
    assert (node3);
    zsock_send (node3, "si", "CONFIGURE", 5670);
    hostname = zstr_recv (node3);
    assert (*hostname);
    freen (hostname);
    
    zsock_send (node1, "sbi", "PUBLISH", "NODE/1", 6, 250);
    zsock_send (node2, "sbi", "PUBLISH", "NODE/2", 6, 250);
    zsock_send (node3, "sbi", "PUBLISH", "RANDOM", 6, 250);
    zsock_send (node1, "sb", "SUBSCRIBE", "NODE", 4);
    
    //  Poll on three API sockets at once
    zpoller_t *poller = zpoller_new (node1, node2, node3, NULL);
    assert (poller);
    int64_t stop_at = zclock_mono () + 1000;
    while (zclock_mono () < stop_at) {
        long timeout = (long) (stop_at - zclock_mono ());
        if (timeout < 0)
            timeout = 0;
        void *which = zpoller_wait (poller, timeout);
        if (which) {
            assert (which == node1);
            char *ipaddress, *received;
            zstr_recvx (node1, &ipaddress, &received, NULL);
            assert (streq (received, "NODE/2"));
            zstr_free (&ipaddress);
            zstr_free (&received);
        }
    }
    zpoller_destroy (&poller);
    
    //  Stop listening
    zstr_sendx (node1, "UNSUBSCRIBE", NULL);
    
    //  Stop all node broadcasts
    zstr_sendx (node1, "SILENCE", NULL);
    zstr_sendx (node2, "SILENCE", NULL);
    zstr_sendx (node3, "SILENCE", NULL);
    
    //  Destroy the test nodes
    zactor_destroy (&node1);
    zactor_destroy (&node2);
    zactor_destroy (&node3);
    
    //  Unset multicast
    zsys_set_ipv4_mcast_address (NULL);
    
    //  Test 1 - two beacons, one speaking, one listening
    //  Create speaker beacon to broadcast our service
    zactor_t *speaker = zactor_new (zbeacon, NULL);
    assert (speaker);
    if (verbose)
        zstr_sendx (speaker, "VERBOSE", NULL);
    
    zsock_send (speaker, "si", "CONFIGURE", 9999);
    char *hostname = zstr_recv (speaker);
    if (!*hostname) {
        printf ("OK (skipping test, no UDP broadcasting)\n");
        zactor_destroy (&speaker);
        freen (hostname);
        return;
    }
    freen (hostname);
    
    //  Create listener beacon on port 9999 to lookup service
    zactor_t *listener = zactor_new (zbeacon, NULL);
    assert (listener);
    if (verbose)
        zstr_sendx (listener, "VERBOSE", NULL);
    zsock_send (listener, "si", "CONFIGURE", 9999);
    hostname = zstr_recv (listener);
    assert (*hostname);
    freen (hostname);
    
    //  We will broadcast the magic value 0xCAFE
    byte announcement [2] = { 0xCA, 0xFE };
    zsock_send (speaker, "sbi", "PUBLISH", announcement, 2, 100);
    //  We will listen to anything (empty subscription)
    zsock_send (listener, "sb", "SUBSCRIBE", "", 0);
    
    //  Wait for at most 1/2 second if there's no broadcasting
    zsock_set_rcvtimeo (listener, 500);
    char *ipaddress = zstr_recv (listener);
    if (ipaddress) {
        zframe_t *content = zframe_recv (listener);
        assert (zframe_size (content) == 2);
        assert (zframe_data (content) [0] == 0xCA);
        assert (zframe_data (content) [1] == 0xFE);
        zframe_destroy (&content);
        zstr_free (&ipaddress);
        zstr_sendx (speaker, "SILENCE", NULL);
    }
    zactor_destroy (&listener);
    zactor_destroy (&speaker);
    
    //  Test subscription filter using a 3-node setup
    zactor_t *node1 = zactor_new (zbeacon, NULL);
    assert (node1);
    zsock_send (node1, "si", "CONFIGURE", 5670);
    hostname = zstr_recv (node1);
    assert (*hostname);
    freen (hostname);
    
    zactor_t *node2 = zactor_new (zbeacon, NULL);
    assert (node2);
    zsock_send (node2, "si", "CONFIGURE", 5670);
    hostname = zstr_recv (node2);
    assert (*hostname);
    freen (hostname);
    
    zactor_t *node3 = zactor_new (zbeacon, NULL);
    assert (node3);
    zsock_send (node3, "si", "CONFIGURE", 5670);
    hostname = zstr_recv (node3);
    assert (*hostname);
    freen (hostname);
    
    zsock_send (node1, "sbi", "PUBLISH", "NODE/1", 6, 250);
    zsock_send (node2, "sbi", "PUBLISH", "NODE/2", 6, 250);
    zsock_send (node3, "sbi", "PUBLISH", "RANDOM", 6, 250);
    zsock_send (node1, "sb", "SUBSCRIBE", "NODE", 4);
    
    //  Poll on three API sockets at once
    zpoller_t *poller = zpoller_new (node1, node2, node3, NULL);
    assert (poller);
    int64_t stop_at = zclock_mono () + 1000;
    while (zclock_mono () < stop_at) {
        long timeout = (long) (stop_at - zclock_mono ());
        if (timeout < 0)
            timeout = 0;
        void *which = zpoller_wait (poller, timeout);
        if (which) {
            assert (which == node1);
            char *ipaddress, *received;
            zstr_recvx (node1, &ipaddress, &received, NULL);
            assert (streq (received, "NODE/2"));
            zstr_free (&ipaddress);
            zstr_free (&received);
        }
    }
    zpoller_destroy (&poller);
    
    //  Stop listening
    zstr_sendx (node1, "UNSUBSCRIBE", NULL);
    
    //  Stop all node broadcasts
    zstr_sendx (node1, "SILENCE", NULL);
    zstr_sendx (node2, "SILENCE", NULL);
    zstr_sendx (node3, "SILENCE", NULL);
    
    //  Destroy the test nodes
    zactor_destroy (&node1);
    zactor_destroy (&node2);
    zactor_destroy (&node3);
    
    #if defined (__WINDOWS__)
    zsys_shutdown();
    #endif

    //  This is a stable class, and may not change except for emergencies. It
    //  is provided in stable builds.
    //  This class has draft methods, which may change over time. They are not
    //  in stable releases, by default. Use --enable-drafts to enable.
    //  Create and initialize a new certificate in memory
    CZMQ_EXPORT zcert_t *
        zcert_new (void);
    
    //  Accepts public/secret key pair from caller
    CZMQ_EXPORT zcert_t *
        zcert_new_from (const byte *public_key, const byte *secret_key);
    
    //  Load certificate from file
    CZMQ_EXPORT zcert_t *
        zcert_load (const char *filename);
    
    //  Destroy a certificate in memory
    CZMQ_EXPORT void
        zcert_destroy (zcert_t **self_p);
    
    //  Return public part of key pair as 32-byte binary string
    CZMQ_EXPORT const byte *
        zcert_public_key (zcert_t *self);
    
    //  Return secret part of key pair as 32-byte binary string
    CZMQ_EXPORT const byte *
        zcert_secret_key (zcert_t *self);
    
    //  Return public part of key pair as Z85 armored string
    CZMQ_EXPORT const char *
        zcert_public_txt (zcert_t *self);
    
    //  Return secret part of key pair as Z85 armored string
    CZMQ_EXPORT const char *
        zcert_secret_txt (zcert_t *self);
    
    //  Set certificate metadata from formatted string.
    CZMQ_EXPORT void
        zcert_set_meta (zcert_t *self, const char *name, const char *format, ...) CHECK_PRINTF (3);
    
    //  Get metadata value from certificate; if the metadata value doesn't
    //  exist, returns NULL.
    CZMQ_EXPORT const char *
        zcert_meta (zcert_t *self, const char *name);
    
    //  Get list of metadata fields from certificate. Caller is responsible for
    //  destroying list. Caller should not modify the values of list items.
    CZMQ_EXPORT zlist_t *
        zcert_meta_keys (zcert_t *self);
    
    //  Save full certificate (public + secret) to file for persistent storage
    //  This creates one public file and one secret file (filename + "_secret").
    CZMQ_EXPORT int
        zcert_save (zcert_t *self, const char *filename);
    
    //  Save public certificate only to file for persistent storage
    CZMQ_EXPORT int
        zcert_save_public (zcert_t *self, const char *filename);
    
    //  Save secret certificate only to file for persistent storage
    CZMQ_EXPORT int
        zcert_save_secret (zcert_t *self, const char *filename);
    
    //  Apply certificate to socket, i.e. use for CURVE security on socket.
    //  If certificate was loaded from public file, the secret key will be
    //  undefined, and this certificate will not work successfully.
    CZMQ_EXPORT void
        zcert_apply (zcert_t *self, void *socket);
    
    //  Return copy of certificate; if certificate is NULL or we exhausted
    //  heap memory, returns NULL.
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT zcert_t *
        zcert_dup (zcert_t *self);
    
    //  Return true if two certificates have the same keys
    CZMQ_EXPORT bool
        zcert_eq (zcert_t *self, zcert_t *compare);
    
    //  Print certificate contents to stdout
    CZMQ_EXPORT void
        zcert_print (zcert_t *self);
    
    //  Self test of this class
    CZMQ_EXPORT void
        zcert_test (bool verbose);
    
    #ifdef CZMQ_BUILD_DRAFT_API
    //  *** Draft method, for development use, may change without warning ***
    //  Accepts public/secret key text pair from caller
    CZMQ_EXPORT zcert_t *
        zcert_new_from_txt (const char *public_txt, const char *secret_txt);
    
    //  *** Draft method, for development use, may change without warning ***
    //  Unset certificate metadata.
    CZMQ_EXPORT void
        zcert_unset_meta (zcert_t *self, const char *name);
    
    #endif // CZMQ_BUILD_DRAFT_API

    
    const char *SELFTEST_DIR_RW = "src/selftest-rw";
    
    const char *testbasedir  = ".test_zcert";
    const char *testfile = "mycert.txt";
    char *basedirpath = NULL;   // subdir in a test, under SELFTEST_DIR_RW
    char *filepath = NULL;      // pathname to testfile in a test, in dirpath
    char *filepath_s = NULL;    // pathname to testfile+secret in a test, in dirpath
    
    basedirpath = zsys_sprintf ("%s/%s", SELFTEST_DIR_RW, testbasedir);
    assert (basedirpath);
    filepath = zsys_sprintf ("%s/%s", basedirpath, testfile);
    assert (filepath);
    filepath_s = zsys_sprintf ("%s_secret", filepath);
    assert (filepath_s);
    
    // Make sure old aborted tests do not hinder us
    zdir_t *dir = zdir_new (basedirpath, NULL);
    if (dir) {
        zdir_remove (dir, true);
        zdir_destroy (&dir);
    }
    zsys_file_delete (filepath);
    zsys_dir_delete  (basedirpath);
    
    //  Create temporary directory for test files
    zsys_dir_create (basedirpath);
    
    //  Create a simple certificate with metadata
    zcert_t *cert = zcert_new ();
    assert (cert);
    zcert_set_meta (cert, "email", "[email protected]");
    zcert_set_meta (cert, "name", "Pieter Hintjens");
    zcert_set_meta (cert, "organization", "iMatix Corporation");
    zcert_set_meta (cert, "version", "%d", 1);
    zcert_set_meta (cert, "delete_me", "now");
    zcert_unset_meta (cert, "delete_me");
    assert (streq (zcert_meta (cert, "email"), "[email protected]"));
    zlist_t *keys = zcert_meta_keys (cert);
    assert (zlist_size (keys) == 4);
    zlist_destroy (&keys);
    
    //  Check the dup and eq methods
    zcert_t *shadow = zcert_dup (cert);
    assert (zcert_eq (cert, shadow));
    zcert_destroy (&shadow);
    
    //  Check we can save and load certificate
    zcert_save (cert, filepath);
    assert (zsys_file_exists (filepath));
    assert (zsys_file_exists (filepath_s));
    
    //  Load certificate, will in fact load secret one
    shadow = zcert_load (filepath);
    assert (shadow);
    assert (zcert_eq (cert, shadow));
    zcert_destroy (&shadow);
    
    //  Delete secret certificate, load public one
    int rc = zsys_file_delete (filepath_s);
    assert (rc == 0);
    shadow = zcert_load (filepath);
    
    //  32-byte null key encodes as 40 '0' characters
    assert (streq (zcert_secret_txt (shadow), FORTY_ZEROES));
    
    #ifdef CZMQ_BUILD_DRAFT_API
    // test zcert_from_txt
    zcert_t *cert2 = zcert_new_from_txt(cert->public_txt, cert->secret_txt);
    assert (cert2);
    assert (zcert_eq (cert, cert2));
    zcert_destroy(&cert2);
    #endif
    
    zcert_destroy (&shadow);
    zcert_destroy (&cert);
    
    //  Delete all test files
    dir = zdir_new (basedirpath, NULL);
    assert (dir);
    zdir_remove (dir, true);
    zdir_destroy (&dir);
    
    zstr_free (&basedirpath);
    zstr_free (&filepath);
    zstr_free (&filepath_s);
    
    #if defined (__WINDOWS__)
    zsys_shutdown();
    #endif

    //  This is a stable class, and may not change except for emergencies. It
    //  is provided in stable builds.
    //  This class has draft methods, which may change over time. They are not
    //  in stable releases, by default. Use --enable-drafts to enable.
    //  Create a new certificate store from a disk directory, loading and
    //  indexing all certificates in that location. The directory itself may be
    //  absent, and created later, or modified at any time. The certificate store
    //  is automatically refreshed on any zcertstore_lookup() call. If the
    //  location is specified as NULL, creates a pure-memory store, which you
    //  can work with by inserting certificates at runtime.
    CZMQ_EXPORT zcertstore_t *
        zcertstore_new (const char *location);
    
    //  Destroy a certificate store object in memory. Does not affect anything
    //  stored on disk.
    CZMQ_EXPORT void
        zcertstore_destroy (zcertstore_t **self_p);
    
    //  Look up certificate by public key, returns zcert_t object if found,
    //  else returns NULL. The public key is provided in Z85 text format.
    CZMQ_EXPORT zcert_t *
        zcertstore_lookup (zcertstore_t *self, const char *public_key);
    
    //  Insert certificate into certificate store in memory. Note that this
    //  does not save the certificate to disk. To do that, use zcert_save()
    //  directly on the certificate. Takes ownership of zcert_t object.
    CZMQ_EXPORT void
        zcertstore_insert (zcertstore_t *self, zcert_t **cert_p);
    
    //  Print list of certificates in store to logging facility
    CZMQ_EXPORT void
        zcertstore_print (zcertstore_t *self);
    
    //  Self test of this class
    CZMQ_EXPORT void
        zcertstore_test (bool verbose);
    
    #ifdef CZMQ_BUILD_DRAFT_API
    // Loaders retrieve certificates from an arbitrary source.
    typedef void (zcertstore_loader) (
        zcertstore_t *self);
    
    // Destructor for loader state.
    typedef void (zcertstore_destructor) (
        void **self_p);
    
    //  *** Draft method, for development use, may change without warning ***
    //  Override the default disk loader with a custom loader fn.
    CZMQ_EXPORT void
        zcertstore_set_loader (zcertstore_t *self, zcertstore_loader loader, zcertstore_destructor destructor, void *state);
    
    //  *** Draft method, for development use, may change without warning ***
    //  Empty certificate hashtable. This wrapper exists to be friendly to bindings,
    //  which don't usually have access to struct internals.
    CZMQ_EXPORT void
        zcertstore_empty (zcertstore_t *self);
    
    //  *** Draft method, for development use, may change without warning ***
    //  Return a list of all the certificates in the store.
    //  The caller takes ownership of the zlistx_t object and is responsible
    //  for destroying it.  The caller does not take ownership of the zcert_t
    //  objects.
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT zlistx_t *
        zcertstore_certs (zcertstore_t *self);
    
    #endif // CZMQ_BUILD_DRAFT_API

    
    const char *SELFTEST_DIR_RW = "src/selftest-rw";
    
    const char *testbasedir  = ".test_zcertstore";
    const char *testfile = "mycert.txt";
    char *basedirpath = NULL;   // subdir in a test, under SELFTEST_DIR_RW
    char *filepath = NULL;      // pathname to testfile in a test, in dirpath
    
    basedirpath = zsys_sprintf ("%s/%s", SELFTEST_DIR_RW, testbasedir);
    assert (basedirpath);
    filepath = zsys_sprintf ("%s/%s", basedirpath, testfile);
    assert (filepath);
    
    // Make sure old aborted tests do not hinder us
    zdir_t *dir = zdir_new (basedirpath, NULL);
    if (dir) {
        zdir_remove (dir, true);
        zdir_destroy (&dir);
    }
    zsys_file_delete (filepath);
    zsys_dir_delete  (basedirpath);
    
    //  Create temporary directory for test files
    zsys_dir_create (basedirpath);
    
    //  Load certificate store from disk; it will be empty
    zcertstore_t *certstore = zcertstore_new (basedirpath);
    assert (certstore);
    
    //  Create a single new certificate and save to disk
    zcert_t *cert = zcert_new ();
    assert (cert);
    char *client_key = strdup (zcert_public_txt (cert));
    assert (client_key);
    zcert_set_meta (cert, "name", "John Doe");
    zcert_save (cert, filepath);
    zcert_destroy (&cert);
    
    //  Check that certificate store refreshes as expected
    cert = zcertstore_lookup (certstore, client_key);
    assert (cert);
    assert (streq (zcert_meta (cert, "name"), "John Doe"));
    
    #ifdef CZMQ_BUILD_DRAFT_API
    //  DRAFT-API: Security
    // Iterate through certs
    zlistx_t *certs = zcertstore_certs(certstore);
    cert = (zcert_t *) zlistx_first(certs);
    int cert_count = 0;
    while (cert) {
        assert (streq (zcert_meta (cert, "name"), "John Doe"));
        cert = (zcert_t *) zlistx_next(certs);
        cert_count++;
    }
    assert(cert_count==1);
    zlistx_destroy(&certs);
    #endif
    
    //  Test custom loader
    test_loader_state *state = (test_loader_state *) zmalloc (sizeof (test_loader_state));
    state->index = 0;
    zcertstore_set_loader (certstore, s_test_loader, s_test_destructor, (void *)state);
    #if (ZMQ_VERSION_MAJOR >= 4)
    cert = zcertstore_lookup (certstore, client_key);
    assert (cert == NULL);
    cert = zcertstore_lookup (certstore, "abcdefghijklmnopqrstuvwxyzabcdefghijklmn");
    assert (cert);
    #endif
    
    freen (client_key);
    
    if (verbose)
        zcertstore_print (certstore);
    zcertstore_destroy (&certstore);
    
    //  Delete all test files
    dir = zdir_new (basedirpath, NULL);
    assert (dir);
    zdir_remove (dir, true);
    zdir_destroy (&dir);
    
    zstr_free (&basedirpath);
    zstr_free (&filepath);
    
    #if defined (__WINDOWS__)
    zsys_shutdown();
    #endif

    //  This is a stable class, and may not change except for emergencies. It
    //  is provided in stable builds.
    //  This class has draft methods, which may change over time. They are not
    //  in stable releases, by default. Use --enable-drafts to enable.
    //  Create a new chunk of the specified size. If you specify the data, it
    //  is copied into the chunk. If you do not specify the data, the chunk is
    //  allocated and left empty, and you can then add data using zchunk_append.
    CZMQ_EXPORT zchunk_t *
        zchunk_new (const void *data, size_t size);
    
    //  Destroy a chunk
    CZMQ_EXPORT void
        zchunk_destroy (zchunk_t **self_p);
    
    //  Resizes chunk max_size as requested; chunk_cur size is set to zero
    CZMQ_EXPORT void
        zchunk_resize (zchunk_t *self, size_t size);
    
    //  Return chunk cur size
    CZMQ_EXPORT size_t
        zchunk_size (zchunk_t *self);
    
    //  Return chunk max size
    CZMQ_EXPORT size_t
        zchunk_max_size (zchunk_t *self);
    
    //  Return chunk data
    CZMQ_EXPORT byte *
        zchunk_data (zchunk_t *self);
    
    //  Set chunk data from user-supplied data; truncate if too large. Data may
    //  be null. Returns actual size of chunk
    CZMQ_EXPORT size_t
        zchunk_set (zchunk_t *self, const void *data, size_t size);
    
    //  Fill chunk data from user-supplied octet
    CZMQ_EXPORT size_t
        zchunk_fill (zchunk_t *self, byte filler, size_t size);
    
    //  Append user-supplied data to chunk, return resulting chunk size. If the
    //  data would exceeded the available space, it is truncated. If you want to
    //  grow the chunk to accommodate new data, use the zchunk_extend method.
    CZMQ_EXPORT size_t
        zchunk_append (zchunk_t *self, const void *data, size_t size);
    
    //  Append user-supplied data to chunk, return resulting chunk size. If the
    //  data would exceeded the available space, the chunk grows in size.
    CZMQ_EXPORT size_t
        zchunk_extend (zchunk_t *self, const void *data, size_t size);
    
    //  Copy as much data from 'source' into the chunk as possible; returns the
    //  new size of chunk. If all data from 'source' is used, returns exhausted
    //  on the source chunk. Source can be consumed as many times as needed until
    //  it is exhausted. If source was already exhausted, does not change chunk.
    CZMQ_EXPORT size_t
        zchunk_consume (zchunk_t *self, zchunk_t *source);
    
    //  Returns true if the chunk was exhausted by consume methods, or if the
    //  chunk has a size of zero.
    CZMQ_EXPORT bool
        zchunk_exhausted (zchunk_t *self);
    
    //  Read chunk from an open file descriptor
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT zchunk_t *
        zchunk_read (FILE *handle, size_t bytes);
    
    //  Write chunk to an open file descriptor
    CZMQ_EXPORT int
        zchunk_write (zchunk_t *self, FILE *handle);
    
    //  Try to slurp an entire file into a chunk. Will read up to maxsize of
    //  the file. If maxsize is 0, will attempt to read the entire file and
    //  fail with an assertion if that cannot fit into memory. Returns a new
    //  chunk containing the file data, or NULL if the file could not be read.
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT zchunk_t *
        zchunk_slurp (const char *filename, size_t maxsize);
    
    //  Create copy of chunk, as new chunk object. Returns a fresh zchunk_t
    //  object, or null if there was not enough heap memory. If chunk is null,
    //  returns null.
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT zchunk_t *
        zchunk_dup (zchunk_t *self);
    
    //  Return chunk data encoded as printable hex string. Caller must free
    //  string when finished with it.
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT char *
        zchunk_strhex (zchunk_t *self);
    
    //  Return chunk data copied into freshly allocated string
    //  Caller must free string when finished with it.
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT char *
        zchunk_strdup (zchunk_t *self);
    
    //  Return TRUE if chunk body is equal to string, excluding terminator
    CZMQ_EXPORT bool
        zchunk_streq (zchunk_t *self, const char *string);
    
    //  Transform zchunk into a zframe that can be sent in a message.
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT zframe_t *
        zchunk_pack (zchunk_t *self);
    
    //  Transform a zframe into a zchunk.
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT zchunk_t *
        zchunk_unpack (zframe_t *frame);
    
    //  Calculate SHA1 digest for chunk, using zdigest class.
    CZMQ_EXPORT const char *
        zchunk_digest (zchunk_t *self);
    
    //  Dump chunk to FILE stream, for debugging and tracing.
    CZMQ_EXPORT void
        zchunk_fprint (zchunk_t *self, FILE *file);
    
    //  Dump message to stderr, for debugging and tracing.
    //  See zchunk_fprint for details
    CZMQ_EXPORT void
        zchunk_print (zchunk_t *self);
    
    //  Probe the supplied object, and report if it looks like a zchunk_t.
    CZMQ_EXPORT bool
        zchunk_is (void *self);
    
    //  Self test of this class.
    CZMQ_EXPORT void
        zchunk_test (bool verbose);
    
    #ifdef CZMQ_BUILD_DRAFT_API
    // Destroy an item
    typedef void (zchunk_destructor_fn) (
        void **hint);
    
    //  *** Draft method, for development use, may change without warning ***
    //  Create a new chunk from memory. Take ownership of the memory and calling the destructor
    //  on destroy.
    CZMQ_EXPORT zchunk_t *
        zchunk_frommem (void *data, size_t size, zchunk_destructor_fn destructor, void *hint);
    
    //  *** Draft method, for development use, may change without warning ***
    //  Transform zchunk into a zframe that can be sent in a message.
    //  Take ownership of the chunk.
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT zframe_t *
        zchunk_packx (zchunk_t **self_p);
    
    #endif // CZMQ_BUILD_DRAFT_API

    zchunk_t *chunk = zchunk_new ("1234567890", 10);
    assert (chunk);
    assert (zchunk_size (chunk) == 10);
    assert (memcmp (zchunk_data (chunk), "1234567890", 10) == 0);
    zchunk_destroy (&chunk);
    
    chunk = zchunk_new (NULL, 10);
    assert (chunk);
    zchunk_append (chunk, "12345678", 8);
    zchunk_append (chunk, "90ABCDEF", 8);
    zchunk_append (chunk, "GHIJKLMN", 8);
    assert (memcmp (zchunk_data (chunk), "1234567890", 10) == 0);
    assert (zchunk_size (chunk) == 10);
    assert (zchunk_streq (chunk, "1234567890"));
    assert (streq (zchunk_digest (chunk), "01B307ACBA4F54F55AAFC33BB06BBBF6CA803E9A"));
    char *string = zchunk_strdup (chunk);
    assert (streq (string, "1234567890"));
    freen (string);
    string = zchunk_strhex (chunk);
    assert (streq (string, "31323334353637383930"));
    freen (string);
    
    zframe_t *frame = zchunk_pack (chunk);
    assert (frame);
    
    zchunk_t *chunk2 = zchunk_unpack (frame);
    assert (chunk2);
    assert (memcmp (zchunk_data (chunk2), "1234567890", 10) == 0);
    zframe_destroy (&frame);
    zchunk_destroy (&chunk2);
    
    zchunk_t *copy = zchunk_dup (chunk);
    assert (copy);
    assert (memcmp (zchunk_data (copy), "1234567890", 10) == 0);
    assert (zchunk_size (copy) == 10);
    zchunk_destroy (&copy);
    zchunk_destroy (&chunk);
    
    chunk = zchunk_new (NULL, 0);
    zchunk_extend (chunk, "12345678", 8);
    zchunk_extend (chunk, "90ABCDEF", 8);
    zchunk_extend (chunk, "GHIJKLMN", 8);
    assert (zchunk_size (chunk) == 24);
    assert (zchunk_streq (chunk, "1234567890ABCDEFGHIJKLMN"));
    zchunk_destroy (&chunk);
    
    copy = zchunk_new ("1234567890abcdefghij", 20);
    assert (copy);
    chunk = zchunk_new (NULL, 8);
    assert (chunk);
    zchunk_consume (chunk, copy);
    assert (!zchunk_exhausted (copy));
    assert (memcmp (zchunk_data (chunk), "12345678", 8) == 0);
    zchunk_set (chunk, NULL, 0);
    zchunk_consume (chunk, copy);
    assert (!zchunk_exhausted (copy));
    assert (memcmp (zchunk_data (chunk), "90abcdef", 8) == 0);
    zchunk_set (chunk, NULL, 0);
    zchunk_consume (chunk, copy);
    assert (zchunk_exhausted (copy));
    assert (zchunk_size (chunk) == 4);
    assert (memcmp (zchunk_data (chunk), "ghij", 4) == 0);
    zchunk_destroy (&copy);
    zchunk_destroy (&chunk);
    
    char str[] = "hello";
    chunk = zchunk_frommem (str, 5, mem_destructor, str);
    assert (chunk);
    zchunk_destroy (&chunk);
    
    //  The destructor doesn't free the memory, only changing the strid,
    //  so we can check if the destructor was invoked
    assert (streq (str, "world"));
    
    chunk = zchunk_new ("1234567890", 10);
    frame = zchunk_packx (&chunk);
    assert (frame);
    assert (chunk == NULL);
    
    chunk = zchunk_unpack (frame);
    assert (chunk);
    assert (memcmp (zchunk_data (chunk), "1234567890", 10) == 0);
    zframe_destroy (&frame);
    zchunk_destroy (&chunk);
    
    #if defined (__WINDOWS__)
    zsys_shutdown();
    #endif

    //  This is a stable class, and may not change except for emergencies. It
    //  is provided in stable builds.
    //  Sleep for a number of milliseconds
    CZMQ_EXPORT void
        zclock_sleep (int msecs);
    
    //  Return current system clock as milliseconds. Note that this clock can
    //  jump backwards (if the system clock is changed) so is unsafe to use for
    //  timers and time offsets. Use zclock_mono for that instead.
    CZMQ_EXPORT int64_t
        zclock_time (void);
    
    //  Return current monotonic clock in milliseconds. Use this when you compute
    //  time offsets. The monotonic clock is not affected by system changes and
    //  so will never be reset backwards, unlike a system clock.
    CZMQ_EXPORT int64_t
        zclock_mono (void);
    
    //  Return current monotonic clock in microseconds. Use this when you compute
    //  time offsets. The monotonic clock is not affected by system changes and
    //  so will never be reset backwards, unlike a system clock.
    CZMQ_EXPORT int64_t
        zclock_usecs (void);
    
    //  Return formatted date/time as fresh string. Free using zstr_free().
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT char *
        zclock_timestr (void);
    
    //  Self test of this class.
    CZMQ_EXPORT void
        zclock_test (bool verbose);
    

    int64_t start = zclock_time ();
    zclock_sleep (10);
    assert ((zclock_time () - start) >= 10);
    start = zclock_mono ();
    int64_t usecs = zclock_usecs ();
    zclock_sleep (10);
    assert ((zclock_mono () - start) >= 10);
    assert ((zclock_usecs () - usecs) >= 10000);
    char *timestr = zclock_timestr ();
    if (verbose)
        puts (timestr);
    freen (timestr);
    
    #if defined (__WINDOWS__)
    zsys_shutdown();
    #endif

    //  This is a stable class, and may not change except for emergencies. It
    //  is provided in stable builds.
    //  This class has draft methods, which may change over time. They are not
    //  in stable releases, by default. Use --enable-drafts to enable.
    //
    typedef int (zconfig_fct) (
        zconfig_t *self, void *arg, int level);
    
    //  Create new config item
    CZMQ_EXPORT zconfig_t *
        zconfig_new (const char *name, zconfig_t *parent);
    
    //  Load a config tree from a specified ZPL text file; returns a zconfig_t
    //  reference for the root, if the file exists and is readable. Returns NULL
    //  if the file does not exist.
    CZMQ_EXPORT zconfig_t *
        zconfig_load (const char *filename);
    
    //  Equivalent to zconfig_load, taking a format string instead of a fixed
    //  filename.
    CZMQ_EXPORT zconfig_t *
        zconfig_loadf (const char *format, ...) CHECK_PRINTF (1);
    
    //  Destroy a config item and all its children
    CZMQ_EXPORT void
        zconfig_destroy (zconfig_t **self_p);
    
    //  Return name of config item
    CZMQ_EXPORT char *
        zconfig_name (zconfig_t *self);
    
    //  Return value of config item
    CZMQ_EXPORT char *
        zconfig_value (zconfig_t *self);
    
    //  Insert or update configuration key with value
    CZMQ_EXPORT void
        zconfig_put (zconfig_t *self, const char *path, const char *value);
    
    //  Equivalent to zconfig_put, accepting a format specifier and variable
    //  argument list, instead of a single string value.
    CZMQ_EXPORT void
        zconfig_putf (zconfig_t *self, const char *path, const char *format, ...) CHECK_PRINTF (3);
    
    //  Get value for config item into a string value; leading slash is optional
    //  and ignored.
    CZMQ_EXPORT char *
        zconfig_get (zconfig_t *self, const char *path, const char *default_value);
    
    //  Set config item name, name may be NULL
    CZMQ_EXPORT void
        zconfig_set_name (zconfig_t *self, const char *name);
    
    //  Set new value for config item. The new value may be a string, a printf
    //  format, or NULL. Note that if string may possibly contain '%', or if it
    //  comes from an insecure source, you must use '%s' as the format, followed
    //  by the string.
    CZMQ_EXPORT void
        zconfig_set_value (zconfig_t *self, const char *format, ...) CHECK_PRINTF (2);
    
    //  Find our first child, if any
    CZMQ_EXPORT zconfig_t *
        zconfig_child (zconfig_t *self);
    
    //  Find our first sibling, if any
    CZMQ_EXPORT zconfig_t *
        zconfig_next (zconfig_t *self);
    
    //  Find a config item along a path; leading slash is optional and ignored.
    CZMQ_EXPORT zconfig_t *
        zconfig_locate (zconfig_t *self, const char *path);
    
    //  Locate the last config item at a specified depth
    CZMQ_EXPORT zconfig_t *
        zconfig_at_depth (zconfig_t *self, int level);
    
    //  Execute a callback for each config item in the tree; returns zero if
    //  successful, else -1.
    CZMQ_EXPORT int
        zconfig_execute (zconfig_t *self, zconfig_fct handler, void *arg);
    
    //  Add comment to config item before saving to disk. You can add as many
    //  comment lines as you like. If you use a null format, all comments are
    //  deleted.
    CZMQ_EXPORT void
        zconfig_set_comment (zconfig_t *self, const char *format, ...) CHECK_PRINTF (2);
    
    //  Return comments of config item, as zlist.
    CZMQ_EXPORT zlist_t *
        zconfig_comments (zconfig_t *self);
    
    //  Save a config tree to a specified ZPL text file, where a filename
    //  "-" means dump to standard output.
    CZMQ_EXPORT int
        zconfig_save (zconfig_t *self, const char *filename);
    
    //  Equivalent to zconfig_save, taking a format string instead of a fixed
    //  filename.
    CZMQ_EXPORT int
        zconfig_savef (zconfig_t *self, const char *format, ...) CHECK_PRINTF (2);
    
    //  Report filename used during zconfig_load, or NULL if none
    CZMQ_EXPORT const char *
        zconfig_filename (zconfig_t *self);
    
    //  Reload config tree from same file that it was previously loaded from.
    //  Returns 0 if OK, -1 if there was an error (and then does not change
    //  existing data).
    CZMQ_EXPORT int
        zconfig_reload (zconfig_t **self_p);
    
    //  Load a config tree from a memory chunk
    CZMQ_EXPORT zconfig_t *
        zconfig_chunk_load (zchunk_t *chunk);
    
    //  Save a config tree to a new memory chunk
    CZMQ_EXPORT zchunk_t *
        zconfig_chunk_save (zconfig_t *self);
    
    //  Load a config tree from a null-terminated string
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT zconfig_t *
        zconfig_str_load (const char *string);
    
    //  Save a config tree to a new null terminated string
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT char *
        zconfig_str_save (zconfig_t *self);
    
    //  Return true if a configuration tree was loaded from a file and that
    //  file has changed in since the tree was loaded.
    CZMQ_EXPORT bool
        zconfig_has_changed (zconfig_t *self);
    
    //  Print the config file to open stream
    CZMQ_EXPORT void
        zconfig_fprint (zconfig_t *self, FILE *file);
    
    //  Print properties of object
    CZMQ_EXPORT void
        zconfig_print (zconfig_t *self);
    
    //  Self test of this class
    CZMQ_EXPORT void
        zconfig_test (bool verbose);
    
    #ifdef CZMQ_BUILD_DRAFT_API
    //  *** Draft method, for development use, may change without warning ***
    //  Create copy of zconfig, caller MUST free the value
    //  Create copy of config, as new zconfig object. Returns a fresh zconfig_t
    //  object. If config is null, or memory was exhausted, returns null.
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT zconfig_t *
        zconfig_dup (zconfig_t *self);
    
    //  *** Draft method, for development use, may change without warning ***
    //  Destroy subtree (all children)
    CZMQ_EXPORT void
        zconfig_remove_subtree (zconfig_t *self);
    
    //  *** Draft method, for development use, may change without warning ***
    //  Destroy node and subtree (all children)
    CZMQ_EXPORT void
        zconfig_remove (zconfig_t **self_p);
    
    #endif // CZMQ_BUILD_DRAFT_API

    
    const char *SELFTEST_DIR_RW = "src/selftest-rw";
    
    const char *testbasedir  = ".test_zconfig";
    const char *testfile = "test.cfg";
    char *basedirpath = NULL;   // subdir in a test, under SELFTEST_DIR_RW
    char *filepath = NULL;      // pathname to testfile in a test, in dirpath
    
    basedirpath = zsys_sprintf ("%s/%s", SELFTEST_DIR_RW, testbasedir);
    assert (basedirpath);
    filepath = zsys_sprintf ("%s/%s", basedirpath, testfile);
    assert (filepath);
    
    // Make sure old aborted tests do not hinder us
    zdir_t *dir = zdir_new (basedirpath, NULL);
    if (dir) {
        zdir_remove (dir, true);
        zdir_destroy (&dir);
    }
    zsys_file_delete (filepath);
    zsys_dir_delete  (basedirpath);
    
    //  Create temporary directory for test files
    zsys_dir_create (basedirpath);
    
    zconfig_t *root = zconfig_new ("root", NULL);
    assert (root);
    zconfig_t *section, *item;
    
    section = zconfig_new ("headers", root);
    assert (section);
    item = zconfig_new ("email", section);
    assert (item);
    zconfig_set_value (item, "[email protected]");
    item = zconfig_new ("name", section);
    assert (item);
    zconfig_set_value (item, "Justin Kayce");
    zconfig_putf (root, "/curve/secret-key", "%s", "Top Secret");
    zconfig_set_comment (root, "   CURVE certificate");
    zconfig_set_comment (root, "   -----------------");
    assert (zconfig_comments (root));
    zconfig_save (root, filepath);
    zconfig_destroy (&root);
    root = zconfig_load (filepath);
    if (verbose)
        zconfig_save (root, "-");
    assert (streq (zconfig_filename (root), filepath));
    
    char *email = zconfig_get (root, "/headers/email", NULL);
    assert (email);
    assert (streq (email, "[email protected]"));
    char *passwd = zconfig_get (root, "/curve/secret-key", NULL);
    assert (passwd);
    assert (streq (passwd, "Top Secret"));
    
    zconfig_savef (root, "%s/%s", basedirpath, testfile);
    assert (!zconfig_has_changed (root));
    int rc = zconfig_reload (&root);
    assert (rc == 0);
    assert (!zconfig_has_changed (root));
    zconfig_destroy (&root);
    
    //  Test chunk load/save
    root = zconfig_new ("root", NULL);
    assert (root);
    section = zconfig_new ("section", root);
    assert (section);
    item = zconfig_new ("value", section);
    assert (item);
    zconfig_set_value (item, "somevalue");
    zconfig_t *search = zconfig_locate (root, "section/value");
    assert (search == item);
    zchunk_t *chunk = zconfig_chunk_save (root);
    assert (strlen ((char *) zchunk_data (chunk)) == 32);
    char *string = zconfig_str_save (root);
    assert (string);
    assert (streq (string, (char *) zchunk_data (chunk)));
    freen (string);
    assert (chunk);
    zconfig_destroy (&root);
    
    root = zconfig_chunk_load (chunk);
    assert (root);
    char *value = zconfig_get (root, "/section/value", NULL);
    assert (value);
    assert (streq (value, "somevalue"));
    
    //  Test config can't be saved to a file in a path that doesn't
    //  exist or isn't writable
    rc = zconfig_savef (root, "%s/path/that/doesnt/exist/%s", basedirpath, testfile);
    assert (rc == -1);
    
    zconfig_destroy (&root);
    zchunk_destroy (&chunk);
    
    // Test str_load
    zconfig_t *config = zconfig_str_load (
        "malamute\n"
        "    endpoint = ipc://@/malamute\n"
        "    producer = STREAM\n"
        "    consumer\n"
        "        STREAM2 = .*\n"
        "        STREAM3 = HAM\n"
        "server\n"
        "    verbose = true\n"
        );
    assert (config);
    assert (streq (zconfig_get (config, "malamute/endpoint", NULL), "ipc://@/malamute"));
    assert (streq (zconfig_get (config, "malamute/producer", NULL), "STREAM"));
    assert (zconfig_locate (config, "malamute/consumer"));
    
    zconfig_t *c = zconfig_child (zconfig_locate (config, "malamute/consumer"));
    assert (c);
    assert (streq (zconfig_name (c), "STREAM2"));
    assert (streq (zconfig_value (c), ".*"));
    
    c = zconfig_next (c);
    assert (c);
    assert (streq (zconfig_name (c), "STREAM3"));
    assert (streq (zconfig_value (c), "HAM"));
    
    c = zconfig_next (c);
    assert (!c);
    
    assert (streq (zconfig_get (config, "server/verbose", NULL), "true"));
    
    zconfig_t *dup = zconfig_dup (config);
    assert (dup);
    assert (streq (zconfig_get (dup, "server/verbose", NULL), "true"));
    zconfig_destroy (&dup);
    
    zconfig_destroy (&config);
    
    //  Test subtree removal
    	{
    		zconfig_t *root = zconfig_str_load (
    			"context\n"
    			"    iothreads = 1\n"
    			"    verbose = 1      #   Ask for a trace\n"
    			"main\n"
    			"    type = zqueue    #  ZMQ_DEVICE type\n"
    			"    frontend\n"
    			"        option\n"
    			"            hwm = 1000\n"
    			"            swap = 25000000     #  25MB\n"
    			"        bind = 'inproc://addr1'\n"
    			"        bind = 'ipc://addr2'\n"
    			"    backend\n"
    			"        bind = inproc://addr3\n"
    		);
    
        //  no subtree
        zconfig_t *to_delete = zconfig_locate (root, "context/iothreads");
        assert (to_delete);
    
        zconfig_remove_subtree (to_delete);
    
        zconfig_t *check = zconfig_locate (root, "context/iothreads");
        assert (check);
        assert (streq (zconfig_value (check), "1"));
    
        check = zconfig_locate (root, "context/verbose");
        assert (check);
        assert (streq (zconfig_value (check), "1"));
    
        //  existing subtree
        to_delete = zconfig_locate (root, "main/frontend/option");
        assert (to_delete);
    
        zconfig_remove_subtree (to_delete);
    
        check = zconfig_locate (root, "main/frontend/option/hwm");
        assert (check == NULL);
        check = zconfig_locate (root, "main/frontend/option/swap");
        assert (check == NULL);
        check = zconfig_locate (root, "main/frontend/option");
        assert (check);
        assert (streq (zconfig_value (check), ""));
        check = zconfig_next (check);
        assert (check);
        assert (streq (zconfig_name (check), "bind"));
        assert (streq (zconfig_value (check), "inproc://addr1"));
        check = zconfig_next (check);
        assert (check);
        assert (streq (zconfig_name (check), "bind"));
        assert (streq (zconfig_value (check), "ipc://addr2"));
        assert (zconfig_next (check) == NULL);
    
        to_delete = zconfig_locate (root, "main/frontend");
        assert (to_delete);
    
        zconfig_remove_subtree (to_delete);
    
        check = zconfig_locate (root, "main/frontend/option/hwm");
        assert (check == NULL);
        check = zconfig_locate (root, "main/frontend/option/swap");
        assert (check == NULL);
        check = zconfig_locate (root, "main/frontend/option");
        assert (check == NULL);
        check = zconfig_locate (root, "main/frontend/bind");
        assert (check == NULL);
        check = zconfig_locate (root, "main/frontend");
        assert (check);
        assert (streq (zconfig_value (check), ""));
        assert (zconfig_child (check) == NULL);
        check = zconfig_next (check);
        assert (check);
        assert (streq (zconfig_name (check), "backend"));
        assert (streq (zconfig_value (check), ""));
    
        to_delete = zconfig_locate (root, "main");
        assert (to_delete);
    
        zconfig_remove_subtree (to_delete);
    
        check = zconfig_locate (root, "main/type");
        assert (check == NULL);
        check = zconfig_locate (root, "main/frontend");
        assert (check == NULL);
        check = zconfig_locate (root, "main/backend");
        assert (check == NULL);
        check = zconfig_locate (root, "main");
        assert (check);
    
        //  root
        zconfig_remove_subtree (root);
    
        assert (root);
        assert (zconfig_child (root) == NULL);
        check = zconfig_locate (root, "main");
        assert (check == NULL);
        check = zconfig_locate (root, "context");
        assert (check == NULL);
    
        zconfig_destroy (&root);
    }
    
    //  Test node and subtree removal
    	{
    		zconfig_t *root = zconfig_str_load (
    			"A1 = abc\n"
    			"    x\n"
    			"        1\n"
    			"        2\n"
    			"    y = 1      #   Ask for a trace\n"
    			"A2\n"
    			"    B1 = zqueue    #  ZMQ_DEVICE type\n"
    			"    B2\n"
    			"        C1\n"
    			"            hwm = 1000\n"
    			"            swap = 25000000     #  25MB\n"
    			"        C2 = 50\n"
    			"        C3\n"
    			"            bind = addr3\n"
    			"    B3\n"
    			"        bind = inproc://addr4\n"
    			"    B4 = Ignac\n"
    			"        z = 5\n"
    			"A3\n"
    			"A4\n"
    		);
    
        zconfig_t *to_delete = zconfig_locate (root, "A2/B2/C3");
        assert (to_delete);
    
        zconfig_remove (&to_delete);
    
        zconfig_t *check = zconfig_locate (root, "A2/B2/C2");
        assert (check);
        assert (streq (zconfig_value (check), "50"));
        assert (zconfig_next (check) == NULL);
        assert (zconfig_locate (root, "A2/B2/C3/bind") == NULL);
        assert (zconfig_locate (root, "A2/B2/C3") == NULL);
    
        to_delete = zconfig_locate (root, "A2/B2");
        assert (to_delete);
    
        zconfig_remove (&to_delete);
    
        check = zconfig_locate (root, "A2");
        assert (check);
        check = zconfig_child (check);
        assert (check);
        assert (streq (zconfig_name (check), "B1"));
        assert (streq (zconfig_value (check), "zqueue"));
        check = zconfig_next (check);
        assert (check);
        assert (streq (zconfig_name (check), "B3"));
        assert (streq (zconfig_value (check), ""));
        assert (zconfig_locate (root, "A2/B2/C1") == NULL);
        assert (zconfig_locate (root, "A2/B2/C2") == NULL);
        assert (zconfig_locate (root, "A2/B2") == NULL);
        assert (zconfig_locate (root, "A2/B4"));
    
        to_delete = zconfig_locate (root, "A2/B1");
        assert (to_delete);
    
        zconfig_remove (&to_delete);
    
        check = zconfig_locate (root, "A2");
        assert (check);
        check = zconfig_child (check);
        assert (check);
        assert (streq (zconfig_name (check), "B3"));
        assert (streq (zconfig_value (check), ""));
        check = zconfig_next (check);
        assert (check);
        assert (streq (zconfig_name (check), "B4"));
        assert (streq (zconfig_value (check), "Ignac"));
        assert (zconfig_next (check) == NULL);
        assert (zconfig_locate (root, "A2/B1") == NULL);
        assert (zconfig_locate (root, "A2/B2") == NULL);
    
        to_delete = zconfig_locate (root, "A2/B3");
        assert (to_delete);
    
        zconfig_remove (&to_delete);
    
        check = zconfig_locate (root, "A2");
        assert (check);
        check = zconfig_child (check);
        assert (check);
        assert (streq (zconfig_name (check), "B4"));
        assert (streq (zconfig_value (check), "Ignac"));
        assert (zconfig_next (check) == NULL);
    
        to_delete = zconfig_locate (root, "A2");
        assert (to_delete);
    
        zconfig_remove (&to_delete);
    
        check = zconfig_locate (root, "A1");
        assert (check);
        check = zconfig_next (check);
        assert (check);
        assert (streq (zconfig_name (check), "A3"));
        assert (zconfig_locate (root, "A2/B4") == NULL);
        assert (zconfig_locate (root, "A2") == NULL);
    
        to_delete = zconfig_locate (root, "A1");
        assert (to_delete);
    
        zconfig_remove (&to_delete);
    
        check = zconfig_child (root);
        assert (check);
        assert (streq (zconfig_name (check), "A3"));
        assert (zconfig_locate (root, "A1/x/1") == NULL);
        assert (zconfig_locate (root, "A1/x") == NULL);
        assert (zconfig_locate (root, "A1/y") == NULL);
        assert (zconfig_locate (root, "A3"));
        assert (zconfig_locate (root, "A4"));
    
        //  called on root should be equivalent to zconfig_destroy (&root)
        zconfig_remove (&root);
    }
    
    //  Delete all test files
    dir = zdir_new (basedirpath, NULL);
    assert (dir);
    zdir_remove (dir, true);
    zdir_destroy (&dir);
    
    zstr_free (&basedirpath);
    zstr_free (&filepath);
    
    #if defined (__WINDOWS__)
    zsys_shutdown();
    #endif

    //  This is a stable class, and may not change except for emergencies. It
    //  is provided in stable builds.
    //  Constructor - creates new digest object, which you use to build up a
    //  digest by repeatedly calling zdigest_update() on chunks of data.
    CZMQ_EXPORT zdigest_t *
        zdigest_new (void);
    
    //  Destroy a digest object
    CZMQ_EXPORT void
        zdigest_destroy (zdigest_t **self_p);
    
    //  Add buffer into digest calculation
    CZMQ_EXPORT void
        zdigest_update (zdigest_t *self, const byte *buffer, size_t length);
    
    //  Return final digest hash data. If built without crypto support,
    //  returns NULL.
    CZMQ_EXPORT const byte *
        zdigest_data (zdigest_t *self);
    
    //  Return final digest hash size
    CZMQ_EXPORT size_t
        zdigest_size (zdigest_t *self);
    
    //  Return digest as printable hex string; caller should not modify nor
    //  free this string. After calling this, you may not use zdigest_update()
    //  on the same digest. If built without crypto support, returns NULL.
    CZMQ_EXPORT char *
        zdigest_string (zdigest_t *self);
    
    //  Self test of this class.
    CZMQ_EXPORT void
        zdigest_test (bool verbose);
    

    byte *buffer = (byte *) zmalloc (1024);
    memset (buffer, 0xAA, 1024);
    
    zdigest_t *digest = zdigest_new ();
    assert (digest);
    zdigest_update (digest, buffer, 1024);
    const byte *data = zdigest_data (digest);
    assert (data [0] == 0xDE);
    assert (data [1] == 0xB2);
    assert (data [2] == 0x38);
    assert (data [3] == 0x07);
    assert (streq (zdigest_string (digest),
                   "DEB23807D4FE025E900FE9A9C7D8410C3DDE9671"));
    zdigest_destroy (&digest);
    freen (buffer);
    
    #if defined (__WINDOWS__)
    zsys_shutdown();
    #endif

    //  This is a stable class, and may not change except for emergencies. It
    //  is provided in stable builds.
    //  Create a new directory item that loads in the full tree of the specified
    //  path, optionally located under some parent path. If parent is "-", then
    //  loads only the top-level directory, and does not use parent as a path.
    CZMQ_EXPORT zdir_t *
        zdir_new (const char *path, const char *parent);
    
    //  Destroy a directory tree and all children it contains.
    CZMQ_EXPORT void
        zdir_destroy (zdir_t **self_p);
    
    //  Return directory path
    CZMQ_EXPORT const char *
        zdir_path (zdir_t *self);
    
    //  Return last modification time for directory.
    CZMQ_EXPORT time_t
        zdir_modified (zdir_t *self);
    
    //  Return total hierarchy size, in bytes of data contained in all files
    //  in the directory tree.
    CZMQ_EXPORT off_t
        zdir_cursize (zdir_t *self);
    
    //  Return directory count
    CZMQ_EXPORT size_t
        zdir_count (zdir_t *self);
    
    //  Returns a sorted list of zfile objects; Each entry in the list is a pointer
    //  to a zfile_t item already allocated in the zdir tree. Do not destroy the
    //  original zdir tree until you are done with this list.
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT zlist_t *
        zdir_list (zdir_t *self);
    
    //  Remove directory, optionally including all files that it contains, at
    //  all levels. If force is false, will only remove the directory if empty.
    //  If force is true, will remove all files and all subdirectories.
    CZMQ_EXPORT void
        zdir_remove (zdir_t *self, bool force);
    
    //  Calculate differences between two versions of a directory tree.
    //  Returns a list of zdir_patch_t patches. Either older or newer may
    //  be null, indicating the directory is empty/absent. If alias is set,
    //  generates virtual filename (minus path, plus alias).
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT zlist_t *
        zdir_diff (zdir_t *older, zdir_t *newer, const char *alias);
    
    //  Return full contents of directory as a zdir_patch list.
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT zlist_t *
        zdir_resync (zdir_t *self, const char *alias);
    
    //  Load directory cache; returns a hash table containing the SHA-1 digests
    //  of every file in the tree. The cache is saved between runs in .cache.
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT zhash_t *
        zdir_cache (zdir_t *self);
    
    //  Print contents of directory to open stream
    CZMQ_EXPORT void
        zdir_fprint (zdir_t *self, FILE *file, int indent);
    
    //  Print contents of directory to stdout
    CZMQ_EXPORT void
        zdir_print (zdir_t *self, int indent);
    
    //  Create a new zdir_watch actor instance:
    //
    //      zactor_t *watch = zactor_new (zdir_watch, NULL);
    //
    //  Destroy zdir_watch instance:
    //
    //      zactor_destroy (&watch);
    //
    //  Enable verbose logging of commands and activity:
    //
    //      zstr_send (watch, "VERBOSE");
    //
    //  Subscribe to changes to a directory path:
    //
    //      zsock_send (watch, "ss", "SUBSCRIBE", "directory_path");
    //
    //  Unsubscribe from changes to a directory path:
    //
    //      zsock_send (watch, "ss", "UNSUBSCRIBE", "directory_path");
    //
    //  Receive directory changes:
    //      zsock_recv (watch, "sp", &path, &patches);
    //
    //      // Delete the received data.
    //      free (path);
    //      zlist_destroy (&patches);
    CZMQ_EXPORT void
        zdir_watch (zsock_t *pipe, void *unused);
    
    //  Self test of this class.
    CZMQ_EXPORT void
        zdir_test (bool verbose);
    

    
    const char *SELFTEST_DIR_RW = "src/selftest-rw";
    
    const char *testbasedir  = "zdir-test-dir";
    const char *testfile1 = "initial_file";
    const char *testfile2 = "test_abc";
    char *basedirpath = NULL;   // subdir in a test, under SELFTEST_DIR_RW
    char *filepath1 = NULL;      // pathname to testfile in a test, in dirpath
    char *filepath2 = NULL;      // pathname to testfile in a test, in dirpath
    
    basedirpath = zsys_sprintf ("%s/%s", SELFTEST_DIR_RW, testbasedir);
    assert (basedirpath);
    filepath1 = zsys_sprintf ("%s/%s", basedirpath, testfile1);
    assert (filepath1);
    filepath2 = zsys_sprintf ("%s/%s", basedirpath, testfile2);
    assert (filepath2);
    
    /*
    char *relfilepath2 = NULL;      // pathname to testfile in a test, in dirpath
    relfilepath2 = zsys_sprintf ("%s/%s", testbasedir, testfile2);
    assert (relfilepath2);
    */
    
    // Make sure old aborted tests do not hinder us
    zdir_t *dir = zdir_new (basedirpath, NULL);
    if (dir) {
        zdir_remove (dir, true);
        zdir_destroy (&dir);
    }
    zsys_file_delete (filepath1);
    zsys_file_delete (filepath2);
    zsys_dir_delete  (basedirpath);
    
    dir = zdir_new ("does-not-exist", NULL);
    if (dir) {
        zdir_remove (dir, true);
        zdir_destroy (&dir);
    }
    
    // need to create a file in the test directory we're watching
    // in order to ensure the directory exists
    zfile_t *initfile = zfile_new (basedirpath, testfile1);
    assert (initfile);
    zfile_output (initfile);
    fprintf (zfile_handle (initfile), "initial file\n");
    zfile_close (initfile);
    zfile_destroy (&initfile);
    
    zdir_t *older = zdir_new (basedirpath, NULL);
    assert (older);
    if (verbose) {
        printf ("\n");
        zdir_dump (older, 0);
    }
    zdir_t *newer = zdir_new (SELFTEST_DIR_RW, NULL);
    assert (newer);
    zlist_t *patches = zdir_diff (older, newer, "/");
    assert (patches);
    while (zlist_size (patches)) {
        zdir_patch_t *patch = (zdir_patch_t *) zlist_pop (patches);
        zdir_patch_destroy (&patch);
    }
    zlist_destroy (&patches);
    zdir_destroy (&older);
    zdir_destroy (&newer);
    
    zdir_t *nosuch = zdir_new ("does-not-exist", NULL);
    assert (nosuch == NULL);
    
    // zdir_watch test:
    zactor_t *watch = zactor_new (zdir_watch, NULL);
    assert (watch);
    
    int synced;
    if (verbose) {
        zsock_send (watch, "s", "VERBOSE");
        synced = zsock_wait(watch);
        assert ( synced == 0);
    }
    
    // wait for initial file to become 'stable'
    #ifdef CZMQ_BUILD_DRAFT_API
    zclock_sleep ((int)zsys_file_stable_age_msec() + 50);
    #else
    zclock_sleep (5050);
    #endif
    
    zsock_send (watch, "si", "TIMEOUT", 100);
    synced = zsock_wait(watch);
    assert (synced == 0);
    
    zsock_send (watch, "ss", "SUBSCRIBE", basedirpath);
    synced = zsock_wait(watch);
    assert(synced == 0);
    
    zsock_send (watch, "ss", "UNSUBSCRIBE", basedirpath);
    synced = zsock_wait(watch);
    assert(synced == 0);
    
    zsock_send (watch, "ss", "SUBSCRIBE", basedirpath);
    synced = zsock_wait(watch);
    assert(synced == 0);
    
    zfile_t *newfile = zfile_new (basedirpath, testfile2);
    zfile_output (newfile);
    fprintf (zfile_handle (newfile), "test file\n");
    zfile_close (newfile);
    
    zpoller_t *watch_poll = zpoller_new (watch, NULL);
    
    // poll for a certain timeout before giving up and failing the test
    void* polled = NULL;
    #ifdef CZMQ_BUILD_DRAFT_API
    polled = zpoller_wait(watch_poll, (int)zsys_file_stable_age_msec() + 150);
    #else
    polled = zpoller_wait(watch_poll, 5150);
    #endif
    assert (polled == watch);
    
    // wait for notification of the file being added
    char *path;
    int rc = zsock_recv (watch, "sp", &path, &patches);
    assert (rc == 0);
    
    assert (streq (path, basedirpath));
    freen (path);
    
    if (verbose)
        zsys_debug("zdir_test() : added : zlist_size (patches)=%d",
            zlist_size (patches) );
    assert (zlist_size (patches) == 1);
    
    zdir_patch_t *patch = (zdir_patch_t *) zlist_pop (patches);
    if (verbose)
        zsys_debug("zdir_test() : added : zdir_patch_path (patch)='%s'",
            zdir_patch_path (patch) );
    assert (streq (zdir_patch_path (patch), basedirpath));
    
    zfile_t *patch_file = zdir_patch_file (patch);
    if (verbose)
        zsys_debug("zdir_test() : added : zfile_filename (patch_file, \"\")='%s'",
            zfile_filename (patch_file, "") );
    assert (streq (zfile_filename (patch_file, ""), filepath2));
    
    zdir_patch_destroy (&patch);
    zlist_destroy (&patches);
    
    // remove the file
    zfile_remove (newfile);
    zfile_destroy (&newfile);
    
    // poll for a certain timeout before giving up and failing the test.
    #ifdef CZMQ_BUILD_DRAFT_API
    polled = zpoller_wait(watch_poll, (int)zsys_file_stable_age_msec() + 150);
    #else
    polled = zpoller_wait(watch_poll, 5150);
    #endif
    assert (polled == watch);
    
    // wait for notification of the file being removed
    rc = zsock_recv (watch, "sp", &path, &patches);
    assert (rc == 0);
    
    assert (streq (path, basedirpath));
    freen (path);
    
    if (verbose)
        zsys_debug("zdir_test() : removed : zlist_size (patches)=%d",
            zlist_size (patches) );
    assert (zlist_size (patches) == 1);
    
    patch = (zdir_patch_t *) zlist_pop (patches);
    if (verbose)
        zsys_debug("zdir_test() : removed : zdir_patch_path (patch)='%s'",
            zdir_patch_path (patch) );
    assert (streq (zdir_patch_path (patch), basedirpath));
    
    patch_file = zdir_patch_file (patch);
    if (verbose)
        zsys_debug("zdir_test() : removed : zfile_filename (patch_file, \"\")='%s'",
            zfile_filename (patch_file, "") );
    assert (streq (zfile_filename (patch_file, ""), filepath2));
    
    zdir_patch_destroy (&patch);
    zlist_destroy (&patches);
    
    zpoller_destroy (&watch_poll);
    zactor_destroy (&watch);
    
    // clean up by removing the test directory.
    dir = zdir_new (basedirpath, NULL);
    assert (dir);
    zdir_remove (dir, true);
    zdir_destroy (&dir);
    
    zstr_free (&basedirpath);
    zstr_free (&filepath1);
    zstr_free (&filepath2);
    
    #if defined (__WINDOWS__)
    zsys_shutdown();
    #endif

    //  This is a stable class, and may not change except for emergencies. It
    //  is provided in stable builds.
    #define ZDIR_PATCH_CREATE 1                  // Creates a new file
    #define ZDIR_PATCH_DELETE 2                  // Delete a file
    
    //  Create new patch
    CZMQ_EXPORT zdir_patch_t *
        zdir_patch_new (const char *path, zfile_t *file, int op, const char *alias);
    
    //  Destroy a patch
    CZMQ_EXPORT void
        zdir_patch_destroy (zdir_patch_t **self_p);
    
    //  Create copy of a patch. If the patch is null, or memory was exhausted,
    //  returns null.
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT zdir_patch_t *
        zdir_patch_dup (zdir_patch_t *self);
    
    //  Return patch file directory path
    CZMQ_EXPORT const char *
        zdir_patch_path (zdir_patch_t *self);
    
    //  Return patch file item
    CZMQ_EXPORT zfile_t *
        zdir_patch_file (zdir_patch_t *self);
    
    //  Return operation
    CZMQ_EXPORT int
        zdir_patch_op (zdir_patch_t *self);
    
    //  Return patch virtual file path
    CZMQ_EXPORT const char *
        zdir_patch_vpath (zdir_patch_t *self);
    
    //  Calculate hash digest for file (create only)
    CZMQ_EXPORT void
        zdir_patch_digest_set (zdir_patch_t *self);
    
    //  Return hash digest for patch file
    CZMQ_EXPORT const char *
        zdir_patch_digest (zdir_patch_t *self);
    
    //  Self test of this class.
    CZMQ_EXPORT void
        zdir_patch_test (bool verbose);
    

    
    const char *SELFTEST_DIR_RW = "src/selftest-rw";
    
    const char *testfile = "bilbo";
    const char *prefix   = "/";
    char *prefixed_testfile = zsys_sprintf ("%s%s", prefix, testfile);
    assert (prefixed_testfile);
    
    // Make sure old aborted tests do not hinder us
    zsys_file_delete (prefixed_testfile);
    
    zfile_t *file = zfile_new (SELFTEST_DIR_RW, testfile);
    assert (file);
    zdir_patch_t *patch = zdir_patch_new (SELFTEST_DIR_RW, file, patch_create, prefix);
    assert (patch);
    zfile_destroy (&file);
    
    file = zdir_patch_file (patch);
    assert (file);
    assert (streq (zfile_filename (file, SELFTEST_DIR_RW), testfile));
    assert (streq (zdir_patch_vpath (patch), prefixed_testfile));
    zdir_patch_destroy (&patch);
    
    zstr_free (&prefixed_testfile);
    
    #if defined (__WINDOWS__)
    zsys_shutdown();
    #endif

    //  This is a stable class, and may not change except for emergencies. It
    //  is provided in stable builds.
    //  If file exists, populates properties. CZMQ supports portable symbolic
    //  links, which are files with the extension ".ln". A symbolic link is a
    //  text file containing one line, the filename of a target file. Reading
    //  data from the symbolic link actually reads from the target file. Path
    //  may be NULL, in which case it is not used.
    CZMQ_EXPORT zfile_t *
        zfile_new (const char *path, const char *name);
    
    //  Destroy a file item
    CZMQ_EXPORT void
        zfile_destroy (zfile_t **self_p);
    
    //  Duplicate a file item, returns a newly constructed item. If the file
    //  is null, or memory was exhausted, returns null.
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT zfile_t *
        zfile_dup (zfile_t *self);
    
    //  Return file name, remove path if provided
    CZMQ_EXPORT const char *
        zfile_filename (zfile_t *self, const char *path);
    
    //  Refresh file properties from disk; this is not done automatically
    //  on access methods, otherwise it is not possible to compare directory
    //  snapshots.
    CZMQ_EXPORT void
        zfile_restat (zfile_t *self);
    
    //  Return when the file was last modified. If you want this to reflect the
    //  current situation, call zfile_restat before checking this property.
    CZMQ_EXPORT time_t
        zfile_modified (zfile_t *self);
    
    //  Return the last-known size of the file. If you want this to reflect the
    //  current situation, call zfile_restat before checking this property.
    CZMQ_EXPORT off_t
        zfile_cursize (zfile_t *self);
    
    //  Return true if the file is a directory. If you want this to reflect
    //  any external changes, call zfile_restat before checking this property.
    CZMQ_EXPORT bool
        zfile_is_directory (zfile_t *self);
    
    //  Return true if the file is a regular file. If you want this to reflect
    //  any external changes, call zfile_restat before checking this property.
    CZMQ_EXPORT bool
        zfile_is_regular (zfile_t *self);
    
    //  Return true if the file is readable by this process. If you want this to
    //  reflect any external changes, call zfile_restat before checking this
    //  property.
    CZMQ_EXPORT bool
        zfile_is_readable (zfile_t *self);
    
    //  Return true if the file is writeable by this process. If you want this
    //  to reflect any external changes, call zfile_restat before checking this
    //  property.
    CZMQ_EXPORT bool
        zfile_is_writeable (zfile_t *self);
    
    //  Check if file has stopped changing and can be safely processed.
    //  Updates the file statistics from disk at every call.
    CZMQ_EXPORT bool
        zfile_is_stable (zfile_t *self);
    
    //  Return true if the file was changed on disk since the zfile_t object
    //  was created, or the last zfile_restat() call made on it.
    CZMQ_EXPORT bool
        zfile_has_changed (zfile_t *self);
    
    //  Remove the file from disk
    CZMQ_EXPORT void
        zfile_remove (zfile_t *self);
    
    //  Open file for reading
    //  Returns 0 if OK, -1 if not found or not accessible
    CZMQ_EXPORT int
        zfile_input (zfile_t *self);
    
    //  Open file for writing, creating directory if needed
    //  File is created if necessary; chunks can be written to file at any
    //  location. Returns 0 if OK, -1 if error.
    CZMQ_EXPORT int
        zfile_output (zfile_t *self);
    
    //  Read chunk from file at specified position. If this was the last chunk,
    //  sets the eof property. Returns a null chunk in case of error.
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT zchunk_t *
        zfile_read (zfile_t *self, size_t bytes, off_t offset);
    
    //  Returns true if zfile_read() just read the last chunk in the file.
    CZMQ_EXPORT bool
        zfile_eof (zfile_t *self);
    
    //  Write chunk to file at specified position
    //  Return 0 if OK, else -1
    CZMQ_EXPORT int
        zfile_write (zfile_t *self, zchunk_t *chunk, off_t offset);
    
    //  Read next line of text from file. Returns a pointer to the text line,
    //  or NULL if there was nothing more to read from the file.
    CZMQ_EXPORT const char *
        zfile_readln (zfile_t *self);
    
    //  Close file, if open
    CZMQ_EXPORT void
        zfile_close (zfile_t *self);
    
    //  Return file handle, if opened
    CZMQ_EXPORT FILE *
        zfile_handle (zfile_t *self);
    
    //  Calculate SHA1 digest for file, using zdigest class.
    CZMQ_EXPORT const char *
        zfile_digest (zfile_t *self);
    
    //  Self test of this class.
    CZMQ_EXPORT void
        zfile_test (bool verbose);
    
    #ifdef CZMQ_BUILD_DRAFT_API
    //  *** Draft method, for development use, may change without warning ***
    //  Create new temporary file for writing via tmpfile. File is automatically
    //  deleted on destroy
    CZMQ_EXPORT zfile_t *
        zfile_tmp (void);
    
    #endif // CZMQ_BUILD_DRAFT_API
    //  These methods are deprecated, and now moved to zsys class.
    CZMQ_EXPORT bool
        zfile_exists (const char *filename);
    CZMQ_EXPORT ssize_t
        zfile_size   (const char *filename);
    CZMQ_EXPORT mode_t
        zfile_mode   (const char *filename);
    CZMQ_EXPORT int
        zfile_delete (const char *filename);
    CZMQ_EXPORT bool
        zfile_stable (const char *filename);
    CZMQ_EXPORT int
        zfile_mkdir  (const char *pathname);
    CZMQ_EXPORT int
        zfile_rmdir  (const char *pathname);
    CZMQ_EXPORT void
        zfile_mode_private (void);
    CZMQ_EXPORT void
        zfile_mode_default (void);

    
    const char *SELFTEST_DIR_RW = "src/selftest-rw";
    
    const char *testbasedir  = "this";
    const char *testsubdir  = "is/a/test";
    const char *testfile = "bilbo";
    const char *testlink = "bilbo.ln";
    char *basedirpath = NULL;   // subdir in a test, under SELFTEST_DIR_RW
    char *dirpath = NULL;       // subdir in a test, under basedirpath
    char *filepath = NULL;      // pathname to testfile in a test, in dirpath
    char *linkpath = NULL;      // pathname to testlink in a test, in dirpath
    
    basedirpath = zsys_sprintf ("%s/%s", SELFTEST_DIR_RW, testbasedir);
    assert (basedirpath);
    dirpath = zsys_sprintf ("%s/%s", basedirpath, testsubdir);
    assert (dirpath);
    filepath = zsys_sprintf ("%s/%s", dirpath, testfile);
    assert (filepath);
    linkpath = zsys_sprintf ("%s/%s", dirpath, testlink);
    assert (linkpath);
    
    // This subtest is specifically for NULL as current directory, so
    // no SELFTEST_DIR_RW here; testfile should have no slashes inside.
    // Normally tests clean up in zfile_destroy(), but if a selftest run
    // dies e.g. on assert(), workspace remains dirty. Better clean it up.
    if (zfile_exists (testfile) ) {
        if (verbose)
            zsys_debug ("zfile_test() has to remove ./%s that should not have been here", testfile);
        zfile_delete (testfile);
    }
    zfile_t *file = zfile_new (NULL, testfile);
    assert (file);
    assert (streq (zfile_filename (file, "."), testfile));
    assert (zfile_is_readable (file) == false);
    zfile_destroy (&file);
    
    //  Create a test file in some random subdirectory
    if (verbose)
        zsys_debug ("zfile_test() at timestamp %" PRIi64 ": "
            "Creating new zfile %s",
            zclock_time(), filepath );
    
    if (zfile_exists (filepath) ) {
        if (verbose)
            zsys_debug ("zfile_test() has to remove %s that should not have been here", filepath);
        zfile_delete (filepath);
    }
    
    file = zfile_new (dirpath, testfile);
    assert (file);
    int rc = zfile_output (file);
    assert (rc == 0);
    zchunk_t *chunk = zchunk_new (NULL, 100);
    assert (chunk);
    zchunk_fill (chunk, 0, 100);
    
    //  Write 100 bytes at position 1,000,000 in the file
    if (verbose)
        zsys_debug ("zfile_test() at timestamp %" PRIi64 ": "
            "Writing 100 bytes at position 1,000,000 in the file",
            zclock_time() );
    rc = zfile_write (file, chunk, 1000000);
    if (verbose)
        zsys_debug ("zfile_test() at timestamp %" PRIi64 ": "
            "Wrote 100 bytes at position 1,000,000 in the file, result code %d",
            zclock_time(), rc );
    assert (rc == 0);
    zchunk_destroy (&chunk);
    zfile_close (file);
    assert (zfile_is_readable (file));
    assert (zfile_cursize (file) == 1000100);
    if (verbose)
        zsys_debug ("zfile_test() at timestamp %" PRIi64 ": "
            "Testing if file is NOT stable (is younger than 1 sec)",
            zclock_time() );
    assert (!zfile_is_stable (file));
    if (verbose)
        zsys_debug ("zfile_test() at timestamp %" PRIi64 ": "
            "Passed the lag-dependent tests",
            zclock_time() );
    assert (zfile_digest (file));
    
    //  Now truncate file from outside
    int handle = open (filepath, O_WRONLY | O_TRUNC | O_BINARY, 0);
    assert (handle >= 0);
    rc = write (handle, "Hello, World\n", 13);
    assert (rc == 13);
    close (handle);
    assert (zfile_has_changed (file));
    #ifdef CZMQ_BUILD_DRAFT_API
    zclock_sleep ((int)zsys_file_stable_age_msec() + 50);
    #else
    zclock_sleep (5050);
    #endif
    assert (zfile_has_changed (file));
    
    assert (!zfile_is_stable (file));
    zfile_restat (file);
    assert (zfile_is_stable (file));
    assert (streq (zfile_digest (file), "4AB299C8AD6ED14F31923DD94F8B5F5CB89DFB54"));
    
    //  Check we can read from file
    rc = zfile_input (file);
    assert (rc == 0);
    chunk = zfile_read (file, 1000100, 0);
    assert (chunk);
    assert (zchunk_size (chunk) == 13);
    zchunk_destroy (&chunk);
    zfile_close (file);
    
    //  Check we can read lines from file
    rc = zfile_input (file);
    assert (rc == 0);
    const char *line = zfile_readln (file);
    assert (streq (line, "Hello, World"));
    line = zfile_readln (file);
    assert (line == NULL);
    zfile_close (file);
    
    //  Try some fun with symbolic links
    zfile_t *link = zfile_new (dirpath, testlink);
    assert (link);
    rc = zfile_output (link);
    assert (rc == 0);
    fprintf (zfile_handle (link), "%s\n", filepath);
    zfile_destroy (&link);
    
    link = zfile_new (dirpath, testlink);
    assert (link);
    rc = zfile_input (link);
    assert (rc == 0);
    chunk = zfile_read (link, 1000100, 0);
    assert (chunk);
    assert (zchunk_size (chunk) == 13);
    zchunk_destroy (&chunk);
    zfile_destroy (&link);
    
    //  Remove file and directory
    zdir_t *dir = zdir_new (basedirpath, NULL);
    assert (dir);
    assert (zdir_cursize (dir) == 26);
    zdir_remove (dir, true);
    assert (zdir_cursize (dir) == 0);
    zdir_destroy (&dir);
    
    //  Check we can no longer read from file
    assert (zfile_is_readable (file));
    zfile_restat (file);
    assert (!zfile_is_readable (file));
    rc = zfile_input (file);
    assert (rc == -1);
    zfile_destroy (&file);
    
    // This set of tests is done, free the strings for reuse
    zstr_free (&basedirpath);
    zstr_free (&dirpath);
    zstr_free (&filepath);
    zstr_free (&linkpath);
    
    const char *eof_checkfile = "eof_checkfile";
    filepath = zsys_sprintf ("%s/%s", SELFTEST_DIR_RW, eof_checkfile);
    assert (filepath);
    
    if (zfile_exists (filepath) ) {
        if (verbose)
            zsys_debug ("zfile_test() has to remove %s that should not have been here", filepath);
        zfile_delete (filepath);
    }
    zstr_free (&filepath);
    
    file = zfile_new (SELFTEST_DIR_RW, eof_checkfile);
    assert (file);
    
    //  1. Write something first
    rc = zfile_output (file);
    assert (rc == 0);
    chunk = zchunk_new ("123456789", 9);
    assert (chunk);
    
    rc = zfile_write (file, chunk, 0);
    assert (rc == 0);
    zchunk_destroy (&chunk);
    zfile_close (file);
    assert (zfile_cursize (file) == 9);
    
    // 2. Read the written something
    rc = zfile_input (file);
    assert (rc != -1);
    // try to read more bytes than there is in the file
    chunk = zfile_read (file, 1000, 0);
    assert (zfile_eof(file));
    assert (zchunk_streq (chunk, "123456789"));
    zchunk_destroy (&chunk);
    
    // reading is ok
    chunk = zfile_read (file, 5, 0);
    assert (!zfile_eof(file));
    assert (zchunk_streq (chunk, "12345"));
    zchunk_destroy (&chunk);
    
    // read from non zero offset until the end
    chunk = zfile_read (file, 5, 5);
    assert (zfile_eof(file));
    assert (zchunk_streq (chunk, "6789"));
    zchunk_destroy (&chunk);
    zfile_remove (file);
    zfile_close (file);
    zfile_destroy (&file);
    
    #ifdef CZMQ_BUILD_DRAFT_API
    zfile_t *tempfile = zfile_tmp ();
    assert (tempfile);
    assert (zfile_filename (tempfile, NULL));
    assert (zsys_file_exists (zfile_filename (tempfile, NULL)));
    zchunk_t *tchunk = zchunk_new ("HELLO", 6);
    assert (zfile_write (tempfile, tchunk, 0) == 0);
    zchunk_destroy (&tchunk);
    
    char *filename = strdup (zfile_filename (tempfile, NULL));
    zfile_destroy (&tempfile);
    assert (!zsys_file_exists (filename));
    zstr_free (&filename);
    #endif // CZMQ_BUILD_DRAFT_API
    
    #if defined (__WINDOWS__)
    zsys_shutdown();
    #endif
    

    //  This is a stable class, and may not change except for emergencies. It
    //  is provided in stable builds.
    //  This class has draft methods, which may change over time. They are not
    //  in stable releases, by default. Use --enable-drafts to enable.
    #define ZFRAME_MORE 1                        //
    #define ZFRAME_REUSE 2                       //
    #define ZFRAME_DONTWAIT 4                    //
    
    //  Create a new frame. If size is not null, allocates the frame data
    //  to the specified size. If additionally, data is not null, copies
    //  size octets from the specified data into the frame body.
    CZMQ_EXPORT zframe_t *
        zframe_new (const void *data, size_t size);
    
    //  Create an empty (zero-sized) frame
    CZMQ_EXPORT zframe_t *
        zframe_new_empty (void);
    
    //  Create a frame with a specified string content.
    CZMQ_EXPORT zframe_t *
        zframe_from (const char *string);
    
    //  Receive frame from socket, returns zframe_t object or NULL if the recv
    //  was interrupted. Does a blocking recv, if you want to not block then use
    //  zpoller or zloop.
    CZMQ_EXPORT zframe_t *
        zframe_recv (void *source);
    
    //  Destroy a frame
    CZMQ_EXPORT void
        zframe_destroy (zframe_t **self_p);
    
    //  Send a frame to a socket, destroy frame after sending.
    //  Return -1 on error, 0 on success.
    CZMQ_EXPORT int
        zframe_send (zframe_t **self_p, void *dest, int flags);
    
    //  Return number of bytes in frame data
    CZMQ_EXPORT size_t
        zframe_size (zframe_t *self);
    
    //  Return address of frame data
    CZMQ_EXPORT byte *
        zframe_data (zframe_t *self);
    
    //  Return meta data property for frame
    //  The caller shall not modify or free the returned value, which shall be
    //  owned by the message.
    CZMQ_EXPORT const char *
        zframe_meta (zframe_t *self, const char *property);
    
    //  Create a new frame that duplicates an existing frame. If frame is null,
    //  or memory was exhausted, returns null.
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT zframe_t *
        zframe_dup (zframe_t *self);
    
    //  Return frame data encoded as printable hex string, useful for ØMQ UUIDs.
    //  Caller must free string when finished with it.
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT char *
        zframe_strhex (zframe_t *self);
    
    //  Return frame data copied into freshly allocated string
    //  Caller must free string when finished with it.
    //  Caller owns return value and must destroy it when done.
    CZMQ_EXPORT char *
        zframe_strdup (zframe_t *self);
    
    //  Return TRUE if frame body is equal to string, excluding terminator
    CZMQ_EXPORT bool
        zframe_streq (zframe_t *self, const char *string);
    
    //  Return frame MORE indicator (1 or 0), set when reading frame from socket
    //  or by the zframe_set_more() method
    CZMQ_EXPORT int
        zframe_more (zframe_t *self);
    
    //  Set frame MORE indicator (1 or 0). Note this is NOT used when sending
    //  frame to socket, you have to specify flag explicitly.
    CZMQ_EXPORT void
        zframe_set_more (zframe_t *self, int more);
    
    //  Return TRUE if two frames have identical size and data
    //  If either frame is NULL, equality is always false.
    CZMQ_EXPORT bool
        zframe_eq (zframe_t *self, zframe_t *other);
    
    //  Set new contents for frame
    CZMQ_EXPORT void
        zframe_reset (zframe_t *self, const void *data, size_t size);
    
    //  Send message to zsys log sink (may be stdout, or system facility as
    //  configured by zsys_set_logstream). Prefix shows before frame, if not null.
    //  Long messages are truncated.
    CZMQ_EXPORT void
        zframe_print (zframe_t *self, const char *prefix);
    
    //  Probe the supplied object, and report if it looks like a zframe_t.
    CZMQ_EXPORT bool
        zframe_is (void *self);
    
    //  Self test of this class.
    CZMQ_EXPORT void
        zframe_test (bool verbose);
    
    #ifdef CZMQ_BUILD_DRAFT_API
    // Destroy an item
    typedef void (zframe_destructor_fn) (
        void **hint);
    
    //  *** Draft method, for development use, may change without warning ***
    //  Create a new frame from memory. Take ownership of the memory and calling the destructor
    //  on destroy.
    CZMQ_EXPORT zframe_t *
        zframe_frommem (void *data, size_t size, zframe_destructor_fn destructor, void *hint);
    
    //  *** Draft method, for development use, may change without warning ***
    //  Return frame routing ID, if the frame came from a ZMQ_SERVER socket.
    //  Else returns zero.
    CZMQ_EXPORT uint32_t
        zframe_routing_id (zframe_t *self);
    
    //  *** Draft method, for development use, may change without warning ***
    //  Set routing ID on frame. This is used if/when the frame is sent to a
    //  ZMQ_SERVER socket.
    CZMQ_EXPORT void
        zframe_set_routing_id (zframe_t *self, uint32_t routing_id);
    
    //  *** Draft method, for development use, may change without warning ***
    //  Return frame group of radio-dish pattern.
    CZMQ_EXPORT const char *
        zframe_group (zframe_t *self);
    
    //  *** Draft method, for development use, may change without warning ***
    //  Set group on frame. This is used if/when the frame is sent to a
    //  ZMQ_RADIO socket.
    //  Return -1 on error, 0 on success.
    CZMQ_EXPORT int
        zframe_set_group (zframe_t *self, const char *group);
    
    //  *** Draft method, for development use, may change without warning ***
    //  Send message to zsys log sink (may be stdout, or system facility as
    //  configured by zsys_set_logstream). Prefix shows before frame, if not null.
    //  Message length is specified; no truncation unless length is zero.
    //  Backwards compatible with zframe_print when length is zero.
    CZMQ_EXPORT void
        zframe_print_n (zframe_t *self, const char *prefix, size_t length);
    
    #endif // CZMQ_BUILD_DRAFT_API

    //  Create two PAIR sockets and connect over TCP
    zsock_t *output = zsock_new (ZMQ_PAIR);
    assert (output);
    int port = zsock_bind (output, "tcp://127.0.0.1:*");
    assert (port != -1);
    zsock_t *input = zsock_new (ZMQ_PAIR);
    assert (input);
    rc = zsock_connect (input, "tcp://127.0.0.1:%d", port);
    assert (rc != -1);
    
    //  Send five different frames, test ZFRAME_MORE
    int frame_nbr;
    for (frame_nbr = 0; frame_nbr < 5; frame_nbr++) {
        frame = zframe_new ("Hello", 5);
        assert (frame);
        rc = zframe_send (&frame, output, ZFRAME_MORE);
        assert (rc == 0);
    }
    //  Send same frame five times, test ZFRAME_REUSE
    frame = zframe_new ("Hello", 5);
    assert (frame);
    for (frame_nbr = 0; frame_nbr < 5; frame_nbr++) {
        rc = zframe_send (&frame, output, ZFRAME_MORE + ZFRAME_REUSE);
        assert (rc == 0);
    }
    assert (frame);
    zframe_t *copy = zframe_dup (frame);
    assert (zframe_eq (frame, copy));
    zframe_destroy (&frame);
    assert (!zframe_eq (frame, copy));
    assert (zframe_size (copy) == 5);
    zframe_destroy (&copy);
    assert (!zframe_eq (frame, copy));
    
    //  Test zframe_new_empty
    frame = zframe_new_empty ();
    assert (frame);
    assert (zframe_size (frame) == 0);
    zframe_destroy (&frame);
    
    //  Send END frame
    frame = zframe_new ("NOT", 3);
    assert (frame);
    zframe_reset (frame, "END", 3);
    char *string = zframe_strhex (frame);
    assert (streq (string, "454E44"));
    freen (string);
    string = zframe_strdup (frame);
    assert (streq (string, "END"));
    freen (string);
    rc = zframe_send (&frame, output, 0);
    assert (rc == 0);
    
    //  Read and count until we receive END
    frame_nbr = 0;
    for (frame_nbr = 0;; frame_nbr++) {
        zframe_t *frame = zframe_recv (input);
        if (zframe_streq (frame, "END")) {
            zframe_destroy (&frame);
            break;
        }
        assert (zframe_more (frame));
        zframe_set_more (frame, 0);
        assert (zframe_more (frame) == 0);
        zframe_destroy (&frame);
    }
    assert (frame_nbr == 10);
    
    #if (ZMQ_VERSION >= ZMQ_MAKE_VERSION (4, 1, 0))
    // Test zframe_meta
    frame = zframe_new ("Hello", 5);
    assert (frame);
    rc = zframe_send (&frame, output, 0);
    assert (rc == 0);
    frame = zframe_recv (input);
    const char *meta = zframe_meta (frame, "Socket-Type");
    assert (meta != NULL);
    assert (streq (meta, "PAIR"));
    assert (zframe_meta (frame, "nonexistent") == NULL);
    zframe_destroy (&frame);
    #endif
    
    zsock_destroy (&input);
    zsock_destroy (&output);
    
    #if defined (ZMQ_SERVER)
    //  Create server and client sockets and connect over inproc
    zsock_t *server = zsock_new_server ("inproc://zframe-test-routing");
    assert (server);
    zsock_t *client = zsock_new_client ("inproc://zframe-test-routing");
    assert (client);
    
    //  Send request from client to server
    zframe_t *request = zframe_new ("Hello", 5);
    assert (request);
    rc = zframe_send (&request, client, 0);
    assert (rc == 0);
    assert (!request);
    
    //  Read request and send reply
    request = zframe_recv (server);
    assert (request);
    assert (zframe_streq (request, "Hello"));
    assert (zframe_routing_id (request));
    
    zframe_t *reply = zframe_new ("World", 5);
    assert (reply);
    zframe_set_routing_id (reply, zframe_routing_id (request));
    rc = zframe_send (&reply, server, 0);
    assert (rc == 0);
    zframe_destroy (&request);
    
    //  Read reply
    reply = zframe_recv (client);
    assert (zframe_streq (reply, "World"));
    assert (zframe_routing_id (reply) == 0);
    zframe_destroy (&reply);
    
    //  Client and server disallow multipart
    frame = zframe_new ("Hello", 5);
    rc = zframe_send (&frame, client, ZFRAME_MORE);
    assert (rc == -1);
    rc = zframe_send (&frame, server, ZFRAME_MORE);
    assert (rc == -1);
    zframe_destroy (&frame);
    
    zsock_destroy (&client);
    zsock_destroy (&server);
    #endif
    
    #ifdef ZMQ_RADIO
    //  Create radio and dish sockets and connect over inproc
    zsock_t *radio = zsock_new_radio ("inproc://zframe-test-radio");
    assert (radio);
    zsock_t *dish = zsock_new_dish ("inproc://zframe-test-radio");
    assert (dish);
    
    //  Join the group
    rc = zsock_join (dish, "World");
    assert (rc == 0);
    
    //  Publish message from radio
    zframe_t *message = zframe_new ("Hello", 5);
    assert (message);
    rc = zframe_set_group (message, "World");
    assert (rc == 0);
    rc = zframe_send (&message, radio, 0);
    assert (rc == 0);
    assert (!message);
    
    //  Receive the message from dish
    message = zframe_recv (dish);
    assert (message);
    assert (zframe_streq (message, "Hello"));
    assert (strcmp("World", zframe_group (message)) == 0);
    zframe_destroy (&message);
    
    zsock_destroy (&dish);
    zsock_destroy (&radio);
    #else
    frame = zframe_new ("Hello", 5);
    rc = zframe_set_group (frame, "World");
    assert(rc == -1);
    assert(errno == ENOTSUP);
    zframe_destroy (&frame);
    #endif
    
    char str[] = "hello";
    frame = zframe_frommem (str, 5, mem_destructor, str);
    assert (frame);
    zframe_destroy (&frame);
    
    //  The destructor doesn't free the memory, only changing the strid,
    //  so we can check if the destructor was invoked
    assert (streq (str, "world"));
    
    
    // zframe_print tests
    
    zsys_set_logstream(verbose ? stdout : NULL);
    
    // == No data ==
    frame = zframe_new ("", 0);
    
    // no prefix, backwards compatible
    //  - emits nothing but the timestamp
    zframe_print (frame, "");
    zframe_print_n (frame, "", 0);
    
    // prefix, backwards compatible
    //  - emits nothing but the timestamp
    zframe_print (frame, "Prefix");
    zframe_print_n (frame, "Prefix", 0);
    
    // len > data
    //  - emits nothing but the timestamp
    zframe_print_n (frame, "", 15);
    zframe_print_n (frame, "Prefix", 15);
    
    // max len
    //  - emits nothing but the timestamp
    zframe_print_n (frame, "", -1);
    zframe_print_n (frame, "Prefix", -1);
    
    zframe_destroy (&frame);
    
    
    // == Short data ==
    frame = zframe_new ("Hello there!", 12);
    
    // no prefix, backwards compatible: ellipsis
    //  - "[012] Hello there!"
    zframe_print (frame, "");
    zframe_print_n (frame, "", 0);
    
    // prefix, backwards compatible: ellipsis
    //  - "Prefix[012] Hello there!"
    zframe_print (frame, "Prefix");
    zframe_print_n (frame, "Prefix", 0);
    
    // len < data
    //  - "[012] Hello"
    //  - "Prefix[012] Hello"
    zframe_print_n (frame, "", 5);
    zframe_print_n (frame, "Prefix", 5);
    
    // len > data
    //  - "[012] Hello there!"
    //  - "Prefix[012] Hello there!"
    zframe_print_n (frame, "", 15);
    zframe_print_n (frame, "Prefix", 15);
    
    // max len
    //  - "[012] Hello there!"
    //  - "Prefix[012] Hello there!"
    zframe_print_n (frame, "", -1);
    zframe_print_n (frame, "Prefix", -1);
    
    zframe_destroy (&frame);
    
    
    // == Long data ==
    frame = zframe_new ("Lorem ipsum dolor sit amet, consectetur adipiscing elit. Proin finibus ligula et aliquam tristique. Phasellus consequat, enim et blandit varius, sapien diam faucibus lorem, non ultricies lacus turpis sed lectus. Vivamus id elit urna. In sit amet lacinia mauris. Class aptent taciti sociosqu ad litora torquent per conubia nostra, per inceptos himenaeos. Integer ut cursus diam. Vestibulum semper vel leo eu finibus. Ut urna magna, commodo vel auctor sed, eleifend quis lacus. Aenean quis ipsum et velit auctor ultrices.", 519);
    
    // no prefix, backwards compatible: ellipsis
    //  - "[070] Lorem ipsum dolor sit amet, consectetur adipiscing elit. Proin finibus..."
    zframe_print (frame, "");
    zframe_print_n (frame, "", 0);
    
    // prefix, backwards compatible: ellipsis
    //  - "Prefix[070] Lorem ipsum dolor sit amet, consectetur adipiscing elit. Proin finibus..."
    zframe_print (frame, "Prefix");
    zframe_print_n (frame, "Prefix", 0);
    
    // len < data
    //  - "[519] Lorem"
    //  - "Prefix[519] Lorem"
    zframe_print_n (frame, "", 5);
    zframe_print_n (frame, "Prefix", 5);
    
    // small len
    //  - "[519] Lorem ipsum dolor sit amet"
    //  - "Prefix[519] Lorem ipsum dolor sit amet"
    zframe_print_n (frame, "", 26);
    zframe_print_n (frame, "Prefix", 26);
    
    // mid len
    // - "[519] Lorem ipsum dolor sit amet, consectetur adipiscing elit. Proin finibus"
    //   "[519]  ligula et aliquam tristique. Phasellus consequat, enim et blandit var"
    //   "[519] ius, sapie"
    // - "Prefix[519] Lorem ipsum dolor sit amet, consectetur adipiscing elit. Proin finibus"
    //   "Prefix[519]  ligula et aliquam tristique. Phasellus consequat, enim et blandit var"
    //   "Prefix[519] ius, sapie"
    zframe_print_n (frame, "", 150);
    zframe_print_n (frame, "Prefix", 150);
    
    // len > data
    //  - emits the whole paragraph
    zframe_print_n (frame, "", 1500);
    zframe_print_n (frame, "Prefix", 1500);
    
    // max len
    //  - emits the whole paragraph
    zframe_print_n (frame, "", -1);
    zframe_print_n (frame, "Prefix", -1);
    
    zframe_destroy (&frame);
    
    #if defined (__WINDOWS__)
    zsys_shutdown();
    #endif