text:
#include <nano/core_test/fakes/websocket_client.hpp>
#include <nano/node/websocket.hpp>
#include <nano/test_common/system.hpp>
#include <nano/test_common/telemetry.hpp>
#include <nano/test_common/testutil.hpp>

#include <gtest/gtest.h>

#include <boost/property_tree/json_parser.hpp>

#include <chrono>
#include <cstdlib>
#include <memory>
#include <sstream>
#include <string>
#include <vector>

using namespace std::chrono_literals;

// Tests clients subscribing multiple times or unsubscribing without a subscription
TEST (websocket, subscription_edge)
{
	nano::system system;
	nano::node_config config (nano::get_available_port (), system.logging);
	config.websocket_config.enabled = true;
	config.websocket_config.port = nano::get_available_port ();
	auto node1 (system.add_node (config));

	ASSERT_EQ (0, node1->websocket_server->subscriber_count (nano::websocket::topic::confirmation));

	auto task = ([config, &node1] () {
		fake_websocket_client client (node1->websocket_server->listening_port ());
		client.send_message (R"json({"action": "subscribe", "topic": "confirmation", "ack": true})json");
		client.await_ack ();
		EXPECT_EQ (1, node1->websocket_server->subscriber_count (nano::websocket::topic::confirmation));
		client.send_message (R"json({"action": "subscribe", "topic": "confirmation", "ack": true})json");
		client.await_ack ();
		EXPECT_EQ (1, node1->websocket_server->subscriber_count (nano::websocket::topic::confirmation));
		client.send_message (R"json({"action": "unsubscribe", "topic": "confirmation", "ack": true})json");
		client.await_ack ();
		EXPECT_EQ (0, node1->websocket_server->subscriber_count (nano::websocket::topic::confirmation));
		client.send_message (R"json({"action": "unsubscribe", "topic": "confirmation", "ack": true})json");
		client.await_ack ();
		EXPECT_EQ (0, node1->websocket_server->subscriber_count (nano::websocket::topic::confirmation));
	});
	auto future = std::async (std::launch::async, task);

	ASSERT_TIMELY (5s, future.wait_for (0s) == std::future_status::ready);
}

// Subscribes to block confirmations, confirms a block and then awaits websocket notification
TEST (websocket, confirmation)
{
	nano::system system;
	nano::node_config config (nano::get_available_port (), system.logging);
	config.websocket_config.enabled = true;
	config.websocket_config.port = nano::get_available_port ();
	auto node1 (system.add_node (config));

	std::atomic<bool> ack_ready{ false };
	std::atomic<bool> unsubscribed{ false };
	auto task = ([&ack_ready, &unsubscribed, config, &node1] () {
		fake_websocket_client client (node1->websocket_server->listening_port ());
		client.send_message (R"json({"action": "subscribe", "topic": "confirmation", "ack": true})json");
		client.await_ack ();
		ack_ready = true;
		EXPECT_EQ (1, node1->websocket_server->subscriber_count (nano::websocket::topic::confirmation));
		auto response = client.get_response ();
		EXPECT_TRUE (response);
		boost::property_tree::ptree event;
		std::stringstream stream;
		stream << response.get ();
		boost::property_tree::read_json (stream, event);
		EXPECT_EQ (event.get<std::string> ("topic"), "confirmation");
		client.send_message (R"json({"action": "unsubscribe", "topic": "confirmation", "ack": true})json");
		client.await_ack ();
		unsubscribed = true;
		EXPECT_FALSE (client.get_response (1s));
	});
	auto future = std::async (std::launch::async, task);

	ASSERT_TIMELY (5s, ack_ready);

	nano::keypair key;
	system.wallet (0)->insert_adhoc (nano::dev::genesis_key.prv);
	auto balance = nano::dev::constants.genesis_amount;
	auto send_amount = node1->online_reps.delta () + 1;
	// Quick-confirm a block, legacy blocks should work without filtering
	{
		nano::block_hash previous (node1->latest (nano::dev::genesis_key.pub));
		balance -= send_amount;
		auto send (std::make_shared<nano::send_block> (previous, key.pub, balance, nano::dev::genesis_key.prv, nano::dev::genesis_key.pub, *system.work.generate (previous)));
		node1->process_active (send);
	}

	ASSERT_TIMELY (5s, unsubscribed);

	// Quick confirm a state block
	{
		nano::state_block_builder builder;
		nano::block_hash previous (node1->latest (nano::dev::genesis_key.pub));
		balance -= send_amount;
		auto send = builder
					.account (nano::dev::genesis_key.pub)
					.previous (previous)
					.representative (nano::dev::genesis_key.pub)
					.balance (balance)
					.link (key.pub)
					.sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
					.work (*system.work.generate (previous))
					.build_shared ();

		node1->process_active (send);
	}

	ASSERT_TIMELY (5s, future.wait_for (0s) == std::future_status::ready);
}

// Tests getting notification of an erased election
TEST (websocket, stopped_election)
{
	nano::system system;
	nano::node_config config (nano::get_available_port (), system.logging);
	config.websocket_config.enabled = true;
	config.websocket_config.port = nano::get_available_port ();
	auto node1 (system.add_node (config));

	std::atomic<bool> ack_ready{ false };
	auto task = ([&ack_ready, config, &node1] () {
		fake_websocket_client client (node1->websocket_server->listening_port ());
		client.send_message (R"json({"action": "subscribe", "topic": "stopped_election", "ack": "true"})json");
		client.await_ack ();
		ack_ready = true;
		EXPECT_EQ (1, node1->websocket_server->subscriber_count (nano::websocket::topic::stopped_election));
		return client.get_response ();
	});
	auto future = std::async (std::launch::async, task);

	ASSERT_TIMELY (5s, ack_ready);

	// Create election, then erase it, causing a websocket message to be emitted
	nano::keypair key1;
	auto send1 (std::make_shared<nano::send_block> (nano::dev::genesis->hash (), key1.pub, 0, nano::dev::genesis_key.prv, nano::dev::genesis_key.pub, *system.work.generate (nano::dev::genesis->hash ())));
	nano::publish publish1{ nano::dev::network_params.network, send1 };
	auto channel1 (node1->network.udp_channels.create (node1->network.endpoint ()));
	node1->network.inbound (publish1, channel1);
	node1->block_processor.flush ();
	ASSERT_TIMELY (1s, node1->active.election (send1->qualified_root ()));
	node1->active.erase (*send1);

	ASSERT_TIMELY (5s, future.wait_for (0s) == std::future_status::ready);

	auto response = future.get ();
	ASSERT_TRUE (response);
	boost::property_tree::ptree event;
	std::stringstream stream;
	stream << response.get ();
	boost::property_tree::read_json (stream, event);
	ASSERT_EQ (event.get<std::string> ("topic"), "stopped_election");
}

// Tests the filtering options of block confirmations
TEST (websocket, confirmation_options)
{
	nano::system system;
	nano::node_config config (nano::get_available_port (), system.logging);
	config.websocket_config.enabled = true;
	config.websocket_config.port = nano::get_available_port ();
	auto node1 (system.add_node (config));

	std::atomic<bool> ack_ready{ false };
	auto task1 = ([&ack_ready, config, &node1] () {
		fake_websocket_client client (node1->websocket_server->listening_port ());
		client.send_message (R"json({"action": "subscribe", "topic": "confirmation", "ack": "true", "options": {"confirmation_type": "active_quorum", "accounts": ["xrb_invalid"]}})json");
		client.await_ack ();
		ack_ready = true;
		EXPECT_EQ (1, node1->websocket_server->subscriber_count (nano::websocket::topic::confirmation));
		auto response = client.get_response (1s);
		EXPECT_FALSE (response);
	});
	auto future1 = std::async (std::launch::async, task1);

	ASSERT_TIMELY (5s, ack_ready);

	// Confirm a state block for an in-wallet account
	system.wallet (0)->insert_adhoc (nano::dev::genesis_key.prv);
	nano::keypair key;
	auto balance = nano::dev::constants.genesis_amount;
	auto send_amount = node1->online_reps.delta () + 1;
	nano::block_hash previous (node1->latest (nano::dev::genesis_key.pub));
	{
		balance -= send_amount;
		nano::state_block_builder builder;
		auto send = builder
					.account (nano::dev::genesis_key.pub)
					.previous (previous)
					.representative (nano::dev::genesis_key.pub)
					.balance (balance)
					.link (key.pub)
					.sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
					.work (*system.work.generate (previous))
					.build_shared ();

		node1->process_active (send);
		previous = send->hash ();
	}

	ASSERT_TIMELY (5s, future1.wait_for (0s) == std::future_status::ready);

	ack_ready = false;
	auto task2 = ([&ack_ready, config, &node1] () {
		fake_websocket_client client (node1->websocket_server->listening_port ());
		client.send_message (R"json({"action": "subscribe", "topic": "confirmation", "ack": "true", "options": {"confirmation_type": "active_quorum", "all_local_accounts": "true", "include_election_info": "true"}})json");
		client.await_ack ();
		ack_ready = true;
		EXPECT_EQ (1, node1->websocket_server->subscriber_count (nano::websocket::topic::confirmation));
		return client.get_response ();
	});
	auto future2 = std::async (std::launch::async, task2);

	ASSERT_TIMELY (10s, ack_ready);

	// Quick-confirm another block
	{
		balance -= send_amount;
		nano::state_block_builder builder;
		auto send = builder
					.account (nano::dev::genesis_key.pub)
					.previous (previous)
					.representative (nano::dev::genesis_key.pub)
					.balance (balance)
					.link (key.pub)
					.sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
					.work (*system.work.generate (previous))
					.build_shared ();

		node1->process_active (send);
		previous = send->hash ();
	}

	ASSERT_TIMELY (5s, future2.wait_for (0s) == std::future_status::ready);

	auto response2 = future2.get ();
	ASSERT_TRUE (response2);
	boost::property_tree::ptree event;
	std::stringstream stream;
	stream << response2.get ();
	boost::property_tree::read_json (stream, event);
	ASSERT_EQ (event.get<std::string> ("topic"), "confirmation");
	try
	{
		boost::property_tree::ptree election_info = event.get_child ("message.election_info");
		auto tally (election_info.get<std::string> ("tally"));
		auto final_tally (election_info.get<std::string> ("final"));
		auto time (election_info.get<std::string> ("time"));
		// Duration and request count may be zero on devnet, so we only check that they're present
		ASSERT_EQ (1, election_info.count ("duration"));
		ASSERT_EQ (1, election_info.count ("request_count"));
		ASSERT_EQ (1, election_info.count ("voters"));
		ASSERT_GE (1U, election_info.get<unsigned> ("blocks"));
		// Make sure tally and time are non-zero.
		ASSERT_NE ("0", tally);
		ASSERT_NE ("0", time);
		auto votes_l (election_info.get_child_optional ("votes"));
		ASSERT_FALSE (votes_l.is_initialized ());
	}
	catch (std::runtime_error const & ex)
	{
		FAIL () << ex.what ();
	}

	ack_ready = false;
	auto task3 = ([&ack_ready, config, &node1] () {
		fake_websocket_client client (node1->websocket_server->listening_port ());
		client.send_message (R"json({"action": "subscribe", "topic": "confirmation", "ack": "true", "options": {"confirmation_type": "active_quorum", "all_local_accounts": "true"}})json");
		client.await_ack ();
		ack_ready = true;
		EXPECT_EQ (1, node1->websocket_server->subscriber_count (nano::websocket::topic::confirmation));
		auto response = client.get_response (1s);
		EXPECT_FALSE (response);
	});
	auto future3 = std::async (std::launch::async, task3);

	ASSERT_TIMELY (5s, ack_ready);

	// Confirm a legacy block
	// When filtering options are enabled, legacy blocks are always filtered
	{
		balance -= send_amount;
		auto send (std::make_shared<nano::send_block> (previous, key.pub, balance, nano::dev::genesis_key.prv, nano::dev::genesis_key.pub, *system.work.generate (previous)));
		node1->process_active (send);
		previous = send->hash ();
	}

	ASSERT_TIMELY (5s, future3.wait_for (0s) == std::future_status::ready);
}

TEST (websocket, confirmation_options_votes)
{
	nano::system system;
	nano::node_config config (nano::get_available_port (), system.logging);
	config.websocket_config.enabled = true;
	config.websocket_config.port = nano::get_available_port ();
	auto node1 (system.add_node (config));

	std::atomic<bool> ack_ready{ false };
	auto task1 = ([&ack_ready, config, &node1] () {
		fake_websocket_client client (node1->websocket_server->listening_port ());
		client.send_message (R"json({"action": "subscribe", "topic": "confirmation", "ack": "true", "options": {"confirmation_type": "active_quorum", "include_election_info_with_votes": "true", "include_block": "false"}})json");
		client.await_ack ();
		ack_ready = true;
		EXPECT_EQ (1, node1->websocket_server->subscriber_count (nano::websocket::topic::confirmation));
		return client.get_response ();
	});
	auto future1 = std::async (std::launch::async, task1);

	ASSERT_TIMELY (10s, ack_ready);

	// Confirm a state block for an in-wallet account
	system.wallet (0)->insert_adhoc (nano::dev::genesis_key.prv);
	nano::keypair key;
	auto balance = nano::dev::constants.genesis_amount;
	auto send_amount = node1->config.online_weight_minimum.number () + 1;
	nano::block_hash previous (node1->latest (nano::dev::genesis_key.pub));
	{
		nano::state_block_builder builder;
		balance -= send_amount;
		auto send = builder
					.account (nano::dev::genesis_key.pub)
					.previous (previous)
					.representative (nano::dev::genesis_key.pub)
					.balance (balance)
					.link (key.pub)
					.sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
					.work (*system.work.generate (previous))
					.build_shared ();

		node1->process_active (send);
		previous = send->hash ();
	}

	ASSERT_TIMELY (5s, future1.wait_for (0s) == std::future_status::ready);

	auto response1 = future1.get ();
	ASSERT_TRUE (response1);
	boost::property_tree::ptree event;
	std::stringstream stream;
	stream << response1.get ();
	boost::property_tree::read_json (stream, event);
	ASSERT_EQ (event.get<std::string> ("topic"), "confirmation");
	try
	{
		boost::property_tree::ptree election_info = event.get_child ("message.election_info");
		auto tally (election_info.get<std::string> ("tally"));
		auto time (election_info.get<std::string> ("time"));
		// Duration and request count may be zero on devnet, so we only check that they're present
		ASSERT_EQ (1, election_info.count ("duration"));
		ASSERT_EQ (1, election_info.count ("request_count"));
		ASSERT_EQ (1, election_info.count ("voters"));
		ASSERT_GE (1U, election_info.get<unsigned> ("blocks"));
		// Make sure tally and time are non-zero.
		ASSERT_NE ("0", tally);
		ASSERT_NE ("0", time);
		auto votes_l (election_info.get_child_optional ("votes"));
		ASSERT_TRUE (votes_l.is_initialized ());
		ASSERT_EQ (1, votes_l.get ().size ());
		for (auto & vote : votes_l.get ())
		{
			std::string representative (vote.second.get<std::string> ("representative"));
			ASSERT_EQ (nano::dev::genesis_key.pub.to_account (), representative);
			std::string timestamp (vote.second.get<std::string> ("timestamp"));
			ASSERT_NE ("0", timestamp);
			std::string hash (vote.second.get<std::string> ("hash"));
			ASSERT_EQ (node1->latest (nano::dev::genesis_key.pub).to_string (), hash);
			std::string weight (vote.second.get<std::string> ("weight"));
			ASSERT_EQ (node1->balance (nano::dev::genesis_key.pub).convert_to<std::string> (), weight);
		}
	}
	catch (std::runtime_error const & ex)
	{
		FAIL () << ex.what ();
	}
}

TEST (websocket, confirmation_options_sideband)
{
	nano::system system;
	nano::node_config config (nano::get_available_port (), system.logging);
	config.websocket_config.enabled = true;
	config.websocket_config.port = nano::get_available_port ();
	auto node1 (system.add_node (config));

	std::atomic<bool> ack_ready{ false };
	auto task1 = ([&ack_ready, config, &node1] () {
		fake_websocket_client client (config.websocket_config.port);
		client.send_message (R"json({"action": "subscribe", "topic": "confirmation", "ack": "true", "options": {"confirmation_type": "active_quorum", "include_block": "false", "include_sideband_info": "true"}})json");
		client.await_ack ();
		ack_ready = true;
		EXPECT_EQ (1, node1->websocket_server->subscriber_count (nano::websocket::topic::confirmation));
		return client.get_response ();
	});
	auto future1 = std::async (std::launch::async, task1);

	ASSERT_TIMELY (10s, ack_ready);

	// Confirm a state block for an in-wallet account
	system.wallet (0)->insert_adhoc (nano::dev::genesis_key.prv);
	nano::keypair key;
	auto balance = nano::dev::constants.genesis_amount;
	auto send_amount = node1->config.online_weight_minimum.number () + 1;
	nano::block_hash previous (node1->latest (nano::dev::genesis_key.pub));
	{
		nano::state_block_builder builder;
		balance -= send_amount;
		auto send = builder
					.account (nano::dev::genesis_key.pub)
					.previous (previous)
					.representative (nano::dev::genesis_key.pub)
					.balance (balance)
					.link (key.pub)
					.sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
					.work (*system.work.generate (previous))
					.build_shared ();

		node1->process_active (send);
		previous = send->hash ();
	}

	ASSERT_TIMELY (5s, future1.wait_for (0s) == std::future_status::ready);

	auto response1 = future1.get ();
	ASSERT_TRUE (response1);
	boost::property_tree::ptree event;
	std::stringstream stream;
	stream << response1.get ();
	boost::property_tree::read_json (stream, event);
	ASSERT_EQ (event.get<std::string> ("topic"), "confirmation");
	try
	{
		boost::property_tree::ptree sideband_info = event.get_child ("message.sideband");
		// Check if height and local_timestamp are present
		ASSERT_EQ (1, sideband_info.count ("height"));
		ASSERT_EQ (1, sideband_info.count ("local_timestamp"));
		// Make sure height and local_timestamp are non-zero.
		ASSERT_NE ("0", sideband_info.get<std::string> ("height"));
		ASSERT_NE ("0", sideband_info.get<std::string> ("local_timestamp"));
	}
	catch (std::runtime_error const & ex)
	{
		FAIL () << ex.what ();
	}
}

// Tests updating options of block confirmations
TEST (websocket, confirmation_options_update)
{
	nano::system system;
	nano::node_config config (nano::get_available_port (), system.logging);
	config.websocket_config.enabled = true;
	config.websocket_config.port = nano::get_available_port ();
	auto node1 (system.add_node (config));

	std::atomic<bool> added{ false };
	std::atomic<bool> deleted{ false };
	auto task = ([&added, &deleted, config, &node1] () {
		fake_websocket_client client (node1->websocket_server->listening_port ());
		// Subscribe initially with empty options, everything will be filtered
		client.send_message (R"json({"action": "subscribe", "topic": "confirmation", "ack": "true", "options": {}})json");
		client.await_ack ();
		EXPECT_EQ (1, node1->websocket_server->subscriber_count (nano::websocket::topic::confirmation));
		// Now update filter with an account and wait for a response
		std::string add_message = boost::str (boost::format (R"json({"action": "update", "topic": "confirmation", "ack": "true", "options": {"accounts_add": ["%1%"]}})json") % nano::dev::genesis_key.pub.to_account ());
		client.send_message (add_message);
		client.await_ack ();
		EXPECT_EQ (1, node1->websocket_server->subscriber_count (nano::websocket::topic::confirmation));
		added = true;
		EXPECT_TRUE (client.get_response ());
		// Update the filter again, removing the account
		std::string delete_message = boost::str (boost::format (R"json({"action": "update", "topic": "confirmation", "ack": "true", "options": {"accounts_del": ["%1%"]}})json") % nano::dev::genesis_key.pub.to_account ());
		client.send_message (delete_message);
		client.await_ack ();
		EXPECT_EQ (1, node1->websocket_server->subscriber_count (nano::websocket::topic::confirmation));
		deleted = true;
		EXPECT_FALSE (client.get_response (1s));
	});
	auto future = std::async (std::launch::async, task);

	// Wait for update acknowledgement
	ASSERT_TIMELY (5s, added);

	// Confirm a block
	system.wallet (0)->insert_adhoc (nano::dev::genesis_key.prv);
	nano::keypair key;
	nano::state_block_builder builder;
	auto previous (node1->latest (nano::dev::genesis_key.pub));
	auto send = builder
				.account (nano::dev::genesis_key.pub)
				.previous (previous)
				.representative (nano::dev::genesis_key.pub)
				.balance (nano::dev::constants.genesis_amount - nano::Gxrb_ratio)
				.link (key.pub)
				.sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
				.work (*system.work.generate (previous))
				.build_shared ();

	node1->process_active (send);

	// Wait for delete acknowledgement
	ASSERT_TIMELY (5s, deleted);

	// Confirm another block
	previous = send->hash ();
	auto send2 = builder
				 .make_block ()
				 .account (nano::dev::genesis_key.pub)
				 .previous (previous)
				 .representative (nano::dev::genesis_key.pub)
				 .balance (nano::dev::constants.genesis_amount - 2 * nano::Gxrb_ratio)
				 .link (key.pub)
				 .sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
				 .work (*system.work.generate (previous))
				 .build_shared ();

	node1->process_active (send2);

	ASSERT_TIMELY (5s, future.wait_for (0s) == std::future_status::ready);
}

// Subscribes to votes, sends a block and awaits websocket notification of a vote arrival
TEST (websocket, vote)
{
	nano::system system;
	nano::node_config config (nano::get_available_port (), system.logging);
	config.websocket_config.enabled = true;
	config.websocket_config.port = nano::get_available_port ();
	auto node1 (system.add_node (config));

	std::atomic<bool> ack_ready{ false };
	auto task = ([&ack_ready, config, &node1] () {
		fake_websocket_client client (node1->websocket_server->listening_port ());
		client.send_message (R"json({"action": "subscribe", "topic": "vote", "ack": true})json");
		client.await_ack ();
		ack_ready = true;
		EXPECT_EQ (1, node1->websocket_server->subscriber_count (nano::websocket::topic::vote));
		return client.get_response ();
	});
	auto future = std::async (std::launch::async, task);

	ASSERT_TIMELY (5s, ack_ready);

	// Quick-confirm a block
	nano::keypair key;
	nano::state_block_builder builder;
	system.wallet (0)->insert_adhoc (nano::dev::genesis_key.prv);
	nano::block_hash previous (node1->latest (nano::dev::genesis_key.pub));
	auto send = builder
				.account (nano::dev::genesis_key.pub)
				.previous (previous)
				.representative (nano::dev::genesis_key.pub)
				.balance (nano::dev::constants.genesis_amount - (node1->online_reps.delta () + 1))
				.link (key.pub)
				.sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
				.work (*system.work.generate (previous))
				.build_shared ();

	node1->process_active (send);

	ASSERT_TIMELY (5s, future.wait_for (0s) == std::future_status::ready);

	auto response = future.get ();
	ASSERT_TRUE (response);
	boost::property_tree::ptree event;
	std::stringstream stream;
	stream << response;
	boost::property_tree::read_json (stream, event);
	ASSERT_EQ (event.get<std::string> ("topic"), "vote");
}

// Tests vote subscription options - vote type
TEST (websocket, vote_options_type)
{
	nano::system system;
	nano::node_config config (nano::get_available_port (), system.logging);
	config.websocket_config.enabled = true;
	config.websocket_config.port = nano::get_available_port ();
	auto node1 (system.add_node (config));

	std::atomic<bool> ack_ready{ false };
	auto task = ([&ack_ready, config, &node1] () {
		fake_websocket_client client (node1->websocket_server->listening_port ());
		client.send_message (R"json({"action": "subscribe", "topic": "vote", "ack": true, "options": {"include_replays": "true", "include_indeterminate": "false"}})json");
		client.await_ack ();
		ack_ready = true;
		EXPECT_EQ (1, node1->websocket_server->subscriber_count (nano::websocket::topic::vote));
		return client.get_response ();
	});
	auto future = std::async (std::launch::async, task);

	ASSERT_TIMELY (5s, ack_ready);

	// Custom made votes for simplicity
	auto vote (std::make_shared<nano::vote> (nano::dev::genesis_key.pub, nano::dev::genesis_key.prv, 0, 0, std::vector<nano::block_hash>{ nano::dev::genesis->hash () }));
	nano::websocket::message_builder builder;
	auto msg (builder.vote_received (vote, nano::vote_code::replay));
	node1->websocket_server->broadcast (msg);

	ASSERT_TIMELY (5s, future.wait_for (0s) == std::future_status::ready);

	auto response = future.get ();
	ASSERT_TRUE (response);
	boost::property_tree::ptree event;
	std::stringstream stream;
	stream << response;
	boost::property_tree::read_json (stream, event);
	auto message_contents = event.get_child ("message");
	ASSERT_EQ (1, message_contents.count ("type"));
	ASSERT_EQ ("replay", message_contents.get<std::string> ("type"));
}

// Tests vote subscription options - list of representatives
TEST (websocket, vote_options_representatives)
{
	nano::system system;
	nano::node_config config (nano::get_available_port (), system.logging);
	config.websocket_config.enabled = true;
	config.websocket_config.port = nano::get_available_port ();
	auto node1 (system.add_node (config));

	std::atomic<bool> ack_ready{ false };
	auto task1 = ([&ack_ready, config, &node1] () {
		fake_websocket_client client (node1->websocket_server->listening_port ());
		std::string message = boost::str (boost::format (R"json({"action": "subscribe", "topic": "vote", "ack": "true", "options": {"representatives": ["%1%"]}})json") % nano::dev::genesis_key.pub.to_account ());
		client.send_message (message);
		client.await_ack ();
		ack_ready = true;
		EXPECT_EQ (1, node1->websocket_server->subscriber_count (nano::websocket::topic::vote));
		auto response = client.get_response ();
		EXPECT_TRUE (response);
		boost::property_tree::ptree event;
		std::stringstream stream;
		stream << response;
		boost::property_tree::read_json (stream, event);
		EXPECT_EQ (event.get<std::string> ("topic"), "vote");
	});
	auto future1 = std::async (std::launch::async, task1);

	ASSERT_TIMELY (5s, ack_ready);

	// Quick-confirm a block
	nano::keypair key;
	auto balance = nano::dev::constants.genesis_amount;
	system.wallet (0)->insert_adhoc (nano::dev::genesis_key.prv);
	auto send_amount = node1->online_reps.delta () + 1;
	auto confirm_block = [&] () {
		nano::block_hash previous (node1->latest (nano::dev::genesis_key.pub));
		balance -= send_amount;
		nano::state_block_builder builder;
		auto send = builder
					.account (nano::dev::genesis_key.pub)
					.previous (previous)
					.representative (nano::dev::genesis_key.pub)
					.balance (balance)
					.link (key.pub)
					.sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
					.work (*system.work.generate (previous))
					.build_shared ();
		node1->process_active (send);
	};
	confirm_block ();

	ASSERT_TIMELY (5s, future1.wait_for (0s) == std::future_status::ready);

	ack_ready = false;
	auto task2 = ([&ack_ready, config, &node1] () {
		fake_websocket_client client (node1->websocket_server->listening_port ());
		client.send_message (R"json({"action": "subscribe", "topic": "vote", "ack": "true", "options": {"representatives": ["xrb_invalid"]}})json");
		client.await_ack ();
		ack_ready = true;
		EXPECT_EQ (1, node1->websocket_server->subscriber_count (nano::websocket::topic::vote));
		auto response = client.get_response ();
		// A list of invalid representatives is the same as no filter
		EXPECT_TRUE (response);
	});
	auto future2 = std::async (std::launch::async, task2);

	// Wait for the subscription to be acknowledged
	ASSERT_TIMELY (5s, ack_ready);

	// Confirm another block
	confirm_block ();

	ASSERT_TIMELY (5s, future2.wait_for (0s) == std::future_status::ready);
}

// Test client subscribing to notifications for work generation
TEST (websocket, work)
{
	nano::system system;
	nano::node_config config (nano::get_available_port (), system.logging);
	config.websocket_config.enabled = true;
	config.websocket_config.port = nano::get_available_port ();
	auto node1 (system.add_node (config));

	ASSERT_EQ (0, node1->websocket_server->subscriber_count (nano::websocket::topic::work));

	// Subscribe to work and wait for response asynchronously
	std::atomic<bool> ack_ready{ false };
	auto task = ([&ack_ready, config, &node1] () {
		fake_websocket_client client (node1->websocket_server->listening_port ());
		client.send_message (R"json({"action": "subscribe", "topic": "work", "ack": true})json");
		client.await_ack ();
		ack_ready = true;
		EXPECT_EQ (1, node1->websocket_server->subscriber_count (nano::websocket::topic::work));
		return client.get_response ();
	});
	auto future = std::async (std::launch::async, task);

	// Wait for acknowledge
	ASSERT_TIMELY (5s, ack_ready);
	ASSERT_EQ (1, node1->websocket_server->subscriber_count (nano::websocket::topic::work));

	// Generate work
	nano::block_hash hash{ 1 };
	auto work (node1->work_generate_blocking (hash));
	ASSERT_TRUE (work.is_initialized ());

	// Wait for the work notification
	ASSERT_TIMELY (5s, future.wait_for (0s) == std::future_status::ready);

	// Check the work notification message
	auto response = future.get ();
	ASSERT_TRUE (response);
	std::stringstream stream;
	stream << response;
	boost::property_tree::ptree event;
	boost::property_tree::read_json (stream, event);
	ASSERT_EQ (event.get<std::string> ("topic"), "work");

	auto & contents = event.get_child ("message");
	ASSERT_EQ (contents.get<std::string> ("success"), "true");
	ASSERT_LT (contents.get<unsigned> ("duration"), 10000U);

	ASSERT_EQ (1, contents.count ("request"));
	auto & request = contents.get_child ("request");
	ASSERT_EQ (request.get<std::string> ("version"), nano::to_string (nano::work_version::work_1));
	ASSERT_EQ (request.get<std::string> ("hash"), hash.to_string ());
	ASSERT_EQ (request.get<std::string> ("difficulty"), nano::to_string_hex (node1->default_difficulty (nano::work_version::work_1)));
	ASSERT_EQ (request.get<double> ("multiplier"), 1.0);

	ASSERT_EQ (1, contents.count ("result"));
	auto & result = contents.get_child ("result");
	uint64_t result_difficulty;
	nano::from_string_hex (result.get<std::string> ("difficulty"), result_difficulty);
	ASSERT_GE (result_difficulty, node1->default_difficulty (nano::work_version::work_1));
	ASSERT_NEAR (result.get<double> ("multiplier"), nano::difficulty::to_multiplier (result_difficulty, node1->default_difficulty (nano::work_version::work_1)), 1e-6);
	ASSERT_EQ (result.get<std::string> ("work"), nano::to_string_hex (work.get ()));

	ASSERT_EQ (1, contents.count ("bad_peers"));
	auto & bad_peers = contents.get_child ("bad_peers");
	ASSERT_TRUE (bad_peers.empty ());

	ASSERT_EQ (contents.get<std::string> ("reason"), "");
}

// Test client subscribing to notifications for bootstrap
TEST (websocket, bootstrap)
{
	nano::system system;
	nano::node_config config (nano::get_available_port (), system.logging);
	config.websocket_config.enabled = true;
	config.websocket_config.port = nano::get_available_port ();
	auto node1 (system.add_node (config));

	ASSERT_EQ (0, node1->websocket_server->subscriber_count (nano::websocket::topic::bootstrap));

	// Subscribe to bootstrap and wait for response asynchronously
	std::atomic<bool> ack_ready{ false };
	auto task = ([&ack_ready, config, &node1] () {
		fake_websocket_client client (node1->websocket_server->listening_port ());
		client.send_message (R"json({"action": "subscribe", "topic": "bootstrap", "ack": true})json");
		client.await_ack ();
		ack_ready = true;
		EXPECT_EQ (1, node1->websocket_server->subscriber_count (nano::websocket::topic::bootstrap));
		return client.get_response ();
	});
	auto future = std::async (std::launch::async, task);

	// Wait for acknowledge
	ASSERT_TIMELY (5s, ack_ready);

	// Start bootstrap attempt
	node1->bootstrap_initiator.bootstrap (true, "123abc");
	ASSERT_TIMELY (5s, nullptr == node1->bootstrap_initiator.current_attempt ());

	// Wait for the bootstrap notification
	ASSERT_TIMELY (5s, future.wait_for (0s) == std::future_status::ready);

	// Check the bootstrap notification message
	auto response = future.get ();
	ASSERT_TRUE (response);
	std::stringstream stream;
	stream << response;
	boost::property_tree::ptree event;
	boost::property_tree::read_json (stream, event);
	ASSERT_EQ (event.get<std::string> ("topic"), "bootstrap");

	auto & contents = event.get_child ("message");
	ASSERT_EQ (contents.get<std::string> ("reason"), "started");
	ASSERT_EQ (contents.get<std::string> ("id"), "123abc");
	ASSERT_EQ (contents.get<std::string> ("mode"), "legacy");

	// Wait for bootstrap finish
	ASSERT_TIMELY (5s, !node1->bootstrap_initiator.in_progress ());
}

TEST (websocket, bootstrap_exited)
{
	nano::system system;
	nano::node_config config (nano::get_available_port (), system.logging);
	config.websocket_config.enabled = true;
	config.websocket_config.port = nano::get_available_port ();
	auto node1 (system.add_node (config));

	// Start bootstrap, exit after subscription
	std::atomic<bool> bootstrap_started{ false };
	nano::util::counted_completion subscribed_completion (1);
	std::thread bootstrap_thread ([node1, &system, &bootstrap_started, &subscribed_completion] () {
		std::shared_ptr<nano::bootstrap_attempt> attempt;
		while (attempt == nullptr)
		{
			std::this_thread::sleep_for (50ms);
			node1->bootstrap_initiator.bootstrap (true, "123abc");
			attempt = node1->bootstrap_initiator.current_attempt ();
		}
		ASSERT_NE (nullptr, attempt);
		bootstrap_started = true;
		EXPECT_FALSE (subscribed_completion.await_count_for (5s));
	});

	// Wait for bootstrap start
	ASSERT_TIMELY (5s, bootstrap_started);

	// Subscribe to bootstrap and wait for response asynchronously
	std::atomic<bool> ack_ready{ false };
	auto task = ([&ack_ready, config, &node1] () {
		fake_websocket_client client (node1->websocket_server->listening_port ());
		client.send_message (R"json({"action": "subscribe", "topic": "bootstrap", "ack": true})json");
		client.await_ack ();
		ack_ready = true;
		EXPECT_EQ (1, node1->websocket_server->subscriber_count (nano::websocket::topic::bootstrap));
		return client.get_response ();
	});
	auto future = std::async (std::launch::async, task);

	// Wait for acknowledge
	ASSERT_TIMELY (5s, ack_ready);

	// Wait for the bootstrap notification
	subscribed_completion.increment ();
	bootstrap_thread.join ();
	ASSERT_TIMELY (5s, future.wait_for (0s) == std::future_status::ready);

	// Check the bootstrap notification message
	auto response = future.get ();
	ASSERT_TRUE (response);
	std::stringstream stream;
	stream << response;
	boost::property_tree::ptree event;
	boost::property_tree::read_json (stream, event);
	ASSERT_EQ (event.get<std::string> ("topic"), "bootstrap");

	auto & contents = event.get_child ("message");
	ASSERT_EQ (contents.get<std::string> ("reason"), "exited");
	ASSERT_EQ (contents.get<std::string> ("id"), "123abc");
	ASSERT_EQ (contents.get<std::string> ("mode"), "legacy");
	ASSERT_EQ (contents.get<unsigned> ("total_blocks"), 0U);
	ASSERT_LT (contents.get<unsigned> ("duration"), 15000U);
}

// Tests sending keepalive
TEST (websocket, ws_keepalive)
{
	nano::system system;
	nano::node_config config (nano::get_available_port (), system.logging);
	config.websocket_config.enabled = true;
	config.websocket_config.port = nano::get_available_port ();
	auto node1 (system.add_node (config));

	auto task = ([&node1] () {
		fake_websocket_client client (node1->websocket_server->listening_port ());
		client.send_message (R"json({"action": "ping"})json");
		client.await_ack ();
	});
	auto future = std::async (std::launch::async, task);

	ASSERT_TIMELY (5s, future.wait_for (0s) == std::future_status::ready);
}

// Tests sending telemetry
TEST (websocket, telemetry)
{
	nano::system system;
	nano::node_config config (nano::get_available_port (), system.logging);
	config.websocket_config.enabled = true;
	config.websocket_config.port = nano::get_available_port ();
	nano::node_flags node_flags;
	node_flags.disable_initial_telemetry_requests = true;
	node_flags.disable_ongoing_telemetry_requests = true;
	auto node1 (system.add_node (config, node_flags));
	config.peering_port = nano::get_available_port ();
	config.websocket_config.enabled = true;
	config.websocket_config.port = nano::get_available_port ();
	auto node2 (system.add_node (config, node_flags));

	wait_peer_connections (system);

	std::atomic<bool> done{ false };
	auto task = ([config = node1->config, &node1, &done] () {
		fake_websocket_client client (node1->websocket_server->listening_port ());
		client.send_message (R"json({"action": "subscribe", "topic": "telemetry", "ack": true})json");
		client.await_ack ();
		done = true;
		EXPECT_EQ (1, node1->websocket_server->subscriber_count (nano::websocket::topic::telemetry));
		return client.get_response ();
	});

	auto future = std::async (std::launch::async, task);

	ASSERT_TIMELY (10s, done);

	node1->telemetry->get_metrics_single_peer_async (node1->network.find_channel (node2->network.endpoint ()), [] (auto const & response_a) {
		ASSERT_FALSE (response_a.error);
	});

	ASSERT_TIMELY (10s, future.wait_for (0s) == std::future_status::ready);

	// Check the telemetry notification message
	auto response = future.get ();

	std::stringstream stream;
	stream << response;
	boost::property_tree::ptree event;
	boost::property_tree::read_json (stream, event);
	ASSERT_EQ (event.get<std::string> ("topic"), "telemetry");

	auto & contents = event.get_child ("message");
	nano::jsonconfig telemetry_contents (contents);
	nano::telemetry_data telemetry_data;
	telemetry_data.deserialize_json (telemetry_contents, false);
	compare_default_telemetry_response_data (telemetry_data, node2->network_params, node2->config.bandwidth_limit, node2->default_difficulty (nano::work_version::work_1), node2->node_id);

	ASSERT_EQ (contents.get<std::string> ("address"), node2->network.endpoint ().address ().to_string ());
	ASSERT_EQ (contents.get<uint16_t> ("port"), node2->network.endpoint ().port ());

	// Other node should have no subscribers
	EXPECT_EQ (0, node2->websocket_server->subscriber_count (nano::websocket::topic::telemetry));
}

TEST (websocket, new_unconfirmed_block)
{
	nano::system system;
	nano::node_config config (nano::get_available_port (), system.logging);
	config.websocket_config.enabled = true;
	config.websocket_config.port = nano::get_available_port ();
	auto node1 (system.add_node (config));

	std::atomic<bool> ack_ready{ false };
	auto task = ([&ack_ready, config, node1] () {
		fake_websocket_client client (node1->websocket_server->listening_port ());
		client.send_message (R"json({"action": "subscribe", "topic": "new_unconfirmed_block", "ack": "true"})json");
		client.await_ack ();
		ack_ready = true;
		EXPECT_EQ (1, node1->websocket_server->subscriber_count (nano::websocket::topic::new_unconfirmed_block));
		return client.get_response ();
	});
	auto future = std::async (std::launch::async, task);

	ASSERT_TIMELY (5s, ack_ready);

	nano::state_block_builder builder;
	// Process a new block
	auto send1 = builder
				 .account (nano::dev::genesis_key.pub)
				 .previous (nano::dev::genesis->hash ())
				 .representative (nano::dev::genesis_key.pub)
				 .balance (nano::dev::constants.genesis_amount - 1)
				 .link (nano::dev::genesis_key.pub)
				 .sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
				 .work (*system.work.generate (nano::dev::genesis->hash ()))
				 .build_shared ();

	ASSERT_EQ (nano::process_result::progress, node1->process_local (send1).code);

	ASSERT_TIMELY (5s, future.wait_for (0s) == std::future_status::ready);

	// Check the response
	boost::optional<std::string> response = future.get ();
	ASSERT_TRUE (response);
	std::stringstream stream;
	stream << response;
	boost::property_tree::ptree event;
	boost::property_tree::read_json (stream, event);
	ASSERT_EQ (event.get<std::string> ("topic"), "new_unconfirmed_block");

	auto message_contents = event.get_child ("message");
	ASSERT_EQ ("state", message_contents.get<std::string> ("type"));
	ASSERT_EQ ("send", message_contents.get<std::string> ("subtype"));
}

max_stars_repo_name: CathalT/raiblocks

max_issues_repo_name: CathalT/raiblocks

max_forks_repo_name: CathalT/raiblocks


text:
//  (C) Copyright Matt Borland 2021.
//  Use, modification and distribution are subject to the
//  Boost Software License, Version 1.0. (See accompanying file
//  LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)

#include <boost/math/ccmath/div.hpp>
#include "test_compile_result.hpp"

void compile_and_link_test()
{
   check_result<std::div_t>(boost::math::ccmath::div(1, 1));
   check_result<std::ldiv_t>(boost::math::ccmath::div(1l, 1l));
   check_result<std::lldiv_t>(boost::math::ccmath::div(1ll, 1ll));
}

ext: cpp

max_stars_repo_name: oleg-alexandrov/math

max_issues_repo_name: oleg-alexandrov/math

max_forks_repo_name: oleg-alexandrov/math


text:
// Copyright Louis Dionne 2013-2016
// Distributed under the Boost Software License, Version 1.0.
// (See accompanying file LICENSE.md or copy at http://boost.org/LICENSE_1_0.txt)

#include <boost/hana/all_of.hpp>
#include <boost/hana/assert.hpp>
#include <boost/hana/config.hpp>
#include <boost/hana/ext/std/integral_constant.hpp>
#include <boost/hana/integral_constant.hpp>
#include <boost/hana/mod.hpp>
#include <boost/hana/not.hpp>
#include <boost/hana/not_equal.hpp>
#include <boost/hana/traits.hpp>
#include <boost/hana/tuple.hpp>
#include <boost/hana/type.hpp>

#include <type_traits>
namespace hana = boost::hana;
using namespace hana::literals;

BOOST_HANA_CONSTEXPR_LAMBDA auto is_odd = [](auto x) {
    return x % 2_c != 0_c;
};

int main() {
    BOOST_HANA_CONSTEXPR_CHECK(hana::all_of(hana::make_tuple(1, 3), is_odd));
    BOOST_HANA_CONSTANT_CHECK(!hana::all_of(hana::make_tuple(3_c, 4_c), is_odd));

    BOOST_HANA_CONSTANT_CHECK(
      !hana::all_of(hana::make_tuple(hana::type<void>{}, hana::type<char&>{}),
                    hana::traits::is_void)
    );

    BOOST_HANA_CONSTANT_CHECK(
      hana::all_of(hana::make_tuple(hana::type_c<int>, hana::type_c<char>),
                   hana::traits::is_integral)
    );
}

max_stars_repo_name: metux/boost

max_issues_repo_name: metux/boost

max_forks_repo_name: metux/boost


text:
/***************************************************************************
 * NASA Glenn Research Center, Cleveland, OH
 * Released under the NASA Open Source Agreement (NOSA)
 * May  2021
 ****************************************************************************
 */

#include <iostream>
#include "message.hpp"
#include "ZmqStorageInterface.h"
#include "BundleStorageManagerMT.h"
#include "BundleStorageManagerAsio.h"
#include "Logger.h"
#include <set>
#include <boost/lexical_cast.hpp>
#include <boost/make_unique.hpp>
#include "codec/CustodyIdAllocator.h"
#include "codec/CustodyTransferManager.h"
#include "Uri.h"
#include "CustodyTimers.h"
#include "codec/BundleViewV7.h"

typedef std::pair<cbhe_eid_t, bool> eid_plus_isanyserviceid_pair_t;

ZmqStorageInterface::ZmqStorageInterface() : m_running(false) {}

ZmqStorageInterface::~ZmqStorageInterface() {
    Stop();
}

void ZmqStorageInterface::Stop() {
    m_running = false; //thread stopping criteria
    if (m_threadPtr) {
        m_threadPtr->join();
        m_threadPtr.reset();
    }
}

bool ZmqStorageInterface::Init(const HdtnConfig & hdtnConfig, zmq::context_t * hdtnOneProcessZmqInprocContextPtr) {
    m_hdtnConfig = hdtnConfig;
    //according to ION.pdf v4.0.1 on page 100 it says:
    //  Remember that the format for this argument is ipn:element_number.0 and that
    //  the final 0 is required, as custodial/administration service is always service 0.
    //HDTN shall default m_myCustodialServiceId to 0 although it is changeable in the hdtn config json file
    M_HDTN_EID_CUSTODY.Set(m_hdtnConfig.m_myNodeId, m_hdtnConfig.m_myCustodialServiceId);
    m_hdtnOneProcessZmqInprocContextPtr = hdtnOneProcessZmqInprocContextPtr;

    //{

    m_zmqContextPtr = boost::make_unique<zmq::context_t>();

    //telemetry not implemnted yet
    m_telemetrySockPtr = boost::make_unique<zmq::socket_t>(*m_zmqContextPtr, zmq::socket_type::rep);
    try {
        m_telemetrySockPtr->bind(HDTN_STORAGE_TELEM_PATH);
    }
    catch (const zmq::error_t & ex) {
        std::cerr << "error: cannot bind telemetry socket: " << ex.what() << std::endl;
        hdtn::Logger::getInstance()->logError("storage", "Error: cannot bind telemetry socket: " + std::string(ex.what()));
        return false;
    }

    if (hdtnOneProcessZmqInprocContextPtr) {
        m_zmqPushSock_connectingStorageToBoundEgressPtr = boost::make_unique<zmq::socket_t>(*m_hdtnOneProcessZmqInprocContextPtr, zmq::socket_type::pair);
        m_zmqPullSock_boundEgressToConnectingStoragePtr = boost::make_unique<zmq::socket_t>(*m_hdtnOneProcessZmqInprocContextPtr, zmq::socket_type::pair);
        m_zmqPushSock_connectingStorageToBoundIngressPtr = boost::make_unique<zmq::socket_t>(*m_hdtnOneProcessZmqInprocContextPtr, zmq::socket_type::pair);
        m_zmqPullSock_boundIngressToConnectingStoragePtr = boost::make_unique<zmq::socket_t>(*hdtnOneProcessZmqInprocContextPtr, zmq::socket_type::pair);
        try {
            m_zmqPushSock_connectingStorageToBoundEgressPtr->connect(std::string("inproc://connecting_storage_to_bound_egress")); // egress should bind
            m_zmqPullSock_boundEgressToConnectingStoragePtr->connect(std::string("inproc://bound_egress_to_connecting_storage")); // egress should bind
            m_zmqPushSock_connectingStorageToBoundIngressPtr->connect(std::string("inproc://connecting_storage_to_bound_ingress"));
            m_zmqPullSock_boundIngressToConnectingStoragePtr->connect(std::string("inproc://bound_ingress_to_connecting_storage"));
        }
        catch (const zmq::error_t & ex) {
            std::cerr << "error in ZmqStorageInterface::Init: cannot connect inproc socket: " << ex.what() << std::endl;
            hdtn::Logger::getInstance()->logError("storage", "error in ZmqStorageInterface::Init: cannot connect inproc socket: " + std::string(ex.what()));
            return false;
        }
    }
    else {
        m_zmqPushSock_connectingStorageToBoundEgressPtr = boost::make_unique<zmq::socket_t>(*m_zmqContextPtr, zmq::socket_type::push);
        const std::string connect_connectingStorageToBoundEgressPath(
            std::string("tcp://") +
            m_hdtnConfig.m_zmqEgressAddress +
            std::string(":") +
            boost::lexical_cast<std::string>(m_hdtnConfig.m_zmqConnectingStorageToBoundEgressPortPath));

        m_zmqPullSock_boundEgressToConnectingStoragePtr = boost::make_unique<zmq::socket_t>(*m_zmqContextPtr, zmq::socket_type::pull);
        const std::string connect_boundEgressToConnectingStoragePath(
            std::string("tcp://") +
            m_hdtnConfig.m_zmqEgressAddress +
            std::string(":") +
            boost::lexical_cast<std::string>(m_hdtnConfig.m_zmqBoundEgressToConnectingStoragePortPath));

        m_zmqPushSock_connectingStorageToBoundIngressPtr = boost::make_unique<zmq::socket_t>(*m_zmqContextPtr, zmq::socket_type::push);
        const std::string connect_connectingStorageToBoundIngressPath(
            std::string("tcp://") +
            m_hdtnConfig.m_zmqIngressAddress +
            std::string(":") +
            boost::lexical_cast<std::string>(m_hdtnConfig.m_zmqConnectingStorageToBoundIngressPortPath));

        m_zmqPullSock_boundIngressToConnectingStoragePtr = boost::make_unique<zmq::socket_t>(*m_zmqContextPtr, zmq::socket_type::pull);
        const std::string connect_boundIngressToConnectingStoragePath(
            std::string("tcp://") +
            m_hdtnConfig.m_zmqIngressAddress +
            std::string(":") +
            boost::lexical_cast<std::string>(m_hdtnConfig.m_zmqBoundIngressToConnectingStoragePortPath));

        try {
            m_zmqPushSock_connectingStorageToBoundEgressPtr->connect(connect_connectingStorageToBoundEgressPath); // egress should bind
            m_zmqPullSock_boundEgressToConnectingStoragePtr->connect(connect_boundEgressToConnectingStoragePath); // egress should bind
            m_zmqPushSock_connectingStorageToBoundIngressPtr->connect(connect_connectingStorageToBoundIngressPath);
            m_zmqPullSock_boundIngressToConnectingStoragePtr->connect(connect_boundIngressToConnectingStoragePath);
        }
        catch (const zmq::error_t & ex) {
            std::cerr << "error: cannot connect socket: " << ex.what() << std::endl;
            hdtn::Logger::getInstance()->logError("storage", "Error: cannot connect socket: " + std::string(ex.what()));
            return false;
        }
    }

    m_zmqSubSock_boundReleaseToConnectingStoragePtr = boost::make_unique<zmq::socket_t>(*m_zmqContextPtr, zmq::socket_type::sub);
    const std::string connect_boundSchedulerPubSubPath(
        std::string("tcp://") +
        m_hdtnConfig.m_zmqSchedulerAddress +
        std::string(":") +
        boost::lexical_cast<std::string>(m_hdtnConfig.m_zmqBoundSchedulerPubSubPortPath));
    try {
        m_zmqSubSock_boundReleaseToConnectingStoragePtr->connect(connect_boundSchedulerPubSubPath);// config.releaseWorker);
        m_zmqSubSock_boundReleaseToConnectingStoragePtr->set(zmq::sockopt::subscribe, "");
        std::cout << "release sock connected to " << connect_boundSchedulerPubSubPath << std::endl;
        hdtn::Logger::getInstance()->logNotification("storage", "Release sock connected to " + connect_boundSchedulerPubSubPath);
    }
    catch (const zmq::error_t & ex) {
        std::cerr << "error: cannot connect release socket: " << ex.what() << std::endl;
        hdtn::Logger::getInstance()->logError("storage", "Error: cannot connect release socket: " + std::string(ex.what()));
        return false;
    }

    // Use a form of receive that times out so we can terminate cleanly.
    try {
        static const int timeout = 250;  // milliseconds
        m_zmqPullSock_boundEgressToConnectingStoragePtr->set(zmq::sockopt::rcvtimeo, timeout);
        m_zmqPullSock_boundIngressToConnectingStoragePtr->set(zmq::sockopt::rcvtimeo, timeout);
        m_zmqSubSock_boundReleaseToConnectingStoragePtr->set(zmq::sockopt::rcvtimeo, timeout);
    }
    catch (const zmq::error_t & ex) {
        std::cerr << "error: cannot set timeout on receive sockets: " << ex.what() << std::endl;
        hdtn::Logger::getInstance()->logError("storage", "Error: cannot set timeout on receive sockets: " + std::string(ex.what()));
        return false;
    }

    if (!m_running) {
        m_running = true;
        std::cout << "[ZmqStorageInterface] Launching worker thread ..." << std::endl;
        hdtn::Logger::getInstance()->logNotification("storage", "[ZmqStorageInterface] Launching worker thread");
        m_threadStartupComplete = false;
        m_threadPtr = boost::make_unique<boost::thread>(
            boost::bind(&ZmqStorageInterface::ThreadFunc, this)); //create and start the worker thread
        for (unsigned int attempt = 0; attempt < 10; ++attempt) {
            if (m_threadStartupComplete) {
                break;
            }
            std::cout << "waiting for worker thread to start up...\n";
            boost::this_thread::sleep(boost::posix_time::seconds(1));
        }
        if (!m_threadStartupComplete) {
            std::cout << "error: storage thread took too long to start up.. exiting\n";
            return false;
        }
        else {
            std::cout << "worker thread started\n";
        }
    }
    return true;
}

static bool WriteAcsBundle(BundleStorageManagerBase & bsm, CustodyIdAllocator & custodyIdAllocator,
    const Bpv6CbhePrimaryBlock & primary, const std::vector<uint8_t> & acsBundleSerialized)
{
    const cbhe_eid_t & hdtnSrcEid = primary.m_sourceNodeId;
    const uint64_t newCustodyIdForAcsCustodySignal = custodyIdAllocator.GetNextCustodyIdForNextHopCtebToSend(hdtnSrcEid);

    //write custody signal to disk
    BundleStorageManagerSession_WriteToDisk sessionWrite;
    const uint64_t totalSegmentsRequired = bsm.Push(sessionWrite, primary, acsBundleSerialized.size());
    //std::cout << "totalSegmentsRequired " << totalSegmentsRequired << "\n";
    if (totalSegmentsRequired == 0) {
        const std::string msg = "out of space for acs custody signal";
        std::cerr << msg << "\n";
        hdtn::Logger::getInstance()->logError("storage", msg);
        return false;
    }

    const uint64_t totalBytesPushed = bsm.PushAllSegments(sessionWrite, primary,
        newCustodyIdForAcsCustodySignal, acsBundleSerialized.data(), acsBundleSerialized.size());
    if (totalBytesPushed != acsBundleSerialized.size()) {
        const std::string msg = "totalBytesPushed != acsBundleSerialized.size";
        std::cerr << msg << "\n";
        hdtn::Logger::getInstance()->logError("storage", msg);
        return false;
    }
    return true;
}

static bool Write(zmq::message_t *message, BundleStorageManagerBase & bsm,
    CustodyIdAllocator & custodyIdAllocator, CustodyTransferManager & ctm,
    CustodyTimers & custodyTimers,
    BundleViewV6 & custodySignalRfc5050RenderedBundleView,
    cbhe_eid_t & finalDestEidReturned, ZmqStorageInterface * forStats)
{

    const uint8_t firstByte = ((const uint8_t*)message->data())[0];
    const bool isBpVersion6 = (firstByte == 6);
    const bool isBpVersion7 = (firstByte == ((4U << 5) | 31U));  //CBOR major type 4, additional information 31 (Indefinite-Length Array)
    if (isBpVersion6) {
        BundleViewV6 bv;
        if (!bv.LoadBundle((uint8_t *)message->data(), message->size())) { //invalid bundle
            std::cerr << "malformed bundle\n";
            return false;
        }
        const Bpv6CbhePrimaryBlock & primary = bv.m_primaryBlockView.header;
        finalDestEidReturned = primary.m_destinationEid;

        //admin records pertaining to this hdtn node do not get written to disk.. they signal a deletion from disk
        static const BPV6_BUNDLEFLAG requiredPrimaryFlagsForAdminRecord = BPV6_BUNDLEFLAG::SINGLETON | BPV6_BUNDLEFLAG::ADMINRECORD;
        if (((primary.m_bundleProcessingControlFlags & requiredPrimaryFlagsForAdminRecord) == requiredPrimaryFlagsForAdminRecord) && (finalDestEidReturned == forStats->M_HDTN_EID_CUSTODY)) {
            std::vector<BundleViewV6::Bpv6CanonicalBlockView*> blocks;
            bv.GetCanonicalBlocksByType(BPV6_BLOCK_TYPE_CODE::PAYLOAD, blocks);
            if (blocks.size() != 1) {
                std::cerr << "error admin record does not have a payload block\n";
                return false;
            }
            Bpv6AdministrativeRecord* adminRecordBlockPtr = dynamic_cast<Bpv6AdministrativeRecord*>(blocks[0]->headerPtr.get());
            if (adminRecordBlockPtr == NULL) {
                std::cerr << "error null Bpv6AdministrativeRecord\n";
                return false;
            }
            const BPV6_ADMINISTRATIVE_RECORD_TYPE_CODE adminRecordType = adminRecordBlockPtr->m_adminRecordTypeCode;

            if (adminRecordType == BPV6_ADMINISTRATIVE_RECORD_TYPE_CODE::AGGREGATE_CUSTODY_SIGNAL) {
                ++forStats->m_numAcsPacketsReceived;
                //check acs
                Bpv6AdministrativeRecordContentAggregateCustodySignal * acsPtr = dynamic_cast<Bpv6AdministrativeRecordContentAggregateCustodySignal*>(adminRecordBlockPtr->m_adminRecordContentPtr.get());
                if (acsPtr == NULL) {
                    std::cerr << "error null AggregateCustodySignal\n";
                    return false;
                }
                Bpv6AdministrativeRecordContentAggregateCustodySignal & acs = *(reinterpret_cast<Bpv6AdministrativeRecordContentAggregateCustodySignal*>(acsPtr));
                if (!acs.DidCustodyTransferSucceed()) {
                    std::cerr << "custody transfer failed with reason code " << static_cast<unsigned int>(acs.GetReasonCode()) << "\n";
                    return false;
                }

                //todo figure out what to do with failed custody from next hop
                for (std::set<FragmentSet::data_fragment_t>::const_iterator it = acs.m_custodyIdFills.cbegin(); it != acs.m_custodyIdFills.cend(); ++it) {
                    forStats->m_numAcsCustodyTransfers += (it->endIndex + 1) - it->beginIndex;
                    custodyIdAllocator.FreeCustodyIdRange(it->beginIndex, it->endIndex);
                    for (uint64_t currentCustodyId = it->beginIndex; currentCustodyId <= it->endIndex; ++currentCustodyId) {
                        catalog_entry_t * catalogEntryPtr = bsm.GetCatalogEntryPtrFromCustodyId(currentCustodyId);
                        if (catalogEntryPtr == NULL) {
                            std::cout << "error finding catalog entry for bundle identified by acs custody signal\n";
                            continue;
                        }
                        if (!custodyTimers.CancelCustodyTransferTimer(catalogEntryPtr->destEid, currentCustodyId)) {
                            std::cout << "notice: can't find custody timer associated with bundle identified by acs custody signal\n";
                        }
                        if (!bsm.RemoveReadBundleFromDisk(catalogEntryPtr, currentCustodyId)) {
                            std::cout << "error freeing bundle identified by acs custody signal from disk\n";
                            continue;
                        }
                        ++forStats->m_totalBundlesErasedFromStorageWithCustodyTransfer;
                    }
                }
            }
            else if (adminRecordType == BPV6_ADMINISTRATIVE_RECORD_TYPE_CODE::CUSTODY_SIGNAL) { //rfc5050 style custody transfer
                //check acs
                Bpv6AdministrativeRecordContentCustodySignal * csPtr = dynamic_cast<Bpv6AdministrativeRecordContentCustodySignal*>(adminRecordBlockPtr->m_adminRecordContentPtr.get());
                if (csPtr == NULL) {
                    std::cerr << "error null CustodySignal\n";
                    return false;
                }
                Bpv6AdministrativeRecordContentCustodySignal & cs = *(reinterpret_cast<Bpv6AdministrativeRecordContentCustodySignal*>(csPtr));
                if (!cs.DidCustodyTransferSucceed()) {
                    std::cerr << "custody transfer failed with reason code " << cs.GetReasonCode() << "\n";
                    return false;
                }
                uint64_t * custodyIdPtr;
                if (cs.m_isFragment) {
                    cbhe_bundle_uuid_t uuid;
                    if (!Uri::ParseIpnUriString(cs.m_bundleSourceEid, uuid.srcEid.nodeId, uuid.srcEid.serviceId)) {
                        std::cerr << "error custody signal with bad ipn string\n";
                        return false;
                    }
                    uuid.creationSeconds = cs.m_copyOfBundleCreationTimestamp.secondsSinceStartOfYear2000;
                    uuid.sequence = cs.m_copyOfBundleCreationTimestamp.sequenceNumber;
                    uuid.fragmentOffset = cs.m_fragmentOffsetIfPresent;
                    uuid.dataLength = cs.m_fragmentLengthIfPresent;
                    custodyIdPtr = bsm.GetCustodyIdFromUuid(uuid);
                }
                else {
                    cbhe_bundle_uuid_nofragment_t uuid;
                    if (!Uri::ParseIpnUriString(cs.m_bundleSourceEid, uuid.srcEid.nodeId, uuid.srcEid.serviceId)) {
                        std::cerr << "error custody signal with bad ipn string\n";
                        return false;
                    }
                    uuid.creationSeconds = cs.m_copyOfBundleCreationTimestamp.secondsSinceStartOfYear2000;
                    uuid.sequence = cs.m_copyOfBundleCreationTimestamp.sequenceNumber;
                    //std::cout << "uuid: " << "cs " << uuid.creationSeconds << "  seq " << uuid.sequence << "  " << uuid.srcEid.nodeId << "," << uuid.srcEid.serviceId << std::endl;
                    custodyIdPtr = bsm.GetCustodyIdFromUuid(uuid);
                }
                if (custodyIdPtr == NULL) {
                    std::cerr << "error custody signal does not match a bundle in the storage database\n";
                    return false;
                }
                const uint64_t custodyIdFromRfc5050 = *custodyIdPtr;
                custodyIdAllocator.FreeCustodyId(custodyIdFromRfc5050);
                catalog_entry_t * catalogEntryPtr = bsm.GetCatalogEntryPtrFromCustodyId(custodyIdFromRfc5050);
                if (catalogEntryPtr == NULL) {
                    std::cout << "error finding catalog entry for bundle identified by rfc5050 custody signal\n";
                    return false;
                }
                if (!custodyTimers.CancelCustodyTransferTimer(catalogEntryPtr->destEid, custodyIdFromRfc5050)) {
                    std::cout << "notice: can't find custody timer associated with bundle identified by rfc5050 custody signal\n";
                }
                if (!bsm.RemoveReadBundleFromDisk(catalogEntryPtr, custodyIdFromRfc5050)) {
                    std::cout << "error freeing bundle identified by rfc5050 custody signal from disk\n";
                    return false;
                }
                ++forStats->m_totalBundlesErasedFromStorageWithCustodyTransfer;
                ++forStats->m_numRfc5050CustodyTransfers;
            }
            else {
                std::cerr << "error unknown admin record type\n";
                return false;
            }
            return true; //do not proceed past this point so that the signal is not written to disk
        }

        //write non admin records to disk (unless newly generated below)
        const uint64_t newCustodyId = custodyIdAllocator.GetNextCustodyIdForNextHopCtebToSend(primary.m_sourceNodeId);
        static const BPV6_BUNDLEFLAG requiredPrimaryFlagsForCustody = BPV6_BUNDLEFLAG::SINGLETON | BPV6_BUNDLEFLAG::CUSTODY_REQUESTED;
        if ((primary.m_bundleProcessingControlFlags & requiredPrimaryFlagsForCustody) == requiredPrimaryFlagsForCustody) {
            if (!ctm.ProcessCustodyOfBundle(bv, true, newCustodyId, BPV6_ACS_STATUS_REASON_INDICES::SUCCESS__NO_ADDITIONAL_INFORMATION,
                custodySignalRfc5050RenderedBundleView)) {
                std::cerr << "error unable to process custody\n";
            }
            else if (!bv.Render(message->size() + 200)) { //hdtn modifies bundle for next hop
                std::cerr << "error unable to render new bundle\n";
            }
            else {
                if (custodySignalRfc5050RenderedBundleView.m_renderedBundle.size()) {
                    const cbhe_eid_t & hdtnSrcEid = custodySignalRfc5050RenderedBundleView.m_primaryBlockView.header.m_sourceNodeId;
                    const uint64_t newCustodyIdFor5050CustodySignal = custodyIdAllocator.GetNextCustodyIdForNextHopCtebToSend(hdtnSrcEid);

                    //write custody signal to disk
                    BundleStorageManagerSession_WriteToDisk sessionWrite;
                    const uint64_t totalSegmentsRequired = bsm.Push(sessionWrite,
                        custodySignalRfc5050RenderedBundleView.m_primaryBlockView.header,
                        custodySignalRfc5050RenderedBundleView.m_renderedBundle.size());
                    //std::cout << "totalSegmentsRequired " << totalSegmentsRequired << "\n";
                    if (totalSegmentsRequired == 0) {
                        const std::string msg = "out of space for custody signal";
                        std::cerr << msg << "\n";
                        hdtn::Logger::getInstance()->logError("storage", msg);
                        return false;
                    }

                    const uint64_t totalBytesPushed = bsm.PushAllSegments(sessionWrite, custodySignalRfc5050RenderedBundleView.m_primaryBlockView.header,
                        newCustodyIdFor5050CustodySignal, (const uint8_t*)custodySignalRfc5050RenderedBundleView.m_renderedBundle.data(),
                        custodySignalRfc5050RenderedBundleView.m_renderedBundle.size());
                    if (totalBytesPushed != custodySignalRfc5050RenderedBundleView.m_renderedBundle.size()) {
                        const std::string msg = "totalBytesPushed != custodySignalRfc5050RenderedBundleView.m_renderedBundle.size()";
                        std::cerr << msg << "\n";
                        hdtn::Logger::getInstance()->logError("storage", msg);
                        return false;
                    }
                }
            }
        }

        //write bundle (modified by hdtn if custody requested) to disk
        BundleStorageManagerSession_WriteToDisk sessionWrite;
        uint64_t totalSegmentsRequired = bsm.Push(sessionWrite, primary, bv.m_renderedBundle.size());
        //std::cout << "totalSegmentsRequired " << totalSegmentsRequired << "\n";
        if (totalSegmentsRequired == 0) {
            std::cerr << "out of space\n";
            hdtn::Logger::getInstance()->logError("storage", "Out of space");
            return false;
        }
        //totalSegmentsStoredOnDisk += totalSegmentsRequired;
        //totalBytesWrittenThisTest += size;

        const uint64_t totalBytesPushed = bsm.PushAllSegments(sessionWrite, primary, newCustodyId, (const uint8_t*)bv.m_renderedBundle.data(), bv.m_renderedBundle.size());
        if (totalBytesPushed != bv.m_renderedBundle.size()) {
            const std::string msg = "totalBytesPushed != size";
            std::cerr << msg << "\n";
            hdtn::Logger::getInstance()->logError("storage", msg);
            return false;
        }
        return true;

    }
    else if (isBpVersion7) {
        BundleViewV7 bv;
        if (!bv.LoadBundle((uint8_t *)message->data(), message->size(), true, true)) { //invalid bundle
            std::cerr << "malformed bundle\n";
            return false;
        }
        const Bpv7CbhePrimaryBlock & primary = bv.m_primaryBlockView.header;
        finalDestEidReturned = primary.m_destinationEid;

        //write bundle
        BundleStorageManagerSession_WriteToDisk sessionWrite;
        uint64_t totalSegmentsRequired = bsm.Push(sessionWrite, primary, bv.m_renderedBundle.size());
        //std::cout << "totalSegmentsRequired " << totalSegmentsRequired << "\n";
        if (totalSegmentsRequired == 0) {
            std::cerr << "out of space\n";
            hdtn::Logger::getInstance()->logError("storage", "Out of space");
            return false;
        }
        //totalSegmentsStoredOnDisk += totalSegmentsRequired;
        //totalBytesWrittenThisTest += size;
        const uint64_t newCustodyId = custodyIdAllocator.GetNextCustodyIdForNextHopCtebToSend(primary.m_sourceNodeId);
        const uint64_t totalBytesPushed = bsm.PushAllSegments(sessionWrite, primary, newCustodyId, (const uint8_t*)bv.m_renderedBundle.data(), bv.m_renderedBundle.size());
        if (totalBytesPushed != bv.m_renderedBundle.size()) {
            const std::string msg = "totalBytesPushed != size";
            std::cerr << msg << "\n";
            hdtn::Logger::getInstance()->logError("storage", msg);
            return false;
        }
        return true;
    }
    else {
        std::cout << "error in ZmqStorageInterface Write: unsupported bundle version detected\n";
        return false;
    }

}

//return number of bytes to read for specified links
static uint64_t PeekOne(const std::vector<eid_plus_isanyserviceid_pair_t> & availableDestLinks, BundleStorageManagerBase & bsm) {
    BundleStorageManagerSession_ReadFromDisk  sessionRead;
    const uint64_t bytesToReadFromDisk = bsm.PopTop(sessionRead, availableDestLinks);
    if (bytesToReadFromDisk == 0) { //no more of these links to read
        return 0; //no bytes to read
    }

    //this link has a bundle in the fifo
    bsm.ReturnTop(sessionRead);
    return bytesToReadFromDisk;
}

static void CustomCleanupToEgressHdr(void *data, void *hint) {
    delete static_cast<hdtn::ToEgressHdr*>(hint);
}
static void CustomCleanupStorageAckHdr(void *data, void *hint) {
    delete static_cast<hdtn::StorageAckHdr*>(hint);
}
static void CustomCleanupStdVecUint8(void *data, void *hint) {
    //std::cout << "free " << static_cast<std::vector<uint8_t>*>(hint)->size() << std::endl;
    delete static_cast<std::vector<uint8_t>*>(hint);
}

static bool ReleaseOne_NoBlock(BundleStorageManagerSession_ReadFromDisk & sessionRead,
    const std::vector<eid_plus_isanyserviceid_pair_t> & availableDestLinks,
    zmq::socket_t *egressSock, BundleStorageManagerBase & bsm, const uint64_t maxBundleSizeToRead)
{
    //std::cout << "reading\n";
    const uint64_t bytesToReadFromDisk = bsm.PopTop(sessionRead, availableDestLinks);
    //std::cout << "bytesToReadFromDisk " << bytesToReadFromDisk << "\n";
    if (bytesToReadFromDisk == 0) { //no more of these links to read
        return false;
    }

    //this link has a bundle in the fifo

    //IF YOU DECIDE YOU DON'T WANT TO READ THE BUNDLE AFTER PEEKING AT IT (MAYBE IT'S TOO BIG RIGHT NOW)
    if (bytesToReadFromDisk > maxBundleSizeToRead) {
        std::cerr << "error: bundle to read from disk is too large right now" << std::endl;
        hdtn::Logger::getInstance()->logError("storage", "Error: bundle to read from disk is too large right now");
        bsm.ReturnTop(sessionRead);
        return false;
        //bytesToReadFromDisk = bsm.PopTop(sessionRead, availableDestLinks); //get it back
    }

    std::vector<uint8_t> * vecUint8BundleDataRawPointer = new std::vector<uint8_t>();
    const bool successReadAllSegments = bsm.ReadAllSegments(sessionRead, *vecUint8BundleDataRawPointer);
    zmq::message_t zmqBundleDataMessageWithDataStolen(vecUint8BundleDataRawPointer->data(), vecUint8BundleDataRawPointer->size(), CustomCleanupStdVecUint8, vecUint8BundleDataRawPointer);

    //std::cout << "totalBytesRead " << totalBytesRead << "\n";
    if (!successReadAllSegments) {
        std::cout << "error: unable to read all segments from disk\n";
        hdtn::Logger::getInstance()->logError("storage", "error: unable to read all segments from disk");
        return false;
    }

    //force natural/64-bit alignment
    hdtn::ToEgressHdr * toEgressHdr = new hdtn::ToEgressHdr();
    zmq::message_t zmqMessageToEgressHdrWithDataStolen(toEgressHdr, sizeof(hdtn::ToEgressHdr), CustomCleanupToEgressHdr, toEgressHdr);

    //memset 0 not needed because all values set below
    toEgressHdr->base.type = HDTN_MSGTYPE_EGRESS;
    toEgressHdr->base.flags = 0;
    toEgressHdr->finalDestEid = sessionRead.catalogEntryPtr->destEid;
    toEgressHdr->hasCustody = sessionRead.catalogEntryPtr->HasCustody();
    toEgressHdr->isCutThroughFromIngress = 0;
    toEgressHdr->custodyId = sessionRead.custodyId;

    if (!egressSock->send(std::move(zmqMessageToEgressHdrWithDataStolen), zmq::send_flags::sndmore | zmq::send_flags::dontwait)) {
        std::cout << "error: zmq could not send" << std::endl;
        hdtn::Logger::getInstance()->logError("storage", "Error: zmq could not send");
        bsm.ReturnTop(sessionRead);
        return false;
    }
    if (!egressSock->send(std::move(zmqBundleDataMessageWithDataStolen), zmq::send_flags::dontwait)) {
        std::cout << "error: zmq could not send bundle" << std::endl;
        hdtn::Logger::getInstance()->logError("storage", "Error: zmq could not send bundle");
        bsm.ReturnTop(sessionRead);
        return false;
    }
    /*
    //if you're happy with the bundle data you read back, then officially remove it from the disk
    if (deleteFromDiskNow) {
        bool successRemoveBundle = bsm.RemoveReadBundleFromDisk(sessionRead);
        if (!successRemoveBundle) {
            std::cout << "error freeing bundle from disk\n";
            return false;
        }
    }
        */

    return true;

}

static void PrintReleasedLinks(const std::set<eid_plus_isanyserviceid_pair_t> & availableDestLinksSet) {
    std::string strVals = "[";
    for (std::set<eid_plus_isanyserviceid_pair_t>::const_iterator it = availableDestLinksSet.cbegin(); it != availableDestLinksSet.cend(); ++it) {
        if (it->second) { //any service id
            strVals += Uri::GetIpnUriStringAnyServiceNumber((*it).first.nodeId) + ", ";
        }
        else { //fully qualified
            strVals += Uri::GetIpnUriString((*it).first.nodeId, (*it).first.serviceId) + ", ";
        }
    }
    strVals += "]";
    std::cout << "Currently Releasing Final Destination Eids: " << strVals << std::endl;
    hdtn::Logger::getInstance()->logNotification("storage", "Currently Releasing Final Destination Eids: " + strVals);
}

void ZmqStorageInterface::ThreadFunc() {
    BundleStorageManagerSession_ReadFromDisk sessionRead; //reuse this due to expensive heap allocation
    BundleViewV6 custodySignalRfc5050RenderedBundleView;
    custodySignalRfc5050RenderedBundleView.m_frontBuffer.reserve(2000);
    custodySignalRfc5050RenderedBundleView.m_backBuffer.reserve(2000);
    CustodyIdAllocator custodyIdAllocator;
    CustodyTimers custodyTimers(boost::posix_time::milliseconds(m_hdtnConfig.m_retransmitBundleAfterNoCustodySignalMilliseconds));
    const bool IS_HDTN_ACS_AWARE = m_hdtnConfig.m_isAcsAware;
    const uint64_t ACS_MAX_FILLS_PER_ACS_PACKET = m_hdtnConfig.m_acsMaxFillsPerAcsPacket;

    static const boost::posix_time::time_duration ACS_SEND_PERIOD = boost::posix_time::milliseconds(m_hdtnConfig.m_acsSendPeriodMilliseconds);
    CustodyTransferManager ctm(IS_HDTN_ACS_AWARE, M_HDTN_EID_CUSTODY.nodeId, M_HDTN_EID_CUSTODY.serviceId);
    std::cout << "[storage-worker] Worker thread starting up." << std::endl;
    hdtn::Logger::getInstance()->logNotification("storage", "Worker thread starting up");

    std::unique_ptr<BundleStorageManagerBase> bsmPtr;
    if (m_hdtnConfig.m_storageConfig.m_storageImplementation == "stdio_multi_threaded") {
        std::cout << "[ZmqStorageInterface] Initializing BundleStorageManagerMT ... " << std::endl;
        hdtn::Logger::getInstance()->logNotification("storage", "[ZmqStorageInterface] Initializing BundleStorageManagerMT ... ");
        bsmPtr = boost::make_unique<BundleStorageManagerMT>(boost::make_shared<StorageConfig>(m_hdtnConfig.m_storageConfig));
    }
    else if (m_hdtnConfig.m_storageConfig.m_storageImplementation == "asio_single_threaded") {
        std::cout << "[ZmqStorageInterface] Initializing BundleStorageManagerAsio ... " << std::endl;
        hdtn::Logger::getInstance()->logNotification("storage", "[ZmqStorageInterface] Initializing BundleStorageManagerAsio ... ");
        bsmPtr = boost::make_unique<BundleStorageManagerAsio>(boost::make_shared<StorageConfig>(m_hdtnConfig.m_storageConfig));
    }
    else {
        std::cerr << "error in hdtn::ZmqStorageInterface::ThreadFunc: invalid storage implementation " << m_hdtnConfig.m_storageConfig.m_storageImplementation << std::endl;
        return;
    }
    BundleStorageManagerBase & bsm = *bsmPtr;
    bsm.Start();

    typedef std::set<uint64_t> custodyid_set_t;
    typedef std::map<cbhe_eid_t, custodyid_set_t> finaldesteid_opencustids_map_t;

    std::vector<eid_plus_isanyserviceid_pair_t> availableDestLinksNotCloggedVec;
    availableDestLinksNotCloggedVec.reserve(100); //todo
    std::vector<eid_plus_isanyserviceid_pair_t> availableDestLinksCloggedVec;
    availableDestLinksCloggedVec.reserve(100); //todo

    m_totalBundlesErasedFromStorageNoCustodyTransfer = 0;
    m_totalBundlesErasedFromStorageWithCustodyTransfer = 0;
    m_totalBundlesSentToEgressFromStorage = 0;
    m_numRfc5050CustodyTransfers = 0;
    m_numAcsCustodyTransfers = 0;
    m_numAcsPacketsReceived = 0;
    std::size_t totalEventsAllLinksClogged = 0;
    std::size_t totalEventsNoDataInStorageForAvailableLinks = 0;
    std::size_t totalEventsDataInStorageForCloggedLinks = 0;
    std::size_t numCustodyTransferTimeouts = 0;

    std::set<eid_plus_isanyserviceid_pair_t> availableDestLinksSet;
    finaldesteid_opencustids_map_t finalDestEidToOpenCustIdsMap;

    static constexpr std::size_t minBufSizeBytesReleaseMessages = sizeof(uint64_t) +
        ((sizeof(hdtn::IreleaseStartHdr) > sizeof(hdtn::IreleaseStopHdr)) ? sizeof(hdtn::IreleaseStartHdr) : sizeof(hdtn::IreleaseStopHdr));
    uint64_t rxBufReleaseMessagesAlign64[minBufSizeBytesReleaseMessages / sizeof(uint64_t)];

    zmq::pollitem_t pollItems[4] = {
        {m_zmqPullSock_boundEgressToConnectingStoragePtr->handle(), 0, ZMQ_POLLIN, 0},
        {m_zmqPullSock_boundIngressToConnectingStoragePtr->handle(), 0, ZMQ_POLLIN, 0},
        {m_zmqSubSock_boundReleaseToConnectingStoragePtr->handle(), 0, ZMQ_POLLIN, 0},
        {m_telemetrySockPtr->handle(), 0, ZMQ_POLLIN, 0}
    };
    static const long DEFAULT_BIG_TIMEOUT_POLL = 250;
    long timeoutPoll = DEFAULT_BIG_TIMEOUT_POLL; //0 => no blocking
    boost::posix_time::ptime acsSendNowExpiry = boost::posix_time::microsec_clock::universal_time() + ACS_SEND_PERIOD;
    m_threadStartupComplete = true;
    while (m_running) {
        int rc = 0;
        try {
            rc = zmq::poll(pollItems, 4, timeoutPoll);
        }
        catch (zmq::error_t & e) {
            std::cout << "caught zmq::error_t in hdtn::ZmqStorageInterface::ThreadFunc: " << e.what() << std::endl;
            continue;
        }
        if (rc > 0) {
            if (pollItems[0].revents & ZMQ_POLLIN) { //from egress sock
                hdtn::EgressAckHdr egressAckHdr;
                const zmq::recv_buffer_result_t res = m_zmqPullSock_boundEgressToConnectingStoragePtr->recv(zmq::mutable_buffer(&egressAckHdr, sizeof(egressAckHdr)), zmq::recv_flags::none);
                if (!res) {
                    std::cerr << "[storage-worker] EgressAckHdr not received" << std::endl;
                    hdtn::Logger::getInstance()->logError("storage", "[storage-worker] EgressAckHdr not received");
                    continue;
                }
                else if ((res->truncated()) || (res->size != sizeof(hdtn::EgressAckHdr))) {
                    std::cerr << "[storage-worker] EgressAckHdr wrong size received" << std::endl;
                    hdtn::Logger::getInstance()->logError("storage", "[storage-worker] EgressAckHdr wrong size received");
                    continue;
                }
                else if (egressAckHdr.base.type != HDTN_MSGTYPE_EGRESS_ACK_TO_STORAGE) {
                    std::cerr << "[storage-worker] EgressAckHdr not type HDTN_MSGTYPE_EGRESS_ACK_TO_STORAGE, got " << egressAckHdr.base.type << std::endl;
                    hdtn::Logger::getInstance()->logError("storage", "[storage-worker] EgressAckHdr not type HDTN_MSGTYPE_EGRESS_ACK_TO_STORAGE");
                    continue;
                }
                custodyid_set_t & custodyIdSet = finalDestEidToOpenCustIdsMap[egressAckHdr.finalDestEid];
                custodyid_set_t::iterator it = custodyIdSet.find(egressAckHdr.custodyId);
                if (it != custodyIdSet.end()) {
                    if (egressAckHdr.deleteNow) { //custody not requested, so don't wait on a custody signal to delete the bundle
                        bool successRemoveBundle = bsm.RemoveReadBundleFromDisk(egressAckHdr.custodyId);
                        if (!successRemoveBundle) {
                            std::cout << "error freeing bundle from disk\n";
                            hdtn::Logger::getInstance()->logError("storage", "Error freeing bundle from disk");
                        }
                        else {
                            ++m_totalBundlesErasedFromStorageNoCustodyTransfer;
                        }
                    }
                    custodyIdSet.erase(it);
                }
            }
            if (pollItems[1].revents & ZMQ_POLLIN) { //from ingress bundle data
                hdtn::ToStorageHdr toStorageHeader;
                const zmq::recv_buffer_result_t res = m_zmqPullSock_boundIngressToConnectingStoragePtr->recv(zmq::mutable_buffer(&toStorageHeader, sizeof(hdtn::ToStorageHdr)), zmq::recv_flags::none);
                if (!res) {
                    std::cerr << "error in hdtn::ZmqStorageInterface::ThreadFunc (from ingress bundle data) message hdr not received" << std::endl;
                    hdtn::Logger::getInstance()->logError("storage", "error in hdtn::ZmqStorageInterface::ThreadFunc (from ingress bundle data) message hdr not received");
                    continue;
                }
                else if ((res->truncated()) || (res->size != sizeof(hdtn::ToStorageHdr))) {
                    std::cerr << "error in hdtn::ZmqStorageInterface::ThreadFunc (from ingress bundle data) rhdr.size() != sizeof(hdtn::ToStorageHdr)" << std::endl;
                    hdtn::Logger::getInstance()->logError("storage", "error in hdtn::ZmqStorageInterface::ThreadFunc (from ingress bundle data) rhdr.size() != sizeof(hdtn::ToStorageHdr)");
                    continue;
                }
                else if (toStorageHeader.base.type == HDTN_MSGTYPE_STORAGE_ADD_OPPORTUNISTIC_LINK) {
                    const uint64_t nodeId = toStorageHeader.ingressUniqueId;
                    const std::string msg = "finalDestEid ("
                        + Uri::GetIpnUriStringAnyServiceNumber(nodeId)
                        + ") will be released from storage";
                    std::cout << msg << std::endl;
                    hdtn::Logger::getInstance()->logNotification("storage", msg);
                    availableDestLinksSet.emplace(cbhe_eid_t(nodeId, 0), true); //true => any service id.. 0 is don't care
                    PrintReleasedLinks(availableDestLinksSet);
                    continue;
                }
                else if (toStorageHeader.base.type == HDTN_MSGTYPE_STORAGE_REMOVE_OPPORTUNISTIC_LINK) {
                    const uint64_t nodeId = toStorageHeader.ingressUniqueId;
                    const std::string msg = "finalDestEid ("
                        + Uri::GetIpnUriStringAnyServiceNumber(nodeId)
                        + ") will STOP being released from storage";
                    std::cout << msg << std::endl;
                    hdtn::Logger::getInstance()->logNotification("storage", msg);
                    availableDestLinksSet.erase(eid_plus_isanyserviceid_pair_t(cbhe_eid_t(nodeId, 0), true)); //true => any service id.. 0 is don't care
                    PrintReleasedLinks(availableDestLinksSet);
                    continue;
                }
                else if (toStorageHeader.base.type != HDTN_MSGTYPE_STORE) {
                    std::cerr << "error in hdtn::ZmqStorageInterface::ThreadFunc (from ingress bundle data) message type not HDTN_MSGTYPE_STORE" << std::endl;
                    hdtn::Logger::getInstance()->logError("storage", "error in hdtn::ZmqStorageInterface::ThreadFunc (from ingress bundle data) message type not HDTN_MSGTYPE_STORE");
                    continue;
                }

                storageStats.inBytes += sizeof(hdtn::ToStorageHdr);
                ++storageStats.inMsg;

                zmq::message_t zmqBundleDataReceived;
                if (!m_zmqPullSock_boundIngressToConnectingStoragePtr->recv(zmqBundleDataReceived, zmq::recv_flags::none)) {
                    std::cerr << "error in hdtn::ZmqStorageInterface::ThreadFunc (from ingress bundle data) message not received" << std::endl;
                    hdtn::Logger::getInstance()->logError("storage", "error in hdtn::ZmqStorageInterface::ThreadFunc (from ingress bundle data) message not received");
                    continue;
                }
                storageStats.inBytes += zmqBundleDataReceived.size();

                cbhe_eid_t finalDestEidReturnedFromWrite;
                Write(&zmqBundleDataReceived, bsm, custodyIdAllocator, ctm, custodyTimers, custodySignalRfc5050RenderedBundleView, finalDestEidReturnedFromWrite, this);

                //send ack message to ingress
                //force natural/64-bit alignment
                hdtn::StorageAckHdr * storageAckHdr = new hdtn::StorageAckHdr();
                zmq::message_t zmqMessageStorageAckHdrWithDataStolen(storageAckHdr, sizeof(hdtn::StorageAckHdr), CustomCleanupStorageAckHdr, storageAckHdr);

                //memset 0 not needed because all values set below
                storageAckHdr->base.type = HDTN_MSGTYPE_STORAGE_ACK_TO_INGRESS;
                storageAckHdr->base.flags = 0;
                storageAckHdr->error = 0;
                storageAckHdr->finalDestEid = finalDestEidReturnedFromWrite;
                storageAckHdr->ingressUniqueId = toStorageHeader.ingressUniqueId;

                if (!m_zmqPushSock_connectingStorageToBoundIngressPtr->send(std::move(zmqMessageStorageAckHdrWithDataStolen), zmq::send_flags::dontwait)) {
                    std::cout << "error: zmq could not send ingress an ack from storage" << std::endl;
                    hdtn::Logger::getInstance()->logError("storage", "Error: zmq could not send ingress an ack from storage");
                }
            }
            if (pollItems[2].revents & ZMQ_POLLIN) { //release messages
                //force this hdtn message struct to be aligned on a 64-byte boundary using zmq::mutable_buffer
                const zmq::recv_buffer_result_t res = m_zmqSubSock_boundReleaseToConnectingStoragePtr->recv(
                    zmq::mutable_buffer(rxBufReleaseMessagesAlign64, minBufSizeBytesReleaseMessages), zmq::recv_flags::none);
                if (!res) {
                    std::cerr << "[schedule release] message not received" << std::endl;
                    hdtn::Logger::getInstance()->logError("storage", "[schedule release] message not received");
                    continue;
                }
                else if (res->size < sizeof(hdtn::CommonHdr)) {
                    std::cerr << "[schedule release] res->size < sizeof(hdtn::CommonHdr)" << std::endl;
                    hdtn::Logger::getInstance()->logError("storage", "[schedule release] res->size < sizeof(hdtn::CommonHdr)");
                    continue;
                }

                std::cout << "release message received\n";
                hdtn::Logger::getInstance()->logNotification("storage", "Release message received");
                hdtn::CommonHdr *commonHdr = (hdtn::CommonHdr *)rxBufReleaseMessagesAlign64;
                if (commonHdr->type == HDTN_MSGTYPE_ILINKUP) {
                    if (res->size != sizeof(hdtn::IreleaseStartHdr)) {
                        std::cerr << "[schedule release] res->size != sizeof(hdtn::IreleaseStartHdr)" << std::endl;
                        hdtn::Logger::getInstance()->logError("storage", "[schedule release] res->size != sizeof(hdtn::IreleaseStartHdr)");
                        continue;
                    }

                    hdtn::IreleaseStartHdr * iReleaseStartHdr = (hdtn::IreleaseStartHdr *)rxBufReleaseMessagesAlign64;
                    const std::string msg = "finalDestEid ("
                        + Uri::GetIpnUriString(iReleaseStartHdr->finalDestinationEid.nodeId, iReleaseStartHdr->finalDestinationEid.serviceId)
                        + ") will be released from storage";
                    std::cout << msg << std::endl;
                    hdtn::Logger::getInstance()->logNotification("storage", msg);
                    availableDestLinksSet.emplace(iReleaseStartHdr->finalDestinationEid, false); //false => fully qualified service id
                    availableDestLinksSet.emplace(iReleaseStartHdr->nextHopEid, false);
		}
                else if (commonHdr->type == HDTN_MSGTYPE_ILINKDOWN) {
                    if (res->size != sizeof(hdtn::IreleaseStopHdr)) {
                        std::cerr << "[schedule release] res->size != sizeof(hdtn::IreleaseStopHdr)" << std::endl;
                        hdtn::Logger::getInstance()->logError("storage", "[schedule release] res->size != sizeof(hdtn::IreleaseStopHdr)");
                        continue;
                    }

                    hdtn::IreleaseStopHdr * iReleaseStoptHdr = (hdtn::IreleaseStopHdr *)rxBufReleaseMessagesAlign64;
                    const std::string msg = "finalDestEid (" + boost::lexical_cast<std::string>(iReleaseStoptHdr->finalDestinationEid.nodeId) + ","
                        + boost::lexical_cast<std::string>(iReleaseStoptHdr->finalDestinationEid.serviceId) + ") will STOP BEING released from storage";
                    std::cout << msg << std::endl;
                    hdtn::Logger::getInstance()->logNotification("storage", msg);
                    availableDestLinksSet.erase(eid_plus_isanyserviceid_pair_t(iReleaseStoptHdr->finalDestinationEid, false)); //false => fully qualified service id
                    availableDestLinksSet.erase(eid_plus_isanyserviceid_pair_t(iReleaseStoptHdr->nextHopEid, false)); //false => fully qualified service id

		}
                PrintReleasedLinks(availableDestLinksSet);
            }
            if (pollItems[3].revents & ZMQ_POLLIN) { //telemetry messages
                hdtn::CommonHdr commonHeader;
                const zmq::recv_buffer_result_t res = m_telemetrySockPtr->recv(zmq::mutable_buffer(&commonHeader, sizeof(hdtn::CommonHdr)), zmq::recv_flags::none);
                if (!res) {
                    std::cerr << "error in hdtn::ZmqStorageInterface::ThreadFunc (from telemetry sock) message hdr not received" << std::endl;
                    hdtn::Logger::getInstance()->logError("storage", "error in hdtn::ZmqStorageInterface::ThreadFunc (from telemetry sock) message hdr not received");
                    continue;
                }
                else if ((res->truncated()) || (res->size != sizeof(hdtn::CommonHdr))) {
                    std::cerr << "error in hdtn::ZmqStorageInterface::ThreadFunc (from telemetry sock) size != sizeof(hdtn::CommonHdr)" << std::endl;
                    hdtn::Logger::getInstance()->logError("storage", "error in hdtn::ZmqStorageInterface::ThreadFunc (from telemetry sock) size != sizeof(hdtn::CommonHdr)");
                    continue;
                }

                switch (commonHeader.type) {
                case HDTN_MSGTYPE_CSCHED_REQ:
                    break;
                case HDTN_MSGTYPE_CTELEM_REQ:
                    break;
                }
            }
        }

        const boost::posix_time::ptime nowPtime = boost::posix_time::microsec_clock::universal_time();
        if ((acsSendNowExpiry <= nowPtime) || (ctm.GetLargestNumberOfFills() > ACS_MAX_FILLS_PER_ACS_PACKET)) {
            //std::cout << "send acs, fills = " << ctm.GetLargestNumberOfFills() << "\n";
            //test with generate all
            std::list<BundleViewV6> newAcsRenderedBundleViewList;
            if (ctm.GenerateAllAcsBundlesAndClear(newAcsRenderedBundleViewList)) {
                for(std::list<BundleViewV6>::iterator it = newAcsRenderedBundleViewList.begin(); it != newAcsRenderedBundleViewList.end(); ++it) {
                    WriteAcsBundle(bsm, custodyIdAllocator, it->m_primaryBlockView.header, it->m_frontBuffer);
                }
            }
            acsSendNowExpiry = nowPtime + ACS_SEND_PERIOD;
        }

        uint64_t custodyIdExpiredAndNeedingResent;
        while (custodyTimers.PollOneAndPopAnyExpiredCustodyTimer(custodyIdExpiredAndNeedingResent, nowPtime)) {
            if (bsm.ReturnCustodyIdToAwaitingSend(custodyIdExpiredAndNeedingResent)) {
                ++numCustodyTransferTimeouts;
            }
            else {
                std::cerr << "error unable to return expired custody id " << custodyIdExpiredAndNeedingResent << " to the awaiting send\n";
            }
        }

        //Send and maintain a maximum of 5 unacked bundles (per flow id) to Egress.
        //When a bundle is acked from egress using the head segment Id, the bundle is deleted from disk and a new bundle can be sent.
        static const uint64_t maxBundleSizeToRead = UINT64_MAX;// 65535 * 10;
        if (availableDestLinksSet.empty()) {
            timeoutPoll = DEFAULT_BIG_TIMEOUT_POLL;
        }
        else {
            availableDestLinksNotCloggedVec.resize(0);
            availableDestLinksCloggedVec.resize(0);
            for (std::set<eid_plus_isanyserviceid_pair_t>::const_iterator it = availableDestLinksSet.cbegin(); it != availableDestLinksSet.cend(); ++it) {
                //std::cout << "flow " << flowId << " sz " << flowIdToOpenSessionsMap[flowId].size() << std::endl;
                //const bool isAnyServiceId = it->second;
                if (finalDestEidToOpenCustIdsMap[it->first].size() < 5) {//finaldesteid_opencustids_map_t finalDestEidToOpenCustIdsMap;
                    availableDestLinksNotCloggedVec.push_back(*it);
                }
                else {
                    availableDestLinksCloggedVec.push_back(*it);
                }
            }
            if (availableDestLinksNotCloggedVec.size() > 0) {
                if (ReleaseOne_NoBlock(sessionRead, availableDestLinksNotCloggedVec, m_zmqPushSock_connectingStorageToBoundEgressPtr.get(), bsm, maxBundleSizeToRead)) { //true => (successfully sent to egress)
                    if (finalDestEidToOpenCustIdsMap[sessionRead.catalogEntryPtr->destEid].insert(sessionRead.custodyId).second) {
                        if (sessionRead.catalogEntryPtr->HasCustody()) {
                            custodyTimers.StartCustodyTransferTimer(sessionRead.catalogEntryPtr->destEid, sessionRead.custodyId);
                        }
                        timeoutPoll = 0; //no timeout as we need to keep feeding to egress
                        ++m_totalBundlesSentToEgressFromStorage;
                    }
                    else {
                        std::cerr << "could not insert custody id into finalDestEidToOpenCustIdsMap\n";
                    }
                }
                else if (PeekOne(availableDestLinksCloggedVec, bsm) > 0) { //data available in storage for clogged links
                    timeoutPoll = 1; //shortest timeout 1ms as we wait for acks
                    ++totalEventsDataInStorageForCloggedLinks;
                }
                else { //no data in storage for any available links
                    timeoutPoll = DEFAULT_BIG_TIMEOUT_POLL;
                    ++totalEventsNoDataInStorageForAvailableLinks;
                }
            }
            else { //all links clogged up and need acks
                timeoutPoll = 1; //shortest timeout 1ms as we wait for acks
                ++totalEventsAllLinksClogged;
            }
        }

        //}

        /*hdtn::flow_stats stats = m_storeFlow.stats();
        m_workerStats.flow.disk_wbytes = stats.disk_wbytes;
        m_workerStats.flow.disk_wcount = stats.disk_wcount;
        m_workerStats.flow.disk_rbytes = stats.disk_rbytes;
        m_workerStats.flow.disk_rcount = stats.disk_rcount;*/

    }
    std::cout << "totalEventsAllLinksClogged: " << totalEventsAllLinksClogged << std::endl;
    std::cout << "totalEventsNoDataInStorageForAvailableLinks: " << totalEventsNoDataInStorageForAvailableLinks << std::endl;
    std::cout << "totalEventsDataInStorageForCloggedLinks: " << totalEventsDataInStorageForCloggedLinks << std::endl;
    std::cout << "m_numRfc5050CustodyTransfers: " << m_numRfc5050CustodyTransfers << std::endl;
    std::cout << "m_numAcsCustodyTransfers: " << m_numAcsCustodyTransfers << std::endl;
    std::cout << "m_numAcsPacketsReceived: " << m_numAcsPacketsReceived << std::endl;
    std::cout << "m_totalBundlesErasedFromStorageNoCustodyTransfer: " << m_totalBundlesErasedFromStorageNoCustodyTransfer << std::endl;
    std::cout << "m_totalBundlesErasedFromStorageWithCustodyTransfer: " << m_totalBundlesErasedFromStorageWithCustodyTransfer << std::endl;
    std::cout << "numCustodyTransferTimeouts: " << numCustodyTransferTimeouts << std::endl;
    hdtn::Logger::getInstance()->logInfo("storage", "totalEventsAllLinksClogged: " +
        std::to_string(totalEventsAllLinksClogged));
    hdtn::Logger::getInstance()->logInfo("storage", "totalEventsNoDataInStorageForAvailableLinks: " +
        std::to_string(totalEventsNoDataInStorageForAvailableLinks));
    hdtn::Logger::getInstance()->logInfo("storage", "totalEventsDataInStorageForCloggedLinks: " +
        std::to_string(totalEventsDataInStorageForCloggedLinks));
}

std::size_t ZmqStorageInterface::GetCurrentNumberOfBundlesDeletedFromStorage() {
    return m_totalBundlesErasedFromStorageNoCustodyTransfer + m_totalBundlesErasedFromStorageWithCustodyTransfer;
}