From 1d8fc9e57447b917f165eefc24a2d29b4b5e9d76 Mon Sep 17 00:00:00 2001
From: Thiago Silva <82097354+thsfs@users.noreply.github.com>
Date: Thu, 9 Mar 2023 11:54:47 -0300
Subject: [PATCH] Add the block_publisher class (#4178)

* Add the block_publisher class

* Move the block publishing to the block_publisher class

* Remove unnecessary header
---
 nano/node/CMakeLists.txt      |  2 ++
 nano/node/block_publisher.cpp | 27 +++++++++++++++++++++++++++
 nano/node/block_publisher.hpp | 24 ++++++++++++++++++++++++
 nano/node/blockprocessor.cpp  |  2 --
 nano/node/node.cpp            |  4 +++-
 nano/node/node.hpp            |  2 ++
 6 files changed, 58 insertions(+), 3 deletions(-)
 create mode 100644 nano/node/block_publisher.cpp
 create mode 100644 nano/node/block_publisher.hpp

diff --git a/nano/node/CMakeLists.txt b/nano/node/CMakeLists.txt
index bf20c1ec3b..be0af8f539 100644
--- a/nano/node/CMakeLists.txt
+++ b/nano/node/CMakeLists.txt
@@ -24,6 +24,8 @@ add_library(
   block_arrival.cpp
   block_broadcast.cpp
   block_broadcast.hpp
+  block_publisher.cpp
+  block_publisher.hpp
   blocking_observer.cpp
   blocking_observer.hpp
   blockprocessor.hpp
diff --git a/nano/node/block_publisher.cpp b/nano/node/block_publisher.cpp
new file mode 100644
index 0000000000..1cc77e5edb
--- /dev/null
+++ b/nano/node/block_publisher.cpp
@@ -0,0 +1,27 @@
+#include <nano/node/active_transactions.hpp>
+#include <nano/node/block_publisher.hpp>
+#include <nano/node/blockprocessor.hpp>
+
+nano::block_publisher::block_publisher (nano::active_transactions & active) :
+	active{ active }
+{
+}
+
+void nano::block_publisher::connect (nano::block_processor & block_processor)
+{
+	block_processor.processed.add ([this] (auto const & result, auto const & block) {
+		switch (result.code)
+		{
+			case nano::process_result::fork:
+				observe (block);
+				break;
+			default:
+				break;
+		}
+	});
+}
+
+void nano::block_publisher::observe (std::shared_ptr<nano::block> block)
+{
+	active.publish (block);
+}
diff --git a/nano/node/block_publisher.hpp b/nano/node/block_publisher.hpp
new file mode 100644
index 0000000000..29a2df906b
--- /dev/null
+++ b/nano/node/block_publisher.hpp
@@ -0,0 +1,24 @@
+#pragma once
+
+#include <memory>
+
+namespace nano
+{
+class active_transactions;
+class block_processor;
+class block;
+
+// This class tracks processed blocks to be published.
+class block_publisher
+{
+public:
+	block_publisher (nano::active_transactions & active);
+	void connect (nano::block_processor & block_processor);
+
+private:
+	// Block_processor observer
+	void observe (std::shared_ptr<nano::block> block);
+
+	nano::active_transactions & active;
+};
+}
diff --git a/nano/node/blockprocessor.cpp b/nano/node/blockprocessor.cpp
index 1fa45f7291..1377117eff 100644
--- a/nano/node/blockprocessor.cpp
+++ b/nano/node/blockprocessor.cpp
@@ -1,7 +1,6 @@
 #include <nano/lib/threading.hpp>
 #include <nano/lib/timer.hpp>
 #include <nano/node/blockprocessor.hpp>
-#include <nano/node/election.hpp>
 #include <nano/node/node.hpp>
 #include <nano/node/websocket.hpp>
 #include <nano/secure/store.hpp>
@@ -435,7 +434,6 @@ nano::process_return nano::block_processor::process_one (nano::write_transaction
 		case nano::process_result::fork:
 		{
 			node.stats.inc (nano::stat::type::ledger, nano::stat::detail::fork);
-			events_a.events.emplace_back ([this, block] (nano::transaction const &) { this->node.active.publish (block); });
 			if (node.config.logging.ledger_logging ())
 			{
 				node.logger.try_log (boost::str (boost::format ("Fork for: %1% root: %2%") % hash.to_string () % block->root ().to_string ()));
diff --git a/nano/node/node.cpp b/nano/node/node.cpp
index 1fbcf8807f..da0e94d312 100644
--- a/nano/node/node.cpp
+++ b/nano/node/node.cpp
@@ -205,9 +205,11 @@ nano::node::node (boost::asio::io_context & io_ctx_a, boost::filesystem::path co
 	epoch_upgrader{ *this, ledger, store, network_params, logger },
 	startup_time (std::chrono::steady_clock::now ()),
 	node_seq (seq),
-	block_broadcast{ network, block_arrival, !flags.disable_block_processor_republishing }
+	block_broadcast{ network, block_arrival, !flags.disable_block_processor_republishing },
+	block_publisher{ active }
 {
 	block_broadcast.connect (block_processor);
+	block_publisher.connect (block_processor);
 	unchecked.use_memory = [this] () { return ledger.bootstrap_weight_reached (); };
 	unchecked.satisfied = [this] (nano::unchecked_info const & info) {
 		this->block_processor.add (info.block);
diff --git a/nano/node/node.hpp b/nano/node/node.hpp
index 724b072d35..ab65065f4b 100644
--- a/nano/node/node.hpp
+++ b/nano/node/node.hpp
@@ -8,6 +8,7 @@
 #include <nano/node/bandwidth_limiter.hpp>
 #include <nano/node/block_arrival.hpp>
 #include <nano/node/block_broadcast.hpp>
+#include <nano/node/block_publisher.hpp>
 #include <nano/node/blockprocessor.hpp>
 #include <nano/node/bootstrap/bootstrap.hpp>
 #include <nano/node/bootstrap/bootstrap_attempt.hpp>
@@ -192,6 +193,7 @@ class node final : public std::enable_shared_from_this<nano::node>
 	nano::websocket_server websocket;
 	nano::epoch_upgrader epoch_upgrader;
 	nano::block_broadcast block_broadcast;
+	nano::block_publisher block_publisher;
 
 	std::chrono::steady_clock::time_point const startup_time;
 	std::chrono::seconds unchecked_cutoff = std::chrono::seconds (7 * 24 * 60 * 60); // Week