Repository files navigation

Boost::ASIO low-level redis client (connector),githubgitee

• header only
• zero-copy (currently only for received replies from Redis)
• low-level controls (i.e. you can cancel, or do manual DNS-resolving before a connection)
• unix domain sockets support
• works on linux (clang, gcc) and windows (msvc)
• synchronous & asynchronous interface
• inspired bybeast
• requirements: boostv1.77 minimum

• [feature, breakging] modernize API to use completion token instead ofusing completion handler, which makes it possible to usebredis withcoroutines. Thanks toUsevalad Sauta.Successfully tested with boostv1.77, lower versions might not work

• [feature, possible breakage] addBOOST_ASIO_NO_DEPRECATED definitionfor better support boostv1.74 and modernize boost API usage

• [bugfix] avoid access to protected destructor (c++17 compatibility)

• [bugfix] critical bug in protcol serialization on empty values

• relaxed c++ compiler requirements: c++11 can be used instead of c++14

• minor parsing speed improvements (upto 10% in synthetic tests)
• fix compilation issues on boost::asio 1.70
• make it possible to useDynamicBuffer_v2 (dynamic_string_buffer, dynamic_vector_buffer)from boost::asio 1.70 in addition tostreambuf.DynamicBuffer_v1 was actually neversupported bybredis
• [API breakage]boos::asio::use_future cannot be used withbredis andboost::asiopriorv1.70 (seeissue). If you needuse_future then either upgrade boost::asio or use previousbredis version.

• theparsing_policy::drop_result was documented and made applicable in client code
• updated preformance results
• fixed compliation warnings (-Wall -Wextra -pedantic -Werror)
• added shortcut headerinclude/bredis.hpp
• added redis-streams usage example
• added multi-thread example

• fixed level 4 warning in MSVC
• fixed compilation issues on archlinux
• improved documentation (numerous typos etc.)

• [bugfix] removed unneededtx_buff.commit() onasync_write which corrupted buffer

• improved protocol parser (no memory allocations during input stream validity check)
• more detailed information inprotocol_error_t
• added asyncincr speed test example
• [small API breakage]positive_parse_result_t was enriched with parcing policy;now instead ofpositive_parse_result_t<Iterator> should be written:

using Policy = r::parsing_policy::keep_result;usingresult_t = r::parse_result_mapper_t<Iterator, Policy>;

• [small API breakage]protocol_error_t instead ofstd::string what membernow containsboost::system::error_code code

• added windows support
• added coroutines & futures support
• generalised (templated) buffer support
• changed return type: instead of result of parsing just result markers are returned, extraction of result can be done as separate step
• dropped queing support (queuing policy should be implemented at more higher levels)
• dropped subscription support (can be implemented at higher levels)
• dropped internal buffers (can be implemented at higher levels)
• dropped explicit cancellation (socket reference can be passed to connector, and cancellationcan be done on the socket object outside of the connector)

• initial version

Results achieved withexamples/speed_test_async_multi.cpp for 1 thread, Intel Core i7-8550U, void-linux, gcc 8.3.0

These results are not completely fair, because of the usage of different semantics in theAPIs; however they are still interesting, as they are using differentunderlying event libraries (Boost::ASIO vslibev) as well as redis protocolparsing libraries (written from scratch vshiredis)

(*) bredis with drop_result policy, i.e. replies from redis server arescanned only for formal correctness and never delivered to the caller.

The general idea is that the result of trying to parse a redis reply can be either: not enough data, protocol error (in an extreme case) or some positive parse result. The last one is justmarkers of the result, which is actually stored in thereceive buffer (i.e. outside of markers, and outside of the bredis-connection).

The further work with markers depends on your needs: it is possible to eitherscan the result for the expected results (e.g. for aPONG reply on aPING command, or forOK/QUEUED replies onMULTI/EXEC commands) or toextract the results (the common redis types:nil,string,error,int or a (recursive) array of them).

When the data in the receive buffer is no longer required, it should be consumed.

Scan example:

#include"bredis/MarkerHelpers.hpp"...namespacer= bredis;...using Buffer = boost::asio::streambuf;...Buffer rx_buff;auto result_markers = c.read(rx_buff);/* check for the response*/auto eq_pong = r::marker_helpers::equality<Iterator>("PONG");/* print true or false*/std::cout << boost::apply_visitor(eq_pong, result_markers.result) <<"\n";/* consume the buffers, after finishing work with the markers*/rx_buff.consume(result_markers.consumed);

Forextraction of results it is possible to use either one of the shipped extractors or to write a custom one. Shipped extractors detach (copy / convert) the extraction results from the receive buffer.

#include"bredis/Extract.hpp"...auto result_markers = c.read(rx_buff);auto extract = boost::apply_visitor(r::extractor<Iterator>(), result_markers.result);/* safe to consume buffers now*/rx_buff.consume(result_markers.consumed);/* we know what the type is, safe to unpack to string*/auto &reply_str = boost::get<r::extracts::string_t>(extract);/* print "PONG"*/std::cout << reply_str.str <<"\n";

Custom extractors (visitors) might be useful for performance-sensitive cases, e.g. when JSON is re-constructed in-place by using string reply markerswithout re-allocating the whole JSON-string reply.

The underlying reason for the decision to retrieve the final results in two steps (get markers and then scan/extract results) is that thereceive buffer might be scattered (fragmented). In such cases scan and extraction can be performed without gathering receive buffers (i.e. without flattening / linearizing it) if they are separate steps.

In other words,markers havereference semantics (they refer to memory regions in the buffer, but do not own it), whileextracted results havevalue semantics (they take ownership).

#include"bredis/Connection.hpp"#include"bredis/MarkerHelpers.hpp"#include<boost/variant.hpp>...namespacer= bredis;namespaceasio= boost::asio;.../* define used types*/usingsocket_t = asio::ip::tcp::socket;using Buffer = boost::asio::streambuf;using Iterator =typename r::to_iterator<Buffer>::iterator_t;.../* establishing connection to redis is outside of bredis*/asio::ip::tcp::endpointend_point(asio::ip::make_address("127.0.0.1"), port);socket_tsocket(io_service, end_point.protocol());socket.connect(end_point);/* wrap socket to bredis connection*/r::Connection<socket_t>c(std::move(socket));/* synchronously write command*/c.write("ping");/* buffer is allocated outside of bredis connection*/Buffer rx_buff;/* get the result markers*/auto result_markers = c.read(rx_buff);/* check for the response*/auto eq_pong = r::marker_helpers::equality<Iterator>("PONG");/* print true*/std::cout << boost::apply_visitor(eq_pong, result_markers.result) <<"\n";/* consume the buffers, after finishing work with the markers*/rx_buff.consume(result_markers.consumed);

In the ping example above thePONG reply string from redis is not (re)allocated, but directly scanned from therx_buff using a result markers. This can be useful for performance-sensitive cases, e.g. when JSON is re-constructed in-place by using string reply markerswithout re-allocating the whole JSON-string reply.

In cases where you need toextract the reply (i.e. to detach it fromrx_buff), the following can be done:

#include"bredis/Extract.hpp"...auto result_markers = c.read(rx_buff);/* extract the results*/auto extract = boost::apply_visitor(r::extractor<Iterator>(), result_markers.result);/* safe to consume buffers now*/rx_buff.consume(result_markers.consumed);/* we know what the type is, safe to unpack to string*/auto &reply_str = boost::get<r::extracts::string_t>(extract);/* print "PONG"*/std::cout << reply_str.str <<"\n";

The examples above throw an exception in case of I/O or protocol errors. Another way to use the API is

boost::system::error_code ec;c.write("ping", ec);...parse_result = c.read(rx_buff, ec);

in case you don't want the throw-exception behaviour.

#include"bredis/Connection.hpp"#include"bredis/MarkerHelpers.hpp"...namespacer= bredis;namespaceasio= boost::asio;namespacesys= boost::system;...usingsocket_t = asio::ip::tcp::socket;using Buffer = boost::asio::streambuf;using Iterator =typename r::to_iterator<Buffer>::iterator_t;using Policy = r::parsing_policy::keep_result;usingresult_t = r::parse_result_mapper_t<Iterator, Policy>;.../* establishing the connection to redis is outside of bredis*/asio::ip::tcp::endpointend_point(asio::ip::make_address("127.0.0.1"), port);socket_tsocket(io_service, end_point.protocol());socket.connect(end_point);...Buffer tx_buff, rx_buff;c.async_write(    tx_buff, r::single_command_t{"llen","my-queue"}, [&](const sys::error_code &ec, std::size_t bytes_transferred) {/* tx_buff must be consumed when it is no longer needed*/        tx_buff.consume(bytes_transferred);        c.async_read(rx_buff, [&](const sys::error_code &ec,result_t &&r) {/* see above how to work with the result*/auto extract =boost::apply_visitor(r::extractor<Iterator>(), r.result);auto &queue_size = boost::get<r::extracts::int_t>(extract);            std::cout <<"queue size:" << queue_size <<"\n";            .../* consume rx_buff when it is no longer needed*/            rx_buff.consume(r.consumed);        });    });

In the example above separate receive and transfer buffers are used. In theory you can use only one buffer for both operations, but you must ensure that it will not be used simultaneously for reading and writing, in other words you cannot use thepipelining redis feature.

The same as above, except the underlying socket type must be changed:

usingsocket_t = asio::local::stream_protocol::socket;

There is no specific support for subscriptions, but you can easily build your own like

r::single_command_t subscribe_cmd{"subscribe","some-channel1","some-channel2"};c.write(subscribe_cmd);Buffer rx_buff;/* get the 2 confirmations, as we subscribed to 2 channels*/r::marker_helpers::check_subscription<Iterator> check_subscription{std::move(subscribe_cmd)};for (auto i =0; i <2; ++i){auto result_markers = c.read(rx_buff);bool confirmed =boost::apply_visitor(check_subscription, result_markers.result);if (!confirmed) {// do something!    ...;  }  rx_buff.consume(result_markers.consumed);}while(true) {auto result_markers = c.read(rx_buff);auto extract =boost::apply_visitor(r::extractor<Iterator>(), result_markers.result);  rx_buff.consume(result_markers.consumed);/* process the result*/auto& array_reply = boost::get<r::extracts::array_holder_t>(extract);auto* type_reply = boost::get<r::extracts::string_t>(&array_reply.elements[0]);if (type_reply && type_reply->str =="message") {auto& channel = boost::get<r::extracts::string_t>(array_reply.elements[1]);auto& payload = boost::get<r::extracts::string_t>(array_reply.elements[2]);      ...  }}

Seeexamples/synch-subscription.cpp for the full example.

These work similarly to the synchronous approach. However you have to provide a read callback initially and again after each successfull read

using Policy = r::parsing_policy::keep_result;using ParseResult = r::parse_result_mapper_t<Iterator, Policy>;usingread_callback_t = std::function<void(const boost::system::error_code &error_code, ParseResult &&r)>;using Extractor = r::extractor<Iterator>;.../* we can execute the subscription command synchronously, as it is easier*/c.command("subscribe","channel-1","channel-2");...Buffer rx_buff;read_callback_t notification_callback = [&](const boost::system::error_code,                                            ParseResult &&r) {auto extract =boost::apply_visitor(Extractor(), r.result);    rx_buff.consume(r.consumed);/* process the result, see above*/    .../* re-trigger new message processing*/    c.async_read(rx_buff, notification_callback);};/* initialise listening subscriptions*/c.async_read(rx_buff, notification_callback);

Seeexamples/stream-parse.cpp for the full example.

There is no specific support for transactions in bredis, but you can easily build your own for your needs.

First, wrap your commands into a transaction:

r::command_container_t tx_commands = {r::single_command_t("MULTI"),r::single_command_t("INCR","foo"),r::single_command_t("GET","bar"),r::single_command_t("EXEC"),};r::command_wrapper_tcmd(tx_commands);c.write(cmd);

Then, as above there were4 redis commands, there we should receive4 redisreplies:OK,QUEUED,QUEUED followed by the array of results of the execution of the commandsin the transaction (i.e. results forINCR andGET above)

Buffer rx_buff;c.async_read(rx_buff, [&](const sys::error_code &ec,result_t&& r){auto &replies = boost::get<r::markers::array_holder_t<Iterator>>(r.result);/* scan stream for OK, QUEUED, QUEUED*/    ...assert(replies.elements.size() ==4);auto eq_OK = r::marker_helpers::equality<Iterator>("OK");auto eq_QUEUED = r::marker_helpers::equality<Iterator>("QUEUED");assert(boost::apply_visitor(eq_OK, replies.elements[0]));assert(boost::apply_visitor(eq_QUEUED, replies.elements[1]));assert(boost::apply_visitor(eq_QUEUED, replies.elements[2]));/* get tx replies*/auto &tx_replies = boost::get<r::markers::array_holder_t<Iterator>>(replies.elements[3]);    ...;    rx_buff.consume(r.consumed);},4);/* pay attention here*/

Done in a similiar way as inBoost::ASIO (special thanks to Vinnie Falko for the suggestion)

#include<boost/asio/use_future.hpp>...Buffer rx_buff, tx_buff;auto f_tx_consumed = c.async_write(tx_buff,"ping", asio::use_future);auto f_result_markers = c.async_read(rx_buff, asio::use_future);...tx_buff.consume(f_tx_consumed.get());auto result_markers = f_result_markers.get();/* scan/extract result, and consume rx_buff as usual*/

#include<boost/asio/spawn.hpp>Buffer rx_buff, tx_buff;boost::asio::spawn(    io_service, [&](boost::asio::yield_context yield) mutable {        boost::system::error_code error_code;auto consumed = c.async_write(tx_buff,"ping", yield[error_code]);        tx_buff.consume(consumed);        ...auto parse_result = c.async_read(rx_buff, yield[error_code],1);/* scan/extract result*/        rx_buff.consume(parse_result.consumed);    });

There is no specific support for streams (appeared in redis 5.0) in bredis,they are just usualXADD,XRANGE etc. commands and corresponding replies.

...Buffer rx_buff;c.write(r::single_command_t{"XADD","mystream","*","cpu-temp","23.4","load","2.3" });auto parse_result1 = c.read(rx_buff);auto extract1 = boost::apply_visitor(Extractor(), parse_result1.result);auto id1 = boost::get<r::extracts::string_t>(extract1);c.write(r::single_command_t{"XADD","mystream","*","cpu-temp","23.2","load","2.1" });auto parse_result2 = c.read(rx_buff);auto extract2 = boost::apply_visitor(Extractor(), parse_result2.result);auto id2 = boost::get<r::extracts::string_t>(extract2);rx_buff.consume(parse_result2.consumed);c.write(r::single_command_t{"XRANGE" ,"mystream",  id1.str, id2.str});auto parse_result3 = c.read(rx_buff);auto extract3 = boost::apply_visitor(Extractor(), parse_result3.result);rx_buff.consume(parse_result3.consumed);auto& outer_arr = boost::get<r::extracts::array_holder_t>(extract3);auto& inner_arr1 = boost::get<r::extracts::array_holder_t>(outer_arr.elements[0]);auto& inner_arr2 = boost::get<r::extracts::array_holder_t>(outer_arr.elements[1]);...

Seet/SocketWithLogging.hpp for an example. The main idea is quite simple:Instead of providing a real socket implementation supplied byBoost::ASIO,provide a wrapper (proxy) which willspy on the traffic beforedelegating it to/from aBoost::ASIO socket.

There is nothing specific to this in bredis. If you need low-level socketoperations, instead of movingsocket into bredis connection, you cansimply move areference to it and keep (own) the socket somewhereoutside of the bredis connection.

usingsocket_t = asio::ip::tcp::socket;usingnext_layer_t =socket_t &;...asio::ip::tcp::endpointend_point(asio::ip::make_address("127.0.0.1"), port);socket_tsocket(io_service, end_point.protocol());socket.connect(end_point);r::Connection<next_layer_t>c(socket);...socket.cancel();

bredis itself is thread-agnostic, however the underlying socket (next_layer_t)and used buffers are usually not thread-safe. To handle that in multi-theadenvironment the access to those objects should be sequenced viaasio::io_context::strand . See theexamples/multi-threads-1.cpp.

The performance still can be boosted if it is known beforehand that the response fromredis server is not needed at all. For example, the only possible response toPINGcommand isPONG reply, usually there is no sense it validating thatPONG reply,as soon as it is known, that redis-server alredy delivered ussome reply(in practice it isPONG). Another example isSET command, when redis-serverusually replies withOK.

Withparsing_policy::drop_result the reply result is just verified with formalcompliance to redis protocol, and then it is discarded.

It should be noted, that redis can reply back with error, which aslo correctreply, but the caller side isn't able to see it whenparsing_policy::drop_resultis applied. So, it should be used with care, when you know what your are doing. You havebeen warned.

It is safe, however, to mix different parsing policies on the same connection,i.e. writeSET command and read it's reply withparsing_policy::drop_result andthen writeGET command and read it's reply withparsing_policy::keep_result.See theexamples/speed_test_async_multi.cpp.

There's a convenience header include/bredis.hpp, doing#include "bredis.hpp" will includeevery header under include/bredis/ .

The underlying iterator type used for the dynamic buffer type (e.g.boost::asio::streambuf)

Header:include/bredis/Markers.hpp

Namespace:bredis::markers

boost::variant for the basic types in the redis protocol,i.e. the following marker types :

• nil_t<Iterator>
• int_t<Iterator>
• string_t<Iterator> (simple string and bulk strings)
• error_t<Iterator>
• array_holder_t<Iterator>

The basic type isstring_t<Iterator>, which containsfrom andto members (Iterator)to where the string is held. String does not contain the special redis-protocol symbols or any othermetadata, i.e. it can be used to extract/flatten the whole string.

nil_t<Iterator>,int_t<Iterator>,error_t<Iterator> just have astring memberto point to the underlying string in the redis protocol.

array_holder_t is recursive wrapper for theredis_result_t<Iterator>, it contains aelements member ofstd::array ofredis_result_t<Iterator> type.

Header:include/bredis/Result.hpp

Namespace:bredis

Represents the results of a parse attempt. It is aboost::variant of the following types:

• not_enough_data_t
• protocol_error_t
• positive_parse_result_t<Iterator, Policy>

not_enough_data_t is a empty struct. It means that buffer just does not contain enoughinformation to completely parse it.

protocol_error_t has aboost::system::error_code code member. It describes the errorin the protocol (e.g. when the type in the stream is specified as an integer, but it cannot beconverted to an integer). This error should never occur in production code, meaningthat no (logical) errors are expected in the redis-server nor in the bredis parser. Theerror might occur if the buffer is corrupted.

Policy (namespacebredis::parsing_policy) specifies what to do with the result:Either drop it (bredis::parsing_policy::drop_result) or keep it(bredis::parsing_policy::keep_result). The helperparse_result_mapper_t<Iterator, Policy> helps to get the properpositive_parse_result_t<Iterator, Policy> type.

positive_parse_result_t<Iterator, Policy> contains members:

• markers::redis_result_t<Iterator> result - the result of mark-up buffer; can be usedeither for scanning for particular results or for extraction of results. Valid onlyforkeep_result policy.
• size_t consumed - how many bytes of receive buffer must be consumed afterusing theresult field.

Header:include/bredis/MarkerHelpers.hpp

Namespace:bredis::marker_helpers

Apply thisboost::static_visitor<std::string>s tostringize the result (can be useful for debugging).

Apply thisboost::static_visitor<bool> to find astring in theparsed results (the markup can point to integer types, but as itis transferred as a string anyway, it still can be founded as stringtoo).

Constructor:equality<Iterator>(std::string str)

Thisboost::static_visitor<bool> helper is used to checkwhether the redis reply confirms one of the requested channels. Hence,the constructor ischeck_subscription(single_command_t).

Usually, the redis subscription reply is in the form:

[array] {    [string] "subcribe",    [string] channel_name,    [int] reference}

So it checks that:

1. The redis reply is a 3-element array
2. The 1st reply element is a string, and itcase-insensitivelymatches the command, i.e. it is assumed, thatcommand will besubscribe orpsubscribe depending on the original command
3. That the 3rd reply element is a reference, and it is presentamong the command arguments.

It is possible to reuse the samecheck_subscription<Iterator>onmultiple redis replies to a single subsription command for multiple channels.

Example:

bredis::single_command_t subscribe_cmd{"subscribe","channel-1","channel-2"};...// write the command, so the subscribe_cmd// will be no longer be required...;bredis::marker_helpers::check_subscription<Iterator>    check_subscription{std::move(subscribe_cmd)};...;// get the 1st replyauto parse_result = ...;bool channel_1_ok = boost::apply_visitor(check_subscription, parse_result.result);...;// get the 2nd replyparse_result = ...;bool channel_2_ok = boost::apply_visitor(check_subscription, parse_result.result);

Header:include/bredis/Command.hpp

Namespace:bredis

boost::variant for the basic commands:

• single_command_t
• command_container_t

single_command_t represents a single redis command with all its arguments, e.g.:

// compile-time versionr::single_command_t cmd_ping {"ping"};r::single_command_t cmd_get {"get","queu-name"};...// or runtime-versionstd::vector<std::string> subscription_items {"subscribe","channel-a","channel-b"};r::single_command_t cmd_subscribe {    subscription_items.cbegin(),    subscription_items.cend()};

The arguments must be conversible toboost::string_ref.

command_container_t is astd::vector ofsingle_command_t. It is useful for transactionsor bulk message creation.

Header:include/bredis/Connection.hpp

Namespace:bredis

A thin wrapper aroundNextLayer; represents a connection to redis.NextLayer canbe eitherasio::ip::tcp::socket orasio::ip::tcp::socket& or a custom wrapper, whichfollows the specification ofasio::ip::tcp::socket.

The constructortemplate <typename... Args> Connection(Args &&... args) is used for theconstruction of NextLayer (stream interface).

Stream interface accessors:

• NextLayer &next_layer()
• const NextLayer &next_layer() const

return the underlying stream object.

Performs a synchonous write of a redis command:

• void write(const command_wrapper_t &command)
• void write(const command_wrapper_t &command, boost::system::error_code &ec)

Performs a synchronous read of a redis result until the buffer is parsed orsome error (protocol or I/O) occurs:

• template <typename DynamicBuffer> positive_parse_result_t<Iterator, Policy = bredis::parsing_policy::keep_result> read(DynamicBuffer &rx_buff)
• template <typename DynamicBuffer> positive_parse_result_t<Iterator, Policy = bredis::parsing_policy::keep_result> read(DynamicBuffer &rx_buff, boost::system::error_code &ec);

DynamicBuffer must conform to theboost::asio::streambuf interface.

TheWriteCallback template should be a callable object with the signature:

void (const boost::system::error_code&, std::size_t bytes_transferred)

The asynchronous write has the following signature:

void-or-deducedasync_write(DynamicBuffer &tx_buff,constcommand_wrapper_t &command,                WriteCallback write_callback)

It writes the redis command (or commands) into atransfer buffer, sends themto thenext_layer stream and invokeswrite_callback after completion.

tx_buff must consumebytes_transferred uponwrite_callback invocation.

The client must guarantee thatasync_write is not invoked until the previousinvocation is finished.

ReadCallback template should be a callable object with the signature:

void(boost::system::error_code, r::positive_parse_result_t<Iterator, Policy = bredis::parsing_policy::keep_result>&& result)

The asynchronous read has the following signature:

void-or-deducedasync_read(DynamicBuffer &rx_buff, ReadCallback read_callback,               std::size_t replies_count =1, Policy = bredis::parsing_policy::keep_result{});

It readsreplies_count replies from thenext_layer stream, which will bestored inrx_buff, or until an error (I/O or protocol) is encountered; thenread_callback will be invoked.

Ifreplies_count is greater than1, the result type will always bebredis::array_wrapper_t; if thereplies_count is1 then the result typedepends on redis answer type.

Onread_callback invocation with a successful parse result it is expected,thatrx_buff will consume the amount of bytes specified in theresult.

The client must guarantee thatasync_read is not invoked until the previousinvocation is finished. If you invokeasync_read fromread_callbackdon't forget toconsumerx_buff first, otherwise it leads tosubtle bugs.

MIT

• Derek Colley
• Stefan Hacker
• nkochakian
• Yuval Hager
• Vinnie Falco
• Stephen Coleman
• maxtorm miximtor
• Ronny Nowak
• Stephen Chisholm
• amensel
• Usevalad Sauta

• https://github.com/Cylix/cpp_redis
• https://github.com/hmartiro/redox
• https://github.com/nekipelov/redisclient