No OneTemporary
Actions

Size

41 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/src/client/client.cpp b/src/client/client.cpp
	index a2db61f..66282f5 100644
	--- a/src/client/client.cpp
	+++ b/src/client/client.cpp
	@@ -1,282 +1,322 @@
	/*
	* Copyright (C) 2020-2021 Tusooa Zhu <tusooa@kazv.moe>
	*
	* This file is part of libkazv.
	*
	* libkazv is free software: you can redistribute it and/or modify
	* it under the terms of the GNU Affero General Public License as
	* published by the Free Software Foundation, either version 3 of the
	* License, or (at your option) any later version.
	*
	* libkazv is distributed in the hope that it will be useful,
	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	* GNU Affero General Public License for more details.
	*
	* You should have received a copy of the GNU Affero General Public License
	* along with libkazv. If not, see <https://www.gnu.org/licenses/>.
	*/

	#include <libkazv-config.hpp>

	#include <filesystem>

	#include <lager/constant.hpp>

	#include "client.hpp"

	namespace Kazv
	{
	Client::Client(lager::reader<SdkModel> sdk,
	ContextT ctx, std::nullopt_t)
	: m_sdk(sdk)
	, m_client(sdk.map(&SdkModel::c))
	, m_ctx(std::move(ctx))
	{
	}

	Client::Client(lager::reader<SdkModel> sdk,
	ContextWithDepsT ctx)
	: m_sdk(sdk)
	, m_client(sdk.map(&SdkModel::c))
	, m_ctx(ctx)
	, m_deps(std::move(ctx))
	{
	}

	+ Client::Client(InEventLoopTag,
	+ ContextWithDepsT ctx)
	+ : m_sdk(std::nullopt)
	+ , m_client(std::nullopt)
	+ , m_ctx(ctx)
	+ , m_deps(std::move(ctx))
	+#ifdef KAZV_USE_THREAD_SAFETY_HELPER
	+ , KAZV_ON_EVENT_LOOP_VAR(true)
	+#endif
	+ {
	+ }
	+

	Room Client::room(std::string id) const
	{
	- return Room(m_sdk, lager::make_constant(id), m_ctx);
	+ return Room(sdkCursor(), lager::make_constant(id), m_ctx);
	}

	Room Client::roomByCursor(lager::reader<std::string> id) const
	{
	- return Room(m_sdk, id, m_ctx);
	+ return Room(sdkCursor(), id, m_ctx);
	}

	auto Client::passwordLogin(std::string homeserver, std::string username,
	std::string password, std::string deviceName) const
	-> PromiseT
	{
	auto p1 = m_ctx.dispatch(LoginAction{
	homeserver, username, password, deviceName});
	p1
	.then([*this](auto stat) {
	if (! stat.success()) {
	return;
	}
	// It is meaningless to wait for it in a Promise
	// that is never exposed to the user.
	startSyncing();
	});

	return p1;
	}

	auto Client::tokenLogin(std::string homeserver, std::string username,
	std::string token, std::string deviceId) const
	-> PromiseT
	{
	auto p1 = m_ctx.dispatch(TokenLoginAction{
	homeserver, username, token, deviceId});
	p1
	.then([*this](auto stat) {
	if (! stat.success()) {
	return;
	}
	startSyncing();
	});

	return p1;
	}

	auto Client::createRoom(RoomVisibility v,
	std::optional<std::string> name,
	std::optional<std::string> alias,
	immer::array<std::string> invite,
	std::optional<bool> isDirect,
	bool allowFederate,
	std::optional<std::string> topic,
	JsonWrap powerLevelContentOverride) const
	-> PromiseT
	{
	CreateRoomAction a;
	a.visibility = v;
	a.name = name;
	a.roomAliasName = alias;
	a.invite = invite;
	a.isDirect = isDirect;
	a.topic = topic;
	a.powerLevelContentOverride = powerLevelContentOverride;
	// Synapse won't buy it if we do not provide
	// a creationContent object.
	a.creationContent = json{
	{"m.federate", allowFederate}
	};
	return m_ctx.dispatch(std::move(a));
	}

	auto Client::joinRoomById(std::string roomId) const -> PromiseT
	{
	return m_ctx.dispatch(JoinRoomByIdAction{roomId});
	}

	auto Client::joinRoom(std::string roomId, immer::array<std::string> serverName) const
	-> PromiseT
	{
	return m_ctx.dispatch(JoinRoomAction{roomId, serverName});
	}

	auto Client::uploadContent(immer::box<Bytes> content,
	std::string uploadId,
	std::optional<std::string> filename,
	std::optional<std::string> contentType) const
	-> PromiseT
	{
	return m_ctx.dispatch(UploadContentAction{
	FileDesc(FileContent{content.get().begin(), content.get().end()}),
	filename, contentType, uploadId});
	}

	auto Client::uploadContent(FileDesc file) const
	-> PromiseT
	{
	auto basename = file.name()
	? std::optional(std::filesystem::path(file.name().value()).filename().native())
	: std::nullopt;
	return m_ctx.dispatch(UploadContentAction{
	file,
	// use only basename to prevent path info being leaked
	basename,
	file.contentType(),
	// uploadId unused
	std::string{}});
	}

	auto Client::downloadContent(std::string mxcUri, std::optional<FileDesc> downloadTo) const
	-> PromiseT
	{
	return m_ctx.dispatch(DownloadContentAction{mxcUri, downloadTo});
	}

	auto Client::downloadThumbnail(
	std::string mxcUri,
	int width,
	int height,
	std::optional<ThumbnailResizingMethod> method,
	std::optional<FileDesc> downloadTo) const
	-> PromiseT
	{
	return m_ctx.dispatch(DownloadThumbnailAction{mxcUri, width, height, method, std::nullopt, downloadTo});
	}

	auto Client::startSyncing() const -> PromiseT
	{
	+ KAZV_VERIFY_THREAD_ID();
	using namespace Kazv::CursorOp;

	if (+syncing()) {
	return m_ctx.createResolvedPromise(true);
	}

	auto p1 = m_ctx.createResolvedPromise(true)
	.then([*this](auto) {
	// post filters, if filters are incomplete
	- if ((+m_client[&ClientModel::initialSyncFilterId]).empty()
	- \|\| (+m_client[&ClientModel::incrementalSyncFilterId]).empty()) {
	+ if ((+clientCursor()[&ClientModel::initialSyncFilterId]).empty()
	+ \|\| (+clientCursor()[&ClientModel::incrementalSyncFilterId]).empty()) {
	return m_ctx.dispatch(PostInitialFiltersAction{});
	}
	return m_ctx.createResolvedPromise(true);
	})
	.then([*this](auto stat) {
	if (! stat.success()) {
	return m_ctx.createResolvedPromise(stat);
	}
	// Upload identity keys if we need to
	- if (+m_client[&ClientModel::crypto]
	- && ! +m_client[&ClientModel::identityKeysUploaded]) {
	+ if (+clientCursor()[&ClientModel::crypto]
	+ && ! +clientCursor()[&ClientModel::identityKeysUploaded]) {
	return m_ctx.dispatch(UploadIdentityKeysAction{});
	} else {
	return m_ctx.createResolvedPromise(true);
	}
	});

	p1
	- .then([*this](auto stat) {
	+ .then([m_ctx=m_ctx](auto stat) {
	m_ctx.dispatch(SetShouldSyncAction{true});
	return stat;
	})
	.then([*this](auto stat) {
	if (stat.success()) {
	syncForever();
	}
	});

	return p1;
	}

	auto Client::syncForever(std::optional<int> retryTime) const -> void
	{
	+ KAZV_VERIFY_THREAD_ID();
	+
	// assert (m_deps);
	using namespace CursorOp;

	- bool isInitialSync = ! (+m_client[&ClientModel::syncToken]).has_value();
	+ bool isInitialSync = ! (+clientCursor()[&ClientModel::syncToken]).has_value();

	- bool shouldSync = +m_client[&ClientModel::shouldSync];
	+ bool shouldSync = +clientCursor()[&ClientModel::shouldSync];

	if (! shouldSync) {
	return;
	}
	//

	auto syncRes = m_ctx.dispatch(SyncAction{});

	auto uploadOneTimeKeysRes = syncRes
	.then([*this](auto stat) {
	if (! stat.success()) {
	return m_ctx.createResolvedPromise(stat);
	}
	- bool hasCrypto{+m_client[&ClientModel::crypto]};
	+ bool hasCrypto{+clientCursor()[&ClientModel::crypto]};
	auto p1 = hasCrypto
	? m_ctx.dispatch(GenerateAndUploadOneTimeKeysAction{})
	: m_ctx.createResolvedPromise(true);
	return p1;
	});

	auto queryKeysRes = syncRes
	.then([*this, isInitialSync](auto stat) {
	if (! stat.success()) {
	return m_ctx.createResolvedPromise(stat);
	}
	- bool hasCrypto{+m_client[&ClientModel::crypto]};
	+ bool hasCrypto{+clientCursor()[&ClientModel::crypto]};
	return hasCrypto
	? m_ctx.dispatch(QueryKeysAction{isInitialSync})
	: m_ctx.createResolvedPromise(true);
	});

	m_ctx.promiseInterface()
	.all(std::vector<PromiseT>{uploadOneTimeKeysRes, queryKeysRes})
	.then([*this, retryTime](auto stat) {
	if (stat.success()) {
	syncForever(); // reset retry time
	} else {
	- auto firstRetryTime = +m_client[&ClientModel::firstRetryMs];
	- auto retryTimeFactor = +m_client[&ClientModel::retryTimeFactor];
	- auto maxRetryTime = +m_client[&ClientModel::maxRetryMs];
	+ auto firstRetryTime = +clientCursor()[&ClientModel::firstRetryMs];
	+ auto retryTimeFactor = +clientCursor()[&ClientModel::retryTimeFactor];
	+ auto maxRetryTime = +clientCursor()[&ClientModel::maxRetryMs];
	auto curRetryTime = retryTime ? retryTime.value() : firstRetryTime;
	if (curRetryTime > maxRetryTime) { curRetryTime = maxRetryTime; }
	auto nextRetryTime = curRetryTime * retryTimeFactor;

	kzo.client.warn() << "Sync failed, retrying in " << curRetryTime << "ms" << std::endl;
	auto &jh = getJobHandler(m_deps.value());
	jh.setTimeout([*this, nextRetryTime]() { syncForever(nextRetryTime); },
	curRetryTime);
	}
	});
	}

	void Client::stopSyncing() const
	{
	m_ctx.dispatch(SetShouldSyncAction{false});
	}
	+
	+ lager::reader<ClientModel> Client::clientCursor() const
	+ {
	+ KAZV_VERIFY_THREAD_ID();
	+
	+ if (m_client.has_value()) {
	+ return m_client.value();
	+ } else {
	+ assert(m_deps.has_value());
	+ return lager::get<SdkModelCursorKey>(m_deps.value())->map(&SdkModel::c);
	+ }
	+ }
	+
	+ const lager::reader<SdkModel> &Client::sdkCursor() const
	+ {
	+ KAZV_VERIFY_THREAD_ID();
	+
	+ if (m_sdk.has_value()) {
	+ return m_sdk.value();
	+ } else {
	+ assert(m_deps.has_value());
	+ return *(lager::get<SdkModelCursorKey>(m_deps.value()));
	+ }
	+ }
	+
	}
	diff --git a/src/client/client.hpp b/src/client/client.hpp
	index 2878e40..b4061b3 100644
	--- a/src/client/client.hpp
	+++ b/src/client/client.hpp
	@@ -1,386 +1,405 @@
	/*
	* Copyright (C) 2020-2021 Tusooa Zhu <tusooa@kazv.moe>
	*
	* This file is part of libkazv.
	*
	* libkazv is free software: you can redistribute it and/or modify
	* it under the terms of the GNU Affero General Public License as
	* published by the Free Software Foundation, either version 3 of the
	* License, or (at your option) any later version.
	*
	* libkazv is distributed in the hope that it will be useful,
	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	* GNU Affero General Public License for more details.
	*
	* You should have received a copy of the GNU Affero General Public License
	* along with libkazv. If not, see <https://www.gnu.org/licenses/>.
	*/

	#pragma once
	#include <libkazv-config.hpp>

	#include <lager/reader.hpp>
	#include <immer/box.hpp>
	#include <immer/map.hpp>
	#include <immer/flex_vector.hpp>
	#include <immer/flex_vector_transient.hpp>

	#include "sdk-model.hpp"
	#include "client/client-model.hpp"
	#include "client/actions/content.hpp"
	#include "sdk-model-cursor-tag.hpp"

	#include "room/room.hpp"

	namespace Kazv
	{
	/**
	* Represent a Matrix client.
	*
	* If the Client is constructed from a cursor originated from
	* a root whose event loop is on thread A, then we say that
	* the Client belongs to thread A. If the Client is not constructed
	* from a cursor, then we say that the Client belongs to the thread
	* where the event loop of the context runs.
	*
	* All methods in this class that take a cursor only take a cursor
	* on the same thread as the Client. All methods in this class that
	* return a cursor will return a cursor on the same thread as the Client.
	*
	* All methods in this class must be run on the same thread as the
	* the Client. If the Client is not constructed from a cursor,
	* copy-constructing another Client from this is safe from any thread.
	* If the Client is constructed from a cursor, copy-constructing another
	* Client is safe only from the same thread as this Client.
	*/
	class Client
	{
	public:
	using ActionT = ClientAction;

	using DepsT = lager::deps<JobInterface &, EventInterface &, SdkModelCursorKey
	#ifdef KAZV_USE_THREAD_SAFETY_HELPER
	, EventLoopThreadIdKeeper &
	#endif

	>;
	using ContextT = Context<ActionT>;
	using ContextWithDepsT = Context<ActionT, DepsT>;

	using PromiseT = SingleTypePromise<DefaultRetType>;
	+
	+ struct InEventLoopTag {};
	/**
	* Constructor.
	*
	* Construct the client. Without Deps support.
	*
	* @warning You should not use this directly. Use
	* Sdk::client() instead.
	*/
	Client(lager::reader<SdkModel> sdk,
	ContextT ctx, std::nullopt_t);

	/**
	* Constructor.
	*
	* Construct the client, with Deps support.
	*
	* This enables startSyncing() to work properly.
	*
	* @warning You should not use this directly. Use
	* Sdk::client() instead.
	*/
	Client(lager::reader<SdkModel> sdk,
	ContextWithDepsT ctx);

	+ /**
	+ * Constructor.
	+ *
	+ * Construct the client, with Deps support.
	+ *
	+ * This enables startSyncing() to work properly.
	+ *
	+ * @warning You should not use this directly. Use
	+ * Sdk::client() instead.
	+ */
	+ Client(InEventLoopTag,
	+ ContextWithDepsT ctx);
	+
	+

	/* lager::reader<immer::map<std::string, Room>> */
	inline auto rooms() const {
	- return m_client
	+ return clientCursor()
	[&ClientModel::roomList]
	[&RoomListModel::rooms];
	}

	/* lager::reader<RangeT<std::string>> */
	inline auto roomIds() const {
	return rooms().xform(
	zug::map([](auto m) {
	return intoImmer(
	immer::flex_vector<std::string>{},
	zug::map([](auto val) { return val.first; }),
	m);
	}));
	}

	- KAZV_WRAP_ATTR(ClientModel, m_client, serverUrl)
	- KAZV_WRAP_ATTR(ClientModel, m_client, loggedIn)
	- KAZV_WRAP_ATTR(ClientModel, m_client, userId)
	- KAZV_WRAP_ATTR(ClientModel, m_client, token)
	- KAZV_WRAP_ATTR(ClientModel, m_client, deviceId)
	- KAZV_WRAP_ATTR(ClientModel, m_client, toDevice)
	+ KAZV_WRAP_ATTR(ClientModel, clientCursor(), serverUrl)
	+ KAZV_WRAP_ATTR(ClientModel, clientCursor(), loggedIn)
	+ KAZV_WRAP_ATTR(ClientModel, clientCursor(), userId)
	+ KAZV_WRAP_ATTR(ClientModel, clientCursor(), token)
	+ KAZV_WRAP_ATTR(ClientModel, clientCursor(), deviceId)
	+ KAZV_WRAP_ATTR(ClientModel, clientCursor(), toDevice)

	/**
	* Get the room with @c id .
	*
	* This is equivalent to `roomByCursor(lager::make_constant(id))`.
	*
	* @param id The room id.
	* @return A Room representing the room with `id`.
	*/
	Room room(std::string id) const;

	/**
	* Get the room with `id`.
	*
	* The Room returned will change as the content in `id` changes.
	*
	* For example, you can have the Room that is always the first
	* alphabetically in all rooms by:
	*
	* \code{.cpp}
	* auto someProcessing =
	* zug::map([=](auto ids) {
	* std::sort(ids.begin(), ids.end(), [=](auto id1, auto id2) {
	* using namespace Kazv::CursorOp;
	* return (+client.room(id1).name()) < (+client.room(id2).name());
	* });
	* return ids;
	* });
	* auto room =
	* client.roomByCursor(
	* client.roomIds().xform(someProcessing)[0]);
	* \endcode
	*
	* @param id A lager::reader<std::string> containing the room id.
	* @return A Room representing the room with `id`.
	*/
	Room roomByCursor(lager::reader<std::string> id) const;

	/**
	* Login using the password.
	*
	* This will create a new session on the homeserver.
	*
	* If the returned Promise resolves successfully, this will
	* call `startSyncing()`.
	*
	* @param homeserver The base url of the homeserver. E.g. `https://tusooa.xyz`.
	* @param username The username. This can be the full user id or
	* just the local part. E.g. `tusooa`, `@tusooa:tusooa.xyz`.
	* @param password The password.
	* @param deviceName Optionally, a custom device name. If empty, `libkazv`
	* will be used.
	* @return A Promise that resolves when logging in successfully, or
	* when there is an error.
	*/
	PromiseT passwordLogin(std::string homeserver, std::string username,
	std::string password, std::string deviceName) const;

	/**
	* Login using `token` and `deviceId`.
	*
	* This will not make a request. Library users should make sure
	* the information is correct and the token and the device id are valid.
	*
	* If the returned Promise resolves successfully, this will
	* call `startSyncing()`.
	*
	* @param homeserver The base url of the homeserver. E.g. `https://tusooa.xyz`.
	* @param username The full user id. E.g. `@tusooa:tusooa.xyz`.
	* @param token The access token.
	* @param deviceId The device id that is paired with `token`.
	* @return A Promise that resolves when the account information is filled in.
	*/
	PromiseT tokenLogin(std::string homeserver, std::string username,
	std::string token, std::string deviceId) const;

	/**
	* Create a room.
	*
	* @param v The visibility of the room.
	* @param name The name of the room.
	* @param alias The alias of the room.
	* @param invite User ids to invite to this room.
	* @param isDirect Whether this room is a direct chat.
	* @param allowFederate Whether to allow users from other homeservers
	* to join this room.
	* @param topic The topic of the room.
	* @param powerLevelContentOverride The content of the m.room.power_levels
	* state event to override the default.
	* @return A Promise that resolves when the room is created,
	* or when there is an error.
	*/
	PromiseT createRoom(RoomVisibility v,
	std::optional<std::string> name = {},
	std::optional<std::string> alias = {},
	immer::array<std::string> invite = {},
	std::optional<bool> isDirect = {},
	bool allowFederate = true,
	std::optional<std::string> topic = {},
	JsonWrap powerLevelContentOverride = json::object()) const;

	/**
	* Join a room by its id.
	*
	* @param roomId The id of the room to join.
	* @return A Promise that resolves when the room is joined,
	* or when there is an error.
	*/
	PromiseT joinRoomById(std::string roomId) const;

	/**
	* Join a room by its id or alias.
	*
	* @param roomId The id or alias of the room to join.
	* @param serverName A list of servers to use when joining the room.
	* This corresponds to the `via` parameter in a matrix.to url.
	* @return A Promise that resolves when the room is joined,
	* or when there is an error.
	*/
	PromiseT joinRoom(std::string roomId, immer::array<std::string> serverName) const;

	/**
	* Upload content to the content repository.
	*
	* @param content The content to upload.
	* @param uploadId
	* @param filename The name of the file.
	* @param contentType The content type of the file.
	* @return A Promise that resolves when the upload is successful,
	* or when there is an error. If it successfully resolves to `r`,
	* `r.dataStr("mxcUri")` will be the MXC URI of the uploaded
	* content.
	*/
	PromiseT uploadContent(immer::box<Bytes> content,
	std::string uploadId,
	std::optional<std::string> filename = std::nullopt,
	std::optional<std::string> contentType = std::nullopt) const;

	/**
	* Upload content to the content repository.
	*
	* @param file The file to upload.
	* @return A Promise that resolves when the upload is successful,
	* or when there is an error. If it successfully resolves to `r`,
	* `r.dataStr("mxcUri")` will be the MXC URI of the uploaded
	* content.
	*/
	PromiseT uploadContent(FileDesc file) const;


	/**
	* Convert a MXC URI to an HTTP(s) URI.
	*
	* The converted URI will be using the homeserver of
	* this Client.
	*
	* @param mxcUri The MXC URI to convert.
	* @return The HTTP(s) URI that has the content indicated
	* by `mxcUri`.
	*/
	inline std::string mxcUriToHttp(std::string mxcUri) const {
	using namespace CursorOp;
	auto [serverName, mediaId] = mxcUriToMediaDesc(mxcUri);
	- return (+m_client)
	+ return (+clientCursor())
	.job<GetContentJob>()
	.make(serverName, mediaId).url();
	}

	/**
	* Download content from the content repository
	*
	* After the returned Promise resolves successfully,
	* if @c downloadTo is provided, the content will be available
	* in that file; if it is not provided, `r.dataStr("content")`
	* will contain the content of the downloaded file.
	*
	* @param mxcUri The MXC URI of the content.
	* @param downloadTo The file to write the content to. Must not be
	* an in-memory file.
	* @return A Promise that is resolved after the content
	* is downloaded, or when there is an error.
	*/
	PromiseT downloadContent(std::string mxcUri,
	std::optional<FileDesc> downloadTo = std::nullopt) const;

	/**
	* Download a thumbnail from the content repository
	*
	* After the returned Promise resolves successfully,
	* if @c downloadTo is provided, the content will be available
	* in that file; if it is not provided, `r.dataStr("content")`
	* will contain the content of the downloaded file.
	*
	* @param mxcUri The MXC URI of the content.
	* @param width,height The dimension wanted for the thumbnail
	* @param method The method to generate the thumbnail. Either `Crop`
	* or `Scale`.
	* @param downloadTo The file to write the content to. Must not be
	* an in-memory file.
	* @return A Promise that is resolved after the thumbnail
	* is downloaded, or when there is an error.
	*/
	PromiseT downloadThumbnail(std::string mxcUri,
	int width,
	int height,
	std::optional<ThumbnailResizingMethod> method = std::nullopt,
	std::optional<FileDesc> downloadTo = std::nullopt) const;

	// lager::reader<bool>
	inline auto syncing() const {
	- return m_client[&ClientModel::syncing];
	+ return clientCursor()[&ClientModel::syncing];
	}

	/**
	* Start syncing if the Client is not syncing.
	*
	* Syncing will continue indefinitely, if the preparation of
	* the sync (posting filters and uploading identity keys,
	* if needed) is successful, or until stopSyncing() is called.
	*
	* @return A Promise that resolves when the Client is syncing
	* (more exactly, when syncing() contains true), or when there
	* is an error in the preparation of the sync.
	*/
	PromiseT startSyncing() const;

	/**
	* Stop the indefinite syncing.
	*
	* After this, no more syncing actions will be dispatched.
	*/
	void stopSyncing() const;

	/**
	* Serialize the model to a Boost.Serialization archive.
	*
	* @param ar A Boost.Serialization output archive.
	*
	* This function can be used to save the model. For loading,
	* you should use the makeSdk function. For example:
	*
	* ```c++
	* client.serializeTo(outputAr);
	*
	* SdkModel m;
	* inputAr >> m;
	* auto newSdk = makeSdk(m, ...);
	* ```
	*/
	template<class Archive>
	void serializeTo(Archive &ar) const {
	- ar << m_sdk.get();
	+ ar << sdkCursor().get();
	}

	private:
	void syncForever(std::optional<int> retryTime = std::nullopt) const;

	- lager::reader<SdkModel> m_sdk;
	- lager::reader<ClientModel> m_client;
	+ const lager::reader<SdkModel> &sdkCursor() const;
	+ lager::reader<ClientModel> clientCursor() const;
	+
	+ std::optional<lager::reader<SdkModel>> m_sdk;
	+ std::optional<lager::reader<ClientModel>> m_client;
	ContextT m_ctx;
	std::optional<DepsT> m_deps;
	KAZV_DECLARE_THREAD_ID();
	KAZV_DECLARE_EVENT_LOOP_THREAD_ID_KEEPER(m_deps.has_value() ? &lager::get<EventLoopThreadIdKeeper &>(m_deps.value()) : 0);
	};

	}
	diff --git a/src/client/sdk.hpp b/src/client/sdk.hpp
	index 78fadc3..d9309c5 100644
	--- a/src/client/sdk.hpp
	+++ b/src/client/sdk.hpp
	@@ -1,178 +1,177 @@
	/*
	* Copyright (C) 2020 Tusooa Zhu
	*
	* This file is part of libkazv.
	*
	* libkazv is free software: you can redistribute it and/or modify
	* it under the terms of the GNU Affero General Public License as
	* published by the Free Software Foundation, either version 3 of the
	* License, or (at your option) any later version.
	*
	* libkazv is distributed in the hope that it will be useful,
	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	* GNU Affero General Public License for more details.
	*
	* You should have received a copy of the GNU Affero General Public License
	* along with libkazv. If not, see <https://www.gnu.org/licenses/>.
	*/

	#pragma once
	#include <libkazv-config.hpp>
	#include <lager/store.hpp>

	#include <store.hpp>

	#include "sdk-model.hpp"
	#include "sdk-model-cursor-tag.hpp"
	#include "client.hpp"
	#include "thread-safety-helper.hpp"

	namespace Kazv
	{
	/**
	* Contain the single source of truth of a matrix sdk.
	*/
	template<class EventLoop, class Xform, class ...Enhancers>
	class Sdk
	{
	using ModelT = ::Kazv::SdkModel;
	using ClientT = ::Kazv::ClientModel;
	using ActionT = typename ModelT::Action;
	using CursorT = lager::reader<ModelT>;
	using CursorTSP = std::shared_ptr<const CursorT>;

	using StoreT = decltype(
	makeStore<ActionT>(
	std::declval<ModelT>(),
	&ModelT::update,
	std::declval<EventLoop>(),
	lager::with_deps(
	std::ref(detail::declref<JobInterface>()),
	std::ref(detail::declref<EventInterface>()),
	lager::dep::as<SdkModelCursorKey>(std::declval<std::function<CursorTSP()>>())
	#ifdef KAZV_USE_THREAD_SAFETY_HELPER
	, std::ref(detail::declref<EventLoopThreadIdKeeper>())
	#endif
	),
	std::declval<Enhancers>()...)
	);

	using DepsT = lager::deps<JobInterface &, EventInterface &, SdkModelCursorKey
	#ifdef KAZV_USE_THREAD_SAFETY_HELPER
	, EventLoopThreadIdKeeper &
	#endif
	>;

	using ContextT = Context<ActionT, DepsT>;
	public:
	Sdk(ModelT model,
	JobInterface &jobHandler,
	EventInterface &eventEmitter,
	EventLoop &&eventLoop,
	Xform &&xform,
	Enhancers &&...enhancers)
	: m_d(std::make_unique<Private>(
	std::move(model), jobHandler, eventEmitter,
	std::forward<EventLoop>(eventLoop), std::forward<Xform>(xform),
	std::forward<Enhancers>(enhancers)...)) {}

	/**
	* Get the context associated with this.
	*
	* The returned context is thread-safe if every thread calls with
	* different instances.
	*/
	ContextT context() const {
	return m_d->store;
	}

	/**
	* Get a Client representing this.
	*
	- * The returned Client can only be used in the thread where
	- * the promise handler runs.
	+ * The returned Client belongs to the thread where the promise handler runs.
	*/
	Client client() const {
	- return {*m_d->sdk, ContextT(m_d->store)};
	+ return {Client::InEventLoopTag{}, ContextT(m_d->store)};
	}

	private:
	struct Private
	{
	Private(ModelT model,
	JobInterface &jobHandler,
	EventInterface &eventEmitter,
	EventLoop &&eventLoop,
	Xform &&xform,
	Enhancers &&...enhancers)
	: store(makeStore<ActionT>(
	std::move(model),
	&ModelT::update,
	std::forward<EventLoop>(eventLoop),
	lager::with_deps(
	std::ref(jobHandler),
	std::ref(eventEmitter),
	lager::dep::as<SdkModelCursorKey>(
	std::function<CursorTSP()>([this] { return sdk; }))
	#ifdef KAZV_USE_THREAD_SAFETY_HELPER
	, std::ref(keeper)
	#endif
	),
	std::forward<Enhancers>(enhancers)...))
	, sdk(std::make_shared<lager::reader<ModelT>>(store.reader().xform(std::forward<Xform>(xform))))
	{
	#ifdef KAZV_USE_THREAD_SAFETY_HELPER
	store.context().createResolvedPromise(EffectStatus{})
	.then([this](auto &&) {
	keeper.set(std::this_thread::get_id());
	});
	#endif
	}
	#ifdef KAZV_USE_THREAD_SAFETY_HELPER
	EventLoopThreadIdKeeper keeper;
	#endif
	StoreT store;
	std::shared_ptr<const lager::reader<ModelT>> sdk;
	};

	std::unique_ptr<Private> m_d;
	};

	template<class EventLoop, class Xform, class ...Enhancers>
	inline auto makeSdk(SdkModel sdk,
	JobInterface &jobHandler,
	EventInterface &eventEmitter,
	EventLoop &&eventLoop,
	Xform &&xform,
	Enhancers &&...enhancers)
	-> Sdk<EventLoop, Xform, Enhancers...>
	{
	return { std::move(sdk),
	jobHandler,
	eventEmitter,
	std::forward<EventLoop>(eventLoop),
	std::forward<Xform>(xform),
	std::forward<Enhancers>(enhancers)... };
	}

	template<class EventLoop, class Xform, class ...Enhancers>
	inline auto makeDefaultEncryptedSdk(
	JobInterface &jobHandler,
	EventInterface &eventEmitter,
	EventLoop &&eventLoop,
	Xform &&xform,
	Enhancers &&...enhancers)
	-> Sdk<EventLoop, Xform, Enhancers...>
	{
	auto m = SdkModel{};
	m.client.crypto = Crypto();

	return makeSdk(std::move(m),
	jobHandler,
	eventEmitter,
	std::forward<EventLoop>(eventLoop),
	std::forward<Xform>(xform),
	std::forward<Enhancers>(enhancers)...);
	}
	}
	diff --git a/src/client/thread-safety-helper.hpp b/src/client/thread-safety-helper.hpp
	index 8e4a4c2..0709798 100644
	--- a/src/client/thread-safety-helper.hpp
	+++ b/src/client/thread-safety-helper.hpp
	@@ -1,75 +1,76 @@
	/*
	* This file is part of libkazv.
	* SPDX-FileCopyrightText: 2020-2021 Tusooa Zhu <tusooa@kazv.moe>
	* SPDX-License-Identifier: AGPL-3.0-or-later
	*/

	#pragma once

	#include <libkazv-config.hpp>

	#ifndef NDEBUG
	#ifndef KAZV_USE_THREAD_SAFETY_HELPER
	#define KAZV_USE_THREAD_SAFETY_HELPER
	#endif
	#endif

	#ifdef KAZV_USE_THREAD_SAFETY_HELPER
	#include <mutex>
	struct ThreadNotMatchException // Don't derive from std::exception to avoid catch-all catch clauses
	{
	std::string m_what;

	std::string what() const;
	};

	struct EventLoopThreadIdKeeper
	{
	mutable std::mutex m_mutex;
	std::optional<std::thread::id> m_id;

	/// This should only be called in the event loop thread.
	void set(std::thread::id id) {
	std::lock_guard<std::mutex> g(m_mutex);
	m_id = id;
	}

	std::optional<std::thread::id> get() const {
	std::lock_guard<std::mutex> g(m_mutex);
	return m_id;
	}
	};
	#define KAZV_THREAD_ID_VAR _threadSafetyHelper_threadId
	#define KAZV_ON_EVENT_LOOP_VAR _threadSafetyHelper_onEventLoop
	#define KAZV_EVENT_LOOP_THREAD_ID_KEEPER_VAR _threadSafetyHelper_eventLoopThreadIdKeeper
	#define KAZV_DECLARE_THREAD_ID() bool KAZV_ON_EVENT_LOOP_VAR{false}; \
	std::thread::id KAZV_THREAD_ID_VAR = std::this_thread::get_id();
	#define KAZV_DECLARE_EVENT_LOOP_THREAD_ID_KEEPER(_initializer) \
	EventLoopThreadIdKeeper *KAZV_EVENT_LOOP_THREAD_ID_KEEPER_VAR = _initializer
	//#define KAZV_INIT_THREAD_ID_FROM_KEEPER() KAZV_THREAD_ID_VAR(KAZV_EVENT_LOOP_THREAD_ID_KEEPER_VAR.get())
	#define KAZV_VERIFY_THREAD_ID() \
	do { \
	auto _threadSafetyHelper_local_idActual = std::this_thread::get_id(); \
	\
	if (KAZV_ON_EVENT_LOOP_VAR) { \
	auto _threadSafetyHelper_local_idExpected = \
	KAZV_EVENT_LOOP_THREAD_ID_KEEPER_VAR ? KAZV_EVENT_LOOP_THREAD_ID_KEEPER_VAR->get() : std::nullopt; \
	auto cond = \
	_threadSafetyHelper_local_idExpected.has_value() \
	? _threadSafetyHelper_local_idExpected.value() == _threadSafetyHelper_local_idActual \
	/* if the id is not set yet it means the event loop is not yet run, so it does not matter anyway */ \
	: true; \
	if (!cond) { \
	throw ThreadNotMatchException{"Current object belongs to the event loop, but method is called outside the event loop"}; \
	} \
	} else { \
	if (! (KAZV_THREAD_ID_VAR == _threadSafetyHelper_local_idActual)) { \
	throw ThreadNotMatchException{"Current thread id does not match the id of the thread where it belongs"}; \
	} \
	} \
	} while (false)

	#else
	#define KAZV_DECLARE_THREAD_ID()
	#define KAZV_VERIFY_THREAD_ID()
	+#define KAZV_DECLARE_EVENT_LOOP_THREAD_ID_KEEPER(_initializer)
	#endif
	diff --git a/src/tests/client/thread-safety-test.cpp b/src/tests/client/thread-safety-test.cpp
	index 4eb3b37..163d10e 100644
	--- a/src/tests/client/thread-safety-test.cpp
	+++ b/src/tests/client/thread-safety-test.cpp
	@@ -1,53 +1,85 @@
	/*
	* This file is part of libkazv.
	* SPDX-FileCopyrightText: 2021 Tusooa Zhu <tusooa@kazv.moe>
	* SPDX-License-Identifier: AGPL-3.0-or-later
	*/

	#ifndef NDEBUG
	#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>

	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
	);

	- REQUIRE(sdk.context().has<EventLoopThreadIdKeeper>());
	+ 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());
	+
	+ 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);
	}


	#endif

File Metadata

Mime Type: text/x-diff
Expires: Sat, Sep 19, 1:25 PM (21 h, 10 m)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1769513
Default Alt Text: (41 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions