No OneTemporary
Actions

Authored By

Unknown

Size

34 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/src/job/asio-promise-handler.hpp b/src/job/asio-promise-handler.hpp
	index 83626b4..6f64c5d 100644
	--- a/src/job/asio-promise-handler.hpp
	+++ b/src/job/asio-promise-handler.hpp
	@@ -1,210 +1,210 @@
	/*
	* This file is part of libkazv.
	* SPDX-FileCopyrightText: 2021 Tusooa Zhu <tusooa@kazv.moe>
	* SPDX-License-Identifier: AGPL-3.0-or-later
	*/

	#pragma once

	#include <boost/asio.hpp>

	#include <promise-interface.hpp>

	namespace Kazv
	{
	template<class Exec, class T>
	class AsioPromise;

	namespace detail
	{
	template<class Exec>
	struct AsioPromiseHelper
	{
	template<class T>
	using PromiseType = AsioPromise<Exec, T>;
	};

	struct IdentityFunc
	{
	template<class T>
	constexpr T &&operator()(T &&t) const {
	return std::forward<T>(t);
	}
	};
	}

	template<class Exec, class T>
	class AsioPromise : public AbstractPromise<detail::AsioPromiseHelper<Exec>::template PromiseType, T>
	{
	using BaseT = AbstractPromise<detail::AsioPromiseHelper<Exec>::template PromiseType, T>;

	template<class FuncT, class PromiseT, class ResolveT>
	struct WaitHelper
	{
	void wait() const {
	if (p.ready()) {
	auto res = func(p.get());
	using ResT = decltype(res);
	if constexpr (isPromise<ResT>) {
	boost::asio::post(
	executor,
	[w=WaitHelper<detail::IdentityFunc, ResT, ResolveT>{
	executor,
	res,
	detail::IdentityFunc{},
	resolve,
	}]() mutable {
	w.wait();
	});
	} else {
	resolve(res);
	}
	} else {
	boost::asio::post(
	executor,
	[*this]() mutable {
	wait();
	}
	);
	}
	}

	Exec executor;
	PromiseT p;
	FuncT func;
	ResolveT resolve;
	};

	struct ResolveHelper
	{
	template<class ValT>
	void operator()(ValT val) const {
	using ResT = std::decay_t<decltype(val)>;
	if constexpr (isPromise<ResT>) {
	auto w = WaitHelper<detail::IdentityFunc, ResT, ResolveHelper>{
	executor,
	val,
	detail::IdentityFunc{},
	*this
	};
	w.wait();
	} else {
	p->set_value(std::move(val));
	}
	}

	Exec executor;
	std::shared_ptr<std::promise<T>> p;
	};
	public:
	AsioPromise(Exec executor, T value)
	: BaseT(this)
	, m_executor(std::move(executor)) {
	std::promise<T> p;
	m_val = p.get_future().share();
	p.set_value(std::move(value));
	}

	template<class Func>
	AsioPromise(Exec executor, Func &&callback)
	: BaseT(this)
	, m_executor(std::move(executor)) {
	auto p = std::make_shared<std::promise<T>>();
	m_val = p->get_future().share();
	auto resolve = ResolveHelper{m_executor, p};

	boost::asio::post(
	m_executor,
	[=, callback=std::forward<Func>(callback),
	resolve=std::move(resolve),
	- guard=boost::asio::executor_work_guard(m_executor)]() {
	+ guard=boost::asio::make_work_guard(m_executor)]() {
	callback(resolve);
	});
	}

	// FuncT: (DataT) -> AnotherDataT
	// where AnotherDataT = PromiseThenResult<FuncT, typename BaseT::DataT>
	template<class FuncT>
	auto then(FuncT &&func)
	-> AsioPromise<Exec,
	PromiseThenResult<FuncT, typename BaseT::DataT>> {
	return AsioPromise<Exec,
	PromiseThenResult<FuncT, typename BaseT::DataT>>(
	m_executor,
	[=, func=std::forward<FuncT>(func), *this](auto resolve) {
	auto waitHelper = WaitHelper<std::decay_t<FuncT>,
	AsioPromise,
	std::decay_t<decltype(resolve)>>{
	m_executor,
	*this,
	func,
	resolve
	};
	waitHelper.wait();
	});
	}

	bool ready() const {
	return m_val.wait_for(std::chrono::seconds(0)) == std::future_status::ready;
	}

	T get() const {
	return m_val.get();
	}
	private:
	Exec m_executor;
	std::shared_future<T> m_val;
	};

	template<class Exec>
	class AsioPromiseHandler : public PromiseInterface<AsioPromiseHandler<Exec>,
	detail::AsioPromiseHelper<Exec>::template PromiseType>
	{
	using BaseT = PromiseInterface<AsioPromiseHandler<Exec>,
	detail::AsioPromiseHelper<Exec>::template PromiseType>;
	public:
	template<class T>
	using PromiseT = AsioPromise<Exec, T>;

	AsioPromiseHandler(Exec executor)
	: BaseT(this)
	, m_executor(executor)
	{}

	AsioPromiseHandler(const AsioPromiseHandler &that)
	: BaseT(this)
	, m_executor(that.m_executor)
	{}

	AsioPromiseHandler(AsioPromiseHandler &&that)
	: BaseT(this)
	, m_executor(std::move(that.m_executor))
	{}

	AsioPromiseHandler &operator=(const AsioPromiseHandler &that) {
	m_executor = that.m_executor;
	return *this;
	}

	AsioPromiseHandler &operator=(AsioPromiseHandler &&that) {
	m_executor = std::move(that.m_executor);
	return *this;
	}

	template<class T, class FuncT>
	PromiseT<T> create(FuncT &&func) {
	return PromiseT<T>(m_executor, std::forward<FuncT>(func));
	}

	template<class T>
	PromiseT<T> createResolved(T val) {
	return PromiseT<T>(m_executor, std::move(val));
	}

	private:
	Exec m_executor;
	};

	template<class Exec>
	AsioPromiseHandler(Exec) -> AsioPromiseHandler<Exec>;
	}
	diff --git a/src/job/cprjobhandler.cpp b/src/job/cprjobhandler.cpp
	index ed43a34..efacb88 100644
	--- a/src/job/cprjobhandler.cpp
	+++ b/src/job/cprjobhandler.cpp
	@@ -1,487 +1,487 @@
	/*
	* This file is part of libkazv.
	* SPDX-FileCopyrightText: 2020 Tusooa Zhu
	* SPDX-License-Identifier: AGPL-3.0-or-later
	*/

	#include <libkazv-config.hpp>

	#include <algorithm>
	#include <deque>

	#include <zug/into_vector.hpp>
	#include <zug/transducer/map.hpp>

	#include <cpr.h>
	#include <callback.h>

	#include <asio-std-file-handler.hpp>

	#include "cprjobhandler.hpp"
	#include "debug.hpp"

	namespace {
	using namespace Kazv;
	struct SyncFileStream
	{
	using DataT = FileContent;

	inline SyncFileStream(std::string filename, FileOpenMode mode)
	: m_stream(filename,
	(mode == FileOpenMode::Read
	? std::ios_base::in
	: std::ios_base::out)
	\| std::ios_base::binary)
	{}

	template<class Callback>
	void read(int maxSize, Callback readCallback) {
	auto buf = std::vector<char>(maxSize, '\0');
	auto data = FileContent{};
	auto actualSize = int{};

	try {
	m_stream.read(buf.data(), maxSize);

	actualSize = m_stream.gcount();

	data = FileContent(buf.begin(), buf.begin() + actualSize);
	} catch (const std::exception &) {
	readCallback(FileOpRetCode::Error, FileContent{});
	return;
	}

	if (actualSize > 0) {
	readCallback(FileOpRetCode::Success, data);
	} else if (m_stream.eof()) {
	readCallback(FileOpRetCode::Eof, FileContent{});
	} else {
	readCallback(FileOpRetCode::Error, FileContent{});
	}
	}

	template<class Callback>
	void write(DataT data, Callback writeCallback) {
	for (auto c : data) {
	m_stream.put(c);
	if (m_stream.bad()) {
	writeCallback(FileOpRetCode::Error, 0);
	return;
	}
	}

	writeCallback(FileOpRetCode::Success, data.size());
	}

	std::fstream m_stream;
	};

	struct SyncFileProvider
	{
	using FileStreamT = SyncFileStream;

	std::string filename;

	FileStreamT getStream(FileOpenMode mode) const {
	return FileStreamT(filename, mode);
	}
	};

	struct SyncFileHandler
	{
	using FileProviderT = SyncFileProvider;
	FileProviderT getProviderFor(FileDesc desc) const {
	// assert(desc.name())
	return FileProviderT{desc.name().value()};
	}
	};
	}

	namespace Kazv
	{
	struct CprJobHandler::Private
	{
	CprJobHandler *q;
	boost::asio::io_context::executor_type executor;
	using TimerSP = std::shared_ptr<boost::asio::steady_timer>;
	using TimerSPList = std::vector<TimerSP>;
	using TimerMap = std::unordered_map<std::optional<std::string>, TimerSPList>;
	TimerMap timerMap;

	enum Status
	{
	Waiting,
	Running,
	};
	using Callback = std::function<void(Response)>;
	struct JobDesc
	{
	BaseJob job;
	Callback callback;
	Status status;
	};
	using JobQueue = std::deque<JobDesc>;
	using JobMap = std::unordered_map<std::string, JobQueue>;
	JobMap jobQueues;

	void submitImpl(BaseJob job, std::function<void(Response)> userCallback);

	void addToQueue(BaseJob job, Callback callback) {
	boost::asio::post(
	executor,
	[=] {
	// precondition: job has a queueId
	auto queueId = job.queueId().value();

	jobQueues[queueId].push_back(JobDesc{job, callback, Waiting});
	});
	}

	void clearQueue(std::string queueId) {
	boost::asio::post(
	executor,
	[=] { clearQueueImpl(queueId); });
	}

	void clearQueueImpl(std::string queueId) {
	Response fakeResponse;
	fakeResponse.statusCode = headFailureCancelledStatusCode;
	fakeResponse.body = JsonBody(
	json{ {"errcode", headFailureCancelledErrorCode},
	{"error", headFailureCancelledErrorMsg} }
	);
	kzo.job.dbg() << "clearQueueImpl called with " << queueId << std::endl;

	for (auto [job, callback, status] : jobQueues[queueId]) {
	kzo.job.dbg() << "this job is " << (status == Waiting ? "Waiting" : "Running") << std::endl;
	if (status == Waiting) {
	// Run callback in a different thread, just as in submitImpl().
	q->async([=] { callback(job.genResponse(fakeResponse)); } );
	}
	// if status is Running, the callback is already called
	}
	jobQueues[queueId].clear();
	}

	void popJob(std::string queueId) {
	boost::asio::post(
	executor,
	[=] { popJobImpl(queueId); });
	}

	void popJobImpl(std::string queueId) {
	if (! jobQueues[queueId].empty()) {
	jobQueues[queueId].pop_front();
	}
	}

	void monitorQueues() {
	boost::asio::post(
	executor,
	[=] {
	// precondition: job has a queueId
	for (auto &[queueId, queue] : jobQueues) { // need to change queue
	if (! queue.empty()) {
	auto [job, callback, status] = queue.front();
	if (status == Waiting) {
	queue.front().status = Running;
	submitImpl(
	job,
	[=](Response r) { // in new thread
	callback(r);

	if (! r.success() // should be enough for now
	&& job.queuePolicy() == CancelFutureIfFailed) {
	clearQueue(queueId); // in executor thread
	} else {
	popJob(queueId); // in executor thread
	}
	});
	}
	}
	}
	});
	}

	void addTimerToMap(TimerSP timer, std::optional<std::string> timerId) {
	boost::asio::post(
	executor,
	[=] {
	timerMap[timerId].push_back(timer);
	});
	}

	void clearTimer(TimerSP timer, std::optional<std::string> timerId) {
	boost::asio::post(
	executor,
	[=] {
	std::remove(timerMap[timerId].begin(), timerMap[timerId].end(), timer);
	});
	}

	void cancelAllTimers(std::optional<std::string> timerId) {
	boost::asio::post(
	executor,
	[=] {
	cancelAllTimersImpl(timerId);
	});
	}

	void cancelAllTimersImpl(std::optional<std::string> timerId) {
	auto timers = timerMap[timerId];
	timerMap.erase(timerId);
	for (auto timer : timers) {
	timer->cancel();
	}
	}

	void intervalTimerCallback(TimerSP timer,
	std::function<void()> func,
	int ms,
	const boost::system::error_code &error) {
	if (! error) {
	func();
	auto dur = boost::asio::chrono::milliseconds(ms);
	timer->expires_at(timer->expiry() + dur);
	timer->async_wait(
	[=](const boost::system::error_code &error) {
	intervalTimerCallback(timer, func, ms, error);
	});
	}
	}
	};

	CprJobHandler::CprJobHandler(boost::asio::io_context::executor_type executor)
	: m_d(new Private{this, std::move(executor), Private::TimerMap{}, Private::JobMap{}})
	{
	setInterval(
	[=] {
	m_d->monitorQueues();
	},
	50, // ms
	"-queue-monitor");
	}

	CprJobHandler::~CprJobHandler() = default;

	void CprJobHandler::async(std::function<void()> func)
	{
	- std::thread([func=std::move(func), guard=boost::asio::executor_work_guard(m_d->executor)]() {
	+ std::thread([func=std::move(func), guard=boost::asio::make_work_guard(m_d->executor)]() {
	func();
	}).detach();
	}

	void CprJobHandler::setTimeout(std::function<void()> func, int ms, std::optional<std::string> timerId)
	{
	auto timer=std::make_shared<boost::asio::steady_timer>(
	m_d->executor, boost::asio::chrono::milliseconds(ms));

	m_d->addTimerToMap(timer, timerId);

	timer->async_wait(
	[=, timer=timer](const boost::system::error_code &error){
	if (! error) {
	func();
	this->m_d->clearTimer(timer, timerId);
	}
	});
	}

	void CprJobHandler::setInterval(std::function<void()> func, int ms, std::optional<std::string> timerId)
	{
	auto dur = boost::asio::chrono::milliseconds(ms);
	auto timer = std::make_shared<boost::asio::steady_timer>(m_d->executor, dur);

	m_d->addTimerToMap(timer, timerId);

	timer->async_wait(
	[=](const boost::system::error_code &error) {
	m_d->intervalTimerCallback(timer, func, ms, error);
	});
	}

	void CprJobHandler::cancel(std::string timerId)
	{
	m_d->cancelAllTimers(timerId);
	}

	void CprJobHandler::submit(BaseJob job, std::function<void(Response)> userCallback)
	{
	if (job.queueId()) {
	m_d->addToQueue(job, userCallback);
	} else {
	m_d->submitImpl(job, userCallback);
	}
	}

	void CprJobHandler::Private::submitImpl(BaseJob job, std::function<void(Response)> userCallback)
	{
	cpr::Url url{job.url()};
	auto streamUpload = std::holds_alternative<FileDesc>(job.requestBody());
	cpr::Body body;
	BaseJob::Header origHeader = job.requestHeader();
	cpr::Header header(origHeader.get().begin(), origHeader.get().end());

	auto readCallback = std::function<bool(char *, size_t &)>{};
	auto writeCallback = std::function<bool(std::string)>{};

	if (! std::holds_alternative<FileDesc>(job.requestBody())) {
	body = std::get<BytesBody>(job.requestBody());
	} else {
	auto fileDesc = std::get<FileDesc>(job.requestBody());
	auto typeOpt = fileDesc.contentType();
	if (typeOpt) {
	header.insert_or_assign("Content-Type", typeOpt.value());
	kzo.job.dbg() << "Content-Type is " << header["Content-Type"] << std::endl;
	}
	auto fh = SyncFileHandler{};
	auto provider = fileDesc.provider(fh);
	auto stream = std::make_shared<FileStream>(provider.getStream());

	readCallback =
	[stream](char *buffer, size_t &length) -> bool {
	bool retval = true;
	kzo.job.dbg() << "A buffer of length " << length << " requested." << std::endl;
	stream->read(length,
	[&](FileOpRetCode code, FileContent data) {
	if (code == FileOpRetCode::Success) {
	std::copy_n(data.begin(), data.size(), buffer);
	kzo.job.dbg() << "Read file successful, read " << data.size() << " bytes." << std::endl;
	length = data.size();
	} else if (code == FileOpRetCode::Eof) {
	kzo.job.dbg() << "Got eof." << std::endl;
	length = 0;
	} else {
	kzo.job.dbg() << "Got error reading file." << std::endl;
	length = 0;
	retval = false;
	}
	});
	return retval;
	};
	}

	if (job.responseFile()) {
	auto fileDesc = job.responseFile().value();

	auto fh = SyncFileHandler{};
	auto provider = fileDesc.provider(fh);
	auto stream = std::make_shared<FileStream>(provider.getStream(FileOpenMode::Write));

	writeCallback =
	[stream](std::string buffer) -> bool {
	bool retval = true;
	kzo.job.dbg() << "A buffer of length " << buffer.size() << " responded." << std::endl;
	FileContent data(buffer.begin(), buffer.end());
	stream->write(data,
	[&](FileOpRetCode code, int num) {
	if (code == FileOpRetCode::Success) {
	kzo.job.dbg() << "Write file successful, wrote " << num << " bytes." << std::endl;
	} else {
	kzo.job.dbg() << "Got error writing file." << std::endl;
	retval = false;
	}
	});
	return retval;
	};
	}

	cpr::Parameters params;
	BaseJob::Query query = job.requestQuery();
	BaseJob::ReturnType returnType = job.returnType();
	BaseJob::Method method = job.requestMethod();

	if (! query.empty()) {
	// from cpr/parameters.cpp
	cpr::CurlHolder holder;
	for (const auto kv : query) {
	std::string key = kv.first;
	std::string value = kv.second;
	params.Add(cpr::Parameter(std::move(key), std::move(value)));
	}
	}

	auto callback = [returnType, responseFile=job.responseFile()](cpr::Response r) -> Response {
	Body body;

	if (responseFile) {
	body = responseFile.value();
	} else if (returnType == BaseJob::ReturnType::Json) {
	try {
	body = BaseJob::JsonBody(std::move(json::parse(r.text)));
	} catch (const json::exception &e) {
	// the response is not valid json
	body = r.text;
	kzo.job.dbg() << "body is not correct json: " << e.what() << std::endl;
	}
	} else {
	body = r.text;
	}

	return { r.status_code,
	body,
	BaseJob::Header(r.header.begin(), r.header.end()),
	{} // extraData, will be added in genResponse
	};
	};


	std::shared_future<Response> res = std::visit(lager::visitor{
	[=](BaseJob::Get) {
	if (readCallback) {
	return cpr::GetCallback(callback, url, cpr::ReadCallback(readCallback), header, params);
	} else if (writeCallback) {
	return cpr::GetCallback(callback, url, cpr::WriteCallback(writeCallback), header, body, params);
	} else {
	return cpr::GetCallback(callback, url, header, body, params);
	}
	},
	[=](BaseJob::Post) {
	if (readCallback) {
	return cpr::PostCallback(callback, url, cpr::ReadCallback(readCallback), header, params);
	} else if (writeCallback) {
	return cpr::PostCallback(callback, url, cpr::WriteCallback(writeCallback), header, body, params);
	} else {
	return cpr::PostCallback(callback, url, header, body, params);
	}
	},
	[=](BaseJob::Put) {
	if (readCallback) {
	return cpr::PutCallback(callback, url, cpr::ReadCallback(readCallback), header, params);
	} else if (writeCallback) {
	return cpr::PutCallback(callback, url, cpr::WriteCallback(writeCallback), header, body, params);
	} else {
	return cpr::PutCallback(callback, url, header, body, params);
	}
	},
	[=](BaseJob::Delete) {
	if (readCallback) {
	return cpr::DeleteCallback(callback, url, cpr::ReadCallback(readCallback), header, params);
	} else if (writeCallback) {
	return cpr::DeleteCallback(callback, url, cpr::WriteCallback(writeCallback), header, body, params);
	} else {
	return cpr::DeleteCallback(callback, url, header, body, params);
	}
	}
	}, method).share();

	q->async([=]() {
	userCallback(job.genResponse(res.get()));
	});
	}

	void CprJobHandler::stop()
	{
	boost::asio::post(
	m_d->executor,
	[=] {
	auto ids = zug::into_vector(
	zug::map([](auto i) { return i.first; }),
	m_d->timerMap);
	for (auto id : ids) {
	m_d->cancelAllTimersImpl(id);
	}
	m_d->jobQueues.clear();
	});
	}
	}
	diff --git a/src/tests/client/thread-safety-test.cpp b/src/tests/client/thread-safety-test.cpp
	index 0471d94..cdb76ce 100644
	--- a/src/tests/client/thread-safety-test.cpp
	+++ b/src/tests/client/thread-safety-test.cpp
	@@ -1,208 +1,208 @@
	/*
	* This file is part of libkazv.
	* SPDX-FileCopyrightText: 2021 Tusooa Zhu <tusooa@kazv.moe>
	* SPDX-License-Identifier: AGPL-3.0-or-later
	*/

	#include <libkazv-config.hpp>

	#include <catch2/catch.hpp>

	#include <lager/event_loop/boost_asio.hpp>
	#include <boost/asio.hpp>

	#include <sdk.hpp>
	#include <cprjobhandler.hpp>
	#include <lagerstoreeventemitter.hpp>
	#include <asio-promise-handler.hpp>

	#ifdef KAZV_USE_THREAD_SAFETY_HELPER
	using namespace Kazv;

	TEST_CASE("Thread-safety verification should work", "[client][thread-safety]")
	{
	auto io = boost::asio::io_context{};
	auto jh = Kazv::CprJobHandler{io.get_executor()};
	auto ee = Kazv::LagerStoreEventEmitter(lager::with_boost_asio_event_loop{io.get_executor()});

	auto sdk = Kazv::makeSdk(
	Kazv::SdkModel{},
	jh,
	ee,
	Kazv::AsioPromiseHandler{io.get_executor()},
	zug::identity
	);

	auto ctx = sdk.context();

	REQUIRE(ctx.has<EventLoopThreadIdKeeper>());

	auto client = sdk.client();

	bool thrown = false;
	try {
	client.userId();
	} catch (const ThreadNotMatchException &) {
	thrown = true;
	}
	REQUIRE(! thrown); // event loop has not started yet, so do not throw

	- boost::asio::executor_work_guard g(io.get_executor());
	+ auto g = boost::asio::make_work_guard(io.get_executor());

	std::thread([&io] { io.run(); }).detach();

	// wait till the event loop thread is logged
	std::this_thread::sleep_for(std::chrono::milliseconds(100));

	thrown = false;
	std::thread([=, &thrown] {
	try {
	client.userId();
	} catch (const ThreadNotMatchException &) {
	thrown = true;
	}
	}).join();
	REQUIRE(thrown);

	ctx.createResolvedPromise({})
	.then([](auto &&) {})
	.then([&client](auto &&) {
	client.userId();
	});

	std::this_thread::sleep_for(std::chrono::milliseconds(100));

	thrown = false;
	try {
	client.userId();
	} catch (const ThreadNotMatchException &) {
	thrown = true;
	}
	REQUIRE(thrown);
	}

	TEST_CASE("Thread-safety verification from secondary root should work", "[client][thread-safety]")
	{
	auto io = boost::asio::io_context{};
	auto jh = Kazv::CprJobHandler{io.get_executor()};
	auto ee = Kazv::LagerStoreEventEmitter(lager::with_boost_asio_event_loop{io.get_executor()});

	auto sdk = Kazv::makeSdk(
	Kazv::SdkModel{},
	jh,
	ee,
	Kazv::AsioPromiseHandler{io.get_executor()},
	zug::identity
	);

	auto ctx = sdk.context();

	- boost::asio::executor_work_guard guard(io.get_executor());
	+ auto guard = boost::asio::make_work_guard(io.get_executor());

	std::thread([&io] { io.run(); }).detach();

	auto io2 = boost::asio::io_context{};

	auto sr = sdk.createSecondaryRoot(lager::with_boost_asio_event_loop{io2.get_executor()});

	auto client2 = sdk.clientFromSecondaryRoot(sr);

	bool thrown = false;
	std::thread([client2, &thrown] {
	try {
	client2.userId();
	} catch(const ThreadNotMatchException &) {
	thrown = true;
	}
	}).join();

	REQUIRE(thrown);

	thrown = false;
	std::thread([&client2, &thrown] {
	try {
	client2.userId();
	} catch(const ThreadNotMatchException &) {
	thrown = true;
	}
	}).join();

	REQUIRE(thrown);

	io2.run();
	}

	TEST_CASE("Thread-safety verification should properly verify toEventLoop calls", "[client][thread-safety]")
	{
	auto io = boost::asio::io_context{};
	auto jh = Kazv::CprJobHandler{io.get_executor()};
	auto ee = Kazv::LagerStoreEventEmitter(lager::with_boost_asio_event_loop{io.get_executor()});

	auto sdk = Kazv::makeSdk(
	Kazv::SdkModel{},
	jh,
	ee,
	Kazv::AsioPromiseHandler{io.get_executor()},
	zug::identity
	);

	auto ctx = sdk.context();

	auto io2 = boost::asio::io_context{};

	auto sr = sdk.createSecondaryRoot(lager::with_boost_asio_event_loop{io2.get_executor()});

	auto client2 = sdk.clientFromSecondaryRoot(sr);

	auto room2 = client2.room("!foo:example.org");

	ctx.createResolvedPromise({})
	- .then([client2, room2, guard=boost::asio::executor_work_guard(io.get_executor())](auto &&) {
	+ .then([client2, room2, guard=boost::asio::make_work_guard(io.get_executor())](auto &&) {
	bool thrown = false;

	try {
	client2.userId();
	} catch (const ThreadNotMatchException &) {
	thrown = true;
	}
	REQUIRE(thrown);

	thrown = false;
	try {
	room2.roomId();
	} catch (const ThreadNotMatchException &) {
	thrown = true;
	}
	REQUIRE(thrown);
	});

	ctx.createResolvedPromise({})
	.then([client2=client2.toEventLoop(), room2=room2.toEventLoop(),
	- guard=boost::asio::executor_work_guard(io.get_executor())](auto &&) {
	+ guard=boost::asio::make_work_guard(io.get_executor())](auto &&) {
	bool thrown2 = false;

	try {
	client2.userId();
	} catch (const ThreadNotMatchException &) {
	thrown2 = true;
	}
	REQUIRE(!thrown2);

	thrown2 = false;

	try {
	room2.roomId();
	} catch (const ThreadNotMatchException &) {
	thrown2 = true;
	}
	REQUIRE(!thrown2);
	});

	std::thread([&io] { io.run(); }).detach();

	io2.run();

	std::this_thread::sleep_for(std::chrono::milliseconds(100));
	}

	#endif
	diff --git a/src/tests/event-emitter-test.cpp b/src/tests/event-emitter-test.cpp
	index 3d16bfd..89b742b 100644
	--- a/src/tests/event-emitter-test.cpp
	+++ b/src/tests/event-emitter-test.cpp
	@@ -1,83 +1,83 @@
	/*
	* This file is part of libkazv.
	* SPDX-FileCopyrightText: 2020 Tusooa Zhu <tusooa@kazv.moe>
	* SPDX-License-Identifier: AGPL-3.0-or-later
	*/

	#include <libkazv-config.hpp>

	#include <chrono>

	#include <catch2/catch.hpp>

	#include <eventemitter/lagerstoreeventemitter.hpp>

	#include <lager/event_loop/boost_asio.hpp>


	using namespace Kazv;

	TEST_CASE("Event emitter should work normally", "[eventemitter]")
	{
	boost::asio::io_context ioContext;

	LagerStoreEventEmitter ee{lager::with_boost_asio_event_loop{ioContext.get_executor()}};

	auto watchable = ee.watchable();
	int counter = 0;

	watchable.after<SyncSuccessful>(
	[&](auto) {
	++counter;
	});

	SECTION("Event handlers should be run properly") {
	ee.emit(SyncSuccessful{});
	ee.emit(SyncFailed{});

	ioContext.run();

	REQUIRE(counter == 1);
	}

	SECTION("Event handlers should be called when every event fires off") {
	for (int i = 0; i < 100; ++i) {
	ee.emit(SyncSuccessful{});
	}

	ioContext.run();

	REQUIRE(counter == 100);
	}

	SECTION("Handler should be disconnected once watchable is destroyed") {
	int counter2 = 0;

	- auto guard = boost::asio::executor_work_guard(ioContext.get_executor());
	+ auto guard = boost::asio::make_work_guard(ioContext.get_executor());
	auto thread = std::thread([&] { ioContext.run(); });

	ee.emit(SyncFailed{});
	std::this_thread::sleep_for(std::chrono::milliseconds{100});

	{
	auto watchable2 = ee.watchable();
	watchable2.after<SyncFailed>(
	[&](auto) {
	++counter2;
	});

	ee.emit(SyncSuccessful{});
	ee.emit(SyncFailed{});
	std::this_thread::sleep_for(std::chrono::milliseconds{100});
	}

	ee.emit(SyncFailed{});

	guard.reset();

	thread.join();

	REQUIRE(counter == 1);
	REQUIRE(counter2 == 1);
	}
	}

File Metadata

Mime Type: text/x-diff
Expires: Mon, Nov 2, 10:31 AM (2 h, 52 m)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1804515
Default Alt Text: (34 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions