boost/redis
2
0
Fork 0
mirror of https://github.com/boostorg/redis.git synced 2026-01-19 04:42:09 +00:00
Code Issues Projects Releases Wiki Activity

Removes async_append_some.

Browse Source
This commit is contained in:
Marcelo Zimbres
2025-07-04 22:19:19 +02:00
parent 76129bbcb8
commit 97d71d1d6b
19 changed files with 405 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
test/common.cpp
View File

@@ -50,7 +50,6 @@ boost::redis::config make_test_config()
{
boost::redis::config cfg;
cfg.addr.host = get_server_hostname();
cfg.max_read_size = 1000000;
return cfg;
}
@@ -69,3 +68,10 @@ void run_coroutine_test(net::awaitable<void> op, std::chrono::steady_clock::dura
throw std::runtime_error("Coroutine test did not finish");
}
#endif // BOOST_ASIO_HAS_CO_AWAIT
void append_read_data(boost::redis::detail::reader_fsm& fsm, std::string_view data)
{
auto const buffer = fsm.get_append_buffer();
BOOST_ASSERT(data.size() <= buffer.size());
std::copy(data.begin(), data.end(), buffer.begin());
}

3
test/common.hpp
View File

@@ -1,6 +1,7 @@
#pragma once
#include <boost/redis/connection.hpp>
#include <boost/redis/detail/reader_fsm.hpp>
#include <boost/redis/operation.hpp>
#include <boost/asio/awaitable.hpp>
@@ -34,3 +35,5 @@ void run(
boost::redis::config cfg = make_test_config(),
boost::system::error_code ec = boost::asio::error::operation_aborted,
boost::redis::operation op = boost::redis::operation::receive);
void append_read_data(boost::redis::detail::reader_fsm& fsm, std::string_view data);

16
test/test_exec_fsm.cpp
View File

@@ -15,6 +15,8 @@
#include <boost/core/lightweight_test.hpp>
#include <boost/system/error_code.hpp>
#include "common.hpp"
#include <cstddef>
#include <memory>
#include <optional>
@@ -117,8 +119,7 @@ void test_success()
BOOST_TEST_EQ(mpx.commit_write(), 0u); // all requests expect a response
// Simulate a successful read
mpx.get_read_buffer() = "$5\r\nhello\r\n";
auto req_status = mpx.consume_next(ec);
auto req_status = mpx.consume_next("$5\r\nhello\r\n", ec);
BOOST_TEST_EQ(ec, error_code());
BOOST_TEST_EQ(req_status.first.value(), false); // it wasn't a push
BOOST_TEST_EQ(req_status.second, 11u); // the entire buffer was consumed
@@ -159,10 +160,9 @@ void test_parse_error()
// The second field should be a number (rather than the empty string).
// Note that although part of the buffer was consumed, the multiplexer
// currently throws this information away.
mpx.get_read_buffer() = "*2\r\n$5\r\nhello\r\n:\r\n";
auto req_status = mpx.consume_next(ec);
auto req_status = mpx.consume_next("*2\r\n$5\r\nhello\r\n:\r\n", ec);
BOOST_TEST_EQ(ec, error::empty_field);
BOOST_TEST_EQ(req_status.second, 0u);
BOOST_TEST_EQ(req_status.second, 15u);
BOOST_TEST_EQ(input.done_calls, 1u);
// This will awaken the exec operation, and should complete the operation
@@ -218,8 +218,7 @@ void test_not_connected()
BOOST_TEST_EQ(mpx.commit_write(), 0u); // all requests expect a response
// Simulate a successful read
mpx.get_read_buffer() = "$5\r\nhello\r\n";
auto req_status = mpx.consume_next(ec);
auto req_status = mpx.consume_next("$5\r\nhello\r\n", ec);
BOOST_TEST_EQ(ec, error_code());
BOOST_TEST_EQ(req_status.first.value(), false); // it wasn't a push
BOOST_TEST_EQ(req_status.second, 11u); // the entire buffer was consumed
@@ -342,8 +341,7 @@ void test_cancel_notwaiting_notterminal()
BOOST_TEST_EQ_MSG(act, exec_action_type::wait_for_response, tc.name);
// Simulate a successful read
mpx.get_read_buffer() = "$5\r\nhello\r\n";
auto req_status = mpx.consume_next(ec);
auto req_status = mpx.consume_next("$5\r\nhello\r\n", ec);
BOOST_TEST_EQ_MSG(ec, error_code(), tc.name);
BOOST_TEST_EQ_MSG(req_status.first.value(), false, tc.name); // it wasn't a push
BOOST_TEST_EQ_MSG(req_status.second, 11u, tc.name); // the entire buffer was consumed

1
test/test_low_level.cpp
View File

@@ -528,6 +528,7 @@ BOOST_AUTO_TEST_CASE(cover_error)
check_error("boost.redis", boost::redis::error::sync_receive_push_failed);
check_error("boost.redis", boost::redis::error::incompatible_node_depth);
check_error("boost.redis", boost::redis::error::resp3_hello);
check_error("boost.redis", boost::redis::error::exceeds_maximum_read_buffer_size);
}
std::string get_type_as_str(boost::redis::resp3::type t)

86
test/test_low_level_sync_sans_io.cpp
View File

@@ -11,9 +11,12 @@
#include <boost/redis/resp3/node.hpp>
#include <boost/redis/resp3/serialization.hpp>
#include <boost/redis/resp3/type.hpp>
#include <boost/redis/detail/read_buffer.hpp>
#define BOOST_TEST_MODULE conn_quit
#include <boost/test/included/unit_test.hpp>
#include "common.hpp"
#include <iostream>
#include <string>
@@ -30,6 +33,7 @@ using boost::redis::generic_response;
using boost::redis::resp3::node;
using boost::redis::resp3::to_string;
using boost::redis::any_adapter;
using boost::system::error_code;
BOOST_AUTO_TEST_CASE(low_level_sync_sans_io)
{
@@ -258,10 +262,8 @@ BOOST_AUTO_TEST_CASE(multiplexer_push)
generic_response resp;
mpx.set_receive_response(resp);
mpx.get_read_buffer() = ">2\r\n+one\r\n+two\r\n";
boost::system::error_code ec;
auto const ret = mpx.consume_next(ec);
auto const ret = mpx.consume_next(">2\r\n+one\r\n+two\r\n", ec);
BOOST_TEST(ret.first.value());
BOOST_CHECK_EQUAL(ret.second, 16u);
@@ -282,16 +284,17 @@ BOOST_AUTO_TEST_CASE(multiplexer_push_needs_more)
generic_response resp;
mpx.set_receive_response(resp);
std::string msg;
// Only part of the message.
mpx.get_read_buffer() = ">2\r\n+one\r";
msg += ">2\r\n+one\r";
boost::system::error_code ec;
auto ret = mpx.consume_next(ec);
auto ret = mpx.consume_next(msg, ec);
BOOST_TEST(!ret.first.has_value());
mpx.get_read_buffer().append("\n+two\r\n");
ret = mpx.consume_next(ec);
msg += "\n+two\r\n";
ret = mpx.consume_next(msg, ec);
BOOST_TEST(ret.first.value());
BOOST_CHECK_EQUAL(ret.second, 16u);
@@ -378,20 +381,14 @@ BOOST_AUTO_TEST_CASE(multiplexer_pipeline)
BOOST_TEST(item2.done);
BOOST_TEST(!item3.done);
// Simulates a socket read by putting some data in the read buffer.
mpx.get_read_buffer().append("+one\r\n");
// Consumes the next message in the read buffer.
boost::system::error_code ec;
auto const ret = mpx.consume_next(ec);
auto const ret = mpx.consume_next("+one\r\n", ec);
// The read operation should have been successfull.
BOOST_TEST(ret.first.has_value());
BOOST_TEST(ret.second != 0u);
// The read buffer should also be empty now
BOOST_TEST(mpx.get_read_buffer().empty());
// The last request still did not get a response.
BOOST_TEST(item1.done);
BOOST_TEST(item2.done);
@@ -399,3 +396,64 @@ BOOST_AUTO_TEST_CASE(multiplexer_pipeline)
// TODO: Check the first request was removed from the queue.
}
BOOST_AUTO_TEST_CASE(read_buffer_prepare_error)
{
using boost::redis::detail::read_buffer;
read_buffer buf;
// Usual case, max size is bigger then requested size.
auto ec = buf.prepare_append(10, 10);
BOOST_TEST(!ec);
buf.commit_append(10);
// Corner case, max size is equal to the requested size.
ec = buf.prepare_append(10, 20);
BOOST_TEST(!ec);
buf.commit_append(10);
buf.consume_committed(20);
auto const tmp = buf;
// Error case, max size is smaller to the requested size.
ec = buf.prepare_append(10, 9);
BOOST_TEST(ec == error_code{boost::redis::error::exceeds_maximum_read_buffer_size});
// Check that an error call has no side effects.
auto const res = buf == tmp;
BOOST_TEST(res);
}
BOOST_AUTO_TEST_CASE(read_buffer_prepare_consume_only_committed_data)
{
using boost::redis::detail::read_buffer;
read_buffer buf;
auto ec = buf.prepare_append(10, 10);
BOOST_TEST(!ec);
// No data has been committed yet so nothing can be consummed.
BOOST_CHECK_EQUAL(buf.consume_committed(5), 0u);
buf.commit_append(10);
// All five bytes can be consumed.
BOOST_CHECK_EQUAL(buf.consume_committed(5), 5u);
// Only the remaining five bytes can be consumed
BOOST_CHECK_EQUAL(buf.consume_committed(7), 5u);
}
BOOST_AUTO_TEST_CASE(read_buffer_check_buffer_size)
{
using boost::redis::detail::read_buffer;
read_buffer buf;
auto ec = buf.prepare_append(10, 10);
BOOST_TEST(!ec);
BOOST_CHECK_EQUAL(buf.get_append_buffer().size(), 10u);
}

73
test/test_reader_fsm.cpp
View File

@@ -12,6 +12,8 @@
#include <boost/core/lightweight_test.hpp>
#include <boost/system/error_code.hpp>
#include "common.hpp"
namespace net = boost::asio;
namespace redis = boost::redis;
using boost::system::error_code;
@@ -51,13 +53,15 @@ void test_push()
BOOST_TEST_EQ(act.type_, action::type::append_some);
// The fsm is asking for data.
mpx.get_read_buffer().append(">1\r\n+msg1\r\n");
mpx.get_read_buffer().append(">1\r\n+msg2 \r\n");
mpx.get_read_buffer().append(">1\r\n+msg3 \r\n");
auto const bytes_read = mpx.get_read_buffer().size();
std::string const payload =
">1\r\n+msg1\r\n"
">1\r\n+msg2 \r\n"
">1\r\n+msg3 \r\n";
append_read_data(fsm, payload);
// Deliver the 1st push
act = fsm.resume(bytes_read, ec, cancellation_type_t::none);
act = fsm.resume(payload.size(), ec, cancellation_type_t::none);
BOOST_TEST_EQ(act.type_, action::type::notify_push_receiver);
BOOST_TEST_EQ(act.push_size_, 11u);
BOOST_TEST_EQ(act.ec_, error_code());
@@ -100,20 +104,20 @@ void test_read_needs_more()
std::string const msg[] = {">3\r", "\n+msg1\r\n+ms", "g2\r\n+msg3\r\n"};
// Passes the first part to the fsm.
mpx.get_read_buffer().append(msg[0]);
append_read_data(fsm, msg[0]);
act = fsm.resume(msg[0].size(), ec, cancellation_type_t::none);
BOOST_TEST_EQ(act.type_, action::type::needs_more);
BOOST_TEST_EQ(act.ec_, error_code());
// Passes the second part to the fsm.
mpx.get_read_buffer().append(msg[1]);
append_read_data(fsm, msg[1]);
act = fsm.resume(msg[1].size(), ec, cancellation_type_t::none);
BOOST_TEST_EQ(act.type_, action::type::needs_more);
BOOST_TEST_EQ(act.ec_, error_code());
// Passes the third and last part to the fsm, next it should ask us
// to deliver the message.
mpx.get_read_buffer().append(msg[2]);
append_read_data(fsm, msg[2]);
act = fsm.resume(msg[2].size(), ec, cancellation_type_t::none);
BOOST_TEST_EQ(act.type_, action::type::notify_push_receiver);
BOOST_TEST_EQ(act.push_size_, msg[0].size() + msg[1].size() + msg[2].size());
@@ -141,16 +145,16 @@ void test_read_error()
BOOST_TEST_EQ(act.type_, action::type::append_some);
// The fsm is asking for data.
mpx.get_read_buffer().append(">1\r\n+msg1\r\n");
auto const bytes_read = mpx.get_read_buffer().size();
std::string const payload = ">1\r\n+msg1\r\n";
append_read_data(fsm, payload);
// Deliver the data
act = fsm.resume(bytes_read, {net::error::operation_aborted}, cancellation_type_t::none);
act = fsm.resume(payload.size(), {net::error::operation_aborted}, cancellation_type_t::none);
BOOST_TEST_EQ(act.type_, action::type::cancel_run);
BOOST_TEST_EQ(act.ec_, error_code());
// Finish
act = fsm.resume(bytes_read, ec, cancellation_type_t::none);
act = fsm.resume(0, ec, cancellation_type_t::none);
BOOST_TEST_EQ(act.type_, action::type::done);
BOOST_TEST_EQ(act.ec_, error_code{net::error::operation_aborted});
}
@@ -171,16 +175,16 @@ void test_parse_error()
BOOST_TEST_EQ(act.type_, action::type::append_some);
// The fsm is asking for data.
mpx.get_read_buffer().append(">a\r\n");
auto const bytes_read = mpx.get_read_buffer().size();
std::string const payload = ">a\r\n";
append_read_data(fsm, payload);
// Deliver the data
act = fsm.resume(bytes_read, {}, cancellation_type_t::none);
act = fsm.resume(payload.size(), {}, cancellation_type_t::none);
BOOST_TEST_EQ(act.type_, action::type::cancel_run);
BOOST_TEST_EQ(act.ec_, error_code());
// Finish
act = fsm.resume(bytes_read, {}, cancellation_type_t::none);
act = fsm.resume(0, {}, cancellation_type_t::none);
BOOST_TEST_EQ(act.type_, action::type::done);
BOOST_TEST_EQ(act.ec_, error_code{redis::error::not_a_number});
}
@@ -201,16 +205,16 @@ void test_push_deliver_error()
BOOST_TEST_EQ(act.type_, action::type::append_some);
// The fsm is asking for data.
mpx.get_read_buffer().append(">1\r\n+msg1\r\n");
auto const bytes_read = mpx.get_read_buffer().size();
std::string const payload = ">1\r\n+msg1\r\n";
append_read_data(fsm, payload);
// Deliver the data
act = fsm.resume(bytes_read, {}, cancellation_type_t::none);
act = fsm.resume(payload.size(), {}, cancellation_type_t::none);
BOOST_TEST_EQ(act.type_, action::type::notify_push_receiver);
BOOST_TEST_EQ(act.ec_, error_code());
// Resumes from notifying a push with an error.
act = fsm.resume(bytes_read, net::error::operation_aborted, cancellation_type_t::none);
act = fsm.resume(0, net::error::operation_aborted, cancellation_type_t::none);
BOOST_TEST_EQ(act.type_, action::type::cancel_run);
// Finish
@@ -219,10 +223,39 @@ void test_push_deliver_error()
BOOST_TEST_EQ(act.ec_, error_code{net::error::operation_aborted});
}
void test_max_read_buffer_size()
{
multiplexer mpx;
generic_response resp;
mpx.set_receive_response(resp);
reader_fsm fsm{mpx};
fsm.set_config({5, 7});
error_code ec;
action act;
// Initiate
act = fsm.resume(0, ec, cancellation_type_t::none);
BOOST_TEST_EQ(act.type_, action::type::setup_cancellation);
act = fsm.resume(0, ec, cancellation_type_t::none);
BOOST_TEST_EQ(act.type_, action::type::append_some);
// Passes the first part to the fsm.
std::string const part1 = ">3\r\n";
append_read_data(fsm, part1);
act = fsm.resume(part1.size(), {}, cancellation_type_t::none);
BOOST_TEST_EQ(act.type_, action::type::cancel_run);
BOOST_TEST_EQ(act.ec_, error_code());
act = fsm.resume({}, {}, cancellation_type_t::none);
BOOST_TEST_EQ(act.type_, action::type::done);
BOOST_TEST_EQ(act.ec_, redis::error::exceeds_maximum_read_buffer_size);
}
} // namespace
int main()
{
test_max_read_buffer_size();
test_push_deliver_error();
test_read_needs_more();
test_push();