redis: add ttl command

Add ttl command that returns remaining TTL of the key.

See: https://redis.io/commands/ttl

redis/command_factory.cc

@@ -36,6 +36,7 @@ shared_ptr<abstract_command> command_factory::create(service::storage_proxy& pro
         { "select",  [] (service::storage_proxy& proxy, request&& req) { return commands::select::prepare(proxy, std::move(req)); } }, 
         { "get",  [] (service::storage_proxy& proxy, request&& req) { return commands::get::prepare(proxy, std::move(req)); } }, 
         { "exists", [] (service::storage_proxy& proxy, request&& req) { return commands::exists::prepare(proxy, std::move(req)); } },
+        { "ttl", [] (service::storage_proxy& proxy, request&& req) { return commands::ttl::prepare(proxy, std::move(req)); } },
         { "set",  [] (service::storage_proxy& proxy, request&& req) { return commands::set::prepare(proxy, std::move(req)); } }, 
         { "setex",  [] (service::storage_proxy& proxy, request&& req) { return commands::setex::prepare(proxy, std::move(req)); } },
         { "del",  [] (service::storage_proxy& proxy, request&& req) { return commands::del::prepare(proxy, std::move(req)); } }, 

redis/commands.cc

@@ -74,6 +74,26 @@ future<redis_message> exists::execute(service::storage_proxy& proxy, redis::redi
     });
 }
 
+shared_ptr<abstract_command> ttl::prepare(service::storage_proxy& proxy, request&& req) {
+    if (req.arguments_size() != 1) {
+        throw wrong_arguments_exception(1, req.arguments_size(), req._command);
+    }
+    return seastar::make_shared<ttl> (std::move(req._command), std::move(req._args[0]));
+}
+
+future<redis_message> ttl::execute(service::storage_proxy& proxy, redis::redis_options& options, service_permit permit) {
+    return redis::read_strings(proxy, options, _key, permit).then([] (auto result) {
+        if (result->has_result()) {
+            if (result->has_ttl()) {
+                return redis_message::number(result->ttl().count());
+            }else{
+                return redis_message::number(-1);
+            }
+        }
+        return redis_message::number(-2);
+    });
+}
+
 shared_ptr<abstract_command> set::prepare(service::storage_proxy& proxy, request&& req) {
     if (req.arguments_size() != 2) {
         throw wrong_arguments_exception(2, req.arguments_size(), req._command);

redis/commands.hh

@@ -48,6 +48,17 @@ public:
     virtual future<redis_message> execute(service::storage_proxy&, redis_options&, service_permit) override;
 };
 
+class ttl : public abstract_command {
+    bytes _key;
+public:
+    static shared_ptr<abstract_command> prepare(service::storage_proxy& proxy, request&& req);
+    ttl(bytes&& name, bytes&& key)
+        : abstract_command(std::move(name))
+        , _key(std::move(key)) {
+    }
+    virtual future<redis_message> execute(service::storage_proxy&, redis_options&, service_permit) override;
+};
+
 class set : public abstract_command {
     bytes _key;
     bytes _data;

redis/query_utils.cc

@@ -41,10 +41,13 @@ private:
     void add_cell(const column_definition& col, const std::optional<query::result_atomic_cell_view>& cell)
     {
         if (cell) {
-            cell->value().with_linearized([this, &col] (bytes_view cell_view) {
+            cell->value().with_linearized([this, &col, &cell] (bytes_view cell_view) {
                 auto&& dv = col.type->deserialize_value(cell_view);
                 auto&& d = dv.serialize_nonnull();
                 _data->_result = std::move(d);
+                if (cell->expiry().has_value()) {
+                    _data->_ttl = cell->expiry().value() - gc_clock::now();
+                }
                 _data->_has_result = true;
             });
         }

redis/query_utils.hh

@@ -24,6 +24,7 @@
 #include "seastar/core/shared_ptr.hh"
 #include "seastar/core/future.hh"
 #include "bytes.hh"
+#include "gc_clock.hh"
 
 using namespace seastar;
 
@@ -41,8 +42,11 @@ class redis_options;
 struct strings_result {
     bytes _result;
     bool _has_result;
+    ttl_opt _ttl;
     bytes& result() { return _result; }
     bool has_result() const { return _has_result; }
+    gc_clock::duration ttl() { return _ttl.value(); }
+    bool has_ttl() { return _ttl.has_value(); }
 };
 
 future<lw_shared_ptr<strings_result>> read_strings(service::storage_proxy&, const redis_options&, const bytes&, service_permit);