Merge pull request #3812 from clemahieu/block_deserializer

Extracting nano::bootstrap::block_deserializer class

nano/core_test/bootstrap.cpp

@@ -1,3 +1,4 @@
+#include <nano/node/bootstrap/block_deserializer.hpp>
 #include <nano/node/bootstrap/bootstrap_frontier.hpp>
 #include <nano/node/bootstrap/bootstrap_lazy.hpp>
 #include <nano/test_common/system.hpp>
@@ -1440,15 +1441,14 @@ TEST (bootstrap_processor, wallet_lazy_pending)
 	node0->block_processor.add (send2);
 	node0->block_processor.flush ();
 	// Start wallet lazy bootstrap
-	auto node1 (std::make_shared<nano::node> (system.io_ctx, nano::get_available_port (), nano::unique_path (), system.logging, system.work));
+	auto node1 = system.add_node ();
 	node1->network.udp_channels.insert (node0->network.endpoint (), node1->network_params.network.protocol_version);
 	auto wallet (node1->wallets.create (nano::random_wallet_id ()));
 	ASSERT_NE (nullptr, wallet);
 	wallet->insert_adhoc (key2.prv);
 	node1->bootstrap_wallet ();
 	// Check processed blocks
 	ASSERT_TIMELY (10s, node1->ledger.block_or_pruned_exists (send2->hash ()));
-	node1->stop ();
 }
 
 TEST (bootstrap_processor, multiple_attempts)
@@ -2033,3 +2033,8 @@ TEST (bulk_pull_account, basics)
 		ASSERT_EQ (nullptr, block_data.second.get ());
 	}
 }
+
+TEST (block_deserializer, construction)
+{
+	auto deserializer = std::make_shared<nano::bootstrap::block_deserializer> ();
+}

nano/node/CMakeLists.txt

@@ -18,6 +18,8 @@ add_library(
   active_transactions.cpp
   blockprocessor.hpp
   blockprocessor.cpp
+  bootstrap/block_deserializer.hpp
+  bootstrap/block_deserializer.cpp
   bootstrap/bootstrap_attempt.hpp
   bootstrap/bootstrap_attempt.cpp
   bootstrap/bootstrap_bulk_pull.hpp

nano/node/bootstrap/block_deserializer.cpp

@@ -0,0 +1,65 @@
+#include <nano/lib/blocks.hpp>
+#include <nano/node/bootstrap/block_deserializer.hpp>
+#include <nano/node/socket.hpp>
+#include <nano/secure/buffer.hpp>
+
+nano::bootstrap::block_deserializer::block_deserializer () :
+	read_buffer{ std::make_shared<std::vector<uint8_t>> () }
+{
+}
+
+void nano::bootstrap::block_deserializer::read (nano::socket & socket, callback_type const && callback)
+{
+	debug_assert (callback);
+	read_buffer->resize (1);
+	socket.async_read (read_buffer, 1, [this_l = shared_from_this (), &socket, callback = std::move (callback)] (boost::system::error_code const & ec, std::size_t size_a) {
+		if (ec)
+		{
+			callback (ec, nullptr);
+			return;
+		}
+		if (size_a != 1)
+		{
+			callback (boost::asio::error::fault, nullptr);
+			return;
+		}
+		this_l->received_type (socket, std::move (callback));
+	});
+}
+
+void nano::bootstrap::block_deserializer::received_type (nano::socket & socket, callback_type const && callback)
+{
+	nano::block_type type = static_cast<nano::block_type> (read_buffer->data ()[0]);
+	if (type == nano::block_type::not_a_block)
+	{
+		callback (boost::system::error_code{}, nullptr);
+		return;
+	}
+	auto size = nano::block::size (type);
+	if (size == 0)
+	{
+		callback (boost::asio::error::fault, nullptr);
+		return;
+	}
+	read_buffer->resize (size);
+	socket.async_read (read_buffer, size, [this_l = shared_from_this (), size, type, callback = std::move (callback)] (boost::system::error_code const & ec, std::size_t size_a) {
+		if (ec)
+		{
+			callback (ec, nullptr);
+			return;
+		}
+		if (size_a != size)
+		{
+			callback (boost::asio::error::fault, nullptr);
+			return;
+		}
+		this_l->received_block (type, std::move (callback));
+	});
+}
+
+void nano::bootstrap::block_deserializer::received_block (nano::block_type type, callback_type const && callback)
+{
+	nano::bufferstream stream{ read_buffer->data (), read_buffer->size () };
+	auto block = nano::deserialize_block (stream, type);
+	callback (boost::system::error_code{}, block);
+}

nano/node/bootstrap/block_deserializer.hpp

@@ -0,0 +1,47 @@
+#pragma once
+
+#include <nano/lib/blocks.hpp>
+
+#include <boost/system/error_code.hpp>
+
+#include <memory>
+#include <vector>
+
+namespace nano
+{
+class block;
+class socket;
+namespace bootstrap
+{
+	/**
+	 * Class to read a block-type byte followed by a serialised block from a stream.
+	 * It is typically used to used to read a series of block-types and blocks terminated by a not-a-block type.
+	 */
+	class block_deserializer : public std::enable_shared_from_this<nano::bootstrap::block_deserializer>
+	{
+	public:
+		using callback_type = std::function<void (boost::system::error_code, std::shared_ptr<nano::block>)>;
+
+		block_deserializer ();
+		/**
+		 * Read a type-prefixed block from 'socket' and pass the result, or an error, to 'callback'
+		 * A normal end to series of blocks is a marked by return no error and a nullptr for block.
+		 */
+		void read (nano::socket & socket, callback_type const && callback);
+
+	private:
+		/**
+		 * Called by read method on receipt of a block type byte.
+		 * The type byte will be in the read_buffer.
+		 */
+		void received_type (nano::socket & socket, callback_type const && callback);
+
+		/**
+		 * Called by received_type when a block is received, it parses the block and calls the callback.
+		 */
+		void received_block (nano::block_type type, callback_type const && callback);
+
+		std::shared_ptr<std::vector<uint8_t>> read_buffer;
+	};
+}
+}