Refactors the parser so it is not header-only.

Author: mzimbres

README.md

@@ -42,21 +42,30 @@ auto async_main() -> net::awaitable<void>
 ```
 
 The execution of `connection::async_exec` above is composed with
-`connection::async_run` with the aid of the Asio awaitable operator ||
+`connection::async_run` with the aid of the Asio awaitable `operator ||`
 that ensures that  one operation is cancelled as soon as the other
 completes, these functions play the following roles
 
-* `connection::async_exec`: Execute commands (i.e. write the request and reads the response).
-* `connection::async_run`: Coordinate read and write operations and remains suspended until the connection is lost.
+* `connection::async_exec`: Execute commands by writing the request payload to the underlying stream and reading the response sent back by Redis. It can be called from multiple places in your code concurrently.
+* `connection::async_run`: Coordinate the low-level IO (read and write) operations and remains suspended until the connection is lost.
 
-Let us dig in.
+When a connection is lost, the `async_exec` calls won't automatically
+fail, instead, they will remain suspended until they are either all
+canceled with a call to `connection::cancel(operation::exec)` or a new
+connection is established and `async_run` is called again, in which
+case they will be resent automatically. Users can customise this
+behaviour by carefully choosing the values of
+`aedis::resp3::request::config`.  The role played by `async_run`
+becomes clearer with long-lived connections, which we will cover
+in the next section.
 
 <a name="connection"></a>
 ## Connection
 
-In general we will want to reuse the same connection for multiple
-requests, we can do this with the example above by decoupling the
-HELLO command and the call to `async_run` in a separate coroutine
+For performance reasons we will usually want to perform multiple
+requests on the same connection. We can do this with the example above
+by decoupling the HELLO command and the call to `async_run` in a
+separate coroutine
 
 ```cpp
 auto run(std::shared_ptr<connection> conn) -> net::awaitable<void>
@@ -67,23 +76,33 @@ auto run(std::shared_ptr<connection> conn) -> net::awaitable<void>
    req.push("HELLO", 3); // Upgrade to RESP3
 
    // Notice we use && instead of || so async_run is not cancelled
-   // when the response to HELLO comes.
+   // when the HELLO response arrives. We are also ignoring the
+   // response for simplicity.
    co_await (conn->async_run() && conn->async_exec(req));
 }
 ```
 We can now let `run` run detached in the background while other
-coroutines perform requests on the connection
+coroutines perform requests on the connection, for example
 
 ```cpp
 auto async_main() -> net::awaitable<void>
 {
    auto conn = std::make_shared<connection>(co_await net::this_coro::executor);
 
-   // Calls async_run detached.
-   net::co_spawn(ex, run(conn), net::detached)
+   // Run detached.
+   net::co_spawn(ex, run(conn), net::detached);
+
+   // Here we can use the connection to perform requests and pass it
+   // around to other coroutines so they can make requests.
+
+   resp3::request req;
+   req.push("PING", "Hello world");
+   co_await conn->async_exec(req);
 
-   // Here we can pass conn around to other coroutines so they can make requests.
    ...
+
+   // Cancels the run operation so we can exit.
+   conn->cancel(operation::run);
 }
 ```
 
@@ -138,10 +157,12 @@ auto run(std::shared_ptr<connection> conn) -> net::awaitable<void>
 
    for (;;) {
       co_await connect(conn, "127.0.0.1", "6379");
-      co_await ((conn->async_run() || healthy_checker(conn) || receiver(conn))
-                && conn->async_exec(req));
+      co_await ((conn->async_run() || healthy_checker(conn) || receiver(conn)) && conn->async_exec(req));
 
+      // Prepare the stream to a new connection.
       conn->reset_stream();
+
+      // Waits one second before trying to reconnect.
       timer.expires_after(std::chrono::seconds{1});
       co_await timer.async_wait();
    }
@@ -152,23 +173,23 @@ For failover with sentinels see `resolve_with_sentinel.cpp`.  At
 this point the reasons for why `async_run` was introduced in Aedis
 might have become apparent to the reader
 
-* Provide quick reaction to disconnections and hence faster failover.
+* Provide quick reaction to disconnections and hence faster failovers.
 * Support server pushes and requests in the same connection object, concurrently.
-* Separate requests, handling of server pushes and reconnection operations.
+* Separate requests, handling of server pushes and reconnect operations.
 
 ### Cancellation
 
 Aedis supports both implicit and explicit cancellation of connection
 operations. Explicit cancellation is supported by means of the
 `aedis::connection::cancel` member function. Implicit cancellation,
-like those that may happen when using Asio awaitable operators && and
-|| will be discussed with more detail below.
+like those that may happen when using Asio awaitable operators `&&` and
+`||` will be discussed with more detail below.
 
 ```cpp
 co_await (conn.async_run(...) && conn.async_exec(...))
 ```
 
-* Provide a simple way to send HELLO and perform channel subscription.
+* Provides a simple way to send HELLO and perform channel subscription.
 
 ```cpp
 co_await (conn.async_run(...) || conn.async_exec(...))
@@ -185,7 +206,7 @@ co_await (conn.async_exec(...) || time.async_wait(...))
   should last.
 * The cancellation will be ignored if the request has already
   been written to the socket.
-* It is usually a better idea to have a healthy checker than adding
+* NOTE: It is usually a better idea to have a healthy checker than adding
   per request timeout, see subscriber.cpp for an example.
 
 ```cpp

include/aedis/resp3/detail/impl/parser.ipp

@@ -0,0 +1,155 @@
+/* Copyright (c) 2018-2022 Marcelo Zimbres Silva ([email protected])
+ *
+ * Distributed under the Boost Software License, Version 1.0. (See
+ * accompanying file LICENSE.txt)
+ */
+
+#include <aedis/resp3/detail/parser.hpp>
+
+#include <aedis/error.hpp>
+#include <boost/assert.hpp>
+#include <charconv>
+
+namespace aedis::resp3::detail {
+
+void to_int(int_type& i, std::string_view sv, boost::system::error_code& ec)
+{
+   auto const res = std::from_chars(sv.data(), sv.data() + std::size(sv), i);
+   if (res.ec != std::errc())
+      ec = error::not_a_number;
+}
+
+parser::parser()
+{
+   sizes_[0] = 2; // The sentinel must be more than 1.
+}
+
+auto
+parser::consume(
+   char const* data,
+   std::size_t n,
+   boost::system::error_code& ec) -> std::pair<node_type, std::size_t>
+{
+   node_type ret;
+   if (bulk_expected()) {
+      n = bulk_length_ + 2;
+      ret = {bulk_, 1, depth_, {data, bulk_length_}};
+      bulk_ = type::invalid;
+      --sizes_[depth_];
+
+   } else if (sizes_[depth_] != 0) {
+      auto const t = to_type(*data);
+      switch (t) {
+         case type::streamed_string_part:
+         {
+            to_int(bulk_length_ , std::string_view{data + 1, n - 3}, ec);
+            if (ec)
+               return std::make_pair(node_type{}, 0);
+
+            if (bulk_length_ == 0) {
+               ret = {type::streamed_string_part, 1, depth_, {}};
+               sizes_[depth_] = 0; // We are done.
+               bulk_ = type::invalid;
+            } else {
+               bulk_ = type::streamed_string_part;
+            }
+         } break;
+         case type::blob_error:
+         case type::verbatim_string:
+         case type::blob_string:
+         {
+            if (data[1] == '?') {
+               // NOTE: This can only be triggered with blob_string.
+               // Trick: A streamed string is read as an aggregate
+               // of infinite lenght. When the streaming is done
+               // the server is supposed to send a part with length
+               // 0.
+               sizes_[++depth_] = (std::numeric_limits<std::size_t>::max)();
+               ret = {type::streamed_string, 0, depth_, {}};
+            } else {
+               to_int(bulk_length_ , std::string_view{data + 1, n - 3} , ec);
+               if (ec)
+                  return std::make_pair(node_type{}, 0);
+
+               bulk_ = t;
+            }
+         } break;
+         case type::boolean:
+         {
+            if (n == 3) {
+                ec = error::empty_field;
+                return std::make_pair(node_type{}, 0);
+            }
+
+            if (data[1] != 'f' && data[1] != 't') {
+                ec = error::unexpected_bool_value;
+                return std::make_pair(node_type{}, 0);
+            }
+
+            ret = {t, 1, depth_, {data + 1, n - 3}};
+            --sizes_[depth_];
+         } break;
+         case type::doublean:
+         case type::big_number:
+         case type::number:
+         {
+            if (n == 3) {
+                ec = error::empty_field;
+                return std::make_pair(node_type{}, 0);
+            }
+
+            ret = {t, 1, depth_, {data + 1, n - 3}};
+            --sizes_[depth_];
+         } break;
+         case type::simple_error:
+         case type::simple_string:
+         {
+            ret = {t, 1, depth_, {&data[1], n - 3}};
+            --sizes_[depth_];
+         } break;
+         case type::null:
+         {
+            ret = {type::null, 1, depth_, {}};
+            --sizes_[depth_];
+         } break;
+         case type::push:
+         case type::set:
+         case type::array:
+         case type::attribute:
+         case type::map:
+         {
+            int_type l = -1;
+            to_int(l, std::string_view{data + 1, n - 3}, ec);
+            if (ec)
+               return std::make_pair(node_type{}, 0);
+
+            ret = {t, l, depth_, {}};
+            if (l == 0) {
+               --sizes_[depth_];
+            } else {
+               if (depth_ == max_embedded_depth) {
+                  ec = error::exceeeds_max_nested_depth;
+                  return std::make_pair(node_type{}, 0);
+               }
+
+               ++depth_;
+
+               sizes_[depth_] = l * element_multiplicity(t);
+            }
+         } break;
+         default:
+         {
+            ec = error::invalid_data_type;
+            return std::make_pair(node_type{}, 0);
+         }
+      }
+   }
+   
+   while (sizes_[depth_] == 0) {
+      --depth_;
+      --sizes_[depth_];
+   }
+   
+   return std::make_pair(ret, n);
+}
+} // aedis::resp3::detail