validate builders (#32)

Author: daleiz

client/src/main/java/io/hstream/Subscription.java

@@ -1,5 +1,7 @@
 package io.hstream;
 
+import static com.google.common.base.Preconditions.*;
+
 import java.util.Objects;
 
 /** A class for storing information about subscriptions */
@@ -63,7 +65,7 @@ public static class Builder {
     private String subscriptionId;
     private String streamName;
     private SubscriptionOffset subscriptionOffset;
-    private int ackTimeoutSeconds;
+    private int ackTimeoutSeconds = 600;
 
     public Builder subscription(String subscriptionId) {
       this.subscriptionId = subscriptionId;
@@ -86,6 +88,10 @@ public Builder ackTimeoutSeconds(int ackTimeoutSeconds) {
     }
 
     public Subscription build() {
+      checkNotNull(subscriptionId);
+      checkNotNull(streamName);
+      checkNotNull(subscriptionOffset);
+      checkState(ackTimeoutSeconds > 0 && ackTimeoutSeconds < 36000);
       return new Subscription(subscriptionId, streamName, subscriptionOffset, ackTimeoutSeconds);
     }
   }

client/src/main/java/io/hstream/impl/ConsumerBuilderImpl.java

@@ -1,5 +1,7 @@
 package io.hstream.impl;
 
+import static com.google.common.base.Preconditions.*;
+
 import io.hstream.Consumer;
 import io.hstream.ConsumerBuilder;
 import io.hstream.HRecordReceiver;
@@ -48,6 +50,10 @@ public ConsumerBuilder hRecordReceiver(HRecordReceiver hRecordReceiver) {
 
   @Override
   public Consumer build() {
+    checkNotNull(grpcStub);
+    checkNotNull(grpcBlockingStub);
+    checkNotNull(subscription);
+    checkState(rawRecordReceiver != null || hRecordReceiver != null);
     return new ConsumerImpl(
         grpcStub, grpcBlockingStub, name, subscription, rawRecordReceiver, hRecordReceiver);
   }

client/src/main/java/io/hstream/impl/HStreamClientBuilderImpl.java

@@ -1,5 +1,7 @@
 package io.hstream.impl;
 
+import static com.google.common.base.Preconditions.*;
+
 import io.hstream.HStreamClient;
 import io.hstream.HStreamClientBuilder;
 
@@ -15,6 +17,7 @@ public HStreamClientBuilder serviceUrl(String serviceUrl) {
 
   @Override
   public HStreamClient build() {
+    checkNotNull(serviceUrl);
     return new HStreamClientImpl(serviceUrl);
   }
 }

client/src/main/java/io/hstream/impl/ProducerBuilderImpl.java

@@ -1,5 +1,7 @@
 package io.hstream.impl;
 
+import static com.google.common.base.Preconditions.*;
+
 import io.hstream.Producer;
 import io.hstream.ProducerBuilder;
 import io.hstream.internal.HStreamApiGrpc;
@@ -38,6 +40,8 @@ public ProducerBuilder recordCountLimit(int recordCountLimit) {
 
   @Override
   public Producer build() {
+    checkNotNull(grpcStub);
+    checkNotNull(streamName);
     return new ProducerImpl(grpcStub, streamName, enableBatch, recordCountLimit);
   }
 }

client/src/main/java/io/hstream/impl/QueryerBuilderImpl.java

@@ -1,5 +1,7 @@
 package io.hstream.impl;
 
+import static com.google.common.base.Preconditions.*;
+
 import io.hstream.*;
 import io.hstream.internal.HStreamApiGrpc;
 
@@ -29,6 +31,10 @@ public QueryerBuilder resultObserver(Observer<HRecord> resultObserver) {
 
   @Override
   public Queryer build() {
+    checkNotNull(client);
+    checkNotNull(grpcStub);
+    checkNotNull(sql);
+    checkNotNull(resultObserver);
     return new QueryerImpl(client, grpcStub, sql, resultObserver);
   }
 }