zmqd

A thin wrapper around the low-level C API of the ∅MQ messaging framework, for the D programming language.

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, namely: The names of functions and types in ∅MQD are very similar to those in ∅MQ, but they follow the D naming conventions. For example, zmq_msg_send() becomes zmqd.Message.send() and so on. Thus, the library should feel both familiar to ∅MQ users and natural to D users.

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 page 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 only use the INPROC and IPC transports. 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. Anyway, they are only short snippets that demonstrate the syntax; for more comprehensive and realistic examples, please refer to the ∅MQ Guide.

Version:

0.2 (∅MQ 3.2 compatible)

Authors:

Lars T. Kyllingstad

Licence:

∅MQD is released under a BSD licence (see LICENCE.txt for details).
Please refer to the ∅MQ site for details about ∅MQ licensing.

Members

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

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.

See also:

defaultContext()

Members

static @safe 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 destroy();

Destroys the ∅MQ context.

It is normally not necessary to do this manually, as the context will be destroyed automatically when the last reference to it goes out of scope.

Throws:

ZmqException if ∅MQ reports an error.

Corresponds to:

zmq_ctx_destroy()

Examples:

auto ctx = Context();
assert (ctx.initialized);
ctx.destroy();
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 @safe 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.destroy();
assert (!ctx.initialized);

@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

enum SocketType: int;

The various socket types.

These are described in the zmq_socket() reference.

Members

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

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.
Socket objects can be passed around by value, and two copies will refer to the same socket. The underlying socket is managed using reference counting, so that when the last copy of a Socket goes out of scope, the socket is automatically closed.

Members

this(SocketType type);
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 automatically closed when the last reference to it goes out of scope, so it is often 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("ipc://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("ipc://zmqd_disconnect_example");
// Do some work...
s.disconnect("ipc://zmqd_disconnect_example");

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

Sends a message part.

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

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).

Examples:

auto sck = Socket(SocketType.pub);
sck.send(cast(ubyte[]) [11, 226, 92]);
sck.send("Hello World!");

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

Sends a message part.

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

Throws:

ZmqException if ∅MQ reports an error.

Corresponds to:

zmq_msg_send() (with the ZMQ_DONTWAIT flag, in the case of trySend).

Examples:

auto sck = Socket(SocketType.pub);
auto msg = Message(12);
msg.data.asString()[] = "Hello World!";
sck.send(msg);

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

Receives a message part.

receive blocks until the request can be satisfied, and returns the number of bytes in the message. tryReceive performs the operation in non-blocking mode, and returns a std.typecons.Tuple which contains the size of the message along with a bool value that signifies whether a message 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("ipc://zmqd_receive_example");
snd.send("Hello World!");

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

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

Receives a message part.

receive blocks until the request can be satisfied, and returns the number of bytes in the message. tryReceive performs the operation in non-blocking mode, and returns a std.typecons.Tuple which contains the size of the message along with a bool value that signifies whether a message 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("ipc://zmqd_msg_receive_example");
snd.send("Hello World!");

// Receiver
import std.string: representation;
auto rcv = Socket(SocketType.rep);
rcv.bind("ipc://zmqd_msg_receive_example");
auto msg = Message();
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 data parts 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 @trusted ubyte[] identity();
@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 ipv4Only();
@property @safe void ipv4Only(bool value);
@property @safe bool delayAttachOnConnect();
@property @safe void delayAttachOnConnect(bool value);
@property @safe FD fd();
@property @safe int events();
@property @trusted char[] lastEndpoint();

Misc. socket properties.

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).
  • The fd property is an int on POSIX and a SOCKET on Windows.
  • 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).

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

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

Starts the built-in ∅MQ proxy.

Corresponds to:

zmq_proxy()

struct Message;

An object that encapsulates a ∅MQ message.

This struct is a wrapper around a zmq_msg_t object. Unlike Context and Socket, it does not perform reference counting, because ∅MQ messages have a form of reference counting of their own. A Message cannot be copied by normal assignment; use Message.copy() for this.

A default-initialized Message is not a valid ∅MQ message; it must always be explicitly initialized with Message.opCall() or Message.this(): 
Message msg1;               // Invalid message
auto msg2 = Message();      // Empty message
auto msg3 = Message(1024);  // 1K message
When a Message goes out of scope, zmq_msg_close() is called on the underlying zmq_msg_t.

Members

static @safe Message opCall(size_t size = 0);

Initialises a ∅MQ message.

size specifies the size of the message. If size is zero, the message will be empty.

Throws:

ZmqException if ∅MQ reports an error.

Corresponds to:

zmq_msg_init() if size == 0 and zmq_msg_init_size() if size > 0.

Examples:

auto msg = Message();
assert(msg.size == 0);

Examples:

auto msg = Message(123);
assert(msg.size == 123);

@safe void close();

Releases the ∅MQ message.

Note that the message will be automatically released when the Message 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 Message copy();
@safe void copyTo(ref Message dest);

Copies message content to another message.

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

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 = Message(3);
msg1.data[] = "foo".representation;
auto msg2 = msg1.copy();
assert (msg2.data.asString() == "foo");

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

Moves message content to another message.

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

Throws:

ZmqException if ∅MQ reports an error.

Corresponds to:

zmq_msg_move()

Examples:

import std.string: representation;
auto msg1 = Message(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 content size in bytes.

Corresponds to:

zmq_msg_size()

Examples:

auto msg = Message(123);
assert(msg.size == 123);

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

Retrieves the message content.

Corresponds to:

zmq_msg_data()

Examples:

import std.string: representation;
auto msg = Message(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 parts 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.

enum EventType: int;

Socket event types.

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

Members

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(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.

The function will attempt to detect whether the received message is in fact an event message, by checking that its length is equal to zmq_event_t.sizeof and that the value of the zmq_event_t.event field is valid. If this is not the case, an InvalidEventException is thrown.

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

Members

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 Socket.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

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("ipc://zmqd_asString_example");
auto s2 = Socket(SocketType.pair);
s2.connect("ipc://zmqd_asString_example");

auto msg = Message(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.

Members

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.