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();

	}
}
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 unsubscribing;

	// 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.unsubscribing = false;
		this.unsubscribed = new AtomicBoolean(false);
	}

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

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

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

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

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

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

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

	}
}
refactor field names