ceph/src/seastar/tests/unit/output_stream_test.cc

   1 /*
   2  * This file is open source software, licensed to you under the terms
   3  * of the Apache License, Version 2.0 (the "License").  See the NOTICE file
   4  * distributed with this work for additional information regarding copyright
   5  * ownership.  You may not use this file except in compliance with the License.
   6  *
   7  * You may obtain a copy of the License at
   8  *
   9  *   http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing,
  12  * software distributed under the License is distributed on an
  13  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  14  * KIND, either express or implied.  See the License for the
  15  * specific language governing permissions and limitations
  16  * under the License.
  17  */
  18 /*
  19  * Copyright (C) 2014 Cloudius Systems, Ltd.
  20  */
  21
  22 #include <seastar/core/app-template.hh>
  23 #include <seastar/core/shared_ptr.hh>
  24 #include <seastar/core/vector-data-sink.hh>
  25 #include <seastar/core/loop.hh>
  26 #include <seastar/util/later.hh>
  27 #include <seastar/core/sstring.hh>
  28 #include <seastar/net/packet.hh>
  29 #include <seastar/testing/test_case.hh>
  30 #include <seastar/testing/thread_test_case.hh>
  31 #include <vector>
  32
  33 using namespace seastar;
  34 using namespace net;
  35
  36 static sstring to_sstring(const packet& p) {
  37     sstring res = uninitialized_string(p.len());
  38     auto i = res.begin();
  39     for (auto& frag : p.fragments()) {
  40         i = std::copy(frag.base, frag.base + frag.size, i);
  41     }
  42     return res;
  43 }
  44
  45 struct stream_maker {
  46     output_stream_options opts;
  47     size_t _size;
  48
  49     stream_maker size(size_t size) && {
  50         _size = size;
  51         return std::move(*this);
  52     }
  53
  54     stream_maker trim(bool do_trim) && {
  55         opts.trim_to_size = do_trim;
  56         return std::move(*this);
  57     }
  58
  59     lw_shared_ptr<output_stream<char>> operator()(data_sink sink) {
  60         return make_lw_shared<output_stream<char>>(std::move(sink), _size, opts);
  61     }
  62 };
  63
  64 template <typename T, typename StreamConstructor>
  65 future<> assert_split(StreamConstructor stream_maker, std::initializer_list<T> write_calls,
  66         std::vector<std::string> expected_split) {
  67     static int i = 0;
  68     BOOST_TEST_MESSAGE("checking split: " << i++);
  69     auto sh_write_calls = make_lw_shared<std::vector<T>>(std::move(write_calls));
  70     auto sh_expected_splits = make_lw_shared<std::vector<std::string>>(std::move(expected_split));
  71     auto v = make_shared<std::vector<packet>>();
  72     auto out = stream_maker(data_sink(std::make_unique<vector_data_sink>(*v)));
  73
  74     return do_for_each(sh_write_calls->begin(), sh_write_calls->end(), [out, sh_write_calls] (auto&& chunk) {
  75         return out->write(chunk);
  76     }).then([out, v, sh_expected_splits] {
  77         return out->close().then([out, v, sh_expected_splits] {
  78             BOOST_REQUIRE_EQUAL(v->size(), sh_expected_splits->size());
  79             int i = 0;
  80             for (auto&& chunk : *sh_expected_splits) {
  81                 BOOST_REQUIRE(to_sstring((*v)[i]) == chunk);
  82                 i++;
  83             }
  84         });
  85     });
  86 }
  87
  88 SEASTAR_TEST_CASE(test_splitting) {
  89     auto ctor = stream_maker().trim(false).size(4);
  90     return now()
  91         .then([=] { return assert_split(ctor, {"1"}, {"1"}); })
  92         .then([=] { return assert_split(ctor, {"12", "3"}, {"123"}); })
  93         .then([=] { return assert_split(ctor, {"12", "34"}, {"1234"}); })
  94         .then([=] { return assert_split(ctor, {"12", "345"}, {"1234", "5"}); })
  95         .then([=] { return assert_split(ctor, {"1234"}, {"1234"}); })
  96         .then([=] { return assert_split(ctor, {"12345"}, {"12345"}); })
  97         .then([=] { return assert_split(ctor, {"1234567890"}, {"1234567890"}); })
  98         .then([=] { return assert_split(ctor, {"1", "23456"}, {"1234", "56"}); })
  99         .then([=] { return assert_split(ctor, {"123", "4567"}, {"1234", "567"}); })
 100         .then([=] { return assert_split(ctor, {"123", "45678"}, {"1234", "5678"}); })
 101         .then([=] { return assert_split(ctor, {"123", "4567890"}, {"1234", "567890"}); })
 102         .then([=] { return assert_split(ctor, {"1234", "567"}, {"1234", "567"}); })
 103
 104         .then([] { return assert_split(stream_maker().trim(false).size(3), {"1", "234567", "89"}, {"123", "4567", "89"}); })
 105         .then([] { return assert_split(stream_maker().trim(false).size(3), {"1", "2345", "67"}, {"123", "456", "7"}); })
 106         ;
 107 }
 108
 109 SEASTAR_TEST_CASE(test_splitting_with_trimming) {
 110     auto ctor = stream_maker().trim(true).size(4);
 111     return now()
 112         .then([=] { return assert_split(ctor, {"1"}, {"1"}); })
 113         .then([=] { return assert_split(ctor, {"12", "3"}, {"123"}); })
 114         .then([=] { return assert_split(ctor, {"12", "3456789"}, {"1234", "5678", "9"}); })
 115         .then([=] { return assert_split(ctor, {"12", "3456789", "12"}, {"1234", "5678", "912"}); })
 116         .then([=] { return assert_split(ctor, {"123456789"}, {"1234", "5678", "9"}); })
 117         .then([=] { return assert_split(ctor, {"12345678"}, {"1234", "5678"}); })
 118         .then([=] { return assert_split(ctor, {"12345678", "9"}, {"1234", "5678", "9"}); })
 119         .then([=] { return assert_split(ctor, {"1234", "567890"}, {"1234", "5678", "90"}); })
 120         ;
 121 }
 122
 123 SEASTAR_TEST_CASE(test_flush_on_empty_buffer_does_not_push_empty_packet_down_stream) {
 124     auto v = make_shared<std::vector<packet>>();
 125     auto out = make_shared<output_stream<char>>(
 126         data_sink(std::make_unique<vector_data_sink>(*v)), 8);
 127
 128     return out->flush().then([v, out] {
 129         BOOST_REQUIRE(v->empty());
 130         return out->close();
 131     }).finally([out]{});
 132 }
 133
 134 SEASTAR_THREAD_TEST_CASE(test_simple_write) {
 135     auto vec = std::vector<net::packet>{};
 136     auto out = output_stream<char>(data_sink(std::make_unique<vector_data_sink>(vec)), 8);
 137
 138     auto value1 = sstring("te");
 139     out.write(value1).get();
 140
 141
 142     auto value2 = sstring("st");
 143     out.write(value2).get();
 144
 145     auto value3 = sstring("abcdefgh1234");
 146     out.write(value3).get();
 147
 148     out.close().get();
 149
 150     auto value = value1 + value2 + value3;
 151     auto packets = net::packet{};
 152     for (auto& p : vec) {
 153         packets.append(std::move(p));
 154     }
 155     packets.linearize();
 156     auto buf = packets.release();
 157     BOOST_REQUIRE_EQUAL(buf.size(), 1);
 158     BOOST_REQUIRE_EQUAL(sstring(buf.front().get(), buf.front().size()), value);
 159 }