boostbufferboost-asioc++20nlohmann-json

boost::asio::async_read_until with custom match_char to accept only JSON format

I've been trying to change match_char function to accept only JSON messages when reading data from a socket.

I have 2 implementations (one does not work and the other one works but I don't think it's efficient).

1- First approach (working)

    typedef boost::asio::buffers_iterator<boost::asio::streambuf::const_buffers_type> buffer_iterator;

    static std::pair<buffer_iterator, bool> match_json2(const buffer_iterator begin,
                                                            const buffer_iterator end) {
        buffer_iterator i = begin;
        while (i != end) {
            if ((*i == ']') || (*i == '}')) {
                return std::make_pair(i, true);
            }
            *i++;
        }
        return std::make_pair(i, false);
    }

With this definition, I read in a loop and reconstruct the json. This is a working version, but if I receive a message different from a valid json, I stay in the loop, can't clear tmp_response and never recover from it...

        std::string read_buffer_string() {
            std::string response;
            bool keepReading = true;
            while (keepReading) {
                std::string tmp_response;
                async_read_until(s, ba::dynamic_buffer(tmp_response), match_json2, yc);
                if (!tmp_response.empty()) {
                    response += tmp_response;
                    if (nlohmann::json::accept(response)) {
                        keepReading = false;
                    }
                }
            }
            return response;
        }

  1. Second approach (not working). Ideally I would like something like this one (this implementation doesn't work because begin iterator doesn't always point to the start of the message - I guess some data is already been transferred to the buffer-, and therefore match_json returns invalid values.

     static std::pair<buffer_iterator, bool> match_json(const buffer_iterator begin,
                                                         const buffer_iterator end) {
     buffer_iterator i = begin;
     while (i != end) {
         if ((*i == ']') || (*i == '}')) {
             std::string _message(begin, i);
             std::cout << _message << std::endl;
             if (nlohmann::json::accept(_message)) {
                 return std::make_pair(i, true);
             }
         }
         *i++;
     }
     return std::make_pair(i, false);
 }

And then call it like this:

        std::string read_buffer_string() {
            std::string response;
            async_read_until(s, ba::dynamic_buffer(response), match_json, yc);
            return response;
        }

Does anybody now a more efficient way to do it? Thanks in advance! :)

Solution

  • Of course, right after posting my other answer I remembered that Boost has accepted Boost JSON in 1.75.0.

    It does stream parsing way more gracefully: https://www.boost.org/doc/libs/1_75_0/libs/json/doc/html/json/ref/boost__json__stream_parser.html#json.ref.boost__json__stream_parser.usage

    It actually deals with trailing data as well!

    stream_parser p;                  // construct a parser
std::size_t n;                    // number of characters used
n = p.write_some( "[1,2" );       // parse some of a JSON
assert( n == 4 );                 // all characters consumed
n = p.write_some( ",3,4] null" ); // parse the remainder of the JSON
assert( n == 6 );                 // only some characters consumed
assert( p.done() );               // we have a complete JSON
value jv = p.release();           // take ownership of the value

    I would also submit that this could be a better match for a CompletionCondition: see https://www.boost.org/doc/libs/1_75_0/doc/html/boost_asio/reference/read/overload3.html

    Here's an implementation that I tested with:

    template <typename Buffer, typename SyncReadStream>
static size_t read_json(SyncReadStream& s, Buffer buf,
    boost::json::value& message, boost::json::parse_options options = {})
{
    boost::json::stream_parser p{{}, options};

    size_t total_parsed = 0;
    boost::asio::read(s, buf, [&](boost::system::error_code ec, size_t /*n*/) {
        size_t parsed = 0;

        for (auto& contiguous : buf.data()) {
            parsed += p.write_some(
                boost::asio::buffer_cast<char const*>(contiguous),
                contiguous.size(), ec);
        }
        buf.consume(parsed);
        total_parsed += parsed;
        return ec || p.done(); // true means done
    });

    message = p.release(); // throws if incomplete
    return total_parsed;
}

    Adding a delegating overload for streambufs:

    template <typename SyncReadStream, typename Alloc>
static size_t read_json(SyncReadStream& s,
    boost::asio::basic_streambuf<Alloc>& buf,
    boost::json::value& message,
    boost::json::parse_options options = {})
{
    return read_json(s, boost::asio::basic_streambuf_ref<Alloc>(buf), message, options);
}

    Demo Program

    This demo program adds the test-cases from earlier as well as a socket client with some benchmark statistics added. Arguments:

    Live On Compiler Explorer¹

    #include <boost/spirit/home/x3.hpp>
#include <boost/fusion/adapted.hpp>
#include <iomanip>
#include <iostream>
namespace x3 = boost::spirit::x3;

int main() {
    std::string const s = 
        "? 8==2 : true ! false"
        "? 9==3 : 'book' ! 'library'";

    using expression = std::string;
    using ternary = std::tuple<expression, expression, expression>;
    std::vector<ternary> parsed;

    auto expr_ = x3::lexeme [+~x3::char_("?:!")];
    auto ternary_ = "?" >> expr_ >> ":" >> expr_ >> "!" >> expr_;

    std::cout << "=== parser approach:\n";
    if (x3::phrase_parse(begin(s), end(s), *x3::seek[ ternary_ ], x3::space, parsed)) {

        for (auto [cond, e1, e2] : parsed) {
            std::cout
                << " condition " << std::quoted(cond) << "\n"
                << " true expression " << std::quoted(e1) << "\n"
                << " else expression " << std::quoted(e2) << "\n"
                << "\n";
        }
    } else {
        std::cout << "non matching" << '\n';
    }
}

    With test prints:

     ----- valid test cases
Testing {}                     -> Success {}
Testing {"a":4, "b":5}         -> Success {"a":4,"b":5}
Testing []                     -> Success []
Testing [4, "b"]               -> Success [4,"b"]
 ----- incomplete test cases
Testing {                      -> (incomplete...)
Testing {"a":4, "b"            -> (incomplete...)
Testing [                      -> (incomplete...)
Testing [4, "                  -> (incomplete...)
 ----- invalid test cases
Testing }                      -> syntax error
Testing "a":4 }                -> Success "a" -- remaining `:4 }`
Testing ]                      -> syntax error
 ----- excess input test cases
Testing {}{"a":4, "b":5}       -> Success {} -- remaining `{"a":4, "b":5}`
Testing []["a", "b"]           -> Success [] -- remaining `["a", "b"]`
Testing {} bogus trailing data -> Success {} -- remaining `bogus trailing data`

    With socket client some demos:

    Mean packet size: 16 in 2 packets
Request: 28 bytes
Request: {"a":4,"b":"5"} bytes
Remaining data: "bye
"
took 0.000124839s, ~0.213899MiB/s

    With a large (448MiB) location_history.json:

    Mean packet size: 511.999 in 917791 packets
Request: 469908167 bytes
 (large request output suppressed)
took 3.30509s, ~135.59MiB/s

    enter image description here

    ¹ linking non-header only libraries is not supported on Compiler Explorer