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 <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 <coroutine>
#include <cstdio>
#include <memory>
#include <optional>
#include <string>
#include <utility>
 
namespace uvco {


class StreamBase {
  struct ShutdownAwaiter_;
  struct InStreamAwaiter_;
  struct OutStreamAwaiter_;
 
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 = 4080;

  [[nodiscard]] Promise<std::optional<std::string>>
  read(size_t maxSize = defaultMaxReadSize);

  Promise<size_t> read(std::span<char> buffer);

  [[nodiscard]] Promise<void> write(std::string buf);

  [[nodiscard]] Promise<void> shutdown();

  void close();

  [[nodiscard]] const uv_stream_t *underlying() const { return stream_.get(); }
 
  [[nodiscard]] uv_os_fd_t fd() const;
 
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::coroutine_handle<> reader_;
  std::coroutine_handle<> writer_;
};

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