<commit_before>#pragma once

#include <array>
#include <silicium/exchange.hpp>
#include <warpcoil/protocol.hpp>
#include <warpcoil/cpp/begin_parse_value.hpp>
#include <warpcoil/cpp/integer_parser.hpp>

namespace warpcoil
{
    namespace cpp
    {
        template <class AsyncReadStream>
        struct buffered_read_stream
        {
            AsyncReadStream &input;
            std::array<std::uint8_t, 512> response_buffer;
            std::size_t response_buffer_used;

            explicit buffered_read_stream(AsyncReadStream &input)
                : input(input)
                , response_buffer_used(0)
            {
            }
        };

        template <class AsyncReadStream>
        struct message_splitter
        {
            explicit message_splitter(AsyncReadStream &input)
                : buffer(input)
                , locked(false)
                , parsing_header(false)
            {
            }

            template <class ResponseHandler>
            void wait_for_response(ResponseHandler handler)
            {
                assert(!waiting_for_response);
                waiting_for_response = [handler](boost::system::error_code ec, request_id request) mutable
                {
                    using boost::asio::asio_handler_invoke;
                    asio_handler_invoke(
                        [&]
                        {
                            handler(ec, request);
                        },
                        &handler);
                };
                if (parsing_header)
                {
                    return;
                }
                set_begin_parse_message(handler);
                begin_parse_message();
            }

            template <class RequestHandler>
            void wait_for_request(RequestHandler handler)
            {
                assert(!waiting_for_request);
                waiting_for_request = [handler](boost::system::error_code ec, request_id request) mutable
                {
                    using boost::asio::asio_handler_invoke;
                    asio_handler_invoke(
                        [&]
                        {
                            handler(ec, request);
                        },
                        &handler);
                };
                if (parsing_header)
                {
                    return;
                }
                set_begin_parse_message(handler);
                begin_parse_message();
            }

            buffered_read_stream<AsyncReadStream> &lock_input()
            {
                assert(!locked);
                locked = true;
                return buffer;
            }

            void unlock_input()
            {
                assert(locked);
                locked = false;
                if (!waiting_for_response)
                {
                    return;
                }
                begin_parse_message();
            }

        private:
            buffered_read_stream<AsyncReadStream> buffer;
            bool locked;
            std::function<void()> begin_parse_message;
            std::function<void(boost::system::error_code, request_id)> waiting_for_request;
            std::function<void(boost::system::error_code, request_id)> waiting_for_response;
            bool parsing_header;

            template <class DummyHandler>
            struct parse_message_type_operation
            {
                message_splitter &pipeline;
                DummyHandler dummy;

                explicit parse_message_type_operation(message_splitter &pipeline, DummyHandler dummy)
                    : pipeline(pipeline)
                    , dummy(std::move(dummy))
                {
                }

                void operator()(boost::system::error_code ec, message_type_int const type)
                {
                    if (!!ec)
                    {
                        pipeline.on_error(ec);
                        return;
                    }
                    switch (static_cast<message_type>(type))
                    {
                    case message_type::response:
                        begin_parse_value(pipeline.buffer.input, boost::asio::buffer(pipeline.buffer.response_buffer),
                                          pipeline.buffer.response_buffer_used, integer_parser<request_id>(),
                                          parse_response_operation<DummyHandler>(pipeline, dummy));
                        break;

                    case message_type::request:
                        begin_parse_value(pipeline.buffer.input, boost::asio::buffer(pipeline.buffer.response_buffer),
                                          pipeline.buffer.response_buffer_used, integer_parser<request_id>(),
                                          parse_request_operation<DummyHandler>(pipeline, dummy));
                        break;

                    default:
                        pipeline.on_error(make_invalid_input_error());
                        break;
                    }
                }

                template <class Function>
                friend void asio_handler_invoke(Function &&f, parse_message_type_operation *operation)
                {
                    using boost::asio::asio_handler_invoke;
                    asio_handler_invoke(f, &operation->dummy);
                }
            };

            template <class DummyHandler>
            struct parse_request_operation
            {
                message_splitter &pipeline;
                DummyHandler dummy;

                explicit parse_request_operation(message_splitter &pipeline, DummyHandler dummy)
                    : pipeline(pipeline)
                    , dummy(std::move(dummy))
                {
                }

                void operator()(boost::system::error_code ec, request_id const request)
                {
                    if (!!ec)
                    {
                        pipeline.on_error(ec);
                        return;
                    }

                    assert(pipeline.parsing_header);
                    pipeline.parsing_header = false;

                    bool const continue_ = pipeline.waiting_for_response != nullptr;

                    assert(pipeline.waiting_for_request);
                    Si::exchange(pipeline.waiting_for_request, nullptr)(ec, request);

                    if (continue_ && !pipeline.parsing_header)
                    {
                        pipeline.begin_parse_message();
                    }
                }

                template <class Function>
                friend void asio_handler_invoke(Function &&f, parse_request_operation *operation)
                {
                    using boost::asio::asio_handler_invoke;
                    asio_handler_invoke(f, &operation->dummy);
                }
            };

            template <class DummyHandler>
            struct parse_response_operation
            {
                message_splitter &pipeline;
                DummyHandler dummy;<commit_msg>message_splitter parses also the method in a request<commit_after>#pragma once

#include <array>
#include <silicium/exchange.hpp>
#include <warpcoil/protocol.hpp>
#include <warpcoil/cpp/begin_parse_value.hpp>
#include <warpcoil/cpp/tuple_parser.hpp>
#include <warpcoil/cpp/integer_parser.hpp>
#include <warpcoil/cpp/utf8_parser.hpp>

namespace warpcoil
{
    namespace cpp
    {
        template <class AsyncReadStream>
        struct buffered_read_stream
        {
            AsyncReadStream &input;
            std::array<std::uint8_t, 512> response_buffer;
            std::size_t response_buffer_used;

            explicit buffered_read_stream(AsyncReadStream &input)
                : input(input)
                , response_buffer_used(0)
            {
            }
        };

        template <class AsyncReadStream>
        struct message_splitter
        {
            explicit message_splitter(AsyncReadStream &input)
                : buffer(input)
                , locked(false)
                , parsing_header(false)
            {
            }

            template <class ResponseHandler>
            void wait_for_response(ResponseHandler handler)
            {
                assert(!waiting_for_response);
                waiting_for_response = [handler](boost::system::error_code ec, request_id request) mutable
                {
                    using boost::asio::asio_handler_invoke;
                    asio_handler_invoke(
                        [&]
                        {
                            handler(ec, request);
                        },
                        &handler);
                };
                if (parsing_header)
                {
                    return;
                }
                set_begin_parse_message(handler);
                begin_parse_message();
            }

            template <class RequestHandler>
            void wait_for_request(RequestHandler handler)
            {
                assert(!waiting_for_request);
                waiting_for_request = [handler](boost::system::error_code ec, request_id id, std::string method) mutable
                {
                    using boost::asio::asio_handler_invoke;
                    asio_handler_invoke(
                        [&]
                        {
                            handler(ec, id, std::move(method));
                        },
                        &handler);
                };
                if (parsing_header)
                {
                    return;
                }
                set_begin_parse_message(handler);
                begin_parse_message();
            }

            buffered_read_stream<AsyncReadStream> &lock_input()
            {
                assert(!locked);
                locked = true;
                return buffer;
            }

            void unlock_input()
            {
                assert(locked);
                locked = false;
                if (!waiting_for_response)
                {
                    return;
                }
                begin_parse_message();
            }

        private:
            buffered_read_stream<AsyncReadStream> buffer;
            bool locked;
            std::function<void()> begin_parse_message;
            std::function<void(boost::system::error_code, request_id, std::string)> waiting_for_request;
            std::function<void(boost::system::error_code, request_id)> waiting_for_response;
            bool parsing_header;

            template <class DummyHandler>
            struct parse_message_type_operation
            {
                message_splitter &pipeline;
                DummyHandler dummy;

                explicit parse_message_type_operation(message_splitter &pipeline, DummyHandler dummy)
                    : pipeline(pipeline)
                    , dummy(std::move(dummy))
                {
                }

                void operator()(boost::system::error_code ec, message_type_int const type)
                {
                    if (!!ec)
                    {
                        pipeline.on_error(ec);
                        return;
                    }
                    switch (static_cast<message_type>(type))
                    {
                    case message_type::response:
                        begin_parse_value(pipeline.buffer.input, boost::asio::buffer(pipeline.buffer.response_buffer),
                                          pipeline.buffer.response_buffer_used, integer_parser<request_id>(),
                                          parse_response_operation<DummyHandler>(pipeline, dummy));
                        break;

                    case message_type::request:
                        begin_parse_value(
                            pipeline.buffer.input, boost::asio::buffer(pipeline.buffer.response_buffer),
                            pipeline.buffer.response_buffer_used,
                            tuple_parser<integer_parser<request_id>, utf8_parser<integer_parser<std::uint8_t>>>(),
                            parse_request_operation<DummyHandler>(pipeline, dummy));
                        break;

                    default:
                        pipeline.on_error(make_invalid_input_error());
                        break;
                    }
                }

                template <class Function>
                friend void asio_handler_invoke(Function &&f, parse_message_type_operation *operation)
                {
                    using boost::asio::asio_handler_invoke;
                    asio_handler_invoke(f, &operation->dummy);
                }
            };

            template <class DummyHandler>
            struct parse_request_operation
            {
                message_splitter &pipeline;
                DummyHandler dummy;

                explicit parse_request_operation(message_splitter &pipeline, DummyHandler dummy)
                    : pipeline(pipeline)
                    , dummy(std::move(dummy))
                {
                }

                void operator()(boost::system::error_code ec, std::tuple<request_id, std::string> request)
                {
                    if (!!ec)
                    {
                        pipeline.on_error(ec);
                        return;
                    }

                    assert(pipeline.parsing_header);
                    pipeline.parsing_header = false;

                    bool const continue_ = pipeline.waiting_for_response != nullptr;

                    assert(pipeline.waiting_for_request);
                    Si::exchange(pipeline.waiting_for_request, nullptr)(ec, std::get<0>(request),
                                                                        std::move(std::get<1>(request)));

                    if (continue_ && !pipeline.parsing_header)
                    {
                        pipeline.begin_parse_message();
                    }
                }

                template <class Function>
                friend void asio_handler_invoke(Function &&f, parse_request_operation *operation)
                {
                    using boost::asio::asio_handler_invoke;
                    asio_handler_invoke(f, &operation->dummy);
                }
            };

            template <class DummyHandler>
            struct parse_response_operation
            {
                message_splitter &pipeline;
                DummyHandler dummy;
{'fraction_non_alphanumeric': 0.07429318291256792, 'fraction_numerical': 0.0010318497626745546, 'mean_word_length': 1.2003935220220978, 'pattern_counts': {'":': 0, '<': 61, '<?xml version=': 0, '>': 65, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 15, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1345122', 'n_tokens_mistral': 3461, 'n_tokens_neox': 3196, 'n_words': 515}