void zmq::pgm_receiver_t::activate_in ()
{
    //  It is possible that the most recently used decoder
    //  processed the whole buffer but failed to write
    //  the last message into the pipe.
    if (pending_bytes == 0) {
        if (mru_decoder != NULL)
            mru_decoder->process_buffer (NULL, 0);
        return;
    }

    zmq_assert (mru_decoder != NULL);
    zmq_assert (pending_ptr != NULL);

    //  Ask the decoder to process remaining data.
    size_t n = mru_decoder->process_buffer (pending_ptr, pending_bytes);
    pending_bytes -= n;

    if (pending_bytes > 0)
        return;

    //  Resume polling.
    set_pollin (pipe_handle);
    set_pollin (socket_handle);

    in_event ();
}

void zmq::pgm_sender_t::plug (io_thread_t *io_thread_, session_base_t *session_)
{
    //  Alocate 2 fds for PGM socket.
    fd_t downlink_socket_fd = retired_fd;
    fd_t uplink_socket_fd = retired_fd;
    fd_t rdata_notify_fd = retired_fd;
    fd_t pending_notify_fd = retired_fd;

    encoder.set_msg_source (session_);

    //  Fill fds from PGM transport and add them to the poller.
    pgm_socket.get_sender_fds (&downlink_socket_fd, &uplink_socket_fd,
        &rdata_notify_fd, &pending_notify_fd);

    handle = add_fd (downlink_socket_fd);
    uplink_handle = add_fd (uplink_socket_fd);
    rdata_notify_handle = add_fd (rdata_notify_fd);   
    pending_notify_handle = add_fd (pending_notify_fd);

    //  Set POLLIN. We wont never want to stop polling for uplink = we never
    //  want to stop porocess NAKs.
    set_pollin (uplink_handle);
    set_pollin (rdata_notify_handle);
    set_pollin (pending_notify_handle);

    //  Set POLLOUT for downlink_socket_handle.
    set_pollout (handle);
}

void zmq::pgm_sender_t::plug (i_inout *inout_)
{
    //  Alocate 2 fds for PGM socket.
    int downlink_socket_fd = 0;
    int uplink_socket_fd = 0;
    int rdata_notify_fd = 0;
    int pending_notify_fd = 0;

    encoder.set_inout (inout_);

    //  Fill fds from PGM transport and add them to the poller.
    pgm_socket.get_sender_fds (&downlink_socket_fd, &uplink_socket_fd,
                               &rdata_notify_fd, &pending_notify_fd);

    handle = add_fd (downlink_socket_fd);
    uplink_handle = add_fd (uplink_socket_fd);
    rdata_notify_handle = add_fd (rdata_notify_fd);
    pending_notify_handle = add_fd (pending_notify_fd);

    //  Set POLLIN. We wont never want to stop polling for uplink = we never
    //  want to stop porocess NAKs.
    set_pollin (uplink_handle);
    set_pollin (rdata_notify_handle);
    set_pollin (pending_notify_handle);

    //  Set POLLOUT for downlink_socket_handle.
    set_pollout (handle);
}

void zmq::pgm_receiver_t::plug (i_inout *inout_)
{
    //  Retrieve PGM fds and start polling.
    int socket_fd;
    int waiting_pipe_fd;
    pgm_socket.get_receiver_fds (&socket_fd, &waiting_pipe_fd);
    socket_handle = add_fd (socket_fd);
    pipe_handle = add_fd (waiting_pipe_fd);
    set_pollin (pipe_handle);
    set_pollin (socket_handle);

    inout = inout_;
}

void zmq::pgm_receiver_t::plug (io_thread_t *io_thread_, i_inout *inout_)
{
    //  Retrieve PGM fds and start polling.
    fd_t socket_fd = retired_fd;
    fd_t waiting_pipe_fd = retired_fd;
    pgm_socket.get_receiver_fds (&socket_fd, &waiting_pipe_fd);
    socket_handle = add_fd (socket_fd);
    pipe_handle = add_fd (waiting_pipe_fd);
    set_pollin (pipe_handle);
    set_pollin (socket_handle);

    inout = inout_;
}

void xs::stream_engine_t::activate_in ()
{
    set_pollin (handle);

    //  Speculative read.
    in_event (s);
}

void zmq::zmq_engine_t::activate_in ()
{
    set_pollin (handle);

    //  Speculative read.
    in_event ();
}

void zmq::stream_engine_t::plug (io_thread_t *io_thread_,
    session_base_t *session_)
{
    zmq_assert (!plugged);
    plugged = true;

    //  Connect to session object.
    zmq_assert (!session);
    zmq_assert (session_);
    session = session_;
    socket = session-> get_socket ();

    //  Connect to I/O threads poller object.
    io_object_t::plug (io_thread_);
    handle = add_fd (s);
    io_enabled = true;

    //  Send the 'length' and 'flags' fields of the identity message.
    //  The 'length' field is encoded in the long format.
    outpos = greeting_output_buffer;
    outpos [outsize++] = 0xff;
    put_uint64 (&outpos [outsize], options.identity_size + 1);
    outsize += 8;
    outpos [outsize++] = 0x7f;

    set_pollin (handle);
    set_pollout (handle);
    //  Flush all the data that may have been already received downstream.
    in_event ();
}

void zmq::udp_engine_t::plug (io_thread_t* io_thread_, session_base_t *session_)
{
    zmq_assert (!plugged);
    plugged = true;

    zmq_assert (!session);
    zmq_assert (session_);
    session = session_;

    //  Connect to I/O threads poller object.
    io_object_t::plug (io_thread_);
    handle = add_fd (fd);

    if (send_enabled) {
        if (!options.raw_socket) {
            out_address = address->resolved.udp_addr->dest_addr ();
            out_addrlen = address->resolved.udp_addr->dest_addrlen ();
        }
        else {
            out_address = (sockaddr *) &raw_address;
            out_addrlen = sizeof (sockaddr_in);
        }

        set_pollout (handle);
    }

    if (recv_enabled) {
        int on = 1;
        int rc = setsockopt (fd, SOL_SOCKET, SO_REUSEADDR, (char *) &on, sizeof (on));
#ifdef ZMQ_HAVE_WINDOWS
        wsa_assert (rc != SOCKET_ERROR);
#else
        errno_assert (rc == 0);
#endif

        rc = bind (fd, address->resolved.udp_addr->bind_addr (),
                       address->resolved.udp_addr->bind_addrlen ());
#ifdef ZMQ_HAVE_WINDOWS
        wsa_assert (rc != SOCKET_ERROR);
#else
        errno_assert (rc == 0);
#endif

        if (address->resolved.udp_addr->is_mcast ()) {
            struct ip_mreq mreq;
            mreq.imr_multiaddr = address->resolved.udp_addr->multicast_ip ();
            mreq.imr_interface = address->resolved.udp_addr->interface_ip ();
            rc = setsockopt (fd, IPPROTO_IP, IP_ADD_MEMBERSHIP, (char*) &mreq, sizeof (mreq));
#ifdef ZMQ_HAVE_WINDOWS
            wsa_assert (rc != SOCKET_ERROR);
#else
            errno_assert (rc == 0);
#endif
        }
        set_pollin (handle);

        //  Call restart output to drop all join/leave commands
        restart_output ();
    }
}

//  Called when our pipe is reactivated (able to accept more data).
void zmq::udp_receiver_t::restart_input ()
{
    //  Process any pending data.
    if (pending_bytes > 0) {
        ssize_t processed_bytes = 0;
        //    decoder->process_buffer (pending_p, pending_bytes);
        //  Flush any messages produced by the decoder to the pipe.
        session->flush ();
        if (processed_bytes < pending_bytes) {
            //  Some data (still) could not be written to the pipe.
            pending_bytes -= processed_bytes;
            pending_p += processed_bytes;
            //  Try again later.
            return;
        }
        //  Done with unprocessed data.
        pending_bytes = 0;
    }

    //  Reactivate polling.
    set_pollin (socket_handle);

    //  Read any data that might have showed up on the socket in the mean time.
    in_event (retired_fd);
}

void zmq::stream_engine_t::plug (io_thread_t *io_thread_,
    session_base_t *session_)
{
    zmq_assert (!plugged);
    plugged = true;

    //  Connect to session object.
    zmq_assert (!session);
    zmq_assert (session_);
    session = session_;
    socket = session-> get_socket ();

    //  Connect to I/O threads poller object.
    io_object_t::plug (io_thread_);
    handle = add_fd (s);
    io_error = false;

    if (options.raw_socket) {
        // no handshaking for raw sock, instantiate raw encoder and decoders
        encoder = new (std::nothrow) raw_encoder_t (out_batch_size);
        alloc_assert (encoder);

        decoder = new (std::nothrow) raw_decoder_t (in_batch_size);
        alloc_assert (decoder);

        // disable handshaking for raw socket
        handshaking = false;

        next_msg = &stream_engine_t::pull_msg_from_session;
        process_msg = &stream_engine_t::push_msg_to_session;

        if (options.raw_notify) {
            //  For raw sockets, send an initial 0-length message to the
            // application so that it knows a peer has connected.
            msg_t connector;
            connector.init();
            push_msg_to_session (&connector);
            connector.close();
            session->flush ();
        }
    }
    else {
        // start optional timer, to prevent handshake hanging on no input
        set_handshake_timer ();

        //  Send the 'length' and 'flags' fields of the identity message.
        //  The 'length' field is encoded in the long format.
        outpos = greeting_send;
        outpos [outsize++] = 0xff;
        put_uint64 (&outpos [outsize], options.identity_size + 1);
        outsize += 8;
        outpos [outsize++] = 0x7f;
    }

    set_pollin (handle);
    set_pollout (handle);
    //  Flush all the data that may have been already received downstream.
    in_event ();
}

void zmq::udp_engine_t::restart_input ()
{
    if (!recv_enabled)
        return;

    set_pollin (handle);
    in_event ();
}

void zmq::pgm_receiver_t::plug (io_thread_t *io_thread_,
    session_base_t *session_)
{
    //  Retrieve PGM fds and start polling.
    fd_t socket_fd = retired_fd;
    fd_t waiting_pipe_fd = retired_fd;
    pgm_socket.get_receiver_fds (&socket_fd, &waiting_pipe_fd);
    socket_handle = add_fd (socket_fd);
    pipe_handle = add_fd (waiting_pipe_fd);
    set_pollin (pipe_handle);
    set_pollin (socket_handle);

    session = session_;

    //  If there are any subscriptions already queued in the session, drop them.
    drop_subscriptions ();
}

void zmq::socks_connecter_t::out_event ()
{
    zmq_assert (status == waiting_for_proxy_connection
             || status == sending_greeting
             || status == sending_request);

    if (status == waiting_for_proxy_connection) {
        const int rc = (int) check_proxy_connection ();
        if (rc == -1)
            error ();
        else {
            greeting_encoder.encode (
                socks_greeting_t (socks_no_auth_required));
            status = sending_greeting;
        }
    }
    else
    if (status == sending_greeting) {
        zmq_assert (greeting_encoder.has_pending_data ());
        const int rc = greeting_encoder.output (s);
        if (rc == -1 || rc == 0)
            error ();
        else
        if (!greeting_encoder.has_pending_data ()) {
            reset_pollout (handle);
            set_pollin (handle);
            status = waiting_for_choice;
        }
    }
    else {
        zmq_assert (request_encoder.has_pending_data ());
        const int rc = request_encoder.output (s);
        if (rc == -1 || rc == 0)
            error ();
        else
        if (!request_encoder.has_pending_data ()) {
            reset_pollout (handle);
            set_pollin (handle);
            status = waiting_for_response;
        }
    }
}

void zmq::udp_receiver_t::plug (io_thread_t *io_thread_,
    session_base_t *session_)
{
    //  Start polling.
    socket_handle = add_fd (socket);
    set_pollin (socket_handle);

    session = session_;
    //decoder->set_session (session);

    //  If there are any subscriptions already queued in the session, drop them.
    drop_subscriptions ();
}

void zmq::pgm_receiver_t::restart_input ()
{
    zmq_assert (session != NULL);
    zmq_assert (active_tsi != NULL);

    const peers_t::iterator it = peers.find (*active_tsi);
    zmq_assert (it != peers.end ());
    zmq_assert (it->second.joined);

    //  Push the pending message into the session.
    int rc = session->push_msg (it->second.decoder->msg ());
    errno_assert (rc == 0);

    if (insize > 0) {
        rc = process_input (it->second.decoder);
        if (rc == -1) {
            //  HWM reached; we will try later.
            if (errno == EAGAIN) {
                session->flush ();
                return;
            }
            //  Data error. Delete message decoder, mark the
            //  peer as not joined and drop remaining data.
            it->second.joined = false;
            delete it->second.decoder;
            it->second.decoder = NULL;
            insize = 0;
        }
    }

    //  Resume polling.
    set_pollin (pipe_handle);
    set_pollin (socket_handle);

    active_tsi = NULL;
    in_event ();
}

void zmq::norm_engine_t::plug (io_thread_t* io_thread_, session_base_t *session_)
{
    // TBD - we may assign the NORM engine to an io_thread in the future???
    zmq_session = session_;
    if (is_sender) zmq_output_ready = true;
    if (is_receiver) zmq_input_ready = true;

    fd_t normDescriptor = NormGetDescriptor(norm_instance);
    norm_descriptor_handle = add_fd(normDescriptor);
    // Set POLLIN for notification of pending NormEvents
    set_pollin(norm_descriptor_handle);

    if (is_sender) send_data();

}  // end zmq::norm_engine_t::init()

void zmq::zmq_engine_t::plug (i_inout *inout_)
{
    zmq_assert (!inout);

    encoder.set_inout (inout_);
    decoder.set_inout (inout_);

    handle = add_fd (tcp_socket.get_fd ());
    set_pollin (handle);
    set_pollout (handle);

    inout = inout_;

    //  Flush all the data that may have been already received downstream.
    in_event ();
}

void zmq::stream_engine_t::plug (io_thread_t *io_thread_,
    session_base_t *session_)
{
    zmq_assert (!plugged);
    plugged = true;

    //  Connect to session object.
    zmq_assert (!session);
    zmq_assert (session_);
    session = session_;
    socket = session-> get_socket ();

    //  Connect to I/O threads poller object.
    io_object_t::plug (io_thread_);
    handle = add_fd (s);
    io_error = false;

    if (options.raw_sock) {
        // no handshaking for raw sock, instantiate raw encoder and decoders
        encoder = new (std::nothrow) raw_encoder_t (out_batch_size);
        alloc_assert (encoder);

        decoder = new (std::nothrow) raw_decoder_t (in_batch_size);
        alloc_assert (decoder);

        // disable handshaking for raw socket
        handshaking = false;

        read_msg = &stream_engine_t::pull_msg_from_session;
        write_msg = &stream_engine_t::push_msg_to_session;
    }
    else {
        //  Send the 'length' and 'flags' fields of the identity message.
        //  The 'length' field is encoded in the long format.
        outpos = greeting_send;
        outpos [outsize++] = 0xff;
        put_uint64 (&outpos [outsize], options.identity_size + 1);
        outsize += 8;
        outpos [outsize++] = 0x7f;
    }

    set_pollin (handle);
    set_pollout (handle);
    //  Flush all the data that may have been already received downstream.
    in_event ();
}

void zmq::stream_engine_t::restart_input ()
{
    zmq_assert (input_stopped);
    zmq_assert (session != NULL);
    zmq_assert (decoder != NULL);

    int rc = (this->*process_msg) (decoder->msg ());
    if (rc == -1) {
        if (errno == EAGAIN)
            session->flush ();
        else
            error (protocol_error);
        return;
    }

    while (insize > 0) {
        size_t processed = 0;
        rc = decoder->decode (inpos, insize, processed);
        zmq_assert (processed <= insize);
        inpos += processed;
        insize -= processed;
        if (rc == 0 || rc == -1)
            break;
        rc = (this->*process_msg) (decoder->msg ());
        if (rc == -1)
            break;
    }

    if (rc == -1 && errno == EAGAIN)
        session->flush ();
    else
    if (io_error)
        error (connection_error);
    else
    if (rc == -1)
        error (protocol_error);
    else {
        input_stopped = false;
        set_pollin (handle);
        session->flush ();

        //  Speculative read.
        in_event ();
    }
}