zeromq_handler.d

import core.thread;
import deimos.zmq.zmq;
import std.string;
import std.stdio;
import std.c.string;

class ZeroMQ : Thread
{
    private:
        static void* _context;
        void* _responder;
    protected:
        @property
        void *responder() nothrow
        {
            return this._responder;
        }

        @property
        void responder(void *responder) nothrow
        {
            if (this.responder != null)
                zmq_close(this.responder);
            this._responder = responder;
        }

        void run()
        {
            while(1)
            {
                char buffer[10];
                int rc = zmq_recv(this.responder, cast(void*)buffer, 10, 0);
                debug {
                    char *errorp = zmq_strerror(zmq_errno());
                    string error_str = cast(string)errorp[0..strlen(errorp)];
                    if (rc < 0)
                        writefln("Error: %s", error_str);
                }
                writefln("Received: %s\n", buffer);
                zmq_send(responder, toStringz("World"), 6, 0);
            }
        }

    public:
        this()
        {
            this._responder = null;
            super( &run );
        }

        void initialize()
        {
            this.responder = zmq_socket(this.context, ZMQ_REP);
            int rc = zmq_bind(this.responder, "tcp://*:5555".ptr);
            assert(rc == 0);
        }

        static this()
        {
            this._context = zmq_init(1);
            int major, minor, patch;
            zmq_version(&major, &minor, &patch);
            writefln("%d.%d.%d", major, minor, patch);
        }

        @property
        static void* context() nothrow
        {
            return this._context;
        }

        @property
        static void context(void* aContext) nothrow
        {
            if (this.context != null)
                zmq_term(this.context);
            this._context = aContext;
        }
}