uvco::StreamBase Class Reference

A plain stream, permitting reading, writing, and closing. More...

#include <stream.h>

Inheritance diagram for uvco::StreamBase:

Classes
struct	InStreamAwaiter_
struct	OutStreamAwaiter_
struct	ShutdownAwaiter_

Public Member Functions
template<typename Stream>
	StreamBase (std::unique_ptr< Stream > stream)
	StreamBase (const StreamBase &)=delete
	StreamBase (StreamBase &&)=default
StreamBase &	operator= (const StreamBase &)=delete
StreamBase &	operator= (StreamBase &&)=default
virtual	~StreamBase ()
Promise< std::optional< std::string > >	read (size_t maxSize=defaultMaxReadSize)
Promise< size_t >	read (std::span< char > buffer)
Promise< void >	write (std::string buf)
Promise< void >	shutdown ()
void	close ()
const uv_stream_t *	underlying () const
	Return the underlying UV stream object.
uv_os_fd_t	fd () const

Static Public Attributes
static constexpr size_t	defaultMaxReadSize = 4080

Protected Member Functions
uv_stream_t &	stream ()
void	destroyStream ()

Private Attributes
std::unique_ptr< uv_stream_t, UvHandleDeleter >	stream_
std::coroutine_handle	reader_
std::coroutine_handle	writer_

Detailed Description

A plain stream, permitting reading, writing, and closing.

Constructor & Destructor Documentation

StreamBase() [1/3]

template<typename Stream>

uvco::StreamBase::StreamBase ( std::unique_ptr< Stream > stream )

inlineexplicit

      : stream_{(uv_stream_t *)stream.release()} {}

StreamBase() [2/3]

uvco::StreamBase::StreamBase ( const StreamBase & )

delete

StreamBase() [3/3]

uvco::StreamBase::StreamBase ( StreamBase && )

default

~StreamBase()

uvco::StreamBase::~StreamBase ( )

virtual

Closes the stream if not already closed. It's always best to explicitly call co_await stream.close(), but as a backup, the destructor will close the stream if still open. To do so, it will schedule a callback on the libuv loop.

{ close(); }

Member Function Documentation

close()

void uvco::StreamBase::close

(

)

Informs pending readers and writers of the close and causes them to return an empty optional.

                       {
  if (stream_ != nullptr) {
    closeHandle(stream_.release());
  }
  if (reader_ != nullptr) {
    std::coroutine_handle<> reader = reader_;
    reader_ = nullptr;
    reader.resume();
  }
  if (writer_ != nullptr) {
    std::coroutine_handle<> writer = writer_;
    writer_ = nullptr;
    writer.resume();
  }
}

destroyStream()

void uvco::StreamBase::destroyStream ( )

inlineprotected

                       {
    BOOST_ASSERT(stream_);
    stream_.reset();
  }

fd()

uv_os_fd_t uvco::StreamBase::fd ( ) const

nodiscard

                                              {
  uv_os_fd_t fd{};
  const uv_status status = uv_fileno((uv_handle_t *)stream_.get(), &fd);
  if (status != 0) {
    throw UvcoException(status, "StreamBase::fd(): uv_fileno() failed: " +
                                    std::string{uv_strerror(status)});
  }
  return fd;
}

operator=() [1/2]

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

delete

operator=() [2/2]

StreamBase & uvco::StreamBase::operator= ( StreamBase && )

default

read() [1/2]

Promise< std::optional< std::string > > uvco::StreamBase::read ( size_t maxSize = defaultMaxReadSize )

nodiscard

Read available data (up to maxSize bytes) from stream. Returns std::nullopt on EOF or closed handle (close()).

Throws UvcoException on error.

NOTE: Consider using read(std::span<char>) for better performance.

NOTE: only one reader is allowed to be active at a time. If a read is started while another is still active, uvco will abort the process (in Debug mode), or ignore the first read (in Release mode).

                                                               {
  // This is a promise root function, i.e. origin of a promise.
  std::string buf(maxSize, '\0');
  InStreamAwaiter_ awaiter{*this, buf};
  const size_t nRead = co_await awaiter;
  if (nRead == 0) {
    // EOF.
    co_return std::nullopt;
  }
  buf.resize(nRead);
  co_return buf;
}

read() [2/2]

Promise< size_t > uvco::StreamBase::read

(

std::span< char >

buffer

)

Read available data (up to buffer.size() bytes) from stream. Returns the number of bytes read, or 0 on EOF or closed handle (close()).

Only one read() coroutine may be active at a time. The stream must outlive the coroutine, i.e. live until co_await stream.read(...) returns a result.

Throws UvcoException on error.

                                                     {
  InStreamAwaiter_ awaiter{*this, buffer};
  co_return (co_await awaiter);
}

shutdown()

Promise< void > uvco::StreamBase::shutdown ( )

nodiscard

Shut down stream for writing. This is a half-close; the other side can still write. The result of shutdown() must be co_awaited.

                                   {
  auto shutdownReq = std::make_unique<uv_shutdown_t>();
  ShutdownAwaiter_ awaiter;
  const OnExit _onExit{[req = shutdownReq.get(), &awaiter] {
    if (awaiter.handle_) {
      resetRequestData(req);
    }
  }};
 
  setRequestData(shutdownReq.get(), &awaiter);
  uv_shutdown(shutdownReq.release(), &stream(),
              StreamBase::ShutdownAwaiter_::onShutdown);
  co_await awaiter;
  co_return;
}

stream()

uv_stream_t & uvco::StreamBase::stream ( )

inlineprotected

                        {
    BOOST_ASSERT(stream_);
    return *stream_;
  }

underlying()

const uv_stream_t * uvco::StreamBase::underlying ( ) const

inlinenodiscard

Return the underlying UV stream object.

{ return stream_.get(); }

write()

Promise< void > uvco::StreamBase::write ( std::string buf )

nodiscard

Write a buffer to the stream. A copy of buf is taken because it is undetermined when the actual write will occur. Await the result if the status is important; the write will be executed even without awaiting (as long as the process keeps running).

NOTE: only one writer is allowed to be active at a time. If two writes are started simultaneously, the process will be aborted in Debug mode, or the first write() coroutine will not return in Release mode.

WARNING: due to the interactions with libuv, writes cannot be cancelled. That's why write() always takes ownership of a buffer to avoid use-after-free issues. The safety comes at the cost of increased allocations, obviously.

                                             {
  OutStreamAwaiter_ awaiter{*this, std::move(buf)};
  std::array<uv_buf_t, 1> bufs{};
  bufs[0] = uv_buf_init(const_cast<char *>(awaiter.write_->buffer.data()),
                        awaiter.write_->buffer.size());
 
  setRequestData(awaiter.write_.get(), &awaiter);
  const OnExit _onExit{[write = awaiter.write_.get(), &awaiter] {
    // handle_ is reset by onOutStreamWrite callback. If it's non-null, it's
    // because write() was cancelled.
    if (awaiter.handle_) {
      resetRequestData(write);
    }
  }};
 
  const uv_status status =
      uv_write((uv_write_t *)(awaiter.write_.release()), &stream(), bufs.data(),
               bufs.size(), OutStreamAwaiter_::onOutStreamWrite);
  if (status < 0) {
    throw UvcoException{status,
                        "StreamBase::write() encountered error in uv_write"};
  }
  const uv_status completionStatus = co_await awaiter;
  // may be UV_ECANCELED if the stream has been dropped.
  if (completionStatus < 0) {
    throw UvcoException{
        status, "StreamBase::write() encountered error while awaiting write"};
  }
  co_return;
}

Member Data Documentation

defaultMaxReadSize

size_t uvco::StreamBase::defaultMaxReadSize = 4080

staticconstexpr

reader_

std::coroutine_handle uvco::StreamBase::reader_

private

stream_

std::unique_ptr<uv_stream_t, UvHandleDeleter> uvco::StreamBase::stream_

private

writer_

std::coroutine_handle uvco::StreamBase::writer_

private

---

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

• uvco/stream.h
• uvco/stream.cc