stream.h

Go to the documentation of this file.

// uvco (c) 2023 Lewin Bormann. See LICENSE for specific terms.
 
#pragma once
 
#include <boost/assert.hpp>
#include <fmt/core.h>
#include <span>
#include <string_view>
#include <uv.h>
#include <uv/unix.h>
 
#include "uvco/close.h"
#include "uvco/internal/internal_utils.h"
#include "uvco/loop/loop.h"
#include "uvco/promise/promise.h"
#include "uvco/run.h"
 
#include <array>
#include <coroutine>
#include <cstdio>
#include <memory>
#include <optional>
#include <string>
#include <utility>
 
namespace uvco {
 
 
class StreamBase {
public:
  template <typename Stream>
  explicit StreamBase(std::unique_ptr<Stream> stream)
      : stream_{(uv_stream_t *)stream.release()} {}
  StreamBase(const StreamBase &) = delete;
  StreamBase(StreamBase &&) = default;
  StreamBase &operator=(const StreamBase &) = delete;
  StreamBase &operator=(StreamBase &&) = default;
  virtual ~StreamBase();
 
  static constexpr size_t defaultMaxReadSize = 4096;
 
  [[nodiscard]] Promise<std::optional<std::string>>
  read(size_t maxSize = defaultMaxReadSize);
 
  Promise<size_t> read(std::span<char> buffer);
 
  [[nodiscard]] Promise<size_t> write(std::string buf);
 
  [[nodiscard]] Promise<void> shutdown();
 
  [[nodiscard]] Promise<void> close();
 
  [[nodiscard]] const uv_stream_t *underlying() const { return stream_.get(); }
 
protected:
  uv_stream_t &stream() {
    BOOST_ASSERT(stream_);
    return *stream_;
  }
  void destroyStream() {
    BOOST_ASSERT(stream_);
    stream_.reset();
  }
 
private:
  std::unique_ptr<uv_stream_t, UvHandleDeleter> stream_;
  // Currently suspended readers/writers to be notified on close().
  std::optional<std::coroutine_handle<>> reader_;
  std::optional<std::coroutine_handle<>> writer_;
 
  struct ShutdownAwaiter_ {
    ShutdownAwaiter_() = default;
    static void onShutdown(uv_shutdown_t *req, uv_status status);
 
    bool await_ready();
    bool await_suspend(std::coroutine_handle<> handle);
    void await_resume();
 
    std::optional<std::coroutine_handle<>> handle_;
    std::optional<uv_status> status_;
  };
 
  struct InStreamAwaiter_ {
    explicit InStreamAwaiter_(StreamBase &stream, std::span<char> buffer)
        : stream_{stream}, buffer_{buffer} {}
 
    bool await_ready();
    bool await_suspend(std::coroutine_handle<> handle);
    size_t await_resume();
 
    void start_read();
    void stop_read();
 
    static void allocate(uv_handle_t *handle, size_t suggested_size,
                         uv_buf_t *buf);
    static void onInStreamRead(uv_stream_t *stream, ssize_t nread,
                               const uv_buf_t *buf);
 
    StreamBase &stream_;
    std::span<char> buffer_;
    std::optional<ssize_t> status_;
    std::optional<std::coroutine_handle<>> handle_;
  };
 
  struct OutStreamAwaiter_ {
    OutStreamAwaiter_(StreamBase &stream, std::string_view buffer);
 
    [[nodiscard]] std::array<uv_buf_t, 1> prepare_buffers() const;
 
    bool await_ready();
    bool await_suspend(std::coroutine_handle<> handle);
    uv_status await_resume();
 
    static void onOutStreamWrite(uv_write_t *write, uv_status status);
 
    std::optional<std::coroutine_handle<>> handle_;
    std::optional<uv_status> status_;
 
    // State necessary for both immediate and delayed writing.
    std::string_view buffer_;
    uv_write_t write_{};
    StreamBase &stream_;
  };
};
 
class TtyStream : public StreamBase {
public:
  // Takes ownership of stream.
  TtyStream(TtyStream &&other) = default;
  TtyStream(const TtyStream &other) = delete;
  TtyStream &operator=(TtyStream &&) = default;
  TtyStream &operator=(const TtyStream &) = delete;
  ~TtyStream() override = default;
 
  static TtyStream tty(const Loop &loop, int fd);
  static TtyStream stdin(const Loop &loop) { return tty(loop, 0); }
  static TtyStream stdout(const Loop &loop) { return tty(loop, 1); }
  static TtyStream stderr(const Loop &loop) { return tty(loop, 2); }
 
private:
  explicit TtyStream(std::unique_ptr<uv_tty_t> stream)
      : StreamBase{std::move(stream)} {}
};
 
 
} // namespace uvco