package reactivestreams.commons.processor;

import org.junit.Assert;
import org.junit.Test;
import reactivestreams.commons.test.TestSubscriber;

public class SimpleProcessorTest {

    @Test(expected = NullPointerException.class)
    public void onNextNull() {
        new SimpleProcessor<Integer>().onNext(null);
    }

    @Test(expected = NullPointerException.class)
    public void onErrorNull() {
        new SimpleProcessor<Integer>().onError(null);
    }

    @Test(expected = NullPointerException.class)
    public void onSubscribeNull() {
        new SimpleProcessor<Integer>().onSubscribe(null);
    }

    @Test(expected = NullPointerException.class)
    public void subscribeNull() {
        new SimpleProcessor<Integer>().subscribe(null);
    }

    @Test
    public void normal() {
        TestSubscriber<Integer> ts = new TestSubscriber<>();

        SimpleProcessor<Integer> tp = SimpleProcessor.create();

        tp.subscribe(ts);

        Assert.assertTrue("No subscribers?", tp.hasSubscribers());
        Assert.assertFalse("Completed?", tp.hasCompleted());
        Assert.assertNull("Has error?", tp.getError());
        Assert.assertFalse("Has error?", tp.hasError());

        ts.assertNoValues()
          .assertNoError()
          .assertNotComplete();

        tp.onNext(1);
        tp.onNext(2);

        ts.assertValues(1, 2)
          .assertNotComplete()
          .assertNoError();

        tp.onNext(3);
        tp.onComplete();

        Assert.assertFalse("Subscribers present?", tp.hasSubscribers());
        Assert.assertTrue("Not completed?", tp.hasCompleted());
        Assert.assertNull("Has error?", tp.getError());
        Assert.assertFalse("Has error?", tp.hasError());


        ts.assertValues(1, 2, 3)
          .assertComplete()
          .assertNoError();
    }

    @Test
    public void normalBackpressured() {
        TestSubscriber<Integer> ts = new TestSubscriber<>(0);

        SimpleProcessor<Integer> tp = SimpleProcessor.create();

        tp.subscribe(ts);

        Assert.assertTrue("No subscribers?", tp.hasSubscribers());
        Assert.assertFalse("Completed?", tp.hasCompleted());
        Assert.assertNull("Has error?", tp.getError());
        Assert.assertFalse("Has error?", tp.hasError());

        ts.assertNoValues()
          .assertNoError()
          .assertNotComplete();

        ts.request(10);

        tp.onNext(1);
        tp.onNext(2);
        tp.onComplete();

        Assert.assertFalse("Subscribers present?", tp.hasSubscribers());
        Assert.assertTrue("Not completed?", tp.hasCompleted());
        Assert.assertNull("Has error?", tp.getError());
        Assert.assertFalse("Has error?", tp.hasError());

        ts.assertValues(1, 2)
          .assertNoError()
          .assertComplete();
    }

    @Test
    public void notEnoughRequests() {

        TestSubscriber<Integer> ts = new TestSubscriber<>(0);

        SimpleProcessor<Integer> tp = SimpleProcessor.create();

        tp.subscribe(ts);

        ts.assertNoValues()
          .assertNoError()
          .assertNotComplete();

        ts.request(1);

        tp.onNext(1);
        tp.onNext(2);
        tp.onComplete();

        Assert.assertFalse("Subscribers present?", tp.hasSubscribers());
        Assert.assertTrue("Not completed?", tp.hasCompleted());
        Assert.assertNull("Has error?", tp.getError());
        Assert.assertFalse("Has error?", tp.hasError());

        ts.assertValues(1)
          .assertError(IllegalStateException.class)
          .assertNotComplete();
    }

    @Test
    public void error() {
        TestSubscriber<Integer> ts = new TestSubscriber<>();

        SimpleProcessor<Integer> tp = SimpleProcessor.create();

        tp.subscribe(ts);

        Assert.assertTrue("No subscribers?", tp.hasSubscribers());
        Assert.assertFalse("Completed?", tp.hasCompleted());
        Assert.assertNull("Has error?", tp.getError());
        Assert.assertFalse("Has error?", tp.hasError());

        ts.assertNoValues()
          .assertNoError()
          .assertNotComplete();

        tp.onNext(1);
        tp.onNext(2);

        ts.assertValues(1, 2)
          .assertNotComplete()
          .assertNoError();

        tp.onNext(3);
        tp.onError(new RuntimeException("forced failure"));

        Assert.assertFalse("Subscribers present?", tp.hasSubscribers());
        Assert.assertFalse("Completed?", tp.hasCompleted());
        Assert.assertNotNull("Has error?", tp.getError());
        Assert.assertTrue("No error?", tp.hasError());

        Throwable e = tp.getError();
        Assert.assertTrue("Wrong exception? " + e, RuntimeException.class.isInstance(e));
        Assert.assertEquals("forced failure", e.getMessage());

        ts.assertValues(1, 2, 3)
          .assertNotComplete()
          .assertError(RuntimeException.class)
          .assertErrorMessage("forced failure");
    }

    @Test
    public void terminatedWithError() {
        TestSubscriber<Integer> ts = new TestSubscriber<>();

        SimpleProcessor<Integer> tp = SimpleProcessor.create();
        tp.onError(new RuntimeException("forced failure"));

        tp.subscribe(ts);

        Assert.assertFalse("Subscribers present?", tp.hasSubscribers());
        Assert.assertFalse("Completed?", tp.hasCompleted());
        Assert.assertNotNull("No error?", tp.getError());
        Assert.assertTrue("No error?", tp.hasError());

        Throwable e = tp.getError();
        Assert.assertTrue("Wrong exception? " + e, RuntimeException.class.isInstance(e));
        Assert.assertEquals("forced failure", e.getMessage());

        ts.assertNoValues()
          .assertNotComplete()
          .assertError(RuntimeException.class)
          .assertErrorMessage("forced failure");
    }

    @Test
    public void terminatedNormally() {
        TestSubscriber<Integer> ts = new TestSubscriber<>();

        SimpleProcessor<Integer> tp = SimpleProcessor.create();
        tp.onComplete();

        tp.subscribe(ts);

        Assert.assertFalse("Subscribers present?", tp.hasSubscribers());
        Assert.assertTrue("Not completed?", tp.hasCompleted());
        Assert.assertNull("Has error?", tp.getError());
        Assert.assertFalse("Has error?", tp.hasError());

        ts.assertNoValues()
          .assertComplete()
          .assertNoError();
    }

    @Test
    public void subscriberAlreadyCancelled() {
        TestSubscriber<Integer> ts = new TestSubscriber<>();
        ts.cancel();

        SimpleProcessor<Integer> tp = SimpleProcessor.create();

        tp.subscribe(ts);

        Assert.assertFalse("Subscribers present?", tp.hasSubscribers());

        tp.onNext(1);


        ts.assertNoValues()
          .assertNotComplete()
          .assertNoError();
    }

    @Test
    public void subscriberCancels() {
        TestSubscriber<Integer> ts = new TestSubscriber<>();

        SimpleProcessor<Integer> tp = SimpleProcessor.create();

        tp.subscribe(ts);

        Assert.assertTrue("No Subscribers present?", tp.hasSubscribers());

        tp.onNext(1);

        ts.assertValue(1)
          .assertNoError()
          .assertNotComplete();

        ts.cancel();

        Assert.assertFalse("Subscribers present?", tp.hasSubscribers());

        tp.onNext(2);

        ts.assertValue(1)
          .assertNotComplete()
          .assertNoError();
    }

}

@@ -254,36 +254,32 @@
 ull() %7B%0A        
-new 
 SimpleProcessor%3C
@@ -269,32 +269,33 @@
  SimpleProcessor
+.
 %3CInteger%3E().onNe
@@ -283,24 +283,30 @@
 or.%3CInteger%3E
+create
 ().onNext(nu
@@ -398,36 +398,32 @@
 ull() %7B%0A        
-new 
 SimpleProcessor%3C
@@ -413,32 +413,33 @@
  SimpleProcessor
+.
 %3CInteger%3E().onEr
@@ -427,24 +427,30 @@
 or.%3CInteger%3E
+create
 ().onError(n
@@ -547,36 +547,32 @@
 ull() %7B%0A        
-new 
 SimpleProcessor%3C
@@ -562,32 +562,33 @@
  SimpleProcessor
+.
 %3CInteger%3E().onSu
@@ -580,16 +580,22 @@
 Integer%3E
+create
 ().onSub
@@ -706,20 +706,16 @@
         
-new 
 SimplePr
@@ -713,32 +713,33 @@
  SimpleProcessor
+.
 %3CInteger%3E().subs
@@ -731,16 +731,22 @@
 Integer%3E
+create
 ().subsc

Use static for UnicastProcessor/SimpleProcessor