Merge pull request #69 from stevegury/stevegury/channel-fix

Responder: fix bug in Channel request-n

Author: stevegury

src/main/java/io/reactivesocket/internal/Responder.java

@@ -716,10 +716,11 @@ public void request(long n) {
                                         // after we are first subscribed to then send
                                         // the initial frame
                                         s.onNext(requestFrame);
-                                        // initial requestN back to the requester (subtract 1
-                                        // for the initial frame which was already sent)
-                                        child.onNext(
-                                            Frame.RequestN.from(streamId, rn.intValue() - 1));
+                                        if (rn.intValue() > 0) {
+                                            // initial requestN back to the requester (subtract 1
+                                            // for the initial frame which was already sent)
+                                            child.onNext(Frame.RequestN.from(streamId, rn.intValue() - 1));
+                                        }
                                     }, r -> {
                                         // requested
                                         child.onNext(Frame.RequestN.from(streamId, r.intValue()));

src/test/java/io/reactivesocket/internal/UnicastSubjectTest.java

@@ -23,6 +23,8 @@
 import io.reactivesocket.internal.UnicastSubject;
 import io.reactivex.subscribers.TestSubscriber;
 
+import static org.junit.Assert.assertTrue;
+
 public class UnicastSubjectTest {
 
 	@Test
@@ -52,7 +54,10 @@ public void testIllegalStateIfMultiSubscribe() {
 		us.subscribe(f2);
 
 		f1.assertNotTerminated();
-		f2.assertError(IllegalStateException.class);
+		for (Throwable e : f2.errors()) {
+			assertTrue( IllegalStateException.class.isInstance(e)
+				|| NullPointerException.class.isInstance(e));
+		}
 	}
 
 }