<commit_before>#include "clone.hpp"
#include "http_file_service.hpp"
#include <server/directory_listing.hpp>
#include <silicium/virtualized_source.hpp>
#include <silicium/observable_source.hpp>
#include <silicium/received_from_socket_source.hpp>
#include <silicium/ref.hpp>
#include <silicium/coroutine.hpp>
#include <silicium/total_consumer.hpp>
#include <silicium/open.hpp>
#include <boost/filesystem/operations.hpp>

namespace fileserver
{
	namespace
	{
		boost::system::error_code clone_recursively(file_service &service, unknown_digest const &tree_digest, boost::filesystem::path const &destination, Si::yield_context yield)
		{
			{
				boost::system::error_code ec;
				boost::filesystem::create_directories(destination, ec);
				if (ec)
				{
					return ec;
				}
			}
			auto tree_file_opening = service.open(tree_digest);
			boost::optional<Si::error_or<linear_file>> maybe_tree_file = yield.get_one(tree_file_opening);
			if (maybe_tree_file->is_error())
			{
				return *maybe_tree_file->error();
			}
			linear_file tree_file = std::move(*maybe_tree_file).get();
			auto receiving_source = Si::virtualize_source(Si::make_observable_source(Si::ref(tree_file.content), yield));
			Si::received_from_socket_source content_source(receiving_source);
			auto parsed = deserialize_json(std::move(content_source));
			return Si::visit<boost::system::error_code>(
				parsed,
				[&](std::unique_ptr<directory_listing> const &listing) -> boost::system::error_code
			{
				for (auto const &entry : listing->entries)
				{
					if (entry.second.type == "blob")
					{
						auto maybe_file = Si::create_file(destination / entry.first);
						if (maybe_file.error())
						{
							return *maybe_file.error();
						}
						//TODO write file
					}
					else if (entry.second.type == "json_v1")
					{
						auto ec = clone_recursively(service, to_unknown_digest(entry.second.referenced), destination / entry.first, yield);
						if (ec)
						{
							return ec;
						}
					}
					else
					{
						throw std::logic_error("unknown directory entry type"); //TODO
					}
				}
				return boost::system::error_code();
			},
				[](std::size_t) -> boost::system::error_code
			{
				throw std::logic_error("todo");
			});
		}
	}

	void clone_directory(unknown_digest const &root_digest, boost::filesystem::path const &destination, boost::asio::ip::tcp::endpoint const &server)
	{
		boost::asio::io_service io;
		http_file_service service(io, server);
		auto all = Si::make_total_consumer(Si::make_coroutine<Si::nothing>([&](Si::push_context<Si::nothing> yield)
		{
			auto ec = clone_recursively(service, root_digest, destination, yield);
			if (ec)
			{
				Si::detail::throw_system_error(ec);
			}
		}));
		all.start();
		io.run();
	}
}<commit_msg>write file contents when cloning<commit_after>#include "clone.hpp"
#include "http_file_service.hpp"
#include <server/directory_listing.hpp>
#include <silicium/virtualized_source.hpp>
#include <silicium/observable_source.hpp>
#include <silicium/received_from_socket_source.hpp>
#include <silicium/ref.hpp>
#include <silicium/coroutine.hpp>
#include <silicium/total_consumer.hpp>
#include <silicium/open.hpp>
#include <boost/filesystem/operations.hpp>
#include <boost/asio/posix/stream_descriptor.hpp>
#include <boost/asio/write.hpp>

namespace Si
{
	template <class AsyncStream>
	struct reading_observable
	{
		typedef error_or<incoming_bytes> element_type;

		explicit reading_observable(AsyncStream &stream, boost::iterator_range<char *> buffer)
			: stream(&stream)
		{
			assert(buffer.size() >= 1);
		}

		void async_get_one(observer<element_type> &receiver)
		{
			stream->async_read_some(
				boost::asio::buffer(buffer.begin(), buffer.size()),
				[this, &receiver](boost::system::error_code ec, std::size_t bytes_received)
			{
				if (ec)
				{
					receiver.got_element(ec);
				}
				else
				{
					assert(bytes_received <= buffer.size());
					receiver.got_element(incoming_bytes(buffer.begin(), buffer.begin() + bytes_received));
				}
			});
		}

	private:

		AsyncStream *stream;
		boost::iterator_range<char *> buffer;
	};

	template <class AsyncStream>
	auto make_reading_observable(AsyncStream &stream, boost::iterator_range<char *> buffer)
	{
		return reading_observable<AsyncStream>(stream, buffer);
	}

	template <class AsyncStream>
	struct writing_observable
	{
		typedef boost::system::error_code element_type;

		explicit writing_observable(AsyncStream &stream, boost::iterator_range<char const *> buffer)
			: stream(&stream)
		{
			assert(buffer.size() >= 1);
		}

		void async_get_one(observer<element_type> &receiver)
		{
			boost::asio::async_write(
				*stream,
				boost::asio::buffer(buffer.begin(), buffer.size()),
				[this, &receiver](boost::system::error_code ec, std::size_t bytes_received)
			{
				assert(ec || (bytes_received == buffer.size()));
				receiver.got_element(ec);
			});
		}

	private:

		AsyncStream *stream;
		boost::iterator_range<char const *> buffer;
	};

	template <class AsyncStream>
	auto make_writing_observable(AsyncStream &stream, boost::iterator_range<char const *> buffer)
	{
		return writing_observable<AsyncStream>(stream, buffer);
	}

	inline boost::asio::posix::stream_descriptor stream_from_file(boost::asio::io_service &io, file_descriptor file)
	{
		return boost::asio::posix::stream_descriptor(io, file.release());
	}

	boost::system::error_code write_all(native_file_handle destination, boost::iterator_range<char const *> buffer)
	{
		std::size_t total_written = 0;
		while (total_written < buffer.size())
		{
			ssize_t rc = write(destination, buffer.begin() + total_written, buffer.size() - total_written);
			if (rc < 0)
			{
				return boost::system::error_code(errno, boost::system::native_ecat);
			}
			total_written += static_cast<size_t>(rc);
		}
		return boost::system::error_code();
	}
}

namespace fileserver
{
	namespace
	{
		boost::system::error_code clone_recursively(file_service &service, unknown_digest const &tree_digest, boost::filesystem::path const &destination, Si::yield_context yield, boost::asio::io_service &io)
		{
			{
				boost::system::error_code ec;
				boost::filesystem::create_directories(destination, ec);
				if (ec)
				{
					return ec;
				}
			}
			auto tree_file_opening = service.open(tree_digest);
			boost::optional<Si::error_or<linear_file>> maybe_tree_file = yield.get_one(tree_file_opening);
			if (maybe_tree_file->is_error())
			{
				return *maybe_tree_file->error();
			}
			linear_file tree_file = std::move(*maybe_tree_file).get();
			auto receiving_source = Si::virtualize_source(Si::make_observable_source(Si::ref(tree_file.content), yield));
			Si::received_from_socket_source content_source(receiving_source);
			auto parsed = deserialize_json(std::move(content_source));
			return Si::visit<boost::system::error_code>(
				parsed,
				[&](std::unique_ptr<directory_listing> const &listing) -> boost::system::error_code
			{
				for (auto const &entry : listing->entries)
				{
					if (entry.second.type == "blob")
					{
						auto opening_remote = service.open(to_unknown_digest(entry.second.referenced));
						auto maybe_remote_file = *yield.get_one(opening_remote);
						if (maybe_remote_file.error())
						{
							return *maybe_remote_file.error();
						}
						linear_file remote_file = std::move(maybe_remote_file).get();
						auto maybe_local_file = Si::create_file(destination / entry.first);
						if (maybe_local_file.error())
						{
							return *maybe_local_file.error();
						}
						auto local_file = std::move(maybe_local_file).get();
						file_offset total_written = 0;
						while (total_written < remote_file.size)
						{
							Si::error_or<Si::incoming_bytes> const received = *yield.get_one(remote_file.content);
							if (received.error())
							{
								return *received.error();
							}
							if (received->size() == 0)
							{
								break;
							}
							if (static_cast<file_offset>(received->size()) > (remote_file.size - total_written))
							{
								throw std::logic_error("todo received too much");
							}
							boost::system::error_code const written = Si::write_all(local_file.handle, boost::make_iterator_range(received->begin, received->end));
							if (written)
							{
								return written;
							}
							total_written += received->size();
						}
					}
					else if (entry.second.type == "json_v1")
					{
						auto ec = clone_recursively(service, to_unknown_digest(entry.second.referenced), destination / entry.first, yield, io);
						if (ec)
						{
							return ec;
						}
					}
					else
					{
						throw std::logic_error("unknown directory entry type"); //TODO
					}
				}
				return boost::system::error_code();
			},
				[](std::size_t) -> boost::system::error_code
			{
				throw std::logic_error("todo");
			});
		}
	}

	void clone_directory(unknown_digest const &root_digest, boost::filesystem::path const &destination, boost::asio::ip::tcp::endpoint const &server)
	{
		boost::asio::io_service io;
		http_file_service service(io, server);
		auto all = Si::make_total_consumer(Si::make_coroutine<Si::nothing>([&](Si::push_context<Si::nothing> yield)
		{
			auto ec = clone_recursively(service, root_digest, destination, yield, io);
			if (ec)
			{
				Si::detail::throw_system_error(ec);
			}
		}));
		all.start();
		io.run();
	}
}
{'fraction_non_alphanumeric': 0.14043193295369077, 'fraction_numerical': 0.0008595680670463093, 'mean_word_length': 4.226277372262774, 'pattern_counts': {'":': 0, '<': 58, '<?xml version=': 0, '>': 73, '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': '1315344', 'n_tokens_mistral': 3818, 'n_tokens_neox': 3035, 'n_words': 507}