#pragma once

#include <memory>

#include "envoy/api/io_error.h"

#include "envoy/api/os_sys_calls.h"

#include "envoy/common/platform.h"

#include "envoy/event/dispatcher.h"

#include "envoy/network/address.h"

#include "envoy/network/io_handle.h"

#include "source/common/buffer/watermark_buffer.h"

#include "source/common/common/logger.h"

#include "source/common/network/io_socket_error_impl.h"

#include "source/extensions/io_socket/user_space/file_event_impl.h"

#include "source/extensions/io_socket/user_space/io_handle.h"

namespace Envoy {

namespace Extensions {

namespace IoSocket {

namespace UserSpace {

// Align with Buffer::Reservation::MAX_SLICES_

constexpr uint32_t MAX_FRAGMENT = 8;

// Align with Buffer::Slice::default_slice_size_

constexpr uint64_t FRAGMENT_SIZE = 16 * 1024;

/**

 * Network::IoHandle implementation which provides a buffer as data source. It is designed to used

 * by Network::ConnectionImpl. Some known limitations include

 * 1. It doesn't include a file descriptor. Do not use "fdDoNotUse".

 * 2. It doesn't support socket options. Wrap this in ConnectionSocket and implement the socket

 * getter/setter options.

 * 3. It doesn't support UDP interface.

 * 4. The peer BufferedIoSocket must be scheduled in the same thread to avoid data race because

 *    IoHandleImpl mutates the state of peer handle and no lock is introduced.

 */

class IoHandleImpl final : public Network::IoHandle,

                           public UserSpace::IoHandle,

                           protected Logger::Loggable<Logger::Id::io>,

                           NonCopyable {

public:

  ~IoHandleImpl() override;

  // Network::IoHandle

  Api::IoCallUint64Result close() override;

  bool isOpen() const override;

  bool wasConnected() const override;

  Api::IoCallUint64Result readv(uint64_t max_length, Buffer::RawSlice* slices,

                                uint64_t num_slice) override;

  Api::IoCallUint64Result read(Buffer::Instance& buffer,

                               absl::optional<uint64_t> max_length_opt) override;

  Api::IoCallUint64Result writev(const Buffer::RawSlice* slices, uint64_t num_slice) override;

  Api::IoCallUint64Result write(Buffer::Instance& buffer) override;

  Api::IoCallUint64Result sendmsg(const Buffer::RawSlice* slices, uint64_t num_slice, int flags,

                                  const Network::Address::Ip* self_ip,

                                  const Network::Address::Instance& peer_address) override;

  Api::IoCallUint64Result recvmsg(Buffer::RawSlice* slices, const uint64_t num_slice,

                                  uint32_t self_port,

                                  const Network::IoHandle::UdpSaveCmsgConfig& udp_save_cmsg_config,

                                  RecvMsgOutput& output) override;

  Api::IoCallUint64Result recvmmsg(RawSliceArrays& slices, uint32_t self_port,

                                   const Network::IoHandle::UdpSaveCmsgConfig& udp_save_cmsg_config,

                                   RecvMsgOutput& output) override;

  Api::IoCallUint64Result recv(void* buffer, size_t length, int flags) override;

  bool supportsMmsg() const override;

  bool supportsUdpGro() const override;

  Api::SysCallIntResult bind(Network::Address::InstanceConstSharedPtr address) override;

  Api::SysCallIntResult listen(int backlog) override;

  Network::IoHandlePtr accept(struct sockaddr* addr, socklen_t* addrlen) override;

  Api::SysCallIntResult connect(Network::Address::InstanceConstSharedPtr address) override;

  Api::SysCallIntResult setOption(int level, int optname, const void* optval,

                                  socklen_t optlen) override;

  Api::SysCallIntResult getOption(int level, int optname, void* optval, socklen_t* optlen) override;

  Api::SysCallIntResult ioctl(unsigned long, void*, unsigned long, void*, unsigned long,

                              unsigned long*) override;

  Api::SysCallIntResult setBlocking(bool blocking) override;

  absl::optional<int> domain() override;

  absl::StatusOr<Network::Address::InstanceConstSharedPtr> localAddress() override;

  absl::StatusOr<Network::Address::InstanceConstSharedPtr> peerAddress() override;

  void initializeFileEvent(Event::Dispatcher& dispatcher, Event::FileReadyCb cb,

                           Event::FileTriggerType trigger, uint32_t events) override;

  Network::IoHandlePtr duplicate() override;

  void activateFileEvents(uint32_t events) override;

  void enableFileEvents(uint32_t events) override;

  void resetFileEvents() override;

  Api::SysCallIntResult shutdown(int how) override;

    // Low to high is checked by peer after peer populates the receive buffer.

  // UserSpace::IoHandle

          // Closed ready type is defined as `end of stream`

  // `UserspaceIoHandle`

  // Set the peer which will populate the owned pending_received_data.

    // Swapping peers is undefined behavior.

private:

  friend class IoHandleFactory;

  explicit IoHandleImpl(PassthroughStateSharedPtr passthrough_state = nullptr);

  static const Network::Address::InstanceConstSharedPtr& getCommonInternalAddress();

  // Support isOpen() and close(). Network::IoHandle owner must invoke close() to avoid potential

  // resource leak.

  bool closed_{false};

  // The attached file event with this socket.

  std::unique_ptr<FileEventImpl> user_file_event_;

  // True if pending_received_data_ is not addable. Note that pending_received_data_ may have

  // pending data to drain.

  bool receive_data_end_stream_{false};

  // The buffer owned by this socket. This buffer is populated by the write operations of the peer

  // socket and drained by read operations of this socket.

  Buffer::WatermarkBuffer pending_received_data_;

  // write() calls will populate the receive buffer of the peer handle. Guaranteed to be non-null

  // until close() is called on either this handle or the peer handle.

  UserSpace::IoHandle* peer_handle_{nullptr};

  // Indicates whether this handle has sent EOF to the peer by calling setEof().

  bool sent_eof_{false};

  // Shared state between peer handles.

  PassthroughStateSharedPtr passthrough_state_{nullptr};

};

class PassthroughStateImpl : public PassthroughState, public Logger::Loggable<Logger::Id::io> {

public:

  void initialize(std::unique_ptr<envoy::config::core::v3::Metadata> metadata,

                  const StreamInfo::FilterState::Objects& filter_state_objects) override;

  void mergeInto(envoy::config::core::v3::Metadata& metadata,

                 StreamInfo::FilterState& filter_state) override;

protected:

  enum class State { Created, Initialized, Done };

  State state_{State::Created};

  std::unique_ptr<envoy::config::core::v3::Metadata> metadata_;

  StreamInfo::FilterState::Objects filter_state_objects_;

};

using IoHandleImplPtr = std::unique_ptr<IoHandleImpl>;

class IoHandleFactory {

public:

  /**

   * @return a pair of connected IoHandleImpl instances.

   * @param state optional existing value to use as the shared PassthroughState. If omitted, a

   * newly constructed PassthroughStateImpl will be used.

   */

  static std::pair<IoHandleImplPtr, IoHandleImplPtr>

  createIoHandlePair(PassthroughStatePtr state = nullptr);

  /**

   * @return a pair of connected IoHandleImpl instances with pre-configured watermarks.

   * @param buffer_size buffer watermark size in bytes

   * @param state optional existing value to use as the shared PassthroughState. If omitted, a

   * newly constructed PassthroughStateImpl will be used.

   */

  static std::pair<IoHandleImplPtr, IoHandleImplPtr>

  createBufferLimitedIoHandlePair(uint32_t buffer_size, PassthroughStatePtr state = nullptr);

};

} // namespace UserSpace

} // namespace IoSocket

} // namespace Extensions

} // namespace Envoy