Allow to set arbitrary Integer values to FetchSize

src/main/java/com/github/davidmoten/rx/jdbc/QueryContext.java

@@ -13,13 +13,13 @@ final class QueryContext {
 
     private final Database db;
     private final int batchSize;
-    private final int fetchSize;
+    private final Integer fetchSize;
 
     QueryContext(Database db) {
-        this(db, 1, 0);
+        this(db, 1, null);
     }
 
-    public QueryContext(Database db, int batchSize, int fetchSize) {
+    public QueryContext(Database db, int batchSize, Integer fetchSize) {
         this.db = db;
         this.batchSize = batchSize;
         this.fetchSize = fetchSize;
@@ -79,11 +79,11 @@ int batchSize() {
         return batchSize;
     }
 
-    QueryContext fetchSize(int fetchSize) {
+    QueryContext fetchSize(Integer fetchSize) {
         return new QueryContext(db, batchSize, fetchSize);
     }
 
-    int fetchSize() {
+    Integer fetchSize() {
         return fetchSize;
     }
 

src/main/java/com/github/davidmoten/rx/jdbc/QuerySelect.java

@@ -144,13 +144,11 @@ private <T> Observable<T> executeOnce(final List<Parameter> params,
      */
     public static final class Builder {
 
-        private static final int DEFAULT_FETCH_SIZE = 0;
-
         /**
          * Builds the standard stuff.
          */
         private final QueryBuilder builder;
-        private int fetchSize = DEFAULT_FETCH_SIZE;
+        private Integer fetchSize;
 
         /**
          * The {@link ResultSet} is transformed before use.
@@ -282,7 +280,7 @@ public <T> Observable<T> get(ResultSetMapper<? extends T> function) {
         <T> Observable<T> get(ResultSetMapper<? extends T> function, QueryBuilder builder,
                 Func1<ResultSet, ? extends ResultSet> resultSetTransform) {
             final QueryContext ctxt;
-            if (fetchSize > 1) {
+            if (fetchSize != null) {
                 ctxt = builder.context().fetchSize(fetchSize);
             } else {
                 ctxt = builder.context();

src/main/java/com/github/davidmoten/rx/jdbc/QuerySelectOnSubscribe.java

@@ -107,7 +107,7 @@ private void connectAndPrepareStatement(Subscriber<? super T> subscriber, State
             log.debug("preparing statement,sql={}", query.sql());
             state.ps = state.con.prepareStatement(query.sql(), ResultSet.TYPE_FORWARD_ONLY,
                     ResultSet.CONCUR_READ_ONLY);
-            if (query.context().fetchSize() > 0) {
+            if (query.context().fetchSize() != null) {
                 state.ps.setFetchSize(query.context().fetchSize());
             }
             log.debug("setting parameters");

src/test/java/com/github/davidmoten/rx/jdbc/FetchSizeTest.java

@@ -1,22 +1,31 @@
 package com.github.davidmoten.rx.jdbc;
 
+import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.verify;
 
 import java.sql.Connection;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+
+import org.junit.Before;
 import org.junit.Test;
 import org.mockito.Mockito;
 import rx.schedulers.Schedulers;
 
 public class FetchSizeTest {
 
-  @Test
-  public void testMocked() throws SQLException {
-    String sql = "select name, score from people";
+  private String sql;
+
+  private Database db;
+
+  private PreparedStatement ps;
+
+  @Before
+  public void setup() throws Exception {
+    sql = "select name, score from people";
     Connection con = Mockito.mock(Connection.class);
-    PreparedStatement ps = Mockito.mock(PreparedStatement.class);
+    ps = Mockito.mock(PreparedStatement.class);
     Mockito.when(con.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY,
         ResultSet.CONCUR_READ_ONLY)).thenReturn(ps);
     ResultSet resultSet = Mockito.mock(ResultSet.class);
@@ -25,8 +34,11 @@ public void testMocked() throws SQLException {
     Mockito.when(con.getAutoCommit()).thenReturn(false);
     Mockito.when(con.isClosed()).thenReturn(false);
     ConnectionProvider cp = createConnectionProvider(con);
-    Database db = Database.from(cp);
+    db = Database.from(cp);
+  }
 
+  @Test
+  public void testSetFetchSizePositive() throws SQLException {
     db.select(sql) //
         // set batch size
         .fetchSize(3)
@@ -40,6 +52,49 @@ public void testMocked() throws SQLException {
     verify(ps, Mockito.times(1)).setFetchSize(3);
   }
 
+  @Test
+  public void testSetFetchSizeMinInteger() throws SQLException {
+    db.select(sql) //
+        // set batch size
+        .fetchSize(Integer.MIN_VALUE)
+        // go
+        .count()
+        //
+        .subscribeOn(Schedulers.immediate())
+        // go
+        .subscribe();
+
+    verify(ps, Mockito.times(1)).setFetchSize(Integer.MIN_VALUE);
+  }
+
+  @Test
+  public void testSetFetchSizeMaxInteger() throws SQLException {
+    db.select(sql) //
+        // set batch size
+        .fetchSize(Integer.MAX_VALUE)
+        // go
+        .count()
+        //
+        .subscribeOn(Schedulers.immediate())
+        // go
+        .subscribe();
+
+    verify(ps, Mockito.times(1)).setFetchSize(Integer.MAX_VALUE);
+  }
+
+  @Test
+  public void testNotSetFetchSize() throws SQLException {
+    db.select(sql) //
+        // go
+        .count()
+        //
+        .subscribeOn(Schedulers.immediate())
+        // go
+        .subscribe();
+
+    verify(ps, Mockito.never()).setFetchSize(any(Integer.class));
+  }
+
   private static ConnectionProvider createConnectionProvider(final Connection con) {
     return new ConnectionProvider() {