1.x: increase Coverage of some classes

src/main/java/rx/internal/util/UtilityFunctions.java

@@ -59,7 +59,7 @@ public T call(T o) {
         };
     }
 
-    private enum AlwaysTrue implements Func1<Object, Boolean> {
+    enum AlwaysTrue implements Func1<Object, Boolean> {
         INSTANCE;
 
         @Override
@@ -68,7 +68,7 @@ public Boolean call(Object o) {
         }
     }
 
-    private enum AlwaysFalse implements Func1<Object, Boolean> {
+    enum AlwaysFalse implements Func1<Object, Boolean> {
         INSTANCE;
 
         @Override

src/main/java/rx/schedulers/ImmediateScheduler.java

@@ -24,7 +24,7 @@
 // Class was part of public API.
 public final class ImmediateScheduler extends Scheduler { // NOPMD 
     private ImmediateScheduler() {
-        throw new AssertionError();
+        throw new IllegalStateException("No instances!");
     }
 
     @Override

src/main/java/rx/schedulers/NewThreadScheduler.java

@@ -24,7 +24,7 @@
 // Class was part of public API.
 public final class NewThreadScheduler extends Scheduler { // NOPMD 
     private NewThreadScheduler() {
-        throw new AssertionError();
+        throw new IllegalStateException("No instances!");
     }
 
     @Override

src/main/java/rx/schedulers/Timestamped.java

@@ -61,14 +61,7 @@ public boolean equals(Object obj) {
         if (timestampMillis != other.timestampMillis) {
             return false;
         }
-        if (value == null) {
-            if (other.value != null) {
-                return false;
-            }
-        } else if (!value.equals(other.value)) {
-            return false;
-        }
-        return true;
+        return (value == other.value) || (value != null && value.equals(other.value));
     }
 
     @Override

src/main/java/rx/schedulers/TrampolineScheduler.java

@@ -24,7 +24,7 @@
 // Class was part of public API.
 public final class TrampolineScheduler extends Scheduler { // NOPMD 
     private TrampolineScheduler() {
-        throw new AssertionError();
+        throw new IllegalStateException("No instances!");
     }
 
     @Override

src/test/java/rx/internal/operators/DeferredScalarSubscriberTest.java

@@ -343,8 +343,12 @@ public void call() {
     public void emissionRequestRace2() {
         Worker w = Schedulers.io().createWorker();
         Worker w2 = Schedulers.io().createWorker();
+        int m = 10000;
+        if (Runtime.getRuntime().availableProcessors() < 3) {
+            m = 1000;
+        }
         try {
-            for (int i = 0; i < 10000; i++) {
+            for (int i = 0; i < m; i++) {
 
                 final TestSubscriber<Integer> ts = TestSubscriber.create(0L);
                 TestingDeferredScalarSubscriber ds = new TestingDeferredScalarSubscriber(ts);

src/test/java/rx/internal/producers/ProducerObserverArbiterTest.java

@@ -16,6 +16,7 @@
 package rx.internal.producers;
 
 import org.junit.*;
+import static org.junit.Assert.*;
 
 import rx.*;
 import rx.exceptions.TestException;
@@ -38,6 +39,7 @@ public void negativeRequestThrows() {
     public void nullProducerAccepted() {
         ProducerObserverArbiter<Integer> pa = new ProducerObserverArbiter<Integer>(Subscribers.empty());
         pa.setProducer(null);
+        pa.request(5);
     }
 
     public void failedRequestUnlocksEmitting() {
@@ -160,4 +162,78 @@ public void onNext(Integer t) {
         }
     }
 
+    @Test
+    public void onNextRequests() {
+        @SuppressWarnings("rawtypes")
+        final ProducerObserverArbiter[] o = { null };
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>() {
+            @Override
+            public void onNext(Integer t) {
+                o[0].request(1);
+            }
+        };
+        ProducerObserverArbiter<Integer> poa = new ProducerObserverArbiter<Integer>(ts);
+        poa.request(1);
+        o[0] = poa;
+        try {
+            poa.onNext(1);
+        } catch (TestException ex) {
+            // expected
+        }
+        assertEquals(1, poa.requested);
+    }
+
+    @Test
+    public void requestIsCapped() {
+        ProducerObserverArbiter<Integer> poa = new ProducerObserverArbiter<Integer>(new TestSubscriber<Integer>());
+
+        poa.request(Long.MAX_VALUE - 1);
+        poa.request(2);
+
+        assertEquals(Long.MAX_VALUE, Long.MAX_VALUE);
+    }
+
+    @Test
+    public void onNextChangesProducerNull() {
+        @SuppressWarnings("rawtypes")
+        final ProducerObserverArbiter[] o = { null };
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>() {
+            @Override
+            public void onNext(Integer t) {
+                o[0].setProducer(null);
+            }
+        };
+        ProducerObserverArbiter<Integer> poa = new ProducerObserverArbiter<Integer>(ts);
+        poa.request(1);
+        o[0] = poa;
+        try {
+            poa.onNext(1);
+        } catch (TestException ex) {
+            // expected
+        }
+        assertNull(poa.currentProducer);
+    }
+
+    @Test
+    public void onNextChangesProducerNotNull() {
+        @SuppressWarnings("rawtypes")
+        final ProducerObserverArbiter[] o = { null };
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>() {
+            @SuppressWarnings("unchecked")
+            @Override
+            public void onNext(Integer t) {
+                o[0].setProducer(new SingleProducer<Integer>(o[0].child, 2));
+            }
+        };
+        ProducerObserverArbiter<Integer> poa = new ProducerObserverArbiter<Integer>(ts);
+        poa.request(1);
+        o[0] = poa;
+        try {
+            poa.onNext(1);
+        } catch (TestException ex) {
+            // expected
+        }
+        assertNotNull(poa.currentProducer);
+    }
+
 }

src/test/java/rx/internal/producers/ProducersTest.java

@@ -15,6 +15,7 @@
  */
 package rx.internal.producers;
 
+import static org.junit.Assert.*;
 import static org.mockito.Mockito.*;
 
 import java.util.*;
@@ -25,8 +26,10 @@
 import rx.*;
 import rx.Observable;
 import rx.Observable.*;
+import rx.exceptions.MissingBackpressureException;
 import rx.Observer;
 import rx.functions.*;
+import rx.internal.util.unsafe.SpscArrayQueue;
 import rx.observers.TestSubscriber;
 import rx.schedulers.*;
 import rx.subscriptions.SerialSubscription;
@@ -426,4 +429,95 @@ public void onNext(Integer t) {
             }
         });
     }
+
+    @Test
+    public void queuedProducerRequestNegative() {
+        QueuedProducer<Integer> qp = new QueuedProducer<Integer>(new TestSubscriber<Integer>());
+        try {
+            qp.request(-99);
+        } catch (IllegalArgumentException ex) {
+            assertEquals("n >= 0 required", ex.getMessage());
+        }
+    }
+
+    @Test
+    public void queuedProducerOfferNull() {
+        TestSubscriber<Integer> ts = TestSubscriber.create();
+        QueuedProducer<Integer> qp = new QueuedProducer<Integer>(ts);
+        qp.offer(null);
+
+        qp.request(1);
+
+        ts.assertValue(null);
+    }
+
+    @Test
+    public void queuedProducerFull() {
+        TestSubscriber<Integer> ts = TestSubscriber.create();
+        QueuedProducer<Integer> qp = new QueuedProducer<Integer>(ts, new SpscArrayQueue<Object>(1));
+        assertTrue(qp.offer(1));
+        assertFalse(qp.offer(2));
+
+        qp.request(1);
+
+        ts.assertValue(1);
+    }
+
+    @Test
+    public void queuedProducerOnNextFull() {
+        TestSubscriber<Integer> ts = TestSubscriber.create();
+        QueuedProducer<Integer> qp = new QueuedProducer<Integer>(ts, new SpscArrayQueue<Object>(1));
+
+        qp.onNext(1);
+        qp.onNext(2);
+
+        qp.request(1);
+
+        ts.assertError(MissingBackpressureException.class);
+    }
+
+    @Test
+    public void queuedProducerOnNextFullWithNull() {
+        TestSubscriber<Integer> ts = TestSubscriber.create();
+        QueuedProducer<Integer> qp = new QueuedProducer<Integer>(ts, new SpscArrayQueue<Object>(1));
+
+        qp.onNext(1);
+        qp.onNext(null);
+
+        qp.request(1);
+
+        ts.assertError(MissingBackpressureException.class);
+    }
+
+    @Test
+    public void queuedProducerRequestCompletes() {
+        TestSubscriber<Integer> ts = TestSubscriber.create();
+        QueuedProducer<Integer> qp = new QueuedProducer<Integer>(ts, new SpscArrayQueue<Object>(1));
+
+        qp.onNext(1);
+        qp.onCompleted();
+
+        qp.request(2);
+
+        ts.assertValue(1);
+        ts.assertNoErrors();
+        ts.assertCompleted();
+    }
+
+    @Test
+    public void queuedProducerUnsubscribed() {
+        TestSubscriber<Integer> ts = TestSubscriber.create();
+        ts.unsubscribe();
+        QueuedProducer<Integer> qp = new QueuedProducer<Integer>(ts, new SpscArrayQueue<Object>(1));
+
+        qp.onNext(1);
+        qp.onCompleted();
+
+        qp.request(2);
+
+        ts.assertNoValues();
+        ts.assertNoErrors();
+        ts.assertNotCompleted();
+    }
+
 }

src/test/java/rx/internal/util/ExceptionUtilsTest.java

@@ -15,6 +15,8 @@
  */
 package rx.internal.util;
 
+import static org.junit.Assert.*;
+
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.*;
@@ -51,4 +53,15 @@ public void isTerminated() {
         Assert.assertTrue(ExceptionsUtils.isTerminated(error));
         Assert.assertTrue(ExceptionsUtils.isTerminated(error.get()));
     }
+
+    @Test
+    public void utilityEnum() {
+        assertEquals(0, ExceptionsUtils.values().length);
+        try {
+            ExceptionsUtils.valueOf("INSTANCE");
+            fail("Failed to throw IllegalArgumentException");
+        } catch (IllegalArgumentException ex) {
+            // expected
+        }
+    }
 }

src/test/java/rx/internal/util/UtilityFunctionsTest.java

@@ -16,6 +16,7 @@
 package rx.internal.util;
 
 import org.junit.Test;
+import static org.junit.Assert.*;
 
 import rx.TestUtil;
 
@@ -24,4 +25,19 @@ public class UtilityFunctionsTest {
     public void constructorShouldBePrivate() {
         TestUtil.checkUtilityClass(UtilityFunctions.class);
     }
+
+    @Test
+    public void alwaysFalse() {
+        assertEquals(1, UtilityFunctions.AlwaysFalse.values().length);
+        assertNotNull(UtilityFunctions.AlwaysFalse.valueOf("INSTANCE"));
+        assertFalse(UtilityFunctions.alwaysFalse().call(1));
+    }
+
+    @Test
+    public void alwaysTrue() {
+        assertEquals(1, UtilityFunctions.AlwaysTrue.values().length);
+        assertNotNull(UtilityFunctions.AlwaysTrue.valueOf("INSTANCE"));
+        assertTrue(UtilityFunctions.alwaysTrue().call(1));
+    }
+
 }