first commit

.gitignore

@@ -0,0 +1,60 @@
+# Covers JetBrains IDEs: IntelliJ, RubyMine, PhpStorm, AppCode, PyCharm, CLion, Android Studio and WebStorm
+# Reference: https://intellij-support.jetbrains.com/hc/en-us/articles/206544839
+
+# User-specific stuff
+.idea/*.*
+.idea/**/workspace.xml
+.idea/**/tasks.xml
+.idea/**/dictionaries
+.idea/**/shelf
+
+# Sensitive or high-churn files
+.idea/workspace.xml
+.idea/**/dataSources/
+.idea/**/dataSources.ids
+.idea/**/dataSources.local.xml
+.idea/**/sqlDataSources.xml
+.idea/**/dynamic.xml
+.idea/**/uiDesigner.xml
+.idea/**/dbnavigator.xml
+
+# Gradle
+.idea/**/gradle.xml
+.idea/**/libraries
+
+# CMake
+cmake-build-debug/
+cmake-build-release/
+
+# Mongo Explorer plugin
+.idea/**/mongoSettings.xml
+
+# File-based project format
+*.iws
+
+# IntelliJ
+out/
+filequeue.iml
+
+# mpeltonen/sbt-idea plugin
+.idea_modules/
+
+# JIRA plugin
+atlassian-ide-plugin.xml
+
+# Cursive Clojure plugin
+.idea/replstate.xml
+
+# Crashlytics plugin (for Android Studio and IntelliJ)
+com_crashlytics_export_strings.xml
+crashlytics.properties
+crashlytics-build.properties
+fabric.properties
+
+# Editor-based Rest Client
+.idea/httpRequests
+/.idea/
+target/
+.DS_Store
+/.idea/
+/.idea/

pom.xml

@@ -0,0 +1,49 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <groupId>it.cavallium</groupId>
+    <artifactId>filequeue</artifactId>
+    <name>file queue project</name>
+    <version>3.0.0</version>
+    <packaging>jar</packaging>
+    <description>Light weight, high performance, simple, reliable and persistent queue</description>
+    <modelVersion>4.0.0</modelVersion>
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
+    </properties>
+    <distributionManagement>
+        <repository>
+            <id>mchv-release-distribution</id>
+            <name>MCHV Release Apache Maven Packages Distribution</name>
+            <url>https://mvn.mchv.eu/repository/mchv</url>
+        </repository>
+        <snapshotRepository>
+            <id>mchv-snapshot-distribution</id>
+            <name>MCHV Snapshot Apache Maven Packages Distribution</name>
+            <url>https://mvn.mchv.eu/repository/mchv-snapshot</url>
+        </snapshotRepository>
+    </distributionManagement>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <version>3.7.0</version>
+                <configuration>
+                    <source>11</source>
+                    <target>11</target>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+    <dependencies>
+        <dependency>
+            <groupId>com.squareup.tape2</groupId>
+            <artifactId>tape</artifactId>
+            <version>2.0.0-beta1</version>
+        </dependency>
+    </dependencies>
+
+</project>

src/main/java/it/cavallium/filequeue/Deserializer.java

@@ -0,0 +1,22 @@
+package it.cavallium.filequeue;
+
+import java.io.ByteArrayInputStream;
+import java.io.Closeable;
+import java.io.DataInput;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.util.Map;
+
+public interface Deserializer<T> {
+
+	default T deserialize(byte[] data) throws IOException {
+		ByteArrayInputStream bais = new ByteArrayInputStream(data);
+		return deserialize(data.length, new DataInputStream(bais));
+	}
+
+	default T deserialize(int length, DataInput dataInput) throws IOException {
+		byte[] data = new byte[length];
+		dataInput.readFully(data);
+		return deserialize(data);
+	}
+}

src/main/java/it/cavallium/filequeue/DiskQueueToConsumer.java

@@ -0,0 +1,40 @@
+package it.cavallium.filequeue;
+
+import com.squareup.tape2.QueueFile;
+import java.io.IOException;
+import java.nio.file.Path;
+
+public final class DiskQueueToConsumer<T> implements IQueueToConsumer<T> {
+
+	private final QueueToConsumer<T> queue;
+	private final QueueFile queueFile;
+
+	public DiskQueueToConsumer(Path file,
+			Serializer<T> serializer,
+			Deserializer<T> deserializer,
+			QueueConsumer<T> consumer) throws IOException {
+		QueueFile queueFile = new QueueFile.Builder(file.toFile()).build();
+		this.queueFile = queueFile;
+		this.queue = new QueueToConsumer<>(new SimpleQueueFile<>(queueFile, serializer, deserializer), consumer);
+	}
+
+	@Override
+	public void add(T value) {
+		queue.add(value);
+	}
+
+	@Override
+	public void close() {
+		queue.close();
+		try {
+			queueFile.close();
+		} catch (IOException e) {
+			throw new RuntimeException(e);
+		}
+	}
+
+	@Override
+	public void startQueue() {
+		queue.startQueue();
+	}
+}

src/main/java/it/cavallium/filequeue/IQueueToConsumer.java

@@ -0,0 +1,13 @@
+package it.cavallium.filequeue;
+
+import java.io.Closeable;
+
+public interface IQueueToConsumer<T> extends Closeable {
+
+	void add(T value);
+
+	@Override
+	void close();
+
+	void startQueue();
+}

src/main/java/it/cavallium/filequeue/QueueConsumer.java

@@ -0,0 +1,6 @@
+package it.cavallium.filequeue;
+
+public interface QueueConsumer<T> {
+
+	boolean tryConsume(T value);
+}

src/main/java/it/cavallium/filequeue/QueueToConsumer.java

@@ -0,0 +1,93 @@
+package it.cavallium.filequeue;
+
+import java.time.Duration;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.locks.LockSupport;
+
+class QueueToConsumer<T> implements IQueueToConsumer<T> {
+
+	private final long BACKOFF_NS = Duration.ofMillis(2).toNanos();
+	private final long MAX_BACKOFF_NS = Duration.ofMillis(500).toNanos();
+
+	private final Object lock = new Object();
+	private final Semaphore semaphore = new Semaphore(0);
+	private long queued;
+	private final SimpleQueue<T> queue;
+	private final QueueConsumer<T> consumer;
+	private Manager manager;
+	private volatile boolean closed;
+
+	public QueueToConsumer(SimpleQueue<T> queue, QueueConsumer<T> consumer) {
+		this.queue = queue;
+		this.consumer = consumer;
+		queued = queue.size();
+	}
+
+	public synchronized void startQueue() {
+		if (manager == null) {
+			this.manager = new Manager();
+			manager.setName("queue-manager");
+			manager.start();
+		}
+	}
+
+	@Override
+	public void add(T value) {
+		boolean shouldAdd = true;
+		synchronized (lock) {
+			if (queued == 0 && consumer.tryConsume(value)) {
+				shouldAdd = false;
+			} else {
+				queued++;
+			}
+		}
+		if (shouldAdd && !closed) {
+			synchronized (queue) {
+				queue.add(value);
+			}
+			semaphore.release();
+		}
+	}
+
+	@Override
+	public void close() {
+		closed = true;
+		semaphore.release();
+	}
+
+	private class Manager extends Thread {
+
+		@Override
+		public void run() {
+			try {
+				while (!closed) {
+					boolean shouldRemove = false;
+					T element;
+					synchronized (lock) {
+						if (queued > 0) {
+							queued--;
+							shouldRemove = true;
+						}
+					}
+					semaphore.acquire();
+					if (!closed && shouldRemove) {
+						synchronized (queue) {
+							element = queue.remove();
+						}
+						long nextDelay = BACKOFF_NS;
+						while (!closed && !consumer.tryConsume(element)) {
+							LockSupport.parkNanos(nextDelay);
+							if (nextDelay + BACKOFF_NS <= MAX_BACKOFF_NS) {
+								nextDelay += BACKOFF_NS;
+							} else if (nextDelay < MAX_BACKOFF_NS) {
+								nextDelay = MAX_BACKOFF_NS;
+							}
+						}
+					}
+				}
+			} catch (InterruptedException ex) {
+				closed = true;
+			}
+		}
+	}
+}

src/main/java/it/cavallium/filequeue/Serializer.java

@@ -0,0 +1,22 @@
+package it.cavallium.filequeue;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutput;
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+public interface Serializer<T> {
+
+	default byte[] serialize(T data) throws IOException {
+		try (ByteArrayOutputStream baos = new ByteArrayOutputStream()) {
+			try (DataOutputStream daos = new DataOutputStream(baos)) {
+				serialize(data, daos);
+				return baos.toByteArray();
+			}
+		}
+	}
+
+	default void serialize(T data, DataOutput output) throws IOException {
+		output.write(serialize(data));
+	}
+}

src/main/java/it/cavallium/filequeue/SimpleQueue.java

@@ -0,0 +1,10 @@
+package it.cavallium.filequeue;
+
+interface SimpleQueue<T> {
+
+	void add(T element);
+
+	T remove();
+
+	int size();
+}

src/main/java/it/cavallium/filequeue/SimpleQueueFile.java

@@ -0,0 +1,47 @@
+package it.cavallium.filequeue;
+
+import com.squareup.tape2.QueueFile;
+import java.io.IOException;
+import java.util.NoSuchElementException;
+
+class SimpleQueueFile<T> implements SimpleQueue<T> {
+
+	private final QueueFile queueFile;
+	private final Serializer<T> ser;
+	private final Deserializer<T> des;
+
+	public SimpleQueueFile(QueueFile queueFile, Serializer<T> serializer, Deserializer<T> deserializer) {
+		this.queueFile = queueFile;
+		this.ser = serializer;
+		this.des = deserializer;
+	}
+
+	@Override
+	public void add(T element) {
+		try {
+			queueFile.add(ser.serialize(element));
+		} catch (IOException e) {
+			throw new RuntimeException(e);
+		}
+	}
+
+	@Override
+	public T remove() {
+		try {
+			byte[] element = queueFile.peek();
+			if (element == null) {
+				throw new NoSuchElementException("Queue is empty");
+			}
+			var deserialized = des.deserialize(element);
+			queueFile.remove();
+			return deserialized;
+		} catch (IOException e) {
+			throw new RuntimeException(e);
+		}
+	}
+
+	@Override
+	public int size() {
+		return queueFile.size();
+	}
+}

src/main/java/it/cavallium/filequeue/SimpleQueueJava.java

@@ -0,0 +1,27 @@
+package it.cavallium.filequeue;
+
+import java.util.Queue;
+
+class SimpleQueueJava<T> implements SimpleQueue<T> {
+
+	private final Queue<T> queue;
+
+	public SimpleQueueJava(Queue<T> queue) {
+		this.queue = queue;
+	}
+
+	@Override
+	public void add(T element) {
+		queue.add(element);
+	}
+
+	@Override
+	public T remove() {
+		return queue.remove();
+	}
+
+	@Override
+	public int size() {
+		return queue.size();
+	}
+}