feat: Zero length reads.

- Simplified the noop loop to correctly handle zero length reads.

Author: kcexn

CMakeLists.txt

@@ -24,7 +24,7 @@ CPMAddPackage(
 CPMAddPackage(
   NAME asyncberkeley
   GITHUB_REPOSITORY kcexn/async-berkeley
-  GIT_TAG v0.1.2
+  GIT_TAG v0.1.3
   OPTIONS
   "IO_BUILD_TESTING OFF"
   "IO_BUILD_EXAMPLES OFF"

include/net/service/async_tcp_service.hpp

@@ -24,6 +24,8 @@
 namespace cloudbus::service {
 /**
  * @brief A ServiceLike Async TCP Service.
+ * @tparam StreamHandler The StreamHandler type that derives from
+ * async_tcp_service.
  * @note The default constructor of async_tcp_service is protected
  * so async_tcp_service can't be constructed without a stream handler
  * (which would be UB).
@@ -34,8 +36,6 @@ namespace cloudbus::service {
  * reader to restart the read loop. It also optionally specifies an initialize
  * member that can be used to configure the service socket. See `noop_service`
  * below for an example of how to specialize async_tcp_service.
- * @tparam StreamHandler The StreamHandler type that derives from
- * async_tcp_service.
  * @code
  * struct noop_service : public async_tcp_service<noop_service>
  * {
@@ -52,10 +52,10 @@ namespace cloudbus::service {
  *   }
  *
  *   auto operator()(async_context &ctx, const socket_dialog &socket,
- *                   std::shared_ptr<readbuf> rmsg,
+ *                   std::shared_ptr<read_context> rctx,
  *                   std::span<const std::byte> buf) -> void
  *   {
- *     reader(ctx, socket, std::move(rmsg));
+ *     reader(ctx, socket, std::move(rctx));
  *   }
  * };
  * @endcode
@@ -115,7 +115,7 @@ template <typename TCPStreamHandler> class async_tcp_service {
    * @param rmsg A shared pointer to a mutable read buffer.
    */
   auto reader(async_context &ctx, const socket_dialog &socket,
-              std::shared_ptr<read_context> rmsg) -> void;
+              std::shared_ptr<read_context> rctx) -> void;
 
 protected:
   /** @brief Default constructor. */
@@ -139,8 +139,8 @@ template <typename TCPStreamHandler> class async_tcp_service {
    * @param buf The data read from the socket in the last recvmsg.
    */
   auto emit(async_context &ctx, const socket_dialog &socket,
-            std::shared_ptr<read_context> rmsg,
-            std::span<const std::byte> buf) -> void;
+            std::shared_ptr<read_context> rctx = {},
+            std::span<const std::byte> buf = {}) -> void;
 
   /**
    * @brief Initializes the server socket with options. Delegates to

include/net/service/impl/async_tcp_service_impl.hpp

@@ -87,37 +87,35 @@ auto async_tcp_service<TCPStreamHandler>::acceptor(
 template <typename TCPStreamHandler>
 auto async_tcp_service<TCPStreamHandler>::reader(
     async_context &ctx, const socket_dialog &socket,
-    std::shared_ptr<read_context> rmsg) -> void
+    std::shared_ptr<read_context> rctx) -> void
 {
   using namespace stdexec;
   using namespace io::socket;
-  if (ctx.scope.get_stop_token().stop_requested())
+  if (!rctx || ctx.scope.get_stop_token().stop_requested())
     return;
 
   sender auto recvmsg =
-      io::recvmsg(socket, rmsg->msg, 0) |
-      then([&, socket, rmsg](auto &&len) mutable {
+      io::recvmsg(socket, rctx->msg, 0) |
+      then([&, socket, rctx](auto &&len) mutable {
         using size_type = std::size_t;
         if (!len)
-          return emit(ctx, socket, {}, std::span<std::byte>{});
+          return emit(ctx, socket);
 
-        auto buf = std::span{rmsg->buffer.data(), static_cast<size_type>(len)};
-        emit(ctx, socket, std::move(rmsg), buf);
+        auto buf = std::span{rctx->buffer.data(), static_cast<size_type>(len)};
+        emit(ctx, socket, std::move(rctx), buf);
       }) |
-      upon_error([&, socket](auto &&error) {
-        emit(ctx, socket, {}, std::span<std::byte>{});
-      });
+      upon_error([&, socket](auto &&error) { emit(ctx, socket); });
 
   ctx.scope.spawn(std::move(recvmsg));
 }
 
 template <typename TCPStreamHandler>
 auto async_tcp_service<TCPStreamHandler>::emit(
     async_context &ctx, const socket_dialog &socket,
-    std::shared_ptr<read_context> rmsg, std::span<const std::byte> buf) -> void
+    std::shared_ptr<read_context> rctx, std::span<const std::byte> buf) -> void
 {
   auto &handle = static_cast<TCPStreamHandler &>(*this);
-  handle(ctx, socket, std::move(rmsg), buf);
+  handle(ctx, socket, std::move(rctx), buf);
 }
 
 template <typename TCPStreamHandler>

tests/test_async_tcp_service.cpp

@@ -45,19 +45,19 @@ struct echo_block_service : public async_tcp_service<echo_block_service> {
   }
 
   auto echo(async_context &ctx, const socket_dialog &socket,
-            const std::shared_ptr<read_context> &rmsg,
+            const std::shared_ptr<read_context> &rctx,
             socket_message msg) -> void
   {
     using namespace io::socket;
     using namespace stdexec;
 
     sender auto sendmsg =
         io::sendmsg(socket, msg, 0) |
-        then([&, socket, msg, rmsg](auto &&len) mutable {
-          if (auto buffers = std::move(msg.buffers); buffers += len)
-            return echo(ctx, socket, rmsg, {.buffers = buffers});
+        then([&, socket, rctx, bufs = msg.buffers](auto &&len) mutable {
+          if (bufs += len)
+            return echo(ctx, socket, std::move(rctx), {.buffers = bufs});
 
-          reader(ctx, socket, std::move(rmsg));
+          reader(ctx, socket, std::move(rctx));
         }) |
         upon_error([](auto &&error) {});
 
@@ -68,8 +68,7 @@ struct echo_block_service : public async_tcp_service<echo_block_service> {
                   std::shared_ptr<read_context> rmsg,
                   std::span<const std::byte> buf) -> void
   {
-    if (buf.data())
-      echo(ctx, socket, rmsg, {.buffers = buf});
+    echo(ctx, socket, rmsg, {.buffers = buf});
   }
 };