uvco::Udp Class Reference

A UDP socket. More...

#include <udp.h>

Collaboration diagram for uvco::Udp:

Classes
struct	RecvAwaiter_
struct	SendAwaiter_

Public Member Functions
	Udp (const Loop &loop)
	Set up a UDP object.
	Udp (Udp &&other)=default
Udp &	operator= (Udp &&other)=default
	Udp (const Udp &)=delete
Udp &	operator= (const Udp &)=delete
	~Udp ()
Promise< void >	bind (std::string_view address, uint16_t port, unsigned int flag=0)
	Bind UDP socket to address.
Promise< void >	bind (const AddressHandle &address, unsigned int flag=0)
Promise< void >	connect (std::string_view address, uint16_t port, bool ipv6only=false)
	Connect UDP socket to address.
Promise< void >	connect (const AddressHandle &address)
Promise< void >	send (std::span< const char > buffer, std::optional< AddressHandle > address={})
Promise< std::string >	receiveOne ()
Promise< std::pair< std::string, AddressHandle > >	receiveOneFrom ()
MultiPromise< std::pair< std::string, AddressHandle > >	receiveMany ()
void	stopReceiveMany (MultiPromise< std::pair< std::string, AddressHandle > > &packets)
void	setBroadcast (bool enabled)
	Enable sending to broadcast addresses.
void	setTtl (uint8_t ttl)
	Set TTL on sent packets. TTL must be between 1 and 255.
void	setMulticastInterface (const std::string &interfaceAddress)
	Set multicast interface for sending or receiving data.
void	setMulticastLoop (bool enabled)
void	joinMulticast (const std::string &address, const std::string &interface)
	Join a multicast group.
void	leaveMulticast (const std::string &address, const std::string &interface)
	Leave multicast group.
AddressHandle	getSockname () const
	Obtain locally bound name of socket.
std::optional< AddressHandle >	getPeername () const
	Obtain peer name if connected.
void	close ()

Private Member Functions
int	udpStartReceive ()
void	udpStopReceive ()

Static Private Member Functions
static void	onSendDone (uv_udp_send_t *req, uv_status status)
static void	onReceiveOne (uv_udp_t *handle, ssize_t nread, const uv_buf_t *buf, const struct sockaddr *addr, unsigned int flags)

Private Attributes
const Loop *	loop_
std::unique_ptr< uv_udp_t >	udp_
bool	connected_ = false

Detailed Description

A UDP socket.

Constructor & Destructor Documentation

Udp() [1/3]

uvco::Udp::Udp ( const Loop & loop )

explicit

Set up a UDP object.

                         : loop_{&loop}, udp_{std::make_unique<uv_udp_t>()} {
  uv_udp_init(loop.uvloop(), udp_.get());
}

Udp() [2/3]

uvco::Udp::Udp ( Udp && other )

default

Udp() [3/3]

uvco::Udp::Udp ( const Udp & )

delete

~Udp()

uvco::Udp::~Udp

(

)

          {
  if (udp_) {
    udpStopReceive();
    closeHandle(udp_.release());
  }
}

Member Function Documentation

bind() [1/2]

Promise< void > uvco::Udp::bind	(	const AddressHandle &	address,
		unsigned int	flag = 0 )

                                                                       {
  const uv_status status = uv_udp_bind(udp_.get(), address.sockaddr(), flag);
  if (status != 0) {
    close();
    throw UvcoException{status, "binding UDP socket"};
  }
  co_return;
}

bind() [2/2]

Promise< void > uvco::Udp::bind	(	std::string_view	address,
		uint16_t	port,
		unsigned int	flag = 0 )

Bind UDP socket to address.

close()

void uvco::Udp::close

(

)

Close UDP socket. The socket is not closed immediately; it typically takes another turn of the libuv event loop before the closing operation is finished. In most cases you don't need to care about that.

                {
  if (udp_ == nullptr) {
    return;
  }
  if (!dataIsNull(udp_.get())) {
    auto *const awaiter = getData<RecvAwaiter_>(udp_.get());
    awaiter->buffer_.put(uv_status{UV_ECANCELED});
    if (awaiter->handle_) {
      // Loop::enqueue(awaiter->handle_.value());
      const std::coroutine_handle<> h = awaiter->handle_.value();
      awaiter->handle_.reset();
      // This immediately resumes the receiveMany generator which will get the
      // ECANCELED exception, store it, and make it available to anyone waiting
      // on it.
      h.resume();
    }
  }
  closeHandle(udp_.release());
  connected_ = false;
}

connect() [1/2]

Promise< void > uvco::Udp::connect

(

const AddressHandle &

address

)

                                                       {
  uv_udp_connect(udp_.get(), nullptr);
  const uv_status status = uv_udp_connect(udp_.get(), address.sockaddr());
  if (status != 0) {
    close();
    throw UvcoException{status, "connecting UDP socket"};
  }
  connected_ = true;
  co_return;
}

connect() [2/2]

Promise< void > uvco::Udp::connect	(	std::string_view	address,
		uint16_t	port,
		bool	ipv6only = false )

Connect UDP socket to address.

                                          {
  Resolver resolver{*loop_};
  const int hint = ipv6only ? AF_INET6 : AF_UNSPEC;
  AddressHandle addressHandle = co_await resolver.gai(address, port, hint);
 
  uv_udp_connect(udp_.get(), nullptr);
  const uv_status status = uv_udp_connect(udp_.get(), addressHandle.sockaddr());
  if (status != 0) {
    close();
    throw UvcoException{status, "connecting UDP socket"};
  }
  connected_ = true;
  co_return;
}

getPeername()

std::optional< AddressHandle > uvco::Udp::getPeername ( ) const

nodiscard

Obtain peer name if connected.

                                                  {
  struct sockaddr_storage address{};
  int ss_size = sizeof(struct sockaddr_storage);
  const uv_status status =
      uv_udp_getpeername(udp_.get(), (struct sockaddr *)&address, &ss_size);
  if (status < 0) {
    if (status == UV_ENOTCONN) {
      return std::nullopt;
    }
    throw UvcoException(status, "Error in getpeername");
  }
  AddressHandle addressHandle{(struct sockaddr *)&address};
  return addressHandle;
}

getSockname()

AddressHandle uvco::Udp::getSockname ( ) const

nodiscard

Obtain locally bound name of socket.

                                     {
  struct sockaddr_storage address{};
  int ss_size = sizeof(struct sockaddr_storage);
  const uv_status status =
      uv_udp_getsockname(udp_.get(), (struct sockaddr *)&address, &ss_size);
  if (status < 0) {
    throw UvcoException(status, "Error in getsockname");
  }
  AddressHandle addressHandle{(struct sockaddr *)&address};
  return addressHandle;
}

joinMulticast()

void uvco::Udp::joinMulticast	(	const std::string &	address,
		const std::string &	interface )

Join a multicast group.

                                                    {
  const uv_status status = uv_udp_set_membership(
      udp_.get(), address.c_str(), interface.c_str(), UV_JOIN_GROUP);
  if (status != 0) {
    throw UvcoException(status, "join multicast group");
  }
}

leaveMulticast()

void uvco::Udp::leaveMulticast	(	const std::string &	address,
		const std::string &	interface )

Leave multicast group.

                                                     {
  const uv_status status = uv_udp_set_membership(
      udp_.get(), address.c_str(), interface.c_str(), UV_LEAVE_GROUP);
  if (status != 0) {
    throw UvcoException(status, "join multicast group");
  }
}

onReceiveOne()

void uvco::Udp::onReceiveOne ( uv_udp_t * handle, ssize_t nread, const uv_buf_t * buf, const struct sockaddr * addr, unsigned int flags )

staticprivate

                                                                        {
  auto *awaiter = getDataOrNull<RecvAwaiter_>(handle);
  // cancelled?
  if (awaiter == nullptr) {
    uv_udp_recv_stop(handle);
    freeUvBuf(buf);
    return;
  }
 
  if (addr == nullptr) {
    // Error or asking to free buffers.
    if (0 == (flags & UV_UDP_MMSG_CHUNK)) {
      freeUvBuf(buf);
    }
    return;
  }
 
  if (awaiter->stop_receiving_) {
    uv_udp_recv_stop(handle);
  }
  if (awaiter->buffer_.hasSpace()) {
    if (nread >= 0) {
      awaiter->buffer_.put(RecvAwaiter_::QueueItem_{
          std::make_pair(std::string{buf->base, static_cast<size_t>(nread)},
                         AddressHandle{addr})});
    } else {
      awaiter->buffer_.put(
          RecvAwaiter_::QueueItem_{static_cast<uv_status>(nread)});
    }
  } else {
    fmt::print(stderr, "Udp::onReceiveOne: dropping packet, buffer full\n");
  }
 
  if (0 == (flags & UV_UDP_MMSG_CHUNK)) {
    freeUvBuf(buf);
  }
 
  // Only enqueues once; if this callback is called again, the receiver will
  // already have been resumed.
  if (awaiter->handle_) {
    std::coroutine_handle<void> resumeHandle = *awaiter->handle_;
    awaiter->handle_.reset();
    Loop::enqueue(resumeHandle);
  }
}

onSendDone()

void uvco::Udp::onSendDone ( uv_udp_send_t * req, uv_status status )

staticprivate

                                                         {
  auto *const awaiter = getRequestDataOrNull<SendAwaiter_>(req);
  if (awaiter == nullptr) {
    // Send request cancelled/dropped.
    return;
  }
  awaiter->status_ = status;
  if (awaiter->handle_) {
    std::coroutine_handle<void> resumeHandle = *awaiter->handle_;
    awaiter->handle_.reset();
    Loop::enqueue(resumeHandle);
  }
}

operator=() [1/2]

Udp & uvco::Udp::operator= ( const Udp & )

delete

operator=() [2/2]

Udp & uvco::Udp::operator= ( Udp && other )

default

receiveMany()

MultiPromise< std::pair< std::string, AddressHandle > > uvco::Udp::receiveMany

(

)

Generate packets received on socket. Call stopReceiveMany() when no more packets are desired; otherwise this will continue indefinitely.

Only one coroutine can be receiving at a time. This is currently enforced by assertions.

                                                                 {
  BOOST_ASSERT_MSG(dataIsNull(udp_.get()),
                   "only one coroutine can receive from UDP socket at a time");
  RecvAwaiter_ awaiter{*udp_};
  awaiter.stop_receiving_ = false;
  const OnExit onExit{[this] {
    if (udp_) {
      udpStopReceive();
      resetData(udp_.get());
    }
  }};
 
  const uv_status status = udpStartReceive();
  if (status != 0) {
    throw UvcoException(status, "receiveMany(): uv_udp_recv_start()");
  }
 
  while (uv_is_active((uv_handle_t *)udp_.get()) != 0) {
    // Awaiter returns empty optional on requested stop (stopReceiveMany()).
    std::optional<std::pair<std::string, AddressHandle>> buffer =
        co_await awaiter;
    if (!buffer) {
      break;
    }
    co_yield std::move(buffer.value());
  }
  co_return;
}

receiveOne()

Promise< std::string > uvco::Udp::receiveOne

(

)

Receive a single UDP packet.

TODO: use a better-suited buffer type.

Only one coroutine can be receiving at a time. This is currently enforced by assertions.

                                   {
  std::pair<std::basic_string<char>, AddressHandle> packet =
      co_await receiveOneFrom();
  co_return std::move(packet.first);
}

receiveOneFrom()

Promise< std::pair< std::string, AddressHandle > > uvco::Udp::receiveOneFrom

(

)

Receive a single UDP packet and also return the sender's address.

Only one coroutine can be receiving at a time. This is currently enforced by assertions.

                                                               {
  BOOST_ASSERT_MSG(dataIsNull(udp_.get()),
                   "only one coroutine can receive from UDP socket at a time");
  RecvAwaiter_ awaiter{*udp_, 1};
  awaiter.stop_receiving_ = true;
  const OnExit onExit{[this] { resetData(udp_.get()); }};
 
  const uv_status status = udpStartReceive();
  if (status != 0) {
    throw UvcoException(status, "uv_udp_recv_start()");
  }
 
  // Exception thrown here if occurred.
  std::optional<std::pair<std::string, AddressHandle>> packet =
      co_await awaiter;
 
  // Any exceptions are thrown in RecvAwaiter_::await_resume
  co_return std::move(packet.value());
}

send()

Promise< void > uvco::Udp::send	(	std::span< const char >	buffer,
		std::optional< AddressHandle >	address = {} )

Send to address, or send to connected peer. Must be a mutable buffer because libuv requires it - the buffer will not be modified.

                                                            {
  uv_udp_send_t req{};
  SendAwaiter_ sendAwaiter{req};
 
  std::array<uv_buf_t, 1> bufs{};
  // The buffer is never written to, so this is necessary to interface
  // with the legacy C code.
  bufs[0] = uv_buf_init(const_cast<char *>(buffer.data()), buffer.size_bytes());
 
  const struct sockaddr *addr = nullptr;
  if (address) {
    addr = address->sockaddr();
  }
 
  const uv_status status =
      uv_udp_send(&req, udp_.get(), bufs.begin(), 1, addr, onSendDone);
  if (status != 0) {
    throw UvcoException{status, "uv_udp_send() failed immediately"};
  }
 
  const uv_status status_done = co_await sendAwaiter;
  if (status_done != 0) {
    throw UvcoException{status_done, "uv_udp_send() failed while sending"};
  }
 
  co_return;
}

setBroadcast()

void uvco::Udp::setBroadcast

(

bool

enabled

)

Enable sending to broadcast addresses.

                                   {
  const uv_status status =
      uv_udp_set_broadcast(udp_.get(), static_cast<int>(enabled));
  if (status != 0) {
    throw UvcoException(status, "join multicast group");
  }
}

setMulticastInterface()

void uvco::Udp::setMulticastInterface

(

const std::string &

interfaceAddress

)

Set multicast interface for sending or receiving data.

                                                                 {
  const uv_status status =
      uv_udp_set_multicast_interface(udp_.get(), interfaceAddress.c_str());
  if (status != 0) {
    throw UvcoException(status, "join multicast group");
  }
}

setMulticastLoop()

void uvco::Udp::setMulticastLoop

(

bool

enabled

)

If the loop flag is enabled, sent multicast packets will arrive back on the sending socket.

                                       {
  const uv_status status =
      uv_udp_set_multicast_loop(udp_.get(), static_cast<int>(enabled));
  if (status != 0) {
    throw UvcoException(status, "join multicast group");
  }
}

setTtl()

void uvco::Udp::setTtl

(

uint8_t

ttl

)

Set TTL on sent packets. TTL must be between 1 and 255.

                            {
  const uv_status status = uv_udp_set_ttl(udp_.get(), static_cast<int>(ttl));
  if (status != 0) {
    throw UvcoException(status, "join multicast group");
  }
}

stopReceiveMany()

void uvco::Udp::stopReceiveMany

(

MultiPromise< std::pair< std::string, AddressHandle > > &

packets

)

Stop receiving with receiveMany() by cancelling the receiving generator coroutine. Supply the MultiPromise obtained from receiveMany() in order to guarantee a complete clean-up.

                                                              {
  udpStopReceive();
  // Cancel receiving generator if currently suspended by co_yield.
  packets.cancel();
}

udpStartReceive()

int uvco::Udp::udpStartReceive ( )

private

                         {
  BOOST_ASSERT(udp_);
  return uv_udp_recv_start(udp_.get(), allocator, onReceiveOne);
}

udpStopReceive()

void uvco::Udp::udpStopReceive ( )

private

                         {
  BOOST_ASSERT(udp_);
  uv_udp_recv_stop(udp_.get());
}

Member Data Documentation

connected_

bool uvco::Udp::connected_ = false

private

loop_

const Loop* uvco::Udp::loop_

private

udp_

std::unique_ptr<uv_udp_t> uvco::Udp::udp_

private

---

The documentation for this class was generated from the following files:

• uvco/udp.h
• uvco/udp.cc