void zmq::xrep_t::xattach_pipes (reader_t *inpipe_, writer_t *outpipe_,
    const blob_t &peer_identity_)
{
    if (outpipe_) {

        outpipe_->set_event_sink (this);

        //  TODO: What if new connection has same peer identity as the old one?
        outpipe_t outpipe = {outpipe_, true};
        bool ok = outpipes.insert (outpipes_t::value_type (
            peer_identity_, outpipe)).second;
        zmq_assert (ok);

        if (terminating) {
            register_term_acks (1);
            outpipe_->terminate ();
        }
    }

    if (inpipe_) {

        inpipe_->set_event_sink (this);

        inpipe_t inpipe = {inpipe_, peer_identity_, true};
        inpipes.push_back (inpipe);

        if (terminating) {
            register_term_acks (1);
            inpipe_->terminate ();
        }
    }
}

void zmq::session_t::attach_pipes (class reader_t *inpipe_,
    class writer_t *outpipe_, const blob_t &peer_identity_)
{
    if (inpipe_) {
        zmq_assert (!in_pipe);
        in_pipe = inpipe_;
        in_pipe->set_event_sink (this);
    }

    if (outpipe_) {
        zmq_assert (!out_pipe);
        out_pipe = outpipe_;
        out_pipe->set_event_sink (this);
    }

    //  If we are already terminating, terminate the pipes straight away.
    if (finalised) {
        if (in_pipe) {
            register_term_acks (1);
            in_pipe->terminate ();
        }
        if (out_pipe) {
            register_term_acks (1);
            out_pipe->terminate ();
        }
        return;
    }

    attach_processed = true;
    finalise ();
}

void zmq::session_t::proceed_with_term ()
{
    if (state == terminating)
        return;

    zmq_assert (state == pending);
    state = terminating;

    //  If there's still a pending linger timer, remove it.
    if (has_linger_timer) {
        cancel_timer (linger_timer_id);
        has_linger_timer = false;
    }

    if (in_pipe) {
        register_term_acks (1);
        in_pipe->terminate ();
    }
    if (out_pipe) {
        register_term_acks (1);
        out_pipe->terminate ();
    }

    //  The session has already waited for the linger period. We don't want
    //  the child objects to linger any more thus linger is set to zero.
    own_t::process_term (0);
}

void zmq::pair_t::process_term (int linger_)
{
    terminating = true;

    if (inpipe) {
        register_term_acks (1);
        inpipe->terminate ();
    }

    if (outpipe) {
        register_term_acks (1);
        outpipe->terminate ();
    }

    socket_base_t::process_term (linger_);
}

void zmq::socket_base_t::attach_pipe (pipe_t *pipe_,
    const blob_t &peer_identity_)
{
    //  First, register the pipe so that we can terminate it later on.
    pipe_->set_event_sink (this);
    pipes.push_back (pipe_);

    //  Then, pass the pipe to the specific socket type.
    //  If the peer haven't specified it's identity, let's generate one.
    if (peer_identity_.size ()) {
        xattach_pipe (pipe_, peer_identity_);
    }
    else {
        blob_t identity (17, 0);
        generate_uuid ((unsigned char*) identity.data () + 1);
        xattach_pipe (pipe_, identity);
    }

    //  If the socket is already being closed, ask any new pipes to terminate
    //  straight away.
    if (is_terminating ()) {
        register_term_acks (1);
        pipe_->terminate (false);
    }
}

void zmq::own_t::process_own (own_t *object_)
{
    //  If the object is already being shut down, new owned objects are
    //  immediately asked to terminate. Note that linger is set to zero.
    if (terminating) {
        register_term_acks (1);
        send_term (object_, 0);
        return;
    }

    //  Store the reference to the owned object.
    owned.insert (object_);
}

void zmq::socket_base_t::process_term (int linger_)
{
    //  Unregister all inproc endpoints associated with this socket.
    //  Doing this we make sure that no new pipes from other sockets (inproc)
    //  will be initiated.
    unregister_endpoints (this);

    //  Ask all attached pipes to terminate.
    for (pipes_t::size_type i = 0; i != pipes.size (); ++i)
        pipes [i]->terminate (false);
    register_term_acks ((int) pipes.size ());

    //  Continue the termination process immediately.
    own_t::process_term (linger_);
}

void zmq::xrep_t::process_term (int linger_)
{
    terminating = true;

    register_term_acks (inpipes.size () + outpipes.size ());

    for (inpipes_t::iterator it = inpipes.begin (); it != inpipes.end ();
          ++it)
        it->reader->terminate ();
    for (outpipes_t::iterator it = outpipes.begin (); it != outpipes.end ();
          ++it)
        it->second.writer->terminate ();

    socket_base_t::process_term (linger_);
}

void zmq::own_t::process_term (int linger_)
{
    //  Double termination should never happen.
    zmq_assert (!terminating);

    //  Send termination request to all owned objects.
    for (owned_t::iterator it = owned.begin (); it != owned.end (); ++it)
        send_term (*it, linger_);
    register_term_acks ((int) owned.size ());
    owned.clear ();

    //  Start termination process and check whether by chance we cannot
    //  terminate immediately.
    terminating = true;
    check_term_acks ();
}

void zmq::socket_base_t::attach_pipe (pipe_t *pipe_, bool subscribe_to_all_)
{
    //  First, register the pipe so that we can terminate it later on.
    pipe_->set_event_sink (this);
    pipes.push_back (pipe_);

    //  Let the derived socket type know about new pipe.
    xattach_pipe (pipe_, subscribe_to_all_);

    //  If the socket is already being closed, ask any new pipes to terminate
    //  straight away.
    if (is_terminating ()) {
        register_term_acks (1);
        pipe_->terminate (false);
    }
}

void zmq::pair_t::xattach_pipes (reader_t *inpipe_, writer_t *outpipe_,
    const blob_t &peer_identity_)
{
    zmq_assert (!inpipe && !outpipe);

    inpipe = inpipe_;
    inpipe_alive = true;
    inpipe->set_event_sink (this);

    outpipe = outpipe_;
    outpipe_alive = true;
    outpipe->set_event_sink (this);

    if (terminating) {
        register_term_acks (2);
        inpipe_->terminate ();
        outpipe_->terminate ();
    }
}