Recording

// foo_interface.hpp
#include <boost/noncopyable.hpp>

namespace foo {

class Interface : boost::noncopyable {
public:
    int Bar();
};

}

// foo.hpp
#include "foo_interface.hpp"

#include <boost/shared_ptr.hpp>

namespace foo {

typedef boost::shared_ptr<Interface> Reference;

Reference New();

}

// foo.cpp
#include "foo.hpp"

namespace foo {

class implementation : public Interface {
public:
int Bar() {
    // ...
    return bar0();
}

implementation() {
    // ...
}

private:
    int priv_data0;
    // ...

int bar0() {
    // ...
}

};

int Interface::Bar() {
    BOOST_AUTO(impl, static_cast<implementation>(this));
    return impl->Bar();
}

Reference New() {
    return Reference(new implementation());
}

}

typedef ypipe_t < zmq_msg_t, message_pipe_granularity > pipe_t;

//  Lock-free queue implementation.
//  Only a single thread can read from the pipe at any specific moment.
//  Only a single thread can write to the pipe at any specific moment.
//  T is the type of the object in the queue.
//  N is granularity of the pipe, i.e. how many items are needed to
//  perform next memory allocation.

//  Write an item to the pipe.  Don't flush it yet. If incomplete is
//  set to true the item is assumed to be continued by items
//  subsequently written to the pipe. Incomplete items are never
//  flushed down the stream.
inline void write (const T &value_, bool incomplete_)
{
    //  Place the value to the queue, add new terminator element.
    queue.back () = value_;
    queue.push ();

    //  Move the "flush up to here" poiter.
    if (!incomplete_)
	f = &queue.back ();
}

// Virtual interface to be exposed by object that want to be notified
// about events on file descriptors.
 
struct i_poll_events
{
    virtual ~i_poll_events () {}
 
    // Called by I/O thread when file descriptor is ready for reading.
    virtual void in_event () = 0;
 
    // Called by I/O thread when file descriptor is ready for writing.
    virtual void out_event () = 0;
 
    // Called when timer expires.
    virtual void timer_event (int id_) = 0;
};

if (protocol == "tcp" || protocol == "ipc") {

        //  Choose I/O thread to run the listerner in.
        io_thread_t *io_thread = choose_io_thread (options.affinity);
        if (!io_thread) {
            errno = EMTHREAD;
            return -1;
        }

        //  Create and run the listener.
        zmq_listener_t *listener = new (std::nothrow) zmq_listener_t (
            io_thread, this, options);
        alloc_assert (listener);
        int rc = listener->set_address (protocol.c_str(), address.c_str ());
        if (rc != 0) {
            delete listener;
            return -1;
        }
        launch_child (listener);

        return 0;
    }

int zmq::zmq_listener_t::set_address (const char *protocol_, const char *addr_)
{
     return tcp_listener.set_address (protocol_, addr_, options.backlog);
}

int zmq::tcp_listener_t::set_address (const char *protocol_, const char *addr_,
    int backlog_)
{
    if (strcmp (protocol_, "tcp") == 0 ) {

        //  Resolve the sockaddr to bind to.
        int rc = resolve_ip_interface (&addr, &addr_len, addr_);
        if (rc != 0)
            return -1;

        //  Create a listening socket.
        s = open_socket (addr.ss_family, SOCK_STREAM, IPPROTO_TCP);
        if (s == -1)
            return -1;

        //  Allow reusing of the address.
        int flag = 1;
        rc = setsockopt (s, SOL_SOCKET, SO_REUSEADDR, &flag, sizeof (int));
        errno_assert (rc == 0);

        //  Set the non-blocking flag.
    	flag = fcntl (s, F_GETFL, 0);
    	if (flag == -1)
            flag = 0;
    	rc = fcntl (s, F_SETFL, flag | O_NONBLOCK);
        errno_assert (rc != -1);

        //  Bind the socket to the network interface and port.
        rc = bind (s, (struct sockaddr*) &addr, addr_len);
        if (rc != 0) {
            int err = errno;
            if (close () != 0)
                return -1;
            errno = err;
            return -1;
        }

        //  Listen for incomming connections.
        rc = listen (s, backlog_);
        if (rc != 0) {
            int err = errno;
            if (close () != 0)
                return -1;
            errno = err;
            return -1;
        }

        return 0;
    }
    else if (strcmp (protocol_, "ipc") == 0) {

        //  Get rid of the file associated with the UNIX domain socket that
        //  may have been left behind by the previous run of the application.
        ::unlink (addr_);

        //  Convert the address into sockaddr_un structure.
        int rc = resolve_local_path (&addr, &addr_len, addr_);
        if (rc != 0)
            return -1;

        //  Create a listening socket.
        s = socket (AF_UNIX, SOCK_STREAM, 0);
        if (s == -1)
            return -1;

        //  Set the non-blocking flag.
        int flag = fcntl (s, F_GETFL, 0);
        if (flag == -1) 
            flag = 0;
        rc = fcntl (s, F_SETFL, flag | O_NONBLOCK);
        errno_assert (rc != -1);

        //  Bind the socket to the file path.
        rc = bind (s, (struct sockaddr*) &addr, addr_len);
        if (rc != 0) {
            int err = errno;
            if (close () != 0)
                return -1;
            errno = err;
            return -1;
        }
        has_file = true;

        //  Listen for incomming connections.
        rc = listen (s, backlog_);
        if (rc != 0) {
            int err = errno;
            if (close () != 0)
                return -1;
            errno = err;
            return -1;
        }

        return 0;
    }
    else {
        errno = EPROTONOSUPPORT;
        return -1;
    }    
}

void zmq::zmq_listener_t::process_plug ()
{
    //  Start polling for incoming connections.
    handle = add_fd (tcp_listener.get_fd ());
    set_pollin (handle);
}

zmq::fd_t zmq::tcp_listener_t::get_fd ()
{
    return s;
}

void zmq::zmq_listener_t::in_event ()
{
    fd_t fd = tcp_listener.accept ();

    //  If connection was reset by the peer in the meantime, just ignore it.
    //  TODO: Handle specific errors like ENFILE/EMFILE etc.
    if (fd == retired_fd)
        return;

    //  Choose I/O thread to run connecter in. Given that we are already
    //  running in an I/O thread, there must be at least one available.
    io_thread_t *io_thread = choose_io_thread (options.affinity);
    zmq_assert (io_thread);

    //  Create and launch an init object. 
    zmq_init_t *init = new (std::nothrow) zmq_init_t (io_thread, socket,
        NULL, fd, options);
    alloc_assert (init);
    launch_child (init);
}