/**
 * Copyright 2014 Netflix, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package rx.reactivestreams;

import org.reactivestreams.Publisher;
import org.testng.annotations.Test;
import rx.Observable;
import rx.RxReactiveStreams;
import rx.reactivestreams.test.RsSubscriber;
import rx.reactivestreams.test.RxSubscriber;
import rx.reactivestreams.test.IterablePublisher;

import java.util.Arrays;

import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.AssertJUnit.assertEquals;

public class NonTckTest {

    @Test
    public void canSubscribeToObservableAsPublisher() {
        Observable<Integer> observable = Observable.just(1, 2, 3);
        Publisher<Integer> publisher = RxReactiveStreams.toPublisher(observable);
        RsSubscriber<Integer> subscriber = new RsSubscriber<Integer>();
        publisher.subscribe(subscriber);

        assertEquals("no items sent", 0, subscriber.received.size());
        subscriber.subscription.request(1);
        assertEquals("one item sent", 1, subscriber.received.size());
        subscriber.subscription.request(2);
        assertEquals("two items sent", 3, subscriber.received.size());
        assertTrue(subscriber.complete);
        assertNull(subscriber.error);
    }

    @Test
    public void canSubscribeToPublisherAsObservable() {
        Publisher<Integer> publisher = new IterablePublisher<Integer>(Arrays.asList(1, 2, 3));
        Observable<Integer> observable = RxReactiveStreams.toObservable(publisher);
        RxSubscriber<Integer> subscriber = new RxSubscriber<Integer>(0);
        observable.subscribe(subscriber);

        assertEquals("no items sent", 0, subscriber.received.size());
        subscriber.makeRequest(1);
        assertEquals("one item sent", 1, subscriber.received.size());
        subscriber.makeRequest(2);
        assertEquals("two items sent", 3, subscriber.received.size());
        assertTrue(subscriber.complete);
        assertNull(subscriber.error);
    }

    @Test
    public void rxSubscriberNotMakingInitialRequestConumesPublisher() {
        Publisher<Integer> publisher = new IterablePublisher<Integer>(Arrays.asList(1, 2, 3));
        Observable<Integer> observable = RxReactiveStreams.toObservable(publisher);
        RxSubscriber<Integer> subscriber = new RxSubscriber<Integer>(-1); // -1 means no initial request
        observable.subscribe(subscriber);

        assertEquals("all items sent", 3, subscriber.received.size());
        assertTrue(subscriber.complete);
        assertNull(subscriber.error);
    }

}
@@ -741,16 +741,38 @@
 treams;%0A
+import rx.Subscriber;%0A
 import r
@@ -962,26 +962,17 @@
 .Assert.
-assertNull
+*
 ;%0Aimport
@@ -1000,19 +1000,26 @@
 sert
+JUnit
 .assert
-True
+Equals
 ;%0Aim
@@ -1034,43 +1034,89 @@
 tic 
-org.testng.AssertJUnit.assertEquals
+rx.RxReactiveStreams.toObservable;%0Aimport static rx.RxReactiveStreams.toPublisher
 ;%0A%0Ap
@@ -1313,34 +1313,16 @@
 isher = 
-RxReactiveStreams.
 toPublis
@@ -2039,34 +2039,16 @@
 vable = 
-RxReactiveStreams.
 toObserv
@@ -2765,34 +2765,16 @@
 vable = 
-RxReactiveStreams.
 toObserv
@@ -3095,9 +3095,1680 @@
 %0A    %7D%0A%0A
+    @Test void errorStatePublisherSendsSingleErrorPostSubscribe() %7B%0A        RsSubscriber%3CInteger%3E subscriber = new RsSubscriber%3CInteger%3E();%0A        toPublisher(Observable.%3CInteger%3Eerror(new RuntimeException(%22!%22))).subscribe(subscriber);%0A%0A        // An error state observable always allows subscription, but then immediately sends onError.%0A        // The spec suggests not trying to subscribe but immediately firing onError without an onSubscribe.%0A        // However, this isn't a spec violation.%0A        assertNotNull(subscriber.subscription);%0A%0A        assertFalse(subscriber.complete);%0A        assertEquals(subscriber.error.getClass(), RuntimeException.class);%0A        assertEquals(subscriber.error.getMessage(), %22!%22);%0A    %7D%0A%0A    @Test void rxFailingOnSubscribeSendsSingleErrorPostSubscribe() %7B%0A        RsSubscriber%3CInteger%3E subscriber = new RsSubscriber%3CInteger%3E();%0A        Observable%3CInteger%3E error = Observable.create(new Observable.OnSubscribe%3CInteger%3E() %7B%0A            @Override%0A            public void call(Subscriber%3C? super Integer%3E subscriber) %7B%0A                throw new RuntimeException(%22!%22);%0A            %7D%0A        %7D);%0A%0A        toPublisher(error).subscribe(subscriber);%0A%0A        // An error state observable always allows subscription, but then immediately sends onError.%0A        // The spec suggests not trying to subscribe but immediately firing onError without an onSubscribe.%0A        // However, this isn't a spec violation.%0A        assertNotNull(subscriber.subscription);%0A%0A        assertFalse(subscriber.complete);%0A        assertEquals(subscriber.error.getClass(), RuntimeException.class);%0A        assertEquals(subscriber.error.getMessage(), %22!%22);%0A    %7D%0A%0A
 %7D

Add some coverage for error state observables.