<commit_before>package com.github.davidmoten.rx.internal.operators;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.File;
import java.io.IOError;
import java.io.IOException;
import java.io.Serializable;
import java.util.Queue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

import org.mapdb.DB;
import org.mapdb.DBMaker;
import org.mapdb.Serializer;

import com.github.davidmoten.rx.buffertofile.CacheType;
import com.github.davidmoten.rx.buffertofile.DataSerializer;
import com.github.davidmoten.rx.buffertofile.Options;
import com.github.davidmoten.rx.internal.operators.RollingQueue.Queue2;
import com.github.davidmoten.util.Preconditions;

import rx.Observable;
import rx.Observable.OnSubscribe;
import rx.Observable.Operator;
import rx.Producer;
import rx.Scheduler;
import rx.Scheduler.Worker;
import rx.Subscriber;
import rx.Subscription;
import rx.functions.Action0;
import rx.functions.Func0;
import rx.internal.operators.BackpressureUtils;
import rx.observers.Subscribers;

public final class OperatorBufferToFile<T> implements Operator<T, T> {

	private static final String QUEUE_NAME = "q";

	private final Serializer<T> serializer;
	private final Scheduler scheduler;
	private final Options options;

	public OperatorBufferToFile(DataSerializer<T> dataSerializer, Scheduler scheduler, Options options) {
		Preconditions.checkNotNull(dataSerializer);
		Preconditions.checkNotNull(scheduler);
		Preconditions.checkNotNull(options);
		this.scheduler = scheduler;
		this.serializer = createSerializer(dataSerializer);
		this.options = options;
	}

	private static <T> Serializer<T> createSerializer(DataSerializer<T> dataSerializer) {
		return new MapDbSerializer<T>(dataSerializer);
	}

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

		// create the MapDB queue
		final CloseableQueue<T> queue = createQueue(serializer, options);

		// hold a reference to the queueProducer which will be set on
		// subscription to `source`
		final AtomicReference<QueueProducer<T>> queueProducer = new AtomicReference<QueueProducer<T>>();

		// emissions will propagate to downstream via this worker
		final Worker worker = scheduler.createWorker();

		// set up the observable to read from the MapDB queue
		Observable<T> source = Observable.create(new OnSubscribeFromQueue<T>(queueProducer, queue, worker, options));

		// create the parent subscriber
		Subscriber<T> parentSubscriber = new ParentSubscriber<T>(queueProducer);

		// ensure worker gets unsubscribed
		child.add(worker);

		// link unsubscription
		child.add(parentSubscriber);

		// close and delete database on unsubscription
		child.add(disposer(queue));

		// ensure onStart not called twice
		Subscriber<T> wrappedChild = Subscribers.wrap(child);

		// subscribe to queue
		source.unsafeSubscribe(wrappedChild);

		return parentSubscriber;
	}

	/**
	 * Wraps a Queue (like MapDB Queue) to provide concurrency guarantees around
	 * calls to the close() method.
	 * 
	 * @param <T> type of item on queue
	 */
	private static class Q2<T> implements Queue2<T> {

		private final DB db;
		private final Queue<T> queue;
		// guarded by currentCalls and this
		private boolean closing = false;
		private final AtomicBoolean closed = new AtomicBoolean(false);

		// ensures db.close() doesn't occur until outstanding peek(),offer(),
		// poll(), isEmpty() calls have finished
		private final AtomicInteger currentCalls = new AtomicInteger(0);

		Q2(DB db, Queue<T> queue) {
			this.db = db;
			this.queue = queue;
		}

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

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

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

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

		@Override
		public void close() {
			synchronized (this) {
				// ensure this change is seen by all threads by enclosing in
				// synchronized block
				closing = true;
			}
			checkClosed();
		}

		private void checkClosed() {
			if (closing && currentCalls.get() == 0 && closed.compareAndSet(false, true)) {
				db.close();
			}
		}

	}

	private static <T> CloseableQueue<T> createQueue(final Serializer<T> serializer, final Options options) {
		final Func0<Queue2<T>> queueFactory = new Func0<Queue2<T>>() {
			@Override
			public Queue2<T> call() {
				// create the file to be used for queue storage (and whose file
				// name will determine the names of other files used for
				// storage)
				File file = options.fileFactory().call();

				// create a MapDB database using file
				final DB db = createDb(file, options);

				// create the queue

				// setting useLocks to false means that we don't reuse file
				// system space in the queue but operations are faster.
				// Reclaiming space is handled by RollingQueue so we opt for
				// faster operations.
				boolean useLocks = false;<commit_msg>save allocation by making Q2 extend AtomicBoolean<commit_after>package com.github.davidmoten.rx.internal.operators;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.File;
import java.io.IOError;
import java.io.IOException;
import java.io.Serializable;
import java.util.Queue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

import org.mapdb.DB;
import org.mapdb.DBMaker;
import org.mapdb.Serializer;

import com.github.davidmoten.rx.buffertofile.CacheType;
import com.github.davidmoten.rx.buffertofile.DataSerializer;
import com.github.davidmoten.rx.buffertofile.Options;
import com.github.davidmoten.rx.internal.operators.RollingQueue.Queue2;
import com.github.davidmoten.util.Preconditions;

import rx.Observable;
import rx.Observable.OnSubscribe;
import rx.Observable.Operator;
import rx.Producer;
import rx.Scheduler;
import rx.Scheduler.Worker;
import rx.Subscriber;
import rx.Subscription;
import rx.functions.Action0;
import rx.functions.Func0;
import rx.internal.operators.BackpressureUtils;
import rx.observers.Subscribers;

public final class OperatorBufferToFile<T> implements Operator<T, T> {

	private static final String QUEUE_NAME = "q";

	private final Serializer<T> serializer;
	private final Scheduler scheduler;
	private final Options options;

	public OperatorBufferToFile(DataSerializer<T> dataSerializer, Scheduler scheduler, Options options) {
		Preconditions.checkNotNull(dataSerializer);
		Preconditions.checkNotNull(scheduler);
		Preconditions.checkNotNull(options);
		this.scheduler = scheduler;
		this.serializer = createSerializer(dataSerializer);
		this.options = options;
	}

	private static <T> Serializer<T> createSerializer(DataSerializer<T> dataSerializer) {
		return new MapDbSerializer<T>(dataSerializer);
	}

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

		// create the MapDB queue
		final CloseableQueue<T> queue = createQueue(serializer, options);

		// hold a reference to the queueProducer which will be set on
		// subscription to `source`
		final AtomicReference<QueueProducer<T>> queueProducer = new AtomicReference<QueueProducer<T>>();

		// emissions will propagate to downstream via this worker
		final Worker worker = scheduler.createWorker();

		// set up the observable to read from the MapDB queue
		Observable<T> source = Observable.create(new OnSubscribeFromQueue<T>(queueProducer, queue, worker, options));

		// create the parent subscriber
		Subscriber<T> parentSubscriber = new ParentSubscriber<T>(queueProducer);

		// ensure worker gets unsubscribed
		child.add(worker);

		// link unsubscription
		child.add(parentSubscriber);

		// close and delete database on unsubscription
		child.add(disposer(queue));

		// ensure onStart not called twice
		Subscriber<T> wrappedChild = Subscribers.wrap(child);

		// subscribe to queue
		source.unsafeSubscribe(wrappedChild);

		return parentSubscriber;
	}

	/**
	 * Wraps a Queue (like MapDB 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
	 */
	private static class Q2<T> extends AtomicBoolean implements Queue2<T> {

		private static final long serialVersionUID = -950306777716863302L;

		private final DB db;
		private final Queue<T> queue;
		// guarded by currentCalls and this
		private boolean closing = false;

		// ensures db.close() doesn't occur until outstanding peek(),offer(),
		// poll(), isEmpty() calls have finished
		private final AtomicInteger currentCalls = new AtomicInteger(0);

		Q2(DB db, Queue<T> queue) {
			this.db = db;
			this.queue = queue;
			// store-store barrier
			lazySet(false);
		}

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

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

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

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

		@Override
		public void close() {
			synchronized (this) {
				// ensure this change is seen by all threads by enclosing in
				// synchronized block
				closing = true;
			}
			checkClosed();
		}

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

	}

	private static <T> CloseableQueue<T> createQueue(final Serializer<T> serializer, final Options options) {
		final Func0<Queue2<T>> queueFactory = new Func0<Queue2<T>>() {
			@Override
			public Queue2<T> call() {
				// create the file to be used for queue storage (and whose file
				// name will determine the names of other files used for
				// storage)
				File file = options.fileFactory().call();

				// create a MapDB database using file
				final DB db = createDb(file, options);

				// create the queue

				// setting useLocks to false means that we don't reuse file
				// system space in the queue but operations are faster.
				// Reclaiming space is handled by RollingQueue so we opt for
				// faster operations.
				boolean useLocks = false;
{'fraction_non_alphanumeric': 0.1015193956370427, 'fraction_numerical': 0.0038197097020626434, 'mean_word_length': 3.743156199677939, 'pattern_counts': {'":': 0, '<': 69, '<?xml version=': 0, '>': 69, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2019536', 'n_tokens_mistral': 4171, 'n_tokens_neox': 3497, 'n_words': 994}