Remove Reader::read

include/libp2p/basic/reader.hpp

@@ -20,20 +20,6 @@ namespace libp2p::basic {
 
     virtual ~Reader() = default;
 
-    /**
-     * @brief Reads exactly {@code} min(out.size(), bytes) {@nocode} bytes to
-     * the buffer.
-     * @param out output argument. Read data will be written to this buffer.
-     * @param bytes number of bytes to read
-     * @param cb callback with result of operation
-     *
-     * @note caller should maintain validity of an output buffer until callback
-     * is executed. It is usually done with either wrapping buffer as shared
-     * pointer, or having buffer as part of some class/struct, and using
-     * enable_shared_from_this()
-     */
-    virtual void read(BytesOut out, size_t bytes, ReadCallbackFunc cb) = 0;
-
     /**
      * @brief Reads up to {@code} min(out.size(), bytes) {@nocode} bytes to the
      * buffer.

include/libp2p/connection/loopback_stream.hpp

@@ -40,8 +40,6 @@ namespace libp2p::connection {
         const override;
 
    protected:
-    void read(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
-
     void readSome(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
 
     void writeSome(BytesIn in, size_t bytes, WriteCallbackFunc cb) override;

include/libp2p/layer/websocket/ssl_connection.hpp

@@ -32,7 +32,6 @@ namespace libp2p::connection {
     outcome::result<void> close() override;
     bool isClosed() const override;
 
-    void read(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
     void readSome(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
     void deferReadCallback(outcome::result<size_t> res,
                            ReadCallbackFunc cb) override;

include/libp2p/layer/websocket/ws_connection.hpp

@@ -68,8 +68,6 @@ namespace libp2p::connection {
 
     bool isClosed() const override;
 
-    void read(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
-
     void readSome(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
 
     void deferReadCallback(outcome::result<size_t> res,

include/libp2p/muxer/mplex/mplex_stream.hpp

@@ -50,8 +50,6 @@ namespace libp2p::connection {
 
     ~MplexStream() override = default;
 
-    void read(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
-
     void readSome(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
 
     void deferReadCallback(outcome::result<size_t> res,

include/libp2p/muxer/mplex/mplexed_connection.hpp

@@ -64,7 +64,6 @@ namespace libp2p::connection {
 
     /// usage of these four methods is highly not recommended or even forbidden:
     /// use stream over this connection instead
-    void read(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
     void readSome(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
     void writeSome(BytesIn in, size_t bytes, WriteCallbackFunc cb) override;
 

include/libp2p/muxer/yamux/yamux_stream.hpp

@@ -54,8 +54,6 @@ namespace libp2p::connection {
                 size_t maximum_window_size,
                 size_t write_queue_limit);
 
-    void read(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
-
     void readSome(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
 
     void deferReadCallback(outcome::result<size_t> res,

include/libp2p/muxer/yamux/yamuxed_connection.hpp

@@ -111,7 +111,6 @@ namespace libp2p::connection {
 
     /// usage of these four methods is highly not recommended or even forbidden:
     /// use stream over this connection instead
-    void read(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
     void readSome(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
     void writeSome(BytesIn in, size_t bytes, WriteCallbackFunc cb) override;
 

include/libp2p/security/noise/noise_connection.hpp

@@ -38,8 +38,6 @@ namespace libp2p::connection {
 
     outcome::result<void> close() override;
 
-    void read(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
-
     void readSome(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
 
     void deferReadCallback(outcome::result<size_t> res,

include/libp2p/security/plaintext/plaintext_connection.hpp

@@ -35,8 +35,6 @@ namespace libp2p::connection {
 
     outcome::result<multi::Multiaddress> remoteMultiaddr() override;
 
-    void read(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
-
     void readSome(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
 
     void deferReadCallback(outcome::result<size_t> res,

include/libp2p/security/secio/secio_connection.hpp

@@ -97,8 +97,6 @@ namespace libp2p::connection {
 
     outcome::result<multi::Multiaddress> remoteMultiaddr() override;
 
-    void read(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
-
     void readSome(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
 
     void deferReadCallback(outcome::result<size_t> res,

include/libp2p/transport/quic/connection.hpp

@@ -39,7 +39,6 @@ namespace libp2p::transport {
     void operator=(QuicConnection &&) = delete;
 
     // Reader
-    void read(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
     void readSome(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
     void deferReadCallback(outcome::result<size_t> res,
                            ReadCallbackFunc cb) override;

include/libp2p/transport/quic/stream.hpp

@@ -32,7 +32,6 @@ namespace libp2p::connection {
     void operator=(QuicStream &&) = delete;
 
     // Reader
-    void read(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
     void readSome(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
     void deferReadCallback(outcome::result<size_t> res,
                            ReadCallbackFunc cb) override;

include/libp2p/transport/tcp/tcp_connection.hpp

@@ -61,8 +61,6 @@ namespace libp2p::transport {
                  ConnectCallbackFunc cb,
                  std::chrono::milliseconds timeout);
 
-    void read(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
-
     void readSome(BytesOut out, size_t bytes, ReadCallbackFunc cb) override;
 
     void deferReadCallback(outcome::result<size_t> res,

src/basic/message_read_writer_bigendian.cpp

@@ -11,6 +11,7 @@
 #include <arpa/inet.h>
 #include <boost/assert.hpp>
 #include <libp2p/basic/message_read_writer_error.hpp>
+#include <libp2p/basic/read_return_size.hpp>
 #include <libp2p/basic/write_return_size.hpp>
 #include <libp2p/common/byteutil.hpp>
 
@@ -24,9 +25,9 @@ namespace libp2p::basic {
   void MessageReadWriterBigEndian::read(ReadCallbackFunc cb) {
     auto buffer = std::make_shared<std::vector<uint8_t>>();
     buffer->resize(kLenMarkerSize);
-    conn_->read(
+    readReturnSize(
+        conn_,
         *buffer,
-        kLenMarkerSize,
         [self{shared_from_this()}, buffer, cb{std::move(cb)}](auto &&result) {
           if (not result) {
             return cb(result.error());
@@ -35,8 +36,8 @@ namespace libp2p::basic {
               common::convert<uint32_t>(buffer->data()));
           buffer->resize(msg_len);
           std::fill(buffer->begin(), buffer->end(), 0u);
-          self->conn_->read(
-              *buffer, msg_len, [self, buffer, cb](auto &&result) {
+          readReturnSize(
+              self->conn_, *buffer, [self, buffer, cb](auto &&result) {
                 if (not result) {
                   return cb(result.error());
                 }

src/basic/message_read_writer_uvarint.cpp

@@ -11,6 +11,7 @@
 #include <boost/assert.hpp>
 #include <boost/optional.hpp>
 #include <libp2p/basic/message_read_writer_error.hpp>
+#include <libp2p/basic/read_return_size.hpp>
 #include <libp2p/basic/varint_reader.hpp>
 #include <libp2p/basic/write_return_size.hpp>
 #include <libp2p/multi/uvarint.hpp>
@@ -34,9 +35,9 @@ namespace libp2p::basic {
           auto msg_len = varint_res.value().toUInt64();
           if (0 != msg_len) {
             auto buffer = std::make_shared<std::vector<uint8_t>>(msg_len, 0);
-            self->conn_->read(
+            readReturnSize(
+                self->conn_,
                 *buffer,
-                msg_len,
                 [self, buffer, cb = std::move(cb)](auto &&res) mutable {
                   if (!res) {
                     return cb(res.error());

src/basic/varint_reader.cpp

@@ -5,6 +5,7 @@
  */
 
 #include <libp2p/basic/varint_reader.hpp>
+#include <libp2p/basic/read_return_size.hpp>
 
 #include <vector>
 
@@ -44,10 +45,10 @@ namespace libp2p::basic {
     }
 
     // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-pointer-arithmetic)
-    conn->read(
+    readReturnSize(
+        conn,
         std::span(varint_buf->data() + current_length, 1),
-        1,
-        [c = std::move(conn), cb = std::move(cb), current_length, varint_buf](
+        [c = conn, cb = std::move(cb), current_length, varint_buf](
             auto &&res) mutable {
           if (not res.has_value()) {
             return cb(res.error());

src/connection/loopback_stream.cpp

@@ -77,13 +77,6 @@ namespace libp2p::connection {
     return outcome::success(own_peer_info_.addresses.front());
   }
 
-  void LoopbackStream::read(BytesOut out,
-                            size_t bytes,
-                            libp2p::basic::Reader::ReadCallbackFunc cb) {
-    ambigousSize(out, bytes);
-    readReturnSize(shared_from_this(), out, std::move(cb));
-  }
-
   void LoopbackStream::writeSome(BytesIn in,
                                  size_t bytes,
                                  libp2p::basic::Writer::WriteCallbackFunc cb) {

src/layer/websocket/ssl_connection.cpp

@@ -43,13 +43,6 @@ namespace libp2p::connection {
     return connection_->close();
   }
 
-  void SslConnection::read(BytesOut out,
-                           size_t bytes,
-                           libp2p::basic::Reader::ReadCallbackFunc cb) {
-    ambigousSize(out, bytes);
-    readReturnSize(shared_from_this(), out, std::move(cb));
-  }
-
   void SslConnection::readSome(BytesOut out,
                                size_t bytes,
                                libp2p::basic::Reader::ReadCallbackFunc cb) {

src/layer/websocket/ws_connection.cpp

@@ -85,14 +85,6 @@ namespace libp2p::connection {
     return connection_->close();
   }
 
-  void WsConnection::read(BytesOut out,
-                          size_t bytes,
-                          libp2p::basic::Reader::ReadCallbackFunc cb) {
-    ambigousSize(out, bytes);
-    SL_TRACE(log_, "read {} bytes", bytes);
-    readReturnSize(shared_from_this(), out, std::move(cb));
-  }
-
   void WsConnection::readSome(BytesOut out,
                               size_t bytes,
                               libp2p::basic::Reader::ReadCallbackFunc cb) {

src/muxer/mplex/mplex_frame.cpp

@@ -6,6 +6,7 @@
 
 #include <libp2p/muxer/mplex/mplex_frame.hpp>
 
+#include <libp2p/basic/read_return_size.hpp>
 #include <libp2p/basic/varint_reader.hpp>
 #include <libp2p/multi/uvarint.hpp>
 #include <libp2p/muxer/mplex/mplexed_connection.hpp>
@@ -100,16 +101,16 @@ namespace libp2p::connection {
                 // read data
                 std::shared_ptr<Bytes> data =
                     std::make_shared<Bytes>(length, 0);
-                reader->read(*data,
-                             length,
-                             [id_flag, data, cb{std::move(cb)}](
-                                 auto &&read_res) mutable {
-                               if (!read_res) {
-                                 return cb(read_res.error());
-                               }
+                readReturnSize(reader,
+                               *data,
+                               [id_flag, data, cb{std::move(cb)}](
+                                   auto &&read_res) mutable {
+                                 if (!read_res) {
+                                   return cb(read_res.error());
+                                 }
 
-                               cb(createFrame(id_flag, std::move(*data)));
-                             });
+                                 cb(createFrame(id_flag, std::move(*data)));
+                               });
               });
         });
   }

src/muxer/mplex/mplex_stream.cpp

@@ -34,11 +34,6 @@ namespace libp2p::connection {
                            StreamId stream_id)
       : connection_{std::move(connection)}, stream_id_{stream_id} {}
 
-  void MplexStream::read(BytesOut out, size_t bytes, ReadCallbackFunc cb) {
-    ambigousSize(out, bytes);
-    readReturnSize(shared_from_this(), out, std::move(cb));
-  }
-
   void MplexStream::readDone(outcome::result<size_t> res) {
     auto cb{std::move(reading_->cb)};
     reading_.reset();

src/muxer/mplex/mplexed_connection.cpp

@@ -7,6 +7,7 @@
 #include <libp2p/muxer/mplex/mplexed_connection.hpp>
 
 #include <boost/assert.hpp>
+#include <libp2p/basic/read_return_size.hpp>
 #include <libp2p/basic/write_return_size.hpp>
 #include <libp2p/muxer/mplex/mplex_frame.hpp>
 
@@ -124,12 +125,6 @@ namespace libp2p::connection {
     return !is_active_ || connection_->isClosed();
   }
 
-  void MplexedConnection::read(BytesOut out,
-                               size_t bytes,
-                               ReadCallbackFunc cb) {
-    connection_->read(out, bytes, std::move(cb));
-  }
-
   void MplexedConnection::readSome(BytesOut out,
                                    size_t bytes,
                                    ReadCallbackFunc cb) {

src/muxer/yamux/yamux_stream.cpp

@@ -67,11 +67,6 @@ namespace libp2p::connection {
     assert(write_queue_limit >= maximum_window_size_);
   }
 
-  void YamuxStream::read(BytesOut out, size_t bytes, ReadCallbackFunc cb) {
-    ambigousSize(out, bytes);
-    readReturnSize(shared_from_this(), out, std::move(cb));
-  }
-
   void YamuxStream::readSome(BytesOut out, size_t bytes, ReadCallbackFunc cb) {
     ambigousSize(out, bytes);
     doRead(out, std::move(cb));

src/muxer/yamux/yamuxed_connection.cpp

@@ -163,13 +163,6 @@ namespace libp2p::connection {
     return !started_ || connection_->isClosed();
   }
 
-  void YamuxedConnection::read(BytesOut out,
-                               size_t bytes,
-                               ReadCallbackFunc cb) {
-    log()->error("YamuxedConnection::read : invalid direct call");
-    deferReadCallback(Error::CONNECTION_DIRECT_IO_FORBIDDEN, std::move(cb));
-  }
-
   void YamuxedConnection::readSome(BytesOut out,
                                    size_t bytes,
                                    ReadCallbackFunc cb) {

src/protocol/gossip/impl/stream.cpp

@@ -8,6 +8,7 @@
 
 #include <cassert>
 
+#include <libp2p/basic/read_return_size.hpp>
 #include <libp2p/basic/varint_reader.hpp>
 #include <libp2p/basic/write_return_size.hpp>
 
@@ -83,15 +84,15 @@ namespace libp2p::protocol::gossip {
 
     read_buffer_->resize(msg_len);
 
-    stream_->read(std::span(read_buffer_->data(), msg_len),
-                  msg_len,
-                  [self_wptr = weak_from_this(), this, buffer = read_buffer_](
-                      auto &&res) {
-                    if (self_wptr.expired()) {
-                      return;
-                    }
-                    onMessageRead(std::forward<decltype(res)>(res));
-                  });
+    readReturnSize(stream_,
+                   std::span(read_buffer_->data(), msg_len),
+                   [self_wptr = weak_from_this(), this, buffer = read_buffer_](
+                       auto &&res) {
+                     if (self_wptr.expired()) {
+                       return;
+                     }
+                     onMessageRead(std::forward<decltype(res)>(res));
+                   });
   }
 
   void Stream::onMessageRead(outcome::result<size_t> res) {

src/protocol/ping/ping_client_session.cpp

@@ -7,6 +7,7 @@
 #include <libp2p/protocol/ping/ping_client_session.hpp>
 
 #include <boost/assert.hpp>
+#include <libp2p/basic/read_return_size.hpp>
 #include <libp2p/basic/write_return_size.hpp>
 #include <libp2p/protocol/ping/common.hpp>
 
@@ -84,11 +85,11 @@ namespace libp2p::protocol {
         },
         config_.timeout);
 
-    stream_->read(read_buffer_,
-                  config_.message_size,
-                  [self{shared_from_this()}](outcome::result<size_t> r) {
-                    self->readCompleted(r);
-                  });
+    readReturnSize(stream_,
+                   read_buffer_,
+                   [self{shared_from_this()}](outcome::result<size_t> r) {
+                     self->readCompleted(r);
+                   });
   }
 
   void PingClientSession::readCompleted(outcome::result<size_t> r) {