Param (ii) (#66)

* Params * apply @gkorland's suggestion
RedisTimeSeries · Jul 30, 2021 · 1006688 · 1006688
1 parent 3ee7924
commit 1006688
Show file tree

Hide file tree

Showing 5 changed files with 335 additions and 4 deletions.
diff --git a/src/main/java/com/redislabs/redistimeseries/CreateParams.java b/src/main/java/com/redislabs/redistimeseries/CreateParams.java
@@ -0,0 +1,71 @@
+package com.redislabs.redistimeseries;
+
+import java.util.List;
+import java.util.Map;
+import redis.clients.jedis.Protocol;
+import redis.clients.jedis.util.SafeEncoder;
+
+public class CreateParams {
+
+  private Long retentionTime;
+  private boolean uncompressed;
+  private Long chunkSize;
+  private DuplicatePolicy duplicatePolicy;
+  private Map<String, String> labels;
+
+  public static CreateParams createParams() {
+    return new CreateParams();
+  }
+
+  public CreateParams() {}
+
+  public CreateParams retentionTime(long retentionTime) {
+    this.retentionTime = retentionTime;
+    return this;
+  }
+
+  public CreateParams uncompressed() {
+    this.uncompressed = true;
+    return this;
+  }
+
+  public CreateParams chunkSize(long chunkSize) {
+    this.chunkSize = chunkSize;
+    return this;
+  }
+
+  public CreateParams duplicatePolicy(DuplicatePolicy duplicatePolicy) {
+    this.duplicatePolicy = duplicatePolicy;
+    return this;
+  }
+
+  public CreateParams labels(Map<String, String> labels) {
+    this.labels = labels;
+    return this;
+  }
+
+  public void addOptionalParams(List<byte[]> params) {
+    if (retentionTime != null) {
+      params.add(Keyword.RETENTION.getRaw());
+      params.add(Protocol.toByteArray(retentionTime));
+    }
+    if (uncompressed) {
+      params.add(Keyword.UNCOMPRESSED.getRaw());
+    }
+    if (chunkSize != null) {
+      params.add(Keyword.CHUNK_SIZE.getRaw());
+      params.add(Protocol.toByteArray(chunkSize));
+    }
+    if (duplicatePolicy != null) {
+      params.add(Keyword.DUPLICATE_POLICY.getRaw());
+      params.add(duplicatePolicy.getRaw());
+    }
+    if (labels != null) {
+      params.add(Keyword.LABELS.getRaw());
+      for (Map.Entry<String, String> entry : labels.entrySet()) {
+        params.add(SafeEncoder.encode(entry.getKey()));
+        params.add(SafeEncoder.encode(entry.getValue()));
+      }
+    }
+  }
+}
diff --git a/src/main/java/com/redislabs/redistimeseries/MultiRangeParams.java b/src/main/java/com/redislabs/redistimeseries/MultiRangeParams.java
@@ -0,0 +1,71 @@
+package com.redislabs.redistimeseries;
+
+import java.util.ArrayList;
+import java.util.List;
+import redis.clients.jedis.Protocol;
+import redis.clients.jedis.util.SafeEncoder;
+
+public class MultiRangeParams {
+
+  private Integer count;
+
+  private Aggregation aggregationType;
+  private long timeBucket;
+
+  private boolean withLabels;
+
+  public static MultiRangeParams multiRangeParams() {
+    return new MultiRangeParams();
+  }
+
+  public MultiRangeParams count(int count) {
+    this.count = count;
+    return this;
+  }
+
+  public MultiRangeParams aggregation(Aggregation aggregation, long timeBucket) {
+    this.aggregationType = aggregation;
+    this.timeBucket = timeBucket;
+    return this;
+  }
+
+  public MultiRangeParams withLabels() {
+    this.withLabels = true;
+    return this;
+  }
+
+  public MultiRangeParams withLabels(boolean withLabels) {
+    if (withLabels) {
+      return withLabels();
+    }
+    return this;
+  }
+
+  public byte[][] getByteParams(long from, long to, String... filters) {
+    List<byte[]> params = new ArrayList<>();
+    params.add(Protocol.toByteArray(from));
+    params.add(Protocol.toByteArray(to));
+
+    if (count != null) {
+      params.add(Keyword.COUNT.getRaw());
+      params.add(Protocol.toByteArray(count));
+    }
+
+    if (aggregationType != null) {
+      params.add(Keyword.AGGREGATION.getRaw());
+      params.add(aggregationType.getRaw());
+      params.add(Protocol.toByteArray(timeBucket));
+    }
+
+    if (withLabels) {
+      params.add(Keyword.WITHLABELS.getRaw());
+    }
+
+    params.add(Keyword.FILTER.getRaw());
+    for (String filter : filters) {
+      params.add(SafeEncoder.encode(filter));
+    }
+
+    return params.toArray(new byte[params.size()][]);
+  }
+}
diff --git a/src/main/java/com/redislabs/redistimeseries/RangeParams.java b/src/main/java/com/redislabs/redistimeseries/RangeParams.java
@@ -0,0 +1,49 @@
+package com.redislabs.redistimeseries;
+
+import java.util.ArrayList;
+import java.util.List;
+import redis.clients.jedis.Protocol;
+import redis.clients.jedis.util.SafeEncoder;
+
+public class RangeParams {
+
+  private Integer count;
+
+  private Aggregation aggregationType;
+  private long timeBucket;
+
+  public static RangeParams rangeParams() {
+    return new RangeParams();
+  }
+
+  public RangeParams count(int count) {
+    this.count = count;
+    return this;
+  }
+
+  public RangeParams aggregation(Aggregation aggregation, long timeBucket) {
+    this.aggregationType = aggregation;
+    this.timeBucket = timeBucket;
+    return this;
+  }
+
+  public byte[][] getByteParams(String key, long from, long to) {
+    List<byte[]> params = new ArrayList<>();
+    params.add(SafeEncoder.encode(key));
+    params.add(Protocol.toByteArray(from));
+    params.add(Protocol.toByteArray(to));
+
+    if (count != null) {
+      params.add(Keyword.COUNT.getRaw());
+      params.add(Protocol.toByteArray(count));
+    }
+
+    if (aggregationType != null) {
+      params.add(Keyword.AGGREGATION.getRaw());
+      params.add(aggregationType.getRaw());
+      params.add(Protocol.toByteArray(timeBucket));
+    }
+
+    return params.toArray(new byte[params.size()][]);
+  }
+}
diff --git a/src/main/java/com/redislabs/redistimeseries/RedisTimeSeries.java b/src/main/java/com/redislabs/redistimeseries/RedisTimeSeries.java
@@ -1,6 +1,7 @@
 package com.redislabs.redistimeseries;
 
 import com.redislabs.redistimeseries.information.Info;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -201,6 +202,25 @@ private static byte[][] tsCreateArgs(
     return args;
   }
 
+  /**
+   * TS.CREATE key [RETENTION retentionTime] [UNCOMPRESSED] [CHUNK_SIZE size] [DUPLICATE_POLICY
+   * policy] [LABELS label value..]
+   *
+   * @param key
+   * @param createParams
+   * @return
+   */
+  public boolean create(String key, CreateParams createParams) {
+    try (Jedis conn = getConnection()) {
+      List<byte[]> params = new ArrayList<>();
+      params.add(SafeEncoder.encode(key));
+      createParams.addOptionalParams(params);
+      return sendCommand(conn, Command.CREATE, params.toArray(new byte[params.size()][]))
+          .getStatusCodeReply()
+          .equals("OK");
+    }
+  }
+
   /**
    * TS.DEL key fromTimestamp toTimestamp
    *
@@ -514,6 +534,51 @@ private static byte[][] tsAddArgs(
     return args;
   }
 
+  /**
+   * TS.ADD key timestamp value [RETENTION retentionTime] [UNCOMPRESSED] [CHUNK_SIZE size]
+   * [ON_DUPLICATE policy] [LABELS label value..]
+   *
+   * @param sourceKey
+   * @param value
+   * @param createParams
+   * @return
+   */
+  public long add(String sourceKey, double value, CreateParams createParams) {
+    try (Jedis conn = getConnection()) {
+      List<byte[]> params = new ArrayList<>();
+      params.add(SafeEncoder.encode(sourceKey));
+      params.add(Protocol.BYTES_ASTERISK);
+      params.add(Protocol.toByteArray(value));
+
+      createParams.addOptionalParams(params);
+      return sendCommand(conn, Command.ADD, params.toArray(new byte[params.size()][]))
+          .getIntegerReply();
+    }
+  }
+
+  /**
+   * TS.ADD key timestamp value [RETENTION retentionTime] [UNCOMPRESSED] [CHUNK_SIZE size]
+   * [ON_DUPLICATE policy] [LABELS label value..]
+   *
+   * @param sourceKey
+   * @param timestamp
+   * @param value
+   * @param addParams
+   * @return
+   */
+  public long add(String sourceKey, long timestamp, double value, CreateParams addParams) {
+    try (Jedis conn = getConnection()) {
+      List<byte[]> params = new ArrayList<>();
+      params.add(SafeEncoder.encode(sourceKey));
+      params.add(Protocol.toByteArray(timestamp));
+      params.add(Protocol.toByteArray(value));
+
+      addParams.addOptionalParams(params);
+      return sendCommand(conn, Command.ADD, params.toArray(new byte[params.size()][]))
+          .getIntegerReply();
+    }
+  }
+
   /**
    * TS.MADD key timestamp value [key timestamp value ...]
    *
@@ -624,6 +689,23 @@ public Value[] range(
         Protocol.toByteArray(count));
   }
 
+  /**
+   * TS.RANGE key fromTimestamp toTimestamp [COUNT count] [AGGREGATION aggregationType timeBucket]
+   *
+   * @param key
+   * @param from
+   * @param to
+   * @param rangeParams
+   * @return
+   */
+  public Value[] range(String key, long from, long to, RangeParams rangeParams) {
+    try (Jedis conn = getConnection()) {
+      Object obj =
+          sendCommand(conn, Command.RANGE, rangeParams.getByteParams(key, from, to)).getOne();
+      return Range.parseRange((List<Object>) obj);
+    }
+  }
+
   /**
    * TS.REVRANGE key fromTimestamp toTimestamp
    *
@@ -707,6 +789,24 @@ public Value[] revrange(
         Protocol.toByteArray(count));
   }
 
+  /**
+   * TS.REVRANGE key fromTimestamp toTimestamp [COUNT count] [AGGREGATION aggregationType
+   * timeBucket]
+   *
+   * @param key
+   * @param from
+   * @param to
+   * @param rangeParams
+   * @return
+   */
+  public Value[] revrange(String key, long from, long to, RangeParams rangeParams) {
+    try (Jedis conn = getConnection()) {
+      Object obj =
+          sendCommand(conn, Command.REVRANGE, rangeParams.getByteParams(key, from, to)).getOne();
+      return Range.parseRange((List<Object>) obj);
+    }
+  }
+
   /**
    * TS.MRANGE fromTimestamp toTimestamp FILTER filter. </br> Similar to calling <code>
    * mrange(from, to, null, 0, false, null, filters)</code>
@@ -845,6 +945,25 @@ private Range[] multiRange(
     }
   }
 
+  /**
+   * TS.MRANGE fromTimestamp toTimestamp [COUNT count] [AGGREGATION aggregationType timeBucket]
+   * [WITHLABELS] FILTER filter...
+   *
+   * @param from
+   * @param to
+   * @param multiRangeParams
+   * @param filters
+   * @return
+   */
+  public Range[] mrange(long from, long to, MultiRangeParams multiRangeParams, String... filters) {
+    try (Jedis conn = getConnection()) {
+      Object obj =
+          sendCommand(conn, Command.MRANGE, multiRangeParams.getByteParams(from, to, filters))
+              .getOne();
+      return Range.parseRanges((List<?>) obj);
+    }
+  }
+
   /**
    * TS.MREVRANGE fromTimestamp toTimestamp FILTER filter. </br> Similar to calling <code>
    * mrevrange(from, to, null, 0, false, null, filters)</code>
@@ -938,6 +1057,26 @@ public Range[] mrevrange(
         Command.MREVRANGE, from, to, aggregation, timeBucket, withLabels, count, filters);
   }
 
+  /**
+   * TS.MREVRANGE fromTimestamp toTimestamp [COUNT count] [AGGREGATION aggregationType timeBucket]
+   * [WITHLABELS] FILTER filter...
+   *
+   * @param from
+   * @param to
+   * @param multiRangeParams
+   * @param filters
+   * @return
+   */
+  public Range[] mrevrange(
+      long from, long to, MultiRangeParams multiRangeParams, String... filters) {
+    try (Jedis conn = getConnection()) {
+      Object obj =
+          sendCommand(conn, Command.MREVRANGE, multiRangeParams.getByteParams(from, to, filters))
+              .getOne();
+      return Range.parseRanges((List<?>) obj);
+    }
+  }
+
   /**
    * TS.GET key
    *