test/requests/main.cpp - main - Ant010ff/ffpp

test/requests/main.cpp
274 строки10 KB
Ant010ff
Demo code update; exception-related refactoring (FFPP_TRY / FFPP_CATCH).
13 июн 2026, 15:39
13 июн 2026, 15:39d23feb5
#include "../common/common.hpp"

//Declaring broker policy with wstring identifiers in current test (just for demo purposes).
struct Policy : CommonPolicy {
	struct ResourcePolicy : ffpp::DefaultResourcePolicy {
		struct IdProvider {
			using Id = std::wstring;
			using Hash = std::hash<Id>;
			
			static constexpr Id InvalidId() { return { }; }	
			
			static Id GenerateId() { 
				//Return string representation of default 64-bit integer identifier.
				#if !(defined(_GLIBCXX_RELEASE) && (_GLIBCXX_RELEASE <= 12))
					return std::format(L"{{{:#018X}}}", ffpp::IdProvider::GenerateId());
				#else
					std::wstringstream wss;
					wss << L"0x" << std::setw(16) << std::uppercase << std::setfill(L'0') << std::hex << ffpp::IdProvider::GenerateId();
					return wss.str();
				#endif
			}
		};

		using IdProviderT = IdProvider;
	};

	using ResourcePolicyT = ResourcePolicy;
	using IdProviderT = ResourcePolicyT::IdProviderT;
	using Id = IdProviderT::Id;
};

using BrokerT = ffpp::FunctionalBroker<Policy>;
using ActorT = ffpp::FunctionalActor<true, Policy>;

void TestSimpleReception() {
	using ResponseT = std::vector<int>; //user defined request context (response, or any kind of async result)
	auto [_, sgComplete] = ffpp::BeginJoin();
	Accent::Put("Simulating simple async reception...");
	BrokerT::Id const 
		idReceive1 = BrokerT::Receive<ResponseT>(
		[sgComplete] (ResponseT const& vResponse) {
				Info::Put("Received first:", VectorToString(vResponse));
			}
		).Complete()->GetTaskId(),
		idReceive2 = BrokerT::Receive<ResponseT>(
			[sgComplete = std::move(sgComplete)] (ResponseT const& vResponse) {
				Info::Put("Received second:", VectorToString(vResponse));
			}
		).Complete()->GetTaskId();
	Info::Put("Waiting for response...");
	BrokerT::Emit(idReceive1, ResponseT { 1, 2, 3, 4, 5 });
	BrokerT::Emit(idReceive2, ResponseT { 6, 7, 8, 9, 0 });
}

void TestSimpleJunctions() {
	using namespace std::literals;

	auto [ufJoin, sfComplete] = ffpp::BeginJoinList();
	
	Accent::Put("Simulating basic async junctions (Emit before Receive)...");
	
	BrokerT::Id const 
		idGroup { L"TestSimpleJunctions.Group" }
		, idJunction1 { L"TestSimpleJunctions.Junction1" }
		, idJunction2 { L"TestSimpleJunctions.Junction2" }
	;
	
	*sfComplete += [idGroup] () { BrokerT::Finalize(idGroup); };

	using ContextT = std::tuple<std::vector<int>, std::shared_ptr<ffpp::Finalizer<>>>;
	BrokerT::Emit<ffpp::Flags::Junction>(idJunction1, ContextT { { 0, 9, 8, 7, 6 }, sfComplete });
	BrokerT::Emit<ffpp::Flags::Junction>(idJunction1, ContextT { { 5, 4, 3, 2, 1 }, sfComplete });
	Info::Put("Waiting for handlers...");
	SleepUpTo(2s);
	BrokerT::Receive<ffpp::Flags::Junction, ContextT>(
		idJunction1,
		idGroup,
		[] (auto const& vCtx, auto /*sfComplete*/) mutable {
			Info::Put("Received context 1:", VectorToString(vCtx));
		}
	);
	BrokerT::Emit<ffpp::Flags::Junction>(idJunction1, ContextT { { 5, 4, 3 }, sfComplete });
	BrokerT::Emit<ffpp::Flags::Junction>(idJunction1, ContextT { { 2, 1, 0 }, sfComplete });
	BrokerT::Emit<ffpp::Flags::Junction>(idJunction1, ContextT { { 0, 1, 2 }, sfComplete });

	BrokerT::Emit<ffpp::Flags::Junction>(idJunction2, ContextT { { 5, 4, 3, 2, 1 }, sfComplete });
	BrokerT::Emit<ffpp::Flags::Junction>(idJunction2, ContextT { { 0, 9, 8, 7, 6 }, sfComplete });
	BrokerT::Receive<ffpp::Flags::Junction, ContextT>(
		idJunction2,
		idGroup,
		[] (auto const& vCtx, auto /*sfComplete*/) mutable {
			Info::Put("Received context 2:", VectorToString(vCtx));
		}
	);
	BrokerT::Emit<ffpp::Flags::Junction>(idJunction2, ContextT { { 0, 1, 2 }, sfComplete });
	BrokerT::Emit<ffpp::Flags::Junction>(idJunction2, ContextT { { 2, 1, 0 }, sfComplete });
	BrokerT::Emit<ffpp::Flags::Junction>(idJunction2, ContextT { { 5, 4, 3 }, sfComplete });
	SleepUpTo(2s);
	BrokerT::Emit<ffpp::Flags::Junction>(idJunction2, ContextT { { 9, 0, 1 }, sfComplete });
	BrokerT::Emit<ffpp::Flags::Junction>(idJunction2, ContextT { { 6, 7, 8 }, sfComplete });
	BrokerT::Emit<ffpp::Flags::Junction>(idJunction2, ContextT { { 3, 4, 5 }, sfComplete });

	Info::Put("Waiting for completion...");
	ffpp::FinishJoin(sfComplete, ufJoin);
	Info::Put("Done.");
}

void TestSimpleRequest() {
	using ResponseT = std::vector<int>; //user defined request context (response, or any kind of async result)
	auto [_, sgComplete] = ffpp::BeginJoin();
	Accent::Put("Simulating simple async request...");
	BrokerT::Request<ResponseT>(
		[] (BrokerT::Id idRequest) {
			auto spWorker = std::make_shared<ActorT>(BrokerT::GetPool());
			spWorker->Chain([idRequest] () {
				SimulateLatency(idRequest);
				BrokerT::Emit(idRequest, ResponseT { 1, 2, 3, 4, 5 });
			}).Complete([] (auto) { });
		}, 
		[] (ResponseT const& vResponse) {
			Info::Put("Received request response:", VectorToString(vResponse));
		}
	)
	.Chain([] () { 
		Info::Put("Additional request chain stage.");
	})
	.Complete([sgComplete = std::move(sgComplete)] (auto) { 
		Info::Put("Request chain completed.");
	});	
	Info::Put("Waiting for response...");
}

void TestMixedResponseRequest(bool bRepeat = true, ffpp::SharedFinalizerT<> const& spfExternal = nullptr) {
	using ResponseT = std::tuple<std::string, float, std::vector<int>>;
	auto [_, sgComplete] = ffpp::BeginJoinList();
	if(spfExternal) *sgComplete += [spfExternal] () { };
	Accent::Put("Simulating mixed response async request...");
	BrokerT::Request<ResponseT>(
		[] (BrokerT::Id idRequest) {
			auto spWorker = std::make_shared<ActorT>(BrokerT::GetPool());
			spWorker->Chain([idRequest] () {
				SimulateLatency(idRequest);
				BrokerT::Emit<ResponseT>(idRequest, { "Deferred string", 3.14f, { 6, 7, 8, 9, 0 } });
			});
			return spWorker;
		}, 
		[bRepeat, sgComplete] (auto const& strDeferred, auto fPi, auto const& vInts, auto const& spWorker) {
			if(bRepeat) spWorker->Chain([sgComplete] () { 
				TestMixedResponseRequest(false, sgComplete); 
			}).Complete();
			Info::Put("Request completed with response:", strDeferred, fPi, VectorToString(vInts));
		}
	).Complete([sgComplete = std::move(sgComplete)] (auto) { 
		Info::Put("Request chain completed.");
	});	
	Info::Put("Waiting for response...");
}

void TestMixedTupleResponseRequest() {
	using namespace std::literals;

	using ResponseT = std::tuple<std::string, float, std::vector<int>>;
	auto [_, sgComplete] = ffpp::BeginJoin();
	Accent::Put("Simulating mixed tuple response async request...");
	BrokerT::Request<ResponseT>(
		[] (BrokerT::Id idRequest) {
			auto spWorker = std::make_shared<ActorT>(BrokerT::GetPool());
			spWorker->Chain([idRequest] () {
				SimulateLatency(idRequest);
				BrokerT::Emit<ResponseT>(idRequest, { "Deferred string", 3.14f, { 6, 7, 8, 9, 0 } });
			});
			return std::tuple { spWorker, "Request id = "s + Log<>::Convert(idRequest) };
		}, 
		[] (auto const& strDeferred, auto fPi, auto const& vInts, auto const& spWorker, auto const& strInfo) {
			spWorker->Complete();
			Info::Put("Request completed with response:", strDeferred, fPi, VectorToString(vInts), strInfo);
		}
	).Complete([sgComplete = std::move(sgComplete)] (auto) { });	
	Info::Put("Waiting for response...");
}

template<bool t_bLatency>
void TestRequestLoop(ffpp::Concepts::Duration auto const& drnLoop) {
	using namespace std::literals;

	Accent::Put("Simulating async request loop", t_bLatency ? "(with latency)" : "(no latency)", "for", drnLoop, "...");
	
	auto [ugJoin, sgComplete] = ffpp::BeginJoin([] () {
		Info::Put("Request loop simulation completed.");
	});
	
	auto const idLoop = BrokerT::GenerateId();
	auto spWorker = std::make_shared<ActorT>(BrokerT::GetPool());
	BrokerT::Request<std::string>(
		idLoop,
		idLoop, 
		0,
		[spWorker] (BrokerT::Id idLoop) {
			spWorker->Chain([idLoop] () {
				if constexpr(t_bLatency) SimulateLatency(idLoop);
				if(BrokerT::TestId<std::string>(idLoop)) { //optional check, just a demo
					auto const strRandom = GenerateRandomString(8, 32);
					Trace::Put("Emitting string:", strRandom);
					if(!BrokerT::Emit(idLoop, std::move(strRandom))) {
						Error::Put("Failed to emit value for loop id", idLoop);
					}
				}
			});
			auto const strRandom = GenerateRandomString(2, 8);
			Trace::Put("Passing string:", strRandom);
			return strRandom;
		}, 
		[idLoop, sgComplete = std::move(sgComplete)] (auto const& strRequested, auto const& strPassed) {
			Trace::Put("Request completed with response", std::tuple { strRequested, strPassed });
			BrokerT::Request(idLoop);
		},
		[] (auto nCode, auto idLoop, auto /*idGroup*/) {
			if(nCode != -ETIMEDOUT) {
				Info::Put("Request:", idLoop, ", unknown failure code:", nCode, ", finalizing...");
				return false;
			}
			if constexpr(true) {
				Info::Put("Request:", idLoop, ", timed out, resuming...");
				BrokerT::Request(idLoop);
				return true;
			}
			else {
				Info::Put("Request:", idLoop, ", timed out and finalized, waiting in main thread...");
				return false;
			}
		},
		1000ms
	);	
	
	std::this_thread::sleep_for(drnLoop);
	spWorker->Finalize();
	BrokerT::Finalize(idLoop);
	ffpp::FinishJoin(sgComplete, ugJoin);
}

int main([[maybe_unused]] int argc, [[maybe_unused]] char** argv) {
	FFPP_TRY {
		using namespace std::literals;
		Log<>::Init(
			algy::FilesystemOptions { 
				algy::verbose, 
				algy::c_bitDefaultCaps, 
				std::filesystem::path(argv[0]).parent_path() / "log" , 
				"requests",
				8 * 1024 * 1024
			},
			algy::ConsoleOptions { algy::debug, algy::c_bitDefaultCaps }
		);
		Accent::Put("Starting async requests test...");
		TestSimpleReception();
		TestSimpleJunctions();
		TestSimpleRequest();
		TestMixedResponseRequest();
		TestMixedTupleResponseRequest();
		TestRequestLoop<true>(30s);
		TestRequestLoop<false>(3s);
		BrokerT::Finalize(ffpp::Flags::Wait);
		Accent::Put("Finished.");
		return 0;
	}
	FFPP_CATCH(ffpp::Base::Exception, ex) {
		if(Log<>::IsValid()) Fatal::Put("Top level FFPP exception:", ex);
		return -2;
	}
	FFPP_CATCH(std::exception, ex) {
		if(Log<>::IsValid()) Fatal::Put("Top level exception:", ex);
		return -1;
	}
}//main