package com.github.davidmoten.rx.internal.operators;

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

/**
 * Wraps a Queue (like a file based queue) to provide concurrency guarantees
 * around calls to the close() method. Extends AtomicBoolean to save allocation.
 * The AtomicBoolean represents the closed status of the queue.
 * 
 * @param <T>
 *            type of item on queue
 */
final class QueueWithResourcesNonBlockingUnsubscribe<T> extends AbstractQueueWithResources<T> {

	// currentCalls and this used to manage visibility
	private boolean closing;

	// ensures queue.close() doesn't occur until outstanding peek(),offer(),
	// poll(), isEmpty() calls have finished. When currentCalls is zero a
	// close request can be actioned.
	private final AtomicInteger currentCalls = new AtomicInteger(0);

	private final AtomicBoolean unsubscribed;
	
	QueueWithResourcesNonBlockingUnsubscribe(QueueWithResources<T> queue) {
		super(queue);
		this.closing = false;
		this.unsubscribed = new AtomicBoolean(false);
	}

	@Override
	public T poll() {
		try {
			if (closing) {
				return null;
			} else {
				try {
					currentCalls.incrementAndGet();
					return super.poll();
				} finally {
					currentCalls.decrementAndGet();
				}
			}
		} finally {
			checkClosed();
		}
	}

	@Override
	public boolean offer(T t) {
		try {
			if (closing) {
				return true;
			} else {
				try {
					currentCalls.incrementAndGet();
					return super.offer(t);
				} finally {
					currentCalls.decrementAndGet();
				}
			}
		} finally {
			checkClosed();
		}
	}

	@Override
	public boolean isEmpty() {
		try {
			if (closing) {
				return true;
			} else {
				try {
					currentCalls.incrementAndGet();
					return super.isEmpty();
				} finally {
					currentCalls.decrementAndGet();
				}
			}
		} finally {
			checkClosed();
		}
	}

	@Override
	public void unsubscribe() {
		synchronized (this) {
			// ensure this change is flushed to main memory by enclosing in
			// synchronized block
			closing = true;
		}
		checkClosed();
	}

	private void checkClosed() {
		if (closing && currentCalls.get() == 0 && unsubscribed.compareAndSet(false, true)) {
			super.unsubscribe();
		}
	}

	@Override
	public void freeResources() {
		super.freeResources();
	}

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

	}
}
@@ -598,20 +598,26 @@
 boolean 
-clos
+unsubscrib
 ing;%0A%0A%09/
@@ -1002,20 +1002,26 @@
 %0A%09%09this.
-clos
+unsubscrib
 ing = fa
@@ -1114,36 +1114,42 @@
 %0A%09%09try %7B%0A%09%09%09if (
-clos
+unsubscrib
 ing) %7B%0A%09%09%09%09retur
@@ -1319,38 +1319,43 @@
 nally %7B%0A%09%09%09check
-Closed
+Unsubscribe
 ();%0A%09%09%7D%0A%09%7D%0A%0A%09@Ov
@@ -1397,36 +1397,42 @@
 %0A%09%09try %7B%0A%09%09%09if (
-clos
+unsubscrib
 ing) %7B%0A%09%09%09%09retur
@@ -1604,38 +1604,43 @@
 nally %7B%0A%09%09%09check
-Closed
+Unsubscribe
 ();%0A%09%09%7D%0A%09%7D%0A%0A%09@Ov
@@ -1689,20 +1689,26 @@
 %0A%09%09%09if (
-clos
+unsubscrib
 ing) %7B%0A%09
@@ -1893,30 +1893,35 @@
 y %7B%0A%09%09%09check
-Closed
+Unsubscribe
 ();%0A%09%09%7D%0A%09%7D%0A%0A
@@ -2079,20 +2079,26 @@
 lock%0A%09%09%09
-clos
+unsubscrib
 ing = tr
@@ -2112,22 +2112,27 @@
 %0A%09%09check
-Closed
+Unsubscribe
 ();%0A%09%7D%0A%0A
@@ -2154,14 +2154,19 @@
 heck
-Closed
+Unsubscribe
 () %7B
@@ -2172,20 +2172,26 @@
 %7B%0A%09%09if (
-clos
+unsubscrib
 ing && c

refactor field names