bool net::socket::accept(socket& s, socket_address* addr, socklen_t* addrlen, int timeout)
{
	if (timeout != 0) {
		if (!wait_readable(timeout)) {
			return false;
		}
	}

	int fd;

#if HAVE_ACCEPT4
	if ((fd = accept4(_M_fd, reinterpret_cast<struct sockaddr*>(addr), addrlen, SOCK_NONBLOCK)) < 0) {
		return false;
	}

	s._M_fd = fd;
#else
	if ((fd = ::accept(_M_fd, reinterpret_cast<struct sockaddr*>(addr), addrlen)) < 0) {
		return false;
	}

	s._M_fd = fd;

	if (!s.set_non_blocking()) {
		s.close();
		return false;
	}
#endif

	return true;
}

void Packet::read(socket& socket) {
  std::uint32_t bytes_read = 0;

  if (ext_len > 0) {
    std::vector<char> extras(ext_len);
    while (bytes_read < static_cast<std::uint32_t>(ext_len)) {
      bytes_read += socket.receive(boost::asio::buffer(extras));
    }
    bytes_read = 0;

    if (extras.size() > 4) {
      char flags_buf[4] = { extras[0], extras[1], extras[2], extras[3] };
      flags = readUInt32LE(flags_buf, 0);
    }
  }

  if (key_len > 0) {
    key.resize(key_len);
    while (bytes_read < key_len) {
      bytes_read += socket.receive(boost::asio::buffer(key));
    }
    bytes_read = 0;
  }

  if (bod_len > key_len) {
    std::uint32_t val_len = bod_len - key_len - ext_len;
    val.resize(val_len);
    while (bytes_read < val_len) {
      bytes_read += socket.receive(boost::asio::buffer(val));
    }
  }
}

bool net::tcp_server::on_new_connection(socket& client, const socket_address& addr, struct listener* listener)
{
	if (!allow_connection(addr, listener)) {
		return false;
	}

	if (!client.set_tcp_no_delay(true)) {
		return false;
	}

	tcp_connection* conn = _M_connections[client.fd()];

	// Add timer.
	if (!timer::timers::add(_M_current_msec + (tcp_connection::_M_max_idle_time * 1000), conn, 0, conn->_M_timer)) {
		return false;
	}

	if (!selector::add(client.fd(), fdset::FD_SOCKET, conn, _M_client_writes_first ? READ : WRITE)) {
		// Delete timer.
		del(conn->_M_timer);

		return false;
	}

	conn->fd(client.fd());

	conn->_M_listener = listener;

	conn->_M_timer_set = 1;

	return true;
}

void socket_acceptor::add(const socket& s) const {
	auto hdl = s.native_handle();
#if defined(NG_WINDOWS)
#else
	pimpl->max = std::max(pimpl->max, hdl);
#endif

	FD_SET(hdl, &pimpl->all);
}

// Call from work to establish a proxy between two sockets.
void worker::relay(socket& left, socket& right)
{
    // Blocks until the context is terminated, always returns -1.
    zmq_proxy_steerable(left.self(), right.self(), nullptr, nullptr);

    // Equivalent implementation:
    ////zmq::poller poller;
    ////poller.add(left);
    ////poller.add(right);
    ////
    ////while (!poller.terminated())
    ////{
    ////    const auto signaled = poller.wait();
    ////
    ////    if (signaled.contains(left.id()))
    ////        forward(left, right);
    ////
    ////    if (signaled.contains(right.id()))
    ////        forward(right, left);
    ////}
}

void certificate::apply(socket& sock)
{
    zcert_apply(self_, sock.self());
}

bool operator==(const socket& sock_a, const socket& sock_b)
{
    return sock_a.self() == sock_b.self();
}

// TODO: use non-copying private zmq implementation of forward.
// Call from work to forward a message from one socket to another.
bool worker::forward(socket& from, socket& to)
{
    message packet;
    return !from.receive(packet) && !to.send(packet);
}

 void set_resend_interval(socket &s, milliseconds i)
 { s.setopt(REQ, REQ_RESEND_IVL, static_cast<int>(i.count())); }

 void poll_entry::set(socket &s, int ev) noexcept
 { set(s.fd(), ev); }

 bool poll_entry::is(const socket &s) const noexcept
 { return this->fd == s.fd(); }

 void swap(socket &s1, socket &s2) noexcept
 { s1.swap(s2); }

void poller::add(socket& sock)
{
    zpoller_add(self_, sock.self());
}

void socket_acceptor::remove(const socket& s) const {
	auto hdl = s.native_handle();

	FD_CLR(hdl, &pimpl->all);
	FD_CLR(hdl, &pimpl->ready);
}

void auth::handle_command(socket& pipe) {
    // Get the whole message off the pipe in one go
	message msg;
	pipe.receive(msg);

    if(0 == msg.parts())
        return;     // Interrupted

    // authentication command
	std::string command = msg.get(0);
	if (verbose) {
    	std::cout <<"auth: API command=" << command << std::endl;
    }

    if("ALLOW" == command) {
    	std::string address = msg.get(1);
    	if(verbose) {
    		std::cout << "auth: whitelisting ipaddress=" << address << std::endl;
    	}

    	whitelist.insert(address);
    	pipe.send(signal::ok); 

    } else if("DENY" == command) {
    	std::string address = msg.get(1);
    	if(verbose) {
    		std::cout << "auth: blacklisting ipaddress=" << address << std::endl;
    	}

    	blacklist.insert(address);
    	pipe.send(signal::ok); 
    	
    } else if("DOMAIN" == command) {
    	std::string domain = msg.get(1);
    	if(verbose) {
    		std::cout << "auth: domain=" << domain << std::endl;
    	}

    	this->domain = domain;
    	pipe.send(signal::ok); 
    	
    } else if("PLAIN" == command) {
    	std::string user = msg.get(1);
    	std::string pass = msg.get(2);

        if (verbose) {
            std::cout << "auth: configured PLAIN - user:" << user << std::endl; 
        }

    	passwords.insert(std::make_pair(user, pass));
    	pipe.send(signal::ok); 

    } else if("CURVE" == command) {
    	// If client_public_key is CURVE_ALLOW_ANY, allow all clients. Otherwise
    	// treat client_public_key as client public key certificate.
    	std::string client_public_key = msg.get(1);

    	if("CURVE_ALLOW_ANY" == client_public_key) {
    		curve_allow_any = true;
            if(verbose) {
    		  std::cout << "auth: configured CURVE - allow ALL clients" << std::endl;
            }
    	} else {
    		curve_allow_any = false;
    		client_keys.insert(client_public_key);
            if(verbose) {
    		  std::cout << "auth: configured CURVE - allow client with public key:" << client_public_key << std::endl;
            }
    	}
		pipe.send(signal::ok); 

    } else if("GSSAPI" == command) {
    	// GSSAPI authentication is not yet implemented here
        if(verbose) {
    	   std::cout << "auth: configure GSSAPI authentication is not yet implemented here" << std::endl;
        }
    	pipe.send(signal::ok); 

    } else if("VERBOSE" == command) {
    	std::string verbose_string = msg.get(1);

    	verbose = ("true" == verbose_string)? true : false;
    	pipe.send(signal::ok); 

    } else if("TERMINATE" == command) {
    	std::cout << "auth: Shutdown ZAP Authentication Server" << std::endl;
        
    	terminated = true;
        pipe.send(signal::ok); 

    } else {
        if(verbose) {
    	   std::cout << "auth: Invalid command=" << command << std::endl;
        }
    	assert(false);
    }
}

socket::socket(const socket& sock) {
	_socket = sock.get_rawsocket();
//	sock.set_rawsocket(SOCKET_ERROR);
}

 void handleIncomingMessage(socket& sock, const std::string& msg)
 {
     typedef std::pair<std::string, std::string> NameAndArgs;
     auto nameAndArgs = deserialize<NameAndArgs>(msg);
     sock.write(functions_[nameAndArgs.first](nameAndArgs.second));
 }

 void device(socket &s1, socket &s2)
 {
   nn_device(s1.fd(), s2.fd());
   throw_error();
 }

bool socket_acceptor::is_ready(const socket& s) const {
	return FD_ISSET(s.native_handle(), &pimpl->ready) != 0;
}

 virtual void write(const byte *buf, size_t len) {
   s.write_some(boost::asio::buffer(buf,len));
 }