update dependencies and enhance XodusQueue functionality with additional constructors and locking mechanisms

Author: ralscha

pom.xml

@@ -63,20 +63,20 @@
 		<dependency>
 			<groupId>org.junit.jupiter</groupId>
 			<artifactId>junit-jupiter-api</artifactId>
-			<version>6.0.0</version>
+			<version>6.0.3</version>
 			<scope>test</scope>
 		</dependency>
 		<dependency>
 			<groupId>org.junit.jupiter</groupId>
 			<artifactId>junit-jupiter-engine</artifactId>
-			<version>6.0.0</version>
+			<version>6.0.3</version>
 			<scope>test</scope>
 		</dependency>
 
 		<dependency>
 			<groupId>org.junit.platform</groupId>
 			<artifactId>junit-platform-launcher</artifactId>
-			<version>6.0.0</version>
+			<version>6.0.3</version>
 			<scope>test</scope>
 		</dependency>
 

src/main/java/ch/rasc/xodusqueue/XodusBlockingQueue.java

@@ -16,6 +16,7 @@
 package ch.rasc.xodusqueue;
 
 import java.util.Collection;
+import java.util.Iterator;
 import java.util.Objects;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.TimeUnit;
@@ -38,6 +39,19 @@ public class XodusBlockingQueue<T> extends XodusQueue<T> implements BlockingQueu
 
 	private long capacity;
 
+	public XodusBlockingQueue(String databaseDir, Class<T> entryClass) {
+		this(databaseDir, entryClass, Long.MAX_VALUE);
+	}
+
+	public XodusBlockingQueue(String databaseDir, XodusQueueSerializer<T> serializer) {
+		this(databaseDir, serializer, Long.MAX_VALUE);
+	}
+
+	public XodusBlockingQueue(LogConfig logConfig, EnvironmentConfig environmentConfig,
+			XodusQueueSerializer<T> serializer) {
+		this(logConfig, environmentConfig, serializer, Long.MAX_VALUE);
+	}
+
 	public XodusBlockingQueue(LogConfig logConfig, EnvironmentConfig environmentConfig,
 			XodusQueueSerializer<T> serializer, long capacity) {
 		super(logConfig, environmentConfig, serializer);
@@ -161,7 +175,9 @@ public T poll() {
 		lock.lock();
 		try {
 			T e = super.poll();
-			this.notFull.signal();
+			if (e != null) {
+				this.notFull.signal();
+			}
 			return e;
 		}
 		finally {
@@ -177,7 +193,9 @@ public T take() throws InterruptedException {
 			while (super.sizeLong() == 0) {
 				this.notEmpty.await();
 			}
-			return poll();
+			T e = super.poll();
+			this.notFull.signal();
+			return e;
 		}
 		finally {
 			lock.unlock();
@@ -196,7 +214,9 @@ public T poll(long timeout, TimeUnit unit) throws InterruptedException {
 				}
 				nanos = this.notEmpty.awaitNanos(nanos);
 			}
-			return poll();
+			T e = super.poll();
+			this.notFull.signal();
+			return e;
 		}
 		finally {
 			lock.unlock();
@@ -319,4 +339,113 @@ public boolean retainAll(Collection<?> c) {
 		}
 	}
 
+	@Override
+	public T peek() {
+		final ReentrantLock lock = this.reentrantLock;
+		lock.lock();
+		try {
+			return super.peek();
+		}
+		finally {
+			lock.unlock();
+		}
+	}
+
+	@Override
+	public int size() {
+		final ReentrantLock lock = this.reentrantLock;
+		lock.lock();
+		try {
+			return super.size();
+		}
+		finally {
+			lock.unlock();
+		}
+	}
+
+	@Override
+	public long sizeLong() {
+		final ReentrantLock lock = this.reentrantLock;
+		lock.lock();
+		try {
+			return super.sizeLong();
+		}
+		finally {
+			lock.unlock();
+		}
+	}
+
+	@Override
+	public boolean isEmpty() {
+		final ReentrantLock lock = this.reentrantLock;
+		lock.lock();
+		try {
+			return super.isEmpty();
+		}
+		finally {
+			lock.unlock();
+		}
+	}
+
+	@Override
+	public boolean contains(Object o) {
+		final ReentrantLock lock = this.reentrantLock;
+		lock.lock();
+		try {
+			return super.contains(o);
+		}
+		finally {
+			lock.unlock();
+		}
+	}
+
+	@Override
+	public boolean containsAll(Collection<?> c) {
+		final ReentrantLock lock = this.reentrantLock;
+		lock.lock();
+		try {
+			return super.containsAll(c);
+		}
+		finally {
+			lock.unlock();
+		}
+	}
+
+	@Override
+	public Object[] toArray() {
+		final ReentrantLock lock = this.reentrantLock;
+		lock.lock();
+		try {
+			return super.toArray();
+		}
+		finally {
+			lock.unlock();
+		}
+	}
+
+	@SuppressWarnings({ "unchecked", "hiding" })
+	@Override
+	public <T> T[] toArray(T[] a) {
+		final ReentrantLock lock = this.reentrantLock;
+		lock.lock();
+		try {
+			return super.toArray(a);
+		}
+		finally {
+			lock.unlock();
+		}
+	}
+
+	@Override
+	public Iterator<T> iterator() {
+		final ReentrantLock lock = this.reentrantLock;
+		lock.lock();
+		try {
+			return super.iterator();
+		}
+		finally {
+			lock.unlock();
+		}
+	}
+
 }

src/main/java/ch/rasc/xodusqueue/XodusQueue.java

@@ -22,8 +22,8 @@
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Collections;
 import java.util.Objects;
-import java.util.concurrent.atomic.AtomicLong;
 
 import ch.rasc.xodusqueue.serializer.BigDecimalXodusQueueSerializer;
 import ch.rasc.xodusqueue.serializer.BigIntegerXodusQueueSerializer;
@@ -57,8 +57,6 @@ public class XodusQueue<T> extends AbstractQueue<T> implements AutoCloseable {
 
 	private final XodusQueueSerializer<T> serializer;
 
-	private final AtomicLong key = new AtomicLong(0L);
-
 	@SuppressWarnings("unchecked")
 	public XodusQueue(final String databaseDir, final Class<T> entryClass) {
 		this.env = Environments.newInstance(databaseDir);
@@ -96,42 +94,17 @@ else if (entryClass == BigDecimal.class) {
 		else {
 			this.serializer = new DefaultXodusQueueSerializer<>(entryClass);
 		}
-
-		this.key.set(this.lastKey());
 	}
 
 	public XodusQueue(final String databaseDir, final XodusQueueSerializer<T> serializer) {
 		this.env = Environments.newInstance(databaseDir);
 		this.serializer = serializer;
-
-		this.key.set(this.lastKey());
 	}
 
 	public XodusQueue(final LogConfig logConfig, final EnvironmentConfig environmentConfig,
 			final XodusQueueSerializer<T> serializer) {
 		this.env = Environments.newInstance(logConfig, environmentConfig);
 		this.serializer = serializer;
-
-		this.key.set(this.lastKey());
-	}
-
-	/**
-	 * Returns the last key stored in the queue store or 0 if the store is empty.
-	 */
-	private long lastKey() {
-		return this.env.computeInReadonlyTransaction(txn -> {
-			Store store = this.env.openStore(STORE_NAME, StoreConfig.WITHOUT_DUPLICATES, txn, false);
-			if (store == null) {
-				return 0L;
-			}
-
-			try (Cursor cursor = store.openCursor(txn)) {
-				if (cursor.getLast()) {
-					return LongBinding.entryToLong(cursor.getKey());
-				}
-			}
-			return 0L;
-		});
 	}
 
 	@Override
@@ -149,7 +122,6 @@ public boolean offer(T e) {
 			}
 
 			store.putRight(txn, LongBinding.longToEntry(nextKey), this.serializer.toEntry(e));
-			this.key.set(nextKey);
 		});
 
 		return true;
@@ -182,10 +154,6 @@ public boolean addAll(Collection<? extends T> c) {
 				modified = true;
 			}
 
-			if (modified) {
-				this.key.set(last);
-			}
-
 			return modified;
 		});
 	}
@@ -285,7 +253,7 @@ public Iterator<T> iterator() {
 				}
 			}
 		});
-		return snapshot.iterator();
+		return Collections.unmodifiableList(snapshot).iterator();
 	}
 
 	@Override
@@ -324,14 +292,23 @@ public <T> T[] toArray(T[] a) {
 						r[ix++] = (T) this.serializer.fromEntry(value);
 					}
 				}
+				if (r.length > ix) {
+					r[ix] = null;
+				}
 				return r;
 			}
+			if (a.length > 0) {
+				a[0] = null;
+			}
 			return a;
 		});
 	}
 
 	@Override
 	public boolean remove(Object o) {
+		if (o == null) {
+			return false;
+		}
 		return this.env.computeInExclusiveTransaction(txn -> {
 			Store store = this.env.openStore(STORE_NAME, StoreConfig.WITHOUT_DUPLICATES, txn, false);
 
@@ -352,6 +329,10 @@ public boolean remove(Object o) {
 
 	@Override
 	public boolean containsAll(Collection<?> c) {
+		Objects.requireNonNull(c);
+		if (c.isEmpty()) {
+			return true;
+		}
 		return this.env.computeInReadonlyTransaction(txn -> {
 			Store store = this.env.openStore(STORE_NAME, StoreConfig.WITHOUT_DUPLICATES, txn, false);
 			if (store != null) {
@@ -428,7 +409,6 @@ public boolean retainAll(Collection<?> c) {
 	public void clear() {
 		this.env.executeInExclusiveTransaction(txn -> {
 			this.env.truncateStore(STORE_NAME, txn);
-			this.key.set(0L);
 		});
 	}
 

src/test/java/ch/rasc/xodusqueue/XodusQueueTest.java

@@ -171,7 +171,7 @@ void testAdd() {
 	@Test
 	void testAddAllCollectionOfQextendsT() {
 		try (XodusQueue<Integer> queue = new XodusQueue<>("./test", Integer.class)) {
-			Assertions.assertThrows(NullPointerException.class, null);
+			Assertions.assertThrows(NullPointerException.class, () -> queue.addAll(null));
 
 			Assertions.assertTrue(queue.addAll(Arrays.asList(1, 2, 3, 4, 5)));
 			Assertions.assertEquals(5, queue.size());