serializer: Introduce as_input_stream(bytes_view)

canonical_mutation.cc

@@ -56,13 +56,13 @@ canonical_mutation::canonical_mutation(const mutation& m)
 }
 
 utils::UUID canonical_mutation::column_family_id() const {
-    seastar::simple_input_stream in(reinterpret_cast<const char*>(_data.begin()), _data.size());
+    auto in = ser::as_input_stream(_data);
     auto mv = ser::deserialize(in, boost::type<ser::canonical_mutation_view>());
     return mv.table_id();
 }
 
 mutation canonical_mutation::to_mutation(schema_ptr s) const {
-    seastar::simple_input_stream in(reinterpret_cast<const char*>(_data.begin()), _data.size());
+    auto in = ser::as_input_stream(_data);
     auto mv = ser::deserialize(in, boost::type<ser::canonical_mutation_view>());
 
     auto cf_id = mv.table_id();

db/batchlog_manager.cc

@@ -180,7 +180,7 @@ future<> db::batchlog_manager::replay_all_failed_batches() {
         logger.debug("Replaying batch {}", id);
 
         auto fms = make_lw_shared<std::deque<canonical_mutation>>();
-        seastar::simple_input_stream in(reinterpret_cast<const char*>(data.begin()), data.size());
+        auto in = ser::as_input_stream(data);
         while (in.size()) {
             fms->emplace_back(ser::deserialize(in, boost::type<canonical_mutation>()));
         }

frozen_mutation.cc

@@ -49,14 +49,14 @@ using namespace db;
 
 utils::UUID
 frozen_mutation::column_family_id() const {
-    seastar::simple_input_stream in(reinterpret_cast<const char*>(_bytes.begin()), _bytes.size());
+    auto in = ser::as_input_stream(_bytes);
     auto mv = ser::deserialize(in, boost::type<ser::mutation_view>());
     return mv.table_id();
 }
 
 utils::UUID
 frozen_mutation::schema_version() const {
-    seastar::simple_input_stream in(reinterpret_cast<const char*>(_bytes.begin()), _bytes.size());
+    auto in = ser::as_input_stream(_bytes);
     auto mv = ser::deserialize(in, boost::type<ser::mutation_view>());
     return mv.schema_version();
 }
@@ -72,7 +72,7 @@ frozen_mutation::decorated_key(const schema& s) const {
 }
 
 partition_key frozen_mutation::deserialize_key() const {
-    seastar::simple_input_stream in(reinterpret_cast<const char*>(_bytes.begin()), _bytes.size());
+    auto in = ser::as_input_stream(_bytes);
     auto mv = ser::deserialize(in, boost::type<ser::mutation_view>());
     return mv.key();
 }
@@ -113,7 +113,7 @@ frozen_mutation freeze(const mutation& m) {
 }
 
 mutation_partition_view frozen_mutation::partition() const {
-    seastar::simple_input_stream in(reinterpret_cast<const char*>(_bytes.begin()), _bytes.size());
+    auto in = ser::as_input_stream(_bytes);
     auto mv = ser::deserialize(in, boost::type<ser::mutation_view>());
     return mutation_partition_view::from_view(mv.partition());
 }

frozen_schema.cc

@@ -43,7 +43,7 @@ frozen_schema::frozen_schema(const schema_ptr& s)
 { }
 
 schema_ptr frozen_schema::unfreeze() const {
-    seastar::simple_input_stream in(reinterpret_cast<const char*>(_data.begin()), _data.size());
+    auto in = ser::as_input_stream(_data);
     auto sv = ser::deserialize(in, boost::type<ser::schema_view>());
     return db::schema_tables::create_table_from_mutations(sv.mutations(), sv.version());
 }

serializer_impl.hh

@@ -323,6 +323,11 @@ T deserialize_from_buffer(const Buffer& buf, boost::type<T> type, size_t head_sp
     return deserialize(in, std::move(type));
 }
 
+inline
+seastar::simple_input_stream as_input_stream(bytes_view b) {
+    return seastar::simple_input_stream(reinterpret_cast<const char*>(b.begin()), b.size());
+}
+
 template<typename Output, typename ...T>
 void serialize(Output& out, const boost::variant<T...>& v) {}
 

tests/idl_test.cc

@@ -85,12 +85,12 @@ BOOST_AUTO_TEST_CASE(test_simple_compound)
     BOOST_REQUIRE_EQUAL(buf1.linearize(), buf2.linearize());
 
     auto bv1 = buf1.linearize();
-    seastar::simple_input_stream in1(reinterpret_cast<const char*>(bv1.data()), bv1.size());
+    auto in1 = ser::as_input_stream(bv1);
     auto deser_sc = ser::deserialize(in1, boost::type<simple_compound>());
     BOOST_REQUIRE_EQUAL(sc, deser_sc);
 
     auto bv2 = buf2.linearize();
-    seastar::simple_input_stream in2(reinterpret_cast<const char*>(bv2.data()), bv2.size());
+    auto in2 = ser::as_input_stream(bv2);
     auto sc_view = ser::deserialize(in2, boost::type<ser::writable_simple_compound_view>());
     BOOST_REQUIRE_EQUAL(sc.foo, sc_view.foo());
     BOOST_REQUIRE_EQUAL(sc.bar, sc_view.bar());
@@ -132,13 +132,13 @@ BOOST_AUTO_TEST_CASE(test_vector)
     BOOST_REQUIRE_EQUAL(buf1.linearize(), buf2.linearize());
 
     auto bv1 = buf1.linearize();
-    seastar::simple_input_stream in1(reinterpret_cast<const char*>(bv1.data()), bv1.size());
+    auto in1 = ser::as_input_stream(bv1);
     auto deser_voc = ser::deserialize(in1, boost::type<vectors_of_compounds>());
     BOOST_REQUIRE_EQUAL(voc.first, deser_voc.first);
     BOOST_REQUIRE_EQUAL(voc.second, deser_voc.second);
 
     auto bv2 = buf2.linearize();
-    seastar::simple_input_stream in2(reinterpret_cast<const char*>(bv2.data()), bv2.size());
+    auto in2 = ser::as_input_stream(bv2);
     auto voc_view = ser::deserialize(in2, boost::type<ser::writable_vectors_of_compounds_view>());
     auto&& first_view = voc_view.first();
     BOOST_REQUIRE_EQUAL(vec1.size(), first_view.size());
@@ -180,7 +180,7 @@ BOOST_AUTO_TEST_CASE(test_variant)
     BOOST_REQUIRE_EQUAL(buf.size(), 120);
 
     auto bv = buf.linearize();
-    seastar::simple_input_stream in(reinterpret_cast<const char*>(bv.data()), bv.size());
+    auto in = ser::as_input_stream(bv);
     auto wv_view = ser::deserialize(in, boost::type<ser::writable_variants_view>());
     BOOST_REQUIRE_EQUAL(wv_view.id(), 17);
 

tests/keys_test.cc

@@ -141,7 +141,7 @@ template<typename T>
 inline
 T reserialize(const T& v) {
     auto buf = ser::serialize_to_buffer<bytes>(v);
-    seastar::simple_input_stream in(reinterpret_cast<const char*>(buf.begin()), buf.size());
+    auto in = ser::as_input_stream(buf);
     return ser::deserialize(in, boost::type<T>());
 }