zmqd

Github Page

zmqd

∅MQD – a thin wrapper around the low-level C API of the ∅MQ messaging framework.

Most functions in this module have a one-to-one relationship with functions in the underlying C API. Some adaptations have been made to make the API safer, easier and more pleasant to use; most importantly:

  • Errors are signalled by means of exceptions rather than return codes. In particular, the ZmqException class provides a standard textual message for any error condition, but it also provides access to the errno code set by the C function that reported the error.
  • Functions are appropriately marked with @safe, pure and nothrow, thus facilitating their use in high-level D code.
  • Memory and resources (i.e. contexts, sockets and messages) are automatically managed, thus preventing leaks.
  • Context, socket and message options are implemented as properties.
The names of functions and types in ∅MQD are very similar to those in ∅MQ, but they follow the D naming conventions. Thus, the library should feel both familiar to ∅MQ users and natural to D users. A notable deviation from the C API is that message parts are consistently called "frames". For example, zmq_msg_send() becomes  zmqd.Frame.send() and so on. (Multipart messages were a late addition to ∅MQ, and the "msg" function names were well established at that point. The library's developers have admitted that this is somewhat confusing, and the newer CZMQ API consistently uses "frame" in function names.)

Due to the close correspondence with the C API, this documentation has intentionally been kept sparse. There is really no reason to repeat the contents of the ∅MQ reference manual here. Instead, the documentation for each function contains a "Corresponds to" section that links to the appropriate pages in the ∅MQ reference. Any details given in the present documentation mostly concern the D-specific adaptations that have been made.

Also note that the examples generally only use the INPROC transport. The reason for this is that the examples double as unittests, and we want to avoid firewall troubles and other issues that could arise with the use of network protocols such as TCP, PGM, etc., and the IPC protocol is not supported on Windows. Anyway, they are only short snippets that demonstrate the syntax; for more comprehensive and realistic examples, please refer to the ∅MQ Guide. Many of the examples in the Guide have been translated to D, and can be found in the examples subdirectory of the ∅MQD source repository.

Version
1.0.0 (compatible with ∅MQ >= 4.0.0)
Authors
Lars T. Kyllingstad
License
∅MQD is released under the terms of the Mozilla Public License v. 2.0.
Please refer to the ∅MQ site for details about ∅MQ licensing.

nothrow @safe Tuple!(int, "major", int, "minor", int, "patch")  zmqVersion();

Reports the ∅MQ library version.

Returns
A std.typecons.Tuple with three integer fields that represent the three versioning levels: major, minor and patch.
Corresponds to:
zmq_version()

enum  SocketType: int;

The various socket types.

These are described in the zmq_socket() reference.

 req

Corresponds to ZMQ_REQ

 rep

Corresponds to ZMQ_REP

 dealer

Corresponds to ZMQ_DEALER

 router

Corresponds to ZMQ_ROUTER

 pub

Corresponds to ZMQ_PUB

 sub

Corresponds to ZMQ_SUB

 xpub

Corresponds to ZMQ_XPUB

 xsub

Corresponds to ZMQ_XSUB

 push

Corresponds to ZMQ_PUSH

 pull

Corresponds to ZMQ_PULL

 pair

Corresponds to ZMQ_PAIR

 stream

Corresponds to ZMQ_STREAM

enum  Security: int;

 Security mechanisms.

 none

NULL: No security or confidentiality.

 plain

PLAIN: Clear-text authentication.

 curve

CURVE: Secure authentication and confidentiality.

struct  Socket;

An object that encapsulates a ∅MQ socket.

A default-initialized  Socket is not a valid ∅MQ socket; it must always be explicitly initialized with a constructor (see Socket.this()): 

Socket s;                     // Not a valid socket yet
s = Socket(SocketType.push);  // ...but now it is.
This struct is noncopyable, which means that a socket is always uniquely managed by a single  Socket object. Functions that will inspect or use the socket, but not take ownership of it, should take a ref  Socket parameter. Use std.algorithm.move to move a  Socket to a different location (e.g. into a sink function that takes it by value, or into a new variable).

The socket is automatically closed when the  Socket object goes out of scope.

Linger period:
Note that  Socket by default sets the socket's linger period to zero. This deviates from the ∅MQ default (which is an infinite linger period).

@safe this(SocketType type);
@safe this(Context context, SocketType type);

Creates a new ∅MQ socket.

If context is not specified, the default context (as returned by defaultContext()) is used.

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_socket()
Examples
With default context: 
auto sck = Socket(SocketType.push);
assert (sck.initialized);
Examples
With explicit context: 
auto ctx = Context();
auto sck = Socket(ctx, SocketType.push);
assert (sck.initialized);

@safe void  close();

Closes the ∅MQ socket.

Note that the socket will be closed automatically upon destruction, so it is usually not necessary to call this method manually.

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_close()
Examples
auto s = Socket(SocketType.pair);
assert (s.initialized);
s.close();
assert (!s.initialized);

@safe void  bind(const char[] endpoint);

Starts accepting incoming connections on endpoint.

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_bind()
Examples
auto s = Socket(SocketType.pub);
s.bind("inproc://zmqd_bind_example");

@safe void  unbind(const char[] endpoint);

Stops accepting incoming connections on endpoint.

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_unbind()
Examples
auto s = Socket(SocketType.pub);
s.bind("ipc://zmqd_unbind_example");
// Do some work...
s.unbind("ipc://zmqd_unbind_example");

@safe void  connect(const char[] endpoint);

Creates an outgoing connection to endpoint.

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_connect()
Examples
auto s = Socket(SocketType.sub);
s.connect("inproc://zmqd_connect_example");

@safe void  disconnect(const char[] endpoint);

Disconnects the socket from endpoint.

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_disconnect()
Examples
auto s = Socket(SocketType.sub);
s.connect("inproc://zmqd_disconnect_example");
// Do some work...
s.disconnect("inproc://zmqd_disconnect_example");

@safe void  send(const ubyte[] data, bool more = false);
@safe void  send(const char[] data, bool more = false);
@safe bool  trySend(const ubyte[] data, bool more = false);
@safe bool  trySend(const char[] data, bool more = false);

Sends a message frame.

send blocks until the frame has been queued on the socket. trySend performs the operation in non-blocking mode, and returns a bool value that signifies whether the frame was queued on the socket.

The more parameter specifies whether this is a multipart message and there are more frames to follow.

The char[] overload is a convenience function that simply casts the string argument to ubyte[].

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_send() (with the ZMQ_DONTWAIT flag, in the case of trySend, and with the ZMQ_SNDMORE flag if more == true).
Examples
auto sck = Socket(SocketType.pub);
sck.send(cast(ubyte[]) [11, 226, 92]);
sck.send("Hello World!");

@safe void  send(ref Frame msg, bool more = false);
@safe bool  trySend(ref Frame msg, bool more = false);

Sends a message frame.

send blocks until the frame has been queued on the socket. trySend performs the operation in non-blocking mode, and returns a bool value that signifies whether the frame was queued on the socket.

The more parameter specifies whether this is a multipart message and there are more frames to follow.

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_msg_send() (with the ZMQ_DONTWAIT flag, in the case of trySend, and with the ZMQ_SNDMORE flag if more == true).
Examples
auto sck = Socket(SocketType.pub);
auto msg = Frame(12);
msg.data.asString()[] = "Hello World!";
sck.send(msg);

@safe void  sendConst(immutable ubyte[] data, bool more = false);
@safe void  sendConst(string data, bool more = false);
@safe bool  trySendConst(immutable ubyte[] data, bool more = false);
@safe bool  trySendConst(string data, bool more = false);

Sends a constant-memory message frame.

sendConst blocks until the frame has been queued on the socket. trySendConst performs the operation in non-blocking mode, and returns a bool value that signifies whether the frame was queued on the socket.

The more parameter specifies whether this is a multipart message and there are more frames to follow.

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_send_const() (with the ZMQ_DONTWAIT flag, in the case of trySend, and with the ZMQ_SNDMORE flag if more == true).
Examples
static immutable arr = cast(ubyte[]) [11, 226, 92];
auto sck = Socket(SocketType.pub);
sck.sendConst(arr);
sck.sendConst("Hello World!");

@safe size_t  receive(ubyte[] data);
@safe Tuple!(size_t, bool)  tryReceive(ubyte[] data);

Receives a message frame.

receive blocks until the request can be satisfied, and returns the number of bytes in the frame. tryReceive performs the operation in non-blocking mode, and returns a std.typecons.Tuple which contains the size of the frame along with a bool value that signifies whether a frame was received. (If the latter is false, the former is always set to zero.)

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_recv() (with the ZMQ_DONTWAIT flag, in the case of tryReceive).
Examples
// Sender
auto snd = Socket(SocketType.req);
snd.connect("inproc://zmqd_receive_example");
snd.send("Hello World!");

// Receiver
import std.string: representation;
auto rcv = Socket(SocketType.rep);
rcv.bind("inproc://zmqd_receive_example");
char[256] buf;
immutable len  = rcv.receive(buf.representation);
assert (buf[0 .. len] == "Hello World!");

@safe size_t  receive(ref Frame msg);
@safe Tuple!(size_t, bool)  tryReceive(ref Frame msg);

Receives a message frame.

receive blocks until the request can be satisfied, and returns the number of bytes in the frame. tryReceive performs the operation in non-blocking mode, and returns a std.typecons.Tuple which contains the size of the frame along with a bool value that signifies whether a frame was received. (If the latter is false, the former is always set to zero.)

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_msg_recv() (with the ZMQ_DONTWAIT flag, in the case of tryReceive).
Examples
// Sender
auto snd = Socket(SocketType.req);
snd.connect("inproc://zmqd_msg_receive_example");
snd.send("Hello World!");

// Receiver
import std.string: representation;
auto rcv = Socket(SocketType.rep);
rcv.bind("inproc://zmqd_msg_receive_example");
auto msg = Frame();
rcv.receive(msg);
assert (msg.data.asString() == "Hello World!");

@property @safe SocketType  type();

The socket type.

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_getsockopt() with ZMQ_TYPE.
Examples
auto sck = Socket(SocketType.xpub);
assert (sck.type == SocketType.xpub);

@property @safe bool  more();

Whether there are more message frames to follow.

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_getsockopt() with ZMQ_RCVMORE.

@property @safe int  sendHWM();
@property @safe void  sendHWM(int value);
@property @safe int  receiveHWM();
@property @safe void  receiveHWM(int value);
@property @safe ulong  threadAffinity();
@property @safe void  threadAffinity(ulong value);
@property @safe ubyte[]  identity();
@safe ubyte[]  getIdentity(ubyte[] dest);
@property @safe void  identity(const ubyte[] value);
@property @safe void  identity(const char[] value);
@property @safe int  rate();
@property @safe void  rate(int value);
@property @safe Duration  recoveryInterval();
@property @safe void  recoveryInterval(Duration value);
@property @safe int  sendBufferSize();
@property @safe void  sendBufferSize(int value);
@property @safe int  receiveBufferSize();
@property @safe void  receiveBufferSize(int value);
@property @safe Duration  linger();
@property @safe void  linger(Duration value);
@property @safe Duration  reconnectionInterval();
@property @safe void  reconnectionInterval(Duration value);
@property @safe Duration  maxReconnectionInterval();
@property @safe void  maxReconnectionInterval(Duration value);
@property @safe int  backlog();
@property @safe void  backlog(int value);
@property @safe long  maxMsgSize();
@property @safe void  maxMsgSize(long value);
@property @safe int  multicastHops();
@property @safe void  multicastHops(int value);
@property @safe Duration  receiveTimeout();
@property @safe void  receiveTimeout(Duration value);
@property @safe Duration  sendTimeout();
@property @safe void  sendTimeout(Duration value);
@property @safe bool  ipv6();
@property @safe void  ipv6(bool value);
deprecated @property @safe bool  ipv4Only();
deprecated @property @safe void  ipv4Only(bool value);
@property @safe bool  immediate();
@property @safe void  immediate(bool value);
deprecated @property @safe bool  delayAttachOnConnect();
deprecated @property @safe void  delayAttachOnConnect(bool value);
@property @safe FD  fd();
@property @safe PollFlags  events();
@property @trusted char[]  lastEndpoint();
@property @safe void  routerMandatory(bool value);
@property @safe void  probeRouter(bool value);
@property @safe void  xpubVerbose(bool value);
@property @safe void  reqCorrelate(bool value);
@property @safe void  reqRelaxed(bool value);
@property @safe int  tcpKeepalive();
@property @safe void  tcpKeepalive(int value);
@property @safe int  tcpKeepaliveIdle();
@property @safe void  tcpKeepaliveIdle(int value);
@property @safe int  tcpKeepaliveCnt();
@property @safe void  tcpKeepaliveCnt(int value);
@property @safe int  tcpKeepaliveIntvl();
@property @safe void  tcpKeepaliveIntvl(int value);
@property @safe Security  mechanism();
@property @safe bool  plainServer();
@property @safe void  plainServer(bool value);
@safe char[]  getPlainUsername(char[] dest);
@property @safe void  plainUsername(const(char)[] value);
@safe char[]  getPlainPassword(char[] dest);
@property @safe void  plainPassword(const(char)[] value);
@property @safe bool  curveServer();
@property @safe void  curveServer(bool value);
@property @safe ubyte[]  curvePublicKey();
@safe ubyte[]  getCurvePublicKey(ubyte[] dest);
@property @safe char[]  curvePublicKeyZ85();
@safe char[]  getCurvePublicKeyZ85(char[] dest);
@property @safe void  curvePublicKey(const(ubyte)[] value);
@property @safe void  curvePublicKeyZ85(const(char)[] value);
@property @safe ubyte[]  curveSecretKey();
@safe ubyte[]  getCurveSecretKey(ubyte[] dest);
@property @safe char[]  curveSecretKeyZ85();
@safe char[]  getCurveSecretKeyZ85(char[] dest);
@property @safe void  curveSecretKey(const(ubyte)[] value);
@property @safe void  curveSecretKeyZ85(const(char)[] value);
@property @safe ubyte[]  curveServerKey();
@safe ubyte[]  getCurveServerKey(ubyte[] dest);
@property @safe char[]  curveServerKeyZ85();
@safe char[]  getCurveServerKeyZ85(char[] dest);
@property @safe void  curveServerKey(const(ubyte)[] value);
@property @safe void  curveServerKeyZ85(const(char)[] value);
@property @safe char[]  zapDomain();
@safe char[]  getZapDomain(char[] dest);
@property @safe void  zapDomain(const char[] value);
@property @safe void  conflate(bool value);

Misc. socket options.

Each of these has a one-to-one correspondence with an option passed to zmq_getsockopt() and zmq_setsockopt(). For example, identity corresponds to ZMQ_IDENTITY, receiveBufferSize corresponds to ZMQ_RCVBUF, etc.

Notes:
  • For convenience, the setter for the identity property accepts strings. To retrieve a string with the getter, use the asString() function. 
    sck.identity = "foobar";
assert (sck.identity.asString() == "foobar");
  • The linger, receiveTimeout and sendTimeout properties may have the special value core.time.Duration.max, which in this context specifies an infinite duration. This is translated to an option value of -1 in the C API (and it is also the default value for all of them).
  • Some options have array type, and these allow the user to supply a buffer in which to store the value, to avoid a GC allocation. The return value is then a slice of this buffer. These are not marked as @property, but are prefixed with "get" (e.g. getIdentity()). A user-supplied buffer is required for some options, namely getPlainUsername() and getPlainPassword(), and these do not have @property versions. getCurveXxxKey() and getCurveXxxKeyZ85() require buffers which are at least 32 and 41 bytes long, respectively.
  • The ZMQ_SUBSCRIBE and ZMQ_UNSUBSCRIBE options are treated differently from the others; see Socket.subscribe() and Socket.unsubscribe()
Throws
ZmqException if ∅MQ reports an error.
std.conv.ConvOverflowException if a given Duration is longer than the number of milliseconds that will fit in an int (only applies to properties of core.time.Duration type).
core.exception.RangeError if the dest buffers passed to getCurveXxxKey() or getCurveXxxKeyZ85() are less than 32 or 41 bytes long, respectively.
Corresponds to:
zmq_getsockopt() and zmq_setsockopt().

@safe void  subscribe(const ubyte[] filterPrefix);
@safe void  subscribe(const char[] filterPrefix);

Establishes a message filter.

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_msg_setsockopt() with ZMQ_SUBSCRIBE.
Examples
// Create a subscriber that accepts all messages that start with
// the prefixes "foo" or "bar".
auto sck = Socket(SocketType.sub);
sck.subscribe("foo");
sck.subscribe("bar");

@safe void  unsubscribe(const ubyte[] filterPrefix);
@safe void  unsubscribe(const char[] filterPrefix);

Removes a message filter.

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_msg_setsockopt() with ZMQ_SUBSCRIBE.
Examples
// Subscribe to messages that start with "foo" or "bar".
auto sck = Socket(SocketType.sub);
sck.subscribe("foo");
sck.subscribe("bar");
// ...
// From now on, only accept messages that start with "bar"
sck.unsubscribe("foo");

@safe void  monitor(const char[] endpoint, EventType events = EventType.all);

Spawns a PAIR socket that publishes socket state changes (events) over the INPROC transport to the given endpoint.

Which event types should be published may be selected by bitwise-ORing together different EventType flags in the event parameter.

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_socket_monitor()
See Also
receiveEvent(), which receives and parses event messages.
Examples
auto sck = Socket(SocketType.pub);
sck.monitor("inproc://zmqd_monitor_unittest",
            EventType.accepted | EventType.closed);

inout pure nothrow @property @safe inout(void)*  handle();

The void* pointer used by the underlying C API to refer to the socket.

If the object has not been initialized, this function returns null.

const pure nothrow @property @safe bool  initialized();

Whether this Socket object has been initialized, i.e. whether it refers to a valid ∅MQ socket.

Examples
Socket sck;
assert (!sck.initialized);
sck = Socket(SocketType.sub);
assert (sck.initialized);
sck.close();
assert (!sck.initialized);

alias  FD = int;

The native socket file descriptor type.

This is an alias for SOCKET on Windows and int on POSIX systems.

@safe void  proxy(ref Socket frontend, ref Socket backend);
@safe void  proxy(ref Socket frontend, ref Socket backend, ref Socket capture);

Starts the built-in ∅MQ proxy.

This function never returns normally, but it may throw an exception. This could happen if the context associated with either of the specified sockets is manually destroyed in a different thread.

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_proxy()

@trusted uint  poll(PollItem[] items, Duration timeout = Duration.max);

Input/output multiplexing.

The timeout parameter may have the special value core.time.Duration.max, which in this context specifies an infinite duration. This is translated to an argument value of -1 in the C API.

Returns
The number of PollItem structures with events signalled in PollItem.returnedEvents, or 0 if no events have been signalled.
Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_poll()
Examples
auto socket1 = zmqd.Socket(zmqd.SocketType.pull);
socket1.bind("inproc://zmqd_poll_example");

import std.socket;
auto socket2 = new std.socket.Socket(
    AddressFamily.INET,
    std.socket.SocketType.DGRAM);
socket2.bind(new InternetAddress(InternetAddress.ADDR_ANY, 5678));

auto socket3 = zmqd.Socket(zmqd.SocketType.push);
socket3.connect("inproc://zmqd_poll_example");
socket3.send("test");

import core.thread: Thread;
Thread.sleep(10.msecs);

auto items = [
    PollItem(socket1, PollFlags.pollIn),
    PollItem(socket2, PollFlags.pollIn | PollFlags.pollOut),
    PollItem(socket3, PollFlags.pollIn),
];

const n = poll(items, 100.msecs);
assert (n == 2);
assert (items[0].returnedEvents == PollFlags.pollIn);
assert (items[1].returnedEvents == PollFlags.pollOut);
assert (items[2].returnedEvents == 0);
socket2.close();

enum  PollFlags: int;

poll() event flags.

These are described in the zmq_poll() manual.

 pollIn

Corresponds to ZMQ_POLLIN

 pollOut

Corresponds to ZMQ_POLLOUT

 pollErr

Corresponds to ZMQ_POLLERR

struct  PollItem;

A structure that specifies a socket to be monitored by poll() as well as the events to poll for, and, when poll() returns, the events that occurred.

Warning:
 PollItem objects do not store std.socket.Socket references, only the corresponding native file descriptors. This means that the references have to be stored elsewhere, or the objects may be garbage collected, invalidating the sockets before or while poll() executes. 
// Not OK
auto p1 = PollItem(new std.socket.Socket(/*...*/), PollFlags.pollIn);

// OK
auto s = new std.socket.Socket(/*...*/);
auto p2 = PollItem(s, PollFlags.pollIn);
Corresponds to:
zmq_pollitem_t

nothrow @safe this(ref zmqd.Socket socket, PollFlags events);

Constructs a PollItem for monitoring a ∅MQ socket.

@system this(std.socket.Socket socket, PollFlags events);

Constructs a PollItem for monitoring a standard socket referenced by a std.socket.Socket.

pure nothrow @safe this(FD fd, PollFlags events);

Constructs a PollItem for monitoring a standard socket referenced by a native file descriptor.

pure nothrow @property @safe void  requestedEvents(PollFlags events);
const pure nothrow @property @safe PollFlags  requestedEvents();

Requested events.

Corresponds to:
zmq_pollitem_t.events

const pure nothrow @property @safe PollFlags  returnedEvents();

Returned events.

Corresponds to:
zmq_pollitem_t.revents

struct  Frame;

An object that encapsulates a ∅MQ message frame.

This struct is a wrapper around a zmq_msg_t object. A default-initialized  Frame is not a valid ∅MQ message frame; it should always be explicitly initialized upon construction using Frame.opCall(). Alternatively, it may be initialized later with Frame.rebuild(). 

Frame msg1;                 // Invalid frame
auto msg2 = Frame();        // Empty frame
auto msg3 = Frame(1024);    // 1K frame
msg1.rebuild(2048);         // msg1 now has size 2K
msg2.rebuild(2048);         // ...and so does msg2
When a  Frame goes out of scope, zmq_msg_close() is called on the underlying zmq_msg_t.

A  Frame cannot be copied by normal assignment; use Frame.copy() for this.

static @safe Frame  opCall();

Initializes an empty ∅MQ message frame.

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_msg_init()
Examples
auto msg = Frame();
assert(msg.size == 0);

static @safe Frame  opCall(size_t size);

Initializes a ∅MQ message frame of a specified size.

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_msg_init_size()
Examples
auto msg = Frame(123);
assert(msg.size == 123);

static @safe Frame  opCall(ubyte[] data);

Initializes a ∅MQ message frame from a supplied buffer.

Warning:
Some care must be taken when using this function, as ∅MQ expects to take full ownership of the supplied buffer. Client code should therefore avoid retaining any references to it, including slices that contain, overlap with or are contained in data. ∅MQ makes no guarantee that the buffer is not modified, and it does not specify when the buffer is released.

An additional complication is caused by the fact that most arrays in D are owned by the garbage collector. This is solved by adding the array pointer as a new garbage collector root before passing it to zmq_msg_init_data(), thus preventing the GC from collecting it. The root is then removed again in the deallocator callback function which is called by ∅MQ when it no longer requires the buffer, thus allowing the GC to collect it.
Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_msg_init_data()
Examples
auto buf = new ubyte[123];
auto msg = Frame(buf);
assert(msg.size == buf.length);
assert(msg.data.ptr == buf.ptr);

@safe void  rebuild();

Reinitializes the Frame object as an empty message.

This function will first call Frame.close() to release the resources associated with the message frame, and then it will initialize it anew, exactly as if it were constructed with Frame().

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_msg_close() followed by zmq_msg_init()
Examples
auto msg = Frame(256);
assert (msg.size == 256);
msg.rebuild();
assert (msg.size == 0);

@safe void  rebuild(size_t size);

Reinitializes the Frame object to a specified size.

This function will first call Frame.close() to release the resources associated with the message frame, and then it will initialize it anew, exactly as if it were constructed with Frame(size).

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_msg_close() followed by zmq_msg_init_size().
Examples
auto msg = Frame(256);
assert (msg.size == 256);
msg.rebuild(1024);
assert (msg.size == 1024);

@safe void  rebuild(ubyte[] data);

Reinitializes the Frame object from a supplied buffer.

This function will first call Frame.close() to release the resources associated with the message frame, and then it will initialize it anew, exactly as if it were constructed with Frame(data).

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_msg_close() followed by zmq_msg_init_data().
Examples
auto msg = Frame(256);
assert (msg.size == 256);
auto buf = new ubyte[123];
msg.rebuild(buf);
assert(msg.size == buf.length);
assert(msg.data.ptr == buf.ptr);

@safe void  close();

Releases the ∅MQ message frame.

Note that the frame will be automatically released when the Frame object is destroyed, so it is often not necessary to call this method manually.

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_msg_close()

@safe Frame  copy();
@safe void  copyTo(ref Frame dest);

Copies frame content to another message frame.

 copy() returns a new Frame object, while copyTo(dest) copies the contents of this Frame into dest. dest must be a valid (i.e. initialized) Frame.

Warning:
These functions may not do what you think they do. Please refer to the ∅MQ manual for details.
Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_msg_copy()
Examples
import std.string: representation;
auto msg1 = Frame(3);
msg1.data[] = "foo".representation;
auto msg2 = msg1.copy();
assert (msg2.data.asString() == "foo");

@safe Frame  move();
@safe void  moveTo(ref Frame dest);

Moves frame content to another message frame.

 move() returns a new Frame object, while moveTo(dest) moves the contents of this Frame to dest. dest must be a valid (i.e. initialized) Frame.

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_msg_move()
Examples
import std.string: representation;
auto msg1 = Frame(3);
msg1.data[] = "foo".representation;
auto msg2 = msg1.move();
assert (msg1.size == 0);
assert (msg2.data.asString() == "foo");

nothrow @property @safe size_t  size();

The message frame content  size in bytes.

Corresponds to:
zmq_msg_size()
Examples
auto msg = Frame(123);
assert(msg.size == 123);

nothrow @property @trusted ubyte[]  data();

Retrieves the message frame content.

Corresponds to:
zmq_msg_data()
Examples
import std.string: representation;
auto msg = Frame(3);
assert(msg.data.length == 3);
msg.data[] = "foo".representation; // Slice operator -> array copy.
assert(msg.data.asString() == "foo");

nothrow @property @safe bool  more();

Whether there are  more message frames to retrieve.

Corresponds to:
zmq_msg_more()

inout pure nothrow @property @safe inout(zmq_msg_t)*  handle();

A pointer to the underlying zmq_msg_t.

@trusted Context  defaultContext();

A global context which is used by default by all sockets, unless they are explicitly constructed with a different context.

The ∅MQ Guide has the following to say about context creation:

You should create and use exactly one context in your process. […] If at runtime a process has two contexts, these are like separate ∅MQ instances. If that's explicitly what you want, OK, but otherwise remember: Do one zmq_ctx_new() at the start of your main line code, and one zmq_ctx_destroy() at the end.
By using  defaultContext(), this is exactly what you achieve. The context is created the first time the function is called, and is automatically destroyed when the program ends.

This function is thread safe.

Throws
ZmqException if ∅MQ reports an error.
See Also
Context

struct  Context;

An object that encapsulates a ∅MQ context.

In most programs, it is not necessary to use this type directly, as Socket will use a default global context if not explicitly provided with one. See defaultContext() for details.

A default-initialized  Context is not a valid ∅MQ context; it must always be explicitly initialized with Context.opCall(): 

Context ctx;        // Not a valid context yet
ctx = Context();    // ...but now it is.
 Context objects can be passed around by value, and two copies will refer to the same context. The underlying context is managed using reference counting, so that when the last copy of a  Context goes out of scope, the context is automatically destroyed. The reference counting is performed in a thread safe manner, so that the same context can be shared between multiple threads. (∅MQ guarantees the thread safety of other context operations.)

See Also
defaultContext()

static @trusted Context  opCall();

Creates a new ∅MQ context.

Returns
A Context object that encapsulates the new context.
Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_ctx_new()
Examples
auto ctx = Context();
assert (ctx.initialized);

@safe void  detach();

Detaches from the ∅MQ context.

If this is the last reference to the context, it will be destroyed with zmq_ctx_destroy().

Throws
ZmqException if ∅MQ reports an error.
Examples
auto ctx = Context();
assert (ctx.initialized);
ctx.detach();
assert (!ctx.initialized);

@property @safe int  ioThreads();
@property @safe void  ioThreads(int value);

The number of I/O threads.

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_ctx_get() and zmq_ctx_set() with ZMQ_IO_THREADS.
Examples
auto ctx = Context();
ctx.ioThreads = 3;
assert (ctx.ioThreads == 3);

@property @safe int  maxSockets();
@property @safe void  maxSockets(int value);

The maximum number of sockets.

Throws
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_ctx_get() and zmq_ctx_set() with ZMQ_MAX_SOCKETS.
Examples
auto ctx = Context();
ctx.maxSockets = 512;
assert (ctx.maxSockets == 512);

inout pure nothrow @property @trusted inout(void)*  handle();

The void* pointer used by the underlying C API to refer to the context.

If the object has not been initialized, this function returns null.

const pure nothrow @property @safe bool  initialized();

Whether this Context object has been initialized, i.e. whether it refers to a valid ∅MQ context.

Examples
Context ctx;
assert (!ctx.initialized);
ctx = Context();
assert (ctx.initialized);
ctx.detach();
assert (!ctx.initialized);

enum  EventType: int;

Socket event types.

These are used together with Socket.monitor(), and are described in the zmq_socket_monitor() reference.

 connected

Corresponds to ZMQ_EVENT_CONNECTED.

 connectDelayed

Corresponds to ZMQ_EVENT_CONNECT_DELAYED.

 connectRetried

Corresponds to ZMQ_EVENT_CONNECT_RETRIED.

 listening

Corresponds to ZMQ_EVENT_LISTENING.

 bindFailed

Corresponds to ZMQ_EVENT_BIND_FAILED.

 accepted

Corresponds to ZMQ_EVENT_ACCEPTED.

 acceptFailed

Corresponds to ZMQ_EVENT_ACCEPT_FAILED.

 closed

Corresponds to ZMQ_EVENT_CLOSED.

 closeFailed

Corresponds to ZMQ_EVENT_CLOSE_FAILED.

 disconnected

Corresponds to ZMQ_EVENT_DISCONNECTED.

 all

Corresponds to ZMQ_EVENT_ALL.

@system Event  receiveEvent(ref Socket socket);

Receives a message on the given socket and interprets it as a socket state change event.

socket must be a PAIR socket which is connected to an endpoint created via a Socket.monitor() call.  receiveEvent() receives one message on the socket, parses its contents according to the specification in the zmq_socket_monitor() reference, and returns the event information as an Event object.

Throws
ZmqException if ∅MQ reports an error.
InvalidEventException if the received message could not be interpreted as an event message.
See Also
Socket.monitor(), for monitoring socket state changes.

struct  Event;

Information about a socket state change.

Corresponds to:
zmq_event_t
See Also
receiveEvent()

const pure nothrow @property @safe EventType  type();

The event  type.

Corresponds to:
zmq_event_t.event

const pure nothrow @property @safe string  address();

The peer  address.

Corresponds to:
zmq_event_t.data.xyz.addr, where xyz is the event-specific union.

const pure nothrow @property @safe FD  fd();

The socket file descriptor.

This property function may only be called if Event.type is one of: connected, listening, accepted, closed or disonnected.

Throws
Error if the property is called for a wrong event type.
Corresponds to:
zmq_event_t.data.xyz.addr, where xyz is the event-specific union.

const pure nothrow @property @safe int  errno();

The  errno code for the error which triggered the event.

This property function may only be called if Event.type is either bindFailed, acceptFailed or closeFailed.

Throws
Error if the property is called for a wrong event type.
Corresponds to:
zmq_event_t.data.xyz.addr, where xyz is the event-specific union.

const pure nothrow @property @safe Duration  interval();

The reconnect  interval.

This property function may only be called if Event.type is connectRetried.

Throws
Error if the property is called for a wrong event type.
Corresponds to:
zmq_event_t.data.connect_retried. interval

@safe char[]  z85Encode(ubyte[] data, char[] dest);
@safe char[]  z85Encode(ubyte[] data);

Encodes a binary key as Z85 printable text.

dest must be an array whose length is at least 5*data.length/4 + 1, which will be used to store the return value plus a terminating zero byte. If dest is omitted, a new array will be created.

Returns
An array of size 5*data.length/4 which contains the Z85-encoded text, excluding the terminating zero byte. This will be a slice of dest if it is provided.
Throws
core.exception.RangeError if dest is given but is too small.
ZmqException if ∅MQ reports an error (i.e., if data.length is not a multiple of 4).
Corresponds to:
zmq_z85_encode()

@safe ubyte[]  z85Decode(char[] text, ubyte[] dest);
@safe ubyte[]  z85Decode(char[] text);

Decodes a binary key from Z85 printable text.

dest must be an array whose length is at least 4*data.length/5, which will be used to store the return value. If dest is omitted, a new array will be created.

Note that zmq_z85_decode() expects a zero-terminated string, so a zero byte will be appended to text if it does not contain one already. However, this may trigger a (possibly unwanted) GC allocation. To avoid this, either make sure that the last character in text is '\0', or use assumeSafeAppend on the array before calling this function.

Returns
An array of size 4*data.length/5 which contains the decoded data. This will be a slice of dest if it is provided.
Throws
core.exception.RangeError if dest is given but is too small.
ZmqException if ∅MQ reports an error (i.e., if data.length is not a multiple of 5).
Corresponds to:
zmq_z85_decode()

@safe Tuple!(char[], "publicKey", char[], "secretKey")  curveKeyPair(char[] publicKeyBuf = null, char[] secretKeyBuf = null);

Generates a new CURVE key pair.

To avoid a memory allocation, preallocated buffers may optionally be supplied for the two keys. Each of these must have a length of at least 41 bytes, enough for a 40-character Z85-encoded key plus a terminating zero byte. If either buffer is omitted/null, a new one will be created.

Returns
A tuple that contains the two keys. Each of these will have a length of 40 characters, and will be slices of the input buffers if such have been provided.
Throws
core.exception.RangeError if publicKeyBuf or secretKeyBuf are not null but have a length of less than 41 characters.
ZmqException if ∅MQ reports an error.
Corresponds to:
zmq_curve_keypair()
Examples
auto server = Socket(SocketType.rep);
auto serverKeys = curveKeyPair();
server.curveServer = true;
server.curveSecretKeyZ85 = serverKeys.secretKey;
server.bind("inproc://curveKeyPair_test");

auto client = Socket(SocketType.req);
auto clientKeys = curveKeyPair();
client.curvePublicKeyZ85 = clientKeys.publicKey;
client.curveSecretKeyZ85 = clientKeys.secretKey;
client.curveServerKeyZ85 = serverKeys.publicKey;
client.connect("inproc://curveKeyPair_test");
client.send("hello");

ubyte[5] buf;
assert (server.receive(buf) == 5);
assert (buf.asString() == "hello");

pure @safe inout(char)[]  asString(inout(ubyte)[] data);

Utility function which interprets and validates a byte array as a UTF-8 string.

Most of ∅MQD's message API deals in ubyte[] arrays, but very often, the message data contains plain text.  asString() allows for easy and safe interpretation of raw data as characters. It checks that data is a valid UTF-8 encoded string, and returns a char[] array that refers to the same memory region.

Throws
std.utf.UTFException if data is not a valid UTF-8 string.
See Also
std.string.representation, which performs the opposite operation.
Examples
auto s1 = Socket(SocketType.pair);
s1.bind("inproc://zmqd_asString_example");
auto s2 = Socket(SocketType.pair);
s2.connect("inproc://zmqd_asString_example");

auto msg = Frame(12);
msg.data.asString()[] = "Hello World!";
s1.send(msg);

ubyte[12] buf;
s2.receive(buf);
assert(buf.asString() == "Hello World!");

class  ZmqException: object.Exception;

A class for exceptions thrown when any of the underlying ∅MQ C functions report an error.

The exception provides a standard error message obtained with zmq_strerror(), as well as the errno code set by the ∅MQ function which reported the error.

immutable int  errno;

The  errno code that was set by the ∅MQ function that reported the error.

Corresponds to:
zmq_errno()

class  InvalidEventException: object.Exception;

Exception thrown by receiveEvent() on failure to interpret a received message as an event description.