Add available() function to subscriber

src/sw/redis++/connection.cpp

@@ -18,6 +18,7 @@
 #include <cassert>
 #include <tuple>
 #include <algorithm>
+#include <poll.h>
 #include "sw/redis++/reply.h"
 #include "sw/redis++/command.h"
 #include "sw/redis++/command_args.h"
@@ -253,6 +254,28 @@ ReplyUPtr Connection::recv(bool handle_error_reply) {
     return reply;
 }
 
+bool Connection::avail() {
+    auto *ctx = _context();
+
+    assert(ctx != nullptr);
+
+    auto fd = ctx->fd;
+
+    struct pollfd pfd;
+    pfd.fd = fd;
+    pfd.events = POLLIN;  // Check for readability
+
+    int result = poll(&pfd, 1, 0);  // Timeout set to 0 milliseconds
+    if (result > 0 && (pfd.revents & POLLIN)) {
+        return true;
+    } else if (result == 0) {
+        return false;
+    } else {
+        // An error occurred
+        throw_error(*ctx, strerror(errno));
+    }
+}
+
 #ifdef REDIS_PLUS_PLUS_RESP_VERSION_3
 
 void Connection::set_push_callback(redisPushFn *push_func) {

src/sw/redis++/connection.h

@@ -145,6 +145,8 @@ class Connection {
 
     ReplyUPtr recv(bool handle_error_reply = true);
 
+    bool avail();
+
     const ConnectionOptions& options() const {
         return _opts;
     }

src/sw/redis++/subscriber.cpp

@@ -127,6 +127,12 @@ void Subscriber::consume() {
     }
 }
 
+bool Subscriber::available() {
+    _check_connection();
+
+    return _connection.avail();
+}
+
 Subscriber::MsgType Subscriber::_msg_type(redisReply *reply) const {
     if (reply == nullptr) {
         throw ProtoError("Null type reply.");

src/sw/redis++/subscriber.h

@@ -163,6 +163,8 @@ class Subscriber {
 
     void consume();
 
+    bool available();
+
 private:
     friend class Redis;