uvco::StreamServerBase< UvStreamType, StreamType > Class Template Reference

#include <stream_server_base.h>

Classes
struct	ConnectionAwaiter_

Public Member Functions
	StreamServerBase (const StreamServerBase &)=delete
	StreamServerBase (StreamServerBase &&)=default
StreamServerBase &	operator= (const StreamServerBase &)=delete
StreamServerBase &	operator= (StreamServerBase &&)=default
	~StreamServerBase ()
MultiPromise< StreamType >	listen (int backlog=128)
Promise< void >	close ()
	Close server and stop accepting client connections; must be awaited.

Protected Member Functions
	StreamServerBase (std::unique_ptr< UvStreamType > socket)

Protected Attributes
std::unique_ptr< UvStreamType >	socket_

Static Private Member Functions
static void	onNewConnection (uv_stream_t *stream, uv_status status)

Detailed Description

template<typename UvStreamType, typename StreamType>
class uvco::StreamServerBase< UvStreamType, StreamType >

Not for use in user code; base class for e.g. UnixServer and TcpServer.

Because accepting connections looks the same for Unix and TCP servers, the behavior is defined here and shared by both. However, the implementation must be generic over the stream type, so the actual stream type is a template parameter.

Constructor & Destructor Documentation

StreamServerBase() [1/3]

template<typename UvStreamType , typename StreamType >

uvco::StreamServerBase< UvStreamType, StreamType >::StreamServerBase ( const StreamServerBase< UvStreamType, StreamType > &  )

delete

StreamServerBase() [2/3]

template<typename UvStreamType , typename StreamType >

uvco::StreamServerBase< UvStreamType, StreamType >::StreamServerBase ( StreamServerBase< UvStreamType, StreamType > &&  )

default

~StreamServerBase()

template<typename UvStreamType , typename StreamType >

uvco::StreamServerBase< UvStreamType, StreamType >::~StreamServerBase

(

)

                                                              {
  if (socket_) {
    fmt::print(stderr, "StreamServerBase::~StreamServerBase(): closing server "
                       "in dtor; this will leak memory. "
                       "Please co_await server.close() if possible.\n");
    // Asynchronously close handle. It's better to leak memory than file
    // descriptors.
    closeHandle(socket_.release());
  }
}

StreamServerBase() [3/3]

template<typename UvStreamType , typename StreamType >

uvco::StreamServerBase< UvStreamType, StreamType >::StreamServerBase ( std::unique_ptr< UvStreamType >  socket )

inlineexplicitprotected

      : socket_{std::move(socket)} {}

Member Function Documentation

close()

template<typename UvStreamType , typename StreamType >

Promise< void > uvco::StreamServerBase< UvStreamType, StreamType >::close

(

)

Close server and stop accepting client connections; must be awaited.

                                                                {
  if (!dataIsNull(socket_.get())) {
    auto *awaiter = getData<ConnectionAwaiter_>(socket_.get());
    // Resume listener coroutine and tell it to exit.
    // If awaiter == nullptr, one of two things is true:
    // 1. listener is currently not running
    // 2. listener has yielded and is suspended there: the listener generator
    // will be cancelled when its MultiPromise is dropped.
    if (awaiter->handle_) {
      awaiter->stop();
    }
  }
  co_await closeHandle(socket_.get());
  socket_.reset();
}

listen()

template<typename UvStreamType , typename StreamType >

MultiPromise< StreamType > uvco::StreamServerBase< UvStreamType, StreamType >::listen

(

int

backlog = 128

)

Return client connections as clients connect.

Raises exceptions if errors occur during accepting or listening.

This generator may not be co_awaited on after having called close().

                                                              {
  BOOST_ASSERT(socket_);
  ConnectionAwaiter_ awaiter{*socket_};
  setData(socket_.get(), &awaiter);
 
  const uv_status listenStatus =
      uv_listen((uv_stream_t *)socket_.get(), backlog,
                StreamServerBase<UvStreamType, StreamType>::onNewConnection);
  if (listenStatus != 0) {
    setData(socket_.get(), (void *)nullptr);
    throw UvcoException{listenStatus,
                        "StreamServerBase::listen(): failed to listen"};
  }
 
  while (true) {
    const bool acceptOk = co_await awaiter;
    if (!acceptOk) {
      // At this point, do not touch socket_->data anymore!
      // This is the result of ConnectionAwaiter_::stop(), and
      // data points to a CloseAwaiter_ object.
      break;
    }
 
    for (auto it = awaiter.accepted_.begin(); it != awaiter.accepted_.end();
         it++) {
      auto &streamSlot = *it;
 
      if (streamSlot.index() == 0) {
        const uv_status status = std::get<0>(streamSlot);
        BOOST_ASSERT(status != 0);
        // When the error is handled, the user code can again call listen()
        // and will process the remaining connections. Therefore, first remove
        // the already processed connections.
        awaiter.accepted_.erase(awaiter.accepted_.begin(), it);
        setData(socket_.get(), (void *)nullptr);
        throw UvcoException{status,
                            "UnixStreamServer failed to accept a connection!"};
      } else {
        // Awkward handlign: if the returned MultiPromise is dropped, we will
        // never return from co_yield. However, dropping may happen after
        // calling `close()`, so we cannot rely on socket_ still existing.
        //
        // `close()` also relies on whether `socket_->data` is `nullptr` or not
        // to decide if the socket has been closed already.
        setData(socket_.get(), (void *)nullptr);
        co_yield std::move(std::get<1>(streamSlot));
        setData(socket_.get(), &awaiter);
      }
    }
    awaiter.accepted_.clear();
  }
  setData(socket_.get(), (void *)nullptr);
}

onNewConnection()

template<typename UvStreamType , typename StreamType >

void uvco::StreamServerBase< UvStreamType, StreamType >::onNewConnection ( uv_stream_t *  stream, uv_status  status  )

staticprivate

                                           {
  const auto *server = (UvStreamType *)stream;
  auto *connectionAwaiter = getData<ConnectionAwaiter_>(server);
  uv_loop_t *const loop = connectionAwaiter->socket_.loop;
 
  if (status == 0) {
    auto clientStream = std::make_unique<UvStreamType>();
    UvStreamInitHelper<UvStreamType>::init(loop, clientStream.get());
    const uv_status acceptStatus =
        uv_accept((uv_stream_t *)server, (uv_stream_t *)clientStream.get());
    if (acceptStatus == 0) {
      connectionAwaiter->accepted_.emplace_back(
          StreamType{std::move(clientStream)});
    } else {
      connectionAwaiter->accepted_.emplace_back(acceptStatus);
    }
  } else {
    connectionAwaiter->accepted_.emplace_back(status);
  }
 
  if (connectionAwaiter->handle_) {
    Loop::enqueue(*connectionAwaiter->handle_);
    connectionAwaiter->handle_.reset();
  }
}

operator=() [1/2]

template<typename UvStreamType , typename StreamType >

StreamServerBase & uvco::StreamServerBase< UvStreamType, StreamType >::operator= ( const StreamServerBase< UvStreamType, StreamType > &  )

delete

operator=() [2/2]

template<typename UvStreamType , typename StreamType >

StreamServerBase & uvco::StreamServerBase< UvStreamType, StreamType >::operator= ( StreamServerBase< UvStreamType, StreamType > &&  )

default

Member Data Documentation

socket_

template<typename UvStreamType , typename StreamType >

std::unique_ptr<UvStreamType> uvco::StreamServerBase< UvStreamType, StreamType >::socket_

protected

---

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

• uvco/stream_server_base.h
• uvco/stream_server_base_impl.cc