复现已有算法
This commit is contained in:
cloud
@@ -0,0 +1,378 @@
|
||||
/**
|
||||
*
|
||||
* @file RedisClient.h
|
||||
* @author An Tao
|
||||
*
|
||||
* Copyright 2018, An Tao. All rights reserved.
|
||||
* https://github.com/an-tao/drogon
|
||||
* Use of this source code is governed by a MIT license
|
||||
* that can be found in the License file.
|
||||
*
|
||||
* Drogon
|
||||
*
|
||||
*/
|
||||
#pragma once
|
||||
|
||||
#include <drogon/exports.h>
|
||||
#include <drogon/nosql/RedisResult.h>
|
||||
#include <drogon/nosql/RedisException.h>
|
||||
#include <drogon/nosql/RedisSubscriber.h>
|
||||
#include <string_view>
|
||||
#include <trantor/net/InetAddress.h>
|
||||
#include <trantor/utils/Logger.h>
|
||||
#include <memory>
|
||||
#include <functional>
|
||||
#include <future>
|
||||
#ifdef __cpp_impl_coroutine
|
||||
#include <drogon/utils/coroutine.h>
|
||||
#endif
|
||||
|
||||
namespace drogon
|
||||
{
|
||||
namespace nosql
|
||||
{
|
||||
#ifdef __cpp_impl_coroutine
|
||||
class RedisClient;
|
||||
class RedisTransaction;
|
||||
|
||||
namespace internal
|
||||
{
|
||||
struct [[nodiscard]] RedisAwaiter : public CallbackAwaiter<RedisResult>
|
||||
{
|
||||
using RedisFunction =
|
||||
std::function<void(RedisResultCallback &&, RedisExceptionCallback &&)>;
|
||||
|
||||
explicit RedisAwaiter(RedisFunction &&function)
|
||||
: function_(std::move(function))
|
||||
{
|
||||
}
|
||||
|
||||
void await_suspend(std::coroutine_handle<> handle)
|
||||
{
|
||||
function_(
|
||||
[handle, this](const RedisResult &result) {
|
||||
this->setValue(result);
|
||||
handle.resume();
|
||||
},
|
||||
[handle, this](const RedisException &e) {
|
||||
LOG_ERROR << e.what();
|
||||
this->setException(std::make_exception_ptr(e));
|
||||
handle.resume();
|
||||
});
|
||||
}
|
||||
|
||||
private:
|
||||
RedisFunction function_;
|
||||
};
|
||||
|
||||
struct [[nodiscard]] RedisTransactionAwaiter
|
||||
: public CallbackAwaiter<std::shared_ptr<RedisTransaction>>
|
||||
{
|
||||
RedisTransactionAwaiter(RedisClient *client) : client_(client)
|
||||
{
|
||||
}
|
||||
|
||||
void await_suspend(std::coroutine_handle<> handle);
|
||||
|
||||
private:
|
||||
RedisClient *client_;
|
||||
};
|
||||
} // namespace internal
|
||||
#endif
|
||||
|
||||
class RedisTransaction;
|
||||
|
||||
/**
|
||||
* @brief This class represents a redis client that contains several connections
|
||||
* to a redis server.
|
||||
*
|
||||
*/
|
||||
class DROGON_EXPORT RedisClient
|
||||
{
|
||||
public:
|
||||
/**
|
||||
* @brief Create a new redis client with multiple connections;
|
||||
*
|
||||
* @param serverAddress The server address.
|
||||
* @param numberOfConnections The number of connections. 1 by default.
|
||||
* @param username The username to authenticate if necessary.
|
||||
* @param password The password to authenticate if necessary.
|
||||
* @return std::shared_ptr<RedisClient>
|
||||
*/
|
||||
static std::shared_ptr<RedisClient> newRedisClient(
|
||||
const trantor::InetAddress &serverAddress,
|
||||
size_t numberOfConnections = 1,
|
||||
const std::string &password = "",
|
||||
unsigned int db = 0,
|
||||
const std::string &username = "");
|
||||
/**
|
||||
* @brief Execute a redis command
|
||||
*
|
||||
* @param resultCallback The callback is called when a redis reply is
|
||||
* received successfully.
|
||||
* @param exceptionCallback The callback is called when an error occurs.
|
||||
* @note When a redis reply with REDIS_REPLY_ERROR code is received, this
|
||||
* callback is called.
|
||||
* @param command The command to be executed. the command string can contain
|
||||
* some placeholders for parameters, such as '%s', '%d', etc.
|
||||
* @param ... The command parameters.
|
||||
* For example:
|
||||
* @code
|
||||
redisClientPtr->execCommandAsync([](const RedisResult &r){
|
||||
std::cout << r.getStringForDisplaying() << std::endl;
|
||||
},[](const std::exception &err){
|
||||
std::cerr << err.what() << std::endl;
|
||||
}, "get %s", key.data());
|
||||
@endcode
|
||||
*/
|
||||
virtual void execCommandAsync(RedisResultCallback &&resultCallback,
|
||||
RedisExceptionCallback &&exceptionCallback,
|
||||
std::string_view command,
|
||||
...) noexcept = 0;
|
||||
|
||||
/**
|
||||
* @brief Execute a redis command synchronously
|
||||
*
|
||||
* @param processFunc Function to extract data from redis result.
|
||||
* received successfully.
|
||||
* @param command The command to be executed. the command string can contain
|
||||
* some placeholders for parameters, such as '%s', '%d', etc.
|
||||
* @param ... The command parameters.
|
||||
* @return Returns the same value as process callback.
|
||||
* For example:
|
||||
* @code
|
||||
try
|
||||
{
|
||||
std::string res = redisClientPtr->execCommandSync<std::string>(
|
||||
[](const RedisResult &r){
|
||||
return r.asString();
|
||||
},
|
||||
"get %s",
|
||||
key.data()
|
||||
);
|
||||
}
|
||||
catch (const RedisException & err)
|
||||
{
|
||||
}
|
||||
catch (const std::exception & err)
|
||||
{
|
||||
}
|
||||
@endcode
|
||||
*/
|
||||
template <typename T, typename... Args>
|
||||
T execCommandSync(std::function<T(const RedisResult &)> &&processFunc,
|
||||
std::string_view command,
|
||||
Args &&...args)
|
||||
{
|
||||
return execCommandSync<std::decay_t<decltype(processFunc)>>(
|
||||
std::move(processFunc), command, std::forward<Args>(args)...);
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Execute a redis command synchronously
|
||||
* Return type can be deduced automatically in this version.
|
||||
*/
|
||||
template <typename F, typename... Args>
|
||||
std::invoke_result_t<F, const RedisResult &> execCommandSync(
|
||||
F &&processFunc,
|
||||
std::string_view command,
|
||||
Args &&...args)
|
||||
{
|
||||
using Ret = std::invoke_result_t<F, const RedisResult &>;
|
||||
std::promise<Ret> prom;
|
||||
execCommandAsync(
|
||||
[&processFunc, &prom](const RedisResult &result) {
|
||||
try
|
||||
{
|
||||
prom.set_value(processFunc(result));
|
||||
}
|
||||
catch (...)
|
||||
{
|
||||
prom.set_exception(std::current_exception());
|
||||
}
|
||||
},
|
||||
[&prom](const RedisException &err) {
|
||||
prom.set_exception(std::make_exception_ptr(err));
|
||||
},
|
||||
command,
|
||||
std::forward<Args>(args)...);
|
||||
|
||||
return prom.get_future().get();
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Create a subscriber for redis subscribe commands.
|
||||
*
|
||||
* @return std::shared_ptr<RedisSubscriber>
|
||||
* @note This subscriber creates a new redis connection dedicated to
|
||||
* subscribe commands. This connection is managed by RedisClient.
|
||||
*/
|
||||
virtual std::shared_ptr<RedisSubscriber> newSubscriber() noexcept = 0;
|
||||
|
||||
/**
|
||||
* @brief Create a redis transaction object.
|
||||
*
|
||||
* @return std::shared_ptr<RedisTransaction>
|
||||
* @note An exception with kTimeout code is thrown if the operation is
|
||||
* timed out. see RedisException.h
|
||||
*/
|
||||
virtual std::shared_ptr<RedisTransaction> newTransaction() noexcept(
|
||||
false) = 0;
|
||||
|
||||
/**
|
||||
* @brief Create a transaction object in asynchronous mode.
|
||||
*
|
||||
* @return std::shared_ptr<RedisTransaction>
|
||||
* @note An empty shared_ptr object is returned via the callback if the
|
||||
* operation is timed out.
|
||||
*/
|
||||
virtual void newTransactionAsync(
|
||||
const std::function<void(const std::shared_ptr<RedisTransaction> &)>
|
||||
&callback) = 0;
|
||||
/**
|
||||
* @brief Set the Timeout value of execution of a command.
|
||||
*
|
||||
* @param timeout in seconds, if the result is not returned from the
|
||||
* server within the timeout, a RedisException with "Command execution
|
||||
* timeout" string is generated and returned to the caller.
|
||||
* @note set the timeout value to zero or negative for no limit on time.
|
||||
* The default value is -1.0, this means there is no time limit if this
|
||||
* method is not called.
|
||||
*/
|
||||
virtual void setTimeout(double timeout) = 0;
|
||||
|
||||
virtual ~RedisClient() = default;
|
||||
|
||||
/**
|
||||
* @brief Close all connections in the client. usually used by Drogon in the
|
||||
* quit() method.
|
||||
* */
|
||||
virtual void closeAll() = 0;
|
||||
|
||||
#ifdef __cpp_impl_coroutine
|
||||
/**
|
||||
* @brief Send a Redis command and await the RedisResult in a coroutine.
|
||||
*
|
||||
* @tparam Arguments
|
||||
* @param command
|
||||
* @param args
|
||||
* @return internal::RedisAwaiter that can be awaited in a coroutine.
|
||||
* For example:
|
||||
* @code
|
||||
try
|
||||
{
|
||||
auto result = co_await redisClient->execCommandCoro("get %s",
|
||||
"keyname");
|
||||
std::cout << result.getStringForDisplaying() << "\n";
|
||||
}
|
||||
catch(const RedisException &err)
|
||||
{
|
||||
std::cout << err.what() << "\n";
|
||||
}
|
||||
@endcode
|
||||
*/
|
||||
template <typename... Arguments>
|
||||
internal::RedisAwaiter execCommandCoro(std::string_view command,
|
||||
Arguments... args)
|
||||
{
|
||||
return internal::RedisAwaiter(
|
||||
[command,
|
||||
this,
|
||||
args...](RedisResultCallback &&commandCallback,
|
||||
RedisExceptionCallback &&exceptionCallback) {
|
||||
execCommandAsync(std::move(commandCallback),
|
||||
std::move(exceptionCallback),
|
||||
command,
|
||||
args...);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief await a RedisTransactionPtr in a coroutine.
|
||||
*
|
||||
* @return internal::RedisTransactionAwaiter that can be awaited in a
|
||||
* coroutine.
|
||||
* For example:
|
||||
* @code
|
||||
try
|
||||
{
|
||||
auto transPtr = co_await redisClient->newTransactionCoro();
|
||||
...
|
||||
}
|
||||
catch(const RedisException &err)
|
||||
{
|
||||
std::cout << err.what() << "\n";
|
||||
}
|
||||
@endcode
|
||||
*/
|
||||
internal::RedisTransactionAwaiter newTransactionCoro()
|
||||
{
|
||||
return internal::RedisTransactionAwaiter(this);
|
||||
}
|
||||
#endif
|
||||
};
|
||||
|
||||
class DROGON_EXPORT RedisTransaction : public RedisClient
|
||||
{
|
||||
public:
|
||||
// virtual void cancel() = 0;
|
||||
virtual void execute(RedisResultCallback &&resultCallback,
|
||||
RedisExceptionCallback &&exceptionCallback) = 0;
|
||||
#ifdef __cpp_impl_coroutine
|
||||
/**
|
||||
* @brief Send a "exec" command to execute the transaction and await a
|
||||
* RedisResult in a coroutine.
|
||||
*
|
||||
* @return internal::RedisAwaiter that can be awaited in a coroutine.
|
||||
* For example:
|
||||
* @code
|
||||
try
|
||||
{
|
||||
auto transPtr = co_await redisClient->newTransactionCoro();
|
||||
...
|
||||
auto result = co_await transPtr->executeCoro();
|
||||
std::cout << result.getStringForDisplaying() << "\n";
|
||||
}
|
||||
catch(const RedisException &err)
|
||||
{
|
||||
std::cout << err.what() << "\n";
|
||||
}
|
||||
@endcode
|
||||
*/
|
||||
internal::RedisAwaiter executeCoro()
|
||||
{
|
||||
return internal::RedisAwaiter(
|
||||
[this](RedisResultCallback &&resultCallback,
|
||||
RedisExceptionCallback &&exceptionCallback) {
|
||||
execute(std::move(resultCallback),
|
||||
std::move(exceptionCallback));
|
||||
});
|
||||
}
|
||||
#endif
|
||||
void closeAll() override
|
||||
{
|
||||
}
|
||||
};
|
||||
|
||||
using RedisClientPtr = std::shared_ptr<RedisClient>;
|
||||
using RedisTransactionPtr = std::shared_ptr<RedisTransaction>;
|
||||
|
||||
#ifdef __cpp_impl_coroutine
|
||||
inline void internal::RedisTransactionAwaiter::await_suspend(
|
||||
std::coroutine_handle<> handle)
|
||||
{
|
||||
assert(client_ != nullptr);
|
||||
client_->newTransactionAsync(
|
||||
[this, handle](const std::shared_ptr<RedisTransaction> &transaction) {
|
||||
if (transaction == nullptr)
|
||||
setException(std::make_exception_ptr(RedisException(
|
||||
RedisErrorCode::kTimeout,
|
||||
"Timeout, no connection available for transaction")));
|
||||
else
|
||||
setValue(transaction);
|
||||
handle.resume();
|
||||
});
|
||||
}
|
||||
#endif
|
||||
} // namespace nosql
|
||||
} // namespace drogon
|
||||
@@ -0,0 +1,69 @@
|
||||
/**
|
||||
*
|
||||
* @file RedisException.h
|
||||
* @author An Tao
|
||||
*
|
||||
* Copyright 2018, An Tao. All rights reserved.
|
||||
* https://github.com/an-tao/drogon
|
||||
* Use of this source code is governed by a MIT license
|
||||
* that can be found in the License file.
|
||||
*
|
||||
* Drogon
|
||||
*
|
||||
*/
|
||||
#pragma once
|
||||
|
||||
#include <exception>
|
||||
#include <functional>
|
||||
#include <string>
|
||||
|
||||
namespace drogon
|
||||
{
|
||||
namespace nosql
|
||||
{
|
||||
enum class RedisErrorCode
|
||||
{
|
||||
kNone = 0,
|
||||
kUnknown,
|
||||
kConnectionBroken,
|
||||
kNoConnectionAvailable,
|
||||
kRedisError,
|
||||
kInternalError,
|
||||
kTransactionCancelled,
|
||||
kBadType,
|
||||
kTimeout
|
||||
};
|
||||
|
||||
class RedisException final : public std::exception
|
||||
{
|
||||
public:
|
||||
const char *what() const noexcept override
|
||||
{
|
||||
return message_.data();
|
||||
}
|
||||
|
||||
RedisErrorCode code() const
|
||||
{
|
||||
return code_;
|
||||
}
|
||||
|
||||
RedisException(RedisErrorCode code, const std::string &message)
|
||||
: message_(message), code_(code)
|
||||
{
|
||||
}
|
||||
|
||||
RedisException(RedisErrorCode code, std::string &&message)
|
||||
: message_(std::move(message)), code_(code)
|
||||
{
|
||||
}
|
||||
|
||||
RedisException() = delete;
|
||||
|
||||
private:
|
||||
std::string message_;
|
||||
RedisErrorCode code_{RedisErrorCode::kNone};
|
||||
};
|
||||
|
||||
using RedisExceptionCallback = std::function<void(const RedisException &)>;
|
||||
} // namespace nosql
|
||||
} // namespace drogon
|
||||
@@ -0,0 +1,132 @@
|
||||
/**
|
||||
*
|
||||
* @file RedisResult.h
|
||||
* @author An Tao
|
||||
*
|
||||
* Copyright 2018, An Tao. All rights reserved.
|
||||
* https://github.com/an-tao/drogon
|
||||
* Use of this source code is governed by a MIT license
|
||||
* that can be found in the License file.
|
||||
*
|
||||
* Drogon
|
||||
*
|
||||
*/
|
||||
|
||||
#pragma once
|
||||
|
||||
#include <drogon/exports.h>
|
||||
#include <vector>
|
||||
#include <string>
|
||||
#include <memory>
|
||||
#include <functional>
|
||||
|
||||
struct redisReply;
|
||||
|
||||
namespace drogon
|
||||
{
|
||||
namespace nosql
|
||||
{
|
||||
enum class RedisResultType
|
||||
{
|
||||
kInteger = 0,
|
||||
kString,
|
||||
kArray,
|
||||
kStatus,
|
||||
kNil,
|
||||
kError
|
||||
};
|
||||
|
||||
/**
|
||||
* @brief This class represents a redis reply with no error.
|
||||
* @note Limited by the hiredis library, the RedisResult object is only
|
||||
* available in the context of the result callback, one can't hold or copy or
|
||||
* move a RedisResult object for later use after the callback is returned.
|
||||
*/
|
||||
class DROGON_EXPORT RedisResult
|
||||
{
|
||||
public:
|
||||
explicit RedisResult(redisReply *result) : result_(result)
|
||||
{
|
||||
}
|
||||
|
||||
~RedisResult() = default;
|
||||
|
||||
/**
|
||||
* @brief Return the type of the result_
|
||||
* @return RedisResultType
|
||||
* @note The kError type is never obtained here.
|
||||
*/
|
||||
RedisResultType type() const noexcept;
|
||||
|
||||
/**
|
||||
* @brief Get the string value of the result.
|
||||
*
|
||||
* @return std::string
|
||||
* @note Calling the method of a result object which is the kArray type
|
||||
* throws a runtime exception.
|
||||
*/
|
||||
std::string asString() const noexcept(false);
|
||||
|
||||
/**
|
||||
* @brief Get the array value of the result.
|
||||
*
|
||||
* @return std::vector<RedisResult>
|
||||
* @note Calling the method of a result object whose type is not kArray type
|
||||
* throws a runtime exception.
|
||||
*/
|
||||
std::vector<RedisResult> asArray() const noexcept(false);
|
||||
|
||||
/**
|
||||
* @brief Get the integer value of the result.
|
||||
*
|
||||
* @return long long
|
||||
* @note Calling the method of a result object whose type is not kInteger
|
||||
* type throws a runtime exception.
|
||||
*/
|
||||
long long asInteger() const noexcept(false);
|
||||
|
||||
/**
|
||||
* @brief Get the string for displaying the result.
|
||||
*
|
||||
* @return std::string
|
||||
*/
|
||||
std::string getStringForDisplaying() const noexcept;
|
||||
|
||||
/**
|
||||
* @brief Get the string for displaying with indent.
|
||||
*
|
||||
* @param indent The indent value.
|
||||
* @return std::string
|
||||
*/
|
||||
std::string getStringForDisplayingWithIndent(
|
||||
size_t indent = 0) const noexcept;
|
||||
|
||||
/**
|
||||
* @brief return true if the result object is nil.
|
||||
*
|
||||
* @return true
|
||||
* @return false
|
||||
*/
|
||||
bool isNil() const noexcept;
|
||||
|
||||
/**
|
||||
* @brief Check if the result object is not nil.
|
||||
*
|
||||
* @return true
|
||||
* @return false
|
||||
*/
|
||||
explicit operator bool() const
|
||||
{
|
||||
return !isNil();
|
||||
}
|
||||
|
||||
private:
|
||||
redisReply *result_;
|
||||
};
|
||||
|
||||
using RedisResultCallback = std::function<void(const RedisResult &)>;
|
||||
using RedisMessageCallback =
|
||||
std::function<void(const std::string &channel, const std::string &message)>;
|
||||
|
||||
} // namespace nosql
|
||||
} // namespace drogon
|
||||
@@ -0,0 +1,61 @@
|
||||
/**
|
||||
*
|
||||
* @file RedisSubscriber.h
|
||||
* @author Nitromelon
|
||||
*
|
||||
* Copyright 2022, Nitromelon. All rights reserved.
|
||||
* https://github.com/drogonframework/drogon
|
||||
* Use of this source code is governed by a MIT license
|
||||
* that can be found in the License file.
|
||||
*
|
||||
* Drogon
|
||||
*
|
||||
*/
|
||||
#pragma once
|
||||
|
||||
#include <string>
|
||||
#include <drogon/nosql/RedisResult.h>
|
||||
|
||||
namespace drogon::nosql
|
||||
{
|
||||
class RedisSubscriber
|
||||
{
|
||||
public:
|
||||
/**
|
||||
* @brief Subscribe to a channel. The subscriber will keep the channel
|
||||
* subscribed, until unsubscribe() is called on this channel, or the
|
||||
* subscriber or RedisClient who creates it no longer exists.
|
||||
* This method will not block.
|
||||
*
|
||||
* @param messageCallback The callback is called when a message is received
|
||||
* from the channel.
|
||||
* @param channel The channel to subscribe to.
|
||||
*
|
||||
* @note: Subscribing to same channel multiple times is supported. All
|
||||
* message callbacks will be called in subscription order.
|
||||
* One unsubscribe() call will remove all callbacks.
|
||||
*/
|
||||
virtual void subscribe(const std::string &channel,
|
||||
RedisMessageCallback &&messageCallback) noexcept = 0;
|
||||
|
||||
// Subscribe to channel pattern
|
||||
virtual void psubscribe(
|
||||
const std::string &pattern,
|
||||
RedisMessageCallback &&messageCallback) noexcept = 0;
|
||||
|
||||
/**
|
||||
* @brief Unsubscribe from a channel. Once this function returns, the
|
||||
* messageCallback registered through subscribe() will no longer be called.
|
||||
* This method will not block.
|
||||
*
|
||||
* @param channel The channel to subscribe to.
|
||||
*/
|
||||
virtual void unsubscribe(const std::string &channel) noexcept = 0;
|
||||
|
||||
// Unsubscribe from channel pattern
|
||||
virtual void punsubscribe(const std::string &pattern) noexcept = 0;
|
||||
|
||||
virtual ~RedisSubscriber() = default;
|
||||
};
|
||||
|
||||
} // namespace drogon::nosql
|
||||
@@ -0,0 +1,490 @@
|
||||
/**
|
||||
*
|
||||
* @file RedisClientImpl.cc
|
||||
* @author An Tao
|
||||
*
|
||||
* Copyright 2018, An Tao. All rights reserved.
|
||||
* https://github.com/an-tao/drogon
|
||||
* Use of this source code is governed by a MIT license
|
||||
* that can be found in the License file.
|
||||
*
|
||||
* Drogon
|
||||
*
|
||||
*/
|
||||
|
||||
#include "RedisConnection.h"
|
||||
#include "RedisClientImpl.h"
|
||||
#include "RedisSubscriberImpl.h"
|
||||
#include "RedisTransactionImpl.h"
|
||||
#include "../../lib/src/TaskTimeoutFlag.h"
|
||||
|
||||
using namespace drogon::nosql;
|
||||
|
||||
std::shared_ptr<RedisClient> RedisClient::newRedisClient(
|
||||
const trantor::InetAddress &serverAddress,
|
||||
size_t connectionNumber,
|
||||
const std::string &password,
|
||||
unsigned int db,
|
||||
const std::string &username)
|
||||
{
|
||||
auto client = std::make_shared<RedisClientImpl>(
|
||||
serverAddress, connectionNumber, username, password, db);
|
||||
client->init();
|
||||
return client;
|
||||
}
|
||||
|
||||
RedisClientImpl::RedisClientImpl(const trantor::InetAddress &serverAddress,
|
||||
size_t numberOfConnections,
|
||||
std::string username,
|
||||
std::string password,
|
||||
unsigned int db)
|
||||
: loops_(numberOfConnections < std::thread::hardware_concurrency()
|
||||
? numberOfConnections
|
||||
: std::thread::hardware_concurrency(),
|
||||
"RedisLoop"),
|
||||
serverAddr_(serverAddress),
|
||||
username_(std::move(username)),
|
||||
password_(std::move(password)),
|
||||
db_(db),
|
||||
numberOfConnections_(numberOfConnections)
|
||||
{
|
||||
}
|
||||
|
||||
void RedisClientImpl::init()
|
||||
{
|
||||
loops_.start();
|
||||
for (size_t i = 0; i < numberOfConnections_; ++i)
|
||||
{
|
||||
auto loop = loops_.getNextLoop();
|
||||
loop->queueInLoop([this, loop]() {
|
||||
std::lock_guard<std::mutex> lock(connectionsMutex_);
|
||||
connections_.insert(newConnection(loop));
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
RedisConnectionPtr RedisClientImpl::newConnection(trantor::EventLoop *loop)
|
||||
{
|
||||
auto conn = std::make_shared<RedisConnection>(
|
||||
serverAddr_, username_, password_, db_, loop);
|
||||
std::weak_ptr<RedisClientImpl> thisWeakPtr = shared_from_this();
|
||||
conn->setConnectCallback([thisWeakPtr](RedisConnectionPtr &&conn) {
|
||||
auto thisPtr = thisWeakPtr.lock();
|
||||
if (thisPtr)
|
||||
{
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(thisPtr->connectionsMutex_);
|
||||
thisPtr->readyConnections_.push_back(conn);
|
||||
}
|
||||
thisPtr->handleNextTask(conn);
|
||||
}
|
||||
});
|
||||
conn->setDisconnectCallback([thisWeakPtr](RedisConnectionPtr &&conn) {
|
||||
// assert(status == REDIS_CONNECTED);
|
||||
auto thisPtr = thisWeakPtr.lock();
|
||||
if (thisPtr)
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(thisPtr->connectionsMutex_);
|
||||
thisPtr->connections_.erase(conn);
|
||||
for (auto iter = thisPtr->readyConnections_.begin();
|
||||
iter != thisPtr->readyConnections_.end();
|
||||
++iter)
|
||||
{
|
||||
if (*iter == conn)
|
||||
{
|
||||
thisPtr->readyConnections_.erase(iter);
|
||||
break;
|
||||
}
|
||||
}
|
||||
auto loop = trantor::EventLoop::getEventLoopOfCurrentThread();
|
||||
assert(loop);
|
||||
loop->runAfter(2.0, [thisPtr, loop, conn]() {
|
||||
std::lock_guard<std::mutex> lock(thisPtr->connectionsMutex_);
|
||||
thisPtr->connections_.insert(thisPtr->newConnection(loop));
|
||||
});
|
||||
}
|
||||
});
|
||||
conn->setIdleCallback([thisWeakPtr](const RedisConnectionPtr &connPtr) {
|
||||
auto thisPtr = thisWeakPtr.lock();
|
||||
if (thisPtr)
|
||||
{
|
||||
thisPtr->handleNextTask(connPtr);
|
||||
}
|
||||
});
|
||||
return conn;
|
||||
}
|
||||
|
||||
RedisConnectionPtr RedisClientImpl::newSubscribeConnection(
|
||||
trantor::EventLoop *loop,
|
||||
const std::shared_ptr<RedisSubscriberImpl> &subscriber)
|
||||
{
|
||||
auto conn = std::make_shared<RedisConnection>(
|
||||
serverAddr_, username_, password_, db_, loop);
|
||||
std::weak_ptr<RedisClientImpl> weakThis = shared_from_this();
|
||||
std::weak_ptr<RedisSubscriberImpl> weakSub(subscriber);
|
||||
conn->setConnectCallback([weakThis, weakSub](RedisConnectionPtr &&conn) {
|
||||
auto thisPtr = weakThis.lock();
|
||||
if (!thisPtr)
|
||||
return;
|
||||
auto subPtr = weakSub.lock();
|
||||
|
||||
std::lock_guard<std::mutex> lock(thisPtr->connectionsMutex_);
|
||||
if (subPtr)
|
||||
{
|
||||
subPtr->setConnection(conn);
|
||||
subPtr->subscribeAll();
|
||||
}
|
||||
else
|
||||
{
|
||||
thisPtr->connections_.erase(conn);
|
||||
}
|
||||
});
|
||||
conn->setDisconnectCallback([weakThis, weakSub](RedisConnectionPtr &&conn) {
|
||||
// assert(status == REDIS_CONNECTED);
|
||||
auto thisPtr = weakThis.lock();
|
||||
if (!thisPtr)
|
||||
return;
|
||||
std::lock_guard<std::mutex> lock(thisPtr->connectionsMutex_);
|
||||
thisPtr->connections_.erase(conn);
|
||||
auto subPtr = weakSub.lock();
|
||||
if (!subPtr)
|
||||
return;
|
||||
subPtr->clearConnection();
|
||||
|
||||
auto loop = trantor::EventLoop::getEventLoopOfCurrentThread();
|
||||
assert(loop);
|
||||
loop->runAfter(2.0, [thisPtr, loop, subPtr]() {
|
||||
std::lock_guard<std::mutex> lock(thisPtr->connectionsMutex_);
|
||||
thisPtr->connections_.insert(
|
||||
thisPtr->newSubscribeConnection(loop, subPtr));
|
||||
});
|
||||
});
|
||||
conn->setIdleCallback([weakThis, weakSub](const RedisConnectionPtr &) {
|
||||
auto thisPtr = weakThis.lock();
|
||||
if (!thisPtr)
|
||||
return;
|
||||
auto subPtr = weakSub.lock();
|
||||
if (!subPtr)
|
||||
return;
|
||||
subPtr->subscribeNext();
|
||||
});
|
||||
return conn;
|
||||
}
|
||||
|
||||
void RedisClientImpl::execCommandAsync(
|
||||
RedisResultCallback &&resultCallback,
|
||||
RedisExceptionCallback &&exceptionCallback,
|
||||
std::string_view command,
|
||||
...) noexcept
|
||||
{
|
||||
if (timeout_ > 0.0)
|
||||
{
|
||||
va_list args;
|
||||
va_start(args, command);
|
||||
execCommandAsyncWithTimeout(command,
|
||||
std::move(resultCallback),
|
||||
std::move(exceptionCallback),
|
||||
args);
|
||||
va_end(args);
|
||||
return;
|
||||
}
|
||||
RedisConnectionPtr connPtr;
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(connectionsMutex_);
|
||||
if (!readyConnections_.empty())
|
||||
{
|
||||
if (connectionPos_ >= readyConnections_.size())
|
||||
{
|
||||
connPtr = readyConnections_[0];
|
||||
connectionPos_ = 1;
|
||||
}
|
||||
else
|
||||
{
|
||||
connPtr = readyConnections_[connectionPos_++];
|
||||
}
|
||||
}
|
||||
}
|
||||
if (connPtr)
|
||||
{
|
||||
va_list args;
|
||||
va_start(args, command);
|
||||
connPtr->sendvCommand(command,
|
||||
std::move(resultCallback),
|
||||
std::move(exceptionCallback),
|
||||
args);
|
||||
va_end(args);
|
||||
}
|
||||
else
|
||||
{
|
||||
LOG_TRACE << "no connection available, push command to buffer";
|
||||
va_list args;
|
||||
va_start(args, command);
|
||||
auto formattedCmd = RedisConnection::getFormattedCommand(command, args);
|
||||
va_end(args);
|
||||
std::lock_guard<std::mutex> lock(connectionsMutex_);
|
||||
tasks_.emplace_back(
|
||||
std::make_shared<std::function<void(const RedisConnectionPtr &)>>(
|
||||
[resultCallback = std::move(resultCallback),
|
||||
exceptionCallback = std::move(exceptionCallback),
|
||||
formattedCmd = std::move(formattedCmd)](
|
||||
const RedisConnectionPtr &connPtr) mutable {
|
||||
connPtr->sendFormattedCommand(std::move(formattedCmd),
|
||||
std::move(resultCallback),
|
||||
std::move(exceptionCallback));
|
||||
}));
|
||||
}
|
||||
}
|
||||
|
||||
RedisClientImpl::~RedisClientImpl()
|
||||
{
|
||||
closeAll();
|
||||
}
|
||||
|
||||
void RedisClientImpl::closeAll()
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(connectionsMutex_);
|
||||
for (auto &conn : connections_)
|
||||
{
|
||||
conn->disconnect();
|
||||
}
|
||||
readyConnections_.clear();
|
||||
connections_.clear();
|
||||
}
|
||||
|
||||
void RedisClientImpl::newTransactionAsync(
|
||||
const std::function<void(const std::shared_ptr<RedisTransaction> &)>
|
||||
&callback)
|
||||
{
|
||||
RedisConnectionPtr connPtr;
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(connectionsMutex_);
|
||||
if (!readyConnections_.empty())
|
||||
{
|
||||
connPtr = readyConnections_[readyConnections_.size() - 1];
|
||||
readyConnections_.resize(readyConnections_.size() - 1);
|
||||
}
|
||||
}
|
||||
if (connPtr)
|
||||
{
|
||||
callback(makeTransaction(connPtr));
|
||||
}
|
||||
else
|
||||
{
|
||||
if (timeout_ <= 0.0)
|
||||
{
|
||||
std::weak_ptr<RedisClientImpl> thisWeakPtr = shared_from_this();
|
||||
std::lock_guard<std::mutex> lock(connectionsMutex_);
|
||||
tasks_.emplace_back(
|
||||
std::make_shared<
|
||||
std::function<void(const RedisConnectionPtr &)>>(
|
||||
[callback,
|
||||
thisWeakPtr](const RedisConnectionPtr & /*connPtr*/) {
|
||||
auto thisPtr = thisWeakPtr.lock();
|
||||
if (thisPtr)
|
||||
{
|
||||
thisPtr->newTransactionAsync(callback);
|
||||
}
|
||||
}));
|
||||
}
|
||||
else
|
||||
{
|
||||
auto callbackPtr = std::make_shared<
|
||||
std::function<void(const std::shared_ptr<RedisTransaction> &)>>(
|
||||
callback);
|
||||
auto transCbPtr = std::make_shared<std::weak_ptr<
|
||||
std::function<void(const RedisConnectionPtr &)>>>();
|
||||
auto timeoutFlagPtr = std::make_shared<TaskTimeoutFlag>(
|
||||
loops_.getNextLoop(),
|
||||
std::chrono::duration<double>(timeout_),
|
||||
[callbackPtr, transCbPtr, this]() {
|
||||
auto cbPtr = (*transCbPtr).lock();
|
||||
if (cbPtr)
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(connectionsMutex_);
|
||||
for (auto iter = tasks_.begin(); iter != tasks_.end();
|
||||
++iter)
|
||||
{
|
||||
if (cbPtr == *iter)
|
||||
{
|
||||
tasks_.erase(iter);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
(*callbackPtr)(nullptr);
|
||||
});
|
||||
std::weak_ptr<RedisClientImpl> thisWeakPtr = shared_from_this();
|
||||
auto bufferCbPtr = std::make_shared<
|
||||
std::function<void(const RedisConnectionPtr &)>>(
|
||||
[callbackPtr, timeoutFlagPtr, thisWeakPtr](
|
||||
const RedisConnectionPtr & /*connPtr*/) {
|
||||
auto thisPtr = thisWeakPtr.lock();
|
||||
if (thisPtr)
|
||||
{
|
||||
if (timeoutFlagPtr->done())
|
||||
{
|
||||
return;
|
||||
}
|
||||
thisPtr->newTransactionAsync(*callbackPtr);
|
||||
}
|
||||
});
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(connectionsMutex_);
|
||||
tasks_.emplace_back(bufferCbPtr);
|
||||
}
|
||||
(*transCbPtr) = bufferCbPtr;
|
||||
timeoutFlagPtr->runTimer();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
std::shared_ptr<RedisTransaction> RedisClientImpl::makeTransaction(
|
||||
const RedisConnectionPtr &connPtr)
|
||||
{
|
||||
std::weak_ptr<RedisClientImpl> thisWeakPtr = shared_from_this();
|
||||
auto trans = std::shared_ptr<RedisTransactionImpl>(
|
||||
new RedisTransactionImpl(connPtr),
|
||||
[thisWeakPtr, connPtr](RedisTransactionImpl *p) {
|
||||
delete p;
|
||||
auto thisPtr = thisWeakPtr.lock();
|
||||
if (thisPtr)
|
||||
{
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(
|
||||
thisPtr->connectionsMutex_);
|
||||
thisPtr->readyConnections_.push_back(connPtr);
|
||||
}
|
||||
thisPtr->handleNextTask(connPtr);
|
||||
}
|
||||
});
|
||||
trans->doBegin();
|
||||
return trans;
|
||||
}
|
||||
|
||||
void RedisClientImpl::handleNextTask(const RedisConnectionPtr &connPtr)
|
||||
{
|
||||
std::shared_ptr<std::function<void(const RedisConnectionPtr &)>> taskPtr;
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(connectionsMutex_);
|
||||
if (!tasks_.empty())
|
||||
{
|
||||
taskPtr = std::move(tasks_.front());
|
||||
tasks_.pop_front();
|
||||
}
|
||||
}
|
||||
if (taskPtr && (*taskPtr))
|
||||
{
|
||||
(*taskPtr)(connPtr);
|
||||
}
|
||||
}
|
||||
|
||||
void RedisClientImpl::execCommandAsyncWithTimeout(
|
||||
std::string_view command,
|
||||
RedisResultCallback &&resultCallback,
|
||||
RedisExceptionCallback &&exceptionCallback,
|
||||
va_list ap)
|
||||
{
|
||||
auto expCbPtr =
|
||||
std::make_shared<RedisExceptionCallback>(std::move(exceptionCallback));
|
||||
auto bufferCbPtr = std::make_shared<
|
||||
std::weak_ptr<std::function<void(const RedisConnectionPtr &)>>>();
|
||||
auto timeoutFlagPtr = std::make_shared<TaskTimeoutFlag>(
|
||||
loops_.getNextLoop(),
|
||||
std::chrono::duration<double>(timeout_),
|
||||
[expCbPtr, bufferCbPtr, this]() {
|
||||
auto bfCbPtr = (*bufferCbPtr).lock();
|
||||
if (bfCbPtr)
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(connectionsMutex_);
|
||||
for (auto iter = tasks_.begin(); iter != tasks_.end(); ++iter)
|
||||
{
|
||||
if (bfCbPtr == *iter)
|
||||
{
|
||||
tasks_.erase(iter);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (*expCbPtr)
|
||||
{
|
||||
(*expCbPtr)(RedisException(RedisErrorCode::kTimeout,
|
||||
"Command execution timeout"));
|
||||
}
|
||||
});
|
||||
auto newResultCallback = [resultCallback = std::move(resultCallback),
|
||||
timeoutFlagPtr](const RedisResult &result) {
|
||||
if (timeoutFlagPtr->done())
|
||||
{
|
||||
return;
|
||||
}
|
||||
if (resultCallback)
|
||||
{
|
||||
resultCallback(result);
|
||||
}
|
||||
};
|
||||
auto newExceptionCallback = [expCbPtr,
|
||||
timeoutFlagPtr](const RedisException &err) {
|
||||
if (timeoutFlagPtr->done())
|
||||
{
|
||||
return;
|
||||
}
|
||||
if (*expCbPtr)
|
||||
{
|
||||
(*expCbPtr)(err);
|
||||
}
|
||||
};
|
||||
RedisConnectionPtr connPtr;
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(connectionsMutex_);
|
||||
if (!readyConnections_.empty())
|
||||
{
|
||||
if (connectionPos_ >= readyConnections_.size())
|
||||
{
|
||||
connPtr = readyConnections_[0];
|
||||
connectionPos_ = 1;
|
||||
}
|
||||
else
|
||||
{
|
||||
connPtr = readyConnections_[connectionPos_++];
|
||||
}
|
||||
}
|
||||
}
|
||||
if (connPtr)
|
||||
{
|
||||
connPtr->sendvCommand(command,
|
||||
std::move(newResultCallback),
|
||||
std::move(newExceptionCallback),
|
||||
ap);
|
||||
}
|
||||
else
|
||||
{
|
||||
LOG_TRACE << "no connection available, push command to buffer";
|
||||
auto formattedCmd = RedisConnection::getFormattedCommand(command, ap);
|
||||
auto bfCbPtr =
|
||||
std::make_shared<std::function<void(const RedisConnectionPtr &)>>(
|
||||
[resultCallback = std::move(newResultCallback),
|
||||
exceptionCallback = std::move(newExceptionCallback),
|
||||
formattedCmd = std::move(formattedCmd)](
|
||||
const RedisConnectionPtr &connPtr) mutable {
|
||||
connPtr->sendFormattedCommand(std::move(formattedCmd),
|
||||
std::move(resultCallback),
|
||||
std::move(exceptionCallback));
|
||||
});
|
||||
(*bufferCbPtr) = bfCbPtr;
|
||||
std::lock_guard<std::mutex> lock(connectionsMutex_);
|
||||
tasks_.emplace_back(bfCbPtr);
|
||||
}
|
||||
timeoutFlagPtr->runTimer();
|
||||
}
|
||||
|
||||
std::shared_ptr<RedisSubscriber> RedisClientImpl::newSubscriber() noexcept
|
||||
{
|
||||
auto subscriber = std::make_shared<RedisSubscriberImpl>();
|
||||
auto loop = loops_.getNextLoop();
|
||||
loop->queueInLoop([this, loop, subscriber]() {
|
||||
std::lock_guard<std::mutex> lock(connectionsMutex_);
|
||||
connections_.insert(newSubscribeConnection(loop, subscriber));
|
||||
});
|
||||
|
||||
return subscriber;
|
||||
}
|
||||
@@ -0,0 +1,110 @@
|
||||
/**
|
||||
*
|
||||
* @file RedisClientImpl.h
|
||||
* @author An Tao
|
||||
*
|
||||
* Copyright 2018, An Tao. All rights reserved.
|
||||
* https://github.com/an-tao/drogon
|
||||
* Use of this source code is governed by a MIT license
|
||||
* that can be found in the License file.
|
||||
*
|
||||
* Drogon
|
||||
*
|
||||
*/
|
||||
#pragma once
|
||||
|
||||
#include "RedisConnection.h"
|
||||
#include "RedisSubscriberImpl.h"
|
||||
#include "SubscribeContext.h"
|
||||
#include <drogon/nosql/RedisClient.h>
|
||||
#include <trantor/utils/NonCopyable.h>
|
||||
#include <trantor/net/EventLoopThreadPool.h>
|
||||
#include <vector>
|
||||
#include <unordered_set>
|
||||
#include <list>
|
||||
#include <future>
|
||||
|
||||
namespace drogon
|
||||
{
|
||||
namespace nosql
|
||||
{
|
||||
class RedisConnection;
|
||||
using RedisConnectionPtr = std::shared_ptr<RedisConnection>;
|
||||
|
||||
class RedisClientImpl final
|
||||
: public RedisClient,
|
||||
public trantor::NonCopyable,
|
||||
public std::enable_shared_from_this<RedisClientImpl>
|
||||
{
|
||||
public:
|
||||
RedisClientImpl(const trantor::InetAddress &serverAddress,
|
||||
size_t numberOfConnections,
|
||||
std::string username = "",
|
||||
std::string password = "",
|
||||
unsigned int db = 0);
|
||||
void execCommandAsync(RedisResultCallback &&resultCallback,
|
||||
RedisExceptionCallback &&exceptionCallback,
|
||||
std::string_view command,
|
||||
...) noexcept override;
|
||||
~RedisClientImpl() override;
|
||||
std::shared_ptr<RedisSubscriber> newSubscriber() noexcept override;
|
||||
|
||||
RedisTransactionPtr newTransaction() noexcept(false) override
|
||||
{
|
||||
std::promise<RedisTransactionPtr> prom;
|
||||
auto f = prom.get_future();
|
||||
newTransactionAsync([&prom](const RedisTransactionPtr &transPtr) {
|
||||
prom.set_value(transPtr);
|
||||
});
|
||||
auto trans = f.get();
|
||||
if (!trans)
|
||||
{
|
||||
throw RedisException(
|
||||
RedisErrorCode::kTimeout,
|
||||
"Timeout, no connection available for transaction");
|
||||
}
|
||||
return trans;
|
||||
}
|
||||
|
||||
void newTransactionAsync(
|
||||
const std::function<void(const RedisTransactionPtr &)> &callback)
|
||||
override;
|
||||
|
||||
void setTimeout(double timeout) override
|
||||
{
|
||||
timeout_ = timeout;
|
||||
}
|
||||
|
||||
void init();
|
||||
void closeAll() override;
|
||||
|
||||
private:
|
||||
trantor::EventLoopThreadPool loops_;
|
||||
std::mutex connectionsMutex_;
|
||||
std::unordered_set<RedisConnectionPtr> connections_;
|
||||
std::vector<RedisConnectionPtr> readyConnections_;
|
||||
size_t connectionPos_{0};
|
||||
const trantor::InetAddress serverAddr_;
|
||||
const std::string username_;
|
||||
const std::string password_;
|
||||
const unsigned int db_;
|
||||
const size_t numberOfConnections_;
|
||||
double timeout_{-1.0};
|
||||
std::list<std::shared_ptr<std::function<void(const RedisConnectionPtr &)>>>
|
||||
tasks_;
|
||||
|
||||
RedisConnectionPtr newConnection(trantor::EventLoop *loop);
|
||||
RedisConnectionPtr newSubscribeConnection(
|
||||
trantor::EventLoop *loop,
|
||||
const std::shared_ptr<RedisSubscriberImpl> &subscriber);
|
||||
|
||||
std::shared_ptr<RedisTransaction> makeTransaction(
|
||||
const RedisConnectionPtr &connPtr);
|
||||
void handleNextTask(const RedisConnectionPtr &connPtr);
|
||||
void execCommandAsyncWithTimeout(std::string_view command,
|
||||
RedisResultCallback &&resultCallback,
|
||||
RedisExceptionCallback &&exceptionCallback,
|
||||
va_list ap);
|
||||
};
|
||||
} // namespace nosql
|
||||
} // namespace drogon
|
||||
@@ -0,0 +1,445 @@
|
||||
/**
|
||||
*
|
||||
* @file RedisClientLockFree.cc
|
||||
* @author An Tao
|
||||
*
|
||||
* Copyright 2018, An Tao. All rights reserved.
|
||||
* https://github.com/an-tao/drogon
|
||||
* Use of this source code is governed by a MIT license
|
||||
* that can be found in the License file.
|
||||
*
|
||||
* Drogon
|
||||
*
|
||||
*/
|
||||
|
||||
#include "RedisConnection.h"
|
||||
#include "RedisClientLockFree.h"
|
||||
#include "RedisSubscriberImpl.h"
|
||||
#include "RedisTransactionImpl.h"
|
||||
#include "../../lib/src/TaskTimeoutFlag.h"
|
||||
using namespace drogon::nosql;
|
||||
|
||||
RedisClientLockFree::RedisClientLockFree(
|
||||
const trantor::InetAddress &serverAddress,
|
||||
size_t numberOfConnections,
|
||||
trantor::EventLoop *loop,
|
||||
std::string username,
|
||||
std::string password,
|
||||
unsigned int db)
|
||||
: loop_(loop),
|
||||
serverAddr_(serverAddress),
|
||||
username_(std::move(username)),
|
||||
password_(std::move(password)),
|
||||
db_(db),
|
||||
numberOfConnections_(numberOfConnections)
|
||||
{
|
||||
assert(loop_);
|
||||
for (size_t i = 0; i < numberOfConnections_; ++i)
|
||||
{
|
||||
loop_->queueInLoop([this]() { connections_.insert(newConnection()); });
|
||||
}
|
||||
}
|
||||
|
||||
RedisConnectionPtr RedisClientLockFree::newConnection()
|
||||
{
|
||||
loop_->assertInLoopThread();
|
||||
auto conn = std::make_shared<RedisConnection>(
|
||||
serverAddr_, username_, password_, db_, loop_);
|
||||
std::weak_ptr<RedisClientLockFree> thisWeakPtr = shared_from_this();
|
||||
conn->setConnectCallback([thisWeakPtr](RedisConnectionPtr &&conn) {
|
||||
auto thisPtr = thisWeakPtr.lock();
|
||||
if (thisPtr)
|
||||
{
|
||||
thisPtr->readyConnections_.push_back(conn);
|
||||
thisPtr->handleNextTask(conn);
|
||||
}
|
||||
});
|
||||
conn->setDisconnectCallback([thisWeakPtr](RedisConnectionPtr &&conn) {
|
||||
// assert(status == REDIS_CONNECTED);
|
||||
auto thisPtr = thisWeakPtr.lock();
|
||||
if (thisPtr)
|
||||
{
|
||||
thisPtr->connections_.erase(conn);
|
||||
for (auto iter = thisPtr->readyConnections_.begin();
|
||||
iter != thisPtr->readyConnections_.end();
|
||||
++iter)
|
||||
{
|
||||
if (*iter == conn)
|
||||
{
|
||||
thisPtr->readyConnections_.erase(iter);
|
||||
break;
|
||||
}
|
||||
}
|
||||
thisPtr->loop_->runAfter(2.0, [thisPtr, conn]() {
|
||||
thisPtr->connections_.insert(thisPtr->newConnection());
|
||||
});
|
||||
}
|
||||
});
|
||||
conn->setIdleCallback([thisWeakPtr](const RedisConnectionPtr &connPtr) {
|
||||
auto thisPtr = thisWeakPtr.lock();
|
||||
if (thisPtr)
|
||||
{
|
||||
thisPtr->handleNextTask(connPtr);
|
||||
}
|
||||
});
|
||||
return conn;
|
||||
}
|
||||
|
||||
RedisConnectionPtr RedisClientLockFree::newSubscribeConnection(
|
||||
const std::shared_ptr<RedisSubscriberImpl> &subscriber)
|
||||
{
|
||||
loop_->assertInLoopThread();
|
||||
auto conn = std::make_shared<RedisConnection>(
|
||||
serverAddr_, username_, password_, db_, loop_);
|
||||
std::weak_ptr<RedisClientLockFree> weakThis = shared_from_this();
|
||||
std::weak_ptr<RedisSubscriberImpl> weakSub(subscriber);
|
||||
conn->setConnectCallback([weakThis, weakSub](RedisConnectionPtr &&conn) {
|
||||
conn->getLoop()->assertInLoopThread(); // TODO: remove
|
||||
auto thisPtr = weakThis.lock();
|
||||
if (!thisPtr)
|
||||
return;
|
||||
thisPtr->loop_->assertInLoopThread(); // TODO: remove
|
||||
auto subPtr = weakSub.lock();
|
||||
if (subPtr)
|
||||
{
|
||||
subPtr->setConnection(conn);
|
||||
subPtr->subscribeAll();
|
||||
}
|
||||
else
|
||||
{
|
||||
thisPtr->connections_.erase(conn);
|
||||
}
|
||||
});
|
||||
conn->setDisconnectCallback([weakThis, weakSub](RedisConnectionPtr &&conn) {
|
||||
// assert(status == REDIS_CONNECTED);
|
||||
auto thisPtr = weakThis.lock();
|
||||
if (!thisPtr)
|
||||
return;
|
||||
thisPtr->connections_.erase(conn);
|
||||
auto subPtr = weakSub.lock();
|
||||
if (!subPtr)
|
||||
return;
|
||||
subPtr->clearConnection();
|
||||
|
||||
thisPtr->loop_->runAfter(2.0, [thisPtr, subPtr]() {
|
||||
thisPtr->connections_.insert(
|
||||
thisPtr->newSubscribeConnection(subPtr));
|
||||
});
|
||||
});
|
||||
conn->setIdleCallback(
|
||||
[weakThis, weakSub](const RedisConnectionPtr &connPtr) {
|
||||
auto thisPtr = weakThis.lock();
|
||||
if (!thisPtr)
|
||||
return;
|
||||
auto subPtr = weakSub.lock();
|
||||
if (!subPtr)
|
||||
return;
|
||||
subPtr->subscribeNext();
|
||||
});
|
||||
return conn;
|
||||
}
|
||||
|
||||
void RedisClientLockFree::execCommandAsync(
|
||||
RedisResultCallback &&resultCallback,
|
||||
RedisExceptionCallback &&exceptionCallback,
|
||||
std::string_view command,
|
||||
...) noexcept
|
||||
{
|
||||
loop_->assertInLoopThread();
|
||||
if (timeout_ > 0.0)
|
||||
{
|
||||
va_list args;
|
||||
va_start(args, command);
|
||||
execCommandAsyncWithTimeout(command,
|
||||
std::move(resultCallback),
|
||||
std::move(exceptionCallback),
|
||||
args);
|
||||
va_end(args);
|
||||
return;
|
||||
}
|
||||
RedisConnectionPtr connPtr;
|
||||
{
|
||||
if (!readyConnections_.empty())
|
||||
{
|
||||
if (connectionPos_ >= readyConnections_.size())
|
||||
{
|
||||
connPtr = readyConnections_[0];
|
||||
connectionPos_ = 1;
|
||||
}
|
||||
else
|
||||
{
|
||||
connPtr = readyConnections_[connectionPos_++];
|
||||
}
|
||||
}
|
||||
}
|
||||
if (connPtr)
|
||||
{
|
||||
va_list args;
|
||||
va_start(args, command);
|
||||
connPtr->sendvCommand(command,
|
||||
std::move(resultCallback),
|
||||
std::move(exceptionCallback),
|
||||
args);
|
||||
va_end(args);
|
||||
}
|
||||
else
|
||||
{
|
||||
LOG_TRACE << "no connection available, push command to buffer";
|
||||
std::weak_ptr<RedisClientLockFree> thisWeakPtr = shared_from_this();
|
||||
va_list args;
|
||||
va_start(args, command);
|
||||
auto formattedCmd = RedisConnection::getFormattedCommand(command, args);
|
||||
va_end(args);
|
||||
tasks_.emplace_back(
|
||||
std::make_shared<std::function<void(const RedisConnectionPtr &)>>(
|
||||
[thisWeakPtr,
|
||||
resultCallback = std::move(resultCallback),
|
||||
exceptionCallback = std::move(exceptionCallback),
|
||||
formattedCmd = std::move(formattedCmd)](
|
||||
const RedisConnectionPtr &connPtr) mutable {
|
||||
connPtr->sendFormattedCommand(std::move(formattedCmd),
|
||||
std::move(resultCallback),
|
||||
std::move(exceptionCallback));
|
||||
}));
|
||||
}
|
||||
}
|
||||
|
||||
RedisClientLockFree::~RedisClientLockFree()
|
||||
{
|
||||
closeAll();
|
||||
}
|
||||
|
||||
void RedisClientLockFree::closeAll()
|
||||
{
|
||||
for (auto &conn : connections_)
|
||||
{
|
||||
conn->disconnect();
|
||||
}
|
||||
readyConnections_.clear();
|
||||
connections_.clear();
|
||||
}
|
||||
|
||||
void RedisClientLockFree::newTransactionAsync(
|
||||
const std::function<void(const std::shared_ptr<RedisTransaction> &)>
|
||||
&callback)
|
||||
{
|
||||
loop_->assertInLoopThread();
|
||||
RedisConnectionPtr connPtr;
|
||||
|
||||
if (!readyConnections_.empty())
|
||||
{
|
||||
connPtr = readyConnections_[readyConnections_.size() - 1];
|
||||
readyConnections_.resize(readyConnections_.size() - 1);
|
||||
}
|
||||
|
||||
if (connPtr)
|
||||
{
|
||||
callback(makeTransaction(connPtr));
|
||||
}
|
||||
else
|
||||
{
|
||||
if (timeout_ <= 0.0)
|
||||
{
|
||||
std::weak_ptr<RedisClientLockFree> thisWeakPtr = shared_from_this();
|
||||
tasks_.emplace_back(
|
||||
std::make_shared<
|
||||
std::function<void(const RedisConnectionPtr &)>>(
|
||||
[callback,
|
||||
thisWeakPtr](const RedisConnectionPtr & /*connPtr*/) {
|
||||
auto thisPtr = thisWeakPtr.lock();
|
||||
if (thisPtr)
|
||||
{
|
||||
thisPtr->newTransactionAsync(callback);
|
||||
}
|
||||
}));
|
||||
}
|
||||
else
|
||||
{
|
||||
auto callbackPtr = std::make_shared<
|
||||
std::function<void(const std::shared_ptr<RedisTransaction> &)>>(
|
||||
callback);
|
||||
auto transCbPtr = std::make_shared<std::weak_ptr<
|
||||
std::function<void(const RedisConnectionPtr &)>>>();
|
||||
auto timeoutFlagPtr = std::make_shared<TaskTimeoutFlag>(
|
||||
loop_,
|
||||
std::chrono::duration<double>(timeout_),
|
||||
[callbackPtr, transCbPtr, this]() {
|
||||
auto cbPtr = (*transCbPtr).lock();
|
||||
if (cbPtr)
|
||||
{
|
||||
for (auto iter = tasks_.begin(); iter != tasks_.end();
|
||||
++iter)
|
||||
{
|
||||
if (cbPtr == *iter)
|
||||
{
|
||||
tasks_.erase(iter);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
(*callbackPtr)(nullptr);
|
||||
});
|
||||
std::weak_ptr<RedisClientLockFree> thisWeakPtr = shared_from_this();
|
||||
auto bufferCbPtr = std::make_shared<
|
||||
std::function<void(const RedisConnectionPtr &)>>(
|
||||
[callbackPtr, timeoutFlagPtr, thisWeakPtr](
|
||||
const RedisConnectionPtr & /*connPtr*/) {
|
||||
auto thisPtr = thisWeakPtr.lock();
|
||||
if (thisPtr)
|
||||
{
|
||||
if (timeoutFlagPtr->done())
|
||||
{
|
||||
return;
|
||||
}
|
||||
thisPtr->newTransactionAsync(*callbackPtr);
|
||||
}
|
||||
});
|
||||
tasks_.emplace_back(bufferCbPtr);
|
||||
|
||||
(*transCbPtr) = bufferCbPtr;
|
||||
timeoutFlagPtr->runTimer();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
std::shared_ptr<RedisTransaction> RedisClientLockFree::makeTransaction(
|
||||
const RedisConnectionPtr &connPtr)
|
||||
{
|
||||
std::weak_ptr<RedisClientLockFree> thisWeakPtr = shared_from_this();
|
||||
auto trans = std::shared_ptr<RedisTransactionImpl>(
|
||||
new RedisTransactionImpl(connPtr),
|
||||
[thisWeakPtr, connPtr](RedisTransactionImpl *p) {
|
||||
delete p;
|
||||
auto thisPtr = thisWeakPtr.lock();
|
||||
if (thisPtr)
|
||||
{
|
||||
thisPtr->readyConnections_.push_back(connPtr);
|
||||
thisPtr->handleNextTask(connPtr);
|
||||
}
|
||||
});
|
||||
trans->doBegin();
|
||||
return trans;
|
||||
}
|
||||
|
||||
void RedisClientLockFree::handleNextTask(const RedisConnectionPtr &connPtr)
|
||||
{
|
||||
loop_->assertInLoopThread();
|
||||
std::shared_ptr<std::function<void(const RedisConnectionPtr &)>> taskPtr;
|
||||
|
||||
if (!tasks_.empty())
|
||||
{
|
||||
taskPtr = std::move(tasks_.front());
|
||||
tasks_.pop_front();
|
||||
}
|
||||
|
||||
if (taskPtr && (*taskPtr))
|
||||
{
|
||||
(*taskPtr)(connPtr);
|
||||
}
|
||||
}
|
||||
|
||||
void RedisClientLockFree::execCommandAsyncWithTimeout(
|
||||
std::string_view command,
|
||||
RedisResultCallback &&resultCallback,
|
||||
RedisExceptionCallback &&exceptionCallback,
|
||||
va_list ap)
|
||||
{
|
||||
auto expCbPtr =
|
||||
std::make_shared<RedisExceptionCallback>(std::move(exceptionCallback));
|
||||
auto bufferCbPtr = std::make_shared<
|
||||
std::weak_ptr<std::function<void(const RedisConnectionPtr &)>>>();
|
||||
auto timeoutFlagPtr = std::make_shared<TaskTimeoutFlag>(
|
||||
loop_,
|
||||
std::chrono::duration<double>(timeout_),
|
||||
[expCbPtr, bufferCbPtr, this]() {
|
||||
auto bfCbPtr = (*bufferCbPtr).lock();
|
||||
if (bfCbPtr)
|
||||
{
|
||||
for (auto iter = tasks_.begin(); iter != tasks_.end(); ++iter)
|
||||
{
|
||||
if (bfCbPtr == *iter)
|
||||
{
|
||||
tasks_.erase(iter);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (*expCbPtr)
|
||||
{
|
||||
(*expCbPtr)(RedisException(RedisErrorCode::kTimeout,
|
||||
"Command execution timeout"));
|
||||
}
|
||||
});
|
||||
auto newResultCallback = [resultCallback = std::move(resultCallback),
|
||||
timeoutFlagPtr](const RedisResult &result) {
|
||||
if (timeoutFlagPtr->done())
|
||||
{
|
||||
return;
|
||||
}
|
||||
if (resultCallback)
|
||||
{
|
||||
resultCallback(result);
|
||||
}
|
||||
};
|
||||
auto newExceptionCallback = [expCbPtr,
|
||||
timeoutFlagPtr](const RedisException &err) {
|
||||
if (timeoutFlagPtr->done())
|
||||
{
|
||||
return;
|
||||
}
|
||||
if (*expCbPtr)
|
||||
{
|
||||
(*expCbPtr)(err);
|
||||
}
|
||||
};
|
||||
RedisConnectionPtr connPtr;
|
||||
{
|
||||
if (!readyConnections_.empty())
|
||||
{
|
||||
if (connectionPos_ >= readyConnections_.size())
|
||||
{
|
||||
connPtr = readyConnections_[0];
|
||||
connectionPos_ = 1;
|
||||
}
|
||||
else
|
||||
{
|
||||
connPtr = readyConnections_[connectionPos_++];
|
||||
}
|
||||
}
|
||||
}
|
||||
if (connPtr)
|
||||
{
|
||||
connPtr->sendvCommand(command,
|
||||
std::move(newResultCallback),
|
||||
std::move(newExceptionCallback),
|
||||
ap);
|
||||
}
|
||||
else
|
||||
{
|
||||
LOG_TRACE << "no connection available, push command to buffer";
|
||||
auto formattedCmd = RedisConnection::getFormattedCommand(command, ap);
|
||||
auto bfCbPtr =
|
||||
std::make_shared<std::function<void(const RedisConnectionPtr &)>>(
|
||||
[resultCallback = std::move(newResultCallback),
|
||||
exceptionCallback = std::move(newExceptionCallback),
|
||||
formattedCmd = std::move(formattedCmd)](
|
||||
const RedisConnectionPtr &connPtr) mutable {
|
||||
connPtr->sendFormattedCommand(std::move(formattedCmd),
|
||||
std::move(resultCallback),
|
||||
std::move(exceptionCallback));
|
||||
});
|
||||
(*bufferCbPtr) = bfCbPtr;
|
||||
tasks_.emplace_back(bfCbPtr);
|
||||
}
|
||||
timeoutFlagPtr->runTimer();
|
||||
}
|
||||
|
||||
std::shared_ptr<RedisSubscriber> RedisClientLockFree::newSubscriber() noexcept
|
||||
{
|
||||
auto subscriber = std::make_shared<RedisSubscriberImpl>();
|
||||
loop_->runInLoop([this, subscriber]() {
|
||||
connections_.insert(newSubscribeConnection(subscriber));
|
||||
});
|
||||
|
||||
return subscriber;
|
||||
}
|
||||
@@ -0,0 +1,99 @@
|
||||
/**
|
||||
*
|
||||
* @file RedisClientLockFree.h
|
||||
* @author An Tao
|
||||
*
|
||||
* Copyright 2018, An Tao. All rights reserved.
|
||||
* https://github.com/an-tao/drogon
|
||||
* Use of this source code is governed by a MIT license
|
||||
* that can be found in the License file.
|
||||
*
|
||||
* Drogon
|
||||
*
|
||||
*/
|
||||
#pragma once
|
||||
|
||||
#include "RedisConnection.h"
|
||||
#include "RedisSubscriberImpl.h"
|
||||
#include <drogon/nosql/RedisClient.h>
|
||||
#include <trantor/utils/NonCopyable.h>
|
||||
#include <trantor/net/EventLoopThreadPool.h>
|
||||
#include <vector>
|
||||
#include <unordered_set>
|
||||
#include <list>
|
||||
#include <future>
|
||||
|
||||
namespace drogon
|
||||
{
|
||||
namespace nosql
|
||||
{
|
||||
class RedisConnection;
|
||||
using RedisConnectionPtr = std::shared_ptr<RedisConnection>;
|
||||
|
||||
class RedisClientLockFree final
|
||||
: public RedisClient,
|
||||
public trantor::NonCopyable,
|
||||
public std::enable_shared_from_this<RedisClientLockFree>
|
||||
{
|
||||
public:
|
||||
RedisClientLockFree(const trantor::InetAddress &serverAddress,
|
||||
size_t numberOfConnections,
|
||||
trantor::EventLoop *loop,
|
||||
std::string username = "",
|
||||
std::string password = "",
|
||||
unsigned int db = 0);
|
||||
void execCommandAsync(RedisResultCallback &&resultCallback,
|
||||
RedisExceptionCallback &&exceptionCallback,
|
||||
std::string_view command,
|
||||
...) noexcept override;
|
||||
~RedisClientLockFree() override;
|
||||
std::shared_ptr<RedisSubscriber> newSubscriber() noexcept override;
|
||||
|
||||
RedisTransactionPtr newTransaction() override
|
||||
{
|
||||
LOG_ERROR
|
||||
<< "You can't use the synchronous interface in the fast redis "
|
||||
"client, please use the asynchronous version "
|
||||
"(newTransactionAsync)";
|
||||
assert(0);
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
void newTransactionAsync(
|
||||
const std::function<void(const RedisTransactionPtr &)> &callback)
|
||||
override;
|
||||
|
||||
void setTimeout(double timeout) override
|
||||
{
|
||||
timeout_ = timeout;
|
||||
}
|
||||
|
||||
void closeAll() override;
|
||||
|
||||
private:
|
||||
trantor::EventLoop *loop_;
|
||||
std::unordered_set<RedisConnectionPtr> connections_;
|
||||
std::vector<RedisConnectionPtr> readyConnections_;
|
||||
size_t connectionPos_{0};
|
||||
const trantor::InetAddress serverAddr_;
|
||||
const std::string username_;
|
||||
const std::string password_;
|
||||
const unsigned int db_;
|
||||
const size_t numberOfConnections_;
|
||||
std::list<std::shared_ptr<std::function<void(const RedisConnectionPtr &)>>>
|
||||
tasks_;
|
||||
double timeout_{-1.0};
|
||||
|
||||
RedisConnectionPtr newConnection();
|
||||
RedisConnectionPtr newSubscribeConnection(
|
||||
const std::shared_ptr<RedisSubscriberImpl> &);
|
||||
std::shared_ptr<RedisTransaction> makeTransaction(
|
||||
const RedisConnectionPtr &connPtr);
|
||||
void handleNextTask(const RedisConnectionPtr &connPtr);
|
||||
void execCommandAsyncWithTimeout(std::string_view command,
|
||||
RedisResultCallback &&resultCallback,
|
||||
RedisExceptionCallback &&exceptionCallback,
|
||||
va_list ap);
|
||||
};
|
||||
} // namespace nosql
|
||||
} // namespace drogon
|
||||
@@ -0,0 +1,133 @@
|
||||
/**
|
||||
*
|
||||
* @file RedisClientManager.cc
|
||||
* @author An Tao
|
||||
*
|
||||
* Copyright 2018, An Tao. All rights reserved.
|
||||
* https://github.com/an-tao/drogon
|
||||
* Use of this source code is governed by a MIT license
|
||||
* that can be found in the License file.
|
||||
*
|
||||
* Drogon
|
||||
*
|
||||
*/
|
||||
|
||||
#include "../../lib/src/RedisClientManager.h"
|
||||
#include "RedisClientLockFree.h"
|
||||
#include "RedisClientImpl.h"
|
||||
|
||||
#include <algorithm>
|
||||
|
||||
using namespace drogon::nosql;
|
||||
using namespace drogon;
|
||||
|
||||
void RedisClientManager::createRedisClients(
|
||||
const std::vector<trantor::EventLoop *> &ioLoops)
|
||||
{
|
||||
assert(redisClientsMap_.empty());
|
||||
assert(redisFastClientsMap_.empty());
|
||||
for (auto &redisInfo : redisInfos_)
|
||||
{
|
||||
if (redisInfo.isFast_)
|
||||
{
|
||||
redisFastClientsMap_[redisInfo.name_] =
|
||||
IOThreadStorage<RedisClientPtr>();
|
||||
redisFastClientsMap_[redisInfo.name_].init([&](RedisClientPtr &c,
|
||||
size_t idx) {
|
||||
assert(idx == ioLoops[idx]->index());
|
||||
LOG_TRACE << "create fast redis client for the thread " << idx;
|
||||
c = std::make_shared<RedisClientLockFree>(
|
||||
trantor::InetAddress(redisInfo.addr_, redisInfo.port_),
|
||||
redisInfo.connectionNumber_,
|
||||
ioLoops[idx],
|
||||
redisInfo.username_,
|
||||
redisInfo.password_,
|
||||
redisInfo.db_);
|
||||
if (redisInfo.timeout_ > 0.0)
|
||||
{
|
||||
c->setTimeout(redisInfo.timeout_);
|
||||
}
|
||||
});
|
||||
}
|
||||
else
|
||||
{
|
||||
auto clientPtr = std::make_shared<RedisClientImpl>(
|
||||
trantor::InetAddress(redisInfo.addr_, redisInfo.port_),
|
||||
redisInfo.connectionNumber_,
|
||||
redisInfo.username_,
|
||||
redisInfo.password_,
|
||||
redisInfo.db_);
|
||||
if (redisInfo.timeout_ > 0.0)
|
||||
{
|
||||
clientPtr->setTimeout(redisInfo.timeout_);
|
||||
}
|
||||
clientPtr->init();
|
||||
redisClientsMap_[redisInfo.name_] = std::move(clientPtr);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void RedisClientManager::createRedisClient(const std::string &name,
|
||||
const std::string &addr,
|
||||
unsigned short port,
|
||||
const std::string &username,
|
||||
const std::string &password,
|
||||
const size_t connectionNum,
|
||||
const bool isFast,
|
||||
double timeout,
|
||||
unsigned int db)
|
||||
{
|
||||
RedisInfo info;
|
||||
info.name_ = name;
|
||||
info.addr_ = addr;
|
||||
info.port_ = port;
|
||||
info.username_ = username;
|
||||
info.password_ = password;
|
||||
info.connectionNumber_ = connectionNum;
|
||||
info.isFast_ = isFast;
|
||||
info.timeout_ = timeout;
|
||||
info.db_ = db;
|
||||
|
||||
redisInfos_.emplace_back(std::move(info));
|
||||
}
|
||||
|
||||
// bool RedisClientManager::areAllRedisClientsAvailable() const noexcept
|
||||
//{
|
||||
// for (auto const &pair : redisClientsMap_)
|
||||
// {
|
||||
// if (!(pair.second)->hasAvailableConnections())
|
||||
// return false;
|
||||
// }
|
||||
// auto loop = trantor::EventLoop::getEventLoopOfCurrentThread();
|
||||
// if (loop && loop->index() < app().getThreadNum())
|
||||
// {
|
||||
// for (auto const &pair : redisFastClientsMap_)
|
||||
// {
|
||||
// if (!(*(pair.second))->hasAvailableConnections())
|
||||
// return false;
|
||||
// }
|
||||
// }
|
||||
// return true;
|
||||
//}
|
||||
|
||||
RedisClientManager::~RedisClientManager()
|
||||
{
|
||||
for (auto &pair : redisClientsMap_)
|
||||
{
|
||||
pair.second->closeAll();
|
||||
}
|
||||
for (auto &pair : redisFastClientsMap_)
|
||||
{
|
||||
pair.second.init([](RedisClientPtr &clientPtr, size_t index) {
|
||||
// the main loop;
|
||||
std::promise<void> p;
|
||||
auto f = p.get_future();
|
||||
drogon::getIOThreadStorageLoop(index)->runInLoop(
|
||||
[&clientPtr, &p]() {
|
||||
clientPtr->closeAll();
|
||||
p.set_value();
|
||||
});
|
||||
f.get();
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,522 @@
|
||||
/**
|
||||
*
|
||||
* @file RedisConnection.cc
|
||||
* @author An Tao
|
||||
*
|
||||
* Copyright 2018, An Tao. All rights reserved.
|
||||
* https://github.com/an-tao/drogon
|
||||
* Use of this source code is governed by a MIT license
|
||||
* that can be found in the License file.
|
||||
*
|
||||
* Drogon
|
||||
*
|
||||
*/
|
||||
|
||||
#include "RedisConnection.h"
|
||||
#include <drogon/nosql/RedisResult.h>
|
||||
#include <future>
|
||||
#include <string.h>
|
||||
|
||||
#ifdef _MSC_VER
|
||||
#define strcasecmp _stricmp
|
||||
#endif
|
||||
|
||||
using namespace drogon::nosql;
|
||||
|
||||
RedisConnection::RedisConnection(const trantor::InetAddress &serverAddress,
|
||||
const std::string &username,
|
||||
const std::string &password,
|
||||
unsigned int db,
|
||||
trantor::EventLoop *loop)
|
||||
: serverAddr_(serverAddress),
|
||||
username_(username),
|
||||
password_(password),
|
||||
db_(db),
|
||||
loop_(loop)
|
||||
{
|
||||
assert(loop_);
|
||||
loop_->queueInLoop([this]() { startConnectionInLoop(); });
|
||||
}
|
||||
|
||||
void RedisConnection::startConnectionInLoop()
|
||||
{
|
||||
loop_->assertInLoopThread();
|
||||
assert(!redisContext_);
|
||||
|
||||
redisContext_ =
|
||||
::redisAsyncConnect(serverAddr_.toIp().c_str(), serverAddr_.toPort());
|
||||
status_ = ConnectStatus::kConnecting;
|
||||
if (redisContext_->err)
|
||||
{
|
||||
LOG_ERROR << "Error: " << redisContext_->errstr;
|
||||
|
||||
if (disconnectCallback_)
|
||||
{
|
||||
disconnectCallback_(shared_from_this());
|
||||
}
|
||||
|
||||
// Strange things have happened. In some kinds of connection errors,
|
||||
// such as setsockopt errors, hiredis already set redisContext_->c.fd to
|
||||
// -1, but the tcp connection stays in ESTABLISHED status. And there is
|
||||
// no way for us to obtain the fd of that socket nor close it. This
|
||||
// probably is a bug of hiredis.
|
||||
return;
|
||||
}
|
||||
|
||||
redisContext_->ev.addWrite = addWrite;
|
||||
redisContext_->ev.delWrite = delWrite;
|
||||
redisContext_->ev.addRead = addRead;
|
||||
redisContext_->ev.delRead = delRead;
|
||||
redisContext_->ev.cleanup = cleanup;
|
||||
redisContext_->ev.data = this;
|
||||
|
||||
channel_ = std::make_unique<trantor::Channel>(loop_, redisContext_->c.fd);
|
||||
channel_->setReadCallback([this]() { handleRedisRead(); });
|
||||
channel_->setWriteCallback([this]() { handleRedisWrite(); });
|
||||
redisAsyncSetConnectCallback(
|
||||
redisContext_, [](const redisAsyncContext *context, int status) {
|
||||
auto thisPtr = static_cast<RedisConnection *>(context->ev.data);
|
||||
if (status != REDIS_OK)
|
||||
{
|
||||
LOG_ERROR << "Failed to connect to "
|
||||
<< thisPtr->serverAddr_.toIpPort() << "! "
|
||||
<< context->errstr;
|
||||
thisPtr->handleDisconnect();
|
||||
if (thisPtr->disconnectCallback_)
|
||||
{
|
||||
thisPtr->disconnectCallback_(thisPtr->shared_from_this());
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
LOG_TRACE << "Connected successfully to "
|
||||
<< thisPtr->serverAddr_.toIpPort();
|
||||
if (thisPtr->password_.empty())
|
||||
{
|
||||
if (thisPtr->db_ == 0)
|
||||
{
|
||||
thisPtr->status_ = ConnectStatus::kConnected;
|
||||
if (thisPtr->connectCallback_)
|
||||
{
|
||||
thisPtr->connectCallback_(
|
||||
thisPtr->shared_from_this());
|
||||
}
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
if (thisPtr->username_.empty())
|
||||
{
|
||||
std::weak_ptr<RedisConnection> weakThisPtr =
|
||||
thisPtr->shared_from_this();
|
||||
thisPtr->sendCommand(
|
||||
[weakThisPtr](const RedisResult &r) {
|
||||
auto thisPtr = weakThisPtr.lock();
|
||||
if (!thisPtr)
|
||||
return;
|
||||
if (r.asString() == "OK")
|
||||
{
|
||||
if (thisPtr->db_ == 0)
|
||||
{
|
||||
thisPtr->status_ =
|
||||
ConnectStatus::kConnected;
|
||||
if (thisPtr->connectCallback_)
|
||||
thisPtr->connectCallback_(
|
||||
thisPtr->shared_from_this());
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
LOG_ERROR << r.asString();
|
||||
thisPtr->disconnect();
|
||||
thisPtr->status_ = ConnectStatus::kEnd;
|
||||
}
|
||||
},
|
||||
[weakThisPtr](const std::exception &err) {
|
||||
LOG_ERROR << err.what();
|
||||
auto thisPtr = weakThisPtr.lock();
|
||||
if (!thisPtr)
|
||||
return;
|
||||
thisPtr->disconnect();
|
||||
thisPtr->status_ = ConnectStatus::kEnd;
|
||||
},
|
||||
"auth %s",
|
||||
thisPtr->password_.c_str());
|
||||
}
|
||||
else
|
||||
{
|
||||
std::weak_ptr<RedisConnection> weakThisPtr =
|
||||
thisPtr->shared_from_this();
|
||||
thisPtr->sendCommand(
|
||||
[weakThisPtr](const RedisResult &r) {
|
||||
auto thisPtr = weakThisPtr.lock();
|
||||
if (!thisPtr)
|
||||
return;
|
||||
if (r.asString() == "OK")
|
||||
{
|
||||
if (thisPtr->db_ == 0)
|
||||
{
|
||||
thisPtr->status_ =
|
||||
ConnectStatus::kConnected;
|
||||
if (thisPtr->connectCallback_)
|
||||
thisPtr->connectCallback_(
|
||||
thisPtr->shared_from_this());
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
LOG_ERROR << r.asString();
|
||||
thisPtr->disconnect();
|
||||
thisPtr->status_ = ConnectStatus::kEnd;
|
||||
}
|
||||
},
|
||||
[weakThisPtr](const std::exception &err) {
|
||||
LOG_ERROR << err.what();
|
||||
auto thisPtr = weakThisPtr.lock();
|
||||
if (!thisPtr)
|
||||
return;
|
||||
thisPtr->disconnect();
|
||||
thisPtr->status_ = ConnectStatus::kEnd;
|
||||
},
|
||||
"auth %s %s",
|
||||
thisPtr->username_.c_str(),
|
||||
thisPtr->password_.c_str());
|
||||
}
|
||||
}
|
||||
|
||||
if (thisPtr->db_ != 0)
|
||||
{
|
||||
LOG_TRACE << "redis db:" << thisPtr->db_;
|
||||
std::weak_ptr<RedisConnection> weakThisPtr =
|
||||
thisPtr->shared_from_this();
|
||||
thisPtr->sendCommand(
|
||||
[weakThisPtr](const RedisResult &r) {
|
||||
auto thisPtr = weakThisPtr.lock();
|
||||
if (!thisPtr)
|
||||
return;
|
||||
if (r.asString() == "OK")
|
||||
{
|
||||
thisPtr->status_ = ConnectStatus::kConnected;
|
||||
if (thisPtr->connectCallback_)
|
||||
{
|
||||
thisPtr->connectCallback_(
|
||||
thisPtr->shared_from_this());
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
LOG_ERROR << r.asString();
|
||||
thisPtr->disconnect();
|
||||
thisPtr->status_ = ConnectStatus::kEnd;
|
||||
}
|
||||
},
|
||||
[weakThisPtr](const std::exception &err) {
|
||||
LOG_ERROR << err.what();
|
||||
auto thisPtr = weakThisPtr.lock();
|
||||
if (!thisPtr)
|
||||
return;
|
||||
thisPtr->disconnect();
|
||||
thisPtr->status_ = ConnectStatus::kEnd;
|
||||
},
|
||||
"select %u",
|
||||
thisPtr->db_);
|
||||
}
|
||||
}
|
||||
});
|
||||
redisAsyncSetDisconnectCallback(
|
||||
redisContext_, [](const redisAsyncContext *context, int /*status*/) {
|
||||
auto thisPtr = static_cast<RedisConnection *>(context->ev.data);
|
||||
|
||||
thisPtr->handleDisconnect();
|
||||
if (thisPtr->disconnectCallback_)
|
||||
{
|
||||
thisPtr->disconnectCallback_(thisPtr->shared_from_this());
|
||||
}
|
||||
|
||||
LOG_TRACE << "Disconnected from "
|
||||
<< thisPtr->serverAddr_.toIpPort();
|
||||
});
|
||||
}
|
||||
|
||||
void RedisConnection::handleDisconnect()
|
||||
{
|
||||
LOG_TRACE << "handleDisconnect";
|
||||
loop_->assertInLoopThread();
|
||||
while ((!resultCallbacks_.empty()) && (!exceptionCallbacks_.empty()))
|
||||
{
|
||||
if (exceptionCallbacks_.front())
|
||||
{
|
||||
exceptionCallbacks_.front()(
|
||||
RedisException(RedisErrorCode::kConnectionBroken,
|
||||
"Connection is broken"));
|
||||
}
|
||||
resultCallbacks_.pop();
|
||||
exceptionCallbacks_.pop();
|
||||
}
|
||||
status_ = ConnectStatus::kEnd;
|
||||
channel_->disableAll();
|
||||
channel_->remove();
|
||||
redisContext_->ev.addWrite = nullptr;
|
||||
redisContext_->ev.delWrite = nullptr;
|
||||
redisContext_->ev.addRead = nullptr;
|
||||
redisContext_->ev.delRead = nullptr;
|
||||
redisContext_->ev.cleanup = nullptr;
|
||||
redisContext_->ev.data = nullptr;
|
||||
}
|
||||
|
||||
void RedisConnection::addWrite(void *userData)
|
||||
{
|
||||
auto thisPtr = static_cast<RedisConnection *>(userData);
|
||||
assert(thisPtr->channel_);
|
||||
thisPtr->channel_->enableWriting();
|
||||
}
|
||||
|
||||
void RedisConnection::delWrite(void *userData)
|
||||
{
|
||||
auto thisPtr = static_cast<RedisConnection *>(userData);
|
||||
assert(thisPtr->channel_);
|
||||
thisPtr->channel_->disableWriting();
|
||||
}
|
||||
|
||||
void RedisConnection::addRead(void *userData)
|
||||
{
|
||||
auto thisPtr = static_cast<RedisConnection *>(userData);
|
||||
assert(thisPtr->channel_);
|
||||
thisPtr->channel_->enableReading();
|
||||
}
|
||||
|
||||
void RedisConnection::delRead(void *userData)
|
||||
{
|
||||
auto thisPtr = static_cast<RedisConnection *>(userData);
|
||||
assert(thisPtr->channel_);
|
||||
thisPtr->channel_->disableReading();
|
||||
}
|
||||
|
||||
void RedisConnection::cleanup(void * /*userData*/)
|
||||
{
|
||||
LOG_TRACE << "cleanup";
|
||||
}
|
||||
|
||||
void RedisConnection::handleRedisRead()
|
||||
{
|
||||
if (status_ != ConnectStatus::kEnd)
|
||||
{
|
||||
redisAsyncHandleRead(redisContext_);
|
||||
}
|
||||
}
|
||||
|
||||
void RedisConnection::handleRedisWrite()
|
||||
{
|
||||
if (status_ != ConnectStatus::kEnd)
|
||||
{
|
||||
redisAsyncHandleWrite(redisContext_);
|
||||
}
|
||||
}
|
||||
|
||||
void RedisConnection::sendCommandInLoop(
|
||||
const std::string &command,
|
||||
RedisResultCallback &&resultCallback,
|
||||
RedisExceptionCallback &&exceptionCallback)
|
||||
{
|
||||
resultCallbacks_.emplace(std::move(resultCallback));
|
||||
exceptionCallbacks_.emplace(std::move(exceptionCallback));
|
||||
|
||||
redisAsyncFormattedCommand(
|
||||
redisContext_,
|
||||
[](redisAsyncContext *context, void *r, void * /*userData*/) {
|
||||
auto thisPtr = static_cast<RedisConnection *>(context->ev.data);
|
||||
thisPtr->handleResult(static_cast<redisReply *>(r));
|
||||
},
|
||||
nullptr,
|
||||
command.c_str(),
|
||||
command.length());
|
||||
}
|
||||
|
||||
void RedisConnection::handleResult(redisReply *result)
|
||||
{
|
||||
auto commandCallback = std::move(resultCallbacks_.front());
|
||||
resultCallbacks_.pop();
|
||||
auto exceptionCallback = std::move(exceptionCallbacks_.front());
|
||||
exceptionCallbacks_.pop();
|
||||
if (result && result->type != REDIS_REPLY_ERROR)
|
||||
{
|
||||
commandCallback(RedisResult(result));
|
||||
}
|
||||
else
|
||||
{
|
||||
if (result)
|
||||
{
|
||||
exceptionCallback(
|
||||
RedisException(RedisErrorCode::kRedisError,
|
||||
std::string{result->str, result->len}));
|
||||
}
|
||||
else
|
||||
{
|
||||
exceptionCallback(RedisException(RedisErrorCode::kConnectionBroken,
|
||||
"Network failure"));
|
||||
}
|
||||
}
|
||||
if (resultCallbacks_.empty())
|
||||
{
|
||||
assert(exceptionCallbacks_.empty());
|
||||
if (idleCallback_)
|
||||
{
|
||||
idleCallback_(shared_from_this());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void RedisConnection::disconnect()
|
||||
{
|
||||
auto thisPtr = shared_from_this();
|
||||
loop_->queueInLoop(
|
||||
[thisPtr]() { redisAsyncDisconnect(thisPtr->redisContext_); });
|
||||
}
|
||||
|
||||
void RedisConnection::sendSubscribe(
|
||||
const std::shared_ptr<SubscribeContext> &subCtx)
|
||||
{
|
||||
if (loop_->isInLoopThread())
|
||||
{
|
||||
sendSubscribeInLoop(subCtx);
|
||||
}
|
||||
else
|
||||
{
|
||||
loop_->queueInLoop([this, subCtx]() { sendSubscribeInLoop(subCtx); });
|
||||
}
|
||||
}
|
||||
|
||||
void RedisConnection::sendUnsubscribe(
|
||||
const std::shared_ptr<SubscribeContext> &subCtx)
|
||||
{
|
||||
if (loop_->isInLoopThread())
|
||||
{
|
||||
sendUnsubscribeInLoop(subCtx);
|
||||
}
|
||||
else
|
||||
{
|
||||
loop_->queueInLoop([this, subCtx]() { sendUnsubscribeInLoop(subCtx); });
|
||||
}
|
||||
}
|
||||
|
||||
void RedisConnection::sendSubscribeInLoop(
|
||||
const std::shared_ptr<SubscribeContext> &subCtx)
|
||||
{
|
||||
if (!subCtx->alive())
|
||||
{
|
||||
// Unsub-ed by somewhere else
|
||||
return;
|
||||
}
|
||||
subContexts_.emplace(subCtx->contextId(), subCtx);
|
||||
redisAsyncFormattedCommand(
|
||||
redisContext_,
|
||||
[](redisAsyncContext *context, void *r, void *subCtx) {
|
||||
auto thisPtr = static_cast<RedisConnection *>(context->ev.data);
|
||||
thisPtr->handleSubscribeResult(static_cast<redisReply *>(r),
|
||||
static_cast<SubscribeContext *>(
|
||||
subCtx));
|
||||
},
|
||||
subCtx.get(),
|
||||
subCtx->subscribeCommand().c_str(),
|
||||
subCtx->subscribeCommand().size());
|
||||
}
|
||||
|
||||
void RedisConnection::sendUnsubscribeInLoop(
|
||||
const std::shared_ptr<SubscribeContext> &subCtx)
|
||||
{
|
||||
// There is a Hiredis issue here
|
||||
// The un-sub callback will not be called, sub callback will be called
|
||||
// instead, with first element in result as "unsubscribe".
|
||||
// This problem is fixed in 2021-12-02 commit da5a4ff, but
|
||||
// have not been released as a tag.
|
||||
// Here we just register a same function to deal with both situation.
|
||||
redisAsyncFormattedCommand(
|
||||
redisContext_,
|
||||
[](redisAsyncContext *context, void *r, void *subCtx) {
|
||||
auto thisPtr = static_cast<RedisConnection *>(context->ev.data);
|
||||
thisPtr->handleSubscribeResult(static_cast<redisReply *>(r),
|
||||
static_cast<SubscribeContext *>(
|
||||
subCtx));
|
||||
},
|
||||
subCtx.get(),
|
||||
subCtx->unsubscribeCommand().c_str(),
|
||||
subCtx->unsubscribeCommand().size());
|
||||
}
|
||||
|
||||
void RedisConnection::handleSubscribeResult(redisReply *result,
|
||||
SubscribeContext *subCtx)
|
||||
{
|
||||
if (result && result->type == REDIS_REPLY_ARRAY && result->elements >= 3 &&
|
||||
result->element[0]->type == REDIS_REPLY_STRING)
|
||||
{
|
||||
const char *type = result->element[0]->str;
|
||||
int isPattern = (type[0] == 'p' || type[0] == 'P') ? 1 : 0;
|
||||
if (isPattern)
|
||||
{
|
||||
type += 1;
|
||||
}
|
||||
if (strcasecmp(type, "message") == 0)
|
||||
{
|
||||
std::string channel(result->element[1 + isPattern]->str,
|
||||
result->element[1 + isPattern]->len);
|
||||
std::string message(result->element[2 + isPattern]->str,
|
||||
result->element[2 + isPattern]->len);
|
||||
if (!subCtx->alive())
|
||||
{
|
||||
LOG_DEBUG << "Subscribe callback receive message, but "
|
||||
"context is no "
|
||||
"longer alive"
|
||||
<< ", channel: " << channel
|
||||
<< ", message: " << message;
|
||||
}
|
||||
else
|
||||
{
|
||||
subCtx->onMessage(channel, message);
|
||||
}
|
||||
// Message callback, no need to call idleCallback_
|
||||
return;
|
||||
}
|
||||
|
||||
std::string channel(result->element[1]->str, result->element[1]->len);
|
||||
long long number = result->element[2]->integer;
|
||||
|
||||
// On channel subscribed
|
||||
if (strcasecmp(type, "subscribe") == 0)
|
||||
{
|
||||
subCtx->onSubscribe(channel, number);
|
||||
}
|
||||
// On channel unsubscribed
|
||||
else if (strcasecmp(type, "unsubscribe") == 0)
|
||||
{
|
||||
subCtx->onUnsubscribe(channel, number);
|
||||
subContexts_.erase(subCtx->contextId());
|
||||
}
|
||||
// Should not happen
|
||||
else
|
||||
{
|
||||
LOG_ERROR << "Unknown redis response: " << result->element[0]->str;
|
||||
// Shouldn't let message from another endpoint to abort this
|
||||
// program. So no assert(false) here.
|
||||
}
|
||||
}
|
||||
else if (!result)
|
||||
{
|
||||
// When connection close, if a channel has been subscribed,
|
||||
// this callback will be called with empty result.
|
||||
LOG_DEBUG << "Empty result (connection lost)";
|
||||
}
|
||||
else if (result->type == REDIS_REPLY_ERROR)
|
||||
{
|
||||
LOG_ERROR << "Subscribe callback receive error result: " << result->str;
|
||||
}
|
||||
else
|
||||
{
|
||||
LOG_ERROR << "Subscribe callback receive error result type: "
|
||||
<< result->type;
|
||||
}
|
||||
|
||||
if (idleCallback_)
|
||||
{
|
||||
idleCallback_(shared_from_this());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,229 @@
|
||||
/**
|
||||
*
|
||||
* @file RedisConnection.h
|
||||
* @author An Tao
|
||||
*
|
||||
* Copyright 2018, An Tao. All rights reserved.
|
||||
* https://github.com/an-tao/drogon
|
||||
* Use of this source code is governed by a MIT license
|
||||
* that can be found in the License file.
|
||||
*
|
||||
* Drogon
|
||||
*
|
||||
*/
|
||||
|
||||
#pragma once
|
||||
#include <string_view>
|
||||
#include <drogon/nosql/RedisException.h>
|
||||
#include <drogon/nosql/RedisResult.h>
|
||||
#include <drogon/utils/Utilities.h>
|
||||
#include <trantor/utils/NonCopyable.h>
|
||||
#include <trantor/net/InetAddress.h>
|
||||
#include <trantor/net/EventLoop.h>
|
||||
#include <trantor/net/Channel.h>
|
||||
#include <hiredis/async.h>
|
||||
#include <hiredis/hiredis.h>
|
||||
#include <memory>
|
||||
#include <queue>
|
||||
|
||||
#include "SubscribeContext.h"
|
||||
|
||||
namespace drogon
|
||||
{
|
||||
namespace nosql
|
||||
{
|
||||
enum class ConnectStatus
|
||||
{
|
||||
kNone = 0,
|
||||
kConnecting,
|
||||
kConnected,
|
||||
kEnd
|
||||
};
|
||||
|
||||
class RedisConnection : public trantor::NonCopyable,
|
||||
public std::enable_shared_from_this<RedisConnection>
|
||||
{
|
||||
public:
|
||||
RedisConnection(const trantor::InetAddress &serverAddress,
|
||||
const std::string &username,
|
||||
const std::string &password,
|
||||
unsigned int db,
|
||||
trantor::EventLoop *loop);
|
||||
|
||||
void setConnectCallback(
|
||||
const std::function<void(std::shared_ptr<RedisConnection> &&)>
|
||||
&callback)
|
||||
{
|
||||
connectCallback_ = callback;
|
||||
}
|
||||
|
||||
void setDisconnectCallback(
|
||||
const std::function<void(std::shared_ptr<RedisConnection> &&)>
|
||||
&callback)
|
||||
{
|
||||
disconnectCallback_ = callback;
|
||||
}
|
||||
|
||||
void setIdleCallback(
|
||||
const std::function<void(const std::shared_ptr<RedisConnection> &)>
|
||||
&callback)
|
||||
{
|
||||
idleCallback_ = callback;
|
||||
}
|
||||
|
||||
static std::string getFormattedCommand(const std::string_view &command,
|
||||
va_list ap) noexcept(false)
|
||||
{
|
||||
char *cmd{nullptr};
|
||||
auto len = redisvFormatCommand(&cmd, command.data(), ap);
|
||||
if (len == -1)
|
||||
{
|
||||
throw RedisException(RedisErrorCode::kInternalError,
|
||||
"Out of memory");
|
||||
}
|
||||
else if (len == -2)
|
||||
{
|
||||
throw RedisException(RedisErrorCode::kInternalError,
|
||||
"Invalid format string");
|
||||
}
|
||||
else if (len <= 0)
|
||||
{
|
||||
throw RedisException(RedisErrorCode::kInternalError,
|
||||
"Unknown format error");
|
||||
}
|
||||
std::string fullCommand{cmd, static_cast<size_t>(len)};
|
||||
redisFreeCommand(cmd);
|
||||
return fullCommand;
|
||||
}
|
||||
|
||||
void sendFormattedCommand(std::string &&command,
|
||||
RedisResultCallback &&resultCallback,
|
||||
RedisExceptionCallback &&exceptionCallback)
|
||||
{
|
||||
if (loop_->isInLoopThread())
|
||||
{
|
||||
sendCommandInLoop(command,
|
||||
std::move(resultCallback),
|
||||
std::move(exceptionCallback));
|
||||
}
|
||||
else
|
||||
{
|
||||
loop_->queueInLoop(
|
||||
[this,
|
||||
callback = std::move(resultCallback),
|
||||
exceptionCallback = std::move(exceptionCallback),
|
||||
command = std::move(command)]() mutable {
|
||||
sendCommandInLoop(command,
|
||||
std::move(callback),
|
||||
std::move(exceptionCallback));
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
void sendvCommand(std::string_view command,
|
||||
RedisResultCallback &&resultCallback,
|
||||
RedisExceptionCallback &&exceptionCallback,
|
||||
va_list ap)
|
||||
{
|
||||
LOG_TRACE << "redis command: " << command;
|
||||
try
|
||||
{
|
||||
auto fullCommand = getFormattedCommand(command, ap);
|
||||
if (loop_->isInLoopThread())
|
||||
{
|
||||
sendCommandInLoop(fullCommand,
|
||||
std::move(resultCallback),
|
||||
std::move(exceptionCallback));
|
||||
}
|
||||
else
|
||||
{
|
||||
loop_->queueInLoop(
|
||||
[this,
|
||||
callback = std::move(resultCallback),
|
||||
exceptionCallback = std::move(exceptionCallback),
|
||||
fullCommand = std::move(fullCommand)]() mutable {
|
||||
sendCommandInLoop(fullCommand,
|
||||
std::move(callback),
|
||||
std::move(exceptionCallback));
|
||||
});
|
||||
}
|
||||
}
|
||||
catch (const RedisException &err)
|
||||
{
|
||||
exceptionCallback(err);
|
||||
}
|
||||
}
|
||||
|
||||
void sendSubscribe(const std::shared_ptr<SubscribeContext> &subCtx);
|
||||
void sendUnsubscribe(const std::shared_ptr<SubscribeContext> &subCtx);
|
||||
|
||||
~RedisConnection()
|
||||
{
|
||||
LOG_TRACE << (int)status_;
|
||||
if (redisContext_ && status_ != ConnectStatus::kEnd)
|
||||
redisAsyncDisconnect(redisContext_);
|
||||
}
|
||||
|
||||
void disconnect();
|
||||
|
||||
void sendCommand(RedisResultCallback &&resultCallback,
|
||||
RedisExceptionCallback &&exceptionCallback,
|
||||
std::string_view command,
|
||||
...)
|
||||
{
|
||||
va_list args;
|
||||
va_start(args, command);
|
||||
sendvCommand(command,
|
||||
std::move(resultCallback),
|
||||
std::move(exceptionCallback),
|
||||
args);
|
||||
va_end(args);
|
||||
}
|
||||
|
||||
trantor::EventLoop *getLoop() const
|
||||
{
|
||||
return loop_;
|
||||
}
|
||||
|
||||
private:
|
||||
redisAsyncContext *redisContext_{nullptr};
|
||||
const trantor::InetAddress serverAddr_;
|
||||
const std::string username_;
|
||||
const std::string password_;
|
||||
const unsigned int db_;
|
||||
trantor::EventLoop *loop_{nullptr};
|
||||
std::unique_ptr<trantor::Channel> channel_{nullptr};
|
||||
std::function<void(std::shared_ptr<RedisConnection> &&)> connectCallback_;
|
||||
std::function<void(std::shared_ptr<RedisConnection> &&)>
|
||||
disconnectCallback_;
|
||||
std::function<void(const std::shared_ptr<RedisConnection> &)> idleCallback_;
|
||||
std::queue<RedisResultCallback> resultCallbacks_;
|
||||
std::queue<RedisExceptionCallback> exceptionCallbacks_;
|
||||
ConnectStatus status_{ConnectStatus::kNone};
|
||||
|
||||
// used to keep the lifetime of context object
|
||||
std::unordered_map<unsigned long long, std::shared_ptr<SubscribeContext>>
|
||||
subContexts_;
|
||||
|
||||
void startConnectionInLoop();
|
||||
static void addWrite(void *userData);
|
||||
static void delWrite(void *userData);
|
||||
static void addRead(void *userData);
|
||||
static void delRead(void *userData);
|
||||
static void cleanup(void *userData);
|
||||
void handleRedisRead();
|
||||
void handleRedisWrite();
|
||||
void handleResult(redisReply *result);
|
||||
void sendCommandInLoop(const std::string &command,
|
||||
RedisResultCallback &&resultCallback,
|
||||
RedisExceptionCallback &&exceptionCallback);
|
||||
void sendSubscribeInLoop(const std::shared_ptr<SubscribeContext> &subCtx);
|
||||
void sendUnsubscribeInLoop(const std::shared_ptr<SubscribeContext> &subCtx);
|
||||
void handleSubscribeResult(redisReply *result, SubscribeContext *subCtx);
|
||||
|
||||
void handleDisconnect();
|
||||
};
|
||||
|
||||
using RedisConnectionPtr = std::shared_ptr<RedisConnection>;
|
||||
} // namespace nosql
|
||||
} // namespace drogon
|
||||
@@ -0,0 +1,129 @@
|
||||
/**
|
||||
*
|
||||
* @file RedisResult.cc
|
||||
* @author An Tao
|
||||
*
|
||||
* Copyright 2018, An Tao. All rights reserved.
|
||||
* https://github.com/an-tao/drogon
|
||||
* Use of this source code is governed by a MIT license
|
||||
* that can be found in the License file.
|
||||
*
|
||||
* Drogon
|
||||
*
|
||||
*/
|
||||
|
||||
#include <drogon/nosql/RedisResult.h>
|
||||
#include <drogon/nosql/RedisClient.h>
|
||||
#include <hiredis/hiredis.h>
|
||||
|
||||
using namespace drogon::nosql;
|
||||
|
||||
std::string RedisResult::getStringForDisplaying() const noexcept
|
||||
{
|
||||
return getStringForDisplayingWithIndent(0);
|
||||
}
|
||||
|
||||
std::string RedisResult::getStringForDisplayingWithIndent(
|
||||
size_t indent) const noexcept
|
||||
{
|
||||
switch (result_->type)
|
||||
{
|
||||
case REDIS_REPLY_STRING:
|
||||
return "\"" + std::string{result_->str, result_->len} + "\"";
|
||||
case REDIS_REPLY_STATUS:
|
||||
return std::string{result_->str, result_->len};
|
||||
case REDIS_REPLY_ERROR:
|
||||
return "'ERROR:" + std::string{result_->str, result_->len} + "'";
|
||||
case REDIS_REPLY_NIL:
|
||||
return "(nil)";
|
||||
case REDIS_REPLY_INTEGER:
|
||||
return std::to_string(result_->integer);
|
||||
case REDIS_REPLY_ARRAY:
|
||||
{
|
||||
std::string ret;
|
||||
for (size_t i = 0; i < result_->elements; ++i)
|
||||
{
|
||||
std::string lineNum = std::to_string(i + 1) + ") ";
|
||||
if (i > 0)
|
||||
{
|
||||
ret += std::string(indent, ' ');
|
||||
}
|
||||
ret += lineNum;
|
||||
ret += RedisResult(result_->element[i])
|
||||
.getStringForDisplayingWithIndent(lineNum.length());
|
||||
if (i != result_->elements - 1)
|
||||
{
|
||||
ret += '\n';
|
||||
}
|
||||
}
|
||||
return ret;
|
||||
}
|
||||
default:
|
||||
return "*";
|
||||
}
|
||||
}
|
||||
|
||||
std::string RedisResult::asString() const noexcept(false)
|
||||
{
|
||||
auto rtype = type();
|
||||
if (rtype == RedisResultType::kString ||
|
||||
rtype == RedisResultType::kStatus || rtype == RedisResultType::kError)
|
||||
{
|
||||
return std::string(result_->str, result_->len);
|
||||
}
|
||||
else if (rtype == RedisResultType::kInteger)
|
||||
{
|
||||
return std::to_string(result_->integer);
|
||||
}
|
||||
else
|
||||
{
|
||||
throw RedisException(RedisErrorCode::kBadType, "bad type");
|
||||
}
|
||||
}
|
||||
|
||||
RedisResultType RedisResult::type() const noexcept
|
||||
{
|
||||
switch (result_->type)
|
||||
{
|
||||
case REDIS_REPLY_STRING:
|
||||
return RedisResultType::kString;
|
||||
case REDIS_REPLY_ARRAY:
|
||||
return RedisResultType::kArray;
|
||||
case REDIS_REPLY_INTEGER:
|
||||
return RedisResultType::kInteger;
|
||||
case REDIS_REPLY_NIL:
|
||||
return RedisResultType::kNil;
|
||||
case REDIS_REPLY_STATUS:
|
||||
return RedisResultType::kStatus;
|
||||
case REDIS_REPLY_ERROR:
|
||||
default:
|
||||
return RedisResultType::kError;
|
||||
}
|
||||
}
|
||||
|
||||
std::vector<RedisResult> RedisResult::asArray() const noexcept(false)
|
||||
{
|
||||
auto rtype = type();
|
||||
if (rtype == RedisResultType::kArray)
|
||||
{
|
||||
std::vector<RedisResult> array;
|
||||
for (size_t i = 0; i < result_->elements; ++i)
|
||||
{
|
||||
array.emplace_back(result_->element[i]);
|
||||
}
|
||||
return array;
|
||||
}
|
||||
throw RedisException(RedisErrorCode::kBadType, "bad type");
|
||||
}
|
||||
|
||||
long long RedisResult::asInteger() const noexcept(false)
|
||||
{
|
||||
if (type() == RedisResultType::kInteger)
|
||||
return result_->integer;
|
||||
throw RedisException(RedisErrorCode::kBadType, "bad type");
|
||||
}
|
||||
|
||||
bool RedisResult::isNil() const noexcept
|
||||
{
|
||||
return type() == RedisResultType::kNil;
|
||||
}
|
||||
@@ -0,0 +1,237 @@
|
||||
/**
|
||||
*
|
||||
* @file RedisSubscriberImpl.cpp
|
||||
* @author Nitromelon
|
||||
*
|
||||
* Copyright 2022, Nitromelon. All rights reserved.
|
||||
* https://github.com/drogonframework/drogon
|
||||
* Use of this source code is governed by a MIT license
|
||||
* that can be found in the License file.
|
||||
*
|
||||
* Drogon
|
||||
*
|
||||
*/
|
||||
|
||||
#include "RedisSubscriberImpl.h"
|
||||
|
||||
using namespace drogon::nosql;
|
||||
|
||||
RedisSubscriberImpl::~RedisSubscriberImpl()
|
||||
{
|
||||
RedisConnectionPtr conn;
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
if (conn_)
|
||||
{
|
||||
conn.swap(conn_);
|
||||
conn->getLoop()->runInLoop([conn]() {
|
||||
// Run in self loop to avoid blocking
|
||||
conn->disconnect();
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
void RedisSubscriberImpl::subscribe(
|
||||
const std::string &channel,
|
||||
RedisMessageCallback &&messageCallback) noexcept
|
||||
{
|
||||
LOG_TRACE << "Subscribe " << channel;
|
||||
|
||||
std::shared_ptr<SubscribeContext> subCtx;
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
if (subContexts_.find(channel) != subContexts_.end())
|
||||
{
|
||||
subCtx = subContexts_.at(channel);
|
||||
}
|
||||
else
|
||||
{
|
||||
subCtx = SubscribeContext::newContext(shared_from_this(), channel);
|
||||
subContexts_.emplace(channel, subCtx);
|
||||
}
|
||||
subCtx->addMessageCallback(std::move(messageCallback));
|
||||
}
|
||||
|
||||
RedisConnectionPtr connPtr;
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
connPtr = conn_;
|
||||
}
|
||||
|
||||
if (connPtr)
|
||||
{
|
||||
connPtr->sendSubscribe(subCtx);
|
||||
}
|
||||
else
|
||||
{
|
||||
LOG_TRACE << "no subscribe connection available, wait for connection";
|
||||
// Just wait for connection, all channels will be re-sub
|
||||
}
|
||||
}
|
||||
|
||||
void RedisSubscriberImpl::psubscribe(
|
||||
const std::string &pattern,
|
||||
RedisMessageCallback &&messageCallback) noexcept
|
||||
{
|
||||
LOG_TRACE << "Psubscribe " << pattern;
|
||||
|
||||
std::shared_ptr<SubscribeContext> subCtx;
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
if (psubContexts_.find(pattern) != psubContexts_.end())
|
||||
{
|
||||
subCtx = psubContexts_.at(pattern);
|
||||
}
|
||||
else
|
||||
{
|
||||
subCtx =
|
||||
SubscribeContext::newContext(shared_from_this(), pattern, true);
|
||||
psubContexts_.emplace(pattern, subCtx);
|
||||
}
|
||||
subCtx->addMessageCallback(std::move(messageCallback));
|
||||
}
|
||||
|
||||
RedisConnectionPtr connPtr;
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
connPtr = conn_;
|
||||
}
|
||||
|
||||
if (connPtr)
|
||||
{
|
||||
connPtr->sendSubscribe(subCtx);
|
||||
}
|
||||
else
|
||||
{
|
||||
LOG_TRACE << "no subscribe connection available, wait for connection";
|
||||
// Just wait for connection, all channels will be re-sub
|
||||
}
|
||||
}
|
||||
|
||||
void RedisSubscriberImpl::unsubscribe(const std::string &channel) noexcept
|
||||
{
|
||||
LOG_TRACE << "Unsubscribe " << channel;
|
||||
|
||||
std::shared_ptr<SubscribeContext> subCtx;
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
auto iter = subContexts_.find(channel);
|
||||
if (iter == subContexts_.end())
|
||||
{
|
||||
LOG_DEBUG << "Attempt to unsubscribe from unknown channel "
|
||||
<< channel;
|
||||
return;
|
||||
}
|
||||
subCtx = std::move(iter->second);
|
||||
subContexts_.erase(iter);
|
||||
}
|
||||
subCtx->disable();
|
||||
|
||||
RedisConnectionPtr connPtr;
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
connPtr = conn_;
|
||||
}
|
||||
if (!connPtr)
|
||||
{
|
||||
LOG_TRACE << "Connection unavailable, no need to send unsub command";
|
||||
return;
|
||||
}
|
||||
|
||||
connPtr->sendUnsubscribe(subCtx);
|
||||
}
|
||||
|
||||
void RedisSubscriberImpl::punsubscribe(const std::string &pattern) noexcept
|
||||
{
|
||||
LOG_TRACE << "Punsubscribe " << pattern;
|
||||
|
||||
std::shared_ptr<SubscribeContext> subCtx;
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
auto iter = psubContexts_.find(pattern);
|
||||
if (iter == psubContexts_.end())
|
||||
{
|
||||
LOG_DEBUG << "Attempt to punsubscribe from unknown pattern "
|
||||
<< pattern;
|
||||
return;
|
||||
}
|
||||
subCtx = std::move(iter->second);
|
||||
psubContexts_.erase(iter);
|
||||
}
|
||||
subCtx->disable();
|
||||
|
||||
RedisConnectionPtr connPtr;
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
connPtr = conn_;
|
||||
}
|
||||
if (!connPtr)
|
||||
{
|
||||
LOG_TRACE << "Connection unavailable, no need to send unsub command";
|
||||
return;
|
||||
}
|
||||
|
||||
connPtr->sendUnsubscribe(subCtx);
|
||||
}
|
||||
|
||||
void RedisSubscriberImpl::setConnection(const RedisConnectionPtr &conn)
|
||||
{
|
||||
assert(conn);
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
assert(!conn_);
|
||||
conn_ = conn;
|
||||
}
|
||||
|
||||
void RedisSubscriberImpl::clearConnection()
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
if (conn_)
|
||||
{
|
||||
conn_ = nullptr;
|
||||
tasks_.clear();
|
||||
}
|
||||
}
|
||||
|
||||
void RedisSubscriberImpl::subscribeNext()
|
||||
{
|
||||
RedisConnectionPtr connPtr;
|
||||
std::shared_ptr<std::function<void(const RedisConnectionPtr &)>> taskPtr;
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
if (!conn_ || tasks_.empty())
|
||||
{
|
||||
return;
|
||||
}
|
||||
connPtr = conn_;
|
||||
taskPtr = std::move(tasks_.front());
|
||||
tasks_.pop_front();
|
||||
}
|
||||
(*taskPtr)(connPtr);
|
||||
}
|
||||
|
||||
void RedisSubscriberImpl::subscribeAll()
|
||||
{
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
for (auto &item : subContexts_)
|
||||
{
|
||||
auto subCtx = item.second;
|
||||
tasks_.emplace_back(
|
||||
std::make_shared<
|
||||
std::function<void(const RedisConnectionPtr &)>>(
|
||||
[subCtx](const RedisConnectionPtr &connPtr) mutable {
|
||||
connPtr->sendSubscribe(subCtx);
|
||||
}));
|
||||
}
|
||||
for (auto &item : psubContexts_)
|
||||
{
|
||||
auto subCtx = item.second;
|
||||
tasks_.emplace_back(
|
||||
std::make_shared<
|
||||
std::function<void(const RedisConnectionPtr &)>>(
|
||||
[subCtx](const RedisConnectionPtr &connPtr) mutable {
|
||||
connPtr->sendSubscribe(subCtx);
|
||||
}));
|
||||
}
|
||||
}
|
||||
subscribeNext();
|
||||
}
|
||||
@@ -0,0 +1,62 @@
|
||||
/**
|
||||
*
|
||||
* @file RedisSubscriberImpl.h
|
||||
* @author Nitromelon
|
||||
*
|
||||
* Copyright 2022, Nitromelon. All rights reserved.
|
||||
* https://github.com/drogonframework/drogon
|
||||
* Use of this source code is governed by a MIT license
|
||||
* that can be found in the License file.
|
||||
*
|
||||
* Drogon
|
||||
*
|
||||
*/
|
||||
#pragma once
|
||||
|
||||
#include <drogon/nosql/RedisSubscriber.h>
|
||||
#include "RedisConnection.h"
|
||||
#include "SubscribeContext.h"
|
||||
|
||||
#include <mutex>
|
||||
#include <unordered_map>
|
||||
#include <memory>
|
||||
#include <list>
|
||||
|
||||
namespace drogon::nosql
|
||||
{
|
||||
class RedisSubscriberImpl
|
||||
: public RedisSubscriber,
|
||||
public std::enable_shared_from_this<RedisSubscriberImpl>
|
||||
{
|
||||
public:
|
||||
~RedisSubscriberImpl() override;
|
||||
|
||||
void subscribe(const std::string &channel,
|
||||
RedisMessageCallback &&messageCallback) noexcept override;
|
||||
void psubscribe(const std::string &pattern,
|
||||
RedisMessageCallback &&messageCallback) noexcept override;
|
||||
|
||||
void unsubscribe(const std::string &channel) noexcept override;
|
||||
void punsubscribe(const std::string &pattern) noexcept override;
|
||||
|
||||
// Set a connected connection to subscriber.
|
||||
void setConnection(const RedisConnectionPtr &conn);
|
||||
// Clear connection and task queue.
|
||||
void clearConnection();
|
||||
// Subscribe next channel in task queue.
|
||||
void subscribeNext();
|
||||
// Subscribe all channels.
|
||||
void subscribeAll();
|
||||
|
||||
private:
|
||||
RedisConnectionPtr conn_;
|
||||
std::unordered_map<std::string, std::shared_ptr<SubscribeContext>>
|
||||
subContexts_;
|
||||
std::unordered_map<std::string, std::shared_ptr<SubscribeContext>>
|
||||
psubContexts_;
|
||||
std::list<std::shared_ptr<std::function<void(const RedisConnectionPtr &)>>>
|
||||
tasks_;
|
||||
std::mutex mutex_;
|
||||
};
|
||||
|
||||
} // namespace drogon::nosql
|
||||
@@ -0,0 +1,128 @@
|
||||
/**
|
||||
*
|
||||
* @file RedisConnection.cc
|
||||
* @author An Tao
|
||||
*
|
||||
* Copyright 2018, An Tao. All rights reserved.
|
||||
* https://github.com/an-tao/drogon
|
||||
* Use of this source code is governed by a MIT license
|
||||
* that can be found in the License file.
|
||||
*
|
||||
* Drogon
|
||||
*
|
||||
*/
|
||||
|
||||
#include "RedisTransactionImpl.h"
|
||||
#include "../../lib/src/TaskTimeoutFlag.h"
|
||||
using namespace drogon::nosql;
|
||||
|
||||
RedisTransactionImpl::RedisTransactionImpl(RedisConnectionPtr connPtr) noexcept
|
||||
: connPtr_(std::move(connPtr))
|
||||
{
|
||||
}
|
||||
|
||||
void RedisTransactionImpl::execute(RedisResultCallback &&resultCallback,
|
||||
RedisExceptionCallback &&exceptionCallback)
|
||||
{
|
||||
execCommandAsync(
|
||||
[thisPtr = shared_from_this(),
|
||||
resultCallback =
|
||||
std::move(resultCallback)](const RedisResult &result) {
|
||||
thisPtr->isExecutedOrCancelled_ = true;
|
||||
resultCallback(result);
|
||||
},
|
||||
std::move(exceptionCallback),
|
||||
"EXEC");
|
||||
}
|
||||
|
||||
void RedisTransactionImpl::execCommandAsync(
|
||||
RedisResultCallback &&resultCallback,
|
||||
RedisExceptionCallback &&exceptionCallback,
|
||||
std::string_view command,
|
||||
...) noexcept
|
||||
{
|
||||
if (isExecutedOrCancelled_)
|
||||
{
|
||||
exceptionCallback(RedisException(RedisErrorCode::kTransactionCancelled,
|
||||
"Transaction was cancelled"));
|
||||
return;
|
||||
}
|
||||
if (timeout_ <= 0.0)
|
||||
{
|
||||
va_list args;
|
||||
va_start(args, command);
|
||||
connPtr_->sendvCommand(
|
||||
command,
|
||||
std::move(resultCallback),
|
||||
[thisPtr = shared_from_this(),
|
||||
exceptionCallback =
|
||||
std::move(exceptionCallback)](const RedisException &err) {
|
||||
LOG_ERROR << err.what();
|
||||
thisPtr->isExecutedOrCancelled_ = true;
|
||||
exceptionCallback(err);
|
||||
},
|
||||
args);
|
||||
va_end(args);
|
||||
}
|
||||
else
|
||||
{
|
||||
auto expCbPtr = std::make_shared<RedisExceptionCallback>(
|
||||
std::move(exceptionCallback));
|
||||
auto timeoutFlagPtr = std::make_shared<TaskTimeoutFlag>(
|
||||
connPtr_->getLoop(),
|
||||
std::chrono::duration<double>(timeout_),
|
||||
[expCbPtr]() {
|
||||
if (*expCbPtr)
|
||||
{
|
||||
(*expCbPtr)(RedisException(RedisErrorCode::kTimeout,
|
||||
"Command execution timeout"));
|
||||
}
|
||||
});
|
||||
va_list args;
|
||||
va_start(args, command);
|
||||
connPtr_->sendvCommand(
|
||||
command,
|
||||
[resultCallback = std::move(resultCallback),
|
||||
timeoutFlagPtr](const RedisResult &result) {
|
||||
if (timeoutFlagPtr->done())
|
||||
{
|
||||
return;
|
||||
}
|
||||
resultCallback(result);
|
||||
},
|
||||
[thisPtr = shared_from_this(), expCbPtr, timeoutFlagPtr](
|
||||
const RedisException &err) {
|
||||
if (timeoutFlagPtr->done())
|
||||
{
|
||||
return;
|
||||
}
|
||||
LOG_ERROR << err.what();
|
||||
thisPtr->isExecutedOrCancelled_ = true;
|
||||
if (*expCbPtr)
|
||||
(*expCbPtr)(err);
|
||||
},
|
||||
args);
|
||||
va_end(args);
|
||||
timeoutFlagPtr->runTimer();
|
||||
}
|
||||
}
|
||||
|
||||
void RedisTransactionImpl::doBegin()
|
||||
{
|
||||
assert(!isExecutedOrCancelled_);
|
||||
execCommandAsync([](const RedisResult & /*result*/) {},
|
||||
[](const RedisException & /*err*/) {},
|
||||
"MULTI");
|
||||
}
|
||||
|
||||
RedisTransactionImpl::~RedisTransactionImpl()
|
||||
{
|
||||
if (!isExecutedOrCancelled_)
|
||||
{
|
||||
LOG_WARN << "The transaction is not executed before being destroyed";
|
||||
connPtr_->sendCommand([](const RedisResult & /*result*/) {},
|
||||
[](const RedisException & /*err*/) {},
|
||||
"DISCARD");
|
||||
}
|
||||
LOG_TRACE << "transaction is destroyed";
|
||||
}
|
||||
@@ -0,0 +1,71 @@
|
||||
/**
|
||||
*
|
||||
* @file RedisConnection.h
|
||||
* @author An Tao
|
||||
*
|
||||
* Copyright 2018, An Tao. All rights reserved.
|
||||
* https://github.com/an-tao/drogon
|
||||
* Use of this source code is governed by a MIT license
|
||||
* that can be found in the License file.
|
||||
*
|
||||
* Drogon
|
||||
*
|
||||
*/
|
||||
#pragma once
|
||||
|
||||
#include "RedisConnection.h"
|
||||
#include <drogon/nosql/RedisClient.h>
|
||||
#include <memory>
|
||||
|
||||
namespace drogon
|
||||
{
|
||||
namespace nosql
|
||||
{
|
||||
class RedisTransactionImpl final
|
||||
: public RedisTransaction,
|
||||
public std::enable_shared_from_this<RedisTransactionImpl>
|
||||
{
|
||||
public:
|
||||
explicit RedisTransactionImpl(RedisConnectionPtr connection) noexcept;
|
||||
// virtual void cancel() override;
|
||||
void execute(RedisResultCallback &&resultCallback,
|
||||
RedisExceptionCallback &&exceptionCallback) override;
|
||||
void execCommandAsync(RedisResultCallback &&resultCallback,
|
||||
RedisExceptionCallback &&exceptionCallback,
|
||||
std::string_view command,
|
||||
...) noexcept override;
|
||||
|
||||
std::shared_ptr<RedisSubscriber> newSubscriber() noexcept override
|
||||
{
|
||||
LOG_ERROR << "You can't create subscriber from redis transaction";
|
||||
assert(0);
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
std::shared_ptr<RedisTransaction> newTransaction() override
|
||||
{
|
||||
return shared_from_this();
|
||||
}
|
||||
|
||||
void newTransactionAsync(
|
||||
const std::function<void(const std::shared_ptr<RedisTransaction> &)>
|
||||
&callback) override
|
||||
{
|
||||
callback(shared_from_this());
|
||||
}
|
||||
|
||||
void setTimeout(double timeout) override
|
||||
{
|
||||
timeout_ = timeout;
|
||||
}
|
||||
|
||||
void doBegin();
|
||||
~RedisTransactionImpl() override;
|
||||
|
||||
private:
|
||||
bool isExecutedOrCancelled_{false};
|
||||
RedisConnectionPtr connPtr_;
|
||||
double timeout_{-1.0};
|
||||
};
|
||||
} // namespace nosql
|
||||
} // namespace drogon
|
||||
@@ -0,0 +1,136 @@
|
||||
/**
|
||||
*
|
||||
* @file SubscribeContext.cc
|
||||
* @author Nitromelon
|
||||
*
|
||||
* Copyright 2022, Nitromelon. All rights reserved.
|
||||
* https://github.com/drogonframework/drogon
|
||||
* Use of this source code is governed by a MIT license
|
||||
* that can be found in the License file.
|
||||
*
|
||||
* Drogon
|
||||
*
|
||||
*/
|
||||
#include "SubscribeContext.h"
|
||||
#include <stdio.h>
|
||||
#include <string>
|
||||
#include <utility>
|
||||
|
||||
using namespace drogon::nosql;
|
||||
|
||||
std::atomic<unsigned long long> SubscribeContext::maxContextId_{0};
|
||||
|
||||
enum class SubCommandType
|
||||
{
|
||||
Subscribe,
|
||||
Unsubscribe,
|
||||
Psubscribe,
|
||||
Punsubscribe
|
||||
};
|
||||
|
||||
static std::string formatSubscribeCommand(const std::string &channel,
|
||||
SubCommandType type)
|
||||
{
|
||||
// Avoid using redisvFormatCommand, we don't want to emit unknown error
|
||||
static const char *redisSubFmt =
|
||||
"*2\r\n$9\r\nsubscribe\r\n$%zu\r\n%.*s\r\n";
|
||||
static const char *redisUnsubFmt =
|
||||
"*2\r\n$11\r\nunsubscribe\r\n$%zu\r\n%.*s\r\n";
|
||||
static const char *redisPsubFmt =
|
||||
"*2\r\n$10\r\npsubscribe\r\n$%zu\r\n%.*s\r\n";
|
||||
static const char *redisPunsubFmt =
|
||||
"*2\r\n$12\r\npunsubscribe\r\n$%zu\r\n%.*s\r\n";
|
||||
|
||||
const char *fmt;
|
||||
switch (type)
|
||||
{
|
||||
case SubCommandType::Subscribe:
|
||||
fmt = redisSubFmt;
|
||||
break;
|
||||
case SubCommandType::Unsubscribe:
|
||||
fmt = redisUnsubFmt;
|
||||
break;
|
||||
case SubCommandType::Psubscribe:
|
||||
fmt = redisPsubFmt;
|
||||
break;
|
||||
case SubCommandType::Punsubscribe:
|
||||
fmt = redisPunsubFmt;
|
||||
break;
|
||||
}
|
||||
|
||||
std::string command;
|
||||
if (channel.size() < 32)
|
||||
{
|
||||
char buf[64];
|
||||
size_t bufSize = sizeof(buf);
|
||||
int len = snprintf(buf,
|
||||
bufSize,
|
||||
fmt,
|
||||
channel.size(),
|
||||
(int)channel.size(),
|
||||
channel.c_str());
|
||||
command = std::string(buf, len);
|
||||
}
|
||||
else
|
||||
{
|
||||
size_t bufSize = channel.size() + 64;
|
||||
char *buf = static_cast<char *>(malloc(bufSize));
|
||||
int len = snprintf(buf,
|
||||
bufSize,
|
||||
fmt,
|
||||
channel.size(),
|
||||
(int)channel.size(),
|
||||
channel.c_str());
|
||||
command = std::string(buf, len);
|
||||
free(buf);
|
||||
}
|
||||
return command;
|
||||
}
|
||||
|
||||
SubscribeContext::SubscribeContext(std::weak_ptr<RedisSubscriber> &&weakSub,
|
||||
const std::string &channel,
|
||||
bool isPattern)
|
||||
: contextId_(++maxContextId_),
|
||||
weakSub_(std::move(weakSub)),
|
||||
channel_(channel),
|
||||
isPattern_(isPattern)
|
||||
{
|
||||
if (isPattern)
|
||||
{
|
||||
subscribeCommand_ =
|
||||
formatSubscribeCommand(channel, SubCommandType::Psubscribe);
|
||||
unsubscribeCommand_ =
|
||||
formatSubscribeCommand(channel, SubCommandType::Punsubscribe);
|
||||
}
|
||||
else
|
||||
{
|
||||
subscribeCommand_ =
|
||||
formatSubscribeCommand(channel, SubCommandType::Subscribe);
|
||||
unsubscribeCommand_ =
|
||||
formatSubscribeCommand(channel, SubCommandType::Unsubscribe);
|
||||
}
|
||||
}
|
||||
|
||||
void SubscribeContext::onMessage(const std::string &channel,
|
||||
const std::string &message)
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
for (auto &callback : messageCallbacks_)
|
||||
{
|
||||
callback(channel, message);
|
||||
}
|
||||
}
|
||||
|
||||
void SubscribeContext::onSubscribe(const std::string &channel,
|
||||
long long numChannels)
|
||||
{
|
||||
LOG_DEBUG << "Subscribe success to [" << channel << "], total "
|
||||
<< numChannels;
|
||||
}
|
||||
|
||||
void SubscribeContext::onUnsubscribe(const std::string &channel,
|
||||
long long numChannels)
|
||||
{
|
||||
LOG_DEBUG << "Unsubscribe success from [" << channel << "], total "
|
||||
<< numChannels;
|
||||
}
|
||||
@@ -0,0 +1,115 @@
|
||||
/**
|
||||
*
|
||||
* @file SubscribeContext.h
|
||||
* @author Nitromelon
|
||||
*
|
||||
* Copyright 2022, Nitromelon. All rights reserved.
|
||||
* https://github.com/drogonframework/drogon
|
||||
* Use of this source code is governed by a MIT license
|
||||
* that can be found in the License file.
|
||||
*
|
||||
* Drogon
|
||||
*
|
||||
*/
|
||||
#pragma once
|
||||
|
||||
#include <drogon/nosql/RedisClient.h>
|
||||
#include <list>
|
||||
#include <atomic>
|
||||
#include <mutex>
|
||||
#include <utility>
|
||||
|
||||
namespace drogon::nosql
|
||||
{
|
||||
class SubscribeContext
|
||||
{
|
||||
public:
|
||||
static std::shared_ptr<SubscribeContext> newContext(
|
||||
std::weak_ptr<RedisSubscriber> &&weakSub,
|
||||
const std::string &channel,
|
||||
bool isPattern = false)
|
||||
{
|
||||
return std::shared_ptr<SubscribeContext>(
|
||||
new SubscribeContext(std::move(weakSub), channel, isPattern));
|
||||
}
|
||||
|
||||
unsigned long long contextId() const
|
||||
{
|
||||
return contextId_;
|
||||
}
|
||||
|
||||
const std::string &channel() const
|
||||
{
|
||||
return channel_;
|
||||
}
|
||||
|
||||
const std::string &subscribeCommand() const
|
||||
{
|
||||
return subscribeCommand_;
|
||||
}
|
||||
|
||||
const std::string &unsubscribeCommand() const
|
||||
{
|
||||
return unsubscribeCommand_;
|
||||
}
|
||||
|
||||
void addMessageCallback(RedisMessageCallback &&messageCallback)
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
messageCallbacks_.emplace_back(std::move(messageCallback));
|
||||
}
|
||||
|
||||
void disable()
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
disabled_ = true;
|
||||
messageCallbacks_.clear();
|
||||
}
|
||||
|
||||
void clear()
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
messageCallbacks_.clear();
|
||||
}
|
||||
|
||||
/**
|
||||
* Message callback called by RedisConnection, whenever a message is
|
||||
* published in target channel
|
||||
* @param channel : target channel name
|
||||
* @param message : message from channel
|
||||
*/
|
||||
void onMessage(const std::string &channel, const std::string &message);
|
||||
|
||||
/**
|
||||
* Callback called by RedisConnection, whenever a sub or re-sub is success
|
||||
*/
|
||||
void onSubscribe(const std::string &channel, long long numChannels);
|
||||
|
||||
/**
|
||||
* Callback called by RedisConnection, when unsubscription success.
|
||||
*/
|
||||
void onUnsubscribe(const std::string &channel, long long numChannels);
|
||||
|
||||
bool alive() const
|
||||
{
|
||||
return !disabled_ && weakSub_.lock() != nullptr;
|
||||
}
|
||||
|
||||
private:
|
||||
SubscribeContext(std::weak_ptr<RedisSubscriber> &&weakSub,
|
||||
const std::string &channel,
|
||||
bool isPattern);
|
||||
static std::atomic<unsigned long long> maxContextId_;
|
||||
|
||||
unsigned long long contextId_;
|
||||
std::weak_ptr<RedisSubscriber> weakSub_;
|
||||
std::string channel_;
|
||||
bool isPattern_{false};
|
||||
std::string subscribeCommand_;
|
||||
std::string unsubscribeCommand_;
|
||||
std::mutex mutex_;
|
||||
std::list<RedisMessageCallback> messageCallbacks_;
|
||||
bool disabled_{false};
|
||||
};
|
||||
|
||||
} // namespace drogon::nosql
|
||||
@@ -0,0 +1,20 @@
|
||||
link_libraries(${PROJECT_NAME})
|
||||
if(WIN32)
|
||||
link_libraries(iphlpapi)
|
||||
endif(WIN32)
|
||||
|
||||
add_executable(redis_test
|
||||
redis_test.cc
|
||||
)
|
||||
|
||||
set_property(TARGET redis_test PROPERTY CXX_STANDARD ${DROGON_CXX_STANDARD})
|
||||
set_property(TARGET redis_test PROPERTY CXX_STANDARD_REQUIRED ON)
|
||||
set_property(TARGET redis_test PROPERTY CXX_EXTENSIONS OFF)
|
||||
|
||||
add_executable(redis_subscriber_test
|
||||
redis_subscriber_test.cc
|
||||
)
|
||||
|
||||
set_property(TARGET redis_subscriber_test PROPERTY CXX_STANDARD ${DROGON_CXX_STANDARD})
|
||||
set_property(TARGET redis_subscriber_test PROPERTY CXX_STANDARD_REQUIRED ON)
|
||||
set_property(TARGET redis_subscriber_test PROPERTY CXX_EXTENSIONS OFF)
|
||||
@@ -0,0 +1,118 @@
|
||||
#define DROGON_TEST_MAIN
|
||||
#include <drogon/nosql/RedisClient.h>
|
||||
#include <drogon/drogon_test.h>
|
||||
#include <drogon/drogon.h>
|
||||
#include <iostream>
|
||||
#include <thread>
|
||||
|
||||
using namespace std::chrono_literals;
|
||||
using namespace drogon::nosql;
|
||||
static std::atomic_int nMsgRecv{0};
|
||||
static std::atomic_int nPmsgRecv{0};
|
||||
static std::atomic_int nMsgSent{0};
|
||||
|
||||
RedisClientPtr redisClient;
|
||||
|
||||
DROGON_TEST(RedisSubscriberTest)
|
||||
{
|
||||
redisClient = drogon::nosql::RedisClient::newRedisClient(
|
||||
trantor::InetAddress("127.0.0.1", 6379), 1);
|
||||
REQUIRE(redisClient != nullptr);
|
||||
|
||||
auto subscriber = redisClient->newSubscriber();
|
||||
subscriber->subscribe("test_sub",
|
||||
[](const std::string &channel,
|
||||
const std::string &message) {
|
||||
++nMsgRecv;
|
||||
LOG_INFO << "Channel test_sub receive "
|
||||
<< nMsgRecv << " messages: " << message;
|
||||
});
|
||||
subscriber->psubscribe("test_*",
|
||||
[](const std::string &channel,
|
||||
const std::string &message) {
|
||||
++nPmsgRecv;
|
||||
LOG_INFO << "Channel " << channel << " receive "
|
||||
<< nPmsgRecv
|
||||
<< " pmessages: " << message;
|
||||
});
|
||||
std::this_thread::sleep_for(1s);
|
||||
|
||||
auto fnPublish = [TEST_CTX](const char *channel, int i) {
|
||||
redisClient->execCommandAsync(
|
||||
[TEST_CTX](const drogon::nosql::RedisResult &r) {
|
||||
SUCCESS();
|
||||
++nMsgSent;
|
||||
},
|
||||
[TEST_CTX](const std::exception &err) {
|
||||
MANDATE(err.what());
|
||||
LOG_ERROR << err.what();
|
||||
++nMsgSent;
|
||||
},
|
||||
"publish %s %s%d",
|
||||
channel,
|
||||
"drogon",
|
||||
i);
|
||||
};
|
||||
|
||||
for (int i = 0; i < 5; ++i)
|
||||
{
|
||||
fnPublish("test_sub", i);
|
||||
}
|
||||
for (int i = 5; i < 10; ++i)
|
||||
{
|
||||
fnPublish("test_test", i);
|
||||
}
|
||||
|
||||
while (nMsgSent < 10)
|
||||
{
|
||||
std::this_thread::sleep_for(100ms);
|
||||
}
|
||||
std::this_thread::sleep_for(1s);
|
||||
|
||||
MANDATE(nMsgRecv == 5);
|
||||
MANDATE(nPmsgRecv == 10);
|
||||
|
||||
// Unsub from channel
|
||||
subscriber->unsubscribe("test_sub");
|
||||
fnPublish("test_sub", 11);
|
||||
while (nMsgSent < 11)
|
||||
{
|
||||
std::this_thread::sleep_for(100ms);
|
||||
}
|
||||
std::this_thread::sleep_for(1s);
|
||||
MANDATE(nMsgRecv == 5);
|
||||
MANDATE(nPmsgRecv == 11);
|
||||
|
||||
// Unsub from pattern
|
||||
subscriber->punsubscribe("test_*");
|
||||
fnPublish("test_sub", 12);
|
||||
while (nMsgSent < 12)
|
||||
{
|
||||
std::this_thread::sleep_for(100ms);
|
||||
}
|
||||
std::this_thread::sleep_for(1s);
|
||||
MANDATE(nMsgRecv == 5);
|
||||
MANDATE(nPmsgRecv == 11);
|
||||
}
|
||||
|
||||
int main(int argc, char **argv)
|
||||
{
|
||||
#ifndef USE_REDIS
|
||||
LOG_DEBUG << "Drogon is built without "
|
||||
"Redis. No tests executed.";
|
||||
return 0;
|
||||
#endif
|
||||
std::promise<void> p1;
|
||||
std::future<void> f1 = p1.get_future();
|
||||
|
||||
std::thread thr([&]() {
|
||||
p1.set_value();
|
||||
drogon::app().run();
|
||||
});
|
||||
|
||||
f1.get();
|
||||
int testStatus = drogon::test::run(argc, argv);
|
||||
drogon::app().getLoop()->queueInLoop([]() { drogon::app().quit(); });
|
||||
thr.join();
|
||||
return testStatus;
|
||||
}
|
||||
@@ -0,0 +1,205 @@
|
||||
#define DROGON_TEST_MAIN
|
||||
#include <drogon/nosql/RedisClient.h>
|
||||
#include <drogon/drogon_test.h>
|
||||
#include <drogon/drogon.h>
|
||||
#include <iostream>
|
||||
#include <thread>
|
||||
|
||||
using namespace std::chrono_literals;
|
||||
using namespace drogon::nosql;
|
||||
|
||||
RedisClientPtr redisClient;
|
||||
|
||||
DROGON_TEST(RedisTest)
|
||||
{
|
||||
redisClient = drogon::nosql::RedisClient::newRedisClient(
|
||||
trantor::InetAddress("127.0.0.1", 6379), 1);
|
||||
REQUIRE(redisClient != nullptr);
|
||||
// std::this_thread::sleep_for(1s);
|
||||
redisClient->newTransactionAsync(
|
||||
[TEST_CTX](const RedisTransactionPtr &transPtr) {
|
||||
// 1
|
||||
transPtr->execCommandAsync(
|
||||
[TEST_CTX](const drogon::nosql::RedisResult &r) { SUCCESS(); },
|
||||
[TEST_CTX](const std::exception &err) { MANDATE(err.what()); },
|
||||
"ping");
|
||||
// 2
|
||||
transPtr->execute(
|
||||
[TEST_CTX](const drogon::nosql::RedisResult &r) { SUCCESS(); },
|
||||
[TEST_CTX](const std::exception &err) { MANDATE(err.what()); });
|
||||
});
|
||||
// 3
|
||||
redisClient->execCommandAsync(
|
||||
[TEST_CTX](const drogon::nosql::RedisResult &r) { SUCCESS(); },
|
||||
[TEST_CTX](const std::exception &err) { MANDATE(err.what()); },
|
||||
"set %s %s",
|
||||
"id_123",
|
||||
"drogon");
|
||||
// 4
|
||||
redisClient->execCommandAsync(
|
||||
[TEST_CTX](const drogon::nosql::RedisResult &r) {
|
||||
MANDATE(r.type() == RedisResultType::kArray);
|
||||
MANDATE(r.asArray().size() == 1UL);
|
||||
},
|
||||
[TEST_CTX](const std::exception &err) { MANDATE(err.what()); },
|
||||
"keys id_*");
|
||||
// 5
|
||||
redisClient->execCommandAsync(
|
||||
[TEST_CTX](const drogon::nosql::RedisResult &r) {
|
||||
MANDATE(r.asString() == "hello");
|
||||
},
|
||||
[TEST_CTX](const RedisException &err) { MANDATE(err.what()); },
|
||||
"echo %s",
|
||||
"hello");
|
||||
// 6
|
||||
redisClient->execCommandAsync(
|
||||
[TEST_CTX](const drogon::nosql::RedisResult &r) { SUCCESS(); },
|
||||
[TEST_CTX](const RedisException &err) { MANDATE(err.what()); },
|
||||
"flushall");
|
||||
// 7
|
||||
redisClient->execCommandAsync(
|
||||
|
||||
[TEST_CTX](const drogon::nosql::RedisResult &r) {
|
||||
MANDATE(r.type() == RedisResultType::kNil);
|
||||
},
|
||||
[TEST_CTX](const RedisException &err) { MANDATE(err.what()); },
|
||||
"get %s",
|
||||
"xxxxx");
|
||||
|
||||
#ifdef __cpp_impl_coroutine
|
||||
auto coro_test = [TEST_CTX]() -> drogon::Task<> {
|
||||
// 8
|
||||
try
|
||||
{
|
||||
auto r = co_await redisClient->execCommandCoro("get %s", "haha");
|
||||
MANDATE(r.type() == RedisResultType::kNil);
|
||||
}
|
||||
catch (const RedisException &err)
|
||||
{
|
||||
FAULT(err.what());
|
||||
}
|
||||
};
|
||||
drogon::sync_wait(coro_test());
|
||||
#endif
|
||||
|
||||
// 9. Test sync
|
||||
try
|
||||
{
|
||||
auto res = redisClient->execCommandSync<std::string>(
|
||||
[](const RedisResult &result) { return result.asString(); },
|
||||
"set %s %s",
|
||||
"sync_key",
|
||||
"sync_value");
|
||||
MANDATE(res == "OK");
|
||||
}
|
||||
catch (const RedisException &err)
|
||||
{
|
||||
MANDATE(err.what());
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
auto [isNull, str] =
|
||||
redisClient->execCommandSync<std::pair<bool, std::string>>(
|
||||
[](const RedisResult &result) -> std::pair<bool, std::string> {
|
||||
if (result.isNil())
|
||||
{
|
||||
return {true, ""};
|
||||
}
|
||||
return {false, result.asString()};
|
||||
},
|
||||
"get %s",
|
||||
"sync_key");
|
||||
MANDATE(isNull == false);
|
||||
MANDATE(str == "sync_value");
|
||||
}
|
||||
catch (const RedisException &err)
|
||||
{
|
||||
MANDATE(err.what());
|
||||
}
|
||||
|
||||
// 10. Test sync redis exception
|
||||
try
|
||||
{
|
||||
auto [isNull, str] =
|
||||
redisClient->execCommandSync<std::pair<bool, std::string>>(
|
||||
[](const RedisResult &result) -> std::pair<bool, std::string> {
|
||||
if (result.isNil())
|
||||
{
|
||||
return {true, ""};
|
||||
}
|
||||
return {false, result.asString()};
|
||||
},
|
||||
"get %s %s",
|
||||
"sync_key",
|
||||
"sync_key");
|
||||
MANDATE(false);
|
||||
}
|
||||
catch (const RedisException &err)
|
||||
{
|
||||
LOG_INFO << "Successfully catch sync error: " << err.what();
|
||||
MANDATE(err.code() == RedisErrorCode::kRedisError);
|
||||
SUCCESS();
|
||||
}
|
||||
|
||||
// 11. Test sync process function exception
|
||||
try
|
||||
{
|
||||
auto value = redisClient->execCommandSync<std::string>(
|
||||
[](const RedisResult &result) {
|
||||
if (result.isNil())
|
||||
{
|
||||
throw std::runtime_error("Key not exists");
|
||||
}
|
||||
return result.asString();
|
||||
},
|
||||
"get %s",
|
||||
"not_exists");
|
||||
MANDATE(false);
|
||||
}
|
||||
catch (const RedisException &err)
|
||||
{
|
||||
(void)err;
|
||||
MANDATE(false);
|
||||
}
|
||||
catch (const std::runtime_error &err)
|
||||
{
|
||||
MANDATE(std::string("Key not exists") == err.what());
|
||||
SUCCESS();
|
||||
}
|
||||
|
||||
// 12. Test omit template parameter
|
||||
try
|
||||
{
|
||||
auto i = redisClient->execCommandSync(
|
||||
[](const RedisResult &r) { return r.asInteger(); },
|
||||
"del %s",
|
||||
"sync_key");
|
||||
MANDATE(i == 1);
|
||||
}
|
||||
catch (const RedisException &err)
|
||||
{
|
||||
MANDATE(err.what());
|
||||
}
|
||||
}
|
||||
|
||||
int main(int argc, char **argv)
|
||||
{
|
||||
#ifndef USE_REDIS
|
||||
LOG_DEBUG << "Drogon is built without Redis. No tests executed.";
|
||||
return 0;
|
||||
#endif
|
||||
std::promise<void> p1;
|
||||
std::future<void> f1 = p1.get_future();
|
||||
|
||||
std::thread thr([&]() {
|
||||
p1.set_value();
|
||||
drogon::app().run();
|
||||
});
|
||||
|
||||
f1.get();
|
||||
int testStatus = drogon::test::run(argc, argv);
|
||||
drogon::app().getLoop()->queueInLoop([]() { drogon::app().quit(); });
|
||||
thr.join();
|
||||
return testStatus;
|
||||
}
|
||||
Reference in New Issue
Block a user