actor.hpp

Go to the documentation of this file.

#ifndef ZIO_ACTOR_HPP_SEEN
#define ZIO_ACTOR_HPP_SEEN

#include "zio/cppzmq.hpp"

#include <thread>

namespace zio {

    inline
    std::pair<socket_t, socket_t>
    create_linked_pairs(context_t& ctx)
    {
        std::pair<socket_t, socket_t> ret{socket_t(ctx, socket_type::pair),
                                          socket_t(ctx, socket_type::pair)};

        std::stringstream ss;
        ss << "inproc://link-"
           << std::hex
           << ret.first.handle()
           << "-"
           << ret.second.handle();
        std::string addr = ss.str();
        ret.first.bind(addr.c_str());
        ret.second.connect(addr.c_str());
        return ret;
    }

    class zactor_t {
    public:
        // Template constructor and not class to perform type erasure
        // of the function type.
        template<typename Func, typename... Args>
        zactor_t(context_t &ctx, Func fn, Args... args) {
            socket_t asock;
            std::tie(asock, _sock) = create_linked_pairs(ctx);

            _thread = std::thread(
                [fn = std::forward<Func>(fn)](socket_t asock, Args... args) {
                    fn(asock, std::forward<Args>(args)...);
                    // send notice of exit down link
                    asock.send(message_t{}, send_flags::none);
                },
                std::move(asock), std::forward<Args>(args)...);
            startup();
        }

        ~zactor_t() {
            shutdown();
            _thread.join();
        }

        socket_ref link() { return _sock; }

    private:

        void startup() {
            // The default contract with the actor function is that it
            // shall notify us with a "signal" message that we may
            // continue.
            message_t rmsg; 
            auto res = link().recv(rmsg);
        }

        void shutdown() {
            // The default contract with the actor function is we will
            // notify it to terminate.  If sending that message is
            // successful we wait for the built-in confirmation
            // message.
            auto sres = link().send(message_t("$TERM",5), send_flags::dontwait);
            if (sres) {
                message_t rmsg;
                auto res = link().recv(rmsg, recv_flags::none);
            }
        }

    private:
        socket_t _sock;
        std::thread _thread;
        
    };
}

#endif