<commit_before>#ifndef SILICIUM_REACTIVE_SOCKET_OBSERVABLE_HPP
#define SILICIUM_REACTIVE_SOCKET_OBSERVABLE_HPP

#include <silicium/fast_variant.hpp>
#include <silicium/override.hpp>
#include <silicium/observable.hpp>
#include <silicium/exchange.hpp>
#include <boost/config.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/range/iterator_range.hpp>

namespace Si
{
	struct incoming_bytes
	{
		char const *begin = nullptr, *end = nullptr;

		incoming_bytes() BOOST_NOEXCEPT
		{
		}

		incoming_bytes(char const *begin, char const *end) BOOST_NOEXCEPT
			: begin(begin)
			, end(end)
		{
		}
	};

	typedef Si::fast_variant<incoming_bytes, boost::system::error_code> received_from_socket;

	struct socket_observable : observable<received_from_socket>
	{
		typedef received_from_socket element_type;
		typedef boost::iterator_range<char *> buffer_type;

		socket_observable(boost::asio::ip::tcp::socket &socket, buffer_type buffer)
			: socket(&socket)
			, buffer(buffer)
		{
			assert(!buffer.empty());
		}

		virtual void async_get_one(observer<element_type> &receiver) SILICIUM_OVERRIDE
		{
			assert(!receiver_);
			receiver_ = &receiver;
			socket->async_receive(boost::asio::buffer(buffer.begin(), buffer.size()), [this](boost::system::error_code error, std::size_t bytes_received)
			{
				assert(receiver_);
				if (error)
				{
					if (error == boost::asio::error::operation_aborted)
					{
						return;
					}
					exchange(this->receiver_, nullptr)->got_element(error);
				}
				else
				{
					auto * const receiver = exchange(this->receiver_, nullptr);
					receiver->got_element(incoming_bytes{buffer.begin(), buffer.begin() + bytes_received});
				}
			});
		}

		virtual void cancel() SILICIUM_OVERRIDE
		{
			assert(receiver_);<commit_msg>devirtualize socket_observable<commit_after>#ifndef SILICIUM_REACTIVE_SOCKET_OBSERVABLE_HPP
#define SILICIUM_REACTIVE_SOCKET_OBSERVABLE_HPP

#include <silicium/fast_variant.hpp>
#include <silicium/override.hpp>
#include <silicium/observer.hpp>
#include <silicium/exchange.hpp>
#include <boost/config.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/range/iterator_range.hpp>

namespace Si
{
	struct incoming_bytes
	{
		char const *begin = nullptr, *end = nullptr;

		incoming_bytes() BOOST_NOEXCEPT
		{
		}

		incoming_bytes(char const *begin, char const *end) BOOST_NOEXCEPT
			: begin(begin)
			, end(end)
		{
		}
	};

	typedef Si::fast_variant<incoming_bytes, boost::system::error_code> received_from_socket;

	struct socket_observable
	{
		typedef received_from_socket element_type;
		typedef boost::iterator_range<char *> buffer_type;

		socket_observable(boost::asio::ip::tcp::socket &socket, buffer_type buffer)
			: socket(&socket)
			, buffer(buffer)
		{
			assert(!buffer.empty());
		}

		void async_get_one(observer<element_type> &receiver)
		{
			assert(!receiver_);
			receiver_ = &receiver;
			socket->async_receive(boost::asio::buffer(buffer.begin(), buffer.size()), [this](boost::system::error_code error, std::size_t bytes_received)
			{
				assert(receiver_);
				if (error)
				{
					if (error == boost::asio::error::operation_aborted)
					{
						return;
					}
					exchange(this->receiver_, nullptr)->got_element(error);
				}
				else
				{
					auto * const receiver = exchange(this->receiver_, nullptr);
					receiver->got_element(incoming_bytes{buffer.begin(), buffer.begin() + bytes_received});
				}
			});
		}

		void cancel()
		{
			assert(receiver_);
{'fraction_non_alphanumeric': 0.13022041763341066, 'fraction_numerical': 0.0, 'mean_word_length': 4.748333333333333, 'pattern_counts': {'":': 0, '<': 24, '<?xml version=': 0, '>': 34, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5371348', 'n_tokens_mistral': 1412, 'n_tokens_neox': 1174, 'n_words': 187}