Closes #40. Support full sliding window (#44)

Author: hbxie

src/main/java/com/yahoo/bullet/parsing/Window.java

@@ -73,8 +73,6 @@ public enum Classification {
                                                                  "Please remove include or match it to emit");
     public static final BulletError IMPROPER_FIRST = makeError("The first field should not be set for \"ALL\"",
                                                               "Please remove the first field");
-    public static final BulletError NOT_ONE_RECORD_EMIT = makeError("The emit type was \"RECORD\" but every was not 1",
-                                                                    "Please set every to 1 or change the emit type");
     public static final BulletError NO_RECORD_ALL = makeError("The emit type was \"RECORD\" and include type was \"ALL\"",
                                                               "Please set emit type to \"TIME\" or match include to emit");
 
@@ -121,9 +119,6 @@ public Optional<List<BulletError>> initialize() {
         if (every == null || every.intValue() <= 0) {
             return Optional.of(singletonList(IMPROPER_EVERY));
         }
-        if (emitType == Unit.RECORD && every.intValue() != 1) {
-            return Optional.of(singletonList(NOT_ONE_RECORD_EMIT));
-        }
 
         if (include == null) {
             return Optional.empty();

src/main/java/com/yahoo/bullet/pubsub/rest/RESTPublisher.java

@@ -28,6 +28,7 @@ public abstract class RESTPublisher implements Publisher {
      * Create a RESTQueryPublisher from a {@link CloseableHttpClient}.
      *
      * @param client The client.
+     * @param connectTimeout The minimum time (ms) to wait for a connection to be made.
      */
     public RESTPublisher(CloseableHttpClient client, int connectTimeout) {
         this.client = client;

src/main/java/com/yahoo/bullet/pubsub/rest/RESTQueryPublisher.java

@@ -25,6 +25,7 @@ public class RESTQueryPublisher extends RESTPublisher {
      * @param client The client.
      * @param queryURL The URL to which to POST queries.
      * @param resultURL The URL that will be added to the Metadata (results will be sent to this URL from the backend).
+     * @param connectTimeout The minimum time (ms) to wait for a connection to be made.
      */
     public RESTQueryPublisher(CloseableHttpClient client, String queryURL, String resultURL, int connectTimeout) {
         super(client, connectTimeout);

src/main/java/com/yahoo/bullet/pubsub/rest/RESTResultPublisher.java

@@ -15,6 +15,7 @@ public class RESTResultPublisher extends RESTPublisher {
      * Create a RESTQueryPublisher from a {@link CloseableHttpClient}.
      *
      * @param client The client.
+     * @param connectTimeout The minimum time (ms) to wait for a connection to be made.
      */
     public RESTResultPublisher(CloseableHttpClient client, int connectTimeout) {
         super(client, connectTimeout);

src/main/java/com/yahoo/bullet/querying/Querier.java

@@ -290,7 +290,7 @@ public enum Mode {
     private Scheme window;
 
     // For testing convenience
-    @Getter(AccessLevel.PACKAGE) @Setter(AccessLevel.PACKAGE)
+    @Getter(AccessLevel.PACKAGE)
     private RunningQuery runningQuery;
 
     private BulletConfig config;

src/main/java/com/yahoo/bullet/querying/RateLimiter.java

@@ -18,15 +18,17 @@
  * the rate limit within your time interval but you neglect to check it for a long enough time interval where the burst
  * is spread out over bringing the overall rate lower than your configured maximum and yielding a false negative.
  */
-@Getter
 public class RateLimiter {
+    @Getter
     private final int maximum;
+    @Getter
     private final int timeInterval;
     private final double absoluteRateLimit;
 
     private long count = 0;
     private long lastCount = 0;
     private long lastCheckTime;
+    @Getter
     private boolean exceededRate = false;
 
     public static final int SECOND = 1000;

src/main/java/com/yahoo/bullet/querying/WindowingOperations.java

@@ -11,8 +11,8 @@
 import com.yahoo.bullet.parsing.Window;
 import com.yahoo.bullet.windowing.AdditiveTumbling;
 import com.yahoo.bullet.windowing.Basic;
-import com.yahoo.bullet.windowing.Reactive;
 import com.yahoo.bullet.windowing.Scheme;
+import com.yahoo.bullet.windowing.SlidingRecord;
 import com.yahoo.bullet.windowing.Tumbling;
 
 public class WindowingOperations {
@@ -29,7 +29,7 @@ public static Scheme findScheme(Query query, Strategy strategy, BulletConfig con
          * TODO: Support other windows
          * The windows we support at the moment:
          * 1. No window -> Basic
-         * 2. Window is emit RECORD and include RECORD -> Reactive
+         * 2. Window is emit RECORD and include RECORD -> SlidingRecord
          * 3. Window is emit TIME and include ALL -> Additive Tumbling
          * 4. All other windows -> Tumbling (RAW can be Tumbling too)
          */
@@ -40,7 +40,7 @@ public static Scheme findScheme(Query query, Strategy strategy, BulletConfig con
 
         Window.Classification classification = window.getType();
         if (classification == Window.Classification.RECORD_RECORD) {
-            return new Reactive(strategy, window, config);
+            return new SlidingRecord(strategy, window, config);
         }
         if (classification == Window.Classification.TIME_ALL) {
             return new AdditiveTumbling(strategy, window, config);

src/main/java/com/yahoo/bullet/windowing/Reactive.java

@@ -30,8 +30,8 @@ public class SlidingRecord extends Basic {
 
     public static final BulletError NOT_RECORD = makeError("The \"type\" for \"emit\" was not \"RECORD\"",
                                                            "Please set \"type\" to \"RECORD\"");
-    protected int maxCount;
-    protected int recordCount;
+    private int maxCount;
+    private int recordCount;
 
     @AllArgsConstructor @Getter
     public static class Data implements Serializable {

src/main/java/com/yahoo/bullet/windowing/SlidingRecord.java

@@ -137,13 +137,13 @@ public void testCustomDuration() {
     public void testWindowForced() {
         BulletConfig config = new BulletConfig();
         Query query = new Query();
-        query.setWindow(WindowUtils.makeReactiveWindow());
+        query.setWindow(WindowUtils.makeSlidingWindow(1));
         query.configure(config);
         Assert.assertNotNull(query.getWindow());
 
         config.set(BulletConfig.WINDOW_DISABLE, true);
         config.validate();
-        query.setWindow(WindowUtils.makeReactiveWindow());
+        query.setWindow(WindowUtils.makeSlidingWindow(1));
         query.configure(config);
         Assert.assertNull(query.getWindow());
     }