package com.github.davidmoten.rx.operators;

import com.github.davidmoten.rx.Emitter;
import com.github.davidmoten.rx.EmitterFactory;
import com.github.davidmoten.rx.StandardProducer;
import com.github.davidmoten.rx.util.Next;
import com.github.davidmoten.rx.util.Optional;

import rx.Observable.OnSubscribe;
import rx.Producer;
import rx.Subscriber;

public abstract class AbstractOnSubscribe<T> implements OnSubscribe<T>, Next<T> {

    @Override
    public void call(Subscriber<? super T> subscriber) {
        subscriber.setProducer(createProducer(subscriber));
    }

    private Producer createProducer(Subscriber<? super T> subscriber) {
        EmitterFactory<T> factory = standardEmitterFactory(this);
        return new StandardProducer<T>(subscriber, factory);
    }

    private static <T> EmitterFactory<T> standardEmitterFactory(final Next<T> a) {
        return new EmitterFactory<T>() {

            @Override
            public Emitter create(final Subscriber<? super T> subscriber) {
                return createEmitter(a, subscriber);
            }

        };
    }

    private static <T> Emitter createEmitter(final Next<T> a, final Subscriber<? super T> subscriber) {
        return new Emitter() {

            private boolean completed = false;

            @Override
            public void emitAll() {
                while (!subscriber.isUnsubscribed() && !completed()) {
                    emitOne();
                }
            }

            @Override
            public void emitOne() {
                Optional<T> next = a.next();
                if (next.isPresent()) {
                    subscriber.onNext(next.get());
                    if (completed())
                        subscriber.onCompleted();
                } else {
                    subscriber.onCompleted();
                    completed = true;
                }
            }

            @Override
            public boolean completed() {
                return completed;
            }
        };
    }
}

@@ -38,16 +38,93 @@
 ators;%0A%0A
+import rx.Observable.OnSubscribe;%0Aimport rx.Producer;%0Aimport rx.Subscriber;%0A%0A
 import c
@@ -349,85 +349,8 @@
 l;%0A%0A
-import rx.Observable.OnSubscribe;%0Aimport rx.Producer;%0Aimport rx.Subscriber;%0A%0A
 publ
@@ -1388,18 +1388,16 @@
 ompleted
-()
 ) %7B%0A    
@@ -1433,32 +1433,200 @@
               %7D%0A
+                if (!subscriber.isUnsubscribed() && !completed) %7B%0A                    subscriber.onCompleted();%0A                    completed = true;%0A                %7D%0A
             %7D%0A%0A 
@@ -1763,16 +1763,48 @@
 resent()
+ && !subscriber.isUnsubscribed()
 ) %7B%0A    
@@ -1883,16 +1883,46 @@
 ompleted
+ && !subscriber.isUnsubscribed
 ())%0A    
@@ -1989,16 +1989,50 @@
   %7D else
+ if (!subscriber.isUnsubscribed())
  %7B%0A     

add more unsub checks