package com.github.davidmoten.rx.operators;

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import rx.Observable;
import rx.Observable.OnSubscribe;
import rx.Subscriber;
import rx.Subscription;
import rx.subjects.PublishSubject;
import rx.subscriptions.Subscriptions;

public class OperationShare {

	public static <T> OnSubscribe<T> share(Observable<T> source) {
		return new Share<T>(source);
	}

	private static class Share<T> implements OnSubscribe<T> {

		private final Observable<? extends T> source;
		private final PublishSubject<T> subject = PublishSubject.create();
		private final AtomicInteger observersCount = new AtomicInteger(0);
		private final AtomicReference<Subscription> mainSubscription = new AtomicReference<Subscription>(
				Subscriptions.empty());
		private final Object lock = new Object();

		Share(Observable<? extends T> source) {
			this.source = source;
		}

		@Override
		public void call(Subscriber<? super T> subscriber) {
			subject.subscribe(subscriber);
			subscriber.add(new Subscription() {

				@Override
				public void unsubscribe() {
					synchronized (lock) {
						if (observersCount.decrementAndGet() == 0)
							mainSubscription.get().unsubscribe();
					}
				}

				@Override
				public boolean isUnsubscribed() {
					return mainSubscription.get().isUnsubscribed();
				}
			});

			synchronized (lock) {
				if (observersCount.incrementAndGet() == 1) {
					mainSubscription.set(source.subscribe(subject));
				}
			}
		}
	}
}

@@ -197,16 +197,47 @@
 scribe;%0A
+import rx.Observable.Operator;%0A
 import r
@@ -1556,11 +1556,1138 @@
 %7D%0A%09%09%7D%0A%09%7D
+%0A%0A%09public static class ShareOperator%3CT%3E implements Operator%3CT, T%3E %7B%0A%0A%09%09private final Observable%3C? extends T%3E source;%0A%09%09private final PublishSubject%3CT%3E subject = PublishSubject.create();%0A%09%09private final AtomicInteger observersCount = new AtomicInteger(0);%0A%09%09private final AtomicReference%3CSubscription%3E mainSubscription = new AtomicReference%3CSubscription%3E(%0A%09%09%09%09Subscriptions.empty());%0A%09%09private final Object lock = new Object();%0A%0A%09%09ShareOperator(Observable%3C? extends T%3E source) %7B%0A%09%09%09this.source = source;%0A%09%09%7D%0A%0A%09%09@Override%0A%09%09public Subscriber%3C? super T%3E call(final Subscriber%3C? super T%3E subscriber) %7B%0A%09%09%09subject.subscribe(subscriber);%0A%09%09%09synchronized (lock) %7B%0A%09%09%09%09if (observersCount.incrementAndGet() == 1) %7B%0A%09%09%09%09%09mainSubscription.set(source.subscribe(subject));%0A%09%09%09%09%7D%0A%09%09%09%7D%0A%09%09%09Subscriber%3CT%3E result = new Subscriber%3CT%3E() %7B%0A%0A%09%09%09%09@Override%0A%09%09%09%09public void onCompleted() %7B%0A%09%09%09%09%09subscriber.onCompleted();%0A%09%09%09%09%7D%0A%0A%09%09%09%09@Override%0A%09%09%09%09public void onError(Throwable e) %7B%0A%09%09%09%09%09subscriber.onError(e);%0A%09%09%09%09%7D%0A%0A%09%09%09%09@Override%0A%09%09%09%09public void onNext(T t) %7B%0A%09%09%09%09%09subscriber.onNext(t);%0A%09%09%09%09%7D%0A%09%09%09%7D;%0A%09%09%09subscriber.add(result);%0A%09%09%09return result;%0A%09%09%7D%0A%0A%09%7D
 %0A%7D%0A

add operator