From aab95097754f7e0bb44a8e1b248e91541a8d46ff Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Micha=C5=82=20Chojnowski?= <michal.chojnowski@scylladb.com>
Date: Mon, 29 Mar 2021 12:37:59 +0200
Subject: [PATCH] types: collection: add versions of pack for fragmented
 buffers

We will need them to port the representation of collection types
in cql3/ from bytes to managed_bytes.
The version which takes an iterator of `bytes` as an argument will
be removed after that transition is complete.
---
 types/collection.hh | 56 +++++++++++++++++++++++++++++++++++++++++----
 1 file changed, 52 insertions(+), 4 deletions(-)

diff --git a/types/collection.hh b/types/collection.hh
index 9500d549b2..ea86f96319 100644
--- a/types/collection.hh
+++ b/types/collection.hh
@@ -52,8 +52,18 @@ public:
     lw_shared_ptr<cql3::column_specification> make_collection_receiver(const cql3::column_specification& collection, bool is_key) const;
     virtual bool is_compatible_with_frozen(const collection_type_impl& previous) const = 0;
     virtual bool is_value_compatible_with_frozen(const collection_type_impl& previous) const = 0;
-    template <typename BytesViewIterator>
-    static bytes pack(BytesViewIterator start, BytesViewIterator finish, int elements, cql_serialization_format sf);
+
+    template <typename Iterator>
+    requires requires (Iterator it) { {*it} -> std::convertible_to<bytes_view>; }
+    static bytes pack(Iterator start, Iterator finish, int elements, cql_serialization_format sf);
+
+    template <typename Iterator>
+    requires requires (Iterator it) { {*it} -> std::convertible_to<bytes_view>; }
+    static managed_bytes pack_fragmented(Iterator start, Iterator finish, int elements, cql_serialization_format sf);
+
+    template <typename Iterator>
+    requires requires (Iterator it) { {*it} -> std::convertible_to<managed_bytes_view>; }
+    static managed_bytes pack_fragmented(Iterator start, Iterator finish, int elements, cql_serialization_format sf);
 
 private:
     // Explicitly instantiated in types.cc
@@ -104,9 +114,10 @@ public:
     bytes serialize_map(const map_type_impl& map_type, const data_value& value) const;
 };
 
-template <typename BytesViewIterator>
+template <typename Iterator>
+requires requires (Iterator it) { {*it} -> std::convertible_to<bytes_view>; }
 bytes
-collection_type_impl::pack(BytesViewIterator start, BytesViewIterator finish, int elements, cql_serialization_format sf) {
+collection_type_impl::pack(Iterator start, Iterator finish, int elements, cql_serialization_format sf) {
     size_t len = collection_size_len(sf);
     size_t psz = collection_value_len(sf);
     for (auto j = start; j != finish; j++) {
@@ -120,3 +131,40 @@ collection_type_impl::pack(BytesViewIterator start, BytesViewIterator finish, in
     }
     return out;
 }
+
+// TODO: remove after all collections types in cql3/ are converted to managed_bytes
+template <typename Iterator>
+requires requires (Iterator it) { {*it} -> std::convertible_to<bytes_view>; }
+managed_bytes
+collection_type_impl::pack_fragmented(Iterator start, Iterator finish, int elements, cql_serialization_format sf) {
+    size_t len = collection_size_len(sf);
+    size_t psz = collection_value_len(sf);
+    for (auto j = start; j != finish; j++) {
+        len += j->size() + psz;
+    }
+    managed_bytes out(managed_bytes::initialized_later(), len);
+    managed_bytes_mutable_view v(out);
+    write_collection_size(v, elements, sf);
+    while (start != finish) {
+        write_collection_value(v, sf, *start++);
+    }
+    return out;
+}
+
+template <typename Iterator>
+requires requires (Iterator it) { {*it} -> std::convertible_to<managed_bytes_view>; }
+managed_bytes
+collection_type_impl::pack_fragmented(Iterator start, Iterator finish, int elements, cql_serialization_format sf) {
+    size_t len = collection_size_len(sf);
+    size_t psz = collection_value_len(sf);
+    for (auto j = start; j != finish; j++) {
+        len += j->size() + psz;
+    }
+    managed_bytes out(managed_bytes::initialized_later(), len);
+    managed_bytes_mutable_view v(out);
+    write_collection_size(v, elements, sf);
+    while (start != finish) {
+        write_collection_value(v, sf, *start++);
+    }
+    return out;
+}