Use boost::redis instead of hiredis

omnn/rt/CMakeLists.txt

@@ -21,3 +21,6 @@ if(OPENMIND_USE_OPENCL)
 endif()
 
 lib(${DEPENDENCIES})
+
+# Add storage subdirectory
+add_subdirectory(storage)

omnn/rt/storage/CMakeLists.txt

@@ -0,0 +1,38 @@
+cmake_minimum_required(VERSION 3.15)
+
+# Find required packages
+find_package(Boost REQUIRED COMPONENTS unit_test_framework system thread filesystem)
+find_package(Threads REQUIRED)
+
+# Add Redis cache library
+add_library(redis_cache
+    RedisCache.cpp
+    RedisCache.h
+)
+
+# Find cache_base target
+if(TARGET cache_base)
+    set(CACHE_BASE_LIB cache_base)
+else()
+    # Fallback to direct path if target not available
+    set(CACHE_BASE_LIB ${CMAKE_BINARY_DIR}/omnn/storage/Debug/libcache_base.a)
+endif()
+
+target_link_libraries(redis_cache
+    PUBLIC
+    Boost::system
+    Boost::thread
+    Boost::filesystem
+    Threads::Threads
+    ${CACHE_BASE_LIB}
+    ${CMAKE_DL_LIBS}
+)
+
+target_include_directories(redis_cache
+    PUBLIC
+    ${CMAKE_CURRENT_SOURCE_DIR}
+    ${CMAKE_CURRENT_SOURCE_DIR}/../../storage
+)
+
+# Add tests subdirectory
+add_subdirectory(tests)

omnn/rt/storage/RedisCache.cpp

@@ -0,0 +1,216 @@
+#include "RedisCache.h"
+#include <chrono>
+#include <thread>
+#include <stdexcept>
+#include <sstream>
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/connect.hpp>
+#include <boost/asio/write.hpp>
+#include <boost/asio/read.hpp>
+#include <boost/asio/read_until.hpp>
+#include <boost/asio/streambuf.hpp>
+#include <boost/system/error_code.hpp>
+
+namespace omnn {
+namespace rt {
+namespace storage {
+
+RedisCache::RedisCache(const std::string& host, int port, int timeout_ms,
+                     int retry_count, int retry_delay_ms)
+    : host_(host), port_(port), timeout_ms_(timeout_ms),
+      retry_count_(retry_count), retry_delay_ms_(retry_delay_ms),
+      connected_(false) {
+
+    if (!Connect()) {
+        throw std::runtime_error("Failed to connect to Redis server");
+    }
+}
+
+RedisCache::~RedisCache() {
+    Disconnect();
+}
+
+bool RedisCache::Connect() {
+    Disconnect();  // Ensure any existing connection is closed
+
+    try {
+        socket_ = std::make_unique<boost::asio::ip::tcp::socket>(io_context_);
+
+        boost::asio::ip::tcp::resolver resolver(io_context_);
+        boost::asio::ip::tcp::resolver::query query(host_, std::to_string(port_));
+        boost::asio::ip::tcp::resolver::iterator endpoint_iterator = resolver.resolve(query);
+
+        boost::system::error_code ec;
+        boost::asio::connect(*socket_, endpoint_iterator, ec);
+
+        if (ec) {
+            connected_ = false;
+            return false;
+        }
+
+        connected_ = true;
+        return true;
+    } catch (const std::exception&) {
+        connected_ = false;
+        return false;
+    }
+}
+
+void RedisCache::Disconnect() {
+    if (socket_ && socket_->is_open()) {
+        boost::system::error_code ec;
+        socket_->close(ec);
+        socket_.reset();
+    }
+    connected_ = false;
+}
+
+bool RedisCache::RetryOperation(const std::function<bool()>& operation) {
+    for (int i = 0; i < retry_count_; ++i) {
+        if (operation()) {
+            return true;
+        }
+
+        if (!IsConnected() && !Connect()) {
+            std::this_thread::sleep_for(std::chrono::milliseconds(retry_delay_ms_));
+            continue;
+        }
+
+        std::this_thread::sleep_for(std::chrono::milliseconds(retry_delay_ms_));
+    }
+    return false;
+}
+
+std::string RedisCache::ExecuteCommand(const std::string& command) {
+    if (!IsConnected()) {
+        return "";
+    }
+
+    try {
+        boost::system::error_code ec;
+        boost::asio::write(*socket_, boost::asio::buffer(command), ec);
+
+        if (ec) {
+            return "";
+        }
+
+        boost::asio::streambuf response;
+        boost::asio::read_until(*socket_, response, "\r\n", ec);
+
+        if (ec) {
+            return "";
+        }
+
+        std::istream response_stream(&response);
+        std::string response_line;
+        std::getline(response_stream, response_line);
+
+        if (response_line.empty()) {
+            return "";
+        }
+
+        char type = response_line[0];
+        std::string value = response_line.substr(1);
+
+        switch (type) {
+            case '+': // Simple string
+                return value;
+            case '-': // Error
+                return "";
+            case ':': // Integer
+                return value;
+            case '$': { // Bulk string
+                int length = std::stoi(value);
+                if (length == -1) {
+                    return ""; // Null response
+                }
+
+                std::string bulk_string;
+                bulk_string.resize(length);
+
+                size_t bytes_read = boost::asio::read(*socket_, 
+                    boost::asio::buffer(&bulk_string[0], length), ec);
+
+                if (ec || bytes_read != length) {
+                    return "";
+                }
+
+                // Read the trailing \r\n
+                boost::asio::streambuf trailing;
+                boost::asio::read_until(*socket_, trailing, "\r\n", ec);
+                if (ec) {
+                    return "";
+                }
+
+                return bulk_string;
+            }
+            default:
+                return "";
+        }
+    } catch (const std::exception&) {
+        return "";
+    }
+}
+
+std::string RedisCache::GetOne(const std::string_view& key) {
+    std::string result;
+
+    RetryOperation([this, &key, &result]() {
+        if (!IsConnected()) return false;
+
+        std::stringstream cmd;
+        cmd << "*2\r\n$3\r\nGET\r\n$" << key.size() << "\r\n" << key << "\r\n";
+
+        result = ExecuteCommand(cmd.str());
+        return !result.empty() || result == ""; // Empty string is a valid result for non-existent key
+    });
+
+    return result;
+}
+
+bool RedisCache::Set(const std::string_view& key, const std::string_view& v) {
+    return RetryOperation([this, &key, &v]() {
+        if (!IsConnected()) return false;
+
+        std::stringstream cmd;
+        cmd << "*3\r\n$3\r\nSET\r\n$" << key.size() << "\r\n" << key 
+            << "\r\n$" << v.size() << "\r\n" << v << "\r\n";
+
+        std::string result = ExecuteCommand(cmd.str());
+        return result == "OK";
+    });
+}
+
+bool RedisCache::Clear(const std::string_view& key) {
+    return RetryOperation([this, &key]() {
+        if (!IsConnected()) return false;
+
+        std::stringstream cmd;
+        cmd << "*2\r\n$3\r\nDEL\r\n$" << key.size() << "\r\n" << key << "\r\n";
+
+        std::string result = ExecuteCommand(cmd.str());
+        return !result.empty(); // Any non-empty result indicates success
+    });
+}
+
+bool RedisCache::ResetAllDB(const fs::path& path) {
+    CacheBase::ResetAllDB(path);
+
+    return RetryOperation([this]() {
+        if (!IsConnected()) return false;
+
+        std::string cmd = "*1\r\n$7\r\nFLUSHDB\r\n";
+
+        std::string result = ExecuteCommand(cmd);
+        return result == "OK";
+    });
+}
+
+bool RedisCache::IsConnected() const {
+    return connected_ && socket_ && socket_->is_open();
+}
+
+} // namespace storage
+} // namespace rt
+} // namespace omnn

omnn/rt/storage/RedisCache.h

@@ -0,0 +1,57 @@
+#pragma once
+
+#include <string>
+#include <string_view>
+#include <memory>
+#include <functional>
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/connect.hpp>
+#include <boost/asio/strand.hpp>
+#include "../../storage/CacheBase.h"
+
+namespace omnn {
+namespace rt {
+namespace storage {
+
+namespace fs = boost::filesystem;
+namespace asio = boost::asio;
+
+class RedisCache : public omnn::rt::storage::CacheBase {
+public:
+    RedisCache(const std::string& host = "localhost", int port = 6379,
+               int timeout_ms = 1000, int retry_count = 5, int retry_delay_ms = 1000);
+    ~RedisCache();
+
+    // CacheBase interface implementation
+    std::string GetOne(const std::string_view& key) override;
+    bool Set(const std::string_view& key, const std::string_view& v) override;
+    bool Clear(const std::string_view& key) override;
+    bool ResetAllDB(const fs::path& path) override;
+
+    // Additional Redis-specific methods
+    bool IsConnected() const;
+
+private:
+    bool Connect();
+    void Disconnect();
+    bool RetryOperation(const std::function<bool()>& operation);
+
+    // Execute Redis command and get response
+    std::string ExecuteCommand(const std::string& command);
+
+    // Redis connection using boost::asio
+    asio::io_context io_context_;
+    std::unique_ptr<asio::ip::tcp::socket> socket_;
+
+    std::string host_;
+    int port_;
+    int timeout_ms_;
+    int retry_count_;
+    int retry_delay_ms_;
+    bool connected_;
+};
+
+} // namespace storage
+} // namespace rt
+} // namespace omnn

omnn/rt/storage/tests/CMakeLists.txt

@@ -0,0 +1,29 @@
+cmake_minimum_required(VERSION 3.15)
+
+# Redis cache test
+add_executable(redis_cache_test redis_cache_test.cpp)
+
+target_link_libraries(redis_cache_test
+    PRIVATE
+    redis_cache
+    Boost::system
+    Boost::thread
+    Boost::filesystem
+    Boost::unit_test_framework
+    Threads::Threads
+    ${CMAKE_DL_LIBS}
+)
+
+target_include_directories(redis_cache_test
+    PRIVATE
+    ${CMAKE_CURRENT_SOURCE_DIR}
+)
+
+# Enable the test
+add_test(NAME redis_cache_test COMMAND redis_cache_test)
+
+# Set test properties
+set_tests_properties(redis_cache_test PROPERTIES
+    ENVIRONMENT "OPENMIND_TEST_REDIS_RETRY_COUNT=5;OPENMIND_TEST_REDIS_RETRY_DELAY=1000"
+    TIMEOUT 60
+)