tcp: Added virtual Stream class for exposing a more general TCP Connection

Author: AndreasAakesson

api/net/http/client_connection.hpp

@@ -36,7 +36,7 @@ namespace http {
     using timeout_duration  = std::chrono::milliseconds;
 
   public:
-    explicit Client_connection(Client&, TCP_conn);
+    explicit Client_connection(Client&, Stream_ptr);
 
     bool available() const
     { return on_response_ == nullptr && keep_alive_; }

api/net/http/connection.hpp

@@ -29,29 +29,30 @@ namespace http {
 
   class Connection {
   public:
-    using TCP_conn      = net::tcp::Connection_ptr;
+    using Stream        = net::tcp::Connection::Stream;
+    using Stream_ptr    = std::unique_ptr<Stream>;
     using Peer          = net::tcp::Socket;
     using buffer_t      = net::tcp::buffer_t;
 
   public:
-    inline explicit Connection(TCP_conn tcpconn, bool keep_alive = true);
+    inline explicit Connection(Stream_ptr stream, bool keep_alive = true);
 
     template <typename TCP>
     explicit Connection(TCP&, Peer);
 
     inline explicit Connection() noexcept;
 
     net::tcp::port_t local_port() const noexcept
-    { return (tcpconn_) ? tcpconn_->local_port() : 0; }
+    { return (stream_) ? stream_->local_port() : 0; }
 
     Peer peer() const noexcept
     { return peer_; }
 
     void timeout()
-    { tcpconn_->is_closing() ? tcpconn_->abort() : tcpconn_->close(); }
+    { stream_->is_closing() ? stream_->abort() : stream_->close(); }
 
-    auto&& tcp() const
-    { return tcpconn_; }
+    auto& stream() const
+    { return stream_; }
 
     /**
      * @brief      Shutdown the underlying TCP connection
@@ -64,15 +65,15 @@ namespace http {
      *
      * @return     The underlying TCP connection
      */
-    inline TCP_conn release();
+    inline Stream_ptr release();
 
     /**
      * @brief      Whether the underlying TCP connection has been released or not
      *
      * @return     true if the underlying TCP connection is released
      */
     bool released() const
-    { return tcpconn_ == nullptr; }
+    { return stream_ == nullptr; }
 
     static Connection& empty() noexcept
     {
@@ -101,52 +102,50 @@ namespace http {
     virtual ~Connection() {}
 
   protected:
-    TCP_conn          tcpconn_;
+    Stream_ptr        stream_;
     bool              keep_alive_;
     Peer              peer_;
 
     virtual void close() {}
 
   }; // < class Connection
 
-  inline Connection::Connection(TCP_conn tcpconn, bool keep_alive)
-    : tcpconn_{std::move(tcpconn)},
+  inline Connection::Connection(Stream_ptr stream, bool keep_alive)
+    : stream_{std::move(stream)},
       keep_alive_{keep_alive},
-      peer_{tcpconn_->remote()}
+      peer_{stream_->remote()}
   {
-    Ensures(tcpconn_ != nullptr);
+    Ensures(stream_ != nullptr);
     debug("<http::Connection> Created %u -> %s %p\n", local_port(), peer().to_string().c_str(), this);
   }
 
   template <typename TCP>
   Connection::Connection(TCP& tcp, Peer addr)
-    : Connection(tcp.connect(addr))
+    : Connection(std::make_unique<Stream>(tcp.connect(addr)))
   {
   }
 
   inline Connection::Connection() noexcept
-    : tcpconn_(nullptr),
+    : stream_(nullptr),
       keep_alive_(false),
       peer_{}
   {
   }
 
   inline void Connection::shutdown()
   {
-    if(not released() and not tcpconn_->is_closing())
-      tcpconn_->close();
+    if(not released() and not stream_->is_closing())
+      stream_->close();
   }
 
-  inline Connection::TCP_conn Connection::release()
+  inline Connection::Stream_ptr Connection::release()
   {
-    auto copy = tcpconn_;
+    auto copy = std::move(stream_);
 
     // this is expensive and may be unecessary,
     // but just to be safe for now
     copy->reset_callbacks();
 
-    tcpconn_ = nullptr;
-
     return copy;
   }
 

api/net/http/server.hpp

@@ -63,7 +63,7 @@ namespace http {
      *
      * @param[in]  port  The port to listen on
      */
-    void listen(uint16_t port);
+    virtual void listen(uint16_t port);
 
     /**
      * @brief      Setup handler for when a Request is received
@@ -90,21 +90,6 @@ namespace http {
      */
     Response_ptr create_response(status_t code = http::OK) const;
 
-    /**
-     * @brief      Returns a vector of all TCP connections currently connected to the server
-     *
-     * @return     A vector of TCP connections
-     */
-    std::vector<TCP_conn> active_tcp_connections() const;
-
-    /**
-     * @brief      Reconnects a TCP connection by creating a Server connection
-     *
-     * @param[in]  conn  The TCP connection
-     */
-    void reconnect(TCP_conn conn)
-    { if (conn != nullptr) connect(conn); }
-
     ~Server();
 
   private:
@@ -124,7 +109,10 @@ namespace http {
     Stat& stat_req_bad_;
     Stat& stat_timeouts_;
 
-    void connect(TCP_conn conn);
+    void connect(TCP_conn conn)
+    { connect(std::make_unique<Connection::Stream>(conn)); }
+
+    void connect(Connection::Stream_ptr stream);
 
     void close(Server_connection&);
 

api/net/http/server_connection.hpp

@@ -33,7 +33,7 @@ namespace http {
     static constexpr size_t DEFAULT_BUFSIZE = 1460;
 
   public:
-    explicit Server_connection(Server&, TCP_conn, size_t idx, const size_t bufsize = DEFAULT_BUFSIZE);
+    explicit Server_connection(Server&, Stream_ptr, size_t idx, const size_t bufsize = DEFAULT_BUFSIZE);
 
     void send(Response_ptr res);
 

api/net/tcp/connection.hpp

@@ -56,6 +56,9 @@ class Connection : public std::enable_shared_from_this<Connection> {
   struct Disconnect;
   /** Reason for packet being dropped */
   enum class Drop_reason;
+  /** A Connection stream */
+  class Stream;
+
   using Byte = uint8_t;
 
   using WriteBuffer = Write_queue::WriteBuffer;
@@ -214,6 +217,205 @@ class Connection : public std::enable_shared_from_this<Connection> {
    */
   inline void abort();
 
+  /**
+   * @brief      Exposes a TCP Connection as a Stream with only the most necessary features.
+   *             May be overrided by extensions like TLS etc for additional functionality.
+   */
+  class Stream {
+  public:
+    /**
+     * @brief      Construct a Stream for a Connection ptr
+     *
+     * @param[in]  conn  The connection
+     */
+    Stream(Connection_ptr conn)
+      : ptr{std::move(conn)}
+    {}
+
+    /** Called when the stream is ready to use. */
+    using ConnectCallback = delegate<void(Stream& self)>;
+    /**
+     * @brief      Event when the stream is connected/established/ready to use.
+     *
+     * @param[in]  cb    The connect callback
+     */
+    virtual void on_connect(ConnectCallback cb)
+    {
+      ptr->on_connect(Connection::ConnectCallback::make_packed(
+          [this, cb] (Connection_ptr)
+          { cb(*this); }));
+    }
+
+    /** Called with a shared buffer and the length of the data when received. */
+    using ReadCallback            = delegate<void(buffer_t, size_t)>;
+    /**
+     * @brief      Event when data is received.
+     *
+     * @param[in]  n     The size of the receive buffer
+     * @param[in]  cb    The read callback
+     */
+    virtual void on_read(size_t n, ReadCallback cb)
+    { ptr->on_read(n, cb); }
+
+    /** Called with nothing ¯\_(ツ)_/¯ */
+    using CloseCallback           = delegate<void()>;
+    /**
+     * @brief      Event for when the Stream is being closed.
+     *
+     * @param[in]  cb    The close callback
+     */
+    virtual void on_close(CloseCallback cb)
+    { ptr->on_close(cb); }
+
+    /** Called with the number of bytes written. */
+    using WriteCallback           = delegate<void(size_t)>;
+    /**
+     * @brief      Event for when data has been written.
+     *
+     * @param[in]  cb    The write callback
+     */
+    virtual void on_write(WriteCallback cb)
+    { ptr->on_write(cb); }
+
+    /**
+     * @brief      Async write of a data with a length.
+     *
+     * @param[in]  buf   data
+     * @param[in]  n     length
+     */
+    virtual void write(const void* buf, size_t n)
+    { ptr->write(buf, n); }
+
+    /**
+     * @brief      Async write of a chunk.
+     *
+     * @param[in]  c     A chunk
+     */
+    virtual void write(Chunk c)
+    { ptr->write(c); }
+
+    /**
+     * @brief      Async write of a shared buffer with a length.
+     *             Calls write(Chunk c).
+     *
+     * @param[in]  buffer  shared buffer
+     * @param[in]  n       length
+     */
+    virtual void write(buffer_t buf, size_t n)
+    { ptr->write(buf, n); }
+
+    /**
+     * @brief      Async write of a string.
+     *             Calls write(const void* buf, size_t n)
+     *
+     * @param[in]  str   The string
+     */
+    void write(const std::string& str)
+    { write(str.data(), str.size()); }
+
+    /**
+     * @brief      Closes the stream.
+     */
+    virtual void close()
+    { ptr->close(); }
+
+    /**
+     * @brief      Aborts (terminates) the stream.
+     */
+    virtual void abort()
+    { ptr->abort(); }
+
+    /**
+     * @brief      Resets all callbacks.
+     */
+    virtual void reset_callbacks()
+    { ptr->reset_callbacks(); }
+
+    /**
+     * @brief      Returns the streams local socket.
+     *
+     * @return     A TCP Socket
+     */
+    tcp::Socket local() const
+    { return ptr->local(); }
+
+    /**
+     * @brief      Returns the streams remote socket.
+     *
+     * @return     A TCP Socket
+     */
+    tcp::Socket remote() const
+    { return ptr->remote(); }
+
+    /**
+     * @brief      Returns the local port.
+     *
+     * @return     A TCP port
+     */
+    uint16_t local_port() const
+    { return ptr->local_port(); }
+
+    /**
+     * @brief      Returns a string representation of the stream.
+     *
+     * @return     String representation of the stream.
+     */
+    virtual std::string to_string() const noexcept
+    { return ptr->to_string(); }
+
+    /**
+     * @brief      Determines if connected (established).
+     *
+     * @return     True if connected, False otherwise.
+     */
+    virtual bool is_connected() const noexcept
+    { return ptr->is_connected(); }
+
+    /**
+     * @brief      Determines if writable. (write is allowed)
+     *
+     * @return     True if writable, False otherwise.
+     */
+    virtual bool is_writable() const noexcept
+    { return ptr->is_writable(); }
+
+    /**
+     * @brief      Determines if readable. (data can be received)
+     *
+     * @return     True if readable, False otherwise.
+     */
+    virtual bool is_readable() const noexcept
+    { return ptr->is_readable(); }
+
+    /**
+     * @brief      Determines if closing.
+     *
+     * @return     True if closing, False otherwise.
+     */
+    virtual bool is_closing() const noexcept
+    { return ptr->is_closing(); }
+
+    /**
+     * @brief      Determines if closed.
+     *
+     * @return     True if closed, False otherwise.
+     */
+    virtual bool is_closed() const noexcept
+    { return ptr->is_closed(); };
+
+  protected:
+    /**
+     * @brief      Returns the underlying TCP connection.
+     *
+     * @return     A TCP Connection ptr
+     */
+    Connection_ptr tcp()
+    { return ptr; };
+
+  private:
+    Connection_ptr ptr;
+
+  }; // < class Connection::Stream
 
   /**
    * @brief      Reason for disconnect event.