package au.gov.amsa.streams;

import java.util.Arrays;
import java.util.concurrent.atomic.AtomicBoolean;

import rx.Observable;
import rx.Observable.OnSubscribe;
import rx.Subscriber;
import rx.exceptions.CompositeException;
import rx.functions.Action0;
import rx.functions.Action1;
import rx.functions.Func0;
import rx.functions.Func1;
import rx.subscriptions.Subscriptions;

/**
 * Constructs an observable sequence that depends on a resource object.
 */
public final class OnSubscribeUsing2<T, Resource> implements OnSubscribe<T> {

	private final Func0<Resource> resourceFactory;
	private final Func1<? super Resource, ? extends Observable<? extends T>> observableFactory;
	private final Action1<? super Resource> dispose;
	private final boolean disposeEagerly;

	public OnSubscribeUsing2(
			Func0<Resource> resourceFactory,
			Func1<? super Resource, ? extends Observable<? extends T>> observableFactory,
			Action1<? super Resource> dispose, boolean disposeEagerly) {
		this.resourceFactory = resourceFactory;
		this.observableFactory = observableFactory;
		this.dispose = dispose;
		this.disposeEagerly = disposeEagerly;
	}

	@Override
	public void call(Subscriber<? super T> subscriber) {

		try {
			// create the resource
			final Resource resource = resourceFactory.call();
			// create an action that disposes only once
			final Action0 disposeOnceOnly = createOnceOnlyDisposeAction(resource);
			// dispose on unsubscription
			subscriber.add(Subscriptions.create(disposeOnceOnly));
			// create the observable
			Observable<? extends T> observable = observableFactory
			// create the observable
					.call(resource);
			// supplement with on termination disposal if requested
			if (disposeEagerly)
				observable = observable
				// dispose on completion or error
						.doOnTerminate(disposeOnceOnly);
			try {
				// start
				observable.unsafeSubscribe(subscriber);
			} catch (Throwable e) {
				Throwable disposeError = call(disposeOnceOnly);
				if (disposeError != null)
					subscriber.onError(new CompositeException(Arrays.asList(e,
							disposeError)));
				else
					// propagate error
					subscriber.onError(e);
			}
		} catch (Throwable e) {
			// then propagate error
			subscriber.onError(e);
		}
	}

	private Throwable call(final Action0 disposeOnceOnly) {
		if (disposeEagerly)
			try {
				disposeOnceOnly.call();
				return null;
			} catch (Throwable e) {
				return e;
			}
		else
			return null;
	}

	private Action0 createOnceOnlyDisposeAction(final Resource resource) {
		return new Action0() {

			final AtomicBoolean disposed = new AtomicBoolean(false);

			@Override
			public void call() {
				// only want dispose called once
				if (disposed.compareAndSet(false, true))
					dispose.call(resource);
			}
		};
	}
}

@@ -1521,16 +1521,22 @@
 able%0A%09%09%09
+final 
 Observab
@@ -1551,25 +1551,21 @@
 ends T%3E 
-observabl
+sourc
 e = obse
@@ -1624,24 +1624,69 @@
 (resource);%0A
+%09%09%09final Observable%3C? extends T%3E observable;%0A
 %09%09%09// supple
@@ -1772,25 +1772,21 @@
 vable = 
-observabl
+sourc
 e%0A%09%09%09%09//
@@ -1856,16 +1856,49 @@
 eOnly);%0A
+%09%09%09else%0A%09%09%09%09observable = source;%0A
 %09%09%09try %7B
@@ -2011,20 +2011,41 @@
 Error = 
-call
+disposeEagerlyIfRequested
 (dispose
@@ -2345,20 +2345,41 @@
 rowable 
-call
+disposeEagerlyIfRequested
 (final A

refactor slightly