RedisBuffer.h

#pragma once
#include "AudioTools/CoreAudio/Buffers.h"
 
#define REDIS_RESULT_BUFFER_SIZE (10 * 1024)
 
 
namespace audio_tools {
 
template <typename T>
class RedisBuffer : public BaseBuffer<T> {
 public:
  RedisBuffer(Client& client, const char* key, size_t max_size,
              size_t local_buf_size = 512, int expire_seconds = 60 * 60)
      : client(client),
        key(key),
        max_size(max_size),
        local_buf_size(local_buf_size),
        expire_seconds(expire_seconds),
        write_buf(local_buf_size),
        read_buf(local_buf_size) {}
 
  void setExpire(int seconds) { expire_seconds = seconds; }
 
  bool write(T data) override {
    has_written = true;
    write_buf.write(data);
    if (write_buf.isFull()) {
      flushWrite();  // flush any pending writes first
    }
    return true;
  }
 
  int writeArray(const T data[], int len) override {
    LOGI("RedisBuffer:writeArray: %d", len);
    has_written = true;
    int written = 0;
    for (int i = 0; i < len; ++i) {
      write(data[i]);
      ++written;
    }
    return written;
  }
 
  bool read(T& result) override {
    flushWrite();  // flush any pending writes before reading
    if (read_buf.isEmpty()) {
      read_buf.reset();
      fillReadBuffer();  // fill local buffer from Redis
      if (read_buf.isEmpty()) {
        LOGI("RedisBuffer:read: no data available");
        return false;  // nothing left in Redis
      }
    }
    T val;
    bool rc = read_buf.read(val);  // read from local buffer
    LOGI("Redis LPOP: %d", val);
    result = val;
    return rc;
  }
 
  int readArray(T data[], int len) override {
    flushWrite();  // flush any pending writes before reading
    int read_count = 0;
    while (read_count < len) {
      if (read_buf.isEmpty()) {
        fillReadBuffer();
        if (read_buf.isEmpty()) break;  // nothing left in Redis
      }
      read_buf.read(data[read_count++]);
    }
    return read_count;
  }
 
  bool peek(T& result) override {
    flushWrite();  // flush any pending writes before peeking
 
    // Use LINDEX to peek at the first value in Redis without removing it
    String cmd = redisCommand("LINDEX", key, "0");
    RedisResult resp = sendCommand(cmd);
    if (!resp.ok) return false;
    result = (T)resp.intValue;
    return true;
  }
 
  void reset() override {
    flushWrite();
    String cmd = redisCommand("DEL", key);
    auto rc = sendCommand(cmd);
    LOGI("Redis DEL: %d", rc.intValue);
    read_buf.reset();
    write_buf.reset();
  }
 
  int available() override {
    flushWrite();
    String cmd = redisCommand("LLEN", key);
    RedisResult resp = sendCommand(cmd);
    LOGI("LLEN: %d (ok=%d)", resp.intValue, resp.ok);
    return resp.intValue + read_buf.available();
  }
 
  int availableForWrite() override { return max_size - available(); }
 
  T* address() override { return nullptr; }
 
  size_t size() override { return max_size; }
 
  bool resize(int size) override {
    if (has_written) return false;
    LOGI("RedisBuffer::resize: %d", size);
    max_size = size;
    return true;
  }
 
 protected:
  struct RedisResult {
    int intValue = 0;  
    Vector<String>
        strValues;    
    bool ok = false;  
    operator bool() const { return ok; }  
    // std::vector<T> values;
  };
 
  Client& client;  
  const char* key;         
  size_t max_size;         
  size_t local_buf_size;   
  int expire_seconds = 0;  
  SingleBuffer<T> write_buf;  
  SingleBuffer<T> read_buf;   
  bool has_written = false;   
 
  void clearResponse() {
    while (client.available()) {
      client.read();  // clear any remaining data in the buffer
    }
  }
 
  String redisCommand(const String& cmd, const String& arg1 = "",
                      const String& arg2 = "", const String& arg3 = "") {
    String out = "*" +
                 String(1 + (arg1.length() > 0) + (arg2.length() > 0) +
                        (arg3.length() > 0)) +
                 "\r\n";
    out += "$" + String(cmd.length()) + "\r\n" + cmd + "\r\n";
    if (arg1.length())
      out += "$" + String(arg1.length()) + "\r\n" + arg1 + "\r\n";
    if (arg2.length())
      out += "$" + String(arg2.length()) + "\r\n" + arg2 + "\r\n";
    if (arg3.length())
      out += "$" + String(arg3.length()) + "\r\n" + arg3 + "\r\n";
    return out;
  }
 
  RedisResult sendCommand(const String& cmd) {
    RedisResult result;
    if (!client.connected()) {
      LOGE("Redis not connected");
      result.ok = false;
      return result;
    }
    client.print(cmd);
    client.flush();
    result = readResponse();
    return result;
  }
 
  RedisResult readResponse() {
    RedisResult result;
    result.ok = true;
    uint8_t buffer[REDIS_RESULT_BUFFER_SIZE] = {};
    int n = 0;
    while (n <= 0) {
      n = client.read(buffer, sizeof(buffer));
    }
    buffer[n] = 0;
 
    // build vector of strings
    result.strValues.clear();
    String tail = (char*)buffer;
    int nl_pos = tail.indexOf("\r\n");
    while (nl_pos >= 0) {
      String head = tail.substring(0, nl_pos);
      if (!head.startsWith("$"))
        result.strValues.push_back(head);
      tail = tail.substring(nl_pos + 2);
      tail.trim();
      nl_pos = tail.indexOf("\r\n");
    }
 
    if (!tail.startsWith("$") && tail.length() > 0) {
      result.strValues.push_back(tail);
    }
 
    // if we have more then 2 lines this is a success
    if (result.strValues.size() > 2) {
      result.intValue = result.strValues.size();
    } else {
      // Else try to determine an int value
      StrView line((char*)buffer, sizeof(buffer), n);
 
      if (line.startsWith("$")) {
        int end = line.indexOf("\n");
        line.substring(line.c_str(), end, line.length());
      }
 
      if (line.startsWith(":")) {
        line.replace(":", "");
      }
 
      if (line.startsWith("-")){
        result.ok = false;
      }
 
      if (line.isEmpty()){
        result.intValue = -1;  // no data available
        result.ok = false;
      }
      else {
        result.intValue = line.toInt();
      }
    }
 
    return result;
  }
 
  void flushWrite() {
    if (write_buf.isEmpty()) return;
    int write_size = write_buf.available();
    // Use RPUSH with multiple arguments
    String cmd = "*" + String(2 + write_size) + "\r\n";
    cmd += "$5\r\nRPUSH\r\n";
    cmd += "$" + String(strlen(key)) + "\r\n" + key + "\r\n";
    T value;
    while (!write_buf.isEmpty()) {
      write_buf.read(value);
      String sval = String(value);
      cmd += "$" + String(sval.length()) + "\r\n" + sval + "\r\n";
    }
    write_buf.clear();
    RedisResult resp = sendCommand(cmd);
    LOGI("Redis RPUSH %d entries: %d (ok=%d)", write_size, resp.intValue,
         resp.ok);
 
    if (expire_seconds > 0) {
      String expireCmd = redisCommand("EXPIRE", key, String(expire_seconds));
      RedisResult resp = sendCommand(expireCmd);
      LOGI("Redis EXPIRE: %d (ok=%d)", resp.intValue, resp.ok);
    }
  }
 
  void fillReadBuffer() {
    read_buf.reset();
    int size = read_buf.size();
    LOGI("Redis LPOP: %d", size)
    // Read up to local_buf_size items from Redis
    String cmd = redisCommand("LPOP", key, String(size));
    auto rc = sendCommand(cmd);
    if (!rc.ok) {
      LOGE("Redis LPOP failed: %s", cmd.c_str());
      return;  // no data available
    }
    for (auto& str : rc.strValues) {
      if (str.startsWith("*")) continue;
      if (str.startsWith("$")) continue;
      if (str.length() == 0) continue;
      LOGI("- %s", str.c_str());
      T value = (T)str.toInt();
      read_buf.write(value);
    }
 
    LOGI("RedisBuffer: %d of %d items",(int) read_buf.available(),(int) read_buf.size() );
    // if this fails the 
    if(!read_buf.isFull() ){
      LOGW("RedisBuffer:fillReadBuffer: not enough data read from Redis");
    }
  }
};
 
}  // namespace audio_tools