uvco::Udp Class Reference

#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< 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.
Promise< void >	close ()
uv_udp_t *	underlying () const

Private Member Functions
int	udpStartReceive ()
void	udpStopReceive ()

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

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

Detailed Description

Interface to UDP functionality: can be connected or disconnected datagram client/server.

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 (is_receiving_) {
    fmt::print(stderr, "Udp::~Udp(): please co_await udp.stopReceiveMany() "
                       "before dropping Udp instance.\n");
  }
  if (udp_) {
    fmt::print(stderr, "Udp::~Udp(): closing UDP socket in dtor; "
                       "this will leak memory. "
                       "Please co_await udp.close() if possible.\n");
    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) {
    co_await close();
    throw UvcoException{status, "binding UDP socket"};
  }
}

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()

Promise< void > uvco::Udp::close

(

)

Close UDP socket. Await on the returned promise to ensure that the socket is fully closed.

                         {
  BOOST_ASSERT(udp_);
  if (!dataIsNull(udp_.get())) {
    auto *const awaiter = getData<RecvAwaiter_>(udp_.get());
    fmt::print(stderr, "Udp::close(): stopping receiving. Please instead use "
                       "Udp::stopReceivingMany() explicitly.\n");
    // Force return from receiveMany() generator.
    if (awaiter->handle_) {
      const auto resumeHandle = awaiter->handle_.value();
      awaiter->handle_.reset();
      resumeHandle.resume();
    }
  }
  co_await closeHandle(udp_.get());
  udp_.reset();
  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) {
    co_await close();
    throw UvcoException{status, "connecting UDP socket"};
  }
  connected_ = true;
}

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) {
    co_await close();
    throw UvcoException{status, "connecting UDP socket"};
  }
  connected_ = true;
}

getPeername()

std::optional< AddressHandle > uvco::Udp::getPeername

(

)

const

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

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

                                                                        {
 
  BOOST_ASSERT(!dataIsNull(handle));
  auto *awaiter = getData<RecvAwaiter_>(handle);
 
  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_) {
    auto 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 = getRequestData<SendAwaiter_>(req);
  awaiter->status_ = status;
  if (awaiter->handle_) {
    auto 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.

                                                                 {
  RecvAwaiter_ awaiter{};
  awaiter.stop_receiving_ = false;
  BOOST_ASSERT(dataIsNull(udp_.get()));
  setData(udp_.get(), &awaiter);
 
  const uv_status status = udpStartReceive();
  if (status != 0) {
    setData(udp_.get(), (void *)nullptr);
    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;
    }
    // It's possible that co_yield doesn't resume anymore, therefore clear
    // reference to local awaiter.
    setData(udp_.get(), (void *)nullptr);
    co_yield std::move(buffer.value());
    BOOST_ASSERT(dataIsNull(udp_.get()));
    setData(udp_.get(), &awaiter);
  }
  setData(udp_.get(), (void *)nullptr);
  udpStopReceive();
  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.

                                   {
  auto 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.

                                                               {
  RecvAwaiter_ awaiter{};
  BOOST_ASSERT(dataIsNull(udp_.get()));
  setData(udp_.get(), &awaiter);
  const uv_status status = udpStartReceive();
  if (status != 0) {
    setData(udp_.get(), (void *)nullptr);
    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
  setData(udp_.get(), (void *)nullptr);
  co_return std::move(packet.value());
}

send()

Promise< void > uvco::Udp::send	(	std::span< 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.

                                                            {
  SendAwaiter_ sendAwaiter{};
  uv_udp_send_t req{};
  setRequestData(&req, &sendAwaiter);
 
  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].base = &(*buffer.begin());
  bufs[0].len = 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) {
    setRequestData(&req, (void *)nullptr);
    throw UvcoException{status, "uv_udp_send() failed immediately"};
  }
 
  const uv_status status_done = co_await sendAwaiter;
  if (status_done != 0) {
    setRequestData(&req, (void *)nullptr);
    throw UvcoException{status_done, "uv_udp_send() failed while sending"};
  }
 
  setRequestData(&req, (void *)nullptr);
 
  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();
  if (dataIsNull(udp_.get())) {
    return;
  }
  auto *const currentAwaiter = getData<RecvAwaiter_>(udp_.get());
  // If generator is suspended on co_await, resume it synchronously so it can
  // exit before the Udp instance is possibly destroyed.
  if (currentAwaiter->handle_) {
    // Don't schedule this on the event loop: we must synchronously terminate
    // the onReceiveMany() loop, otherwise it will exist after destruction of
    // the Udp instance and read invalid memory.
    currentAwaiter->handle_->resume();
    // Don't touch currentAwaiter after this!!
  }
}

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());
}

underlying()

uv_udp_t * uvco::Udp::underlying

(

)

const

{ return udp_.get(); }

Member Data Documentation

connected_

bool uvco::Udp::connected_ = false

private

is_receiving_

bool uvco::Udp::is_receiving_ = 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