ALIGN option

src/main/java/com/redislabs/redistimeseries/Keyword.java

@@ -15,7 +15,8 @@ public enum Keyword implements ProtocolCommand {
   UNCOMPRESSED,
   CHUNK_SIZE,
   DUPLICATE_POLICY,
-  ON_DUPLICATE;
+  ON_DUPLICATE,
+  ALIGN;
 
   private final byte[] raw;
 

src/main/java/com/redislabs/redistimeseries/RangeParams.java

@@ -9,6 +9,8 @@ public class RangeParams {
 
   private Integer count;
 
+  private byte[] align;
+
   private Aggregation aggregationType;
   private long timeBucket;
 
@@ -21,6 +23,23 @@ public RangeParams count(int count) {
     return this;
   }
 
+  private RangeParams align(byte[] raw) {
+    this.align = raw;
+    return this;
+  }
+
+  public RangeParams align(long timestamp) {
+    return align(Protocol.toByteArray(timestamp));
+  }
+
+  public RangeParams alignStart() {
+    return align("start".getBytes());
+  }
+
+  public RangeParams alignEnd() {
+    return align("end".getBytes());
+  }
+
   public RangeParams aggregation(Aggregation aggregation, long timeBucket) {
     this.aggregationType = aggregation;
     this.timeBucket = timeBucket;
@@ -38,6 +57,11 @@ public byte[][] getByteParams(String key, long from, long to) {
       params.add(Protocol.toByteArray(count));
     }
 
+    if (align != null) {
+      params.add(Keyword.ALIGN.getRaw());
+      params.add(align);
+    }
+
     if (aggregationType != null) {
       params.add(Keyword.AGGREGATION.getRaw());
       params.add(aggregationType.getRaw());

src/test/java/com/redislabs/redistimeseries/RedisTimeSeriesTest.java

@@ -1,5 +1,6 @@
 package com.redislabs.redistimeseries;
 
+import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
@@ -409,6 +410,39 @@ public void testIncrByDecrBy() throws InterruptedException {
     }
   }
 
+  @Test
+  public void align() {
+    client.add("align", 1, 10d);
+    client.add("align", 3, 5d);
+    client.add("align", 11, 10d);
+    client.add("align", 25, 11d);
+
+    Value[] values =
+        client.range("align", 1, 30, RangeParams.rangeParams().aggregation(Aggregation.COUNT, 10));
+    assertArrayEquals(new Value[] {new Value(1, 2), new Value(11, 1), new Value(21, 1)}, values);
+
+    values =
+        client.range(
+            "align",
+            1,
+            30,
+            RangeParams.rangeParams().alignStart().aggregation(Aggregation.COUNT, 10));
+    assertArrayEquals(new Value[] {new Value(1, 2), new Value(11, 1), new Value(21, 1)}, values);
+
+    values =
+        client.range(
+            "align",
+            1,
+            30,
+            RangeParams.rangeParams().alignEnd().aggregation(Aggregation.COUNT, 10));
+    assertArrayEquals(new Value[] {new Value(1, 2), new Value(11, 1), new Value(21, 1)}, values);
+
+    values =
+        client.range(
+            "align", 1, 30, RangeParams.rangeParams().align(5).aggregation(Aggregation.COUNT, 10));
+    assertArrayEquals(new Value[] {new Value(1, 2), new Value(11, 1), new Value(21, 1)}, values);
+  }
+
   @Test
   public void testGet() {