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

import java.util.Collection;
import java.util.Deque;
import java.util.Iterator;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicLong;

import rx.functions.Func0;

public final class RollingQueue<T> implements CloseableQueue<T> {

	public interface Queue2<T> {
		T peek();

		T poll();

		boolean offer(T t);

		void dispose();

		boolean isEmpty();
	}

	private final Func0<Queue2<T>> queueFactory;
	private final long maxItemsPerQueue;
	private final Deque<Queue2<T>> queues = new LinkedBlockingDeque<Queue2<T>>();

	private final AtomicLong count = new AtomicLong(0);

	public RollingQueue(Func0<Queue2<T>> queueFactory, long maxItemsPerQueue) {
		this.queueFactory = queueFactory;
		this.maxItemsPerQueue = maxItemsPerQueue;
	}

	@Override
	public void close() {
		// thread-safe
		while (true) {
			Queue2<T> q = queues.pollFirst();
			if (q != null) {
				q.dispose();
			} else {
				return;
			}
		}
	}

	@Override
	public boolean offer(T t) {
		// limited thread safety (offer and poll concurrent but not offer
		// and offer)
		long c = count.incrementAndGet();
		if (c == 1 || c == maxItemsPerQueue) {
			count.set(1);
			queues.add(queueFactory.call());
		}
		return queues.peekLast().offer(t);
	}

	@Override
	public T poll() {
		// limited thread safety (offer and poll concurrent but not poll
		// and poll)
		if (queues.isEmpty())
			return null;
		else {
			while (true) {
				T value = queues.peekFirst().poll();
				if (value == null) {
					if (queues.size() <= 1) {
						return null;
					} else {
						Queue2<T> removed = queues.pollFirst();
						removed.dispose();
					}
				} else {
					return value;
				}
			}
		}
	}

	@Override
	public T peek() {
		// thread-safe
		return queues.peekFirst().peek();
	}

	@Override
	public boolean isEmpty() {
		synchronized (queues) {
			if (queues.isEmpty()) {
				return true;
			} else if (queues.peekLast() == queues.peekFirst() && queues.peekFirst().isEmpty()) {
				return true;
			} else
				return false;
		}
	}

	@Override
	public void clear() {
		throw new UnsupportedOperationException();
	}

	@Override
	public int size() {
		throw new UnsupportedOperationException();
	}

	@Override
	public boolean contains(Object o) {
		throw new UnsupportedOperationException();
	}

	@Override
	public Iterator<T> iterator() {
		throw new UnsupportedOperationException();
	}

	@Override
	public Object[] toArray() {
		throw new UnsupportedOperationException();
	}

	@Override
	public <T> T[] toArray(T[] a) {
		throw new UnsupportedOperationException();
	}

	@Override
	public boolean remove(Object o) {
		throw new UnsupportedOperationException();
	}

	@Override
	public boolean containsAll(Collection<?> c) {
		throw new UnsupportedOperationException();
	}

	@Override
	public boolean addAll(Collection<? extends T> c) {
		throw new UnsupportedOperationException();
	}

	@Override
	public boolean removeAll(Collection<?> c) {
		throw new UnsupportedOperationException();
	}

	@Override
	public boolean retainAll(Collection<?> c) {
		throw new UnsupportedOperationException();
	}

	@Override
	public boolean add(T e) {
		throw new UnsupportedOperationException();
	}

	@Override
	public T remove() {
		throw new UnsupportedOperationException();
	}

	@Override
	public T element() {
		throw new UnsupportedOperationException();
	}

}
@@ -252,16 +252,904 @@
 Func0;%0A%0A
+/**%0A * This abstraction around multiple queues exists because to reclaim file system%0A * space taken by MapDB databases the selected algorithm was to use a double%0A * ended queue of database+queue combinations. As the number of entries added to%0A * a queue (regardless of how many are read) meets a threshold another queue is%0A * created on the end of the deque and new entries added to that. As entries are%0A * read from a queue that is not the last queue, it is deleted when empty and%0A * its file resources recovered (deleted).%0A * %0A * MapDB has the facility to reuse space (at significant speed cost) but to%0A * shrink allocated space (due to a surge in queue size) requires a non-trivial%0A * blocking operation (%7B@code DB.compact()%7D) so it seems better to avoid%0A * blocking and incur regular small new DB instance creation costs.%0A * %0A * @param %3CT%3E%0A *            type of item being queued%0A */%0A
 public f
@@ -2757,60 +2757,86 @@
 %7B%0A%09%09
-synchronized (queues) %7B%0A%09%09%09if (queues.isEmpty()
+// thread-safe%0A%09%09Queue2%3CT%3E first = queues.peekFirst();%0A%09%09if (first == null
 ) %7B%0A
-%09
 %09%09%09r
@@ -2839,33 +2839,32 @@
 %09%09%09return true;%0A
-%09
 %09%09%7D else if (que
@@ -2885,48 +2885,22 @@
  == 
-queues.peekF
+f
 irst
-()
  && 
-queues.peekF
+f
 irst
-()
 .isE
@@ -2908,25 +2908,24 @@
 pty()) %7B%0A%09%09%09
-%09
 return true;
@@ -2925,25 +2925,24 @@
 n true;%0A
-%09
 %09%09%7D else
 %0A%09%09%09%09ret
@@ -2933,18 +2933,19 @@
 %09%09%7D else
+ %7B
 %0A
-%09
 %09%09%09retur

add javaodc