Add first version
First version of C++ SDL API.
Refer README file for instructions how to generate API documentation.
Current version will find database via environment variables, thus RIC
DBaaS service must be up and running before starting the application
pod which uses SDL C++ API.
Change-Id: Ia3c95ac856b293827b680a9f6451a229f69f35f9
Signed-off-by: Rolf Badorek <rolf.badorek@nokia.com>
diff --git a/src/abort.cpp b/src/abort.cpp
new file mode 100644
index 0000000..c612f2b
--- /dev/null
+++ b/src/abort.cpp
@@ -0,0 +1,31 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include <ostream>
+#include <sstream>
+#include <string>
+#include "private/abort.hpp"
+#include "private/createlogger.hpp"
+
+using namespace shareddatalayer;
+
+void shareddatalayer::logAndAbort(const std::string& file, int line, const std::string& message) noexcept
+{
+ std::ostringstream msg;
+ msg << "SHAREDDATALAYER_ABORT, file " << file << ", line " << line << ": " << message;
+ logErrorOnce(msg.str());
+ abort();
+}
diff --git a/src/asyncconnection.cpp b/src/asyncconnection.cpp
new file mode 100644
index 0000000..4acfead
--- /dev/null
+++ b/src/asyncconnection.cpp
@@ -0,0 +1,21 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/asyncconnection.hpp"
+
+using namespace shareddatalayer;
+
+const char AsyncConnection::SEPARATOR(',');
diff --git a/src/asyncdummystorage.cpp b/src/asyncdummystorage.cpp
new file mode 100644
index 0000000..24e85a6
--- /dev/null
+++ b/src/asyncdummystorage.cpp
@@ -0,0 +1,86 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/asyncdummystorage.hpp"
+#include <sys/eventfd.h>
+#include "private/engine.hpp"
+
+using namespace shareddatalayer;
+
+AsyncDummyStorage::AsyncDummyStorage(std::shared_ptr<Engine> engine):
+ engine(engine)
+{
+}
+
+int AsyncDummyStorage::fd() const
+{
+ return engine->fd();
+}
+
+void AsyncDummyStorage::handleEvents()
+{
+ engine->handleEvents();
+}
+
+void AsyncDummyStorage::postCallback(const Callback& callback)
+{
+ engine->postCallback(callback);
+}
+
+void AsyncDummyStorage::waitReadyAsync(const Namespace&, const ReadyAck& readyAck)
+{
+ postCallback(std::bind(readyAck, std::error_code()));
+}
+
+void AsyncDummyStorage::setAsync(const Namespace&, const DataMap&, const ModifyAck& modifyAck)
+{
+ postCallback(std::bind(modifyAck, std::error_code()));
+}
+
+void AsyncDummyStorage::setIfAsync(const Namespace&, const Key&, const Data&, const Data&, const ModifyIfAck& modifyIfAck)
+{
+ postCallback(std::bind(modifyIfAck, std::error_code(), true));
+}
+
+void AsyncDummyStorage::setIfNotExistsAsync(const Namespace&, const Key&, const Data&, const ModifyIfAck& modifyIfAck)
+{
+ postCallback(std::bind(modifyIfAck, std::error_code(), true));
+}
+
+void AsyncDummyStorage::getAsync(const Namespace&, const Keys&, const GetAck& getAck)
+{
+ postCallback(std::bind(getAck, std::error_code(), DataMap()));
+}
+
+void AsyncDummyStorage::removeAsync(const Namespace&, const Keys&, const ModifyAck& modifyAck)
+{
+ postCallback(std::bind(modifyAck, std::error_code()));
+}
+
+void AsyncDummyStorage::removeIfAsync(const Namespace&, const Key&, const Data&, const ModifyIfAck& modifyIfAck)
+{
+ postCallback(std::bind(modifyIfAck, std::error_code(), true));
+}
+
+void AsyncDummyStorage::findKeysAsync(const Namespace&, const std::string&, const FindKeysAck& findKeysAck)
+{
+ postCallback(std::bind(findKeysAck, std::error_code(), Keys()));
+}
+
+void AsyncDummyStorage::removeAllAsync(const Namespace&, const ModifyAck& modifyAck)
+{
+ postCallback(std::bind(modifyAck, std::error_code()));
+}
diff --git a/src/asyncstorage.cpp b/src/asyncstorage.cpp
new file mode 100644
index 0000000..3b6fe77
--- /dev/null
+++ b/src/asyncstorage.cpp
@@ -0,0 +1,53 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include <boost/optional.hpp>
+#include "config.h"
+#include "private/abort.hpp"
+#include "private/asyncconnection.hpp"
+#include "private/asyncstorageimpl.hpp"
+#include "private/createlogger.hpp"
+#include "private/engineimpl.hpp"
+#include "private/configurationreader.hpp"
+#include "private/databaseconfigurationimpl.hpp"
+#include "private/logger.hpp"
+#if HAVE_REDIS
+#include "private/redis/asyncredisstorage.hpp"
+#include "private/redis/asyncdatabasediscovery.hpp"
+#endif
+
+using namespace shareddatalayer;
+
+namespace
+{
+
+std::unique_ptr<shareddatalayer::AsyncStorage> createInstance(const boost::optional<shareddatalayer::AsyncConnection::PublisherId>& pId)
+{
+ auto engine(std::make_shared<EngineImpl>());
+ auto logger(createLogger(SDL_LOG_PREFIX));
+ /* clang compilation does not support make_unique */
+ return std::unique_ptr<AsyncStorageImpl>(new AsyncStorageImpl(engine, pId, logger));
+}
+
+}
+
+/* clang compilation produces undefined reference linker error without this */
+constexpr char AsyncStorage::SEPARATOR;
+
+std::unique_ptr<AsyncStorage> AsyncStorage::create()
+{
+ return createInstance(boost::none);
+}
diff --git a/src/asyncstorageimpl.cpp b/src/asyncstorageimpl.cpp
new file mode 100644
index 0000000..b17fcbd
--- /dev/null
+++ b/src/asyncstorageimpl.cpp
@@ -0,0 +1,167 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "config.h"
+#include "private/error.hpp"
+#include "private/abort.hpp"
+#include "private/asyncstorageimpl.hpp"
+#include "private/configurationreader.hpp"
+#include "private/asyncdummystorage.hpp"
+#include "private/engine.hpp"
+#include "private/logger.hpp"
+#if HAVE_REDIS
+#include "private/redis/asyncdatabasediscovery.hpp"
+#include "private/redis/asyncredisstorage.hpp"
+#endif
+
+using namespace shareddatalayer;
+
+AsyncStorageImpl::AsyncStorageImpl(std::shared_ptr<Engine> engine,
+ const boost::optional<PublisherId>& pId,
+ std::shared_ptr<Logger> logger):
+ engine(engine),
+ databaseConfiguration(std::make_shared<DatabaseConfigurationImpl>()),
+ namespaceConfigurations(std::make_shared<NamespaceConfigurationsImpl>()),
+ publisherId(pId),
+ logger(logger)
+{
+ ConfigurationReader configurationReader(logger);
+ configurationReader.readDatabaseConfiguration(std::ref(*databaseConfiguration));
+ configurationReader.readNamespaceConfigurations(std::ref(*namespaceConfigurations));
+}
+
+// Meant for UT usage
+AsyncStorageImpl::AsyncStorageImpl(std::shared_ptr<Engine> engine,
+ const boost::optional<PublisherId>& pId,
+ std::shared_ptr<DatabaseConfiguration> databaseConfiguration,
+ std::shared_ptr<NamespaceConfigurations> namespaceConfigurations,
+ std::shared_ptr<Logger> logger):
+ engine(engine),
+ databaseConfiguration(databaseConfiguration),
+ namespaceConfigurations(namespaceConfigurations),
+ publisherId(pId),
+ logger(logger)
+{
+}
+
+AsyncStorage& AsyncStorageImpl::getRedisHandler()
+{
+#if HAVE_REDIS
+ static AsyncRedisStorage redisHandler{engine,
+ redis::AsyncDatabaseDiscovery::create(
+ engine,
+ boost::none,
+ std::ref(*databaseConfiguration),
+ logger),
+ publisherId,
+ namespaceConfigurations,
+ logger};
+
+ return redisHandler;
+#else
+ logger->error() << "Redis operations cannot be performed, Redis not enabled";
+ SHAREDDATALAYER_ABORT("Invalid configuration.");
+#endif
+}
+
+AsyncStorage& AsyncStorageImpl::getDummyHandler()
+{
+ static AsyncDummyStorage dummyHandler{engine};
+ return dummyHandler;
+}
+
+AsyncStorage& AsyncStorageImpl::getOperationHandler(const std::string& ns)
+{
+ if (namespaceConfigurations->isDbBackendUseEnabled(ns))
+ return getRedisHandler();
+
+ return getDummyHandler();
+}
+
+int AsyncStorageImpl::fd() const
+{
+ return engine->fd();
+}
+
+void AsyncStorageImpl::handleEvents()
+{
+ engine->handleEvents();
+}
+
+void AsyncStorageImpl::waitReadyAsync(const Namespace& ns,
+ const ReadyAck& readyAck)
+{
+ getOperationHandler(ns).waitReadyAsync(ns, readyAck);
+}
+
+void AsyncStorageImpl::setAsync(const Namespace& ns,
+ const DataMap& dataMap,
+ const ModifyAck& modifyAck)
+{
+ getOperationHandler(ns).setAsync(ns, dataMap, modifyAck);
+}
+
+void AsyncStorageImpl::setIfAsync(const Namespace& ns,
+ const Key& key,
+ const Data& oldData,
+ const Data& newData,
+ const ModifyIfAck& modifyIfAck)
+{
+ getOperationHandler(ns).setIfAsync(ns, key, oldData, newData, modifyIfAck);
+}
+
+void AsyncStorageImpl::removeIfAsync(const Namespace& ns,
+ const Key& key,
+ const Data& data,
+ const ModifyIfAck& modifyIfAck)
+{
+ getOperationHandler(ns).removeIfAsync(ns, key, data, modifyIfAck);
+}
+
+void AsyncStorageImpl::setIfNotExistsAsync(const Namespace& ns,
+ const Key& key,
+ const Data& data,
+ const ModifyIfAck& modifyIfAck)
+{
+ getOperationHandler(ns).setIfNotExistsAsync(ns, key, data, modifyIfAck);
+}
+
+void AsyncStorageImpl::getAsync(const Namespace& ns,
+ const Keys& keys,
+ const GetAck& getAck)
+{
+ getOperationHandler(ns).getAsync(ns, keys, getAck);
+}
+
+void AsyncStorageImpl::removeAsync(const Namespace& ns,
+ const Keys& keys,
+ const ModifyAck& modifyAck)
+{
+ getOperationHandler(ns).removeAsync(ns, keys, modifyAck);
+}
+
+void AsyncStorageImpl::findKeysAsync(const Namespace& ns,
+ const std::string& keyPrefix,
+ const FindKeysAck& findKeysAck)
+{
+ getOperationHandler(ns).findKeysAsync(ns, keyPrefix, findKeysAck);
+}
+
+void AsyncStorageImpl::removeAllAsync(const Namespace& ns,
+ const ModifyAck& modifyAck)
+{
+ getOperationHandler(ns).removeAllAsync(ns, modifyAck);
+}
diff --git a/src/backenderror.cpp b/src/backenderror.cpp
new file mode 100644
index 0000000..3086ea4
--- /dev/null
+++ b/src/backenderror.cpp
@@ -0,0 +1,24 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include <sdl/backenderror.hpp>
+
+using namespace shareddatalayer;
+
+BackendError::BackendError(const std::string& error):
+ Exception(error)
+{
+}
diff --git a/src/cli/commandmap.cpp b/src/cli/commandmap.cpp
new file mode 100644
index 0000000..e0e8304
--- /dev/null
+++ b/src/cli/commandmap.cpp
@@ -0,0 +1,185 @@
+#include "private/cli/commandmap.hpp"
+#include <sstream>
+#include <iomanip>
+#include <memory>
+#include <cstdlib>
+#include <boost/io/ios_state.hpp>
+
+using namespace shareddatalayer::cli;
+
+namespace
+{
+ std::string buildCommandNameAlreadyRegisteredError(const std::string& commandName)
+ {
+ std::ostringstream os;
+ os << "command name \"" << commandName << "\" already registered";
+ return os.str();
+ }
+
+ std::string buildUnknownCommandNameError(const std::string& commandName)
+ {
+ std::ostringstream os;
+ os << "unknown command: \"" << commandName << '\"';
+ return os.str();
+ }
+
+ std::string buildCategoryOffsetAlreadyRegisteredError(const std::string& commandName, int offset)
+ {
+ std::ostringstream os;
+ os << commandName << ": Offset " << offset << " already registered";
+ return os.str();
+ }
+
+ std::string getCategoryName(CommandMap::Category category)
+ {
+ switch (category)
+ {
+ case CommandMap::Category::UTIL:
+ return "Utility";
+ default:
+ return "Unknown";
+ }
+ }
+}
+
+CommandMap::CommandNameAlreadyRegistered::CommandNameAlreadyRegistered(const std::string& commandName):
+ Exception(buildCommandNameAlreadyRegisteredError(commandName))
+{
+}
+
+CommandMap::UnknownCommandName::UnknownCommandName(const std::string& commandName):
+ Exception(buildUnknownCommandNameError(commandName))
+{
+}
+
+CommandMap::CategoryOffsetAlreadyRegistered::CategoryOffsetAlreadyRegistered(const std::string& commandName, int offset):
+ Exception(buildCategoryOffsetAlreadyRegisteredError(commandName, offset))
+{
+}
+
+struct CommandMap::Info
+{
+ CommandFunction function;
+ std::string shortHelp;
+ std::string longHelp;
+ boost::program_options::options_description options;
+
+ Info(const CommandFunction& function,
+ const std::string& shortHelp,
+ const std::string& longHelp);
+};
+
+CommandMap::Info::Info(const CommandFunction& function,
+ const std::string& shortHelp,
+ const std::string& longHelp):
+ function(function),
+ shortHelp(shortHelp),
+ longHelp(longHelp)
+{
+}
+
+CommandMap::CommandMap()
+{
+}
+
+CommandMap::~CommandMap()
+{
+}
+
+boost::program_options::options_description&
+CommandMap::registerCommand(const std::string& commandName,
+ const std::string& shortHelp,
+ const std::string& longHelp,
+ const CommandFunction& commandFunction,
+ Category category,
+ int categoryOffset)
+{
+ const auto ret(map.insert(std::make_pair(commandName, Info(commandFunction, shortHelp, longHelp))));
+ if (!ret.second)
+ throw CommandNameAlreadyRegistered(commandName);
+ const auto retCat(categoryMap.insert(std::make_pair(CategoryKey(category, categoryOffset), commandName)));
+ if (!retCat.second)
+ throw CategoryOffsetAlreadyRegistered(commandName, categoryOffset);
+ return ret.first->second.options;
+}
+
+std::vector<std::string> CommandMap::getCommandNames() const
+{
+ std::vector<std::string> ret;
+ for (const auto& i : map)
+ ret.push_back(i.first);
+ return ret;
+}
+
+const boost::program_options::options_description&
+CommandMap::getCommandOptions(const std::string& commandName) const
+{
+ const auto i(map.find(commandName));
+ if (i == map.end())
+ throw UnknownCommandName(commandName);
+ return i->second.options;
+}
+
+void CommandMap::shortHelp(std::ostream& out) const
+{
+ boost::io::ios_all_saver guard(out);
+ size_t maxWidth(0U);
+ Category currentCategory(Category::UNDEFINED);
+ for (const auto& i : map)
+ if (maxWidth < i.first.size())
+ maxWidth = i.first.size();
+ for (const auto& i : categoryMap)
+ {
+ if (currentCategory != i.first.first)
+ {
+ currentCategory = i.first.first;
+ out << std::endl;
+ out << getCategoryName(currentCategory) << " commands:" << std::endl;
+ }
+ out << std::left << std::setw(maxWidth + 2U) << i.second << map.at(i.second).shortHelp << '\n';
+ }
+}
+
+void CommandMap::longHelp(std::ostream& out, const std::string& commandName) const
+{
+ const auto i(map.find(commandName));
+ if (i == map.end())
+ throw UnknownCommandName(commandName);
+ out << i->first;
+ if (!i->second.options.options().empty())
+ {
+ out << " OPTIONS\n\n" << i->second.longHelp << "\n\nOptions:\n";
+ i->second.options.print(out);
+ }
+ else
+ {
+ out << "\n\n" << i->second.longHelp << '\n';
+ }
+}
+
+int CommandMap::execute(const std::string& commandName,
+ std::ostream& out,
+ std::ostream& err,
+ const boost::program_options::variables_map& params,
+ size_t count)
+{
+ const auto i(map.find(commandName));
+ if (i == map.end())
+ {
+ err << "unknown command: \"" << commandName << '\"' << std::endl;
+ return EXIT_FAILURE;
+ }
+ const auto& function(i->second.function);
+ int ret(EXIT_SUCCESS);
+ for (size_t i = 0U; i < count; ++i)
+ if ((ret = function(out, err, params)) != EXIT_SUCCESS)
+ break;
+ return ret;
+}
+
+CommandMap& CommandMap::getCommandMap() noexcept
+{
+ static CommandMap instance;
+ return instance;
+}
+
diff --git a/src/cli/commandparserandexecutor.cpp b/src/cli/commandparserandexecutor.cpp
new file mode 100644
index 0000000..b0d52bf
--- /dev/null
+++ b/src/cli/commandparserandexecutor.cpp
@@ -0,0 +1,176 @@
+#include "config.h"
+#include "private/cli/commandparserandexecutor.hpp"
+#include <ostream>
+#include <set>
+#include <string>
+#include <exception>
+#include <boost/program_options.hpp>
+#include "private/cli/commandmap.hpp"
+
+using namespace shareddatalayer;
+using namespace shareddatalayer::cli;
+namespace po = boost::program_options;
+
+namespace
+{
+ void outputBashCompletion(std::ostream& os,
+ const po::options_description& commonOpts,
+ CommandMap& commandMap)
+ {
+ const auto commandNames(commandMap.getCommandNames());
+ std::set<std::string> optionNames;
+ for (const auto& i : commonOpts.options())
+ optionNames.insert(i->long_name());
+ for (const auto& i : commandNames)
+ for (const auto& j : commandMap.getCommandOptions(i).options())
+ optionNames.insert(j->long_name());
+
+ os << "_sdltool()\n{\n COMPREPLY=()\n local commands=\"";
+ for (const auto& i : commandNames)
+ os << i << ' ';
+ os << "\"\n local options=\"";
+ for (const auto& i : optionNames)
+ os << "--" << i << ' ';
+ os <<
+R"bash("
+ local cur="${COMP_WORDS[COMP_CWORD]}"
+
+ if [[ ${cur} == -* ]] ; then
+ COMPREPLY=( $(compgen -W "${options}" -- ${cur}) )
+ return 0
+ else
+ COMPREPLY=( $(compgen -W "${commands}" -- ${cur}) )
+ return 0
+ fi
+}
+complete -F _sdltool sdltool
+)bash"
+ << std::flush;
+ }
+
+ void printShortHelp(std::ostream& os,
+ const po::options_description& commonOpts,
+ CommandMap& commandMap)
+ {
+ os << "Usage: sdltool [OPTIONS] COMMAND [COMMAND SPECIFIC OPTIONS]\n\n";
+ commonOpts.print(os);
+ os << "\nAvailable commands:\n";
+ commandMap.shortHelp(os);
+ }
+
+ int privateParseAndExecute(int argc,
+ char** argv,
+ std::ostream& out,
+ std::ostream& err,
+ CommandMap& commandMap)
+ {
+ po::options_description commonHiddenOptsDescription("Common hidden options");
+ commonHiddenOptsDescription.add_options()
+ ("pos-command", po::value<std::string>(), "sdltool command to run")
+ ("pos-subargs", po::value<std::vector<std::string>>(), "Arguments for command");
+
+ po::options_description commonVisibleOptsDescription("Common options");
+ commonVisibleOptsDescription.add_options()
+ ("help", "Show help for COMMAND")
+ ("repeat", po::value<size_t>()->default_value(1U), "Times to repeat the command")
+ ("bash-completion", "Generate bash completion script")
+ ("version", "Show version information");
+
+ po::options_description commonOptsDescription("All common options");
+ commonOptsDescription.add(commonHiddenOptsDescription);
+ commonOptsDescription.add(commonVisibleOptsDescription);
+
+ po::positional_options_description commonPosOptsDescription;
+ commonPosOptsDescription
+ .add("pos-command", 1)
+ .add("pos-subargs", -1); // all positional arguments after 'command'
+
+ po::variables_map commonVarsMap;
+ po::parsed_options commonParsed(po::command_line_parser(argc, argv)
+ .options(commonOptsDescription)
+ .positional(commonPosOptsDescription)
+ .allow_unregistered()
+ .run());
+ po::store(commonParsed, commonVarsMap);
+ po::notify(commonVarsMap);
+
+ if (!commonVarsMap.count("pos-command"))
+ {
+ if (commonVarsMap.count("help"))
+ {
+ printShortHelp(out, commonVisibleOptsDescription, commandMap);
+ return EXIT_SUCCESS;
+ }
+ if (commonVarsMap.count("bash-completion"))
+ {
+ outputBashCompletion(out, commonVisibleOptsDescription, commandMap);
+ return EXIT_SUCCESS;
+ }
+ if (commonVarsMap.count("version"))
+ {
+ out << PACKAGE_STRING << std::endl;
+ return EXIT_SUCCESS;
+ }
+ err << "missing command\n\n";
+ printShortHelp(err, commonVisibleOptsDescription, commandMap);
+ return EXIT_FAILURE;
+ }
+
+ auto leftOverOpts = po::collect_unrecognized(commonParsed.options, po::include_positional);
+ if (!leftOverOpts.empty())
+ leftOverOpts.erase(leftOverOpts.begin());
+
+ const auto commandName(commonVarsMap["pos-command"].as<std::string>());
+
+ if (commonVarsMap.count("help"))
+ {
+ bool found(false);
+ try
+ {
+ commandMap.longHelp(out, commandName);
+ out << '\n';
+ found = true;
+ }
+ catch (const CommandMap::UnknownCommandName&)
+ {
+ }
+ if (!found)
+ {
+ err << "unknown command: \"" << commandName << "\"\n";
+ return EXIT_FAILURE;
+ }
+ return EXIT_SUCCESS;
+ }
+
+ const auto& commandOptions(commandMap.getCommandOptions(commandName));
+
+ po::variables_map subcmdVarsMap;
+ auto subcmdParsed(po::command_line_parser(leftOverOpts).options(commandOptions).run());
+ po::store(subcmdParsed, subcmdVarsMap);
+ po::notify(subcmdVarsMap);
+
+ return commandMap.execute(commandName,
+ out,
+ err,
+ subcmdVarsMap,
+ commonVarsMap["repeat"].as<size_t>());
+ }
+}
+
+int shareddatalayer::cli::parseAndExecute(int argc,
+ char** argv,
+ std::ostream& out,
+ std::ostream& err,
+ CommandMap& commandMap)
+{
+ try
+ {
+ return privateParseAndExecute(argc, argv, out, err, commandMap);
+ }
+ catch (const std::exception& e)
+ {
+ err << e.what() << std::endl;
+ return EXIT_FAILURE;
+ }
+}
+
diff --git a/src/cli/dumpconfigurationcommand.cpp b/src/cli/dumpconfigurationcommand.cpp
new file mode 100644
index 0000000..438f075
--- /dev/null
+++ b/src/cli/dumpconfigurationcommand.cpp
@@ -0,0 +1,83 @@
+#include <ostream>
+#include <cstdlib>
+#include <fstream>
+#include <string>
+#include <iostream>
+#include <boost/property_tree/ptree.hpp>
+#include <boost/property_tree/json_parser.hpp>
+#include "private/cli/commandmap.hpp"
+#include "private/configurationpaths.hpp"
+#include "private/configurationreader.hpp"
+
+using namespace shareddatalayer;
+using namespace shareddatalayer::cli;
+
+namespace
+{
+ bool parseConfiguration(const std::string& file)
+ {
+ boost::property_tree::ptree propertyTree;
+
+ try
+ {
+ boost::property_tree::read_json(file, propertyTree);
+ }
+ catch (const boost::property_tree::json_parser::json_parser_error& e)
+ {
+ std::ostringstream os;
+ os << "error in SDL configuration " << file << " at line " << e.line() << ": ";
+ os << e.message();
+ std::cerr << os.str().c_str() << std::endl;
+ return false;
+ }
+ return true;
+ }
+
+
+ int dumpConfigurationCommand(std::ostream& out)
+ {
+ std::string line;
+ bool status(true);
+
+ for (const auto& i : findConfigurationFiles( getDefaultConfDirectories() ))
+ {
+ std::ifstream file(i);
+
+ out << "File: " << i << std::endl;
+
+ unsigned int lineNum = 1;
+
+ if(file.is_open())
+ {
+ bool parseStatus = parseConfiguration(i);
+
+ if(status && !parseStatus)
+ status = false;
+
+ while(getline(file, line))
+ out << lineNum++ << ": " << line << std::endl;
+ file.close();
+ }
+ }
+
+ const auto var(DB_HOST_ENV_VAR_NAME);
+ const auto conf(getenv(var));
+ if (conf == nullptr)
+ out << var << " not set." << std::endl;
+ else
+ out << var << ": " << conf << std::endl;
+
+ if(!status)
+ {
+ return EXIT_FAILURE;
+ }
+ return EXIT_SUCCESS;
+ }
+}
+
+AUTO_REGISTER_COMMAND(std::bind(dumpConfigurationCommand, std::placeholders::_1),
+ "dump-configuration",
+ "Dump configuration",
+ "Find, parse and dump all shareddatalayer configuration files contents.",
+ CommandMap::Category::UTIL, 30000);
+
diff --git a/src/cli/main.cpp b/src/cli/main.cpp
new file mode 100644
index 0000000..9d84536
--- /dev/null
+++ b/src/cli/main.cpp
@@ -0,0 +1,22 @@
+#include <iostream>
+#include <exception>
+#include <cstdlib>
+#include "private/cli/commandmap.hpp"
+#include "private/cli/commandparserandexecutor.hpp"
+
+using namespace shareddatalayer;
+using namespace shareddatalayer::cli;
+
+int main(int argc, char** argv)
+{
+ try
+ {
+ return parseAndExecute(argc, argv, std::cout, std::cerr, CommandMap::getCommandMap());
+ }
+ catch (const std::exception& e)
+ {
+ std::cerr << "unexpected error: " << e.what() << std::endl;
+ return EXIT_FAILURE;
+ }
+}
+
diff --git a/src/cli/testconnectivitycommand.cpp b/src/cli/testconnectivitycommand.cpp
new file mode 100644
index 0000000..7c4b2ad
--- /dev/null
+++ b/src/cli/testconnectivitycommand.cpp
@@ -0,0 +1,196 @@
+#include <ostream>
+#include <cstdlib>
+#include <fstream>
+#include <string>
+#include <iostream>
+#include <boost/property_tree/ptree.hpp>
+#include <boost/property_tree/json_parser.hpp>
+#include <chrono>
+#include <arpa/inet.h>
+#include <sdl/asyncstorage.hpp>
+#include <boost/asio.hpp>
+#include <thread>
+#include "private/cli/commandmap.hpp"
+#include "private/configurationpaths.hpp"
+#include "private/createlogger.hpp"
+#include "private/engineimpl.hpp"
+#include "private/databaseconfigurationimpl.hpp"
+#include "private/configurationreader.hpp"
+#include "private/redis/databaseinfo.hpp"
+#include "private/asyncstorageimpl.hpp"
+#include "private/redis/asyncredisstorage.hpp"
+
+using namespace shareddatalayer;
+using namespace shareddatalayer::cli;
+using namespace shareddatalayer::redis;
+
+namespace
+{
+ void handler(std::shared_ptr<shareddatalayer::AsyncStorage> sdl, boost::asio::posix::stream_descriptor& sd)
+ {
+ sdl->handleEvents();
+ sd.async_read_some(boost::asio::null_buffers(), std::bind(handler, sdl, std::ref(sd)));
+ }
+
+ std::shared_ptr<AsyncStorage> createStorage(const std::string& nsStr, std::ostream& out)
+ {
+ try
+ {
+ std::shared_ptr<AsyncStorage> sdl(AsyncStorage::create());
+ boost::asio::io_service ios;
+ boost::asio::posix::stream_descriptor sd(ios);
+ sd.assign(sdl->fd());
+ sd.async_read_some(boost::asio::null_buffers(), std::bind(handler, sdl, std::ref(sd)));
+ sdl->waitReadyAsync(nsStr, [&ios](const std::error_code& error)
+ {
+ if (error)
+ std::cerr << "SDL waitReadyAsync failed. Error:\n" << error.message() << std::endl;
+ ios.stop();
+ });
+ ios.run();
+ sd.release();
+ out << "Storage to namespace " << nsStr << " created." << std::endl;
+ return sdl;
+ }
+ catch (const shareddatalayer::Exception& error)
+ {
+ out << "Storage create failed: " << error.what() << std::endl;
+ }
+ return nullptr;
+ }
+
+ std::string getHosts(const DatabaseConfiguration::Addresses& databaseAddresses)
+ {
+ std::string hosts("");
+ for (auto i(databaseAddresses.begin()); i != databaseAddresses.end(); ++i)
+ hosts = hosts + i->getHost() + " ";
+ return hosts;
+ }
+
+ std::string getPorts(const DatabaseConfiguration::Addresses& databaseAddresses)
+ {
+ std::string ports("");
+ for (auto i(databaseAddresses.begin()); i != databaseAddresses.end(); ++i)
+ ports = ports + std::to_string(ntohs(i->getPort())) + " ";
+ return ports;
+ }
+
+ void PrintStaticConfiguration(std::ostream& out)
+ {
+ auto engine(std::make_shared<EngineImpl>());
+ DatabaseConfigurationImpl databaseConfigurationImpl;
+ ConfigurationReader configurationReader(createLogger(SDL_LOG_PREFIX));
+ configurationReader.readDatabaseConfiguration(databaseConfigurationImpl);
+ auto staticAddresses(databaseConfigurationImpl.getServerAddresses());
+ auto defaultAddresses(databaseConfigurationImpl.getDefaultServerAddresses());
+ auto staticDbType(databaseConfigurationImpl.getDbType());
+ if (!staticAddresses.empty())
+ {
+ out << "\nStatic Server Addresses:" << std::endl;
+ out << "Static Host: " << getHosts(staticAddresses) << std::endl;
+ out << "Static Port: " << getPorts(staticAddresses) << std::endl;
+ if (staticDbType == DatabaseConfiguration::DbType::REDIS_CLUSTER)
+ out << "Static DB type: redis-cluster" << std::endl;
+ else if (staticDbType == DatabaseConfiguration::DbType::REDIS_STANDALONE)
+ out << "Static DB type: redis-standalone" << std::endl;
+ else
+ out << "Static DB type not defined" << std::endl;
+ }
+ if (!defaultAddresses.empty() && staticAddresses.empty())
+ {
+ out << "\nDefault Server Addresses:" << std::endl;
+ out << "Default Host: " << getHosts(defaultAddresses) << std::endl;
+ out << "Default Port: " << getPorts(defaultAddresses) << std::endl;
+ }
+ const auto var(DB_HOST_ENV_VAR_NAME);
+ const auto conf(getenv(var));
+ if (conf != nullptr)
+ out << var << ": " << conf << std::endl;
+ }
+
+ void PrintDatabaseInfo(const DatabaseInfo& databaseInfo, std::ostream& out)
+ {
+ out << "Used database configuration (databaseInfo):" << std::endl;
+ out << "Host: " << getHosts(databaseInfo.hosts) << std::endl;
+ out << "Port: " << getPorts(databaseInfo.hosts) << std::endl;
+ switch (databaseInfo.type)
+ {
+ case DatabaseInfo::Type::SINGLE:
+ out << "Database type: SINGLE" << std::endl;
+ break;
+ case DatabaseInfo::Type::REDUNDANT:
+ out << "Database type: REDUNDANT" << std::endl;
+ break;
+ case DatabaseInfo::Type::CLUSTER:
+ out << "Database type: CLUSTER" << std::endl;
+ break;
+ }
+ switch (databaseInfo.discovery)
+ {
+ case DatabaseInfo::Discovery::HIREDIS:
+ out << "Discovery type:: HIREDIS" << std::endl;
+ PrintStaticConfiguration(out);
+ break;
+ }
+ }
+
+ [[noreturn]] void timeoutThread(const int& timeout)
+ {
+ std::this_thread::sleep_for(std::chrono::seconds(timeout));
+ std::cerr << "Storage create timeout, aborting after " << timeout << " seconds"<< std::endl;
+ PrintStaticConfiguration(std::cerr);
+ std::exit(EXIT_FAILURE);
+ }
+
+ void setTimeout(const int& timeout)
+ {
+ if (timeout)
+ {
+ std::thread t(timeoutThread, timeout);
+ t.detach();
+ }
+ }
+
+ int TestConnectivityCommand(std::ostream& out,
+ const boost::program_options::variables_map& map)
+ {
+ const auto ns(map["ns"].as<std::string>());
+ const auto timeout(map["timeout"].as<int>());
+ setTimeout(timeout);
+ auto sdl(createStorage(ns, out));
+ if (sdl != nullptr)
+ {
+ auto asyncStorageImpl(std::dynamic_pointer_cast<AsyncStorageImpl>(sdl));
+ if (asyncStorageImpl != nullptr)
+ {
+ AsyncStorage& operationalHandler(asyncStorageImpl->getOperationHandler(ns));
+ AsyncRedisStorage* redisStorage = dynamic_cast<AsyncRedisStorage*>(&operationalHandler);
+ if (redisStorage != nullptr)
+ {
+ auto databaseinfo (redisStorage->getDatabaseInfo());
+ PrintDatabaseInfo(databaseinfo, out);
+ }
+ else
+ {
+ // @TODO Improve output for the case if dummy backend is used.
+ out << "Cannot get AsyncRedisStorage." << std::endl;
+ return EXIT_FAILURE;
+ }
+ }
+ else
+ {
+ out << "Cannot get AsyncStorageImpl." << std::endl;
+ return EXIT_FAILURE;
+ }
+ }
+ return EXIT_SUCCESS;
+ }
+}
+
+AUTO_REGISTER_COMMAND(std::bind(TestConnectivityCommand, std::placeholders::_1, std::placeholders::_3),
+ "test-connectivity",
+ "Test SDL backend connectivity",
+ "Check that SDL database backend is available and show discovered redis host address and port",
+ CommandMap::Category::UTIL, 30020,
+ ("ns", boost::program_options::value<std::string>()->default_value("sdltoolns"), "Used namespace")
+ ("timeout", boost::program_options::value<int>()->default_value(0), "Timeout (in seconds), Default is no timeout"));
diff --git a/src/cli/testgetsetcommand.cpp b/src/cli/testgetsetcommand.cpp
new file mode 100644
index 0000000..81bdd1c
--- /dev/null
+++ b/src/cli/testgetsetcommand.cpp
@@ -0,0 +1,144 @@
+#include <ostream>
+#include <cstdlib>
+#include <fstream>
+#include <string>
+#include <iostream>
+#include <boost/property_tree/ptree.hpp>
+#include <boost/property_tree/json_parser.hpp>
+#include <chrono>
+#include <thread>
+#include "private/cli/commandmap.hpp"
+#include "private/configurationpaths.hpp"
+#include <sdl/syncstorage.hpp>
+
+using namespace shareddatalayer;
+using namespace shareddatalayer::cli;
+
+namespace
+{
+ std::shared_ptr<shareddatalayer::SyncStorage> createSyncStorage(std::ostream& out)
+ {
+ try
+ {
+ auto sdl(shareddatalayer::SyncStorage::create());
+ return sdl;
+ }
+ catch (const shareddatalayer::Exception& error)
+ {
+ out << "SyncStorage create failed: " << error.what() << std::endl;
+ }
+
+ out << "Test suspended!" << std::endl;
+ return nullptr;
+ }
+
+ void execRemoveAll(shareddatalayer::SyncStorage& sdl, const std::string& nsStr, std::ostream& out)
+ {
+ try
+ {
+ sdl.removeAll(nsStr);
+ }
+ catch (const shareddatalayer::Exception& error)
+ {
+ out << "removeAll failed: " << error.what() << std::endl;
+ }
+ }
+
+ void execSet(shareddatalayer::SyncStorage& sdl, const std::string& nsStr, const std::string& key, const std::vector<uint8_t>& val, std::ostream& out)
+ {
+ try
+ {
+ sdl.set(nsStr, { { key, val } });
+ }
+ catch (const shareddatalayer::Exception& error)
+ {
+ out << "Set " << key << " failed: " << error.what() << std::endl;
+ }
+ }
+
+ void execGet(shareddatalayer::SyncStorage& sdl, const std::string& nsStr, const std::string& key, const std::vector<uint8_t>& val, std::ostream& out)
+ {
+ try
+ {
+ auto map(sdl.get(nsStr, { key }));
+ auto i(map.find(key));
+ if (i == map.end())
+ out << "Get " << key << ": Not found!" << std::endl;
+ else if (i->second != val)
+ out << "Get " << key << ": Wrong value!" << std::endl;
+ }
+ catch (const shareddatalayer::Exception& error)
+ {
+ out << "Get " << key << " failed: " << error.what() << std::endl;
+ }
+ }
+
+ void timeoutThread(const int& timeout)
+ {
+ std::this_thread::sleep_for(std::chrono::seconds(timeout));
+ std::cerr << "SyncStorage create timeout, aborting after " << timeout << " seconds"<< std::endl;
+ std::exit(EXIT_FAILURE);
+ }
+
+ void setTimeout(const int& timeout)
+ {
+ if (timeout)
+ {
+ std::thread t(timeoutThread, timeout);
+ t.detach();
+ }
+ }
+
+ int TestGetSetCommand(std::ostream& out, const boost::program_options::variables_map& map)
+ {
+ auto keyCount(map["key-count"].as<int>());
+ const auto timeout(map["timeout"].as<int>());
+ auto ns("sdltoolns");
+ setTimeout(timeout);
+ auto sdl(createSyncStorage(out));
+ if (sdl == nullptr)
+ return EXIT_FAILURE;
+
+ out << "namespace\t"
+ << "key\t"
+ << "value\t"
+ << "Write\tRead" << std::endl;
+
+ for (uint8_t val(0); 0 < keyCount--; ++val)
+ {
+ auto key("key_" + std::to_string(val));
+
+ auto wStart(std::chrono::high_resolution_clock::now());
+ execSet(std::ref(*sdl), ns, key, {val}, out);
+ auto wEnd(std::chrono::high_resolution_clock::now());
+ auto writeLatency_us = std::chrono::duration_cast<std::chrono::microseconds>(wEnd - wStart);
+
+ auto rStart(std::chrono::high_resolution_clock::now());
+ execGet(std::ref(*sdl), ns, key, {val}, out);
+ auto rEnd(std::chrono::high_resolution_clock::now());
+ auto readLatency_us = std::chrono::duration_cast<std::chrono::microseconds>(rEnd - rStart);
+
+ out << ns << '\t'
+ << key << '\t'
+ << std::dec << static_cast<int>(val) << "\t"
+ << std::dec << writeLatency_us.count() << "\t"
+ << std::dec << readLatency_us.count() << std::endl;
+ }
+
+ auto start(std::chrono::high_resolution_clock::now());
+ execRemoveAll(std::ref(*sdl), ns, out);
+ auto end(std::chrono::high_resolution_clock::now());
+ auto used_us = std::chrono::duration_cast<std::chrono::microseconds>(end - start);
+ out << "All test keys removed in " << used_us.count() << " microseconds" << std::endl;
+
+ return EXIT_SUCCESS;
+ }
+}
+
+AUTO_REGISTER_COMMAND(std::bind(TestGetSetCommand, std::placeholders::_1, std::placeholders::_3),
+ "test-get-set",
+ "Write and read to DB and check latency",
+ "Check that basic SDL api commands (set/get) works normally and measure latency.",
+ CommandMap::Category::UTIL, 30010,
+ ("key-count", boost::program_options::value<int>()->default_value(10), "Number of write/read keys")
+ ("timeout", boost::program_options::value<int>()->default_value(0), "Timeout (in seconds), Default is no timeout"));
diff --git a/src/configurationpaths.cpp b/src/configurationpaths.cpp
new file mode 100644
index 0000000..7ffaabb
--- /dev/null
+++ b/src/configurationpaths.cpp
@@ -0,0 +1,59 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "config.h"
+#include "private/configurationpaths.hpp"
+#include <algorithm>
+#include <boost/filesystem.hpp>
+#include <boost/range/iterator_range.hpp>
+#include <boost/algorithm/string/predicate.hpp>
+
+using namespace shareddatalayer;
+
+namespace
+{
+ void findConfigurationFilesFromDirectory(const std::string& path, std::vector<std::string>& paths)
+ {
+ try
+ {
+ for(const auto& i : boost::make_iterator_range(boost::filesystem::directory_iterator(path), { }))
+ {
+ const auto filename(i.path().filename().native());
+ if ((!boost::filesystem::is_directory(i.path())) &&
+ (filename[0] != '.') &&
+ (boost::algorithm::ends_with(filename, ".json")))
+ paths.push_back(i.path().native());
+ }
+ }
+ catch (const boost::filesystem::filesystem_error &)
+ {
+ }
+ }
+}
+
+Directories shareddatalayer::getDefaultConfDirectories()
+{
+ return Directories({ SDL_CONF_DIR, "/run/" PACKAGE_NAME ".d" });
+}
+
+Files shareddatalayer::findConfigurationFiles(const Directories& directories)
+{
+ std::vector<std::string> paths;
+ for (const auto& i : directories)
+ findConfigurationFilesFromDirectory(i, paths);
+ std::sort(paths.begin(), paths.end());
+ return paths;
+}
diff --git a/src/configurationreader.cpp b/src/configurationreader.cpp
new file mode 100644
index 0000000..85ab627
--- /dev/null
+++ b/src/configurationreader.cpp
@@ -0,0 +1,323 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/abort.hpp"
+#include "private/configurationreader.hpp"
+#include <boost/property_tree/json_parser.hpp>
+#include <sdl/exception.hpp>
+#include "private/createlogger.hpp"
+#include "private/databaseconfiguration.hpp"
+#include "private/logger.hpp"
+#include "private/namespaceconfigurations.hpp"
+#include "private/namespacevalidator.hpp"
+#include "private/system.hpp"
+
+using namespace shareddatalayer;
+
+namespace
+{
+ template <typename T>
+ T get(const boost::property_tree::ptree& ptree, const std::string& param, const std::string& sourceName)
+ {
+ try
+ {
+ return ptree.get<T>(param);
+ }
+ catch (const boost::property_tree::ptree_bad_path&)
+ {
+ std::ostringstream os;
+ os << "Configuration error in " << sourceName << ": "
+ << "missing \"" << param << '\"';
+ throw Exception(os.str());
+ }
+ catch (const boost::property_tree::ptree_bad_data& e)
+ {
+ std::ostringstream os;
+ os << "Configuration error in " << sourceName << ": "
+ << "invalid \"" << param << "\": \"" << e.data<boost::property_tree::ptree::data_type>() << '\"';
+ throw Exception(os.str());
+ }
+ }
+
+ void validateAndSetDbType(const std::string& type, DatabaseConfiguration& databaseConfiguration,
+ const std::string& sourceName)
+ {
+ try
+ {
+ databaseConfiguration.checkAndApplyDbType(type);
+ }
+ catch (const std::exception& e)
+ {
+ std::ostringstream os;
+ os << "Configuration error in " << sourceName << ": "
+ << e.what();
+ throw Exception(os.str());
+ }
+ }
+
+ void validateAndSetDbServerAddress(const std::string& address, DatabaseConfiguration& databaseConfiguration,
+ const std::string& sourceName)
+ {
+ try
+ {
+ databaseConfiguration.checkAndApplyServerAddress(address);
+ }
+ catch (const std::exception& e)
+ {
+ std::ostringstream os;
+ os << "Configuration error in " << sourceName << ": "
+ << "invalid \"address\": \"" << address << "\" " << e.what();
+ throw Exception(os.str());
+ }
+ }
+
+ void parseDatabaseServerConfiguration(DatabaseConfiguration& databaseConfiguration,
+ const boost::property_tree::ptree& ptree,
+ const std::string& sourceName)
+ {
+ const auto address(get<std::string>(ptree, "address", sourceName));
+ validateAndSetDbServerAddress(address, databaseConfiguration, sourceName);
+ }
+
+ void parseDatabaseServersConfiguration(DatabaseConfiguration& databaseConfiguration,
+ const boost::property_tree::ptree& ptree,
+ const std::string& sourceName)
+ {
+ const auto servers(ptree.get_child_optional("servers"));
+ if (servers)
+ for(const auto& server : *servers)
+ parseDatabaseServerConfiguration(databaseConfiguration, server.second, sourceName);
+ else
+ {
+ std::ostringstream os;
+ os << "Configuration error in " << sourceName << ": "
+ << "missing \"servers\"";
+ throw Exception(os.str());
+ }
+ }
+
+ void parseDatabaseConfiguration(DatabaseConfiguration& databaseConfiguration,
+ const boost::property_tree::ptree& ptree,
+ const std::string& sourceName)
+ {
+ const auto type(get<std::string>(ptree, "type", sourceName));
+ validateAndSetDbType(type, databaseConfiguration, sourceName);
+
+ parseDatabaseServersConfiguration(databaseConfiguration, ptree, sourceName);
+ }
+
+ void parseDatabaseConfigurationTree(DatabaseConfiguration& databaseConfiguration,
+ const boost::optional<boost::property_tree::ptree>& databaseConfigurationPtree,
+ const std::string& sourceName)
+ {
+ if (databaseConfigurationPtree)
+ parseDatabaseConfiguration(databaseConfiguration, *databaseConfigurationPtree, sourceName);
+ }
+
+ void parseDatabaseServersConfigurationFromString(DatabaseConfiguration& databaseConfiguration,
+ const std::string& serverConfiguration,
+ const std::string& sourceName)
+ {
+ size_t base(0);
+ auto done(false);
+ do
+ {
+ auto split = serverConfiguration.find(',', base);
+ done = std::string::npos == split;
+ validateAndSetDbServerAddress(serverConfiguration.substr(base, done ? std::string::npos : split-base),
+ databaseConfiguration,
+ sourceName);
+ base = split+1;
+ } while (!done);
+ }
+
+ void validateNamespacePrefix(const std::string& prefix,
+ const std::string& sourceName)
+ {
+ if (!isValidNamespaceSyntax(prefix))
+ {
+ std::ostringstream os;
+ os << "Configuration error in " << sourceName << ": "
+ << "\"namespacePrefix\": \"" << prefix << "\""
+ << " contains some of these disallowed characters: "
+ << getDisallowedCharactersInNamespace();
+ throw Exception(os.str());
+ }
+ }
+
+ void validateEnableNotifications(bool enableNotifications, bool useDbBackend,
+ const std::string& sourceName)
+ {
+ if (enableNotifications && !useDbBackend)
+ {
+ std::ostringstream os;
+ os << "Configuration error in " << sourceName << ": "
+ << "\"enableNotifications\" cannot be true, when \"useDbBackend\" is false";
+ throw Exception(os.str());
+ }
+ }
+
+ void parseNsConfiguration(NamespaceConfigurations& namespaceConfigurations,
+ const std::string& namespacePrefix,
+ const boost::property_tree::ptree& ptree,
+ const std::string& sourceName)
+ {
+ const auto useDbBackend(get<bool>(ptree, "useDbBackend", sourceName));
+ const auto enableNotifications(get<bool>(ptree, "enableNotifications", sourceName));
+
+ validateNamespacePrefix(namespacePrefix, sourceName);
+ validateEnableNotifications(enableNotifications, useDbBackend, sourceName);
+
+ namespaceConfigurations.addNamespaceConfiguration({namespacePrefix, useDbBackend, enableNotifications, sourceName});
+ }
+
+ void parseNsConfigurationMap(NamespaceConfigurations& namespaceConfigurations,
+ std::unordered_map<std::string, std::pair<boost::property_tree::ptree, std::string>>& namespaceConfigurationMap)
+ {
+ for (const auto &namespaceConfigurationMapItem : namespaceConfigurationMap )
+ parseNsConfiguration(namespaceConfigurations, namespaceConfigurationMapItem.first, namespaceConfigurationMapItem.second.first, namespaceConfigurationMapItem.second.second);
+ }
+}
+
+ConfigurationReader::ConfigurationReader(std::shared_ptr<Logger> logger):
+ ConfigurationReader(getDefaultConfDirectories(), System::getSystem(), logger)
+{
+}
+
+ConfigurationReader::ConfigurationReader(const Directories& directories,
+ System& system,
+ std::shared_ptr<Logger> logger):
+ dbHostEnvVariableName(DB_HOST_ENV_VAR_NAME),
+ dbHostEnvVariableValue({}),
+ dbPortEnvVariableName(DB_PORT_ENV_VAR_NAME),
+ dbPortEnvVariableValue({}),
+ jsonDatabaseConfiguration(boost::none),
+ logger(logger)
+{
+ auto envStr = system.getenv(dbHostEnvVariableName.c_str());
+ if (envStr)
+ {
+ dbHostEnvVariableValue = envStr;
+ sourceForDatabaseConfiguration = dbHostEnvVariableName;
+ auto envStr = system.getenv(dbPortEnvVariableName.c_str());
+ if (envStr)
+ dbPortEnvVariableValue = envStr;
+ }
+
+ readConfigurationFromDirectories(directories);
+}
+
+ConfigurationReader::~ConfigurationReader()
+{
+}
+
+void ConfigurationReader::readConfigurationFromDirectories(const Directories& directories)
+{
+ for (const auto& i : findConfigurationFiles(directories))
+ readConfiguration(i, i);
+}
+
+void ConfigurationReader::readConfigurationFromInputStream(const std::istream& input)
+{
+ jsonNamespaceConfigurations.clear();
+ readConfiguration(const_cast<std::istream&>(input), "<istream>");
+}
+
+template<typename T>
+void ConfigurationReader::readConfiguration(T& input, const std::string& currentSourceName)
+{
+ boost::property_tree::ptree propertyTree;
+
+ try
+ {
+ boost::property_tree::read_json(input, propertyTree);
+ }
+ catch (const boost::property_tree::json_parser::json_parser_error& e)
+ {
+ std::ostringstream os;
+ os << "error in SDL configuration " << currentSourceName << " at line " << e.line() << ": ";
+ os << e.message();
+ logger->error() << os.str();
+ throw Exception(os.str());
+ }
+
+ // Environment variable configuration overrides json configuration
+ if (sourceForDatabaseConfiguration != dbHostEnvVariableName)
+ {
+ const auto databaseConfiguration(propertyTree.get_child_optional("database"));
+ if (databaseConfiguration)
+ {
+ jsonDatabaseConfiguration = databaseConfiguration;
+ sourceForDatabaseConfiguration = currentSourceName;
+ }
+ }
+
+ const auto namespaceConfigurations(propertyTree.get_child_optional("sharedDataLayer"));
+ if (namespaceConfigurations)
+ {
+ for(const auto& namespaceConfiguration : *namespaceConfigurations)
+ {
+ const auto namespacePrefix = get<std::string>(namespaceConfiguration.second, "namespacePrefix", currentSourceName);
+ jsonNamespaceConfigurations[namespacePrefix] = std::make_pair(namespaceConfiguration.second, currentSourceName);
+ }
+ }
+}
+
+void ConfigurationReader::readDatabaseConfiguration(DatabaseConfiguration& databaseConfiguration)
+{
+ if (!databaseConfiguration.isEmpty())
+ SHAREDDATALAYER_ABORT("Database configuration can be read only to empty container");
+
+ try
+ {
+ if (sourceForDatabaseConfiguration == dbHostEnvVariableName)
+ {
+ // Currently hard coded to redis-standalone, because RIC dbaas does not support Redis cluster configuration.
+ validateAndSetDbType("redis-standalone", databaseConfiguration, sourceForDatabaseConfiguration);
+ if (dbPortEnvVariableValue.empty())
+ parseDatabaseServersConfigurationFromString(databaseConfiguration, dbHostEnvVariableValue, sourceForDatabaseConfiguration);
+ else
+ parseDatabaseServersConfigurationFromString(databaseConfiguration, dbHostEnvVariableValue + ":" + dbPortEnvVariableValue, sourceForDatabaseConfiguration);
+ }
+ else
+ parseDatabaseConfigurationTree(databaseConfiguration, jsonDatabaseConfiguration, sourceForDatabaseConfiguration);
+ }
+ catch (const std::exception& e)
+ {
+ logger->error() << e.what();
+ throw;
+ }
+}
+
+void ConfigurationReader::readNamespaceConfigurations(NamespaceConfigurations& namespaceConfigurations)
+{
+ if (!namespaceConfigurations.isEmpty())
+ SHAREDDATALAYER_ABORT("Namespace configurations can be read only to empty container");
+
+ try
+ {
+ parseNsConfigurationMap(namespaceConfigurations, jsonNamespaceConfigurations);
+ }
+ catch(const std::exception& e)
+ {
+ logger->error() << e.what();
+ throw;
+ }
+}
+
+
+template void ConfigurationReader::readConfiguration(const std::string&, const std::string&);
+template void ConfigurationReader::readConfiguration(std::istream&, const std::string&);
diff --git a/src/createlogger.cpp b/src/createlogger.cpp
new file mode 100644
index 0000000..cd85f02
--- /dev/null
+++ b/src/createlogger.cpp
@@ -0,0 +1,52 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "config.h"
+#include <memory>
+#include <string>
+#include "private/createlogger.hpp"
+#if HAVE_SYSTEMLOGGER
+#include "private/systemlogger.hpp"
+#endif
+
+using namespace shareddatalayer;
+
+std::shared_ptr<Logger> shareddatalayer::createLogger(const std::string& prefix)
+{
+#if HAVE_SYSTEMLOGGER
+ return std::shared_ptr<Logger>(new SystemLogger(prefix));
+#else
+#error "Need to compile with at least one logging backend"
+#endif
+}
+
+void shareddatalayer::logDebugOnce(const std::string& msg) noexcept
+{
+ auto logger(createLogger(SDL_LOG_PREFIX));
+ logger->debug() << msg;
+}
+
+void shareddatalayer::logErrorOnce(const std::string& msg) noexcept
+{
+ auto logger(createLogger(SDL_LOG_PREFIX));
+ logger->error() << msg;
+}
+
+void shareddatalayer::logInfoOnce(const std::string& msg) noexcept
+{
+ auto logger(createLogger(SDL_LOG_PREFIX));
+ logger->info() << msg;
+}
diff --git a/src/databaseconfiguration.cpp b/src/databaseconfiguration.cpp
new file mode 100644
index 0000000..f51203b
--- /dev/null
+++ b/src/databaseconfiguration.cpp
@@ -0,0 +1,37 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/databaseconfiguration.hpp"
+#include <sstream>
+
+using namespace shareddatalayer;
+
+namespace
+{
+ std::string buildInvalidDbTypeError(const std::string& type)
+ {
+ std::ostringstream os;
+ os << "invalid database type: '" << type << "'. ";
+ os << "Allowed types are: 'redis-standalone' or 'redis-cluster'";
+ return os.str();
+ }
+}
+
+DatabaseConfiguration::InvalidDbType::InvalidDbType(const std::string& type):
+ Exception(buildInvalidDbTypeError(type))
+{
+}
+
diff --git a/src/databaseconfigurationimpl.cpp b/src/databaseconfigurationimpl.cpp
new file mode 100644
index 0000000..f1fb2be
--- /dev/null
+++ b/src/databaseconfigurationimpl.cpp
@@ -0,0 +1,75 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/databaseconfigurationimpl.hpp"
+#include <arpa/inet.h>
+
+using namespace shareddatalayer;
+
+namespace
+{
+ const std::string& getDefaultHost()
+ {
+ static const std::string defaultHost("localhost");
+ return defaultHost;
+ }
+
+ const uint16_t DEFAULT_PORT(6379U);
+}
+
+DatabaseConfigurationImpl::DatabaseConfigurationImpl():
+ dbType(DbType::UNKNOWN)
+{
+}
+
+DatabaseConfigurationImpl::~DatabaseConfigurationImpl()
+{
+}
+
+void DatabaseConfigurationImpl::checkAndApplyDbType(const std::string& type)
+{
+ if (type == "redis-standalone")
+ dbType = DatabaseConfiguration::DbType::REDIS_STANDALONE;
+ else if (type == "redis-cluster")
+ dbType = DatabaseConfiguration::DbType::REDIS_CLUSTER;
+ else
+ throw DatabaseConfiguration::InvalidDbType(type);
+}
+
+DatabaseConfiguration::DbType DatabaseConfigurationImpl::getDbType() const
+{
+ return dbType;
+}
+
+void DatabaseConfigurationImpl::checkAndApplyServerAddress(const std::string& address)
+{
+ serverAddresses.push_back(HostAndPort(address, htons(DEFAULT_PORT)));
+}
+
+bool DatabaseConfigurationImpl::isEmpty() const
+{
+ return serverAddresses.empty();
+}
+
+DatabaseConfiguration::Addresses DatabaseConfigurationImpl::getServerAddresses() const
+{
+ return serverAddresses;
+}
+
+DatabaseConfiguration::Addresses DatabaseConfigurationImpl::getDefaultServerAddresses() const
+{
+ return { HostAndPort(getDefaultHost(), htons(DEFAULT_PORT)) };
+}
diff --git a/src/emptynamespace.cpp b/src/emptynamespace.cpp
new file mode 100644
index 0000000..8ec4ebb
--- /dev/null
+++ b/src/emptynamespace.cpp
@@ -0,0 +1,24 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include <sdl/emptynamespace.hpp>
+
+using namespace shareddatalayer;
+
+EmptyNamespace::EmptyNamespace():
+ Exception("empty namespace string given")
+{
+}
diff --git a/src/engine.cpp b/src/engine.cpp
new file mode 100644
index 0000000..0e2c067
--- /dev/null
+++ b/src/engine.cpp
@@ -0,0 +1,25 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/engine.hpp"
+#include <sys/epoll.h>
+
+using namespace shareddatalayer;
+
+const unsigned int Engine::EVENT_IN(EPOLLIN);
+const unsigned int Engine::EVENT_OUT(EPOLLOUT);
+const unsigned int Engine::EVENT_ERR(EPOLLERR);
+const unsigned int Engine::EVENT_HUP(EPOLLHUP);
diff --git a/src/engineimpl.cpp b/src/engineimpl.cpp
new file mode 100644
index 0000000..07bbe53
--- /dev/null
+++ b/src/engineimpl.cpp
@@ -0,0 +1,155 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/engineimpl.hpp"
+#include "private/abort.hpp"
+#include "private/timerfd.hpp"
+#include "private/eventfd.hpp"
+#include "private/system.hpp"
+
+using namespace shareddatalayer;
+
+EngineImpl::EngineImpl():
+ EngineImpl(System::getSystem())
+{
+}
+
+EngineImpl::EngineImpl(System& system):
+ system(system),
+ stopped(false),
+ epollFD(system, system.epoll_create1(EPOLL_CLOEXEC))
+{
+}
+
+EngineImpl::~EngineImpl()
+{
+}
+
+void EngineImpl::handleEvents()
+{
+ if (!handlers.empty())
+ epollWait(0);
+}
+
+void EngineImpl::epollWait(int timeout)
+{
+ ebuffer.resize(handlers.size());
+ const int count(system.epoll_wait(epollFD, ebuffer.data(), static_cast<int>(ebuffer.size()), timeout));
+ if (count <= 0)
+ return;
+ ebuffer.resize(count);
+ for (const auto& i : ebuffer)
+ if (i.data.fd != -1)
+ callHandler(i);
+}
+
+void EngineImpl::callHandler(const epoll_event& e)
+{
+ handlers[e.data.fd](e.events);
+}
+
+void EngineImpl::addMonitoredFD(int fd, unsigned int events, const EventHandler& eh)
+{
+ if (handlers.find(fd) != handlers.end())
+ SHAREDDATALAYER_ABORT("Monitored fd has already been added");
+
+ epoll_event e = { };
+ e.events = events;
+ e.data.fd = fd;
+ system.epoll_ctl(epollFD, EPOLL_CTL_ADD, fd, &e);
+ handlers.insert(std::make_pair(fd, eh));
+}
+
+void EngineImpl::addMonitoredFD(FileDescriptor& fd, unsigned int events, const EventHandler& eh)
+{
+ int native(fd);
+ addMonitoredFD(native, events, eh);
+
+ fd.atClose([this] (int fd)
+ {
+ deleteMonitoredFD(fd);
+ });
+}
+
+void EngineImpl::modifyMonitoredFD(int fd, unsigned int events)
+{
+ if (handlers.find(fd) == handlers.end())
+ SHAREDDATALAYER_ABORT("Modified monitored fd does not exist");
+
+ epoll_event e = { };
+ e.events = events;
+ e.data.fd = fd;
+ system.epoll_ctl(epollFD, EPOLL_CTL_MOD, fd, &e);
+}
+
+void EngineImpl::deleteMonitoredFD(int fd)
+{
+ const auto i(handlers.find(fd));
+ if (i == handlers.end())
+ SHAREDDATALAYER_ABORT("Monitored (to be deleted) fd does not exist");
+
+ handlers.erase(i);
+ for (auto& i : ebuffer)
+ if (i.data.fd == fd)
+ {
+ i.data.fd = -1;
+ break;
+ }
+ system.epoll_ctl(epollFD, EPOLL_CTL_DEL, fd, nullptr);
+}
+
+void EngineImpl::armTimer(Timer& timer, const Timer::Duration& duration, const Timer::Callback& cb)
+{
+ getTimerFD().arm(timer, duration, cb);
+}
+
+void EngineImpl::disarmTimer(const Timer& timer)
+{
+ getTimerFD().disarm(timer);
+}
+
+void EngineImpl::postCallback(const Callback& callback)
+{
+ getEventFD().post(callback);
+}
+
+void EngineImpl::run()
+{
+ while (!stopped)
+ epollWait(-1);
+ stopped = false;
+}
+
+void EngineImpl::stop()
+{
+ postCallback([this] () { stopped = true; });
+}
+
+TimerFD& EngineImpl::getTimerFD()
+{
+ if (!timerFD)
+ timerFD.reset(new TimerFD(system, *this));
+
+ return *timerFD;
+}
+
+EventFD& EngineImpl::getEventFD()
+{
+ if (!eventFD)
+ eventFD.reset(new EventFD(system, *this));
+
+ return *eventFD;
+}
diff --git a/src/error.cpp b/src/error.cpp
new file mode 100644
index 0000000..347e651
--- /dev/null
+++ b/src/error.cpp
@@ -0,0 +1,182 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include <sstream>
+#include "private/createlogger.hpp"
+#include "private/error.hpp"
+
+using namespace shareddatalayer;
+using namespace shareddatalayer::redis;
+
+namespace
+{
+ /* Error codes under this category are not set directly. All SDL implementation specific error codes can be
+ * mapped to these error codes. These error codes are further on mapped to error codes which are available to
+ * SDL clients (shareddatalayer::Error category).
+ * We could directly map implementation specific error codes to client error codes but having this intermediate
+ * mapping gives some benefits:
+ * - We can easily provide more error categories for clients (e.g. severity, etc.) than just client error code
+ * classification if needed.
+ * - We can implement SDL internal error handling logic based on these internal error codes if needed.
+ * - If implementation specific error would be directly mapped to client error codes, mapping implementation would
+ * easily be quite complicated (especially if the amount of implementation specific errors/error categories increases).
+ */
+ class InternalErrorCategory : public std::error_category
+ {
+ public:
+ InternalErrorCategory() = default;
+ const char* name() const noexcept override;
+ std::string message(int condition) const override;
+ };
+
+ const char* InternalErrorCategory::name() const noexcept
+ {
+ return "SDL-internal-errorcodes";
+ }
+
+ std::string InternalErrorCategory::message(int) const
+ {
+ return "Only for SDL internal usage.";
+ }
+
+ const std::error_category& getInternalErrorCategory() noexcept
+ {
+ static const InternalErrorCategory theInternalErrorCategory;
+ return theInternalErrorCategory;
+ }
+
+ /* This error category is used by both AsyncHiredisCommandDispatcher and AsyncHiredisClusterCommandDispatcher,
+ * thus it is defined here. Error categories related to single class are defined in the files of the corresponding class.
+ * AsyncHiredisCommandDispatcher and AsyncHiredisClusterCommandDispatcher can use common error category as error
+ * handling is identical in those two classes. Also, only one of those classes is always used at a time (depending
+ * on deployment).
+ */
+ class AsyncRedisCommandDispatcherErrorCategory: public std::error_category
+ {
+ public:
+ AsyncRedisCommandDispatcherErrorCategory() = default;
+
+ const char* name() const noexcept override;
+
+ std::string message(int condition) const override;
+
+ std::error_condition default_error_condition(int condition) const noexcept override;
+ };
+
+ const char* AsyncRedisCommandDispatcherErrorCategory::name() const noexcept
+ {
+ /* As the correct dispacther is selected during runtime, we do not known here (without additional implementation)
+ * which dispatcher (redis/rediscluster) is currently in use. At least for now, we do not indicate in error
+ * category name the exact dispacther type but return the same name for all dispatchers.
+ * Main reason for this decision was that in error investigation situations we anyway need to have some other efficient
+ * way to figure out what kind of deployment was used as there can be error situations which do not generate any error
+ * code. Thus it was not seen worth the errort to implement correct dispatcher name display to error category name.
+ * Detailed dispacther name display can be added later if a need for that arises.
+ */
+ return "asyncrediscommanddispatcher";
+ }
+
+ std::string AsyncRedisCommandDispatcherErrorCategory::message(int condition) const
+ {
+ switch (static_cast<AsyncRedisCommandDispatcherErrorCode>(condition))
+ {
+ case AsyncRedisCommandDispatcherErrorCode::SUCCESS:
+ return std::error_code().message();
+ case AsyncRedisCommandDispatcherErrorCode::CONNECTION_LOST:
+ return "redis connection lost";
+ case AsyncRedisCommandDispatcherErrorCode::PROTOCOL_ERROR:
+ return "redis protocol error";
+ case AsyncRedisCommandDispatcherErrorCode::OUT_OF_MEMORY:
+ return "redis out of memory";
+ case AsyncRedisCommandDispatcherErrorCode::DATASET_LOADING:
+ return "redis dataset still being loaded into memory";
+ case AsyncRedisCommandDispatcherErrorCode::NOT_CONNECTED:
+ return "not connected to redis, SDL operation not started";
+ case AsyncRedisCommandDispatcherErrorCode::UNKNOWN_ERROR:
+ return "redis error";
+ case AsyncRedisCommandDispatcherErrorCode::IO_ERROR:
+ return "redis I/O error";
+ case AsyncRedisCommandDispatcherErrorCode::END_MARKER:
+ logErrorOnce("AsyncRedisCommandDispatcherErrorCode::END_MARKER is not meant to be queried (it is only for enum loop control)");
+ return "unsupported error code for message()";
+ default:
+ return "description missing for AsyncRedisCommandDispatcherErrorCategory error: " + std::to_string(condition);
+ }
+ }
+
+ std::error_condition AsyncRedisCommandDispatcherErrorCategory::default_error_condition(int condition) const noexcept
+ {
+ switch (static_cast<AsyncRedisCommandDispatcherErrorCode>(condition))
+ {
+ case AsyncRedisCommandDispatcherErrorCode::SUCCESS:
+ return InternalError::SUCCESS;
+ case AsyncRedisCommandDispatcherErrorCode::CONNECTION_LOST:
+ return InternalError::BACKEND_CONNECTION_LOST;
+ case AsyncRedisCommandDispatcherErrorCode::PROTOCOL_ERROR:
+ return InternalError::BACKEND_REJECTED_REQUEST;
+ case AsyncRedisCommandDispatcherErrorCode::OUT_OF_MEMORY:
+ return InternalError::BACKEND_ERROR;
+ case AsyncRedisCommandDispatcherErrorCode::DATASET_LOADING:
+ return InternalError::BACKEND_NOT_READY;
+ case AsyncRedisCommandDispatcherErrorCode::NOT_CONNECTED:
+ return InternalError::SDL_NOT_CONNECTED_TO_BACKEND;
+ case AsyncRedisCommandDispatcherErrorCode::UNKNOWN_ERROR:
+ return InternalError::BACKEND_ERROR;
+ case AsyncRedisCommandDispatcherErrorCode::IO_ERROR:
+ return InternalError::BACKEND_ERROR;
+ case AsyncRedisCommandDispatcherErrorCode::END_MARKER:
+ logErrorOnce("AsyncRedisCommandDispatcherErrorCode::END_MARKER is not meant to be mapped to InternalError (it is only for enum loop control)");
+ return InternalError::SDL_ERROR_CODE_LOGIC_ERROR;
+ default:
+ std::ostringstream msg;
+ msg << "default error condition missing for AsyncRedisCommandDispatcherErrorCategory error: "
+ << condition;
+ logErrorOnce(msg.str());
+ return InternalError::SDL_ERROR_CODE_LOGIC_ERROR;
+ }
+ }
+
+ const std::error_category& getAsyncRedisCommandDispatcherErrorCategory() noexcept
+ {
+ static const AsyncRedisCommandDispatcherErrorCategory theAsyncRedisCommandDispatcherErrorCategory;
+ return theAsyncRedisCommandDispatcherErrorCategory;
+ }
+}
+
+namespace shareddatalayer
+{
+ std::error_condition make_error_condition(InternalError errorCode)
+ {
+ return {static_cast<int>(errorCode), getInternalErrorCategory()};
+ }
+
+ namespace redis
+ {
+ std::error_code make_error_code(AsyncRedisCommandDispatcherErrorCode errorCode)
+ {
+ return std::error_code(static_cast<int>(errorCode), getAsyncRedisCommandDispatcherErrorCategory());
+ }
+
+ AsyncRedisCommandDispatcherErrorCode& operator++ (AsyncRedisCommandDispatcherErrorCode& ecEnum)
+ {
+ if (ecEnum == AsyncRedisCommandDispatcherErrorCode::END_MARKER)
+ throw std::out_of_range("for AsyncRedisCommandDispatcherErrorCode& operator ++");
+
+ ecEnum = AsyncRedisCommandDispatcherErrorCode(static_cast<std::underlying_type<AsyncRedisCommandDispatcherErrorCode>::type>(ecEnum) + 1);
+ return ecEnum;
+ }
+ }
+}
diff --git a/src/errorqueries.cpp b/src/errorqueries.cpp
new file mode 100644
index 0000000..55ba2c1
--- /dev/null
+++ b/src/errorqueries.cpp
@@ -0,0 +1,120 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include <sstream>
+#include "config.h"
+#include <sdl/errorqueries.hpp>
+#include "private/createlogger.hpp"
+#include "private/error.hpp"
+
+using namespace shareddatalayer;
+
+namespace
+{
+ class SharedDataLayerErrorCategory : public std::error_category
+ {
+ public:
+ SharedDataLayerErrorCategory() = default;
+ const char* name() const noexcept override;
+ std::string message(int condition) const override;
+ bool equivalent(const std::error_code& ec, int condition) const noexcept override;
+ };
+
+ const char* SharedDataLayerErrorCategory::name() const noexcept
+ {
+ return "shareddatalayer-errorcodes";
+ }
+
+ std::string SharedDataLayerErrorCategory::message(int condition) const
+ {
+ switch (static_cast<shareddatalayer::Error>(condition))
+ {
+ case shareddatalayer::Error::SUCCESS:
+ return std::error_code().message();
+ case shareddatalayer::Error::NOT_CONNECTED:
+ return "shareddatalayer not connected to backend data storage";
+ case shareddatalayer::Error::OPERATION_INTERRUPTED:
+ return "shareddatalayer sent the request to backend data storage but did not receive reply";
+ case shareddatalayer::Error::BACKEND_FAILURE:
+ return "backend data storage failed to process the request";
+ case shareddatalayer::Error::REJECTED_BY_BACKEND:
+ return "backend data storage rejected the request";
+ case shareddatalayer::Error::REJECTED_BY_SDL:
+ return "SDL rejected the request";
+ default:
+ return "description missing for SharedDataLayerErrorCategory error: " + std::to_string(condition);
+ }
+ }
+
+ const std::error_category& getSharedDataLayerErrorCategory() noexcept
+ {
+ static const SharedDataLayerErrorCategory theSharedDataLayerErrorCategory;
+ return theSharedDataLayerErrorCategory;
+ }
+
+
+ bool SharedDataLayerErrorCategory::equivalent(const std::error_code& ec, int condition) const noexcept
+ {
+ /* Reasoning for possibly not self-evident mappings:
+ - InternalError::BACKEND_NOT_READY is mapped to shareddatalayer::Error::NOT_CONNECTED
+ even though we are connected to backend in that situation. Client handling for InternalError::BACKEND_NOT_READY
+ situation is identical than handling for other shareddatalayer::Error::NOT_CONNECTED client error cases.
+ To have minimal amount of client error codes we map it to that.
+ - InternalError::SDL_ERROR_CODE_LOGIC_ERROR is mapped to shareddatalayer::Error::BACKEND_FAILURE.
+ That internal failure should never happen. If it does happen, we cannot know here which kind of error
+ has really happened. Thus we map to the most severe client error code (other places will write logs about this).
+ */
+ switch (static_cast<shareddatalayer::Error>(condition))
+ {
+ case shareddatalayer::Error::SUCCESS:
+ return ec == InternalError::SUCCESS ||
+ ec == std::error_code();
+ case shareddatalayer::Error::NOT_CONNECTED:
+ return ec == InternalError::SDL_NOT_CONNECTED_TO_BACKEND ||
+ ec == InternalError::BACKEND_NOT_READY ||
+ ec == InternalError::SDL_NOT_READY;
+ case shareddatalayer::Error::OPERATION_INTERRUPTED:
+ return ec == InternalError::BACKEND_CONNECTION_LOST;
+ case shareddatalayer::Error::BACKEND_FAILURE:
+ return ec == InternalError::BACKEND_ERROR ||
+ ec == InternalError::SDL_ERROR_CODE_LOGIC_ERROR;
+ case shareddatalayer::Error::REJECTED_BY_BACKEND:
+ return ec == InternalError::BACKEND_REJECTED_REQUEST;
+ case shareddatalayer::Error::REJECTED_BY_SDL:
+ return ec == InternalError::SDL_RECEIVED_INVALID_PARAMETER;
+ default:
+ /* Since clients can compare shareddatalayer::Error conditions against any std::error_code based ec, this error log
+ * can occur without fault in SDL implementation. If error log appears:
+ * - First check is the problematic ec set in SDL implementation
+ * - If yes, do needed updates to SDL (update error mappings for given ec or change SDL to set some error_code)
+ * - If no, ask client to check why they are comparing non-SDL originated error_code against shareddatalayer::Error
+ */
+ std::ostringstream msg;
+ msg << "SharedDataLayerErrorCategory::equivalent no mapping for error: " << ec.category().name()
+ << ec.value();
+ logErrorOnce(msg.str());
+ return false;
+ }
+ }
+}
+
+namespace shareddatalayer
+{
+ std::error_condition make_error_condition(shareddatalayer::Error errorCode)
+ {
+ return {static_cast<int>(errorCode), getSharedDataLayerErrorCategory()};
+ }
+}
diff --git a/src/eventfd.cpp b/src/eventfd.cpp
new file mode 100644
index 0000000..6a0b14a
--- /dev/null
+++ b/src/eventfd.cpp
@@ -0,0 +1,96 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/eventfd.hpp"
+#include <sys/eventfd.h>
+#include "private/abort.hpp"
+#include "private/engine.hpp"
+#include "private/system.hpp"
+
+using namespace shareddatalayer;
+
+namespace
+{
+ /*
+ * Simple wrapper for executing the given callback without expecting
+ * exceptions. If callback throws, we'll crash.
+ */
+ void execute(const EventFD::Callback& callback) noexcept
+ {
+ callback();
+ }
+}
+
+EventFD::EventFD(Engine& engine):
+ EventFD(System::getSystem(), engine)
+{
+}
+
+EventFD::EventFD(System& system, Engine& engine):
+ system(system),
+ fd(system, system.eventfd(0U, EFD_CLOEXEC | EFD_NONBLOCK))
+{
+ engine.addMonitoredFD(fd, Engine::EVENT_IN, std::bind(&EventFD::handleEvents, this));
+}
+
+EventFD::~EventFD()
+{
+}
+
+void EventFD::post(const Callback& callback)
+{
+ if (!callback)
+ SHAREDDATALAYER_ABORT("A null callback was provided");
+
+ atomicPushBack(callback);
+ static const uint64_t value(1U);
+ system.write(fd, &value, sizeof(value));
+}
+
+void EventFD::atomicPushBack(const Callback& callback)
+{
+ std::lock_guard<std::mutex> guard(callbacksMutex);
+ callbacks.push_back(callback);
+}
+
+EventFD::Callbacks EventFD::atomicPopAll()
+{
+ std::lock_guard<std::mutex> guard(callbacksMutex);
+ Callbacks extractedCallbacks;
+ std::swap(callbacks, extractedCallbacks);
+ return extractedCallbacks;
+}
+
+void EventFD::handleEvents()
+{
+ uint64_t value;
+ system.read(fd, &value, sizeof(value));
+ executeCallbacks();
+}
+
+void EventFD::executeCallbacks()
+{
+ Callbacks callbacks(atomicPopAll());
+ while (!callbacks.empty())
+ popAndExecuteFirstCallback(callbacks);
+}
+
+void EventFD::popAndExecuteFirstCallback(Callbacks& callbacks)
+{
+ const auto callback(callbacks.front());
+ callbacks.pop_front();
+ execute(callback);
+}
diff --git a/src/exception.cpp b/src/exception.cpp
new file mode 100644
index 0000000..0c8da72
--- /dev/null
+++ b/src/exception.cpp
@@ -0,0 +1,29 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include <sdl/exception.hpp>
+
+using namespace shareddatalayer;
+
+Exception::Exception(const std::string& message):
+ std::runtime_error(message)
+{
+}
+
+Exception::Exception(const char* message):
+ std::runtime_error(message)
+{
+}
diff --git a/src/filedescriptor.cpp b/src/filedescriptor.cpp
new file mode 100644
index 0000000..10dec70
--- /dev/null
+++ b/src/filedescriptor.cpp
@@ -0,0 +1,77 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/filedescriptor.hpp"
+#include "private/system.hpp"
+
+using namespace shareddatalayer;
+
+namespace
+{
+ int moveFD(int& fd)
+ {
+ const int ret(fd);
+ fd = -1;
+ return ret;
+ }
+}
+
+FileDescriptor::FileDescriptor(int fd) noexcept:
+ FileDescriptor(System::getSystem(), fd)
+{
+}
+
+FileDescriptor::FileDescriptor(System& system, int fd) noexcept:
+ system(&system),
+ fd(fd)
+{
+}
+
+FileDescriptor::FileDescriptor(FileDescriptor&& fd) noexcept:
+ system(fd.system),
+ fd(moveFD(fd.fd)),
+ atCloseCb(std::move(fd.atCloseCb))
+{
+}
+
+FileDescriptor& FileDescriptor::operator = (FileDescriptor&& fd) noexcept
+{
+ close();
+ system = fd.system;
+ this->fd = moveFD(fd.fd);
+ atCloseCb = std::move(fd.atCloseCb);
+ return *this;
+}
+
+FileDescriptor::~FileDescriptor()
+{
+ close();
+}
+
+void FileDescriptor::atClose(std::function<void(int)> atCloseCb)
+{
+ this->atCloseCb = atCloseCb;
+}
+
+void FileDescriptor::close()
+{
+ if (fd < 0)
+ return;
+ if (atCloseCb)
+ atCloseCb(fd);
+ system->close(fd);
+ fd = -1;
+}
diff --git a/src/hostandport.cpp b/src/hostandport.cpp
new file mode 100644
index 0000000..de045ed
--- /dev/null
+++ b/src/hostandport.cpp
@@ -0,0 +1,158 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/hostandport.hpp"
+#include <sstream>
+#include <netdb.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <boost/lexical_cast.hpp>
+
+using namespace shareddatalayer;
+
+namespace
+{
+ uint16_t stringToPort(const std::string& port)
+ {
+ if (port.empty())
+ throw HostAndPort::EmptyPort();
+ try
+ {
+ return htons(boost::lexical_cast<uint16_t>(port));
+ }
+ catch (const boost::bad_lexical_cast&)
+ {
+ char buf[1024];
+ servent resultbuf = { };
+ servent *result(nullptr);
+ getservbyname_r(port.c_str(), nullptr, &resultbuf, buf, sizeof(buf), &result);
+ if (result == nullptr)
+ throw HostAndPort::InvalidPort(port);
+ return static_cast<uint16_t>(result->s_port);
+ }
+ }
+
+ bool isInBrackets(const std::string& str)
+ {
+ return ((!str.empty()) && (str.front() == '[') && (str.back() == ']'));
+ }
+
+ std::string removeBrackets(const std::string& str)
+ {
+ return str.substr(1U, str.size() - 2U);
+ }
+
+ bool isLiteralIPv6(const std::string& str)
+ {
+ in6_addr tmp;
+ return inet_pton(AF_INET6, str.c_str(), &tmp) == 1;
+ }
+
+ std::string buildInvalidPortError(const std::string& port)
+ {
+ std::ostringstream os;
+ os << "invalid port: " << port;
+ return os.str();
+ }
+}
+
+HostAndPort::InvalidPort::InvalidPort(const std::string& port):
+ Exception(buildInvalidPortError(port))
+{
+}
+
+HostAndPort::EmptyPort::EmptyPort():
+ Exception("empty port")
+{
+}
+
+HostAndPort::EmptyHost::EmptyHost():
+ Exception("empty host")
+{
+}
+
+HostAndPort::HostAndPort(const std::string& addressAndOptionalPort, uint16_t defaultPort)
+{
+ if (isInBrackets(addressAndOptionalPort))
+ parse(removeBrackets(addressAndOptionalPort), defaultPort);
+ else
+ parse(addressAndOptionalPort, defaultPort);
+}
+
+void HostAndPort::parse(const std::string& addressAndOptionalPort, uint16_t defaultPort)
+{
+ const auto pos(addressAndOptionalPort.rfind(':'));
+ if (pos == std::string::npos)
+ {
+ host = addressAndOptionalPort;
+ port = defaultPort;
+ literalIPv6 = false;
+ }
+ else if (isLiteralIPv6(addressAndOptionalPort))
+ {
+ host = addressAndOptionalPort;
+ port = defaultPort;
+ literalIPv6 = true;
+ }
+ else
+ {
+ host = addressAndOptionalPort.substr(0, pos);
+ if (isInBrackets(host))
+ host = removeBrackets(host);
+ literalIPv6 = isLiteralIPv6(host);
+ port = stringToPort(addressAndOptionalPort.substr(pos + 1U, addressAndOptionalPort.size() - pos - 1U));
+ }
+ if (host.empty())
+ throw EmptyHost();
+}
+
+const std::string& HostAndPort::getHost() const
+{
+ return host;
+}
+
+uint16_t HostAndPort::getPort() const
+{
+ return port;
+}
+
+std::string HostAndPort::getString() const
+{
+ std::ostringstream os;
+ if (literalIPv6)
+ os << '[' << host << "]:" << ntohs(port);
+ else
+ os << host << ':' << ntohs(port);
+ return os.str();
+}
+
+bool HostAndPort::operator==(const HostAndPort& hp) const
+{
+ return this->getPort() == hp.getPort() && this->getHost() == hp.getHost();
+}
+
+bool HostAndPort::operator!=(const HostAndPort& hp) const
+{
+ return !(hp == *this);
+}
+
+bool HostAndPort::operator<(const HostAndPort& hp) const
+{
+ if (this->getHost() == hp.getHost())
+ return this->getPort() < hp.getPort();
+ else
+ return this->getHost() < hp.getHost();
+}
diff --git a/src/invalidnamespace.cpp b/src/invalidnamespace.cpp
new file mode 100644
index 0000000..988d2ee
--- /dev/null
+++ b/src/invalidnamespace.cpp
@@ -0,0 +1,36 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include <sdl/invalidnamespace.hpp>
+#include <sstream>
+#include "private/namespacevalidator.hpp"
+
+using namespace shareddatalayer;
+
+namespace
+{
+ std::string buildErrorMessage(const std::string& ns)
+ {
+ std::ostringstream oss;
+ oss << "namespace string: \"" << ns << "\" is invalid. Disallowed characters: " << getDisallowedCharactersInNamespace();
+ return oss.str();
+ }
+}
+
+InvalidNamespace::InvalidNamespace(const std::string& ns):
+ Exception(buildErrorMessage(ns))
+{
+}
diff --git a/src/namespaceconfigurationsimpl.cpp b/src/namespaceconfigurationsimpl.cpp
new file mode 100644
index 0000000..7a4692c
--- /dev/null
+++ b/src/namespaceconfigurationsimpl.cpp
@@ -0,0 +1,118 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/abort.hpp"
+#include "private/namespaceconfigurationsimpl.hpp"
+#include <sstream>
+#include <iomanip>
+
+using namespace shareddatalayer;
+
+namespace
+{
+ const NamespaceConfiguration getDefaultNamespaceConfiguration()
+ {
+ return {"", true, false, "<default>"};
+ }
+}
+
+NamespaceConfigurationsImpl::NamespaceConfigurationsImpl():
+ namespaceConfigurations({getDefaultNamespaceConfiguration()}),
+ //Current implementation assumes that this index is zero. If changing the index, do needed modifications to implementation.
+ defaultConfigurationIndex(0),
+ namespaceConfigurationsLookupTable()
+{
+}
+
+NamespaceConfigurationsImpl::~NamespaceConfigurationsImpl()
+{
+}
+
+void NamespaceConfigurationsImpl::addNamespaceConfiguration(const NamespaceConfiguration& namespaceConfiguration)
+{
+ if (!namespaceConfigurationsLookupTable.empty())
+ SHAREDDATALAYER_ABORT("Cannot add namespace configurations after lookup table is initialized");
+
+ namespaceConfigurations.push_back(namespaceConfiguration);
+}
+
+std::string NamespaceConfigurationsImpl::getDescription(const std::string& ns) const
+{
+ NamespaceConfiguration namespaceConfiguration = findConfigurationForNamespace(ns);
+ std::string sourceInfo = namespaceConfiguration.sourceName;
+
+ if (!namespaceConfiguration.namespacePrefix.empty())
+ sourceInfo += (" prefix: " + namespaceConfiguration.namespacePrefix);
+
+ std::ostringstream os;
+ os << std::boolalpha;
+ os << sourceInfo << ", ";
+ os << "useDbBackend: " << namespaceConfiguration.dbBackendIsUsed << ", ";
+ os << "enableNotifications: " << namespaceConfiguration.notificationsAreEnabled;
+ return os.str();
+}
+
+bool NamespaceConfigurationsImpl::isDbBackendUseEnabled(const std::string& ns) const
+{
+ return findConfigurationForNamespace(ns).dbBackendIsUsed;
+}
+
+bool NamespaceConfigurationsImpl::areNotificationsEnabled(const std::string& ns) const
+{
+ return findConfigurationForNamespace(ns).notificationsAreEnabled;
+}
+
+const NamespaceConfiguration& NamespaceConfigurationsImpl::findConfigurationForNamespace(const std::string& ns) const
+{
+ if (namespaceConfigurationsLookupTable.count(ns) > 0)
+ return namespaceConfigurations.at(namespaceConfigurationsLookupTable.at(ns));
+
+ size_t longestMatchingPrefixLen(0);
+ std::vector<int>::size_type foundIndex(0);
+ bool configurationFound(false);
+
+ for(std::vector<int>::size_type i = (defaultConfigurationIndex + 1); i < namespaceConfigurations.size(); i++)
+ {
+ const auto prefixLen(namespaceConfigurations[i].namespacePrefix.length());
+
+ if (ns.compare(0, prefixLen, namespaceConfigurations[i].namespacePrefix) == 0)
+ {
+ if (prefixLen >= longestMatchingPrefixLen)
+ {
+ configurationFound = true;
+ foundIndex = i;
+ longestMatchingPrefixLen = prefixLen;
+ }
+ }
+ }
+
+ if (!configurationFound)
+ foundIndex = defaultConfigurationIndex;
+
+ namespaceConfigurationsLookupTable[ns] = foundIndex;
+ return namespaceConfigurations[foundIndex];
+}
+
+bool NamespaceConfigurationsImpl::isEmpty() const
+{
+ /* Empty when contains only default configuration */
+ return namespaceConfigurations.size() == 1;
+}
+
+bool NamespaceConfigurationsImpl::isNamespaceInLookupTable(const std::string& ns) const
+{
+ return namespaceConfigurationsLookupTable.count(ns) > 0;
+}
diff --git a/src/namespacevalidator.cpp b/src/namespacevalidator.cpp
new file mode 100644
index 0000000..7e14d00
--- /dev/null
+++ b/src/namespacevalidator.cpp
@@ -0,0 +1,48 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/namespacevalidator.hpp"
+#include <sdl/emptynamespace.hpp>
+#include <sdl/invalidnamespace.hpp>
+#include "private/asyncconnection.hpp"
+
+
+namespace shareddatalayer
+{
+ const std::string& getDisallowedCharactersInNamespace()
+ {
+ static const std::string disallowedCharacters{AsyncConnection::SEPARATOR, '{', '}'};
+ return disallowedCharacters;
+ }
+
+ bool isValidNamespaceSyntax(const Namespace& ns)
+ {
+ return (ns.find_first_of(getDisallowedCharactersInNamespace()) == std::string::npos);
+ }
+
+ void validateNamespace(const Namespace& ns)
+ {
+ if (ns.empty())
+ throw EmptyNamespace();
+ else if (!isValidNamespaceSyntax(ns))
+ throw InvalidNamespace(ns);
+ }
+
+ bool isValidNamespace(const Namespace& ns)
+ {
+ return (!ns.empty() && isValidNamespaceSyntax(ns));
+ }
+}
diff --git a/src/notconnected.cpp b/src/notconnected.cpp
new file mode 100644
index 0000000..a15ae7c
--- /dev/null
+++ b/src/notconnected.cpp
@@ -0,0 +1,24 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include <sdl/notconnected.hpp>
+
+using namespace shareddatalayer;
+
+NotConnected::NotConnected(const std::string& error):
+ Exception(error)
+{
+}
diff --git a/src/operationinterrupted.cpp b/src/operationinterrupted.cpp
new file mode 100644
index 0000000..46141a3
--- /dev/null
+++ b/src/operationinterrupted.cpp
@@ -0,0 +1,24 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include <sdl/operationinterrupted.hpp>
+
+using namespace shareddatalayer;
+
+OperationInterrupted::OperationInterrupted(const std::string& error):
+ Exception(error)
+{
+}
diff --git a/src/publisherid.cpp b/src/publisherid.cpp
new file mode 100644
index 0000000..c8fd52a
--- /dev/null
+++ b/src/publisherid.cpp
@@ -0,0 +1,19 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include <sdl/publisherid.hpp>
+
+const shareddatalayer::PublisherId shareddatalayer::NO_PUBLISHER("shareddatalayer::NO_PUBLISHER");
diff --git a/src/redis/asynccommanddispatcher.cpp b/src/redis/asynccommanddispatcher.cpp
new file mode 100644
index 0000000..25ec890
--- /dev/null
+++ b/src/redis/asynccommanddispatcher.cpp
@@ -0,0 +1,67 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/redis/asynccommanddispatcher.hpp"
+#include <cstdlib>
+#include "config.h"
+#include "private/redis/databaseinfo.hpp"
+#if HAVE_HIREDIS_VIP
+#include "private/redis/asynchiredisclustercommanddispatcher.hpp"
+#include "private/redis/asynchirediscommanddispatcher.hpp"
+#elif HAVE_HIREDIS
+#include "private/redis/asynchirediscommanddispatcher.hpp"
+#endif
+#include "private/abort.hpp"
+#include "private/engine.hpp"
+
+using namespace shareddatalayer::redis;
+
+std::shared_ptr<AsyncCommandDispatcher> AsyncCommandDispatcher::create(Engine& engine,
+ const DatabaseInfo& databaseInfo,
+ std::shared_ptr<ContentsBuilder> contentsBuilder,
+ bool usePermanentCommandCallbacks,
+ std::shared_ptr<Logger> logger)
+{
+#if HAVE_HIREDIS_VIP
+ if (databaseInfo.type == DatabaseInfo::Type::CLUSTER)
+ {
+ return std::make_shared<AsyncHiredisClusterCommandDispatcher>(engine,
+ databaseInfo.ns,
+ databaseInfo.hosts,
+ contentsBuilder,
+ usePermanentCommandCallbacks,
+ logger);
+ }
+ else
+ return std::make_shared<AsyncHiredisCommandDispatcher>(engine,
+ databaseInfo.hosts.at(0).getHost(),
+ databaseInfo.hosts.at(0).getPort(),
+ contentsBuilder,
+ usePermanentCommandCallbacks,
+ logger);
+#elif HAVE_HIREDIS
+ if (databaseInfo.type == DatabaseInfo::Type::CLUSTER)
+ SHAREDDATALAYER_ABORT("Not implemented.");
+ return std::make_shared<AsyncHiredisCommandDispatcher>(engine,
+ databaseInfo.hosts.at(0).getHost(),
+ databaseInfo.hosts.at(0).getPort(),
+ contentsBuilder,
+ usePermanentCommandCallbacks,
+ logger);
+#else
+ SHAREDDATALAYER_ABORT("Not implemented.");
+#endif
+}
diff --git a/src/redis/asyncdatabasediscovery.cpp b/src/redis/asyncdatabasediscovery.cpp
new file mode 100644
index 0000000..892e7cc
--- /dev/null
+++ b/src/redis/asyncdatabasediscovery.cpp
@@ -0,0 +1,63 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/redis/asyncdatabasediscovery.hpp"
+#include "private/databaseconfiguration.hpp"
+#include "private/logger.hpp"
+#include <cstdlib>
+#include "config.h"
+#if HAVE_HIREDIS
+#include "private/redis/asynchiredisdatabasediscovery.hpp"
+#endif
+#include "private/abort.hpp"
+
+using namespace shareddatalayer::redis;
+
+std::shared_ptr<AsyncDatabaseDiscovery> AsyncDatabaseDiscovery::create(std::shared_ptr<Engine> engine,
+ const boost::optional<Namespace>& ns,
+ const DatabaseConfiguration& staticDatabaseConfiguration,
+ std::shared_ptr<Logger> logger)
+{
+ auto staticAddresses(staticDatabaseConfiguration.getServerAddresses());
+
+ if (staticAddresses.empty())
+ staticAddresses = staticDatabaseConfiguration.getDefaultServerAddresses();
+
+ auto staticDbType(staticDatabaseConfiguration.getDbType());
+
+ if (staticDbType == DatabaseConfiguration::DbType::REDIS_CLUSTER)
+#if HAVE_HIREDIS_VIP
+ return std::make_shared<AsyncHiredisDatabaseDiscovery>(engine,
+ ns,
+ DatabaseInfo::Type::CLUSTER,
+ staticAddresses,
+ logger);
+#else
+ SHAREDDATALAYER_ABORT("No Hiredis vip for Redis cluster configuration");
+#endif
+ else
+#if HAVE_HIREDIS
+ return std::make_shared<AsyncHiredisDatabaseDiscovery>(engine,
+ ns,
+ DatabaseInfo::Type::SINGLE,
+ staticAddresses,
+ logger);
+#else
+ static_cast<void>(logger);
+ SHAREDDATALAYER_ABORT("No Hiredis");
+#endif
+}
+
diff --git a/src/redis/asynchiredisclustercommanddispatcher.cpp b/src/redis/asynchiredisclustercommanddispatcher.cpp
new file mode 100644
index 0000000..7270886
--- /dev/null
+++ b/src/redis/asynchiredisclustercommanddispatcher.cpp
@@ -0,0 +1,334 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/redis/asynchiredisclustercommanddispatcher.hpp"
+#include <algorithm>
+#include <cstring>
+#include <cerrno>
+#include <sstream>
+#include "private/abort.hpp"
+#include "private/createlogger.hpp"
+#include "private/error.hpp"
+#include "private/logger.hpp"
+#include "private/redis/asyncredisreply.hpp"
+#include "private/redis/reply.hpp"
+#include "private/redis/hiredisclustersystem.hpp"
+#include "private/engine.hpp"
+#include "private/redis/hiredisclusterepolladapter.hpp"
+#include "private/redis/contents.hpp"
+#include "private/redis/redisgeneral.hpp"
+
+using namespace shareddatalayer;
+using namespace shareddatalayer::redis;
+
+namespace
+{
+ void connectCb(const redisClusterAsyncContext*, const redisAsyncContext* ac, int status)
+ {
+ if (!status)
+ {
+ std::ostringstream msg;
+ msg << "redis cluster instance connected, fd: " << ac->c.fd;
+ logDebugOnce(msg.str());
+ }
+ }
+
+ void disconnectCb(const redisClusterAsyncContext* acc, const redisAsyncContext* ac, int status)
+ {
+ if (status)
+ {
+ std::ostringstream msg;
+ msg << "redis cluster instance disconnected, fd: " << ac->c.fd
+ << ", status: " << ac->err;
+ logDebugOnce(msg.str());
+ }
+ auto instance(static_cast<AsyncHiredisClusterCommandDispatcher*>(acc->data));
+ instance->handleDisconnect(ac);
+ }
+
+ void cb(redisClusterAsyncContext* acc, void* rr, void* pd)
+ {
+ auto instance(static_cast<AsyncHiredisClusterCommandDispatcher*>(acc->data));
+ auto reply(static_cast<redisReply*>(rr));
+ auto cb(static_cast<AsyncHiredisClusterCommandDispatcher::CommandCb*>(pd));
+ if (instance->isClientCallbacksEnabled())
+ instance->handleReply(*cb, getRedisError(acc->err, acc->errstr, reply), reply);
+ }
+}
+
+AsyncHiredisClusterCommandDispatcher::AsyncHiredisClusterCommandDispatcher(Engine& engine,
+ const boost::optional<std::string>& ns,
+ const DatabaseConfiguration::Addresses& addresses,
+ std::shared_ptr<ContentsBuilder> contentsBuilder,
+ bool usePermanentCommandCallbacks,
+ std::shared_ptr<Logger> logger):
+ AsyncHiredisClusterCommandDispatcher(engine,
+ ns,
+ addresses,
+ contentsBuilder,
+ usePermanentCommandCallbacks,
+ HiredisClusterSystem::getInstance(),
+ std::make_shared<HiredisClusterEpollAdapter>(engine),
+ logger)
+{
+}
+
+AsyncHiredisClusterCommandDispatcher::AsyncHiredisClusterCommandDispatcher(Engine& engine,
+ const boost::optional<std::string>& ns,
+ const DatabaseConfiguration::Addresses& addresses,
+ std::shared_ptr<ContentsBuilder> contentsBuilder,
+ bool usePermanentCommandCallbacks,
+ HiredisClusterSystem& hiredisClusterSystem,
+ std::shared_ptr<HiredisClusterEpollAdapter> adapter,
+ std::shared_ptr<Logger> logger):
+ engine(engine),
+ initialNamespace(ns),
+ addresses(addresses),
+ contentsBuilder(contentsBuilder),
+ usePermanentCommandCallbacks(usePermanentCommandCallbacks),
+ hiredisClusterSystem(hiredisClusterSystem),
+ adapter(adapter),
+ acc(nullptr),
+ serviceState(ServiceState::DISCONNECTED),
+ clientCallbacksEnabled(true),
+ connectionRetryTimer(engine),
+ connectionRetryTimerDuration(std::chrono::seconds(1)),
+ logger(logger)
+{
+ connect();
+}
+
+AsyncHiredisClusterCommandDispatcher::~AsyncHiredisClusterCommandDispatcher()
+{
+ disconnectHiredisCluster();
+}
+
+void AsyncHiredisClusterCommandDispatcher::connect()
+{
+ // Disconnect and free possibly (in re-connecting scenarios) already existing Redis cluster context.
+ disconnectHiredisCluster();
+ acc = hiredisClusterSystem.redisClusterAsyncConnect(formatToClusterSyntax(addresses).c_str(),
+ HIRCLUSTER_FLAG_ROUTE_USE_SLOTS);
+ if (acc == nullptr)
+ {
+ logger->error() << "SDL: connecting to redis cluster failed, null context returned";
+ armConnectionRetryTimer();
+ return;
+ }
+ if (acc->err)
+ {
+ logger->error() << "SDL: connecting to redis cluster failed, error: " << acc->err;
+ armConnectionRetryTimer();
+ return;
+ }
+ acc->data = this;
+ adapter->setup(acc);
+ hiredisClusterSystem.redisClusterAsyncSetConnectCallback(acc, connectCb);
+ hiredisClusterSystem.redisClusterAsyncSetDisconnectCallback(acc, disconnectCb);
+ verifyConnection();
+}
+
+void AsyncHiredisClusterCommandDispatcher::verifyConnection()
+{
+ /* redisClusterAsyncConnect only queries available cluster nodes but it does
+ * not connect to any cluster node (as it does not know to which node it should connect to).
+ * As we use same cluster node for one SDL namespace (namespace is a hash tag), it is already
+ * determined to which cluster node this instance will connect to. We do initial operation
+ * to get connection to right redis node established already now. This also verifies that
+ * connection really works. When Redis has max amount of users, it will still accept new
+ * connections but is will close them immediately. Therefore, we need to verify that just
+ * established connection really works.
+ */
+ /* Connection setup/verification is now done by doing redis command list query. Because we anyway
+ * need to verify that Redis has required commands, we can now combine these two operations
+ * (command list query and connection setup/verification). If either one of the functionalities
+ * is not needed in the future and it is removed, remember to still leave the other one.
+ */
+ /* Non namespace-specific command list query can be used for connection setup purposes,
+ * because SDL uses redisClusterAsyncCommandArgvWithKey which adds namespace to all
+ * commands dispacthed.
+ */
+
+ /* If initial namespace was not given during dispatcher creation (multi namespace API),
+ * verification is sent to hardcoded namespace. This works for verification purposes
+ * because in our environment cluster is configured to operate only if all nodes
+ * are working (so we can do verification to any node). However, this is not optimal
+ * because we do not necessarily connect to such cluster node which will later be
+ * used by client. Also our cluster configuration can change. This needs to be
+ * optimized later (perhaps to connect to all nodes). */
+ std::string nsForVerification;
+ if (initialNamespace)
+ nsForVerification = *initialNamespace;
+ else
+ nsForVerification = "namespace";
+
+ dispatchAsync(std::bind(&AsyncHiredisClusterCommandDispatcher::verifyConnectionReply,
+ this,
+ std::placeholders::_1,
+ std::placeholders::_2),
+ nsForVerification,
+ contentsBuilder->build("COMMAND"),
+ false);
+}
+
+void AsyncHiredisClusterCommandDispatcher::verifyConnectionReply(const std::error_code& error, const redis::Reply& reply)
+{
+ if(error)
+ {
+ logger->error() << "AsyncHiredisClusterCommandDispatcher: connection verification failed: "
+ << error.message();
+ armConnectionRetryTimer();
+ }
+ else
+ {
+ if (checkRedisModuleCommands(parseCommandListReply(reply)))
+ setConnected();
+ else
+ SHAREDDATALAYER_ABORT("Required Redis module extension commands not available.");
+ }
+}
+
+void AsyncHiredisClusterCommandDispatcher::waitConnectedAsync(const ConnectAck& connectAck)
+{
+ this->connectAck = connectAck;
+ if (serviceState == ServiceState::CONNECTED)
+ engine.postCallback(connectAck);
+}
+
+void AsyncHiredisClusterCommandDispatcher::registerDisconnectCb(const DisconnectCb& disconnectCb)
+{
+ disconnectCallback = disconnectCb;
+}
+
+void AsyncHiredisClusterCommandDispatcher::dispatchAsync(const CommandCb& commandCb,
+ const AsyncConnection::Namespace& ns,
+ const Contents& contents)
+{
+ dispatchAsync(commandCb, ns, contents, true);
+}
+
+void AsyncHiredisClusterCommandDispatcher::dispatchAsync(const CommandCb& commandCb,
+ const AsyncConnection::Namespace& ns,
+ const Contents& contents,
+ bool checkConnectionState)
+{
+ if (checkConnectionState && serviceState != ServiceState::CONNECTED)
+ {
+ engine.postCallback(std::bind(&AsyncHiredisClusterCommandDispatcher::callCommandCbWithError,
+ this,
+ commandCb,
+ std::error_code(AsyncRedisCommandDispatcherErrorCode::NOT_CONNECTED)));
+ return;
+ }
+ cbs.push_back(commandCb);
+ std::vector<const char*> chars;
+ std::transform(contents.stack.begin(), contents.stack.end(),
+ std::back_inserter(chars), [](const std::string& str){ return str.c_str(); });
+ if (hiredisClusterSystem.redisClusterAsyncCommandArgvWithKey(acc, cb, &cbs.back(), ns.c_str(), static_cast<int>(ns.size()),
+ static_cast<int>(contents.stack.size()), &chars[0],
+ &contents.sizes[0]) != REDIS_OK)
+ {
+ removeCb(cbs.back());
+ engine.postCallback(std::bind(&AsyncHiredisClusterCommandDispatcher::callCommandCbWithError,
+ this,
+ commandCb,
+ getRedisError(acc->err, acc->errstr, nullptr)));
+ }
+}
+
+void AsyncHiredisClusterCommandDispatcher::disableCommandCallbacks()
+{
+ clientCallbacksEnabled = false;
+}
+
+void AsyncHiredisClusterCommandDispatcher::callCommandCbWithError(const CommandCb& commandCb, const std::error_code& error)
+{
+ commandCb(error, AsyncRedisReply());
+}
+
+void AsyncHiredisClusterCommandDispatcher::setConnected()
+{
+ serviceState = ServiceState::CONNECTED;
+
+ if (connectAck)
+ {
+ connectAck();
+ connectAck = ConnectAck();
+ }
+}
+
+void AsyncHiredisClusterCommandDispatcher::armConnectionRetryTimer()
+{
+ connectionRetryTimer.arm(connectionRetryTimerDuration,
+ [this] () { connect(); });
+
+}
+
+void AsyncHiredisClusterCommandDispatcher::handleReply(const CommandCb& commandCb,
+ const std::error_code& error,
+ const redisReply* rr)
+{
+ if (!isValidCb(commandCb))
+ SHAREDDATALAYER_ABORT("Invalid callback function.");
+ if (error)
+ commandCb(error, AsyncRedisReply());
+ else
+ commandCb(error, AsyncRedisReply(*rr));
+ if (!usePermanentCommandCallbacks)
+ removeCb(commandCb);
+}
+
+bool AsyncHiredisClusterCommandDispatcher::isClientCallbacksEnabled() const
+{
+ return clientCallbacksEnabled;
+}
+
+bool AsyncHiredisClusterCommandDispatcher::isValidCb(const CommandCb& commandCb)
+{
+ for (auto i(cbs.begin()); i != cbs.end(); ++i)
+ if (&*i == &commandCb)
+ return true;
+ return false;
+}
+
+void AsyncHiredisClusterCommandDispatcher::removeCb(const CommandCb& commandCb)
+{
+ for (auto i(cbs.begin()); i != cbs.end(); ++i)
+ if (&*i == &commandCb)
+ {
+ cbs.erase(i);
+ break;
+ }
+}
+
+void AsyncHiredisClusterCommandDispatcher::handleDisconnect(const redisAsyncContext* ac)
+{
+ adapter->detach(ac);
+
+ if (disconnectCallback)
+ disconnectCallback();
+}
+
+void AsyncHiredisClusterCommandDispatcher::disconnectHiredisCluster()
+{
+ /* hiredis sometimes crashes if redisClusterAsyncFree is called without being connected (even
+ * if acc is a valid pointer).
+ */
+ if (serviceState == ServiceState::CONNECTED)
+ hiredisClusterSystem.redisClusterAsyncFree(acc);
+
+ serviceState = ServiceState::DISCONNECTED;
+}
diff --git a/src/redis/asynchirediscommanddispatcher.cpp b/src/redis/asynchirediscommanddispatcher.cpp
new file mode 100644
index 0000000..e4195a9
--- /dev/null
+++ b/src/redis/asynchirediscommanddispatcher.cpp
@@ -0,0 +1,328 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/redis/asynchirediscommanddispatcher.hpp"
+#include <algorithm>
+#include <cstring>
+#include <cerrno>
+#include <sstream>
+#include <arpa/inet.h>
+#include "private/abort.hpp"
+#include "private/createlogger.hpp"
+#include "private/engine.hpp"
+#include "private/error.hpp"
+#include "private/logger.hpp"
+#include "private/redis/asyncredisreply.hpp"
+#include "private/redis/reply.hpp"
+#include "private/redis/hiredissystem.hpp"
+#include "private/redis/hiredisepolladapter.hpp"
+#include "private/redis/contents.hpp"
+#include "private/redis/redisgeneral.hpp"
+
+using namespace shareddatalayer;
+using namespace shareddatalayer::redis;
+
+namespace
+{
+ void connectCb(const redisAsyncContext* ac, int status)
+ {
+ bool isConnected = !status;
+ auto instance(static_cast<AsyncHiredisCommandDispatcher*>(ac->data));
+
+ if (isConnected)
+ {
+ std::ostringstream msg;
+ msg << "redis connected, fd: " << ac->c.fd;
+ logInfoOnce(msg.str());
+ instance->verifyConnection();
+ }
+ else
+ instance->setDisconnected();
+
+ }
+
+ void disconnectCb(const redisAsyncContext* ac, int status)
+ {
+ if (status) {
+ std::ostringstream msg;
+ msg << "redis disconnected, status: " << ac->err << ", " << ac->errstr << ", fd: " << ac->c.fd;
+ logErrorOnce(msg.str());
+ }
+ auto instance(static_cast<AsyncHiredisCommandDispatcher*>(ac->data));
+ instance->setDisconnected();
+ }
+
+ void cb(redisAsyncContext* ac, void* rr, void* pd)
+ {
+ auto instance(static_cast<AsyncHiredisCommandDispatcher*>(ac->data));
+ auto reply(static_cast<redisReply*>(rr));
+ auto cb(static_cast<AsyncHiredisCommandDispatcher::CommandCb*>(pd));
+ if (instance->isClientCallbacksEnabled())
+ instance->handleReply(*cb, getRedisError(ac->err, ac->errstr, reply), reply);
+ }
+}
+
+AsyncHiredisCommandDispatcher::AsyncHiredisCommandDispatcher(Engine& engine,
+ const std::string& address,
+ uint16_t port,
+ std::shared_ptr<ContentsBuilder> contentsBuilder,
+ bool usePermanentCommandCallbacks,
+ std::shared_ptr<Logger> logger):
+ AsyncHiredisCommandDispatcher(engine,
+ address,
+ port,
+ contentsBuilder,
+ usePermanentCommandCallbacks,
+ HiredisSystem::getHiredisSystem(),
+ std::make_shared<HiredisEpollAdapter>(engine),
+ logger)
+{
+}
+
+AsyncHiredisCommandDispatcher::AsyncHiredisCommandDispatcher(Engine& engine,
+ const std::string& address,
+ uint16_t port,
+ std::shared_ptr<ContentsBuilder> contentsBuilder,
+ bool usePermanentCommandCallbacks,
+ HiredisSystem& hiredisSystem,
+ std::shared_ptr<HiredisEpollAdapter> adapter,
+ std::shared_ptr<Logger> logger):
+ engine(engine),
+ address(address),
+ port(ntohs(port)),
+ contentsBuilder(contentsBuilder),
+ usePermanentCommandCallbacks(usePermanentCommandCallbacks),
+ hiredisSystem(hiredisSystem),
+ adapter(adapter),
+ ac(nullptr),
+ serviceState(ServiceState::DISCONNECTED),
+ clientCallbacksEnabled(true),
+ connectionRetryTimer(engine),
+ connectionRetryTimerDuration(std::chrono::seconds(1)),
+ connectionVerificationRetryTimerDuration(std::chrono::seconds(10)),
+ logger(logger)
+
+{
+ connect();
+}
+
+AsyncHiredisCommandDispatcher::~AsyncHiredisCommandDispatcher()
+{
+ disconnectHiredis();
+}
+
+void AsyncHiredisCommandDispatcher::connect()
+{
+ ac = hiredisSystem.redisAsyncConnect(address.c_str(), port);
+ if (ac == nullptr || ac->err)
+ {
+ setDisconnected();
+ return;
+ }
+ ac->data = this;
+ adapter->attach(ac);
+ hiredisSystem.redisAsyncSetConnectCallback(ac, connectCb);
+ hiredisSystem.redisAsyncSetDisconnectCallback(ac, disconnectCb);
+}
+
+void AsyncHiredisCommandDispatcher::verifyConnection()
+{
+ /* When Redis has max amount of users, it will still accept new connections but will
+ * close them immediately. Therefore, we need to verify that just established connection
+ * really works. This prevents calling client readyAck callback for a connection that
+ * will be terminated immediately.
+ */
+ /* Connection verification is now done by doing redis command list query. Because we anyway
+ * need to verify that Redis has required commands, we can now combine these two operations
+ * (command list query and connection verification). If either one of the functionalities
+ * is not needed in the future and it is removed, remember to still leave the other one.
+ */
+ serviceState = ServiceState::CONNECTION_VERIFICATION;
+ /* Disarm retry timer as now we are connected to hiredis. This ensures timer disarm if
+ * we are spontaneously connected to redis while timer is running. If connection verification
+ * fails, timer is armed again (normal handling in connection verification).
+ */
+ connectionRetryTimer.disarm();
+ dispatchAsync(std::bind(&AsyncHiredisCommandDispatcher::verifyConnectionReply,
+ this,
+ std::placeholders::_1,
+ std::placeholders::_2),
+ contentsBuilder->build("COMMAND"),
+ false);
+}
+
+void AsyncHiredisCommandDispatcher::verifyConnectionReply(const std::error_code& error,
+ const redis::Reply& reply)
+{
+ if(error)
+ {
+ logger->error() << "AsyncHiredisCommandDispatcher: connection verification failed: "
+ << error.message();
+
+ if (!connectionRetryTimer.isArmed())
+ {
+ /* Typically if connection verification fails, hiredis will call disconnect callback and
+ * whole connection establishment procedure will be restarted via that. To ensure that
+ * we will retry verification even if connection would not be disconnected this timer
+ * is set. If connection is later disconnected, this timer is disarmed (when disconnect
+ * callback handling arms this timer again).
+ */
+ armConnectionRetryTimer(connectionVerificationRetryTimerDuration,
+ std::bind(&AsyncHiredisCommandDispatcher::verifyConnection, this));
+ }
+ }
+ else
+ {
+ if (checkRedisModuleCommands(parseCommandListReply(reply)))
+ setConnected();
+ else
+ SHAREDDATALAYER_ABORT("Required Redis module extension commands not available.");
+ }
+}
+
+void AsyncHiredisCommandDispatcher::waitConnectedAsync(const ConnectAck& connectAck)
+{
+ this->connectAck = connectAck;
+ if (serviceState == ServiceState::CONNECTED)
+ engine.postCallback(connectAck);
+}
+
+void AsyncHiredisCommandDispatcher::registerDisconnectCb(const DisconnectCb& disconnectCb)
+{
+ disconnectCallback = disconnectCb;
+}
+
+void AsyncHiredisCommandDispatcher::dispatchAsync(const CommandCb& commandCb,
+ const AsyncConnection::Namespace&,
+ const Contents& contents)
+{
+ dispatchAsync(commandCb, contents, true);
+}
+
+void AsyncHiredisCommandDispatcher::dispatchAsync(const CommandCb& commandCb,
+ const Contents& contents,
+ bool checkConnectionState)
+{
+ if (checkConnectionState && serviceState != ServiceState::CONNECTED)
+ {
+ engine.postCallback(std::bind(&AsyncHiredisCommandDispatcher::callCommandCbWithError,
+ this,
+ commandCb,
+ std::error_code(AsyncRedisCommandDispatcherErrorCode::NOT_CONNECTED)));
+ return;
+ }
+ cbs.push_back(commandCb);
+ std::vector<const char*> chars;
+ std::transform(contents.stack.begin(), contents.stack.end(),
+ std::back_inserter(chars), [](const std::string& str){ return str.c_str(); });
+ if (hiredisSystem.redisAsyncCommandArgv(ac, cb, &cbs.back(), static_cast<int>(contents.stack.size()),
+ &chars[0], &contents.sizes[0]) != REDIS_OK)
+ {
+ removeCb(cbs.back());
+ engine.postCallback(std::bind(&AsyncHiredisCommandDispatcher::callCommandCbWithError,
+ this,
+ commandCb,
+ getRedisError(ac->err, ac->errstr, nullptr)));
+ }
+}
+
+void AsyncHiredisCommandDispatcher::disableCommandCallbacks()
+{
+ clientCallbacksEnabled = false;
+}
+
+void AsyncHiredisCommandDispatcher::callCommandCbWithError(const CommandCb& commandCb,
+ const std::error_code& error)
+{
+ commandCb(error, AsyncRedisReply());
+}
+
+void AsyncHiredisCommandDispatcher::setConnected()
+{
+ serviceState = ServiceState::CONNECTED;
+
+ if (connectAck)
+ {
+ connectAck();
+ connectAck = ConnectAck();
+ }
+}
+
+void AsyncHiredisCommandDispatcher::setDisconnected()
+{
+ serviceState = ServiceState::DISCONNECTED;
+
+ if (disconnectCallback)
+ disconnectCallback();
+
+ armConnectionRetryTimer(connectionRetryTimerDuration,
+ std::bind(&AsyncHiredisCommandDispatcher::connect, this));
+}
+
+void AsyncHiredisCommandDispatcher::handleReply(const CommandCb& commandCb,
+ const std::error_code& error,
+ const redisReply* rr)
+{
+ if (!isValidCb(commandCb))
+ SHAREDDATALAYER_ABORT("Invalid callback function.");
+ if (error)
+ commandCb(error, AsyncRedisReply());
+ else
+ commandCb(error, AsyncRedisReply(*rr));
+ if (!usePermanentCommandCallbacks)
+ removeCb(commandCb);
+}
+
+bool AsyncHiredisCommandDispatcher::isClientCallbacksEnabled() const
+{
+ return clientCallbacksEnabled;
+}
+
+bool AsyncHiredisCommandDispatcher::isValidCb(const CommandCb& commandCb)
+{
+ for (auto i(cbs.begin()); i != cbs.end(); ++i)
+ if (&*i == &commandCb)
+ return true;
+ return false;
+}
+
+void AsyncHiredisCommandDispatcher::removeCb(const CommandCb& commandCb)
+{
+ for (auto i(cbs.begin()); i != cbs.end(); ++i)
+ if (&*i == &commandCb)
+ {
+ cbs.erase(i);
+ break;
+ }
+}
+
+void AsyncHiredisCommandDispatcher::disconnectHiredis()
+{
+ /* hiredis sometimes crashes if redisAsyncFree is called without being connected (even
+ * if ac is a valid pointer).
+ */
+ if (serviceState == ServiceState::CONNECTED || serviceState == ServiceState::CONNECTION_VERIFICATION)
+ hiredisSystem.redisAsyncFree(ac);
+
+ //disconnect callback handler will update serviceState
+}
+
+void AsyncHiredisCommandDispatcher::armConnectionRetryTimer(Timer::Duration duration,
+ std::function<void()> retryAction)
+{
+ connectionRetryTimer.arm(duration,
+ [retryAction] () { retryAction(); });
+}
diff --git a/src/redis/asynchiredisdatabasediscovery.cpp b/src/redis/asynchiredisdatabasediscovery.cpp
new file mode 100644
index 0000000..7060995
--- /dev/null
+++ b/src/redis/asynchiredisdatabasediscovery.cpp
@@ -0,0 +1,52 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/asyncconnection.hpp"
+#include "private/redis/asynchiredisdatabasediscovery.hpp"
+#include "private/engine.hpp"
+#include "private/logger.hpp"
+#include "private/redis/asynccommanddispatcher.hpp"
+#include "private/redis/contentsbuilder.hpp"
+
+using namespace shareddatalayer;
+using namespace shareddatalayer::redis;
+
+AsyncHiredisDatabaseDiscovery::AsyncHiredisDatabaseDiscovery(std::shared_ptr<Engine> engine,
+ const boost::optional<std::string>& ns,
+ DatabaseInfo::Type databaseType,
+ const DatabaseConfiguration::Addresses& databaseAddresses,
+ std::shared_ptr<Logger> logger):
+ engine(engine),
+ ns(ns),
+ databaseType(databaseType),
+ databaseAddresses(databaseAddresses),
+ logger(logger)
+{
+}
+
+void AsyncHiredisDatabaseDiscovery::setStateChangedCb(const StateChangedCb& stateChangedCb)
+{
+ /* Note: Current implementation does not monitor possible database configuration changes.
+ * If that is needed, we would probably need to monitor json configuration file changes.
+ */
+ const DatabaseInfo databaseInfo = { databaseAddresses, databaseType, ns, DatabaseInfo::Discovery::HIREDIS };
+ engine->postCallback(std::bind(stateChangedCb,
+ databaseInfo));
+}
+
+void AsyncHiredisDatabaseDiscovery::clearStateChangedCb()
+{
+}
diff --git a/src/redis/asyncredisreply.cpp b/src/redis/asyncredisreply.cpp
new file mode 100644
index 0000000..cfa0fab
--- /dev/null
+++ b/src/redis/asyncredisreply.cpp
@@ -0,0 +1,89 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/redis/asyncredisreply.hpp"
+
+using namespace shareddatalayer;
+using namespace shareddatalayer::redis;
+
+AsyncRedisReply::AsyncRedisReply():
+ type(Type::NIL),
+ integer(0),
+ dataItem { { }, 0 }
+{
+}
+
+AsyncRedisReply::AsyncRedisReply(const redisReply& rr):
+ integer(0),
+ dataItem { { }, 0 },
+ typeMap { { REDIS_REPLY_NIL, Type::NIL }, { REDIS_REPLY_INTEGER, Type::INTEGER },
+ { REDIS_REPLY_STATUS, Type::STATUS }, { REDIS_REPLY_STRING, Type::STRING },
+ { REDIS_REPLY_ARRAY, Type::ARRAY } }
+{
+ auto res(typeMap.find(rr.type));
+ if (res != typeMap.end())
+ {
+ type = res->second;
+ parseReply(rr);
+ }
+}
+
+AsyncRedisReply::Type AsyncRedisReply::getType() const
+{
+ return type;
+}
+
+long long AsyncRedisReply::getInteger() const
+{
+ return integer;
+}
+
+const AsyncRedisReply::DataItem* AsyncRedisReply::getString() const
+{
+ return &dataItem;
+}
+
+const AsyncRedisReply::ReplyVector* AsyncRedisReply::getArray() const
+{
+ return &replyVector;
+}
+
+void AsyncRedisReply::parseReply(const redisReply& rr)
+{
+ switch (type)
+ {
+ case Type::INTEGER:
+ integer = rr.integer;
+ break;
+ case Type::STATUS:
+ case Type::STRING:
+ dataItem.str = std::string(rr.str, static_cast<size_t>(rr.len));
+ dataItem.len = rr.len;
+ break;
+ case Type::ARRAY:
+ parseArray(rr);
+ break;
+ case Type::NIL:
+ default:
+ break;
+ }
+}
+
+void AsyncRedisReply::parseArray(const redisReply& rr)
+{
+ for (auto i(0U); i < rr.elements; ++i)
+ replyVector.push_back(std::make_shared<AsyncRedisReply>(*rr.element[i]));
+}
diff --git a/src/redis/asyncredisstorage.cpp b/src/redis/asyncredisstorage.cpp
new file mode 100644
index 0000000..22d4b2b
--- /dev/null
+++ b/src/redis/asyncredisstorage.cpp
@@ -0,0 +1,570 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "config.h"
+#include <sstream>
+#include "private/error.hpp"
+#include <sdl/emptynamespace.hpp>
+#include <sdl/invalidnamespace.hpp>
+#include <sdl/publisherid.hpp>
+#include "private/abort.hpp"
+#include "private/createlogger.hpp"
+#include "private/engine.hpp"
+#include "private/logger.hpp"
+#include "private/namespacevalidator.hpp"
+#include "private/configurationreader.hpp"
+#include "private/redis/asynccommanddispatcher.hpp"
+#include "private/redis/asyncdatabasediscovery.hpp"
+#include "private/redis/asyncredisstorage.hpp"
+#include "private/redis/contents.hpp"
+#include "private/redis/contentsbuilder.hpp"
+#include "private/redis/redisgeneral.hpp"
+#include "private/redis/reply.hpp"
+
+using namespace shareddatalayer;
+using namespace shareddatalayer::redis;
+
+/* TODO: This implementation contains lot of duplicated code with old API (asyncRedisConnection).
+ * When this new API is fully ready and tested old API implementation could be changed to utilize this
+ * (bit like sync API utilizes async API).
+ */
+
+namespace
+{
+ std::shared_ptr<AsyncCommandDispatcher> asyncCommandDispatcherCreator(Engine& engine,
+ const DatabaseInfo& databaseInfo,
+ std::shared_ptr<ContentsBuilder> contentsBuilder,
+ std::shared_ptr<Logger> logger)
+ {
+ return AsyncCommandDispatcher::create(engine,
+ databaseInfo,
+ contentsBuilder,
+ false,
+ logger);
+ }
+
+ class AsyncRedisStorageErrorCategory: public std::error_category
+ {
+ public:
+ AsyncRedisStorageErrorCategory() = default;
+
+ const char* name() const noexcept override;
+
+ std::string message(int condition) const override;
+
+ std::error_condition default_error_condition(int condition) const noexcept override;
+ };
+
+ const char* AsyncRedisStorageErrorCategory::name() const noexcept
+ {
+ return "asyncredisstorage";
+ }
+
+ std::string AsyncRedisStorageErrorCategory::message(int condition) const
+ {
+ switch (static_cast<AsyncRedisStorage::ErrorCode>(condition))
+ {
+ case AsyncRedisStorage::ErrorCode::SUCCESS:
+ return std::error_code().message();
+ case AsyncRedisStorage::ErrorCode::REDIS_NOT_YET_DISCOVERED:
+ return "connection to the underlying data storage not yet available";
+ case AsyncRedisStorage::ErrorCode::INVALID_NAMESPACE:
+ return "invalid namespace identifier passed to SDL API";
+ case AsyncRedisStorage::ErrorCode::END_MARKER:
+ logErrorOnce("AsyncRedisStorage::ErrorCode::END_MARKER is not meant to be queried (it is only for enum loop control)");
+ return "unsupported error code for message()";
+ default:
+ return "description missing for AsyncRedisStorageErrorCategory error: " + std::to_string(condition);
+ }
+ }
+
+ std::error_condition AsyncRedisStorageErrorCategory::default_error_condition(int condition) const noexcept
+ {
+ switch (static_cast<AsyncRedisStorage::ErrorCode>(condition))
+ {
+ case AsyncRedisStorage::ErrorCode::SUCCESS:
+ return InternalError::SUCCESS;
+ case AsyncRedisStorage::ErrorCode::REDIS_NOT_YET_DISCOVERED:
+ return InternalError::SDL_NOT_READY;
+ case AsyncRedisStorage::ErrorCode::INVALID_NAMESPACE:
+ return InternalError::SDL_RECEIVED_INVALID_PARAMETER;
+ case AsyncRedisStorage::ErrorCode::END_MARKER:
+ logErrorOnce("AsyncRedisStorage::ErrorCode::END_MARKER is not meant to be mapped to InternalError (it is only for enum loop control)");
+ return InternalError::SDL_ERROR_CODE_LOGIC_ERROR;
+ default:
+ std::ostringstream msg;
+ msg << "default error condition missing for AsyncRedisStorageErrorCategory error: "
+ << condition;
+ logErrorOnce(msg.str());
+ return InternalError::SDL_ERROR_CODE_LOGIC_ERROR;
+ }
+ }
+
+ AsyncStorage::DataMap buildDataMap(const AsyncStorage::Keys& keys, const Reply::ReplyVector& replyVector)
+ {
+ AsyncStorage::DataMap dataMap;
+ auto i(0U);
+ for (const auto& j : keys)
+ {
+ if (replyVector[i]->getType() == Reply::Type::STRING)
+ {
+ AsyncStorage::Data data;
+ auto dataStr(replyVector[i]->getString());
+ for (ReplyStringLength k(0); k < dataStr->len; ++k)
+ data.push_back(static_cast<uint8_t>(dataStr->str[static_cast<size_t>(k)]));
+ dataMap.insert({ j, data });
+ }
+ ++i;
+ }
+ return dataMap;
+ }
+
+ AsyncStorage::Key getKey(const Reply::DataItem& item)
+ {
+ std::string str(item.str.c_str(), static_cast<size_t>(item.len));
+ auto res(str.find(AsyncRedisStorage::SEPARATOR));
+ return str.substr(res + 1);
+ }
+
+ AsyncStorage::Keys getKeys(const Reply::ReplyVector& replyVector)
+ {
+ AsyncStorage::Keys keys;
+ for (const auto& i : replyVector)
+ {
+ if (i->getType() == Reply::Type::STRING)
+ keys.insert(getKey(*i->getString()));
+ }
+ return keys;
+ }
+
+ void escapeRedisSearchPatternCharacters(std::string& stringToProcess)
+ {
+ const std::string redisSearchPatternCharacters = R"(*?[]\)";
+
+ std::size_t foundPosition = stringToProcess.find_first_of(redisSearchPatternCharacters);
+
+ while (foundPosition != std::string::npos)
+ {
+ stringToProcess.insert(foundPosition, R"(\)");
+ foundPosition = stringToProcess.find_first_of(redisSearchPatternCharacters, foundPosition + 2);
+ }
+ }
+}
+
+AsyncRedisStorage::ErrorCode& shareddatalayer::operator++ (AsyncRedisStorage::ErrorCode& ecEnum)
+{
+ if (ecEnum == AsyncRedisStorage::ErrorCode::END_MARKER)
+ throw std::out_of_range("for AsyncRedisStorage::ErrorCode& operator ++");
+ ecEnum = AsyncRedisStorage::ErrorCode(static_cast<std::underlying_type<AsyncRedisStorage::ErrorCode>::type>(ecEnum) + 1);
+ return ecEnum;
+}
+
+std::error_code shareddatalayer::make_error_code(AsyncRedisStorage::ErrorCode errorCode)
+{
+ return std::error_code(static_cast<int>(errorCode), AsyncRedisStorage::errorCategory());
+}
+
+const std::error_category& AsyncRedisStorage::errorCategory() noexcept
+{
+ static const AsyncRedisStorageErrorCategory theAsyncRedisStorageErrorCategory;
+ return theAsyncRedisStorageErrorCategory;
+}
+
+AsyncRedisStorage::AsyncRedisStorage(std::shared_ptr<Engine> engine,
+ std::shared_ptr<AsyncDatabaseDiscovery> discovery,
+ const boost::optional<PublisherId>& pId,
+ std::shared_ptr<NamespaceConfigurations> namespaceConfigurations,
+ std::shared_ptr<Logger> logger):
+ AsyncRedisStorage(engine,
+ discovery,
+ pId,
+ namespaceConfigurations,
+ ::asyncCommandDispatcherCreator,
+ std::make_shared<redis::ContentsBuilder>(SEPARATOR),
+ logger)
+{
+}
+
+AsyncRedisStorage::AsyncRedisStorage(std::shared_ptr<Engine> engine,
+ std::shared_ptr<redis::AsyncDatabaseDiscovery> discovery,
+ const boost::optional<PublisherId>& pId,
+ std::shared_ptr<NamespaceConfigurations> namespaceConfigurations,
+ const AsyncCommandDispatcherCreator& asyncCommandDispatcherCreator,
+ std::shared_ptr<redis::ContentsBuilder> contentsBuilder,
+ std::shared_ptr<Logger> logger):
+ engine(engine),
+ dispatcher(nullptr),
+ discovery(discovery),
+ publisherId(pId),
+ asyncCommandDispatcherCreator(asyncCommandDispatcherCreator),
+ contentsBuilder(contentsBuilder),
+ namespaceConfigurations(namespaceConfigurations),
+ logger(logger)
+{
+ if(publisherId && (*publisherId).empty())
+ {
+ throw std::invalid_argument("AsyncRedisStorage: empty publisher ID string given");
+ }
+
+ discovery->setStateChangedCb([this](const redis::DatabaseInfo& databaseInfo)
+ {
+ serviceStateChanged(databaseInfo);
+ });
+}
+
+AsyncRedisStorage::~AsyncRedisStorage()
+{
+ if (discovery)
+ discovery->clearStateChangedCb();
+ if (dispatcher)
+ dispatcher->disableCommandCallbacks();
+}
+
+redis::DatabaseInfo& AsyncRedisStorage::getDatabaseInfo()
+{
+ return dbInfo;
+}
+
+void AsyncRedisStorage::serviceStateChanged(const redis::DatabaseInfo& newDatabaseInfo)
+{
+ dispatcher = asyncCommandDispatcherCreator(*engine,
+ newDatabaseInfo,
+ contentsBuilder,
+ logger);
+ if (readyAck)
+ dispatcher->waitConnectedAsync([this]()
+ {
+ readyAck(std::error_code());
+ readyAck = ReadyAck();
+ });
+ dbInfo = newDatabaseInfo;
+}
+
+int AsyncRedisStorage::fd() const
+{
+ return engine->fd();
+}
+
+void AsyncRedisStorage::handleEvents()
+{
+ engine->handleEvents();
+}
+
+bool AsyncRedisStorage::canOperationBePerformed(const Namespace& ns,
+ boost::optional<bool> noKeysGiven,
+ std::error_code& ecToReturn)
+{
+ if (!::isValidNamespace(ns))
+ {
+ logErrorOnce("Invalid namespace identifier: " + ns + " passed to SDL");
+ ecToReturn = std::error_code(ErrorCode::INVALID_NAMESPACE);
+ return false;
+ }
+ if (noKeysGiven && *noKeysGiven)
+ {
+ ecToReturn = std::error_code();
+ return false;
+ }
+ if (!dispatcher)
+ {
+ ecToReturn = std::error_code(ErrorCode::REDIS_NOT_YET_DISCOVERED);
+ return false;
+ }
+
+ ecToReturn = std::error_code();
+ return true;
+}
+
+void AsyncRedisStorage::waitReadyAsync(const Namespace&,
+ const ReadyAck& readyAck)
+{
+ if (dispatcher)
+ dispatcher->waitConnectedAsync([readyAck]()
+ {
+ readyAck(std::error_code());
+ });
+ else
+ this->readyAck = readyAck;
+}
+
+void AsyncRedisStorage::setAsync(const Namespace& ns,
+ const DataMap& dataMap,
+ const ModifyAck& modifyAck)
+{
+ std::error_code ec;
+
+ if (!canOperationBePerformed(ns, dataMap.empty(), ec))
+ {
+ engine->postCallback(std::bind(modifyAck, ec));
+ return;
+ }
+
+ if (namespaceConfigurations->areNotificationsEnabled(ns))
+ dispatcher->dispatchAsync(std::bind(&AsyncRedisStorage::modificationCommandCallback,
+ this,
+ std::placeholders::_1,
+ std::placeholders::_2,
+ modifyAck),
+ ns,
+ contentsBuilder->build("MSETPUB", ns, dataMap, ns, getPublishMessage()));
+ else
+ dispatcher->dispatchAsync(std::bind(&AsyncRedisStorage::modificationCommandCallback,
+ this,
+ std::placeholders::_1,
+ std::placeholders::_2,
+ modifyAck),
+ ns,
+ contentsBuilder->build("MSET", ns, dataMap));
+}
+
+void AsyncRedisStorage::modificationCommandCallback(const std::error_code& error,
+ const Reply&,
+ const ModifyAck& modifyAck )
+{
+ modifyAck(error);
+}
+
+void AsyncRedisStorage::conditionalCommandCallback(const std::error_code& error,
+ const Reply& reply,
+ const ModifyIfAck& modifyIfAck)
+{
+ auto type(reply.getType());
+ if (error ||
+ (type == Reply::Type::NIL) || // SETIE(PUB)
+ ((type == Reply::Type::INTEGER) && (reply.getInteger() != 1))) // SETNX(PUB) and DELIE(PUB)
+ modifyIfAck(error, false);
+ else
+ modifyIfAck(error, true);
+}
+
+void AsyncRedisStorage::setIfAsync(const Namespace& ns,
+ const Key& key,
+ const Data& oldData,
+ const Data& newData,
+ const ModifyIfAck& modifyIfAck)
+{
+ std::error_code ec;
+
+ if (!canOperationBePerformed(ns, boost::none, ec))
+ {
+ engine->postCallback(std::bind(modifyIfAck, ec, false));
+ return;
+ }
+
+ if (namespaceConfigurations->areNotificationsEnabled(ns))
+ dispatcher->dispatchAsync(std::bind(&AsyncRedisStorage::conditionalCommandCallback,
+ this,
+ std::placeholders::_1,
+ std::placeholders::_2,
+ modifyIfAck),
+ ns,
+ contentsBuilder->build("SETIEPUB", ns, key, newData, oldData, ns, getPublishMessage()));
+ else
+ dispatcher->dispatchAsync(std::bind(&AsyncRedisStorage::conditionalCommandCallback,
+ this,
+ std::placeholders::_1,
+ std::placeholders::_2,
+ modifyIfAck),
+ ns,
+ contentsBuilder->build("SETIE", ns, key, newData, oldData));
+}
+
+void AsyncRedisStorage::removeIfAsync(const Namespace& ns,
+ const Key& key,
+ const Data& data,
+ const ModifyIfAck& modifyIfAck)
+{
+ std::error_code ec;
+
+ if (!canOperationBePerformed(ns, boost::none, ec))
+ {
+ engine->postCallback(std::bind(modifyIfAck, ec, false));
+ return;
+ }
+
+ if (namespaceConfigurations->areNotificationsEnabled(ns))
+ dispatcher->dispatchAsync(std::bind(&AsyncRedisStorage::conditionalCommandCallback,
+ this,
+ std::placeholders::_1,
+ std::placeholders::_2,
+ modifyIfAck),
+ ns,
+ contentsBuilder->build("DELIEPUB", ns, key, data, ns, getPublishMessage()));
+ else
+ dispatcher->dispatchAsync(std::bind(&AsyncRedisStorage::conditionalCommandCallback,
+ this,
+ std::placeholders::_1,
+ std::placeholders::_2,
+ modifyIfAck),
+ ns,
+ contentsBuilder->build("DELIE", ns, key, data));
+}
+
+std::string AsyncRedisStorage::getPublishMessage() const
+{
+ if(publisherId)
+ return *publisherId;
+ else
+ return NO_PUBLISHER;
+}
+
+void AsyncRedisStorage::setIfNotExistsAsync(const Namespace& ns,
+ const Key& key,
+ const Data& data,
+ const ModifyIfAck& modifyIfAck)
+{
+ std::error_code ec;
+
+ if (!canOperationBePerformed(ns, boost::none, ec))
+ {
+ engine->postCallback(std::bind(modifyIfAck, ec, false));
+ return;
+ }
+
+ if (namespaceConfigurations->areNotificationsEnabled(ns))
+ dispatcher->dispatchAsync(std::bind(&AsyncRedisStorage::conditionalCommandCallback,
+ this,
+ std::placeholders::_1,
+ std::placeholders::_2,
+ modifyIfAck),
+ ns,
+ contentsBuilder->build("SETNXPUB", ns, key, data, ns ,getPublishMessage()));
+ else
+ dispatcher->dispatchAsync(std::bind(&AsyncRedisStorage::conditionalCommandCallback,
+ this,
+ std::placeholders::_1,
+ std::placeholders::_2,
+ modifyIfAck),
+ ns,
+ contentsBuilder->build("SETNX", ns, key, data));
+}
+
+void AsyncRedisStorage::getAsync(const Namespace& ns,
+ const Keys& keys,
+ const GetAck& getAck)
+{
+ std::error_code ec;
+
+ if (!canOperationBePerformed(ns, keys.empty(), ec))
+ {
+ engine->postCallback(std::bind(getAck, ec, DataMap()));
+ return;
+ }
+
+ dispatcher->dispatchAsync([getAck, keys](const std::error_code& error,
+ const Reply& reply)
+ {
+ if (error)
+ getAck(error, DataMap());
+ else
+ getAck(std::error_code(), buildDataMap(keys, *reply.getArray()));
+ },
+ ns,
+ contentsBuilder->build("MGET", ns, keys));
+}
+
+void AsyncRedisStorage::removeAsync(const Namespace& ns,
+ const Keys& keys,
+ const ModifyAck& modifyAck)
+{
+ std::error_code ec;
+
+ if (!canOperationBePerformed(ns, keys.empty(), ec))
+ {
+ engine->postCallback(std::bind(modifyAck, ec));
+ return;
+ }
+
+ if (namespaceConfigurations->areNotificationsEnabled(ns))
+ dispatcher->dispatchAsync(std::bind(&AsyncRedisStorage::modificationCommandCallback,
+ this,
+ std::placeholders::_1,
+ std::placeholders::_2,
+ modifyAck),
+ ns,
+ contentsBuilder->build("DELPUB", ns, keys, ns, getPublishMessage()));
+ else
+ dispatcher->dispatchAsync(std::bind(&AsyncRedisStorage::modificationCommandCallback,
+ this,
+ std::placeholders::_1,
+ std::placeholders::_2,
+ modifyAck),
+ ns,
+ contentsBuilder->build("DEL", ns, keys));
+}
+
+void AsyncRedisStorage::findKeysAsync(const Namespace& ns,
+ const std::string& keyPrefix,
+ const FindKeysAck& findKeysAck)
+{
+ //TODO: update to more optimal solution than current KEYS-based one.
+ std::error_code ec;
+
+ if (!canOperationBePerformed(ns, boost::none, ec))
+ {
+ engine->postCallback(std::bind(findKeysAck, ec, Keys()));
+ return;
+ }
+
+ dispatcher->dispatchAsync([findKeysAck](const std::error_code& error, const Reply& reply)
+ {
+ if (error)
+ findKeysAck(error, Keys());
+ else
+ findKeysAck(std::error_code(), getKeys(*reply.getArray()));
+ },
+ ns,
+ contentsBuilder->build("KEYS", buildKeyPrefixSearchPattern(ns, keyPrefix)));
+}
+
+void AsyncRedisStorage::removeAllAsync(const Namespace& ns,
+ const ModifyAck& modifyAck)
+{
+ std::error_code ec;
+
+ if (!canOperationBePerformed(ns, boost::none, ec))
+ {
+ engine->postCallback(std::bind(modifyAck, ec));
+ return;
+ }
+
+ dispatcher->dispatchAsync([this, modifyAck, ns](const std::error_code& error, const Reply& reply)
+ {
+ if (error)
+ {
+ modifyAck(error);
+ return;
+ }
+ const auto& array(*reply.getArray());
+ if (array.empty())
+ modifyAck(std::error_code());
+ else
+ {
+ removeAsync(ns, getKeys(array), modifyAck);
+ }
+ },
+ ns,
+ contentsBuilder->build("KEYS", buildKeyPrefixSearchPattern(ns, "")));
+}
+
+std::string AsyncRedisStorage::buildKeyPrefixSearchPattern(const Namespace& ns, const std::string& keyPrefix) const
+{
+ std::string escapedKeyPrefix = keyPrefix;
+ escapeRedisSearchPatternCharacters(escapedKeyPrefix);
+ std::ostringstream oss;
+ oss << '{' << ns << '}' << SEPARATOR << escapedKeyPrefix << "*";
+ return oss.str();
+}
diff --git a/src/redis/contentsbuilder.cpp b/src/redis/contentsbuilder.cpp
new file mode 100644
index 0000000..6370be1
--- /dev/null
+++ b/src/redis/contentsbuilder.cpp
@@ -0,0 +1,208 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/redis/contentsbuilder.hpp"
+#include "private/redis/contents.hpp"
+
+using namespace shareddatalayer;
+using namespace shareddatalayer::redis;
+
+ContentsBuilder::ContentsBuilder(const char nsKeySeparator):
+ nsKeySeparator(nsKeySeparator)
+{
+}
+
+ContentsBuilder::~ContentsBuilder()
+{
+}
+
+Contents ContentsBuilder::build(const std::string& string) const
+{
+ Contents contents;
+ addString(contents, string);
+ return contents;
+}
+
+Contents ContentsBuilder::build(const std::string& string,
+ const std::string& string2) const
+{
+ Contents contents;
+ addString(contents, string);
+ addString(contents, string2);
+ return contents;
+}
+
+Contents ContentsBuilder::build(const std::string& string,
+ const std::string& string2,
+ const std::string& string3) const
+{
+ Contents contents;
+ addString(contents, string);
+ addString(contents, string2);
+ addString(contents, string3);
+ return contents;
+}
+
+Contents ContentsBuilder::build(const std::string& string,
+ const AsyncConnection::Namespace& ns,
+ const AsyncConnection::DataMap& dataMap) const
+{
+ Contents contents;
+ addString(contents, string);
+ addDataMap(contents, ns, dataMap);
+ return contents;
+}
+
+Contents ContentsBuilder::build(const std::string& string,
+ const AsyncConnection::Namespace& ns,
+ const AsyncConnection::DataMap& dataMap,
+ const std::string& string2,
+ const std::string& string3) const
+{
+ Contents contents;
+ addString(contents, string);
+ addDataMap(contents, ns, dataMap);
+ addString(contents, string2);
+ addString(contents, string3);
+ return contents;
+}
+
+Contents ContentsBuilder::build(const std::string& string,
+ const AsyncConnection::Namespace& ns,
+ const AsyncConnection::Key& key,
+ const AsyncConnection::Data& data) const
+{
+ Contents contents;
+ addString(contents, string);
+ addKey(contents, ns, key);
+ addData(contents, data);
+ return contents;
+}
+
+Contents ContentsBuilder::build(const std::string& string,
+ const AsyncConnection::Namespace& ns,
+ const AsyncConnection::Key& key,
+ const AsyncConnection::Data& data,
+ const std::string& string2,
+ const std::string& string3) const
+{
+ Contents contents;
+ addString(contents, string);
+ addKey(contents, ns, key);
+ addData(contents, data);
+ addString(contents, string2);
+ addString(contents, string3);
+ return contents;
+}
+
+Contents ContentsBuilder::build(const std::string& string,
+ const AsyncConnection::Namespace& ns,
+ const AsyncConnection::Key& key,
+ const AsyncConnection::Data& data,
+ const AsyncConnection::Data& data2) const
+{
+ Contents contents;
+ addString(contents, string);
+ addKey(contents, ns, key);
+ addData(contents, data);
+ addData(contents, data2);
+ return contents;
+}
+
+Contents ContentsBuilder::build(const std::string& string,
+ const AsyncConnection::Namespace& ns,
+ const AsyncConnection::Key& key,
+ const AsyncConnection::Data& data,
+ const AsyncConnection::Data& data2,
+ const std::string& string2,
+ const std::string& string3) const
+{
+ Contents contents;
+ addString(contents, string);
+ addKey(contents, ns, key);
+ addData(contents, data);
+ addData(contents, data2);
+ addString(contents, string2);
+ addString(contents, string3);
+ return contents;
+}
+
+Contents ContentsBuilder::build(const std::string& string,
+ const AsyncConnection::Namespace& ns,
+ const AsyncConnection::Keys& keys) const
+{
+ Contents contents;
+ addString(contents, string);
+ addKeys(contents, ns, keys);
+ return contents;
+}
+
+Contents ContentsBuilder::build(const std::string& string,
+ const AsyncConnection::Namespace& ns,
+ const AsyncConnection::Keys& keys,
+ const std::string& string2,
+ const std::string& string3) const
+{
+ Contents contents;
+ addString(contents, string);
+ addKeys(contents, ns, keys);
+ addString(contents, string2);
+ addString(contents, string3);
+ return contents;
+}
+
+void ContentsBuilder::addString(Contents& contents,
+ const std::string& string) const
+{
+ contents.stack.push_back(string);
+ contents.sizes.push_back(string.size());
+}
+
+void ContentsBuilder::addDataMap(Contents& contents,
+ const AsyncConnection::Namespace& ns,
+ const AsyncConnection::DataMap& dataMap) const
+{
+ for (const auto& i : dataMap)
+ {
+ addKey(contents, ns, i.first);
+ addData(contents, i.second);
+ }
+}
+
+void ContentsBuilder::addKey(Contents& contents,
+ const AsyncConnection::Namespace& ns,
+ const AsyncConnection::Key& key) const
+{
+ auto content('{' + ns + '}' + nsKeySeparator + key);
+ contents.stack.push_back(content);
+ contents.sizes.push_back((content).size());
+}
+
+void ContentsBuilder::addData(Contents& contents,
+ const AsyncConnection::Data& data) const
+{
+ contents.stack.push_back(std::string(reinterpret_cast<const char*>(data.data()),
+ static_cast<size_t>(data.size())));
+ contents.sizes.push_back(data.size());
+}
+
+void ContentsBuilder::addKeys(Contents& contents,
+ const AsyncConnection::Namespace& ns,
+ const AsyncConnection::Keys& keys) const
+{
+ for (const auto& i : keys)
+ addKey(contents, ns, i);
+}
diff --git a/src/redis/hiredisclusterepolladapter.cpp b/src/redis/hiredisclusterepolladapter.cpp
new file mode 100644
index 0000000..10d3a57
--- /dev/null
+++ b/src/redis/hiredisclusterepolladapter.cpp
@@ -0,0 +1,177 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/redis/hiredisclusterepolladapter.hpp"
+#include <sys/epoll.h>
+#include "private/engine.hpp"
+#include "private/redis/hiredisclustersystem.hpp"
+
+using namespace shareddatalayer;
+using namespace shareddatalayer::redis;
+
+namespace
+{
+ int attachFunction(redisAsyncContext* ac, void* data)
+ {
+ auto instance(static_cast<HiredisClusterEpollAdapter*>(data));
+ instance->attach(ac);
+ return REDIS_OK;
+ }
+
+ void addReadWrap(void* data)
+ {
+ auto instance(static_cast<HiredisClusterEpollAdapter::Node*>(data));
+ instance->addRead();
+ }
+
+ void addWriteWrap(void* data)
+ {
+ auto instance(static_cast<HiredisClusterEpollAdapter::Node*>(data));
+ instance->addWrite();
+ }
+
+ void delReadWrap(void* data)
+ {
+ auto instance(static_cast<HiredisClusterEpollAdapter::Node*>(data));
+ instance->delRead();
+ }
+
+ void delWriteWrap(void* data)
+ {
+ auto instance(static_cast<HiredisClusterEpollAdapter::Node*>(data));
+ instance->delWrite();
+ }
+
+ void cleanupWrap(void* data)
+ {
+ auto instance(static_cast<HiredisClusterEpollAdapter::Node*>(data));
+ instance->cleanup();
+ }
+}
+
+HiredisClusterEpollAdapter::HiredisClusterEpollAdapter(Engine& engine):
+ HiredisClusterEpollAdapter(engine, HiredisClusterSystem::getInstance())
+{
+}
+
+HiredisClusterEpollAdapter::HiredisClusterEpollAdapter(Engine& engine, HiredisClusterSystem& hiredisClusterSystem):
+ engine(engine),
+ hiredisClusterSystem(hiredisClusterSystem)
+{
+}
+
+void HiredisClusterEpollAdapter::setup(redisClusterAsyncContext* acc)
+{
+ acc->adapter = this;
+ acc->attach_fn = attachFunction;
+}
+
+void HiredisClusterEpollAdapter::attach(redisAsyncContext* ac)
+{
+ detach(ac);
+ nodes.insert(std::make_pair(ac->c.fd,
+ std::unique_ptr<Node>(new Node(engine,
+ ac,
+ hiredisClusterSystem))));
+}
+
+void HiredisClusterEpollAdapter::detach(const redisAsyncContext* ac)
+{
+ auto it = nodes.find(ac->c.fd);
+ if (it != nodes.end())
+ nodes.erase(it);
+}
+
+HiredisClusterEpollAdapter::Node::Node(Engine& engine,
+ redisAsyncContext* ac,
+ HiredisClusterSystem& hiredisClusterSystem):
+ engine(engine),
+ hiredisClusterSystem(hiredisClusterSystem),
+ ac(ac),
+ eventState(0),
+ reading(false),
+ writing(false)
+{
+ this->ac->ev.data = this;
+ this->ac->ev.addRead = addReadWrap;
+ this->ac->ev.addWrite = addWriteWrap;
+ this->ac->ev.delRead = delReadWrap;
+ this->ac->ev.delWrite = delWriteWrap;
+ this->ac->ev.cleanup = cleanupWrap;
+ engine.addMonitoredFD(ac->c.fd,
+ eventState,
+ std::bind(&HiredisClusterEpollAdapter::Node::eventHandler,
+ this,
+ std::placeholders::_1));
+ isMonitoring = true;
+}
+
+HiredisClusterEpollAdapter::Node::~Node()
+{
+ if (isMonitoring)
+ cleanup();
+}
+
+void HiredisClusterEpollAdapter::Node::eventHandler(unsigned int events)
+{
+ if (events & Engine::EVENT_IN)
+ if (reading && isMonitoring)
+ hiredisClusterSystem.redisAsyncHandleRead(ac);
+ if (events & Engine::EVENT_OUT)
+ if (writing && isMonitoring)
+ hiredisClusterSystem.redisAsyncHandleWrite(ac);
+}
+
+void HiredisClusterEpollAdapter::Node::addRead()
+{
+ if (reading)
+ return;
+ reading = true;
+ eventState |= Engine::EVENT_IN;
+ engine.modifyMonitoredFD(ac->c.fd, eventState);
+}
+
+void HiredisClusterEpollAdapter::Node::addWrite()
+{
+ if (writing)
+ return;
+ writing = true;
+ eventState |= Engine::EVENT_OUT;
+ engine.modifyMonitoredFD(ac->c.fd, eventState);
+}
+
+void HiredisClusterEpollAdapter::Node::delRead()
+{
+ reading = false;
+ eventState &= ~Engine::EVENT_IN;
+ engine.modifyMonitoredFD(ac->c.fd, eventState);
+}
+
+void HiredisClusterEpollAdapter::Node::delWrite()
+{
+ writing = false;
+ eventState &= ~Engine::EVENT_OUT;
+ engine.modifyMonitoredFD(ac->c.fd, eventState);
+}
+
+void HiredisClusterEpollAdapter::Node::cleanup()
+{
+ reading = false;
+ writing = false;
+ eventState = 0;
+ engine.deleteMonitoredFD(ac->c.fd);
+ isMonitoring = false;
+}
diff --git a/src/redis/hiredisclustersystem.cpp b/src/redis/hiredisclustersystem.cpp
new file mode 100644
index 0000000..7b3b25f
--- /dev/null
+++ b/src/redis/hiredisclustersystem.cpp
@@ -0,0 +1,76 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/redis/hiredisclustersystem.hpp"
+
+using namespace shareddatalayer::redis;
+
+redisClusterAsyncContext* HiredisClusterSystem::redisClusterAsyncConnect(const char* addrs, int flags)
+{
+ return ::redisClusterAsyncConnect(addrs, flags);
+}
+
+int HiredisClusterSystem::redisClusterAsyncSetConnectCallback(redisClusterAsyncContext* acc,
+ redisClusterInstanceConnectCallback* fn)
+{
+ return ::redisClusterAsyncSetConnectCallback(acc, fn);
+}
+
+int HiredisClusterSystem::redisClusterAsyncSetDisconnectCallback(redisClusterAsyncContext* acc,
+ redisClusterInstanceDisconnectCallback* fn)
+{
+ return ::redisClusterAsyncSetDisconnectCallback(acc, fn);
+}
+
+int HiredisClusterSystem::redisClusterAsyncCommandArgv(redisClusterAsyncContext* acc, redisClusterCallbackFn* fn,
+ void* privdata, int argc, const char** argv,
+ const size_t* argvlen)
+{
+ return ::redisClusterAsyncCommandArgv(acc, fn, privdata, argc, argv, argvlen);
+}
+
+int HiredisClusterSystem::redisClusterAsyncCommandArgvWithKey(redisClusterAsyncContext *acc, redisClusterCallbackFn *fn,
+ void *privdata, const char *key, int keylen, int argc,
+ const char **argv, const size_t *argvlen)
+{
+ return ::redisClusterAsyncCommandArgvWithKey(acc, fn, privdata, key, keylen, argc, argv, argvlen);
+}
+
+void HiredisClusterSystem::redisAsyncHandleRead(redisAsyncContext* ac)
+{
+ ::redisAsyncHandleRead(ac);
+}
+
+void HiredisClusterSystem::redisAsyncHandleWrite(redisAsyncContext* ac)
+{
+ ::redisAsyncHandleWrite(ac);
+}
+
+void HiredisClusterSystem::redisClusterAsyncDisconnect(redisClusterAsyncContext* acc)
+{
+ ::redisClusterAsyncDisconnect(acc);
+}
+
+void HiredisClusterSystem::redisClusterAsyncFree(redisClusterAsyncContext* acc)
+{
+ ::redisClusterAsyncFree(acc);
+}
+
+HiredisClusterSystem& HiredisClusterSystem::getInstance()
+{
+ static HiredisClusterSystem instance;
+ return instance;
+}
diff --git a/src/redis/hiredisepolladapter.cpp b/src/redis/hiredisepolladapter.cpp
new file mode 100644
index 0000000..11c2252
--- /dev/null
+++ b/src/redis/hiredisepolladapter.cpp
@@ -0,0 +1,150 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/redis/hiredisepolladapter.hpp"
+#include <sys/epoll.h>
+#include "private/engine.hpp"
+#include "private/redis/hiredissystem.hpp"
+
+using namespace shareddatalayer;
+using namespace shareddatalayer::redis;
+
+
+namespace
+{
+ void addReadWrap(void* data)
+ {
+ auto instance(static_cast<HiredisEpollAdapter*>(data));
+ instance->addRead();
+ }
+
+ void delReadWrap(void* data)
+ {
+ auto instance(static_cast<HiredisEpollAdapter*>(data));
+ instance->delRead();
+ }
+
+ void addWriteWrap(void* data)
+ {
+ auto instance(static_cast<HiredisEpollAdapter*>(data));
+ instance->addWrite();
+ }
+
+ void delWriteWrap(void* data)
+ {
+ auto instance(static_cast<HiredisEpollAdapter*>(data));
+ instance->delWrite();
+ }
+
+ void cleanUpWrap(void* data)
+ {
+ auto instance(static_cast<HiredisEpollAdapter*>(data));
+ instance->cleanUp();
+ }
+}
+
+HiredisEpollAdapter::HiredisEpollAdapter(Engine& engine):
+ HiredisEpollAdapter(engine, HiredisSystem::getHiredisSystem())
+{
+}
+
+HiredisEpollAdapter::HiredisEpollAdapter(Engine& engine, HiredisSystem& hiredisSystem):
+ engine(engine),
+ hiredisSystem(hiredisSystem),
+ ac(nullptr),
+ eventState(0),
+ reading(false),
+ writing(false),
+ isMonitoring(false)
+{
+}
+
+HiredisEpollAdapter::~HiredisEpollAdapter()
+{
+ if (ac && isMonitoring)
+ cleanUp();
+}
+
+void HiredisEpollAdapter::attach(redisAsyncContext* ac)
+{
+ eventState = 0;
+ reading = false;
+ writing = false;
+ this->ac = ac;
+ this->ac->ev.data = this;
+ this->ac->ev.addRead = addReadWrap;
+ this->ac->ev.delRead = delReadWrap;
+ this->ac->ev.addWrite = addWriteWrap;
+ this->ac->ev.delWrite = delWriteWrap;
+ this->ac->ev.cleanup = cleanUpWrap;
+ engine.addMonitoredFD(ac->c.fd,
+ eventState,
+ std::bind(&HiredisEpollAdapter::eventHandler,
+ this,
+ std::placeholders::_1));
+ isMonitoring = true;
+}
+
+void HiredisEpollAdapter::eventHandler(unsigned int events)
+{
+ if (events & Engine::EVENT_IN)
+ if (reading)
+ hiredisSystem.redisAsyncHandleRead(ac);
+ if (events & Engine::EVENT_OUT)
+ if (writing)
+ hiredisSystem.redisAsyncHandleWrite(ac);
+}
+
+void HiredisEpollAdapter::addRead()
+{
+ if (reading)
+ return;
+ reading = true;
+ eventState |= Engine::EVENT_IN;
+ engine.modifyMonitoredFD(ac->c.fd, eventState);
+}
+
+void HiredisEpollAdapter::delRead()
+{
+ reading = false;
+ eventState &= ~Engine::EVENT_IN;
+ engine.modifyMonitoredFD(ac->c.fd, eventState);
+}
+
+void HiredisEpollAdapter::addWrite()
+{
+ if (writing)
+ return;
+ writing = true;
+ eventState |= Engine::EVENT_OUT;
+ engine.modifyMonitoredFD(ac->c.fd, eventState);
+}
+
+void HiredisEpollAdapter::delWrite()
+{
+ writing = false;
+ eventState &= ~Engine::EVENT_OUT;
+ engine.modifyMonitoredFD(ac->c.fd, eventState);
+}
+
+void HiredisEpollAdapter::cleanUp()
+{
+ reading = false;
+ writing = false;
+ eventState = 0;
+ engine.deleteMonitoredFD(ac->c.fd);
+ isMonitoring = false;
+}
diff --git a/src/redis/hiredissystem.cpp b/src/redis/hiredissystem.cpp
new file mode 100644
index 0000000..2d37920
--- /dev/null
+++ b/src/redis/hiredissystem.cpp
@@ -0,0 +1,90 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/redis/hiredissystem.hpp"
+
+using namespace shareddatalayer::redis;
+
+redisContext* HiredisSystem::redisConnect(const char* ip, int port)
+{
+ return ::redisConnect(ip, port);
+}
+
+void* HiredisSystem::redisCommandArgv(redisContext* context, int argc, const char** argv, const size_t* argvlen)
+{
+ return ::redisCommandArgv(context, argc, argv, argvlen);
+}
+
+void HiredisSystem::freeReplyObject(void* reply)
+{
+ ::freeReplyObject(reply);
+}
+
+void HiredisSystem::redisFree(redisContext* context)
+{
+ ::redisFree(context);
+}
+
+redisAsyncContext* HiredisSystem::redisAsyncConnect(const char* ip, int port)
+{
+ return ::redisAsyncConnect(ip, port);
+}
+
+int HiredisSystem::redisAsyncSetConnectCallback(redisAsyncContext* ac, redisConnectCallback* fn)
+{
+ return ::redisAsyncSetConnectCallback(ac, fn);
+}
+
+int HiredisSystem::redisAsyncSetDisconnectCallback(redisAsyncContext* ac, redisDisconnectCallback* fn)
+{
+ return ::redisAsyncSetDisconnectCallback(ac, fn);
+}
+
+int HiredisSystem::redisAsyncCommandArgv(redisAsyncContext* ac,
+ redisCallbackFn* fn,
+ void* privdata,
+ int argc,
+ const char** argv,
+ const size_t* argvlen)
+{
+ return ::redisAsyncCommandArgv(ac, fn, privdata, argc, argv, argvlen);
+}
+
+void HiredisSystem::redisAsyncHandleRead(redisAsyncContext* ac)
+{
+ ::redisAsyncHandleRead(ac);
+}
+
+void HiredisSystem::redisAsyncHandleWrite(redisAsyncContext* ac)
+{
+ ::redisAsyncHandleWrite(ac);
+}
+
+void HiredisSystem::redisAsyncDisconnect(redisAsyncContext* ac)
+{
+ ::redisAsyncDisconnect(ac);
+}
+
+void HiredisSystem::redisAsyncFree(redisAsyncContext* ac)
+{
+ ::redisAsyncFree(ac);
+}
+
+HiredisSystem& HiredisSystem::getHiredisSystem() noexcept
+{
+ static HiredisSystem hiredisSystem;
+ return hiredisSystem;
+}
diff --git a/src/redis/redisgeneral.cpp b/src/redis/redisgeneral.cpp
new file mode 100644
index 0000000..eb837ae
--- /dev/null
+++ b/src/redis/redisgeneral.cpp
@@ -0,0 +1,194 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "config.h"
+#include "private/createlogger.hpp"
+#include "private/error.hpp"
+#include "private/redis/redisgeneral.hpp"
+#include <arpa/inet.h>
+#include <cstring>
+#if HAVE_HIREDIS_VIP
+#include <hircluster.h>
+#endif
+#include <hiredis/hiredis.h>
+#include <sstream>
+
+using namespace shareddatalayer;
+using namespace shareddatalayer::redis;
+
+namespace
+{
+ bool equals(const std::string& s1, const char* s2, size_t s2Len)
+ {
+ if (s2 == nullptr)
+ {
+ logErrorOnce("redisGeneral: null pointer passed to equals function");
+ return false;
+ }
+
+ return ((s1.size() == s2Len) && (std::memcmp(s1.data(), s2, s2Len) == 0));
+ }
+
+ bool startsWith(const std::string& s1, const char* s2, size_t s2Len)
+ {
+ if (s2 == nullptr)
+ {
+ logErrorOnce("redisGeneral: null pointer passed to startsWith function");
+ return false;
+ }
+
+ return ((s1.size() <= s2Len) && (std::memcmp(s1.data(), s2, s1.size()) == 0));
+ }
+
+ AsyncRedisCommandDispatcherErrorCode mapRedisReplyErrorToSdlError(const redisReply* rr)
+ {
+ if (equals("LOADING Redis is loading the dataset in memory", rr->str, static_cast<size_t>(rr->len)))
+ return AsyncRedisCommandDispatcherErrorCode::DATASET_LOADING;
+
+ /* This error reply comes when some cluster node(s) is down and rest of the cluster
+ * nodes cannot operate due to that. This error reply typically comes from nodes
+ * which are working but cannot handle requests because other node(s) are down.
+ * Nodes which are actually down (under failover handling) typically return
+ * CLUSTER_ERROR_NOT_CONNECTED or CLUSTER_ERROR_CONNECTION_LOST.
+ */
+ if (startsWith("CLUSTERDOWN", rr->str, static_cast<size_t>(rr->len)))
+ return AsyncRedisCommandDispatcherErrorCode::NOT_CONNECTED;
+
+ if (startsWith("ERR Protocol error", rr->str, static_cast<size_t>(rr->len)))
+ return AsyncRedisCommandDispatcherErrorCode::PROTOCOL_ERROR;
+
+ std::ostringstream oss;
+ oss << "redis reply error: " << std::string(rr->str, static_cast<size_t>(rr->len));
+ logErrorOnce(oss.str());
+ return AsyncRedisCommandDispatcherErrorCode::UNKNOWN_ERROR;
+ }
+
+ AsyncRedisCommandDispatcherErrorCode mapRedisContextErrorToSdlError(int redisContextErr, const char* redisContextErrstr)
+ {
+ switch (redisContextErr)
+ {
+ /* From hiredis/read.h:
+ * When an error occurs, the err flag in a context is set to hold the type of
+ * error that occurred. REDIS_ERR_IO means there was an I/O error and you
+ * should use the "errno" variable to find out what is wrong.
+ * For other values, the "errstr" field will hold a description. */
+ case REDIS_ERR_IO:
+ if (errno == ECONNRESET)
+ return AsyncRedisCommandDispatcherErrorCode::CONNECTION_LOST;
+ logErrorOnce("redis io error. Errno: " + std::to_string(errno));
+ return AsyncRedisCommandDispatcherErrorCode::IO_ERROR;
+ case REDIS_ERR_EOF:
+ return AsyncRedisCommandDispatcherErrorCode::CONNECTION_LOST;
+ case REDIS_ERR_PROTOCOL:
+ return AsyncRedisCommandDispatcherErrorCode::PROTOCOL_ERROR;
+ case REDIS_ERR_OOM:
+ return AsyncRedisCommandDispatcherErrorCode::OUT_OF_MEMORY;
+#if HAVE_HIREDIS_VIP
+ /* hiredis_vip returns CLUSTER_ERROR_NOT_CONNECTED when cluster node is disconnected
+ * but failover handling has not started yet (node_timeout not elapsed yet). In
+ * this situation hiredis_vip does not send request to redis (as it is clear that
+ * request cannot succeed), therefore we can map this error to NOT_CONNECTED error
+ * as it best descripes this situation.
+ */
+ case CLUSTER_ERROR_NOT_CONNECTED:
+ return AsyncRedisCommandDispatcherErrorCode::NOT_CONNECTED;
+ /* hiredis_vip returns CLUSTER_ERROR_CONNECTION_LOST when connection is lost while
+ * hiredis is waiting for reply from redis.
+ */
+ case CLUSTER_ERROR_CONNECTION_LOST:
+ return AsyncRedisCommandDispatcherErrorCode::CONNECTION_LOST;
+#endif
+ default:
+ std::ostringstream oss;
+ oss << "redis error: "
+ << redisContextErrstr
+ << " (" << redisContextErr << ")";
+ logErrorOnce(oss.str());
+ return AsyncRedisCommandDispatcherErrorCode::UNKNOWN_ERROR;
+ }
+ }
+}
+
+namespace shareddatalayer
+{
+ namespace redis
+ {
+ std::string formatToClusterSyntax(const DatabaseConfiguration::Addresses& addresses)
+ {
+ std::ostringstream oss;
+ for (auto i(addresses.begin()); i != addresses.end(); ++i)
+ {
+ oss << i->getHost() << ':' << ntohs(i->getPort());
+ if (i == --addresses.end())
+ break;
+ else
+ oss << ',';
+ }
+ return oss.str();
+ }
+
+ const std::set<std::string>& getRequiredRedisModuleCommands()
+ {
+ static const std::set<std::string> requiredRedisModuleCommands({"msetpub","setie","setiepub","setnxpub","delpub","delie","deliepub"});
+ return requiredRedisModuleCommands;
+ }
+
+ std::error_code getRedisError(int redisContextErr, const char* redisContextErrstr, const redisReply* rr)
+ {
+ if (rr != nullptr)
+ {
+ if (rr->type != REDIS_REPLY_ERROR)
+ return std::error_code();
+
+ return std::error_code(mapRedisReplyErrorToSdlError(rr));
+ }
+
+ return std::error_code(mapRedisContextErrorToSdlError(redisContextErr, redisContextErrstr));
+ }
+
+ std::set<std::string> parseCommandListReply(const redis::Reply& reply)
+ {
+ std::set<std::string> availableCommands;
+ auto replyArray(reply.getArray());
+ for (const auto& j : *replyArray)
+ {
+ auto element = j->getArray();
+ auto command = element->front()->getString();
+ availableCommands.insert(command->str);
+ }
+ return availableCommands;
+ }
+
+ bool checkRedisModuleCommands(const std::set<std::string>& availableCommands)
+ {
+ std::set<std::string> missingCommands;
+
+ for (const auto& i : getRequiredRedisModuleCommands())
+ {
+ const auto it = availableCommands.find(i);
+ if (it == availableCommands.end())
+ missingCommands.insert(i);
+ }
+ if (!missingCommands.empty())
+ {
+ logErrorOnce("Missing Redis module extension commands:");
+ for (const auto& i : missingCommands)
+ logErrorOnce(i);
+ }
+ return missingCommands.empty();
+ }
+ }
+}
diff --git a/src/rejectedbybackend.cpp b/src/rejectedbybackend.cpp
new file mode 100644
index 0000000..8529086
--- /dev/null
+++ b/src/rejectedbybackend.cpp
@@ -0,0 +1,24 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include <sdl/rejectedbybackend.hpp>
+
+using namespace shareddatalayer;
+
+RejectedByBackend::RejectedByBackend(const std::string& error):
+ Exception(error)
+{
+}
diff --git a/src/rejectedbysdl.cpp b/src/rejectedbysdl.cpp
new file mode 100644
index 0000000..09a864c
--- /dev/null
+++ b/src/rejectedbysdl.cpp
@@ -0,0 +1,24 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include <sdl/rejectedbysdl.hpp>
+
+using namespace shareddatalayer;
+
+RejectedBySdl::RejectedBySdl(const std::string& error):
+ Exception(error)
+{
+}
diff --git a/src/syncstorage.cpp b/src/syncstorage.cpp
new file mode 100644
index 0000000..a1d7e54
--- /dev/null
+++ b/src/syncstorage.cpp
@@ -0,0 +1,26 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/syncstorageimpl.hpp"
+#include <sdl/asyncstorage.hpp>
+#include <sdl/syncstorage.hpp>
+
+using namespace shareddatalayer;
+
+std::unique_ptr<SyncStorage> SyncStorage::create()
+{
+ return std::unique_ptr<SyncStorageImpl>(new SyncStorageImpl(AsyncStorage::create()));
+}
diff --git a/src/syncstorageimpl.cpp b/src/syncstorageimpl.cpp
new file mode 100644
index 0000000..b73029c
--- /dev/null
+++ b/src/syncstorageimpl.cpp
@@ -0,0 +1,238 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include <sstream>
+#include <sys/poll.h>
+#include <sdl/asyncstorage.hpp>
+#include <sdl/backenderror.hpp>
+#include <sdl/errorqueries.hpp>
+#include <sdl/invalidnamespace.hpp>
+#include <sdl/notconnected.hpp>
+#include <sdl/operationinterrupted.hpp>
+#include <sdl/rejectedbybackend.hpp>
+#include <sdl/rejectedbysdl.hpp>
+#include "private/redis/asyncredisstorage.hpp"
+#include "private/syncstorageimpl.hpp"
+#include "private/system.hpp"
+
+using namespace shareddatalayer;
+
+namespace
+{
+ void throwExceptionForErrorCode[[ noreturn ]](const std::error_code& ec)
+ {
+ if (ec == shareddatalayer::Error::BACKEND_FAILURE)
+ throw BackendError(ec.message());
+ else if (ec == shareddatalayer::Error::NOT_CONNECTED)
+ throw NotConnected(ec.message());
+ else if (ec == shareddatalayer::Error::OPERATION_INTERRUPTED)
+ throw OperationInterrupted(ec.message());
+ else if (ec == shareddatalayer::Error::REJECTED_BY_BACKEND)
+ throw RejectedByBackend(ec.message());
+ else if (ec == AsyncRedisStorage::ErrorCode::INVALID_NAMESPACE)
+ throw InvalidNamespace(ec.message());
+ else if (ec == shareddatalayer::Error::REJECTED_BY_SDL)
+ throw RejectedBySdl(ec.message());
+
+ std::ostringstream os;
+ os << "No corresponding SDL exception found for error code: " << ec.category().name() << " " << ec.value();
+ throw std::range_error(os.str());
+ }
+}
+
+SyncStorageImpl::SyncStorageImpl(std::unique_ptr<AsyncStorage> asyncStorage):
+ SyncStorageImpl(std::move(asyncStorage), System::getSystem())
+{
+}
+
+SyncStorageImpl::SyncStorageImpl(std::unique_ptr<AsyncStorage> pAsyncStorage,
+ System& system):
+ asyncStorage(std::move(pAsyncStorage)),
+ system(system),
+ pFd(asyncStorage->fd()),
+ localStatus(false),
+ synced(false)
+{
+}
+
+void SyncStorageImpl::modifyAck(const std::error_code& error)
+{
+ synced = true;
+ localError = error;
+}
+
+void SyncStorageImpl::modifyIfAck(const std::error_code& error, bool status)
+{
+ synced = true;
+ localError = error;
+ localStatus = status;
+}
+
+void SyncStorageImpl::getAck(const std::error_code& error, const DataMap& dataMap)
+{
+ synced = true;
+ localError = error;
+ localMap = dataMap;
+}
+
+void SyncStorageImpl::findKeysAck(const std::error_code& error, const Keys& keys)
+{
+ synced = true;
+ localError = error;
+ localKeys = keys;
+}
+
+void SyncStorageImpl::verifyBackendResponse()
+{
+ if(localError)
+ throwExceptionForErrorCode(localError);
+}
+
+void SyncStorageImpl::waitForCallback()
+{
+ struct pollfd events { pFd, POLLIN, 0 };
+ while(!synced)
+ if (system.poll(&events, 1, -1) > 0 && (events.revents & POLLIN))
+ asyncStorage->handleEvents();
+}
+
+void SyncStorageImpl::waitSdlToBeReady(const Namespace& ns)
+{
+ synced = false;
+ asyncStorage->waitReadyAsync(ns,
+ std::bind(&shareddatalayer::SyncStorageImpl::modifyAck,
+ this,
+ std::error_code()));
+ waitForCallback();
+ verifyBackendResponse();
+}
+
+void SyncStorageImpl::set(const Namespace& ns, const DataMap& dataMap)
+{
+ waitSdlToBeReady(ns);
+ synced = false;
+ asyncStorage->setAsync(ns,
+ dataMap,
+ std::bind(&shareddatalayer::SyncStorageImpl::modifyAck,
+ this,
+ std::placeholders::_1));
+ waitForCallback();
+ verifyBackendResponse();
+}
+
+bool SyncStorageImpl::setIf(const Namespace& ns, const Key& key, const Data& oldData, const Data& newData)
+{
+ waitSdlToBeReady(ns);
+ synced = false;
+ asyncStorage->setIfAsync(ns,
+ key,
+ oldData,
+ newData,
+ std::bind(&shareddatalayer::SyncStorageImpl::modifyIfAck,
+ this,
+ std::placeholders::_1,
+ std::placeholders::_2));
+ waitForCallback();
+ verifyBackendResponse();
+ return localStatus;
+}
+
+bool SyncStorageImpl::setIfNotExists(const Namespace& ns, const Key& key, const Data& data)
+{
+ waitSdlToBeReady(ns);
+ synced = false;
+ asyncStorage->setIfNotExistsAsync(ns,
+ key,
+ data,
+ std::bind(&shareddatalayer::SyncStorageImpl::modifyIfAck,
+ this,
+ std::placeholders::_1,
+ std::placeholders::_2));
+ waitForCallback();
+ verifyBackendResponse();
+ return localStatus;
+}
+
+SyncStorageImpl::DataMap SyncStorageImpl::get(const Namespace& ns, const Keys& keys)
+{
+ waitSdlToBeReady(ns);
+ synced = false;
+ asyncStorage->getAsync(ns,
+ keys,
+ std::bind(&shareddatalayer::SyncStorageImpl::getAck,
+ this,
+ std::placeholders::_1,
+ std::placeholders::_2));
+ waitForCallback();
+ verifyBackendResponse();
+ return localMap;
+}
+
+void SyncStorageImpl::remove(const Namespace& ns, const Keys& keys)
+{
+ waitSdlToBeReady(ns);
+ synced = false;
+ asyncStorage->removeAsync(ns,
+ keys,
+ std::bind(&shareddatalayer::SyncStorageImpl::modifyAck,
+ this,
+ std::placeholders::_1));
+ waitForCallback();
+ verifyBackendResponse();
+}
+
+bool SyncStorageImpl::removeIf(const Namespace& ns, const Key& key, const Data& data)
+{
+ waitSdlToBeReady(ns);
+ synced = false;
+ asyncStorage->removeIfAsync(ns,
+ key,
+ data,
+ std::bind(&shareddatalayer::SyncStorageImpl::modifyIfAck,
+ this,
+ std::placeholders::_1,
+ std::placeholders::_2));
+ waitForCallback();
+ verifyBackendResponse();
+ return localStatus;
+}
+
+SyncStorageImpl::Keys SyncStorageImpl::findKeys(const Namespace& ns, const std::string& keyPrefix)
+{
+ waitSdlToBeReady(ns);
+ synced = false;
+ asyncStorage->findKeysAsync(ns,
+ keyPrefix,
+ std::bind(&shareddatalayer::SyncStorageImpl::findKeysAck,
+ this,
+ std::placeholders::_1,
+ std::placeholders::_2));
+ waitForCallback();
+ verifyBackendResponse();
+ return localKeys;
+}
+
+void SyncStorageImpl::removeAll(const Namespace& ns)
+{
+ waitSdlToBeReady(ns);
+ synced = false;
+ asyncStorage->removeAllAsync(ns,
+ std::bind(&shareddatalayer::SyncStorageImpl::modifyAck,
+ this,
+ std::placeholders::_1));
+ waitForCallback();
+ verifyBackendResponse();
+}
diff --git a/src/system.cpp b/src/system.cpp
new file mode 100644
index 0000000..e0852e9
--- /dev/null
+++ b/src/system.cpp
@@ -0,0 +1,129 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/system.hpp"
+#include <system_error>
+#include <cerrno>
+#include <cstring>
+#include <sstream>
+#include <sys/epoll.h>
+#include <sys/timerfd.h>
+#include <sys/eventfd.h>
+#include "private/abort.hpp"
+#include "private/createlogger.hpp"
+
+using namespace shareddatalayer;
+
+int System::poll(struct pollfd *fds, nfds_t nfds, int timeout)
+{
+ const int ret(::poll(fds, nfds, timeout));
+ if (ret == -1 && errno != EINTR)
+ throw std::system_error(errno, std::system_category(), "poll");
+ return ret;
+}
+
+int System::epoll_create1(int flags)
+{
+ const int ret(::epoll_create1(flags));
+ if (ret == -1)
+ throw std::system_error(errno, std::system_category(), "epoll_create1");
+ return ret;
+}
+
+void System::epoll_ctl(int epfd, int op, int fd, epoll_event* event)
+{
+ const int ret(::epoll_ctl(epfd, op, fd, event));
+ if (ret == -1)
+ throw std::system_error(errno, std::system_category(), "epoll_ctl");
+}
+
+int System::epoll_wait(int epfd, epoll_event* events, int maxevents, int timeout)
+{
+ const int ret(::epoll_wait(epfd, events, maxevents, timeout));
+ if ((ret == -1) && (errno != EINTR))
+ throw std::system_error(errno, std::system_category(), "epoll_wait");
+ return ret;
+}
+
+int System::timerfd_create(int clockid, int flags)
+{
+ const int ret(::timerfd_create(clockid, flags));
+ if (ret == -1)
+ throw std::system_error(errno, std::system_category(), "timerfd_create");
+ return ret;
+}
+
+void System::timerfd_settime(int fd, int flags, const itimerspec* new_value, itimerspec* old_value)
+{
+ const int ret(::timerfd_settime(fd, flags, new_value, old_value));
+ if (ret == -1)
+ throw std::system_error(errno, std::system_category(), "timerfd_settime");
+}
+
+ssize_t System::read(int fd, void* buf, size_t count)
+{
+ const ssize_t ret(::read(fd, buf, count));
+ if ((ret == -1) && (errno != EINTR) && (errno != EAGAIN))
+ throw std::system_error(errno, std::system_category(), "read");
+ return ret;
+}
+
+int System::eventfd(unsigned int initval, int flags)
+{
+ const int ret(::eventfd(initval, flags));
+ if (ret == -1)
+ throw std::system_error(errno, std::system_category(), "eventfd");
+ return ret;
+}
+
+ssize_t System::write(int fd, const void* buf, size_t count)
+{
+ ssize_t ret;
+ do
+ ret = ::write(fd, buf, count);
+ while ((ret == -1) && (errno == EINTR));
+ if (ret == -1)
+ throw std::system_error(errno, std::system_category(), "write");
+ return ret;
+}
+
+std::chrono::steady_clock::duration System::time_since_epoch()
+{
+ return std::chrono::steady_clock::now().time_since_epoch();
+}
+
+void System::close(int fd)
+{
+ if (::close(fd) == -1)
+ {
+ int errno_saved = errno;
+ std::ostringstream msg;
+ msg << "close(" << fd << ") failed: " << strerror(errno_saved);
+ logErrorOnce(msg.str());
+ SHAREDDATALAYER_ABORT("close failed");
+ }
+}
+
+const char* System::getenv(const char* name)
+{
+ return ::getenv(name);
+}
+
+System& System::getSystem() noexcept
+{
+ static System system;
+ return system;
+}
diff --git a/src/systemlogger.cpp b/src/systemlogger.cpp
new file mode 100644
index 0000000..8b8336b
--- /dev/null
+++ b/src/systemlogger.cpp
@@ -0,0 +1,114 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/systemlogger.hpp"
+#include <ostream>
+#include <sstream>
+#include <syslog.h>
+#include <boost/iostreams/stream.hpp>
+#include <boost/iostreams/concepts.hpp>
+
+using namespace shareddatalayer;
+
+namespace
+{
+ class Sink: public boost::iostreams::sink
+ {
+ public:
+ Sink(const std::string& prefix, int level): prefix(prefix), level(level) { }
+
+ ~Sink() { }
+
+ std::streamsize write(const char* s, std::streamsize n);
+
+ private:
+ const std::string prefix;
+ const int level;
+ };
+}
+
+std::streamsize Sink::write(const char* s, std::streamsize n)
+{
+ std::ostringstream os;
+ os << "%s: %." << n << 's';
+ syslog(level, os.str().c_str(), prefix.c_str(), s);
+ return n;
+}
+
+SystemLogger::SystemLogger(const std::string& prefix):
+ prefix(prefix)
+{
+}
+
+SystemLogger::~SystemLogger()
+{
+}
+
+std::ostream& SystemLogger::emerg()
+{
+ if (osEmerg == nullptr)
+ osEmerg.reset(new boost::iostreams::stream<Sink>(Sink(prefix, LOG_EMERG)));
+ return *osEmerg;
+}
+
+std::ostream& SystemLogger::alert()
+{
+ if (osAlert == nullptr)
+ osAlert.reset(new boost::iostreams::stream<Sink>(Sink(prefix, LOG_ALERT)));
+ return *osAlert;
+}
+
+std::ostream& SystemLogger::crit()
+{
+ if (osCrit == nullptr)
+ osCrit.reset(new boost::iostreams::stream<Sink>(Sink(prefix, LOG_CRIT)));
+ return *osCrit;
+}
+
+std::ostream& SystemLogger::error()
+{
+ if (osError == nullptr)
+ osError.reset(new boost::iostreams::stream<Sink>(Sink(prefix, LOG_ERR)));
+ return *osError;
+}
+
+std::ostream& SystemLogger::warning()
+{
+ if (osWarning == nullptr)
+ osWarning.reset(new boost::iostreams::stream<Sink>(Sink(prefix, LOG_WARNING)));
+ return *osWarning;
+}
+
+std::ostream& SystemLogger::notice()
+{
+ if (osNotice == nullptr)
+ osNotice.reset(new boost::iostreams::stream<Sink>(Sink(prefix, LOG_NOTICE)));
+ return *osNotice;
+}
+
+std::ostream& SystemLogger::info()
+{
+ if (osInfo == nullptr)
+ osInfo.reset(new boost::iostreams::stream<Sink>(Sink(prefix, LOG_INFO)));
+ return *osInfo;
+}
+
+std::ostream& SystemLogger::debug()
+{
+ if (osDebug == nullptr)
+ osDebug.reset(new boost::iostreams::stream<Sink>(Sink(prefix, LOG_DEBUG)));
+ return *osDebug;
+}
diff --git a/src/timer.cpp b/src/timer.cpp
new file mode 100644
index 0000000..7b4d41a
--- /dev/null
+++ b/src/timer.cpp
@@ -0,0 +1,51 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/timer.hpp"
+#include "private/engine.hpp"
+#include "private/abort.hpp"
+
+using namespace shareddatalayer;
+
+Timer::Timer(Engine& engine):
+ engine(engine),
+ armed(false)
+{
+}
+
+Timer::~Timer()
+{
+ disarm();
+}
+
+void Timer::arm(const Duration& duration, const Callback& cb)
+{
+ if (!cb)
+ SHAREDDATALAYER_ABORT("Timer::arm: a null callback");
+
+ disarm();
+ engine.armTimer(*this, duration, [this, cb] () { armed = false; cb(); });
+ armed = true;
+}
+
+void Timer::disarm()
+{
+ if (!armed)
+ return;
+
+ engine.disarmTimer(*this);
+ armed = false;
+}
diff --git a/src/timerfd.cpp b/src/timerfd.cpp
new file mode 100644
index 0000000..1356c50
--- /dev/null
+++ b/src/timerfd.cpp
@@ -0,0 +1,126 @@
+/*
+ Copyright (c) 2018-2019 Nokia.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+#include "private/timerfd.hpp"
+#include "private/engine.hpp"
+#include "private/system.hpp"
+
+using namespace shareddatalayer;
+
+TimerFD::TimerFD(Engine& engine):
+ TimerFD(System::getSystem(), engine)
+{
+}
+
+TimerFD::TimerFD(System& system, Engine& engine):
+ system(system),
+ fd(system, system.timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC))
+{
+ engine.addMonitoredFD(fd, Engine::EVENT_IN, std::bind(&TimerFD::handleEvents, this));
+}
+
+TimerFD::~TimerFD()
+{
+}
+
+void TimerFD::arm(Timer& timer, const Timer::Duration& duration, const Timer::Callback& cb)
+{
+ const auto absolute(toAbsolute(duration));
+ const auto i(queue.insert(std::make_pair(absolute, std::make_pair(&timer, cb))));
+ timer.iterator = i;
+ if (isFirst(i))
+ armTimerFD(absolute);
+}
+
+bool TimerFD::isFirst(Queue::iterator it) const
+{
+ return queue.begin() == it;
+}
+
+void TimerFD::disarm(const Timer& timer)
+{
+ const bool wasFirst(isFirst(timer.iterator));
+ queue.erase(timer.iterator);
+
+ if (queue.empty())
+ disarmTimerFD();
+ else if (wasFirst)
+ armTimerFD(nextTrigger());
+}
+
+Timer::Duration TimerFD::toAbsolute(const Timer::Duration& duration)
+{
+ return std::chrono::duration_cast<Timer::Duration>(system.time_since_epoch()) + duration;
+}
+
+void TimerFD::handleEvents()
+{
+ if (timerExpired())
+ handleExpiredTimers();
+}
+
+bool TimerFD::timerExpired() const
+{
+ uint64_t count;
+ return (system.read(fd, &count, sizeof(count)) == sizeof(count)) && (count > 0U);
+}
+
+void TimerFD::handleExpiredTimers()
+{
+ const auto now(system.time_since_epoch());
+ do
+ {
+ popAndExecuteFirstTimer();
+ if (queue.empty())
+ {
+ disarmTimerFD();
+ return;
+ }
+ }
+ while (queue.begin()->first <= now);
+ armTimerFD(nextTrigger());
+}
+
+void TimerFD::popAndExecuteFirstTimer()
+{
+ const auto i(queue.begin());
+ const auto cb(i->second.second);
+ queue.erase(i);
+ cb();
+}
+
+Timer::Duration TimerFD::nextTrigger() const
+{
+ return queue.begin()->first;
+}
+
+void TimerFD::disarmTimerFD()
+{
+ setTimeForTimerFD(0, 0);
+}
+
+void TimerFD::armTimerFD(const Timer::Duration& duration)
+{
+ static const long int NANOSECONDS_IN_ONE_SECOND(1E9);
+ setTimeForTimerFD(std::chrono::duration_cast<std::chrono::seconds>(duration).count(),
+ std::chrono::duration_cast<std::chrono::nanoseconds>(duration).count() % NANOSECONDS_IN_ONE_SECOND);
+}
+
+void TimerFD::setTimeForTimerFD(time_t seconds, long int nanoseconds)
+{
+ const itimerspec ts{ { 0, 0 }, { seconds, nanoseconds } };
+ system.timerfd_settime(fd, TFD_TIMER_ABSTIME, &ts, nullptr);
+}
diff --git a/src/tools/collectsdlinfo.in b/src/tools/collectsdlinfo.in
new file mode 100644
index 0000000..583d8b9
--- /dev/null
+++ b/src/tools/collectsdlinfo.in
@@ -0,0 +1,26 @@
+#!/bin/sh
+
+if test "${#}" != 1 ; then
+ echo "Usage: collectsdlinfo OUTPUTDIR" 1>&2
+ exit 1
+fi
+if test "${1}" = "-h" || test "${1}" = "--help" ; then
+ cat << EOF
+Usage: collectsdlinfo OUTPUTDIR
+
+collectsdlinfo finds SDL configuration files and collects debug info.
+The results are saved into the given OUTPUTDIR directory.
+EOF
+ exit 0
+fi
+
+OUTPUTDIR="${1}"
+mkdir -p "${OUTPUTDIR}" || exit 1
+outputfile="${OUTPUTDIR}/shareddatalayer_configuration.txt"
+sdltool dump-configuration > "${outputfile}" 2>&1
+
+outputfile="${OUTPUTDIR}/shareddatalayer_write_read_latency.txt"
+sdltool test-get-set -- timeout 10 > "${outputfile}" 2>&1
+
+outputfile="${OUTPUTDIR}/shareddatalayer_backend_connectivity.txt"
+sdltool test-connectivity -- timeout 10 > "${outputfile}" 2>&1