<NME> zsock.c <BEF> /* ========================================================================= zsock - high-level socket API that hides libzmq contexts and sockets Copyright (c) the Contributors as noted in the AUTHORS file. This file is part of CZMQ, the high-level C binding for 0MQ: http://czmq.zeromq.org. This Source Code Form is subject to the terms of the Mozilla Public License, v. 2.0. If a copy of the MPL was not distributed with this file, You can obtain one at http://mozilla.org/MPL/2.0/. ========================================================================= */ /* @header The zsock class wraps the libzmq socket handle (a void *) with a proper structure that follows the CLASS rules for construction and destruction. Some zsock methods take a void * "polymorphic" reference, which can be either a zsock_t or a zactor_t reference, or a libzmq void *. @discuss @end */ #define ZSOCK_NOCHECK // we are defining the methods here, so don't redirect symbols. #include "czmq_classes.h" #include "zsock_option.inc" // zsock_t instances always have this tag as the first 4 octets of // their data, which lets us do runtime object typing & validation. #define ZSOCK_TAG 0xcafe0004 // This port range is defined by IANA for dynamic or private ports // We use this when choosing a port for dynamic binding. #define DYNAMIC_FIRST 0xc000 // 49152 #define DYNAMIC_LAST 0xffff // 65535 #define ZSOCK_BSEND_MAX_FRAMES 32 // Arbitrary limit, for now // Structure of our class struct _zsock_t { uint32_t tag; // Object tag for runtime detection void *handle; // The libzmq socket handle char *endpoint; // Last bound endpoint, if any char *cache; // Holds last zsock_brecv strings int type; // Socket type size_t cache_size; // Current size of cache uint32_t routing_id; // Routing ID for server sockets }; #ifndef CZMQ_BUILD_DRAFT_API CZMQ_PRIVATE zsock_t * zsock_new_server_checked (const char *endpoint, const char *filename, size_t line_nbr); CZMQ_PRIVATE zsock_t * zsock_new_client_checked (const char *endpoint, const char *filename, size_t line_nbr); CZMQ_PRIVATE zsock_t * zsock_new_radio_checked (const char *endpoint, const char *filename, size_t line_nbr); CZMQ_PRIVATE zsock_t * zsock_new_dish_checked (const char *endpoint, const char *filename, size_t line_nbr); CZMQ_PRIVATE zsock_t * zsock_new_gather_checked (const char *endpoint, const char *filename, size_t line_nbr); CZMQ_PRIVATE zsock_t * zsock_new_scatter_checked (const char *endpoint, const char *filename, size_t line_nbr); #endif // CZMQ_BUILD_DRAFT_API // -------------------------------------------------------------------------- // Create a new socket. This macro passes the caller source and line // number so that CZMQ can report socket leaks intelligently. To switch // off this checking, which may be costly if you use a LOT of sockets, // define ZSOCK_NOCHECK before compiling your code. Returns the new // socket, or NULL if the new socket could not be created. zsock_t * zsock_new_checked (int type, const char *filename, size_t line_nbr) { zsock_t *self = (zsock_t *) zmalloc (sizeof (zsock_t)); assert (self); self->tag = ZSOCK_TAG; self->type = type; self->handle = zsys_socket (type, filename, line_nbr); assert (self->handle); return self; } zsock_t * zsock_new (int type) { return zsock_new_checked (type, NULL, 0); } // -------------------------------------------------------------------------- // Destroy the socket. You must use this for any socket created via the // zsock_new method. void zsock_destroy_checked (zsock_t **self_p, const char *filename, size_t line_nbr) { assert (self_p); if (*self_p) { zsock_t *self = *self_p; assert (zsock_is (self)); self->tag = 0xDeadBeef; int rc = zsys_close (self->handle, filename, line_nbr); assert (rc == 0); freen (self->endpoint); freen (self->cache); freen (self); *self_p = NULL; } } void zsock_destroy (zsock_t **self_p) { zsock_destroy_checked (self_p, NULL, 0); } // -------------------------------------------------------------------------- // This interface includes some smart constructors, which create sockets with // additional set-up. In all of these, the endpoint is NULL, or starts with // '@' (bind) or '>' (connect). Multiple endpoints are allowed, separated by // commas. If endpoint does not start with '@' or '>', default action depends // on socket type. // Create a PUB socket. Default action is bind. zsock_t * zsock_new_pub_checked (const char *endpoints, const char *filename, size_t line_nbr) { zsock_t *sock = zsock_new_checked (ZMQ_PUB, filename, line_nbr); if (zsock_attach (sock, endpoints, true)) zsock_destroy (&sock); return sock; } zsock_t * zsock_new_pub (const char *endpoints) { return zsock_new_pub_checked (endpoints, NULL, 0); } // -------------------------------------------------------------------------- // Create a SUB socket, and optionally subscribe to some prefix string. Default // action is connect. zsock_t * zsock_new_sub_checked (const char *endpoints, const char *subscribe, const char *filename, size_t line_nbr) { zsock_t *sock = zsock_new_checked (ZMQ_SUB, filename, line_nbr); if (zsock_attach (sock, endpoints, false) == 0) { if (subscribe) zsock_set_subscribe (sock, subscribe); } else zsock_destroy (&sock); return sock; } zsock_t * zsock_new_sub (const char *endpoints, const char *subscribe) { return zsock_new_sub_checked (endpoints, subscribe, NULL, 0); } // -------------------------------------------------------------------------- // Create a REQ socket. Default action is connect. zsock_t * zsock_new_req_checked (const char *endpoints, const char *filename, size_t line_nbr) { zsock_t *sock = zsock_new_checked (ZMQ_REQ, filename, line_nbr); if (zsock_attach (sock, endpoints, false)) zsock_destroy (&sock); return sock; } zsock_t * zsock_new_req (const char *endpoints) { return zsock_new_req_checked (endpoints, NULL, 0); } // -------------------------------------------------------------------------- // Create a REP socket. Default action is bind. zsock_t * zsock_new_rep_checked (const char *endpoints, const char *filename, size_t line_nbr) { zsock_t *sock = zsock_new_checked (ZMQ_REP, filename, line_nbr); if (zsock_attach (sock, endpoints, true)) zsock_destroy (&sock); return sock; } zsock_t * zsock_new_rep (const char *endpoints) { return zsock_new_rep_checked (endpoints, NULL, 0); } // -------------------------------------------------------------------------- // Create a DEALER socket. Default action is connect. zsock_t * zsock_new_dealer_checked (const char *endpoints, const char *filename, size_t line_nbr) { zsock_t *sock = zsock_new_checked (ZMQ_DEALER, filename, line_nbr); if (zsock_attach (sock, endpoints, false)) zsock_destroy (&sock); return sock; } zsock_t * zsock_new_dealer (const char *endpoints) { return zsock_new_dealer_checked (endpoints, NULL, 0); } // -------------------------------------------------------------------------- // Create a ROUTER socket. Default action is bind. zsock_t * zsock_new_router_checked (const char *endpoints, const char *filename, size_t line_nbr) { zsock_t *sock = zsock_new_checked (ZMQ_ROUTER, filename, line_nbr); if (zsock_attach (sock, endpoints, true)) zsock_destroy (&sock); return sock; } zsock_t * zsock_new_router (const char *endpoints) { return zsock_new_router_checked (endpoints, NULL, 0); } // -------------------------------------------------------------------------- // Create a PUSH socket. Default action is connect. zsock_t * zsock_new_push_checked (const char *endpoints, const char *filename, size_t line_nbr) { zsock_t *sock = zsock_new_checked (ZMQ_PUSH, filename, line_nbr); if (zsock_attach (sock, endpoints, false)) zsock_destroy (&sock); return sock; } zsock_t * zsock_new_push (const char *endpoints) { return zsock_new_push_checked(endpoints, NULL, 0); } // -------------------------------------------------------------------------- // Create a PULL socket. Default action is bind. zsock_t * zsock_new_pull_checked (const char *endpoints, const char *filename, size_t line_nbr) { zsock_t *sock = zsock_new_checked (ZMQ_PULL, filename, line_nbr); if (zsock_attach (sock, endpoints, true)) zsock_destroy (&sock); return sock; } zsock_t * zsock_new_pull (const char *endpoints) { return zsock_new_pull_checked (endpoints, NULL, 0); } // -------------------------------------------------------------------------- // Create an XPUB socket. Default action is bind. zsock_t * zsock_new_xpub_checked (const char *endpoints, const char *filename, size_t line_nbr) { #if defined ZMQ_XPUB zsock_t *sock = zsock_new_checked (ZMQ_XPUB, filename, line_nbr); if (zsock_attach (sock, endpoints, true)) zsock_destroy (&sock); return sock; #else // errno = ENOTSUP; // too late, stable API would be broken return NULL; // Not implemented #endif } zsock_t * zsock_new_xpub (const char *endpoints) { return zsock_new_xpub_checked (endpoints, NULL, 0); } // -------------------------------------------------------------------------- // Create an XSUB socket. Default action is connect. zsock_t * zsock_new_xsub_checked (const char *endpoints, const char *filename, size_t line_nbr) { #if defined ZMQ_XSUB zsock_t *sock = zsock_new_checked (ZMQ_XSUB, filename, line_nbr); if (zsock_attach (sock, endpoints, false)) zsock_destroy (&sock); return sock; #else // errno = ENOTSUP; // too late, stable API would be broken return NULL; // Not implemented #endif } zsock_t * zsock_new_xsub (const char *endpoints) { return zsock_new_xsub_checked (endpoints, NULL, 0); } // -------------------------------------------------------------------------- // Create a PAIR socket. Default action is connect. zsock_t * zsock_new_pair_checked (const char *endpoints, const char *filename, size_t line_nbr) { zsock_t *sock = zsock_new_checked (ZMQ_PAIR, filename, line_nbr); if (zsock_attach (sock, endpoints, false)) zsock_destroy (&sock); return sock; } zsock_t * zsock_new_pair (const char *endpoints) { assert (zsock_is (self)); #if (ZMQ_VERSION >= ZMQ_MAKE_VERSION (3,2,0)) char endpoint [256]; va_list argptr; va_start (argptr, format); vsnprintf (endpoint, 256, format, argptr); va_end (argptr); return zmq_unbind (self->handle, endpoint); #else return -1; #endif return sock; #else // errno = ENOTSUP; // too late, stable API would be broken return NULL; // Not implemented #endif } zsock_t * zsock_new_stream (const char *endpoints) { assert (self); assert (zsock_is (self)); char endpoint [256]; va_list argptr; va_start (argptr, format); vsnprintf (endpoint, 256, format, argptr); va_end (argptr); int rc = zmq_connect (self->handle, endpoint); #if (ZMQ_VERSION < ZMQ_MAKE_VERSION (4,0,0)) int retries = 4; while (rc == -1 && zmq_errno () == ECONNREFUSED && retries) { return sock; #else errno = ENOTSUP; return NULL; #endif retries--; } #endif return rc; } // -------------------------------------------------------------------------- // Create a CLIENT socket. Default action is connect. zsock_t * zsock_new_client_checked (const char *endpoints, const char *filename, size_t line_nbr) { #if defined ZMQ_CLIENT zsock_t *sock = zsock_new_checked (ZMQ_CLIENT, filename, line_nbr); assert (zsock_is (self)); #if (ZMQ_VERSION >= ZMQ_MAKE_VERSION (3,2,0)) char endpoint [256]; va_list argptr; va_start (argptr, format); vsnprintf (endpoint, 256, format, argptr); va_end (argptr); return zmq_disconnect (self->handle, endpoint); #else return -1; #endif // -------------------------------------------------------------------------- // Create a RADIO socket. Default action is bind. zsock_t * zsock_new_radio_checked (const char *endpoints, const char *filename, size_t line_nbr) { #if defined ZMQ_RADIO zsock_t *sock = zsock_new_checked (ZMQ_RADIO, filename, line_nbr); if (zsock_attach (sock, endpoints, true)) zsock_destroy (&sock); return sock; #else errno = ENOTSUP; return NULL; #endif } zsock_t * zsock_new_radio (const char *endpoints) { return zsock_new_radio_checked (endpoints, NULL, 0); } // -------------------------------------------------------------------------- // Create a DISH socket. Default action is connect. zsock_t * zsock_new_dish_checked (const char *endpoints, const char *filename, size_t line_nbr) { #if defined ZMQ_DISH zsock_t *sock = zsock_new_checked (ZMQ_DISH, filename, line_nbr); if (zsock_attach (sock, endpoints, false)) zsock_destroy (&sock); return sock; #else errno = ENOTSUP; return NULL; #endif } zsock_t * zsock_new_dish (const char *endpoints) { return zsock_new_dish_checked (endpoints, NULL, 0); } // -------------------------------------------------------------------------- // Create a GATHER socket. Default action is bind. zsock_t * zsock_new_gather_checked (const char *endpoints, const char *filename, size_t line_nbr) { #if defined ZMQ_GATHER zsock_t *sock = zsock_new_checked (ZMQ_GATHER, filename, line_nbr); if (zsock_attach (sock, endpoints, true)) zsock_destroy (&sock); return sock; #else errno = ENOTSUP; return NULL; #endif } zsock_t * zsock_new_gather (const char *endpoints) { return zsock_new_gather_checked (endpoints, NULL, 0); } // -------------------------------------------------------------------------- // Create a scatter socket. Default action is connect. zsock_t * zsock_new_scatter_checked (const char *endpoints, const char *filename, size_t line_nbr) { #if defined ZMQ_SCATTER zsock_t *sock = zsock_new_checked (ZMQ_SCATTER, filename, line_nbr); if (zsock_attach (sock, endpoints, false)) zsock_destroy (&sock); return sock; #else errno = ENOTSUP; return NULL; #endif } zsock_t * zsock_new_scatter (const char *endpoints) { return zsock_new_scatter_checked (endpoints, NULL, 0); } // -------------------------------------------------------------------------- // Create a dgram socket. Default action is bind. zsock_t * zsock_new_dgram_checked (const char *endpoints, const char *filename, size_t line_nbr) { #if defined ZMQ_DGRAM zsock_t *sock = zsock_new_checked (ZMQ_DGRAM, filename, line_nbr); if (zsock_attach (sock, endpoints, true)) zsock_destroy (&sock); return sock; #else errno = ENOTSUP; return NULL; #endif } // *** Draft method, for development use, may change without warning *** // Create a DGRAM socket. Default action is bind. zsock_t * zsock_new_dgram (const char *endpoint) { return zsock_new_dgram_checked ( endpoint, NULL, 0 ); } // -------------------------------------------------------------------------- // Bind a socket to a formatted endpoint. For tcp:// endpoints, supports // ephemeral ports, if you specify the port number as "*". By default // zsock uses the IANA designated range from C000 (49152) to FFFF (65535). // To override this range, follow the "*" with "[first-last]". Either or // both first and last may be empty. To bind to a random port within the // range, use "!" in place of "*". // // Examples: // tcp://127.0.0.1:* bind to first free port from C000 up // tcp://127.0.0.1:! bind to random port from C000 to FFFF // tcp://127.0.0.1:*[60000-] bind to first free port from 60000 up // tcp://127.0.0.1:![-60000] bind to random port from C000 to 60000 // tcp://127.0.0.1:![55000-55999] bind to random port from 55000-55999 // // On success, returns the actual port number used, for tcp:// endpoints, // and 0 for other transports. On failure, returns -1. Note that when using // ephemeral ports, a port may be reused by different services without // clients being aware. Protocols that run on ephemeral ports should take // this into account. int zsock_bind (zsock_t *self, const char *format, ...) { assert (self); assert (zsock_is (self)); // Expand format to get full endpoint va_list argptr; va_start (argptr, format); char *endpoint = zsys_vprintf (format, argptr); va_end (argptr); if (!endpoint) return -1; int rc; // If tcp:// endpoint, parse to get or make port number zrex_t *rex = zrex_new (NULL); if (zrex_eq (rex, endpoint, "^tcp://.*:(\\d+)$")) { assert (zrex_hits (rex) == 2); int port = atoi (zrex_hit (rex, 1)); #if defined (HAVE_LIBSYSTEMD) && (ZMQ_VERSION >= ZMQ_MAKE_VERSION (4, 2, 0)) if (zsys_auto_use_fd ()) { int last_handle = SD_LISTEN_FDS_START + sd_listen_fds (0); int handle; for (handle = SD_LISTEN_FDS_START; handle < last_handle; ++handle) if (sd_is_socket_inet (handle, AF_UNSPEC, SOCK_STREAM, 1, (uint16_t) port) > 0) { zsock_set_use_fd (self, handle); break; } } #endif if (zmq_bind (self->handle, endpoint) == 0) rc = port; else rc = -1; } else if (zrex_eq (rex, endpoint, "^(tcp://.*):([*!])(\\[(\\d+)?-(\\d+)?\\])?$")) { assert (zrex_hits (rex) == 6); const char *hostname, *opcode, *group, *first_str, *last_str; zrex_fetch (rex, &hostname, &opcode, &group, &first_str, &last_str, NULL); int first = *first_str? atoi (first_str): DYNAMIC_FIRST; int last = *last_str? atoi (last_str): DYNAMIC_LAST; // This is how many times we'll try before giving up int attempts = last - first + 1; // If operator is '*', take first available port. // If operator is '!', take a random leap into our port space; we'll // still scan sequentially to make sure we find a free slot rapidly. int port = first; if (streq (opcode, "!")) port += randof (attempts); rc = -1; // Assume we don't succeed while (rc == -1 && attempts--) { zstr_free (&endpoint); endpoint = zsys_sprintf ("%s:%d", hostname, port); if (!endpoint) break; if (zmq_bind (self->handle, endpoint) == 0) rc = port; if (++port > last) port = first; } } else { #if defined (HAVE_LIBSYSTEMD) && (ZMQ_VERSION >= ZMQ_MAKE_VERSION (4, 2, 0)) if (zsys_auto_use_fd () && zrex_eq (rex, endpoint, "^ipc://(.*)$")) { assert (zrex_hits (rex) == 2); const char *sock_path; zrex_fetch (rex, &sock_path, NULL); int last_handle = SD_LISTEN_FDS_START + sd_listen_fds (0); int handle; for (handle = SD_LISTEN_FDS_START; handle < last_handle; ++handle) if (sd_is_socket_unix (handle, SOCK_STREAM, 1, sock_path, 0) > 0) { zsock_set_use_fd (self, handle); break; } } #endif rc = zmq_bind (self->handle, endpoint); } // Store successful endpoint for later reference if (rc >= 0) { zstr_free (&self->endpoint); self->endpoint = endpoint; } else zstr_free (&endpoint); zrex_destroy (&rex); return rc; } // -------------------------------------------------------------------------- // Returns last bound endpoint, if any. const char * zsock_endpoint (zsock_t *self) { assert (self); return self->endpoint; } // -------------------------------------------------------------------------- // Unbind a socket from a formatted endpoint. // Returns 0 if OK, -1 if the endpoint was invalid or the function // isn't supported. int zsock_unbind (zsock_t *self, const char *format, ...) { assert (self); assert (zsock_is (self)); #if (ZMQ_VERSION >= ZMQ_MAKE_VERSION (3, 2, 0)) // Expand format to get full endpoint va_list argptr; va_start (argptr, format); char *endpoint = zsys_vprintf (format, argptr); va_end (argptr); if (!endpoint) return -1; int rc = zmq_unbind (self->handle, endpoint); zstr_free (&endpoint); return rc; #else return -1; #endif } // -------------------------------------------------------------------------- // Connect a socket to a formatted endpoint // Returns 0 if the endpoint is valid, -1 if the connect failed. int zsock_connect (zsock_t *self, const char *format, ...) { assert (self); assert (zsock_is (self)); // Expand format to get full endpoint va_list argptr; va_start (argptr, format); char *endpoint = zsys_vprintf (format, argptr); va_end (argptr); if (!endpoint) return -1; int rc = zmq_connect (self->handle, endpoint); #if (ZMQ_VERSION < ZMQ_MAKE_VERSION (4, 0, 0)) int retries = 4; while (rc == -1 && zmq_errno () == ECONNREFUSED && retries) { // This bruteforces a synchronization between connecting and // binding threads on ZeroMQ v3.2 and earlier, where the bind // MUST happen before the connect on inproc transports. zclock_sleep (250); rc = zmq_connect (self->handle, endpoint); retries--; } #endif zstr_free (&endpoint); return rc; } // -------------------------------------------------------------------------- // Disconnect a socket from a formatted endpoint // Returns 0 if disconnection is complete -1 if the disconnection failed. int zsock_disconnect (zsock_t *self, const char *format, ...) { assert (self); assert (zsock_is (self)); #if (ZMQ_VERSION >= ZMQ_MAKE_VERSION (3, 2, 0)) // Expand format to get full endpoint va_list argptr; va_start (argptr, format); char *endpoint = zsys_vprintf (format, argptr); va_end (argptr); if (!endpoint) return -1; int rc = zmq_disconnect (self->handle, endpoint); zstr_free (&endpoint); return rc; #else return -1; #endif } // -------------------------------------------------------------------------- // Attach a socket to zero or more endpoints. If endpoints is not null, // parses as list of ZeroMQ endpoints, separated by commas, and prefixed by // '@' (to bind the socket) or '>' (to attach the socket). Returns 0 if all // endpoints were valid, or -1 if there was a syntax error. If the endpoint // does not start with '@' or '>', the serverish argument defines whether // it is used to bind (serverish = true) or connect (serverish = false). int zsock_attach (zsock_t *self, const char *endpoints, bool serverish) { assert (self); if (!endpoints) return 0; // We hold each individual endpoint here char endpoint [256]; while (*endpoints) { const char *delimiter = strchr (endpoints, ','); if (!delimiter) delimiter = endpoints + strlen (endpoints); if (delimiter - endpoints > 255) return -1; memcpy (endpoint, endpoints, delimiter - endpoints); endpoint [delimiter - endpoints] = 0; int rc; if (endpoint [0] == '@') rc = zsock_bind (self, "%s", endpoint + 1); else if (endpoint [0] == '>') rc = zsock_connect (self, "%s", endpoint + 1); else if (serverish) rc = zsock_bind (self, "%s", endpoint); else rc = zsock_connect (self, "%s", endpoint); if (rc == -1) return -1; // Bad endpoint syntax if (*delimiter == 0) break; endpoints = delimiter + 1; } return 0; } // -------------------------------------------------------------------------- // Returns socket type as printable constant string. const char * zsock_type_str (zsock_t *self) { assert (self); return zsys_sockname (self->type); } // -------------------------------------------------------------------------- // Send a 'picture' message to the socket (or actor). The picture is a // string that defines the type of each frame. This makes it easy to send // a complex multiframe message in one call. The picture can contain any // of these characters, each corresponding to one or two arguments: // // i = int (signed) // 1 = uint8_t // 2 = uint16_t // 4 = uint32_t // 8 = uint64_t // s = char * // b = byte *, int (2 arguments) // c = zchunk_t * // f = zframe_t * // h = zhashx_t * // l = zlistx_t * // U = zuuid_t * // p = void * (sends the pointer value, only meaningful over inproc) // m = zmsg_t * (sends all frames in the zmsg) // z = sends zero-sized frame (0 arguments) // u = uint (deprecated) // // Note that s, b, c, and f are encoded the same way and the choice is // offered as a convenience to the sender, which may or may not already // have data in a zchunk or zframe. Does not change or take ownership of // any arguments. Returns 0 if successful, -1 if sending failed for any // reason. int zsock_send (void *self, const char *picture, ...) { va_list argptr; va_start (argptr, picture); int rc = zsock_vsend (self, picture, argptr); va_end (argptr); return rc; } // -------------------------------------------------------------------------- // Send a 'picture' message to the socket (or actor). This is a va_list // version of zsock_send (), so please consult its documentation for the // details. int zsock_vsend (void *self, const char *picture, va_list argptr) { assert (self); assert (picture); zmsg_t *msg = zmsg_new (); while (*picture) { if (*picture == 'i') zmsg_addstrf (msg, "%d", va_arg (argptr, int)); else if (*picture == '1') zmsg_addstrf (msg, "%" PRIu8, (uint8_t) va_arg (argptr, int)); else if (*picture == '2') zmsg_addstrf (msg, "%" PRIu16, (uint16_t) va_arg (argptr, int)); else if (*picture == '4') zmsg_addstrf (msg, "%" PRIu32, va_arg (argptr, uint32_t)); else if (*picture == '8') zmsg_addstrf (msg, "%" PRIu64, va_arg (argptr, uint64_t)); else if (*picture == 'u') // Deprecated, use 4 or 8 instead zmsg_addstrf (msg, "%ud", va_arg (argptr, uint)); else if (*picture == 's') zmsg_addstr (msg, va_arg (argptr, char *)); else if (*picture == 'b') { // Note function arguments may be expanded in reverse order, // so we cannot use va_arg macro twice in a single call byte *data = va_arg (argptr, byte *); zmsg_addmem (msg, data, va_arg (argptr, int)); } else if (*picture == 'c') { zchunk_t *chunk = va_arg (argptr, zchunk_t *); assert (zchunk_is (chunk)); zmsg_addmem (msg, zchunk_data (chunk), zchunk_size (chunk)); } else if (*picture == 'f') { zframe_t *frame = va_arg (argptr, zframe_t *); assert (zframe_is (frame)); zmsg_addmem (msg, zframe_data (frame), zframe_size (frame)); } else if (*picture == 'U') { zuuid_t *uuid = va_arg (argptr, zuuid_t *); zmsg_addmem (msg, zuuid_data (uuid), zuuid_size (uuid)); } else if (*picture == 'p') { void *pointer = va_arg (argptr, void *); zmsg_addmem (msg, &pointer, sizeof (void *)); } else if (*picture == 'h') { zhashx_t *hash = va_arg (argptr, zhashx_t *); zframe_t *frame = zhashx_pack (hash); zmsg_append (msg, &frame); } #ifdef CZMQ_BUILD_DRAFT_API else if (*picture == 'l') { zlistx_t *list = va_arg (argptr, zlistx_t *); zframe_t *frame = zlistx_pack (list); zmsg_append (msg, &frame); } #endif else if (*picture == 'm') { zframe_t *frame; zmsg_t *zmsg = va_arg (argptr, zmsg_t *); for (frame = zmsg_first (zmsg); frame; frame = zmsg_next (zmsg) ) { zframe_t *frame_dup = zframe_dup (frame); zmsg_append (msg, &frame_dup); } } else if (*picture == 'z') zmsg_addmem (msg, NULL, 0); else { zsys_error ("zsock: invalid picture element '%c'", *picture); assert (false); } picture++; } int rc = zmsg_send (&msg, self); if (rc != 0) zmsg_destroy (&msg); return rc; } // -------------------------------------------------------------------------- // Receive a 'picture' message to the socket (or actor). See zsock_send for // the format and meaning of the picture. Returns the picture elements into // a series of pointers as provided by the caller: // // i = int * (stores signed integer) // 1 = uint8_t * (stores 8-bit unsigned integer) // 2 = uint16_t * (stores 16-bit unsigned integer) // 4 = uint32_t * (stores 32-bit unsigned integer) // 8 = uint64_t * (stores 64-bit unsigned integer) // s = char ** (allocates new string) // b = byte **, size_t * (2 arguments) (allocates memory) // c = zchunk_t ** (creates zchunk) // f = zframe_t ** (creates zframe) // h = zhashx_t ** (creates zhashx) // l = zlistx_t ** (creates zlistx) // U = zuuid_t * (creates a zuuid with the data) // p = void ** (stores pointer) // m = zmsg_t ** (creates a zmsg with the remaing frames) // z = null, asserts empty frame (0 arguments) // u = uint * (stores unsigned integer, deprecated) // // Note that zsock_recv creates the returned objects, and the caller must // destroy them when finished with them. The supplied pointers do not need // to be initialized. Returns 0 if successful, or -1 if it failed to recv // a message, in which case the pointers are not modified. When message // frames are truncated (a short message), sets return values to zero/null. // If an argument pointer is NULL, does not store any value (skips it). // An 'n' picture matches an empty frame; if the message does not match, // the method will return -1. int zsock_recv (void *self, const char *picture, ...) { va_list argptr; va_start (argptr, picture); int rc = zsock_vrecv (self, picture, argptr); va_end (argptr); return rc; } // Receive a 'picture' message from the socket (or actor). This is a // va_list version of zsock_recv (), so please consult its documentation // for the details. int zsock_vrecv (void *self, const char *picture, va_list argptr) { assert (self); assert (picture); zmsg_t *msg = zmsg_recv (self); if (!msg) return -1; // Interrupted // Filter a signal that may come from a dying actor if (zmsg_signal (msg) >= 0) { zmsg_destroy (&msg); return -1; } // Now parse message according to picture argument int rc = 0; while (*picture) { if (*picture == 'i') { char *string = zmsg_popstr (msg); int *int_p = va_arg (argptr, int *); if (int_p) *int_p = string? atoi (string): 0; freen (string); } else if (*picture == '1') { char *string = zmsg_popstr (msg); uint8_t *uint8_p = va_arg (argptr, uint8_t *); if (uint8_p) *uint8_p = string? (uint8_t) atoi (string): 0; freen (string); } else if (*picture == '2') { char *string = zmsg_popstr (msg); uint16_t *uint16_p = va_arg (argptr, uint16_t *); if (uint16_p) *uint16_p = string? (uint16_t) atol (string): 0; freen (string); } else if (*picture == '4') { char *string = zmsg_popstr (msg); uint32_t *uint32_p = va_arg (argptr, uint32_t *); if (uint32_p) *uint32_p = string? (uint32_t) strtoul (string, NULL, 10): 0; freen (string); } else if (*picture == '8') { char *string = zmsg_popstr (msg); uint64_t *uint64_p = va_arg (argptr, uint64_t *); if (uint64_p) *uint64_p = string? (uint64_t) strtoull (string, NULL, 10): 0; freen (string); } else if (*picture == 'u') { // Deprecated, use 4 or 8 instead char *string = zmsg_popstr (msg); uint *uint_p = va_arg (argptr, uint *); if (uint_p) *uint_p = string? (uint) strtoul (string, NULL, 10): 0; freen (string); } else if (*picture == 's') { char *string = zmsg_popstr (msg); char **string_p = va_arg (argptr, char **); if (string_p) *string_p = string; else freen (string); } else if (*picture == 'b') { zframe_t *frame = zmsg_pop (msg); byte **data_p = va_arg (argptr, byte **); size_t *size = va_arg (argptr, size_t *); if (data_p) { if (frame) { *size = zframe_size (frame); *data_p = (byte *) malloc (*size); memcpy (*data_p, zframe_data (frame), *size); } else { *data_p = NULL; *size = 0; } } zframe_destroy (&frame); } else if (*picture == 'c') { zframe_t *frame = zmsg_pop (msg); zchunk_t **chunk_p = va_arg (argptr, zchunk_t **); if (chunk_p) { if (frame) *chunk_p = zchunk_new (zframe_data (frame), zframe_size (frame)); else *chunk_p = NULL; } zframe_destroy (&frame); } else if (*picture == 'f') { zframe_t *frame = zmsg_pop (msg); zframe_t **frame_p = va_arg (argptr, zframe_t **); if (frame_p) *frame_p = frame; else zframe_destroy (&frame); } else if (*picture == 'U') { zframe_t *frame = zmsg_pop (msg); zuuid_t **uuid_p = va_arg (argptr, zuuid_t **); if (uuid_p) { if (frame) { *uuid_p = zuuid_new (); zuuid_set (*uuid_p, zframe_data (frame)); } else *uuid_p = NULL; } zframe_destroy (&frame); } else if (*picture == 'p') { zframe_t *frame = zmsg_pop (msg); void **pointer_p = va_arg (argptr, void **); if (pointer_p) { if (frame) { if (zframe_size (frame) == sizeof (void *)) *pointer_p = *((void **) zframe_data (frame)); else rc = -1; } else *pointer_p = NULL; } zframe_destroy (&frame); } else if (*picture == 'h') { zframe_t *frame = zmsg_pop (msg); zhashx_t **hash_p = va_arg (argptr, zhashx_t **); if (hash_p) { if (frame) *hash_p = zhashx_unpack (frame); else *hash_p = NULL; } zframe_destroy (&frame); } #ifdef CZMQ_BUILD_DRAFT_API else if (*picture == 'l') { zframe_t *frame = zmsg_pop (msg); zlistx_t **list_p = va_arg (argptr, zlistx_t **); if (list_p) { if (frame) *list_p = zlistx_unpack (frame); else *list_p = NULL; } zframe_destroy (&frame); } #endif else if (*picture == 'm') { zmsg_t **zmsg_p = va_arg (argptr, zmsg_t **); if (zmsg_p) { *zmsg_p = zmsg_new (); zframe_t *frame; while ((frame = zmsg_pop (msg))) zmsg_append (*zmsg_p, &frame); } } else if (*picture == 'z') { zframe_t *frame = zmsg_pop (msg); if (frame && zframe_size (frame) != 0) rc = -1; zframe_destroy (&frame); } else { zsys_error ("zsock: invalid picture element '%c'", *picture); assert (false); } picture++; } zmsg_destroy (&msg); return rc; } // -------------------------------------------------------------------------- // Network data encoding macros that we use in bsend/brecv // Put a 1-byte number to the frame #define PUT_NUMBER1(host) { \ *(byte *) needle = (host); \ needle++; \ } // Put a 2-byte number to the frame #define PUT_NUMBER2(host) { \ needle [0] = (byte) (((host) >> 8) & 255); \ needle [1] = (byte) (((host)) & 255); \ needle += 2; \ } // Put a 4-byte number to the frame #define PUT_NUMBER4(host) { \ needle [0] = (byte) (((host) >> 24) & 255); \ needle [1] = (byte) (((host) >> 16) & 255); \ needle [2] = (byte) (((host) >> 8) & 255); \ needle [3] = (byte) (((host)) & 255); \ needle += 4; \ } // Put a 8-byte number to the frame #define PUT_NUMBER8(host) { \ needle [0] = (byte) (((host) >> 56) & 255); \ needle [1] = (byte) (((host) >> 48) & 255); \ needle [2] = (byte) (((host) >> 40) & 255); \ needle [3] = (byte) (((host) >> 32) & 255); \ needle [4] = (byte) (((host) >> 24) & 255); \ needle [5] = (byte) (((host) >> 16) & 255); \ needle [6] = (byte) (((host) >> 8) & 255); \ needle [7] = (byte) (((host)) & 255); \ needle += 8; \ } // Get a 1-byte number from the frame #define GET_NUMBER1(host) { \ if (needle + 1 > ceiling) \ goto malformed; \ (host) = *(byte *) needle; \ needle++; \ } // Get a 2-byte number from the frame #define GET_NUMBER2(host) { \ if (needle + 2 > ceiling) \ goto malformed; \ (host) = ((uint16_t) (needle [0]) << 8) \ + (uint16_t) (needle [1]); \ needle += 2; \ } // Get a 4-byte number from the frame #define GET_NUMBER4(host) { \ if (needle + 4 > ceiling) \ goto malformed; \ (host) = ((uint32_t) (needle [0]) << 24) \ + ((uint32_t) (needle [1]) << 16) \ + ((uint32_t) (needle [2]) << 8) \ + (uint32_t) (needle [3]); \ needle += 4; \ } // Get a 8-byte number from the frame #define GET_NUMBER8(host) { \ if (needle + 8 > ceiling) \ goto malformed; \ (host) = ((uint64_t) (needle [0]) << 56) \ + ((uint64_t) (needle [1]) << 48) \ + ((uint64_t) (needle [2]) << 40) \ + ((uint64_t) (needle [3]) << 32) \ + ((uint64_t) (needle [4]) << 24) \ + ((uint64_t) (needle [5]) << 16) \ + ((uint64_t) (needle [6]) << 8) \ + (uint64_t) (needle [7]); \ needle += 8; \ } // -------------------------------------------------------------------------- // Send a binary encoded 'picture' message to the socket (or actor). This // method is similar to zsock_send, except the arguments are encoded in a // binary format that is compatible with zproto, and is designed to reduce // memory allocations. The pattern argument is a string that defines the // type of each argument. Supports these argument types: // // pattern C type zproto type: // 1 uint8_t type = "number" size = "1" // 2 uint16_t type = "number" size = "2" // 4 uint32_t type = "number" size = "3" // 8 uint64_t type = "number" size = "4" // s char *, 0-255 chars type = "string" // S char *, 0-2^32-1 chars type = "longstr" // c zchunk_t * type = "chunk" // f zframe_t * type = "frame" // u zuuid_t * type = "uuid" // m zmsg_t * type = "msg" // p void *, sends pointer value, only over inproc // // Does not change or take ownership of any arguments. Returns 0 if // successful, -1 if sending failed for any reason. int zsock_bsend (void *self, const char *picture, ...) { assert (self); assert (picture); // Pass 1: calculate total size of data frame size_t frame_size = 0; zframe_t *frames [ZSOCK_BSEND_MAX_FRAMES]; // Non-data frames to send size_t nbr_frames = 0; // Size of this table va_list argptr; va_start (argptr, picture); const char *picptr = picture; while (*picptr) { if (*picptr == '1') { va_arg (argptr, int); frame_size += 1; } else if (*picptr == '2') { va_arg (argptr, int); frame_size += 2; } else if (*picptr == '4') { va_arg (argptr, uint32_t); frame_size += 4; } else if (*picptr == '8') { va_arg (argptr, uint64_t); frame_size += 8; } else if (*picptr == 's') { char *string = va_arg (argptr, char *); frame_size += 1 + (string? strlen (string): 0); } else if (*picptr == 'S') { char *string = va_arg (argptr, char *); frame_size += 4 + (string? strlen (string): 0); } else if (*picptr == 'c') { zchunk_t *chunk = va_arg (argptr, zchunk_t *); frame_size += 4 + (chunk? zchunk_size (chunk): 0); } else if (*picptr == 'p') { va_arg (argptr, void *); frame_size += sizeof (void *); } else if (*picptr == 'f') { zframe_t *frame = va_arg (argptr, zframe_t *); assert (nbr_frames < ZSOCK_BSEND_MAX_FRAMES - 1); frames [nbr_frames++] = frame; } else if (*picptr == 'u') { zuuid_t *uuid = va_arg (argptr, zuuid_t *); frame_size += zuuid_size (uuid); } else if (*picptr == 'm') { if (picptr [1]) { zsys_error ("zsock_bsend: 'm' (zmsg) only valid at end of picptr"); assert (false); } zmsg_t *msg = va_arg (argptr, zmsg_t *); if (msg) { zframe_t *frame = zmsg_first (msg); while (frame) { assert (nbr_frames < ZSOCK_BSEND_MAX_FRAMES - 1); frames [nbr_frames++] = frame; frame = zmsg_next (msg); } } else frames [nbr_frames++] = zframe_new_empty (); } else { zsys_error ("zsock_bsend: invalid picptr element '%c'", *picptr); assert (false); } picptr++; } va_end (argptr); // Pass 2: encode data into data frame zmq_msg_t msg; zmq_msg_init_size (&msg, frame_size); byte *needle = (byte *) zmq_msg_data (&msg); // Set routing id if self is zsock #if defined ZMQ_SERVER if (zsock_is (self) && zsock_routing_id ((zsock_t *)self) != 0) zmq_msg_set_routing_id (&msg, zsock_routing_id ((zsock_t *)self)); #endif va_start (argptr, picture); picptr = picture; while (*picptr) { if (*picptr == '1') { int number1 = va_arg (argptr, int); PUT_NUMBER1 (number1); } else if (*picptr == '2') { int number2 = va_arg (argptr, int); PUT_NUMBER2 (number2); } else if (*picptr == '4') { uint32_t number4 = va_arg (argptr, uint32_t); PUT_NUMBER4 (number4); } else if (*picptr == '8') { uint64_t number8 = va_arg (argptr, uint64_t); PUT_NUMBER8 (number8); } else if (*picptr == 'p') { void *pointer = va_arg (argptr, void *); memcpy (needle, &pointer, sizeof (void *)); needle += sizeof (void *); } else if (*picptr == 's') { char *string = va_arg (argptr, char *); if (!string) string = ""; size_t string_size = strlen (string); PUT_NUMBER1 ((byte) string_size); memcpy (needle, string, string_size); needle += string_size; } else if (*picptr == 'S') { char *string = va_arg (argptr, char *); if (!string) string = ""; size_t string_size = strlen (string); PUT_NUMBER4 (string_size); memcpy (needle, string, string_size); needle += string_size; } else if (*picptr == 'c') { zchunk_t *chunk = va_arg (argptr, zchunk_t *); if (chunk) { PUT_NUMBER4 (zchunk_size (chunk)); memcpy (needle, zchunk_data (chunk), zchunk_size (chunk)); needle += zchunk_size (chunk); } } else if (*picptr == 'u') { zuuid_t *uuid = va_arg (argptr, zuuid_t *); if (uuid) { memcpy (needle, zuuid_data (uuid), zuuid_size (uuid)); needle += zuuid_size (uuid); } } picptr++; } va_end (argptr); // Now send the data frame void *handle = zsock_resolve (self); int rc = zmq_msg_send (&msg, handle, nbr_frames? ZMQ_SNDMORE: 0); if (rc >= 0) { // Now send any additional frames unsigned int frame_nbr; for (frame_nbr = 0; frame_nbr < nbr_frames; frame_nbr++) { bool more = frame_nbr < nbr_frames - 1; rc = zframe_send (&frames [frame_nbr], self, ZFRAME_REUSE + (more? ZFRAME_MORE: 0)); if (rc == -1) break; } } return rc >= 0? 0: -1; } // -------------------------------------------------------------------------- // Receive a binary encoded 'picture' message from the socket (or actor). // This method is similar to zsock_recv, except the arguments are encoded // in a binary format that is compatible with zproto, and is designed to // reduce memory allocations. The pattern argument is a string that defines // the type of each argument. See zsock_bsend for the supported argument // types. All arguments must be pointers; this call sets them to point to // values held on a per-socket basis. // For types 1, 2, 4 and 8 the caller must allocate the memory itself before // calling zsock_brecv. // For types S, the caller must free the value once finished with it, as // zsock_brecv will allocate the buffer. // For type s, the caller must not free the value as it is stored in a // local cache for performance purposes. // For types c, f, u and m the caller must call the appropriate destructor // depending on the object as zsock_brecv will create new objects. // For type p the caller must coordinate with the sender, as it is just a // pointer value being passed. // This is the largest size we allow for an incoming longstr or chunk (1M) #define MAX_ALLOC_SIZE 1024 * 1024 int zsock_brecv (void *selfish, const char *picture, ...) { assert (selfish); // Kind of self assert (picture); zmq_msg_t msg; zmq_msg_init (&msg); if (zmq_msg_recv (&msg, zsock_resolve (selfish), 0) == -1) return -1; // Interrupted // If we don't have a string cache, create one now with arbitrary // value; this will grow if needed. Do not use an initial size less // than 256, or cache expansion will not work properly. zsock_t *self = (zsock_t *) selfish; if (zactor_is (selfish)) self = zactor_sock ((zactor_t *) selfish); if (!self->cache) { self->cache = (char *) malloc (512); self->cache_size = 512; } // Last received strings are cached per socket uint cache_used = 0; byte *needle = (byte *) zmq_msg_data (&msg); byte *ceiling = needle + zmq_msg_size (&msg); // If selfish is zsock get routing id from msg #if defined ZMQ_SERVER if (zsock_is (selfish) && zsock_type (self) == ZMQ_SERVER) zsock_set_routing_id (self, zmq_msg_routing_id (&msg)); #endif va_list argptr; va_start (argptr, picture); const char *picptr = picture; while (*picptr) { if (*picptr == '1') { uint8_t *number1_p = va_arg (argptr, uint8_t *); GET_NUMBER1 (*number1_p); } else if (*picptr == '2') { uint16_t *number2_p = va_arg (argptr, uint16_t *); GET_NUMBER2 (*number2_p); } else if (*picptr == '4') { uint32_t *number4_p = va_arg (argptr, uint32_t *); GET_NUMBER4 (*number4_p); } else if (*picptr == '8') { uint64_t *number8_p = va_arg (argptr, uint64_t *); GET_NUMBER8 (*number8_p); } else if (*picptr == 'p') { void **pointer_p = va_arg (argptr, void **); memcpy (pointer_p, needle, sizeof (void *)); needle += sizeof (void *); } else if (*picptr == 's') { char **string_p = va_arg (argptr, char **); uint string_size; GET_NUMBER1 (string_size); if (needle + string_size > ceiling) goto malformed; // Expand cache if we need to; string is guaranteed to fit into // expansion space if (cache_used + string_size > self->cache_size) { puts ("REALLOC"); self->cache_size *= 2; self->cache = (char *) realloc (self->cache, self->cache_size); assert (self->cache); } *string_p = self->cache + cache_used; memcpy (*string_p, needle, string_size); cache_used += string_size; self->cache [cache_used++] = 0; needle += string_size; } else if (*picptr == 'S') { char **string_p = va_arg (argptr, char **); size_t string_size; GET_NUMBER4 (string_size); if (string_size > MAX_ALLOC_SIZE || needle + string_size > (ceiling)) goto malformed; *string_p = (char *) malloc (string_size + 1); assert (string_p); memcpy (*string_p, needle, string_size); (*string_p) [string_size] = 0; needle += string_size; } else if (*picptr == 'c') { zchunk_t **chunk_p = va_arg (argptr, zchunk_t **); size_t chunk_size; GET_NUMBER4 (chunk_size); if (chunk_size > MAX_ALLOC_SIZE || needle + chunk_size > (ceiling)) goto malformed; *chunk_p = zchunk_new (needle, chunk_size); needle += chunk_size; } else if (*picptr == 'u') { zuuid_t **uuid_p = va_arg (argptr, zuuid_t **); *uuid_p = zuuid_new (); if (zuuid_size (*uuid_p) > MAX_ALLOC_SIZE || needle + zuuid_size (*uuid_p) > (ceiling)) goto malformed; zuuid_set (*uuid_p, needle); needle += zuuid_size (*uuid_p); } else if (*picptr == 'f') { zframe_t **frame_p = va_arg (argptr, zframe_t **); // Get next frame off socket if (!zsock_rcvmore (self)) goto malformed; *frame_p = zframe_recv (self); } else if (*picptr == 'm') { if (picptr [1]) { zsys_error ("zsock_brecv: 'm' (zmsg) only valid at end of picptr"); assert (false); } zmsg_t **msg_p = va_arg (argptr, zmsg_t **); // Get zero or more remaining frames if (!zsock_rcvmore (self)) goto malformed; *msg_p = zmsg_recv (self); } else { zsys_error ("zsock_brecv: invalid picptr element '%c'", *picptr); assert (false); } picptr++; } va_end (argptr); zmq_msg_close (&msg); return 0; // Error return malformed: zmq_msg_close (&msg); return -1; // Invalid message } // -------------------------------------------------------------------------- // Return socket routing ID if any. This returns 0 if the socket is not // of type ZMQ_SERVER or if no request was already received on it. uint32_t zsock_routing_id (zsock_t *self) { return self->routing_id; } // -------------------------------------------------------------------------- // Set routing ID on socket. The socket MUST be of type ZMQ_SERVER. // This will be used when sending messages on the socket via the zsock API. void zsock_set_routing_id (zsock_t *self, uint32_t routing_id) { assert (self); self->routing_id = routing_id; } // -------------------------------------------------------------------------- // Set socket to use unbounded pipes (HWM=0); use this in cases when you are // totally certain the message volume can fit in memory. This method works // across all versions of ZeroMQ. Takes a polymorphic socket reference. void zsock_set_unbounded (void *self) { #if (ZMQ_VERSION_MAJOR == 2) zsock_set_hwm (self, 0); #else zsock_set_sndhwm (self, 0); zsock_set_rcvhwm (self, 0); #endif } // -------------------------------------------------------------------------- // Send a signal over a socket. A signal is a short message carrying a // success/failure code (by convention, 0 means OK). Signals are encoded // to be distinguishable from "normal" messages. Accepts a zsock_t or a // zactor_t argument, and returns 0 if successful, -1 if the signal could // not be sent. Takes a polymorphic socket reference. int zsock_signal (void *self, byte status) { assert (self); zmsg_t *msg = zmsg_new_signal (status); if (zmsg_send (&msg, self) < 0) { zmsg_destroy (&msg); return -1; } else return 0; } // -------------------------------------------------------------------------- // Wait on a signal. Use this to coordinate between threads, over pipe // pairs. Blocks until the signal is received. Returns -1 on error, 0 or // greater on success. Accepts a zsock_t or a zactor_t as argument. // Takes a polymorphic socket reference. int zsock_wait (void *self) { assert (self); // Loop and discard messages until we get a signal value or interrupt. while (true) { zmsg_t *msg = zmsg_recv (self); if (!msg) return -1; int rc = zmsg_signal (msg); zmsg_destroy (&msg); if (rc >= 0) return rc; } return -1; } // -------------------------------------------------------------------------- // If there is a partial message still waiting on the socket, remove and // discard it. This is useful when reading partial messages, to get specific // message types. void zsock_flush (void *self) { if (zsock_rcvmore (self)) { zmsg_t *msg = zmsg_recv (self); zmsg_destroy (&msg); } } // -------------------------------------------------------------------------- // Join a group for the RADIO-DISH pattern. Call only on ZMQ_DISH. // Returns 0 if OK, -1 if failed. int zsock_join (void *self, const char *group) { assert (self); #ifdef ZMQ_DISH return zmq_join (zsock_resolve (self), group); #else errno = ENOTSUP; return -1; #endif } // -------------------------------------------------------------------------- // Leave a group for the RADIO-DISH pattern. Call only on ZMQ_DISH. // Returns 0 if OK, -1 if failed. int zsock_leave (void *self, const char *group) { assert (self); #ifdef ZMQ_DISH return zmq_leave (zsock_resolve (self), group); #else errno = ENOTSUP; return -1; #endif } // -------------------------------------------------------------------------- // Probe the supplied object, and report if it looks like a zsock_t. // Takes a polymorphic socket reference. bool zsock_is (void *self) { assert (self); return ((zsock_t *) self)->tag == ZSOCK_TAG; } // -------------------------------------------------------------------------- // Probe the supplied 'self' pointer. Takes a polymorphic socket reference. // If self is a zactor_t, zsock_t, or libzmq socket handle, returns the // libzmq socket handle. If self is a valid file descriptor, returns NULL. // Else returns self as-is. void * zsock_resolve (void *self) { assert (self); if (zactor_is (self)) return zactor_resolve (self); if (zsock_is (self)) return ((zsock_t *) self)->handle; // Check if we have a valid ZMQ socket by probing the socket type int type; size_t option_len = sizeof (int); if (zmq_getsockopt (self, ZMQ_TYPE, &type, &option_len) == 0) return self; // Check if self is a valid FD or socket FD // TODO: this code should move to zsys_isfd () as we don't like // non-portable code outside of that class. int sock_type = -1; #if defined (__WINDOWS__) int sock_type_size = sizeof (int); int rc = getsockopt (*(SOCKET *) self, SOL_SOCKET, SO_TYPE, (char *) &sock_type, &sock_type_size); if (rc == 0) return NULL; // It's a socket descriptor #else socklen_t sock_type_size = sizeof (socklen_t); int rc = getsockopt (*(SOCKET *) self, SOL_SOCKET, SO_TYPE, (char *) &sock_type, &sock_type_size); if (rc == 0 || (rc == -1 && errno == ENOTSOCK)) return NULL; // It's a socket FD or FD #endif // Socket appears to be something else, return it as-is return self; } // -------------------------------------------------------------------------- // Check whether the socket has available message to read. bool zsock_has_in (void *self) { int events = zsock_events (self); return (events & ZMQ_POLLIN) != 0; } // We use the gossip messages for some test cases #include "zgossip_msg.h" // -------------------------------------------------------------------------- // Selftest void zsock_test (bool verbose) { printf (" * zsock: "); if (verbose) printf ("\n"); // @selftest zsock_t *writer = zsock_new (ZMQ_PUSH); assert (writer); int port = zsock_bind (writer, "tcp://127.0.0.1:*"); assert (port != -1); assert (zsock_resolve (writer) != writer); assert (streq (zsock_type_str (writer), "PUSH")); int rc; #if (ZMQ_VERSION >= ZMQ_MAKE_VERSION (3, 2, 0)) // Check unbind rc = zsock_unbind (writer, "tcp://127.0.0.1:%d", port); assert (rc == 0); // In some cases and especially when running under Valgrind, doing // a bind immediately after an unbind causes an EADDRINUSE error. // Even a short sleep allows the OS to release the port for reuse. zclock_sleep (100); // Bind again rc = zsock_bind (writer, "tcp://127.0.0.1:%d", port); assert (rc == port); char endpoint [40]; sprintf (endpoint, "tcp://127.0.0.1:%d", port); assert (streq (zsock_endpoint (writer), endpoint)); #endif zsock_t *reader = zsock_new (ZMQ_PULL); assert (reader); rc = zsock_connect (reader, "tcp://127.0.0.1:%d", port); assert (rc != -1); assert (zsock_resolve (reader) != reader); assert (streq (zsock_type_str (reader), "PULL")); // Basic Hello, World zstr_send (writer, "Hello, World"); zmsg_t *msg = zmsg_recv (reader); assert (msg); char *string = zmsg_popstr (msg); assert (streq (string, "Hello, World")); freen (string); zmsg_destroy (&msg); // Test resolve libzmq socket #if (ZMQ_VERSION >= ZMQ_MAKE_VERSION (3, 2, 0)) void *zmq_ctx = zmq_ctx_new (); #else void *zmq_ctx = zmq_ctx_new (1); #endif assert (zmq_ctx); void *zmq_sock = zmq_socket (zmq_ctx, ZMQ_PUB); assert (zmq_sock); assert (zsock_resolve (zmq_sock) == zmq_sock); zmq_close (zmq_sock); zmq_ctx_term (zmq_ctx); // Test resolve zsock zsock_t *resolve = zsock_new_pub("@tcp://127.0.0.1:*"); assert (resolve); assert (zsock_resolve (resolve) == resolve->handle); zsock_destroy (&resolve); // Test resolve FD SOCKET fd = zsock_fd (reader); assert (zsock_resolve ((void *) &fd) == NULL); // Test binding to ephemeral ports, sequential and random port = zsock_bind (writer, "tcp://127.0.0.1:*"); assert (port >= DYNAMIC_FIRST && port <= DYNAMIC_LAST); port = zsock_bind (writer, "tcp://127.0.0.1:*[50000-]"); assert (port >= 50000 && port <= DYNAMIC_LAST); port = zsock_bind (writer, "tcp://127.0.0.1:*[-50001]"); assert (port >= DYNAMIC_FIRST && port <= 50001); port = zsock_bind (writer, "tcp://127.0.0.1:*[60000-60500]"); assert (port >= 60000 && port <= 60500); port = zsock_bind (writer, "tcp://127.0.0.1:!"); assert (port >= DYNAMIC_FIRST && port <= DYNAMIC_LAST); port = zsock_bind (writer, "tcp://127.0.0.1:![50000-]"); assert (port >= 50000 && port <= DYNAMIC_LAST); port = zsock_bind (writer, "tcp://127.0.0.1:![-50001]"); assert (port >= DYNAMIC_FIRST && port <= 50001); port = zsock_bind (writer, "tcp://127.0.0.1:![60000-60500]"); assert (port >= 60000 && port <= 60500); // Test zsock_attach method zsock_t *dealer = zsock_new (ZMQ_DEALER); assert (dealer); rc = zsock_attach (dealer, "@inproc://myendpoint,tcp://127.0.0.1:*,inproc://others", true); assert (rc == 0); rc = zsock_attach (dealer, "", false); assert (rc == 0); rc = zsock_attach (dealer, NULL, true); assert (rc == 0); rc = zsock_attach (dealer, ">a,@b, c,, ", false); assert (rc == -1); zsock_destroy (&dealer); // Test zsock_endpoint method rc = zsock_bind (writer, "inproc://test.%s", "writer"); assert (rc == 0); assert (streq (zsock_endpoint (writer), "inproc://test.writer")); // Test error state when connecting to an invalid socket type // ('txp://' instead of 'tcp://', typo intentional) rc = zsock_connect (reader, "txp://127.0.0.1:5560"); assert (rc == -1); // Test signal/wait methods rc = zsock_signal (writer, 123); assert (rc == 0); rc = zsock_wait (reader); assert (rc == 123); // Test zsock_send/recv pictures uint8_t number1 = 123; uint16_t number2 = 123 * 123; uint32_t number4 = 123 * 123; number4 *= 123; uint32_t number4_MAX = UINT32_MAX; uint64_t number8 = 123 * 123; number8 *= 123; number8 *= 123; uint64_t number8_MAX = UINT64_MAX; zchunk_t *chunk = zchunk_new ("HELLO", 5); assert (chunk); zframe_t *frame = zframe_new ("WORLD", 5); assert (frame); zhashx_t *hash = zhashx_new (); assert (hash); #ifdef CZMQ_BUILD_DRAFT_API zlistx_t *list = zlistx_new (); assert (list); #endif zuuid_t *uuid = zuuid_new (); assert (uuid); zhashx_set_destructor (hash, (zhashx_destructor_fn *) zstr_free); zhashx_set_duplicator (hash, (zhashx_duplicator_fn *) strdup); zhashx_insert (hash, "1", "value A"); zhashx_insert (hash, "2", "value B"); #ifdef CZMQ_BUILD_DRAFT_API zlistx_set_destructor (list, (zlistx_destructor_fn *) zstr_free); zlistx_set_duplicator (list, (zlistx_duplicator_fn *) strdup); zlistx_add_end (list, "1"); zlistx_add_end (list, "2"); #endif char *original = "pointer"; // Test zsock_recv into each supported type #ifdef CZMQ_BUILD_DRAFT_API zsock_send (writer, "i124488zsbcfUhlp", #else zsock_send (writer, "i124488zsbcfUhp", #endif -12345, number1, number2, number4, number4_MAX, number8, number8_MAX, "This is a string", "ABCDE", 5, #ifdef CZMQ_BUILD_DRAFT_API chunk, frame, uuid, hash, list, original); #else chunk, frame, uuid, hash, original); #endif char *uuid_str = strdup (zuuid_str (uuid)); zchunk_destroy (&chunk); zframe_destroy (&frame); zuuid_destroy (&uuid); zhashx_destroy (&hash); #ifdef CZMQ_BUILD_DRAFT_API zlistx_destroy (&list); #endif int integer; byte *data; size_t size; char *pointer; number8_MAX = number8 = number4_MAX = number4 = number2 = number1 = 0ULL; #ifdef CZMQ_BUILD_DRAFT_API rc = zsock_recv (reader, "i124488zsbcfUhlp", #else rc = zsock_recv (reader, "i124488zsbcfUhp", #endif &integer, &number1, &number2, &number4, &number4_MAX, &number8, &number8_MAX, &string, &data, &size, &chunk, #ifdef CZMQ_BUILD_DRAFT_API &frame, &uuid, &hash, &list, &pointer); #else &frame, &uuid, &hash, &pointer); #endif assert (rc == 0); assert (integer == -12345); assert (number1 == 123); assert (number2 == 123 * 123); assert (number4 == 123 * 123 * 123); assert (number4_MAX == UINT32_MAX); assert (number8 == 123 * 123 * 123 * 123); assert (number8_MAX == UINT64_MAX); assert (streq (string, "This is a string")); assert (memcmp (data, "ABCDE", 5) == 0); assert (size == 5); assert (memcmp (zchunk_data (chunk), "HELLO", 5) == 0); assert (zchunk_size (chunk) == 5); assert (streq (uuid_str, zuuid_str (uuid))); assert (memcmp (zframe_data (frame), "WORLD", 5) == 0); assert (zframe_size (frame) == 5); char *value = (char *) zhashx_lookup (hash, "1"); assert (streq (value, "value A")); value = (char *) zhashx_lookup (hash, "2"); assert (streq (value, "value
{'commit': 'diff --git a/src/zsock.c b/src/zsock.c\nindex b0f22242..03e9a07a 100644\n--- a/src/zsock.c\n+++ b/src/zsock.c\n@@ -352,12 +352,14 @@ zsock_unbind (zsock_t *self, const char *format, ...)\n     assert (zsock_is (self));\n \n #if (ZMQ_VERSION >= ZMQ_MAKE_VERSION (3,2,0))\n-    char endpoint [256];\n+    //  Expand format to get full endpoint\n     va_list argptr;\n     va_start (argptr, format);\n-    vsnprintf (endpoint, 256, format, argptr);\n+    char *endpoint = zsys_vprintf (format, argptr);\n     va_end (argptr);\n-    return zmq_unbind (self->handle, endpoint);\n+    int rc = zmq_unbind (self->handle, endpoint);\n+    free (endpoint);\n+    return rc;\n #else\n     return -1;\n #endif\n@@ -374,13 +376,13 @@ zsock_connect (zsock_t *self, const char *format, ...)\n     assert (self);\n     assert (zsock_is (self));\n \n-    char endpoint [256];\n+    //  Expand format to get full endpoint\n     va_list argptr;\n     va_start (argptr, format);\n-    vsnprintf (endpoint, 256, format, argptr);\n+    char *endpoint = zsys_vprintf (format, argptr);\n     va_end (argptr);\n-    \n     int rc = zmq_connect (self->handle, endpoint);\n+    \n #if (ZMQ_VERSION < ZMQ_MAKE_VERSION (4,0,0))\n     int retries = 4;\n     while (rc == -1 && zmq_errno () == ECONNREFUSED && retries) {\n@@ -392,6 +394,7 @@ zsock_connect (zsock_t *self, const char *format, ...)\n         retries--;\n     }\n #endif\n+    free (endpoint);\n     return rc;\n }\n \n@@ -407,12 +410,14 @@ zsock_disconnect (zsock_t *self, const char *format, ...)\n     assert (zsock_is (self));\n \n #if (ZMQ_VERSION >= ZMQ_MAKE_VERSION (3,2,0))\n-    char endpoint [256];\n+    //  Expand format to get full endpoint\n     va_list argptr;\n     va_start (argptr, format);\n-    vsnprintf (endpoint, 256, format, argptr);\n+    char *endpoint = zsys_vprintf (format, argptr);\n     va_end (argptr);\n-    return zmq_disconnect (self->handle, endpoint);\n+    int rc = zmq_disconnect (self->handle, endpoint);\n+    free (endpoint);\n+    return rc;\n #else\n     return -1;\n #endif\n', 'commit_subject': 'Problem: vsnprintf use for endpoints is clumsy', 'repo_name': 'zeromq/czmq', 'committer': {'name': 'Pieter Hintjens', 'email': '8df74dda184c19df19d4755602108540e071bf38@imatix.com', 'time_sec': '1409088123', 'tz_offset': '120', 'date': {'seconds': '1409088123'}}, 'commit_tree': '76ad98bd8875b1c67b7fd0ca700a4cb84db91d41', 'commit_message': 'Problem: vsnprintf use for endpoints is clumsy\n\nSolution: use zsys_sprintf and remove size restriction on endpoints.\n\nNote that _attach still uses fixed 256 limit; this should change at\nsome point to use zrex and no fixed size limit on endpoints.\n', 'language_name': 'Perl', 'license': 'mpl-2.0', 'hunks': '@@ -352,12 +352,14 @@ zsock_unbind (zsock_t *self, const char *format, ...)\n     assert (zsock_is (self));\n \n #if (ZMQ_VERSION >= ZMQ_MAKE_VERSION (3,2,0))\n-    char endpoint [256];\n+    //  Expand format to get full endpoint\n     va_list argptr;\n     va_start (argptr, format);\n-    vsnprintf (endpoint, 256, format, argptr);\n+    char *endpoint = zsys_vprintf (format, argptr);\n     va_end (argptr);\n-    return zmq_unbind (self->handle, endpoint);\n+    int rc = zmq_unbind (self->handle, endpoint);\n+    free (endpoint);\n+    return rc;\n #else\n     return -1;\n #endif\n@@ -374,13 +376,13 @@ zsock_connect (zsock_t *self, const char *format, ...)\n     assert (self);\n     assert (zsock_is (self));\n \n-    char endpoint [256];\n+    //  Expand format to get full endpoint\n     va_list argptr;\n     va_start (argptr, format);\n-    vsnprintf (endpoint, 256, format, argptr);\n+    char *endpoint = zsys_vprintf (format, argptr);\n     va_end (argptr);\n-    \n     int rc = zmq_connect (self->handle, endpoint);\n+    \n #if (ZMQ_VERSION < ZMQ_MAKE_VERSION (4,0,0))\n     int retries = 4;\n     while (rc == -1 && zmq_errno () == ECONNREFUSED && retries) {\n@@ -392,6 +394,7 @@ zsock_connect (zsock_t *self, const char *format, ...)\n         retries--;\n     }\n #endif\n+    free (endpoint);\n     return rc;\n }\n \n@@ -407,12 +410,14 @@ zsock_disconnect (zsock_t *self, const char *format, ...)\n     assert (zsock_is (self));\n \n #if (ZMQ_VERSION >= ZMQ_MAKE_VERSION (3,2,0))\n-    char endpoint [256];\n+    //  Expand format to get full endpoint\n     va_list argptr;\n     va_start (argptr, format);\n-    vsnprintf (endpoint, 256, format, argptr);\n+    char *endpoint = zsys_vprintf (format, argptr);\n     va_end (argptr);\n-    return zmq_disconnect (self->handle, endpoint);\n+    int rc = zmq_disconnect (self->handle, endpoint);\n+    free (endpoint);\n+    return rc;\n #else\n     return -1;\n #endif\n', 'addition_count': 14, 'deletion_count': 9, 'src_file': 'a/src/zsock.c', 'tgt_file': 'b/src/zsock.c', 'file_extension': '.c', 'before_file': '/*  =========================================================================\n    zsock - high-level socket API that hides libzmq contexts and sockets\n\n    Copyright (c) the Contributors as noted in the AUTHORS file.\n    This file is part of CZMQ, the high-level C binding for 0MQ:\n    http://czmq.zeromq.org.\n\n    This Source Code Form is subject to the terms of the Mozilla Public\n    License, v. 2.0. If a copy of the MPL was not distributed with this\n    file, You can obtain one at http://mozilla.org/MPL/2.0/.\n    =========================================================================\n*/\n\n/*\n@header\n    The zsock class wraps the libzmq socket handle (a void *) with a proper\n    structure that follows the CLASS rules for construction and destruction.\n    Some zsock methods take a void * "polymorphic" reference, which can be\n    either a zsock_t or a zactor_t reference, or a libzmq void *.\n@discuss\n@end\n*/\n\n#define ZSOCK_NOCHECK // we are defining the methods here, so don\'t redirect symbols.\n\n#include "czmq_classes.h"\n#include "zsock_option.inc"\n\n//  zsock_t instances always have this tag as the first 4 octets of\n//  their data, which lets us do runtime object typing & validation.\n#define ZSOCK_TAG           0xcafe0004\n\n//  This port range is defined by IANA for dynamic or private ports\n//  We use this when choosing a port for dynamic binding.\n#define DYNAMIC_FIRST       0xc000    // 49152\n#define DYNAMIC_LAST        0xffff    // 65535\n\n#define ZSOCK_BSEND_MAX_FRAMES 32  // Arbitrary limit, for now\n\n//  Structure of our class\n\nstruct _zsock_t {\n    uint32_t tag;               //  Object tag for runtime detection\n    void *handle;               //  The libzmq socket handle\n    char *endpoint;             //  Last bound endpoint, if any\n    char *cache;                //  Holds last zsock_brecv strings\n    int type;                   //  Socket type\n    size_t cache_size;          //  Current size of cache\n    uint32_t routing_id;        //  Routing ID for server sockets\n};\n\n#ifndef CZMQ_BUILD_DRAFT_API\nCZMQ_PRIVATE zsock_t *\n    zsock_new_server_checked (const char *endpoint, const char *filename, size_t line_nbr);\n\nCZMQ_PRIVATE zsock_t *\n    zsock_new_client_checked (const char *endpoint, const char *filename, size_t line_nbr);\n\nCZMQ_PRIVATE zsock_t *\n    zsock_new_radio_checked (const char *endpoint, const char *filename, size_t line_nbr);\n\nCZMQ_PRIVATE zsock_t *\n    zsock_new_dish_checked (const char *endpoint, const char *filename, size_t line_nbr);\n\nCZMQ_PRIVATE zsock_t *\n    zsock_new_gather_checked (const char *endpoint, const char *filename, size_t line_nbr);\n\nCZMQ_PRIVATE zsock_t *\n    zsock_new_scatter_checked (const char *endpoint, const char *filename, size_t line_nbr);\n#endif // CZMQ_BUILD_DRAFT_API\n\n\n//  --------------------------------------------------------------------------\n//  Create a new socket. This macro passes the caller source and line\n//  number so that CZMQ can report socket leaks intelligently. To switch\n//  off this checking, which may be costly if you use a LOT of sockets,\n//  define ZSOCK_NOCHECK before compiling your code. Returns the new\n//  socket, or NULL if the new socket could not be created.\n\nzsock_t *\nzsock_new_checked (int type, const char *filename, size_t line_nbr)\n{\n    zsock_t *self = (zsock_t *) zmalloc (sizeof (zsock_t));\n    assert (self);\n    self->tag = ZSOCK_TAG;\n    self->type = type;\n    self->handle = zsys_socket (type, filename, line_nbr);\n    assert (self->handle);\n    return self;\n}\n\nzsock_t *\nzsock_new (int type)\n{\n    return zsock_new_checked (type, NULL, 0);\n}\n\n\n//  --------------------------------------------------------------------------\n//  Destroy the socket. You must use this for any socket created via the\n//  zsock_new method.\n\nvoid\nzsock_destroy_checked (zsock_t **self_p, const char *filename, size_t line_nbr)\n{\n    assert (self_p);\n    if (*self_p) {\n        zsock_t *self = *self_p;\n        assert (zsock_is (self));\n        self->tag = 0xDeadBeef;\n        int rc = zsys_close (self->handle, filename, line_nbr);\n        assert (rc == 0);\n        freen (self->endpoint);\n        freen (self->cache);\n        freen (self);\n        *self_p = NULL;\n    }\n}\n\nvoid zsock_destroy (zsock_t **self_p)\n{\n    zsock_destroy_checked (self_p, NULL, 0);\n}\n\n\n//  --------------------------------------------------------------------------\n//  This interface includes some smart constructors, which create sockets with\n//  additional set-up. In all of these, the endpoint is NULL, or starts with\n//  \'@\' (bind) or \'>\' (connect). Multiple endpoints are allowed, separated by\n//  commas. If endpoint does not start with \'@\' or \'>\', default action depends\n//  on socket type.\n\n//  Create a PUB socket. Default action is bind.\n\nzsock_t *\nzsock_new_pub_checked (const char *endpoints, const char *filename, size_t line_nbr)\n{\n    zsock_t *sock = zsock_new_checked (ZMQ_PUB, filename, line_nbr);\n    if (zsock_attach (sock, endpoints, true))\n        zsock_destroy (&sock);\n    return sock;\n}\n\nzsock_t *\nzsock_new_pub (const char *endpoints)\n{\n    return zsock_new_pub_checked (endpoints, NULL, 0);\n}\n\n\n//  --------------------------------------------------------------------------\n//  Create a SUB socket, and optionally subscribe to some prefix string. Default\n//  action is connect.\n\nzsock_t *\nzsock_new_sub_checked (const char *endpoints, const char *subscribe, const char *filename, size_t line_nbr)\n{\n    zsock_t *sock = zsock_new_checked (ZMQ_SUB, filename, line_nbr);\n    if (zsock_attach (sock, endpoints, false) == 0) {\n        if (subscribe)\n            zsock_set_subscribe (sock, subscribe);\n    }\n    else\n        zsock_destroy (&sock);\n    return sock;\n}\n\nzsock_t *\nzsock_new_sub (const char *endpoints, const char *subscribe)\n{\n    return zsock_new_sub_checked (endpoints, subscribe, NULL, 0);\n}\n\n\n//  --------------------------------------------------------------------------\n//  Create a REQ socket. Default action is connect.\n\nzsock_t *\nzsock_new_req_checked (const char *endpoints, const char *filename, size_t line_nbr)\n{\n    zsock_t *sock = zsock_new_checked (ZMQ_REQ, filename, line_nbr);\n    if (zsock_attach (sock, endpoints, false))\n        zsock_destroy (&sock);\n    return sock;\n}\n\nzsock_t *\nzsock_new_req (const char *endpoints)\n{\n    return zsock_new_req_checked (endpoints, NULL, 0);\n}\n\n\n//  --------------------------------------------------------------------------\n//  Create a REP socket. Default action is bind.\n\nzsock_t *\nzsock_new_rep_checked (const char *endpoints, const char *filename, size_t line_nbr)\n{\n    zsock_t *sock = zsock_new_checked (ZMQ_REP, filename, line_nbr);\n    if (zsock_attach (sock, endpoints, true))\n        zsock_destroy (&sock);\n    return sock;\n}\n\nzsock_t *\nzsock_new_rep (const char *endpoints)\n{\n    return zsock_new_rep_checked (endpoints, NULL, 0);\n}\n\n\n//  --------------------------------------------------------------------------\n//  Create a DEALER socket. Default action is connect.\n\nzsock_t *\nzsock_new_dealer_checked (const char *endpoints, const char *filename, size_t line_nbr)\n{\n    zsock_t *sock = zsock_new_checked (ZMQ_DEALER, filename, line_nbr);\n    if (zsock_attach (sock, endpoints, false))\n        zsock_destroy (&sock);\n    return sock;\n}\n\nzsock_t *\nzsock_new_dealer (const char *endpoints)\n{\n    return zsock_new_dealer_checked (endpoints, NULL, 0);\n}\n\n\n//  --------------------------------------------------------------------------\n//  Create a ROUTER socket. Default action is bind.\n\nzsock_t *\nzsock_new_router_checked (const char *endpoints, const char *filename, size_t line_nbr)\n{\n    zsock_t *sock = zsock_new_checked (ZMQ_ROUTER, filename, line_nbr);\n    if (zsock_attach (sock, endpoints, true))\n        zsock_destroy (&sock);\n    return sock;\n}\n\nzsock_t *\nzsock_new_router (const char *endpoints)\n{\n    return zsock_new_router_checked (endpoints, NULL, 0);\n}\n\n\n//  --------------------------------------------------------------------------\n//  Create a PUSH socket. Default action is connect.\n\nzsock_t *\nzsock_new_push_checked (const char *endpoints, const char *filename, size_t line_nbr)\n{\n    zsock_t *sock = zsock_new_checked (ZMQ_PUSH, filename, line_nbr);\n    if (zsock_attach (sock, endpoints, false))\n        zsock_destroy (&sock);\n    return sock;\n}\n\nzsock_t *\nzsock_new_push (const char *endpoints)\n{\n    return zsock_new_push_checked(endpoints, NULL, 0);\n}\n\n\n//  --------------------------------------------------------------------------\n//  Create a PULL socket. Default action is bind.\n\nzsock_t *\nzsock_new_pull_checked (const char *endpoints, const char *filename, size_t line_nbr)\n{\n    zsock_t *sock = zsock_new_checked (ZMQ_PULL, filename, line_nbr);\n    if (zsock_attach (sock, endpoints, true))\n        zsock_destroy (&sock);\n    return sock;\n}\n\nzsock_t *\nzsock_new_pull (const char *endpoints)\n{\n    return zsock_new_pull_checked (endpoints, NULL, 0);\n}\n\n\n//  --------------------------------------------------------------------------\n//  Create an XPUB socket. Default action is bind.\n\nzsock_t *\nzsock_new_xpub_checked (const char *endpoints, const char *filename, size_t line_nbr)\n{\n#if defined ZMQ_XPUB\n    zsock_t *sock = zsock_new_checked (ZMQ_XPUB, filename, line_nbr);\n    if (zsock_attach (sock, endpoints, true))\n        zsock_destroy (&sock);\n    return sock;\n#else\n    // errno = ENOTSUP;      // too late, stable API would be broken\n    return NULL;            //  Not implemented\n#endif\n}\n\nzsock_t *\nzsock_new_xpub (const char *endpoints)\n{\n    return zsock_new_xpub_checked (endpoints, NULL, 0);\n}\n\n\n//  --------------------------------------------------------------------------\n//  Create an XSUB socket. Default action is connect.\n\nzsock_t *\nzsock_new_xsub_checked (const char *endpoints, const char *filename, size_t line_nbr)\n{\n#if defined ZMQ_XSUB\n    zsock_t *sock = zsock_new_checked (ZMQ_XSUB, filename, line_nbr);\n    if (zsock_attach (sock, endpoints, false))\n        zsock_destroy (&sock);\n    return sock;\n#else\n    // errno = ENOTSUP;      // too late, stable API would be broken\n    return NULL;            //  Not implemented\n#endif\n}\n\nzsock_t *\nzsock_new_xsub (const char *endpoints)\n{\n    return zsock_new_xsub_checked (endpoints, NULL, 0);\n}\n\n\n//  --------------------------------------------------------------------------\n//  Create a PAIR socket. Default action is connect.\n\nzsock_t *\nzsock_new_pair_checked (const char *endpoints, const char *filename, size_t line_nbr)\n{\n    zsock_t *sock = zsock_new_checked (ZMQ_PAIR, filename, line_nbr);\n    if (zsock_attach (sock, endpoints, false))\n        zsock_destroy (&sock);\n    return sock;\n}\n\nzsock_t *\nzsock_new_pair (const char *endpoints)\n{\n    assert (zsock_is (self));\n\n#if (ZMQ_VERSION >= ZMQ_MAKE_VERSION (3,2,0))\n    char endpoint [256];\n    va_list argptr;\n    va_start (argptr, format);\n    vsnprintf (endpoint, 256, format, argptr);\n    va_end (argptr);\n    return zmq_unbind (self->handle, endpoint);\n#else\n    return -1;\n#endif\n    return sock;\n#else\n    // errno = ENOTSUP;      // too late, stable API would be broken\n    return NULL;            //  Not implemented\n#endif\n}\n\nzsock_t *\nzsock_new_stream (const char *endpoints)\n{\n    assert (self);\n    assert (zsock_is (self));\n\n    char endpoint [256];\n    va_list argptr;\n    va_start (argptr, format);\n    vsnprintf (endpoint, 256, format, argptr);\n    va_end (argptr);\n    \n    int rc = zmq_connect (self->handle, endpoint);\n#if (ZMQ_VERSION < ZMQ_MAKE_VERSION (4,0,0))\n    int retries = 4;\n    while (rc == -1 && zmq_errno () == ECONNREFUSED && retries) {\n    return sock;\n#else\n    errno = ENOTSUP;\n    return NULL;\n#endif\n        retries--;\n    }\n#endif\n    return rc;\n}\n\n\n//  --------------------------------------------------------------------------\n//  Create a CLIENT socket. Default action is connect.\n\nzsock_t *\nzsock_new_client_checked (const char *endpoints, const char *filename, size_t line_nbr)\n{\n#if defined ZMQ_CLIENT\n    zsock_t *sock = zsock_new_checked (ZMQ_CLIENT, filename, line_nbr);\n    assert (zsock_is (self));\n\n#if (ZMQ_VERSION >= ZMQ_MAKE_VERSION (3,2,0))\n    char endpoint [256];\n    va_list argptr;\n    va_start (argptr, format);\n    vsnprintf (endpoint, 256, format, argptr);\n    va_end (argptr);\n    return zmq_disconnect (self->handle, endpoint);\n#else\n    return -1;\n#endif\n\n\n//  --------------------------------------------------------------------------\n//  Create a RADIO socket. Default action is bind.\n\nzsock_t *\nzsock_new_radio_checked (const char *endpoints, const char *filename, size_t line_nbr)\n{\n#if defined ZMQ_RADIO\n    zsock_t *sock = zsock_new_checked (ZMQ_RADIO, filename, line_nbr);\n    if (zsock_attach (sock, endpoints, true))\n        zsock_destroy (&sock);\n    return sock;\n#else\n    errno = ENOTSUP;\n    return NULL;\n#endif\n}\n\nzsock_t *\nzsock_new_radio (const char *endpoints)\n{\n    return zsock_new_radio_checked (endpoints, NULL, 0);\n}\n\n\n//  --------------------------------------------------------------------------\n//  Create a DISH socket. Default action is connect.\n\nzsock_t *\nzsock_new_dish_checked (const char *endpoints, const char *filename, size_t line_nbr)\n{\n#if defined ZMQ_DISH\n    zsock_t *sock = zsock_new_checked (ZMQ_DISH, filename, line_nbr);\n    if (zsock_attach (sock, endpoints, false))\n        zsock_destroy (&sock);\n    return sock;\n#else\n    errno = ENOTSUP;\n    return NULL;\n#endif\n}\n\nzsock_t *\nzsock_new_dish (const char *endpoints)\n{\n    return zsock_new_dish_checked (endpoints, NULL, 0);\n}\n\n\n//  --------------------------------------------------------------------------\n//  Create a GATHER socket. Default action is bind.\n\nzsock_t *\nzsock_new_gather_checked (const char *endpoints, const char *filename, size_t line_nbr)\n{\n#if defined ZMQ_GATHER\n    zsock_t *sock = zsock_new_checked (ZMQ_GATHER, filename, line_nbr);\n    if (zsock_attach (sock, endpoints, true))\n        zsock_destroy (&sock);\n    return sock;\n#else\n    errno = ENOTSUP;\n    return NULL;\n#endif\n}\n\nzsock_t *\nzsock_new_gather (const char *endpoints)\n{\n    return zsock_new_gather_checked (endpoints, NULL, 0);\n}\n\n\n//  --------------------------------------------------------------------------\n//  Create a scatter socket. Default action is connect.\n\nzsock_t *\nzsock_new_scatter_checked (const char *endpoints, const char *filename, size_t line_nbr)\n{\n#if defined ZMQ_SCATTER\n    zsock_t *sock = zsock_new_checked (ZMQ_SCATTER, filename, line_nbr);\n    if (zsock_attach (sock, endpoints, false))\n        zsock_destroy (&sock);\n    return sock;\n#else\n    errno = ENOTSUP;\n    return NULL;\n#endif\n}\n\nzsock_t *\nzsock_new_scatter (const char *endpoints)\n{\n    return zsock_new_scatter_checked (endpoints, NULL, 0);\n}\n\n//  --------------------------------------------------------------------------\n//  Create a dgram socket. Default action is bind.\n\nzsock_t *\nzsock_new_dgram_checked (const char *endpoints, const char *filename, size_t line_nbr)\n{\n#if defined ZMQ_DGRAM\n    zsock_t *sock = zsock_new_checked (ZMQ_DGRAM, filename, line_nbr);\n    if (zsock_attach (sock, endpoints, true))\n        zsock_destroy (&sock);\n    return sock;\n#else\n    errno = ENOTSUP;\n    return NULL;\n#endif\n}\n\n//  *** Draft method, for development use, may change without warning ***\n//  Create a DGRAM socket. Default action is bind.\nzsock_t *\nzsock_new_dgram (const char *endpoint)\n{\n    return zsock_new_dgram_checked ( endpoint, NULL, 0 );\n}\n\n\n//  --------------------------------------------------------------------------\n//  Bind a socket to a formatted endpoint. For tcp:// endpoints, supports\n//  ephemeral ports, if you specify the port number as "*". By default\n//  zsock uses the IANA designated range from C000 (49152) to FFFF (65535).\n//  To override this range, follow the "*" with "[first-last]". Either or\n//  both first and last may be empty. To bind to a random port within the\n//  range, use "!" in place of "*".\n//\n//  Examples:\n//  tcp://127.0.0.1:*                bind to first free port from C000 up\n//  tcp://127.0.0.1:!                bind to random port from C000 to FFFF\n//  tcp://127.0.0.1:*[60000-]        bind to first free port from 60000 up\n//  tcp://127.0.0.1:![-60000]        bind to random port from C000 to 60000\n//  tcp://127.0.0.1:![55000-55999]   bind to random port from 55000-55999\n//\n//  On success, returns the actual port number used, for tcp:// endpoints,\n//  and 0 for other transports. On failure, returns -1. Note that when using\n//  ephemeral ports, a port may be reused by different services without\n//  clients being aware. Protocols that run on ephemeral ports should take\n//  this into account.\n\nint\nzsock_bind (zsock_t *self, const char *format, ...)\n{\n    assert (self);\n    assert (zsock_is (self));\n\n    //  Expand format to get full endpoint\n    va_list argptr;\n    va_start (argptr, format);\n    char *endpoint = zsys_vprintf (format, argptr);\n    va_end (argptr);\n    if (!endpoint)\n        return -1;\n    int rc;\n\n    //  If tcp:// endpoint, parse to get or make port number\n    zrex_t *rex = zrex_new (NULL);\n    if (zrex_eq (rex, endpoint, "^tcp://.*:(\\\\d+)$")) {\n        assert (zrex_hits (rex) == 2);\n        int port = atoi (zrex_hit (rex, 1));\n#if defined (HAVE_LIBSYSTEMD) && (ZMQ_VERSION >= ZMQ_MAKE_VERSION (4, 2, 0))\n        if (zsys_auto_use_fd ()) {\n            int last_handle = SD_LISTEN_FDS_START + sd_listen_fds (0);\n            int handle;\n            for (handle = SD_LISTEN_FDS_START; handle < last_handle; ++handle)\n                if (sd_is_socket_inet (handle, AF_UNSPEC, SOCK_STREAM, 1,\n                        (uint16_t) port) > 0) {\n                    zsock_set_use_fd (self, handle);\n                    break;\n                }\n        }\n#endif\n        if (zmq_bind (self->handle, endpoint) == 0)\n            rc = port;\n        else\n            rc = -1;\n    }\n    else\n    if (zrex_eq (rex, endpoint, "^(tcp://.*):([*!])(\\\\[(\\\\d+)?-(\\\\d+)?\\\\])?$")) {\n        assert (zrex_hits (rex) == 6);\n        const char *hostname, *opcode, *group, *first_str, *last_str;\n        zrex_fetch (rex, &hostname, &opcode, &group, &first_str, &last_str, NULL);\n\n        int first = *first_str? atoi (first_str): DYNAMIC_FIRST;\n        int last = *last_str? atoi (last_str): DYNAMIC_LAST;\n\n        //  This is how many times we\'ll try before giving up\n        int attempts = last - first + 1;\n\n        //  If operator is \'*\', take first available port.\n        //  If operator is \'!\', take a random leap into our port space; we\'ll\n        //  still scan sequentially to make sure we find a free slot rapidly.\n        int port = first;\n        if (streq (opcode, "!"))\n            port += randof (attempts);\n\n        rc = -1;                //  Assume we don\'t succeed\n        while (rc == -1 && attempts--) {\n            zstr_free (&endpoint);\n            endpoint = zsys_sprintf ("%s:%d", hostname, port);\n            if (!endpoint)\n                break;\n            if (zmq_bind (self->handle, endpoint) == 0)\n                rc = port;\n            if (++port > last)\n                port = first;\n        }\n    }\n    else {\n#if defined (HAVE_LIBSYSTEMD) && (ZMQ_VERSION >= ZMQ_MAKE_VERSION (4, 2, 0))\n        if (zsys_auto_use_fd () && zrex_eq (rex, endpoint, "^ipc://(.*)$")) {\n            assert (zrex_hits (rex) == 2);\n            const char *sock_path;\n            zrex_fetch (rex, &sock_path, NULL);\n\n            int last_handle = SD_LISTEN_FDS_START + sd_listen_fds (0);\n            int handle;\n            for (handle = SD_LISTEN_FDS_START; handle < last_handle; ++handle)\n                if (sd_is_socket_unix (handle, SOCK_STREAM, 1, sock_path, 0) > 0) {\n                    zsock_set_use_fd (self, handle);\n                    break;\n                }\n        }\n#endif\n        rc = zmq_bind (self->handle, endpoint);\n    }\n\n    //  Store successful endpoint for later reference\n    if (rc >= 0) {\n        zstr_free (&self->endpoint);\n        self->endpoint = endpoint;\n    }\n    else\n        zstr_free (&endpoint);\n\n    zrex_destroy (&rex);\n    return rc;\n}\n\n\n//  --------------------------------------------------------------------------\n//  Returns last bound endpoint, if any.\n\nconst char *\nzsock_endpoint (zsock_t *self)\n{\n    assert (self);\n    return self->endpoint;\n}\n\n\n//  --------------------------------------------------------------------------\n//  Unbind a socket from a formatted endpoint.\n//  Returns 0 if OK, -1 if the endpoint was invalid or the function\n//  isn\'t supported.\n\nint\nzsock_unbind (zsock_t *self, const char *format, ...)\n{\n    assert (self);\n    assert (zsock_is (self));\n\n#if (ZMQ_VERSION >= ZMQ_MAKE_VERSION (3, 2, 0))\n    //  Expand format to get full endpoint\n    va_list argptr;\n    va_start (argptr, format);\n    char *endpoint = zsys_vprintf (format, argptr);\n    va_end (argptr);\n    if (!endpoint)\n        return -1;\n\n    int rc = zmq_unbind (self->handle, endpoint);\n    zstr_free (&endpoint);\n    return rc;\n#else\n    return -1;\n#endif\n}\n\n\n//  --------------------------------------------------------------------------\n//  Connect a socket to a formatted endpoint\n//  Returns 0 if the endpoint is valid, -1 if the connect failed.\n\nint\nzsock_connect (zsock_t *self, const char *format, ...)\n{\n    assert (self);\n    assert (zsock_is (self));\n\n    //  Expand format to get full endpoint\n    va_list argptr;\n    va_start (argptr, format);\n    char *endpoint = zsys_vprintf (format, argptr);\n    va_end (argptr);\n    if (!endpoint)\n        return -1;\n    int rc = zmq_connect (self->handle, endpoint);\n\n#if (ZMQ_VERSION < ZMQ_MAKE_VERSION (4, 0, 0))\n    int retries = 4;\n    while (rc == -1 && zmq_errno () == ECONNREFUSED && retries) {\n        //  This bruteforces a synchronization between connecting and\n        //  binding threads on ZeroMQ v3.2 and earlier, where the bind\n        //  MUST happen before the connect on inproc transports.\n        zclock_sleep (250);\n        rc = zmq_connect (self->handle, endpoint);\n        retries--;\n    }\n#endif\n    zstr_free (&endpoint);\n    return rc;\n}\n\n\n//  --------------------------------------------------------------------------\n//  Disconnect a socket from a formatted endpoint\n//  Returns 0 if disconnection is complete -1 if the disconnection failed.\n\nint\nzsock_disconnect (zsock_t *self, const char *format, ...)\n{\n    assert (self);\n    assert (zsock_is (self));\n\n#if (ZMQ_VERSION >= ZMQ_MAKE_VERSION (3, 2, 0))\n    //  Expand format to get full endpoint\n    va_list argptr;\n    va_start (argptr, format);\n    char *endpoint = zsys_vprintf (format, argptr);\n    va_end (argptr);\n    if (!endpoint)\n        return -1;\n    int rc = zmq_disconnect (self->handle, endpoint);\n    zstr_free (&endpoint);\n    return rc;\n#else\n    return -1;\n#endif\n}\n\n\n//  --------------------------------------------------------------------------\n//  Attach a socket to zero or more endpoints. If endpoints is not null,\n//  parses as list of ZeroMQ endpoints, separated by commas, and prefixed by\n//  \'@\' (to bind the socket) or \'>\' (to attach the socket). Returns 0 if all\n//  endpoints were valid, or -1 if there was a syntax error. If the endpoint\n//  does not start with \'@\' or \'>\', the serverish argument defines whether\n//  it is used to bind (serverish = true) or connect (serverish = false).\n\nint\nzsock_attach (zsock_t *self, const char *endpoints, bool serverish)\n{\n    assert (self);\n    if (!endpoints)\n        return 0;\n\n    //  We hold each individual endpoint here\n    char endpoint [256];\n    while (*endpoints) {\n        const char *delimiter = strchr (endpoints, \',\');\n        if (!delimiter)\n            delimiter = endpoints + strlen (endpoints);\n        if (delimiter - endpoints > 255)\n            return -1;\n        memcpy (endpoint, endpoints, delimiter - endpoints);\n        endpoint [delimiter - endpoints] = 0;\n\n        int rc;\n        if (endpoint [0] == \'@\')\n            rc = zsock_bind (self, "%s", endpoint + 1);\n        else\n        if (endpoint [0] == \'>\')\n            rc = zsock_connect (self, "%s", endpoint + 1);\n        else\n        if (serverish)\n            rc = zsock_bind (self, "%s", endpoint);\n        else\n            rc = zsock_connect (self, "%s", endpoint);\n\n        if (rc == -1)\n            return -1;          //  Bad endpoint syntax\n\n        if (*delimiter == 0)\n            break;\n        endpoints = delimiter + 1;\n    }\n    return 0;\n}\n\n\n//  --------------------------------------------------------------------------\n//  Returns socket type as printable constant string.\n\nconst char *\nzsock_type_str (zsock_t *self)\n{\n    assert (self);\n    return zsys_sockname (self->type);\n}\n\n\n//  --------------------------------------------------------------------------\n//  Send a \'picture\' message to the socket (or actor). The picture is a\n//  string that defines the type of each frame. This makes it easy to send\n//  a complex multiframe message in one call. The picture can contain any\n//  of these characters, each corresponding to one or two arguments:\n//\n//      i = int (signed)\n//      1 = uint8_t\n//      2 = uint16_t\n//      4 = uint32_t\n//      8 = uint64_t\n//      s = char *\n//      b = byte *, int (2 arguments)\n//      c = zchunk_t *\n//      f = zframe_t *\n//      h = zhashx_t *\n//      l = zlistx_t *\n//      U = zuuid_t *\n//      p = void * (sends the pointer value, only meaningful over inproc)\n//      m = zmsg_t * (sends all frames in the zmsg)\n//      z = sends zero-sized frame (0 arguments)\n//      u = uint (deprecated)\n//\n//  Note that s, b, c, and f are encoded the same way and the choice is\n//  offered as a convenience to the sender, which may or may not already\n//  have data in a zchunk or zframe. Does not change or take ownership of\n//  any arguments. Returns 0 if successful, -1 if sending failed for any\n//  reason.\n\nint\nzsock_send (void *self, const char *picture, ...)\n{\n    va_list argptr;\n    va_start (argptr, picture);\n    int rc = zsock_vsend (self, picture, argptr);\n    va_end (argptr);\n    return rc;\n}\n\n\n//  --------------------------------------------------------------------------\n//  Send a \'picture\' message to the socket (or actor). This is a va_list\n//  version of zsock_send (), so please consult its documentation for the\n//  details.\n\nint\nzsock_vsend (void *self, const char *picture, va_list argptr)\n{\n    assert (self);\n    assert (picture);\n\n    zmsg_t *msg = zmsg_new ();\n    while (*picture) {\n        if (*picture == \'i\')\n            zmsg_addstrf (msg, "%d", va_arg (argptr, int));\n        else\n        if (*picture == \'1\')\n            zmsg_addstrf (msg, "%" PRIu8, (uint8_t) va_arg (argptr, int));\n        else\n        if (*picture == \'2\')\n            zmsg_addstrf (msg, "%" PRIu16, (uint16_t) va_arg (argptr, int));\n        else\n        if (*picture == \'4\')\n            zmsg_addstrf (msg, "%" PRIu32, va_arg (argptr, uint32_t));\n        else\n        if (*picture == \'8\')\n            zmsg_addstrf (msg, "%" PRIu64, va_arg (argptr, uint64_t));\n        else\n        if (*picture == \'u\')    //  Deprecated, use 4 or 8 instead\n            zmsg_addstrf (msg, "%ud", va_arg (argptr, uint));\n        else\n        if (*picture == \'s\')\n            zmsg_addstr (msg, va_arg (argptr, char *));\n        else\n        if (*picture == \'b\') {\n            //  Note function arguments may be expanded in reverse order,\n            //  so we cannot use va_arg macro twice in a single call\n            byte *data = va_arg (argptr, byte *);\n            zmsg_addmem (msg, data, va_arg (argptr, int));\n        }\n        else\n        if (*picture == \'c\') {\n            zchunk_t *chunk = va_arg (argptr, zchunk_t *);\n            assert (zchunk_is (chunk));\n            zmsg_addmem (msg, zchunk_data (chunk), zchunk_size (chunk));\n        }\n        else\n        if (*picture == \'f\') {\n            zframe_t *frame = va_arg (argptr, zframe_t *);\n            assert (zframe_is (frame));\n            zmsg_addmem (msg, zframe_data (frame), zframe_size (frame));\n        }\n        else\n        if (*picture == \'U\') {\n            zuuid_t *uuid = va_arg (argptr, zuuid_t *);\n            zmsg_addmem (msg, zuuid_data (uuid), zuuid_size (uuid));\n        }\n        else\n        if (*picture == \'p\') {\n            void *pointer = va_arg (argptr, void *);\n            zmsg_addmem (msg, &pointer, sizeof (void *));\n        }\n        else\n        if (*picture == \'h\') {\n            zhashx_t *hash = va_arg (argptr, zhashx_t *);\n            zframe_t *frame = zhashx_pack (hash);\n            zmsg_append (msg, &frame);\n        }\n#ifdef CZMQ_BUILD_DRAFT_API\n        else\n        if (*picture == \'l\') {\n            zlistx_t *list = va_arg (argptr, zlistx_t *);\n            zframe_t *frame = zlistx_pack (list);\n            zmsg_append (msg, &frame);\n        }\n#endif\t\n        else\n        if (*picture == \'m\') {\n            zframe_t *frame;\n            zmsg_t *zmsg = va_arg (argptr, zmsg_t *);\n            for (frame = zmsg_first (zmsg); frame;\n                 frame = zmsg_next (zmsg) ) {\n                zframe_t *frame_dup = zframe_dup (frame);\n                zmsg_append (msg, &frame_dup);\n            }\n        }\n        else\n        if (*picture == \'z\')\n            zmsg_addmem (msg, NULL, 0);\n        else {\n            zsys_error ("zsock: invalid picture element \'%c\'", *picture);\n            assert (false);\n        }\n        picture++;\n    }\n\n    int rc = zmsg_send (&msg, self);\n    if (rc != 0)\n        zmsg_destroy (&msg);\n\n    return rc;\n}\n\n\n//  --------------------------------------------------------------------------\n//  Receive a \'picture\' message to the socket (or actor). See zsock_send for\n//  the format and meaning of the picture. Returns the picture elements into\n//  a series of pointers as provided by the caller:\n//\n//      i = int * (stores signed integer)\n//      1 = uint8_t * (stores 8-bit unsigned integer)\n//      2 = uint16_t * (stores 16-bit unsigned integer)\n//      4 = uint32_t * (stores 32-bit unsigned integer)\n//      8 = uint64_t * (stores 64-bit unsigned integer)\n//      s = char ** (allocates new string)\n//      b = byte **, size_t * (2 arguments) (allocates memory)\n//      c = zchunk_t ** (creates zchunk)\n//      f = zframe_t ** (creates zframe)\n//      h = zhashx_t ** (creates zhashx)\n//      l = zlistx_t ** (creates zlistx)\n//      U = zuuid_t * (creates a zuuid with the data)\n//      p = void ** (stores pointer)\n//      m = zmsg_t ** (creates a zmsg with the remaing frames)\n//      z = null, asserts empty frame (0 arguments)\n//      u = uint * (stores unsigned integer, deprecated)\n//\n//  Note that zsock_recv creates the returned objects, and the caller must\n//  destroy them when finished with them. The supplied pointers do not need\n//  to be initialized. Returns 0 if successful, or -1 if it failed to recv\n//  a message, in which case the pointers are not modified. When message\n//  frames are truncated (a short message), sets return values to zero/null.\n//  If an argument pointer is NULL, does not store any value (skips it).\n//  An \'n\' picture matches an empty frame; if the message does not match,\n//  the method will return -1.\n\nint\nzsock_recv (void *self, const char *picture, ...)\n{\n    va_list argptr;\n    va_start (argptr, picture);\n    int rc = zsock_vrecv (self, picture, argptr);\n    va_end (argptr);\n    return rc;\n}\n\n\n//  Receive a \'picture\' message from the socket (or actor). This is a\n//  va_list version of zsock_recv (), so please consult its documentation\n//  for the details.\n\nint\nzsock_vrecv (void *self, const char *picture, va_list argptr)\n{\n    assert (self);\n    assert (picture);\n    zmsg_t *msg = zmsg_recv (self);\n    if (!msg)\n        return -1;              //  Interrupted\n\n    //  Filter a signal that may come from a dying actor\n    if (zmsg_signal (msg) >= 0) {\n        zmsg_destroy (&msg);\n        return -1;\n    }\n    //  Now parse message according to picture argument\n    int rc = 0;\n    while (*picture) {\n        if (*picture == \'i\') {\n            char *string = zmsg_popstr (msg);\n            int *int_p = va_arg (argptr, int *);\n            if (int_p)\n                *int_p = string? atoi (string): 0;\n            freen (string);\n        }\n        else\n        if (*picture == \'1\') {\n            char *string = zmsg_popstr (msg);\n            uint8_t *uint8_p = va_arg (argptr, uint8_t *);\n            if (uint8_p)\n                *uint8_p = string? (uint8_t) atoi (string): 0;\n            freen (string);\n        }\n        else\n        if (*picture == \'2\') {\n            char *string = zmsg_popstr (msg);\n            uint16_t *uint16_p = va_arg (argptr, uint16_t *);\n            if (uint16_p)\n                *uint16_p = string? (uint16_t) atol (string): 0;\n            freen (string);\n        }\n        else\n        if (*picture == \'4\') {\n            char *string = zmsg_popstr (msg);\n            uint32_t *uint32_p = va_arg (argptr, uint32_t *);\n            if (uint32_p)\n                *uint32_p = string? (uint32_t) strtoul (string, NULL, 10): 0;\n            freen (string);\n        }\n        else\n        if (*picture == \'8\') {\n            char *string = zmsg_popstr (msg);\n            uint64_t *uint64_p = va_arg (argptr, uint64_t *);\n            if (uint64_p)\n                *uint64_p = string? (uint64_t) strtoull (string, NULL, 10): 0;\n            freen (string);\n        }\n        else\n        if (*picture == \'u\') {  //  Deprecated, use 4 or 8 instead\n            char *string = zmsg_popstr (msg);\n            uint *uint_p = va_arg (argptr, uint *);\n            if (uint_p)\n                *uint_p = string? (uint) strtoul (string, NULL, 10): 0;\n            freen (string);\n        }\n        else\n        if (*picture == \'s\') {\n            char *string = zmsg_popstr (msg);\n            char **string_p = va_arg (argptr, char **);\n            if (string_p)\n                *string_p = string;\n            else\n                freen (string);\n        }\n        else\n        if (*picture == \'b\') {\n            zframe_t *frame = zmsg_pop (msg);\n            byte **data_p = va_arg (argptr, byte **);\n            size_t *size = va_arg (argptr, size_t *);\n            if (data_p) {\n                if (frame) {\n                    *size = zframe_size (frame);\n                    *data_p = (byte *) malloc (*size);\n                    memcpy (*data_p, zframe_data (frame), *size);\n                }\n                else {\n                    *data_p = NULL;\n                    *size = 0;\n                }\n            }\n            zframe_destroy (&frame);\n        }\n        else\n        if (*picture == \'c\') {\n            zframe_t *frame = zmsg_pop (msg);\n            zchunk_t **chunk_p = va_arg (argptr, zchunk_t **);\n            if (chunk_p) {\n                if (frame)\n                    *chunk_p = zchunk_new (zframe_data (frame), zframe_size (frame));\n                else\n                    *chunk_p = NULL;\n            }\n            zframe_destroy (&frame);\n        }\n        else\n        if (*picture == \'f\') {\n            zframe_t *frame = zmsg_pop (msg);\n            zframe_t **frame_p = va_arg (argptr, zframe_t **);\n            if (frame_p)\n                *frame_p = frame;\n            else\n                zframe_destroy (&frame);\n        }\n        else\n        if (*picture == \'U\') {\n            zframe_t *frame = zmsg_pop (msg);\n            zuuid_t **uuid_p = va_arg (argptr, zuuid_t **);\n            if (uuid_p) {\n                if (frame) {\n                    *uuid_p = zuuid_new ();\n                    zuuid_set (*uuid_p, zframe_data (frame));\n                }\n                else\n                    *uuid_p = NULL;\n            }\n            zframe_destroy (&frame);\n        }\n        else\n        if (*picture == \'p\') {\n            zframe_t *frame = zmsg_pop (msg);\n            void **pointer_p = va_arg (argptr, void **);\n            if (pointer_p) {\n                if (frame) {\n                    if (zframe_size (frame) == sizeof (void *))\n                        *pointer_p = *((void **) zframe_data (frame));\n                    else\n                        rc = -1;\n                }\n                else\n                    *pointer_p = NULL;\n            }\n            zframe_destroy (&frame);\n        }\n        else\n        if (*picture == \'h\') {\n            zframe_t *frame = zmsg_pop (msg);\n            zhashx_t **hash_p = va_arg (argptr, zhashx_t **);\n            if (hash_p) {\n                if (frame)\n                    *hash_p = zhashx_unpack (frame);\n                else\n                    *hash_p = NULL;\n            }\n            zframe_destroy (&frame);\n        }\n#ifdef CZMQ_BUILD_DRAFT_API\t\n        else\n        if (*picture == \'l\') {\n            zframe_t *frame = zmsg_pop (msg);\n            zlistx_t **list_p = va_arg (argptr, zlistx_t **);\n            if (list_p) {\n                if (frame)\n                    *list_p = zlistx_unpack (frame);\n                else\n                    *list_p = NULL;\n            }\n\t    zframe_destroy (&frame);\n        }\n#endif\n        else\n        if (*picture == \'m\') {\n            zmsg_t **zmsg_p = va_arg (argptr, zmsg_t **);\n            if (zmsg_p) {\n                *zmsg_p = zmsg_new ();\n                zframe_t *frame;\n                while ((frame = zmsg_pop (msg)))\n                    zmsg_append (*zmsg_p, &frame);\n            }\n        }\n        else\n        if (*picture == \'z\') {\n            zframe_t *frame = zmsg_pop (msg);\n            if (frame && zframe_size (frame) != 0)\n                rc = -1;\n            zframe_destroy (&frame);\n        }\n        else {\n            zsys_error ("zsock: invalid picture element \'%c\'", *picture);\n            assert (false);\n        }\n        picture++;\n    }\n    zmsg_destroy (&msg);\n    return rc;\n}\n\n\n//  --------------------------------------------------------------------------\n//  Network data encoding macros that we use in bsend/brecv\n\n//  Put a 1-byte number to the frame\n#define PUT_NUMBER1(host) { \\\n    *(byte *) needle = (host); \\\n    needle++; \\\n}\n\n//  Put a 2-byte number to the frame\n#define PUT_NUMBER2(host) { \\\n    needle [0] = (byte) (((host) >> 8)  & 255); \\\n    needle [1] = (byte) (((host))       & 255); \\\n    needle += 2; \\\n}\n\n//  Put a 4-byte number to the frame\n#define PUT_NUMBER4(host) { \\\n    needle [0] = (byte) (((host) >> 24) & 255); \\\n    needle [1] = (byte) (((host) >> 16) & 255); \\\n    needle [2] = (byte) (((host) >> 8)  & 255); \\\n    needle [3] = (byte) (((host))       & 255); \\\n    needle += 4; \\\n}\n\n//  Put a 8-byte number to the frame\n#define PUT_NUMBER8(host) { \\\n    needle [0] = (byte) (((host) >> 56) & 255); \\\n    needle [1] = (byte) (((host) >> 48) & 255); \\\n    needle [2] = (byte) (((host) >> 40) & 255); \\\n    needle [3] = (byte) (((host) >> 32) & 255); \\\n    needle [4] = (byte) (((host) >> 24) & 255); \\\n    needle [5] = (byte) (((host) >> 16) & 255); \\\n    needle [6] = (byte) (((host) >> 8)  & 255); \\\n    needle [7] = (byte) (((host))       & 255); \\\n    needle += 8; \\\n}\n\n//  Get a 1-byte number from the frame\n#define GET_NUMBER1(host) { \\\n    if (needle + 1 > ceiling) \\\n        goto malformed; \\\n    (host) = *(byte *) needle; \\\n    needle++; \\\n}\n\n//  Get a 2-byte number from the frame\n#define GET_NUMBER2(host) { \\\n    if (needle + 2 > ceiling) \\\n        goto malformed; \\\n    (host) = ((uint16_t) (needle [0]) << 8) \\\n           +  (uint16_t) (needle [1]); \\\n    needle += 2; \\\n}\n\n//  Get a 4-byte number from the frame\n#define GET_NUMBER4(host) { \\\n    if (needle + 4 > ceiling) \\\n        goto malformed; \\\n    (host) = ((uint32_t) (needle [0]) << 24) \\\n           + ((uint32_t) (needle [1]) << 16) \\\n           + ((uint32_t) (needle [2]) << 8) \\\n           +  (uint32_t) (needle [3]); \\\n    needle += 4; \\\n}\n\n//  Get a 8-byte number from the frame\n#define GET_NUMBER8(host) { \\\n    if (needle + 8 > ceiling) \\\n        goto malformed; \\\n    (host) = ((uint64_t) (needle [0]) << 56) \\\n           + ((uint64_t) (needle [1]) << 48) \\\n           + ((uint64_t) (needle [2]) << 40) \\\n           + ((uint64_t) (needle [3]) << 32) \\\n           + ((uint64_t) (needle [4]) << 24) \\\n           + ((uint64_t) (needle [5]) << 16) \\\n           + ((uint64_t) (needle [6]) << 8) \\\n           +  (uint64_t) (needle [7]); \\\n    needle += 8; \\\n}\n\n\n//  --------------------------------------------------------------------------\n//  Send a binary encoded \'picture\' message to the socket (or actor). This\n//  method is similar to zsock_send, except the arguments are encoded in a\n//  binary format that is compatible with zproto, and is designed to reduce\n//  memory allocations. The pattern argument is a string that defines the\n//  type of each argument. Supports these argument types:\n//\n//   pattern    C type                  zproto type:\n//      1       uint8_t                 type = "number" size = "1"\n//      2       uint16_t                type = "number" size = "2"\n//      4       uint32_t                type = "number" size = "3"\n//      8       uint64_t                type = "number" size = "4"\n//      s       char *, 0-255 chars     type = "string"\n//      S       char *, 0-2^32-1 chars  type = "longstr"\n//      c       zchunk_t *              type = "chunk"\n//      f       zframe_t *              type = "frame"\n//      u       zuuid_t *               type = "uuid"\n//      m       zmsg_t *                type = "msg"\n//      p       void *, sends pointer value, only over inproc\n//\n//  Does not change or take ownership of any arguments. Returns 0 if\n//  successful, -1 if sending failed for any reason.\n\nint\nzsock_bsend (void *self, const char *picture, ...)\n{\n    assert (self);\n    assert (picture);\n\n    //  Pass 1: calculate total size of data frame\n    size_t frame_size = 0;\n    zframe_t *frames [ZSOCK_BSEND_MAX_FRAMES]; //  Non-data frames to send\n    size_t nbr_frames = 0;                     //  Size of this table\n    va_list argptr;\n    va_start (argptr, picture);\n    const char *picptr = picture;\n    while (*picptr) {\n        if (*picptr == \'1\') {\n            va_arg (argptr, int);\n            frame_size += 1;\n        }\n        else\n        if (*picptr == \'2\') {\n            va_arg (argptr, int);\n            frame_size += 2;\n        }\n        else\n        if (*picptr == \'4\') {\n            va_arg (argptr, uint32_t);\n            frame_size += 4;\n        }\n        else\n        if (*picptr == \'8\') {\n            va_arg (argptr, uint64_t);\n            frame_size += 8;\n        }\n        else\n        if (*picptr == \'s\') {\n            char *string = va_arg (argptr, char *);\n            frame_size += 1 + (string? strlen (string): 0);\n        }\n        else\n        if (*picptr == \'S\') {\n            char *string = va_arg (argptr, char *);\n            frame_size += 4 + (string? strlen (string): 0);\n        }\n        else\n        if (*picptr == \'c\') {\n            zchunk_t *chunk = va_arg (argptr, zchunk_t *);\n            frame_size += 4 + (chunk? zchunk_size (chunk): 0);\n        }\n        else\n        if (*picptr == \'p\') {\n            va_arg (argptr, void *);\n            frame_size += sizeof (void *);\n        }\n        else\n        if (*picptr == \'f\') {\n            zframe_t *frame = va_arg (argptr, zframe_t *);\n            assert (nbr_frames < ZSOCK_BSEND_MAX_FRAMES - 1);\n            frames [nbr_frames++] = frame;\n        }\n        else\n        if (*picptr == \'u\') {\n            zuuid_t *uuid = va_arg (argptr, zuuid_t *);\n            frame_size += zuuid_size (uuid);\n        }\n        else\n        if (*picptr == \'m\') {\n            if (picptr [1]) {\n                zsys_error ("zsock_bsend: \'m\' (zmsg) only valid at end of picptr");\n                assert (false);\n            }\n            zmsg_t *msg = va_arg (argptr, zmsg_t *);\n            if (msg) {\n                zframe_t *frame = zmsg_first (msg);\n                while (frame) {\n                    assert (nbr_frames < ZSOCK_BSEND_MAX_FRAMES - 1);\n                    frames [nbr_frames++] = frame;\n                    frame = zmsg_next (msg);\n                }\n            }\n            else\n                frames [nbr_frames++] = zframe_new_empty ();\n        }\n        else {\n            zsys_error ("zsock_bsend: invalid picptr element \'%c\'", *picptr);\n            assert (false);\n        }\n        picptr++;\n    }\n    va_end (argptr);\n\n    //  Pass 2: encode data into data frame\n    zmq_msg_t msg;\n    zmq_msg_init_size (&msg, frame_size);\n    byte *needle = (byte *) zmq_msg_data (&msg);\n\n    //  Set routing id if self is zsock\n#if defined ZMQ_SERVER\n    if (zsock_is (self) && zsock_routing_id ((zsock_t *)self) != 0)\n        zmq_msg_set_routing_id (&msg, zsock_routing_id ((zsock_t *)self));\n#endif\n\n    va_start (argptr, picture);\n    picptr = picture;\n    while (*picptr) {\n        if (*picptr == \'1\') {\n            int number1 = va_arg (argptr, int);\n            PUT_NUMBER1 (number1);\n        }\n        else\n        if (*picptr == \'2\') {\n            int number2 = va_arg (argptr, int);\n            PUT_NUMBER2 (number2);\n        }\n        else\n        if (*picptr == \'4\') {\n            uint32_t number4 = va_arg (argptr, uint32_t);\n            PUT_NUMBER4 (number4);\n        }\n        else\n        if (*picptr == \'8\') {\n            uint64_t number8 = va_arg (argptr, uint64_t);\n            PUT_NUMBER8 (number8);\n        }\n        else\n        if (*picptr == \'p\') {\n            void *pointer = va_arg (argptr, void *);\n            memcpy (needle, &pointer, sizeof (void *));\n            needle += sizeof (void *);\n        }\n        else\n        if (*picptr == \'s\') {\n            char *string = va_arg (argptr, char *);\n            if (!string)\n                string = "";\n            size_t string_size = strlen (string);\n            PUT_NUMBER1 ((byte) string_size);\n            memcpy (needle, string, string_size);\n            needle += string_size;\n        }\n        else\n        if (*picptr == \'S\') {\n            char *string = va_arg (argptr, char *);\n            if (!string)\n                string = "";\n            size_t string_size = strlen (string);\n            PUT_NUMBER4 (string_size);\n            memcpy (needle, string, string_size);\n            needle += string_size;\n        }\n        else\n        if (*picptr == \'c\') {\n            zchunk_t *chunk = va_arg (argptr, zchunk_t *);\n            if (chunk) {\n                PUT_NUMBER4 (zchunk_size (chunk));\n                memcpy (needle, zchunk_data (chunk), zchunk_size (chunk));\n                needle += zchunk_size (chunk);\n            }\n        }\n        else\n        if (*picptr == \'u\') {\n            zuuid_t *uuid = va_arg (argptr, zuuid_t *);\n            if (uuid) {\n                memcpy (needle, zuuid_data (uuid), zuuid_size (uuid));\n                needle += zuuid_size (uuid);\n            }\n        }\n        picptr++;\n    }\n    va_end (argptr);\n\n    //  Now send the data frame\n    void *handle = zsock_resolve (self);\n    int rc = zmq_msg_send (&msg, handle, nbr_frames? ZMQ_SNDMORE: 0);\n    if (rc >= 0) {\n        //  Now send any additional frames\n        unsigned int frame_nbr;\n        for (frame_nbr = 0; frame_nbr < nbr_frames; frame_nbr++) {\n            bool more = frame_nbr < nbr_frames - 1;\n            rc = zframe_send (&frames [frame_nbr], self,\n                              ZFRAME_REUSE + (more? ZFRAME_MORE: 0));\n            if (rc == -1)\n                break;\n        }\n    }\n    return rc >= 0? 0: -1;\n}\n\n\n//  --------------------------------------------------------------------------\n//  Receive a binary encoded \'picture\' message from the socket (or actor).\n//  This method is similar to zsock_recv, except the arguments are encoded\n//  in a binary format that is compatible with zproto, and is designed to\n//  reduce memory allocations. The pattern argument is a string that defines\n//  the type of each argument. See zsock_bsend for the supported argument\n//  types. All arguments must be pointers; this call sets them to point to\n//  values held on a per-socket basis.\n//  For types 1, 2, 4 and 8 the caller must allocate the memory itself before\n//  calling zsock_brecv.\n//  For types S, the caller must free the value once finished with it, as\n//  zsock_brecv will allocate the buffer.\n//  For type s, the caller must not free the value as it is stored in a\n//  local cache for performance purposes.\n//  For types c, f, u and m the caller must call the appropriate destructor\n//  depending on the object as zsock_brecv will create new objects.\n//  For type p the caller must coordinate with the sender, as it is just a\n//  pointer value being passed.\n\n//  This is the largest size we allow for an incoming longstr or chunk (1M)\n#define MAX_ALLOC_SIZE      1024 * 1024\n\nint\nzsock_brecv (void *selfish, const char *picture, ...)\n{\n    assert (selfish);           //  Kind of self\n    assert (picture);\n\n    zmq_msg_t msg;\n    zmq_msg_init (&msg);\n    if (zmq_msg_recv (&msg, zsock_resolve (selfish), 0) == -1)\n        return -1;              //  Interrupted\n\n    //  If we don\'t have a string cache, create one now with arbitrary\n    //  value; this will grow if needed. Do not use an initial size less\n    //  than 256, or cache expansion will not work properly.\n    zsock_t *self = (zsock_t *) selfish;\n    if (zactor_is (selfish))\n        self = zactor_sock ((zactor_t *) selfish);\n\n    if (!self->cache) {\n        self->cache = (char *) malloc (512);\n        self->cache_size = 512;\n    }\n    //  Last received strings are cached per socket\n    uint cache_used = 0;\n    byte *needle = (byte *) zmq_msg_data (&msg);\n    byte *ceiling = needle + zmq_msg_size (&msg);\n\n    //  If selfish is zsock get routing id from msg\n#if defined ZMQ_SERVER\n    if (zsock_is (selfish) && zsock_type (self) == ZMQ_SERVER)\n        zsock_set_routing_id (self, zmq_msg_routing_id (&msg));\n#endif\n\n    va_list argptr;\n    va_start (argptr, picture);\n    const char *picptr = picture;\n    while (*picptr) {\n        if (*picptr == \'1\') {\n            uint8_t *number1_p = va_arg (argptr, uint8_t *);\n            GET_NUMBER1 (*number1_p);\n        }\n        else\n        if (*picptr == \'2\') {\n            uint16_t *number2_p = va_arg (argptr, uint16_t *);\n            GET_NUMBER2 (*number2_p);\n        }\n        else\n        if (*picptr == \'4\') {\n            uint32_t *number4_p = va_arg (argptr, uint32_t *);\n            GET_NUMBER4 (*number4_p);\n        }\n        else\n        if (*picptr == \'8\') {\n            uint64_t *number8_p = va_arg (argptr, uint64_t *);\n            GET_NUMBER8 (*number8_p);\n        }\n        else\n        if (*picptr == \'p\') {\n            void **pointer_p = va_arg (argptr, void **);\n            memcpy (pointer_p, needle, sizeof (void *));\n            needle += sizeof (void *);\n        }\n        else\n        if (*picptr == \'s\') {\n            char **string_p = va_arg (argptr, char **);\n            uint string_size;\n            GET_NUMBER1 (string_size);\n            if (needle + string_size > ceiling)\n                goto malformed;\n            //  Expand cache if we need to; string is guaranteed to fit into\n            //  expansion space\n            if (cache_used + string_size > self->cache_size) {\n                puts ("REALLOC");\n                self->cache_size *= 2;\n                self->cache = (char *) realloc (self->cache, self->cache_size);\n                assert (self->cache);\n            }\n            *string_p = self->cache + cache_used;\n            memcpy (*string_p, needle, string_size);\n            cache_used += string_size;\n            self->cache [cache_used++] = 0;\n            needle += string_size;\n        }\n        else\n        if (*picptr == \'S\') {\n            char **string_p = va_arg (argptr, char **);\n            size_t string_size;\n            GET_NUMBER4 (string_size);\n            if (string_size > MAX_ALLOC_SIZE\n            ||  needle + string_size > (ceiling))\n                goto malformed;\n            *string_p = (char *) malloc (string_size + 1);\n            assert (string_p);\n            memcpy (*string_p, needle, string_size);\n            (*string_p) [string_size] = 0;\n            needle += string_size;\n        }\n        else\n        if (*picptr == \'c\') {\n            zchunk_t **chunk_p = va_arg (argptr, zchunk_t **);\n            size_t chunk_size;\n            GET_NUMBER4 (chunk_size);\n            if (chunk_size > MAX_ALLOC_SIZE\n            ||  needle + chunk_size > (ceiling))\n                goto malformed;\n            *chunk_p = zchunk_new (needle, chunk_size);\n            needle += chunk_size;\n        }\n        else\n        if (*picptr == \'u\') {\n            zuuid_t **uuid_p = va_arg (argptr, zuuid_t **);\n            *uuid_p = zuuid_new ();\n            if (zuuid_size (*uuid_p) > MAX_ALLOC_SIZE\n            ||  needle + zuuid_size (*uuid_p) > (ceiling))\n                goto malformed;\n            zuuid_set (*uuid_p, needle);\n            needle += zuuid_size (*uuid_p);\n        }\n        else\n        if (*picptr == \'f\') {\n            zframe_t **frame_p = va_arg (argptr, zframe_t **);\n            //  Get next frame off socket\n            if (!zsock_rcvmore (self))\n                goto malformed;\n            *frame_p = zframe_recv (self);\n        }\n        else\n        if (*picptr == \'m\') {\n            if (picptr [1]) {\n                zsys_error ("zsock_brecv: \'m\' (zmsg) only valid at end of picptr");\n                assert (false);\n            }\n            zmsg_t **msg_p = va_arg (argptr, zmsg_t **);\n            //  Get zero or more remaining frames\n            if (!zsock_rcvmore (self))\n                goto malformed;\n            *msg_p = zmsg_recv (self);\n        }\n        else {\n            zsys_error ("zsock_brecv: invalid picptr element \'%c\'", *picptr);\n            assert (false);\n        }\n        picptr++;\n    }\n    va_end (argptr);\n    zmq_msg_close (&msg);\n    return 0;\n\n    //  Error return\n    malformed:\n        zmq_msg_close (&msg);\n        return -1;              //  Invalid message\n}\n\n\n//  --------------------------------------------------------------------------\n//  Return socket routing ID if any. This returns 0 if the socket is not\n//  of type ZMQ_SERVER or if no request was already received on it.\n\nuint32_t\nzsock_routing_id (zsock_t *self)\n{\n    return self->routing_id;\n}\n\n\n//  --------------------------------------------------------------------------\n//  Set routing ID on socket. The socket MUST be of type ZMQ_SERVER.\n//  This will be used when sending messages on the socket via the zsock API.\n\nvoid\nzsock_set_routing_id (zsock_t *self, uint32_t routing_id)\n{\n    assert (self);\n    self->routing_id = routing_id;\n}\n\n\n//  --------------------------------------------------------------------------\n//  Set socket to use unbounded pipes (HWM=0); use this in cases when you are\n//  totally certain the message volume can fit in memory. This method works\n//  across all versions of ZeroMQ. Takes a polymorphic socket reference.\n\nvoid\nzsock_set_unbounded (void *self)\n{\n#if (ZMQ_VERSION_MAJOR == 2)\n    zsock_set_hwm (self, 0);\n#else\n    zsock_set_sndhwm (self, 0);\n    zsock_set_rcvhwm (self, 0);\n#endif\n}\n\n//  --------------------------------------------------------------------------\n//  Send a signal over a socket. A signal is a short message carrying a\n//  success/failure code (by convention, 0 means OK). Signals are encoded\n//  to be distinguishable from "normal" messages. Accepts a zsock_t or a\n//  zactor_t argument, and returns 0 if successful, -1 if the signal could\n//  not be sent. Takes a polymorphic socket reference.\n\nint\nzsock_signal (void *self, byte status)\n{\n    assert (self);\n    zmsg_t *msg = zmsg_new_signal (status);\n    if (zmsg_send (&msg, self) < 0) {\n        zmsg_destroy (&msg);\n        return -1;\n    }\n    else\n        return 0;\n}\n\n\n//  --------------------------------------------------------------------------\n//  Wait on a signal. Use this to coordinate between threads, over pipe\n//  pairs. Blocks until the signal is received. Returns -1 on error, 0 or\n//  greater on success. Accepts a zsock_t or a zactor_t as argument.\n//  Takes a polymorphic socket reference.\n\nint\nzsock_wait (void *self)\n{\n    assert (self);\n\n    //  Loop and discard messages until we get a signal value or interrupt.\n    while (true) {\n        zmsg_t *msg = zmsg_recv (self);\n        if (!msg)\n            return -1;\n        int rc = zmsg_signal (msg);\n        zmsg_destroy (&msg);\n        if (rc >= 0)\n            return rc;\n    }\n    return -1;\n}\n\n\n//  --------------------------------------------------------------------------\n//  If there is a partial message still waiting on the socket, remove and\n//  discard it. This is useful when reading partial messages, to get specific\n//  message types.\n\nvoid\nzsock_flush (void *self)\n{\n    if (zsock_rcvmore (self)) {\n        zmsg_t *msg = zmsg_recv (self);\n        zmsg_destroy (&msg);\n    }\n}\n\n\n//  --------------------------------------------------------------------------\n//  Join a group for the RADIO-DISH pattern. Call only on ZMQ_DISH.\n//  Returns 0 if OK, -1 if failed.\n\nint\nzsock_join (void *self, const char *group)\n{\n    assert (self);\n#ifdef ZMQ_DISH\n    return zmq_join (zsock_resolve (self), group);\n#else\n    errno = ENOTSUP;\n    return -1;\n#endif\n}\n\n\n//  --------------------------------------------------------------------------\n//  Leave a group for the RADIO-DISH pattern. Call only on ZMQ_DISH.\n//  Returns 0 if OK, -1 if failed.\n\nint\nzsock_leave (void *self, const char *group)\n{\n    assert (self);\n#ifdef ZMQ_DISH\n    return zmq_leave (zsock_resolve (self), group);\n#else\n    errno = ENOTSUP;\n    return -1;\n#endif\n}\n\n\n//  --------------------------------------------------------------------------\n//  Probe the supplied object, and report if it looks like a zsock_t.\n//  Takes a polymorphic socket reference.\n\nbool\nzsock_is (void *self)\n{\n    assert (self);\n    return ((zsock_t *) self)->tag == ZSOCK_TAG;\n}\n\n\n//  --------------------------------------------------------------------------\n//  Probe the supplied \'self\' pointer. Takes a polymorphic socket reference.\n//  If self is a zactor_t, zsock_t, or libzmq socket handle, returns the\n//  libzmq socket handle. If self is a valid file descriptor, returns NULL.\n//  Else returns self as-is.\n\nvoid *\nzsock_resolve (void *self)\n{\n    assert (self);\n    if (zactor_is (self))\n        return zactor_resolve (self);\n\n    if (zsock_is (self))\n        return ((zsock_t *) self)->handle;\n\n    //  Check if we have a valid ZMQ socket by probing the socket type\n    int type;\n    size_t option_len = sizeof (int);\n    if (zmq_getsockopt (self, ZMQ_TYPE, &type, &option_len) == 0)\n        return self;\n\n    //  Check if self is a valid FD or socket FD\n    //  TODO: this code should move to zsys_isfd () as we don\'t like\n    //  non-portable code outside of that class.\n    int sock_type = -1;\n#if defined (__WINDOWS__)\n    int sock_type_size = sizeof (int);\n    int rc = getsockopt (*(SOCKET *) self, SOL_SOCKET, SO_TYPE, (char *) &sock_type, &sock_type_size);\n    if (rc == 0)\n        return NULL;        //  It\'s a socket descriptor\n#else\n    socklen_t sock_type_size = sizeof (socklen_t);\n    int rc = getsockopt (*(SOCKET *) self, SOL_SOCKET, SO_TYPE, (char *) &sock_type, &sock_type_size);\n    if (rc == 0 || (rc == -1 && errno == ENOTSOCK))\n        return NULL;        //  It\'s a socket FD or FD\n#endif\n    //  Socket appears to be something else, return it as-is\n    return self;\n}\n\n\n//  --------------------------------------------------------------------------\n//  Check whether the socket has available message to read.\n\nbool\nzsock_has_in (void *self) {\n    int events = zsock_events (self);\n    return (events & ZMQ_POLLIN) != 0;\n}\n\n\n//  We use the gossip messages for some test cases\n#include "zgossip_msg.h"\n\n//  --------------------------------------------------------------------------\n//  Selftest\n\nvoid\nzsock_test (bool verbose)\n{\n    printf (" * zsock: ");\n    if (verbose)\n        printf ("\\n");\n\n    //  @selftest\n    zsock_t *writer = zsock_new (ZMQ_PUSH);\n    assert (writer);\n    int port = zsock_bind (writer, "tcp://127.0.0.1:*");\n    assert (port != -1);\n    assert (zsock_resolve (writer) != writer);\n    assert (streq (zsock_type_str (writer), "PUSH"));\n\n    int rc;\n#if (ZMQ_VERSION >= ZMQ_MAKE_VERSION (3, 2, 0))\n    //  Check unbind\n    rc = zsock_unbind (writer, "tcp://127.0.0.1:%d", port);\n    assert (rc == 0);\n\n    //  In some cases and especially when running under Valgrind, doing\n    //  a bind immediately after an unbind causes an EADDRINUSE error.\n    //  Even a short sleep allows the OS to release the port for reuse.\n    zclock_sleep (100);\n\n    //  Bind again\n    rc = zsock_bind (writer, "tcp://127.0.0.1:%d", port);\n    assert (rc == port);\n    char endpoint [40];\n    sprintf (endpoint, "tcp://127.0.0.1:%d", port);\n    assert (streq (zsock_endpoint (writer), endpoint));\n#endif\n\n    zsock_t *reader = zsock_new (ZMQ_PULL);\n    assert (reader);\n    rc = zsock_connect (reader, "tcp://127.0.0.1:%d", port);\n    assert (rc != -1);\n    assert (zsock_resolve (reader) != reader);\n    assert (streq (zsock_type_str (reader), "PULL"));\n\n    //  Basic Hello, World\n    zstr_send (writer, "Hello, World");\n    zmsg_t *msg = zmsg_recv (reader);\n    assert (msg);\n    char *string = zmsg_popstr (msg);\n    assert (streq (string, "Hello, World"));\n    freen (string);\n    zmsg_destroy (&msg);\n\n    //  Test resolve libzmq socket\n#if (ZMQ_VERSION >= ZMQ_MAKE_VERSION (3, 2, 0))\n    void *zmq_ctx = zmq_ctx_new ();\n#else\n    void *zmq_ctx = zmq_ctx_new (1);\n#endif\n    assert (zmq_ctx);\n    void *zmq_sock = zmq_socket (zmq_ctx, ZMQ_PUB);\n    assert (zmq_sock);\n    assert (zsock_resolve (zmq_sock) == zmq_sock);\n    zmq_close (zmq_sock);\n    zmq_ctx_term (zmq_ctx);\n\n    //  Test resolve zsock\n    zsock_t *resolve = zsock_new_pub("@tcp://127.0.0.1:*");\n    assert (resolve);\n    assert (zsock_resolve (resolve) == resolve->handle);\n    zsock_destroy (&resolve);\n\n    //  Test resolve FD\n    SOCKET fd = zsock_fd (reader);\n    assert (zsock_resolve ((void *) &fd) == NULL);\n\n    //  Test binding to ephemeral ports, sequential and random\n    port = zsock_bind (writer, "tcp://127.0.0.1:*");\n    assert (port >= DYNAMIC_FIRST && port <= DYNAMIC_LAST);\n    port = zsock_bind (writer, "tcp://127.0.0.1:*[50000-]");\n    assert (port >= 50000 && port <= DYNAMIC_LAST);\n    port = zsock_bind (writer, "tcp://127.0.0.1:*[-50001]");\n    assert (port >= DYNAMIC_FIRST && port <= 50001);\n    port = zsock_bind (writer, "tcp://127.0.0.1:*[60000-60500]");\n    assert (port >= 60000 && port <= 60500);\n\n    port = zsock_bind (writer, "tcp://127.0.0.1:!");\n    assert (port >= DYNAMIC_FIRST && port <= DYNAMIC_LAST);\n    port = zsock_bind (writer, "tcp://127.0.0.1:![50000-]");\n    assert (port >= 50000 && port <= DYNAMIC_LAST);\n    port = zsock_bind (writer, "tcp://127.0.0.1:![-50001]");\n    assert (port >= DYNAMIC_FIRST && port <= 50001);\n    port = zsock_bind (writer, "tcp://127.0.0.1:![60000-60500]");\n    assert (port >= 60000 && port <= 60500);\n\n    //  Test zsock_attach method\n    zsock_t *dealer = zsock_new (ZMQ_DEALER);\n    assert (dealer);\n    rc = zsock_attach (dealer, "@inproc://myendpoint,tcp://127.0.0.1:*,inproc://others", true);\n    assert (rc == 0);\n    rc = zsock_attach (dealer, "", false);\n    assert (rc == 0);\n    rc = zsock_attach (dealer, NULL, true);\n    assert (rc == 0);\n    rc = zsock_attach (dealer, ">a,@b, c,, ", false);\n    assert (rc == -1);\n    zsock_destroy (&dealer);\n\n    //  Test zsock_endpoint method\n    rc = zsock_bind (writer, "inproc://test.%s", "writer");\n    assert (rc == 0);\n    assert (streq (zsock_endpoint (writer), "inproc://test.writer"));\n\n    //  Test error state when connecting to an invalid socket type\n    //  (\'txp://\' instead of \'tcp://\', typo intentional)\n    rc = zsock_connect (reader, "txp://127.0.0.1:5560");\n    assert (rc == -1);\n\n    //  Test signal/wait methods\n    rc = zsock_signal (writer, 123);\n    assert (rc == 0);\n    rc = zsock_wait (reader);\n    assert (rc == 123);\n\n    //  Test zsock_send/recv pictures\n    uint8_t  number1 = 123;\n    uint16_t number2 = 123 * 123;\n    uint32_t number4 = 123 * 123;\n    number4 *= 123;\n    uint32_t number4_MAX = UINT32_MAX;\n    uint64_t number8 = 123 * 123;\n    number8 *= 123;\n    number8 *= 123;\n    uint64_t number8_MAX = UINT64_MAX;\n\n    zchunk_t *chunk = zchunk_new ("HELLO", 5);\n    assert (chunk);\n    zframe_t *frame = zframe_new ("WORLD", 5);\n    assert (frame);\n    zhashx_t *hash = zhashx_new ();\n    assert (hash);\n#ifdef CZMQ_BUILD_DRAFT_API\n    zlistx_t *list = zlistx_new ();\n    assert (list);\n#endif\n    zuuid_t *uuid = zuuid_new ();\n    assert (uuid);\n    zhashx_set_destructor (hash, (zhashx_destructor_fn *) zstr_free);\n    zhashx_set_duplicator (hash, (zhashx_duplicator_fn *) strdup);\n    zhashx_insert (hash, "1", "value A");\n    zhashx_insert (hash, "2", "value B");\n#ifdef CZMQ_BUILD_DRAFT_API\n    zlistx_set_destructor (list, (zlistx_destructor_fn *) zstr_free);\n    zlistx_set_duplicator (list, (zlistx_duplicator_fn *) strdup);\n    zlistx_add_end (list, "1");\n    zlistx_add_end (list, "2");\n#endif\n    char *original = "pointer";\n\n    //  Test zsock_recv into each supported type\n#ifdef CZMQ_BUILD_DRAFT_API\n    zsock_send (writer, "i124488zsbcfUhlp",\n#else\n    zsock_send (writer, "i124488zsbcfUhp",\n#endif\n                -12345, number1, number2, number4, number4_MAX,\n                number8, number8_MAX,\n                "This is a string", "ABCDE", 5,\n#ifdef CZMQ_BUILD_DRAFT_API\n                chunk, frame, uuid, hash, list, original);\n#else\n                chunk, frame, uuid, hash, original);\n#endif\n    char *uuid_str = strdup (zuuid_str (uuid));\n    zchunk_destroy (&chunk);\n    zframe_destroy (&frame);\n    zuuid_destroy (&uuid);\n    zhashx_destroy (&hash);\n#ifdef CZMQ_BUILD_DRAFT_API\n    zlistx_destroy (&list);\n#endif\n\n    int integer;\n    byte *data;\n    size_t size;\n    char *pointer;\n    number8_MAX = number8 = number4_MAX = number4 = number2 = number1 = 0ULL;\n#ifdef CZMQ_BUILD_DRAFT_API\n    rc = zsock_recv (reader, "i124488zsbcfUhlp",\n#else\n    rc = zsock_recv (reader, "i124488zsbcfUhp",\n#endif\n                     &integer, &number1, &number2, &number4, &number4_MAX,\n                     &number8, &number8_MAX, &string, &data, &size, &chunk,\n#ifdef CZMQ_BUILD_DRAFT_API\n                     &frame, &uuid, &hash, &list, &pointer);\n#else\n                     &frame, &uuid, &hash, &pointer);\n#endif\n    assert (rc == 0);\n    assert (integer == -12345);\n    assert (number1 == 123);\n    assert (number2 == 123 * 123);\n    assert (number4 == 123 * 123 * 123);\n    assert (number4_MAX == UINT32_MAX);\n    assert (number8 == 123 * 123 * 123 * 123);\n    assert (number8_MAX == UINT64_MAX);\n    assert (streq (string, "This is a string"));\n    assert (memcmp (data, "ABCDE", 5) == 0);\n    assert (size == 5);\n    assert (memcmp (zchunk_data (chunk), "HELLO", 5) == 0);\n    assert (zchunk_size (chunk) == 5);\n    assert (streq (uuid_str, zuuid_str (uuid)));\n    assert (memcmp (zframe_data (frame), "WORLD", 5) == 0);\n    assert (zframe_size (frame) == 5);\n    char *value = (char *) zhashx_lookup (hash, "1");\n    assert (streq (value, "value A"));\n    value = (char *) zhashx_lookup (hash, "2");\n    assert (streq (value, "value B"));\n#ifdef CZMQ_BUILD_DRAFT_API\n    value = (char *) zlistx_first (list);\n    assert (streq (value, "1"));\n    value = (char *) zlistx_last (list);\n    assert (streq (value, "2"));\n#endif\n    assert (original == pointer);\n    freen (string);\n    freen (data);\n    freen (uuid_str);\n    zframe_destroy (&frame);\n    zchunk_destroy (&chunk);\n    zhashx_destroy (&hash);\n#ifdef CZMQ_BUILD_DRAFT_API\n    zlistx_destroy (&list);\n#endif\n    zuuid_destroy (&uuid);\n\n    //  Test zsock_recv of short message; this lets us return a failure\n    //  with a status code and then nothing else; the receiver will get\n    //  the status code and NULL/zero for all other values\n    zsock_send (writer, "i", -1);\n    zsock_recv (reader, "izsbcfp",\n        &integer, &string, &data, &size, &chunk, &frame, &pointer);\n    assert (integer == -1);\n    assert (string == NULL);\n    assert (data == NULL);\n    assert (size == 0);\n    assert (chunk == NULL);\n    assert (frame == NULL);\n    assert (pointer == NULL);\n\n    msg = zmsg_new ();\n    zmsg_addstr (msg, "frame 1");\n    zmsg_addstr (msg, "frame 2");\n    zsock_send (writer, "szm", "header", msg);\n    zmsg_destroy (&msg);\n\n    zsock_recv (reader, "szm", &string, &msg);\n\n    assert (streq ("header", string));\n    assert (zmsg_size (msg) == 2);\n    assert (zframe_streq (zmsg_first (msg), "frame 1"));\n    assert (zframe_streq (zmsg_next (msg), "frame 2"));\n    zstr_free (&string);\n    zmsg_destroy (&msg);\n\n    //  Test zsock_recv with null arguments\n    chunk = zchunk_new ("HELLO", 5);\n    assert (chunk);\n    frame = zframe_new ("WORLD", 5);\n    assert (frame);\n    zsock_send (writer, "izsbcfp",\n                -12345, "This is a string", "ABCDE", 5, chunk, frame, original);\n    zframe_destroy (&frame);\n    zchunk_destroy (&chunk);\n    zsock_recv (reader, "izsbcfp", &integer, NULL, NULL, NULL, &chunk, NULL, NULL);\n    assert (integer == -12345);\n    assert (memcmp (zchunk_data (chunk), "HELLO", 5) == 0);\n    assert (zchunk_size (chunk) == 5);\n    zchunk_destroy (&chunk);\n\n    //  Test zsock_bsend/brecv pictures with binary encoding\n    frame = zframe_new ("Hello", 5);\n    chunk = zchunk_new ("World", 5);\n\n    msg = zmsg_new ();\n    zmsg_addstr (msg, "Hello");\n    zmsg_addstr (msg, "World");\n\n    zsock_bsend (writer, "1248sSpcfm",\n                 number1, number2, number4, number8,\n                 "Hello, World",\n                 "Goodbye cruel World!",\n                 original,\n                 chunk, frame, msg);\n    zchunk_destroy (&chunk);\n    zframe_destroy (&frame);\n    zmsg_destroy (&msg);\n\n    number8 = number4 = number2 = number1 = 0;\n    char *longstr;\n    zsock_brecv (reader, "1248sSpcfm",\n                 &number1, &number2, &number4, &number8,\n                 &string, &longstr,\n                 &pointer,\n                 &chunk, &frame, &msg);\n    assert (number1 == 123);\n    assert (number2 == 123 * 123);\n    assert (number4 == 123 * 123 * 123);\n    assert (number8 == 123 * 123 * 123 * 123);\n    assert (streq (string, "Hello, World"));\n    assert (streq (longstr, "Goodbye cruel World!"));\n    assert (pointer == original);\n    zstr_free (&longstr);\n    zchunk_destroy (&chunk);\n    zframe_destroy (&frame);\n    zmsg_destroy (&msg);\n\n#ifdef ZMQ_STREAM\n    zsock_t *streamrecv = zsock_new(ZMQ_STREAM);\n    assert (streamrecv);\n    port = zsock_bind(streamrecv, "tcp://*:*");\n    assert(port > 0);\n\n    zsock_t *streamsender = zsock_new(ZMQ_STREAM);\n    assert (streamsender);\n    rc = zsock_connect(streamsender, "tcp://127.0.0.1:%i", port);\n    assert (rc == 0);\n\n    zmsg_t *connectmsg = zmsg_recv(streamrecv); // receiver gets a message on connect\n    zframe_t *id = zmsg_pop(connectmsg);        // first frame is id\n    assert (id);\n    assert (zframe_size(id) == 5);              // the id is 5 bytes, does this always hold?\n    //byte *vid = zframe_data(id);                // access to the actual id value\n    zframe_t *empty = zmsg_pop(connectmsg);     // second frame is empty\n    assert (empty);\n    assert (zframe_size(empty) == 0);\n    zframe_destroy(&id);\n    zframe_destroy(&empty);\n    zmsg_destroy(&connectmsg);\n\n    zmsg_t *connectmsg2 = zmsg_recv(streamsender); // sender also receives a message on connect\n    zframe_t *id2 = zmsg_pop(connectmsg2);         // first frame is id\n    assert (id2);\n    assert (zframe_size(id2) == 5);                // the id is 5 bytes\n    //byte *vid2 = zframe_data(id2);                 // access to the actual id value\n    zframe_t *empty2 = zmsg_pop(connectmsg2);      // second frame is empty\n    assert (empty2);\n    assert (zframe_size(empty2) == 0);\n    zframe_destroy(&id2);\n    zframe_destroy(&empty2);\n    zmsg_destroy(&connectmsg2);\n\n    // send a http request, first get our id\n    uint8_t rid [256];\n    size_t rid_size = 256;\n    rc = zmq_getsockopt (zsock_resolve(streamsender), ZMQ_ROUTING_ID, rid, &rid_size);\n    assert (rc == 0);\n    zmsg_t *request = zmsg_new();\n    assert (request);\n    rc = zmsg_addmem(request, rid, rid_size);\n    assert (rc == 0);\n    rc = zmsg_addstr(request, "GET /\\n\\n");\n    assert (rc == 0);\n    rc = zmsg_send(&request, streamsender);\n    assert (rc == 0);\n\n    // receive the request\n    zmsg_t * recvreq = zmsg_recv(streamrecv);\n    assert (recvreq);\n    zframe_t* ridframe = zmsg_pop(recvreq); // first frame is routing id\n    byte *rid2 = (byte *)zframe_data(ridframe);\n    assert (zframe_size(ridframe) == 5);\n    char *httpreq = zmsg_popstr(recvreq);\n    assert (streq(httpreq, "GET /\\n\\n"));\n    zstr_free (&httpreq);\n    zmsg_destroy(&recvreq);\n\n    //  Send reply back to client\n    char http_response[] = "HTTP/1.0 200 OK\\r\\n"\n                           "Content-Type: text/plain\\r\\n"\n                           "\\r\\n"\n                           "Hello, World!";\n    rc = zsock_send(streamrecv, "bs", rid2, 5, http_response);\n    assert (rc == 0);\n\n    // send empty frame to signal disconnect\n    rc = zsock_send(streamrecv, "bz", rid2, 5);\n    assert (rc == 0);\n    zframe_destroy(&ridframe);\n\n    // Get reply at sender and check that it\'s complete\n    zmsg_t *httpmsg = zmsg_recv(streamsender);\n    assert (httpmsg);\n    zframe_t *httpid = zmsg_pop(httpmsg);\n    assert (zframe_size(httpid) == 5);\n    char *httpresp = zmsg_popstr(httpmsg);\n    assert (httpresp);\n    assert ( streq(httpresp, http_response) );\n    zframe_destroy(&httpid);\n    zmsg_destroy(&httpmsg);\n    zstr_free(&httpresp);\n\n    // streamrecv sent disconnect so we should have received an empty frame\n    zmsg_t *disconnectmsg = zmsg_recv(streamrecv); // sender also receives a message on connect\n    zframe_t *id3 = zmsg_pop(disconnectmsg);         // first frame is id\n    assert (id3);\n    assert (zframe_size(id3) == 5);                // the id is 5 bytes\n    zframe_t *empty3 = zmsg_pop(disconnectmsg);      // second frame is empty\n    assert (empty3);\n    assert (zframe_size(empty3) == 0);\n    zframe_destroy(&id3);\n    zframe_destroy(&empty3);\n    zmsg_destroy(&disconnectmsg);\n\n    zsock_destroy(&streamsender);\n    zsock_destroy(&streamrecv);\n\n#endif\n\n#ifdef ZMQ_SERVER\n\n    //  Test zsock_bsend/brecv pictures with binary encoding on SERVER and CLIENT sockets\n    zsock_t *server = zsock_new (ZMQ_SERVER);\n    assert (server);\n    port = zsock_bind (server, "tcp://127.0.0.1:*");\n    assert (port != -1);\n    zsock_t* client = zsock_new (ZMQ_CLIENT);\n    assert (client);\n    rc = zsock_connect (client, "tcp://127.0.0.1:%d", port);\n    assert (rc != -1);\n\n    //  From client to server\n    chunk = zchunk_new ("World", 5);\n    zsock_bsend (client, "1248sSpc",\n                 number1, number2, number4, number8,\n                 "Hello, World",\n                 "Goodbye cruel World!",\n                 original,\n                 chunk);\n    zchunk_destroy (&chunk);\n\n    number8 = number4 = number2 = number1 = 0;\n    zsock_brecv (server, "1248sSpc",\n                 &number1, &number2, &number4, &number8,\n                 &string, &longstr,\n                 &pointer,\n                 &chunk);\n    assert (number1 == 123);\n    assert (number2 == 123 * 123);\n    assert (number4 == 123 * 123 * 123);\n    assert (number8 == 123 * 123 * 123 * 123);\n    assert (streq (string, "Hello, World"));\n    assert (streq (longstr, "Goodbye cruel World!"));\n    assert (pointer == original);\n    assert (zsock_routing_id (server));\n    zstr_free (&longstr);\n    zchunk_destroy (&chunk);\n\n    //  From server to client\n    chunk = zchunk_new ("World", 5);\n    zsock_bsend (server, "1248sSpc",\n                 number1, number2, number4, number8,\n                 "Hello, World",\n                 "Goodbye cruel World!",\n                 original,\n                 chunk);\n    zchunk_destroy (&chunk);\n\n    number8 = number4 = number2 = number1 = 0;\n    zsock_brecv (client, "1248sSpc",\n                 &number1, &number2, &number4, &number8,\n                 &string, &longstr,\n                 &pointer,\n                 &chunk);\n    assert (number1 == 123);\n    assert (number2 == 123 * 123);\n    assert (number4 == 123 * 123 * 123);\n    assert (number8 == 123 * 123 * 123 * 123);\n    assert (streq (string, "Hello, World"));\n    assert (streq (longstr, "Goodbye cruel World!"));\n    assert (pointer == original);\n    assert (zsock_routing_id (client) == 0);\n    zstr_free (&longstr);\n    zchunk_destroy (&chunk);\n\n    zsock_destroy (&client);\n    zsock_destroy (&server);\n\n#else\n    errno = 0;\n    zsock_t* server = zsock_new_server (NULL);\n    assert(server == NULL);\n    assert(errno == ENOTSUP);\n\n    errno = 0;\n    zsock_t* client = zsock_new_client (NULL);\n    assert(client == NULL);\n    assert(errno == ENOTSUP);\n#endif\n\n#ifdef ZMQ_SCATTER\n\n    zsock_t* gather = zsock_new_gather ("inproc://test-gather-scatter");\n    assert (gather);\n    zsock_t* scatter = zsock_new_scatter ("inproc://test-gather-scatter");\n    assert (scatter);\n\n    rc = zstr_send (scatter, "HELLO");\n    assert (rc == 0);\n\n    char* message;\n    message = zstr_recv (gather);\n    assert (streq(message, "HELLO"));\n    zstr_free (&message);    \n\n    zsock_destroy (&gather);\n    zsock_destroy (&scatter);\n#else\n    errno = 0;\n    zsock_t* scatter = zsock_new_scatter (NULL);\n    assert(scatter == NULL);\n    assert(errno == ENOTSUP);\n\n    errno = 0;\n    zsock_t* gather = zsock_new_gather (NULL);\n    assert(gather == NULL);\n    assert(errno == ENOTSUP);\n#endif\n\n#ifndef ZMQ_RADIO\n    errno = 0;\n    zsock_t* radio = zsock_new_radio (NULL);\n    assert(radio == NULL);\n    assert(errno == ENOTSUP);\n\n    errno = 0;\n    zsock_t* dish = zsock_new_dish (NULL);\n    assert(dish == NULL);\n    assert(errno == ENOTSUP);\n\n    errno = 0;\n    zsock_t* sock = zsock_new_req (NULL); // any supported socket type\n    rc = zsock_join (sock, "group1");\n    assert(rc == -1);\n    assert(errno == ENOTSUP);\n    errno = 0;\n    rc = zsock_leave (sock, "group1");\n    assert(rc == -1);\n    assert(errno == ENOTSUP);\n    zsock_destroy (&sock);\n#endif\n\n    //  Check that we can send a zproto format message\n    zsock_bsend (writer, "1111sS4", 0xAA, 0xA0, 0x02, 0x01, "key", "value", 1234);\n    zgossip_msg_t *gossip = zgossip_msg_new ();\n    zgossip_msg_recv (gossip, reader);\n    assert (zgossip_msg_id (gossip) == ZGOSSIP_MSG_PUBLISH);\n    zgossip_msg_destroy (&gossip);\n\n    zsock_destroy (&reader);\n    zsock_destroy (&writer);\n\n#ifdef ZMQ_DGRAM\n    // ZMQ_DGRAM ipv4 unicast test\n    zsock_t* dgramr = zsock_new_dgram ("udp://*:7777");\n    assert (dgramr);\n    assert ( streq( "DGRAM", zsys_sockname(ZMQ_DGRAM)) );\n    //zsock_t* dgrams = zsock_new_dgram ("udp://*:*");\n    zsock_t* dgrams = zsock_new (ZMQ_DGRAM);\n    zsock_bind (dgrams, "udp://127.0.0.1:7778" );\n\n    assert (dgrams);\n    // perhaps sleep a little for sockets to setup?\n\n    rc = zstr_sendm( dgrams, "127.0.0.1:7777" );\n    assert (rc == 0);\n    rc = zstr_send (dgrams, "HELLO");\n    assert (rc == 0);\n\n    char *dmessage, *addr;\n\n    zmsg_t *dmsg = zmsg_recv( dgramr );\n    assert (dmsg);\n    addr = zmsg_popstr (dmsg);\n    dmessage = zmsg_popstr (dmsg);\n    assert (streq(dmessage, "HELLO"));\n    zmsg_destroy ( &dmsg );\n    zsock_destroy (&dgrams);\n    zsock_destroy (&dgramr);\n    zstr_free (&dmessage);\n    zstr_free (&addr);\n    zstr_free (&message);\n\n#ifndef __APPLE__\n    // ZMQ_DGRAM ipv4 multicast test\n    zsock_t* mdgramr = zsock_new_dgram ("udp://225.25.25.25:7777");\n    assert (mdgramr);\n    zsock_t* mdgrams = zsock_new_dgram ("udp://*:*");\n    assert (mdgrams);\n\n    rc = zstr_sendm( mdgrams, "225.25.25.25:7777" );\n    assert (rc == 0);\n    rc = zstr_send (mdgrams, "HELLO");\n    assert (rc == 0);\n\n    char *mdmessage, *maddr;\n\n    // this call blocks forever on MacOS\n    zmsg_t *mdmsg = zmsg_recv( mdgramr );\n    assert (mdmsg);\n    maddr = zmsg_popstr (mdmsg);\n    mdmessage = zmsg_popstr (mdmsg);\n    assert (streq(mdmessage, "HELLO"));\n    zmsg_destroy ( &mdmsg );\n    zsock_destroy (&mdgrams);\n    zsock_destroy (&mdgramr);\n    zstr_free (&mdmessage);\n    zstr_free (&maddr);\n    zstr_free (&mdmessage);\n#endif\n\n//    // ipv6 (not supported yet)\n//    zsys_set_ipv6(1);\n//    zsock_t* dgramr6 = zsock_new_dgram ("udp://*:7777");\n//    assert (dgramr6);\n//    zsock_t* dgrams6 = zsock_new_dgram ("udp://*:*");\n//    assert (dgrams6);\n//    // perhaps sleep a little for sockets to setup?\n//    zclock_sleep(100);\n\n//    rc = zstr_sendm( dgrams6, "::1:7777" );\n//    assert (rc == 0);\n//    rc = zstr_send (dgrams6, "HELLO");\n//    assert (rc == 0);\n\n//    char *dmessage6, *addr6;\n//    zmsg_t *dmsg6 = zmsg_recv( dgramr6 );\n//    assert (dmsg6);\n//    addr6 = zmsg_popstr (dmsg6);\n//    dmessage6 = zmsg_popstr(dmsg6);\n//    assert (streq(dmessage6, "HELLO"));\n//    zmsg_destroy( &dmsg6 );\n//    zsock_destroy (&dgrams6);\n//    zsock_destroy (&dgramr6);\n\n//    zstr_free (&dmessage6);\n//    zstr_free (&addr6);\n//    zstr_free (&dmessage6);\n#endif\n\n    //  @end\n    printf ("OK\\n");\n    zsock_option_test (verbose);\n\n#if defined (__WINDOWS__)\n    zsys_shutdown();\n#endif\n}\n', 'message': 'Problem: vsnprintf use for endpoints is clumsy\n\nSolution: use zsys_sprintf and remove size restriction on endpoints.\n\nNote that _attach still uses fixed 256 limit; this should change at\nsome point to use zrex and no fixed size limit on endpoints.\n', 'author_name': 'Pieter Hintjens', 'lang': 'c'}