Updated AsyncWorker (and defined AsyncProcessor) so that we get back …

api/src/main/resources/log4j.properties

@@ -2,7 +2,7 @@ log4j.rootLogger=TRACE, stdout
 log4j.appender.stdout=org.apache.log4j.ConsoleAppender
 log4j.appender.stdout.Target=System.out
 log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
-log4j.appender.stdout.layout.ConversionPattern=%d %-5p %c{1}:%L - %m%n
+log4j.appender.stdout.layout.ConversionPattern=%d [%t] %-5p %c{1}:%L - %m%n
 # Mute Undertow and its dependencies
 log4j.logger.io.undertow=WARN
 log4j.logger.org.jboss=WARN

common/src/main/java/xyz/gianlu/librespot/common/AsyncProcessor.java

@@ -0,0 +1,55 @@
+package xyz.gianlu.librespot.common;
+
+import org.apache.log4j.Logger;
+import org.jetbrains.annotations.NotNull;
+
+import java.io.Closeable;
+import java.util.concurrent.*;
+import java.util.function.Function;
+
+/**
+ * Simple worker thread that processes tasks sequentially
+ * @param <REQ> The type of task/input that AsyncProcessor handles.
+ * @param <RES> Return type of our processor implementation
+ */
+public class AsyncProcessor<REQ, RES> implements Closeable {
+
+    private static final Logger LOGGER = Logger.getLogger(AsyncProcessor.class);
+    private final String name;
+    private final Function<REQ, RES> processor;
+    private final ExecutorService executor;
+    private volatile boolean running;
+
+    /**
+     * @param name name of async processor - used for thread name and logging
+     * @param processor actual processing implementation ran on background thread
+     */
+    public AsyncProcessor(@NotNull String name, @NotNull Function<REQ, RES> processor) {
+        executor = Executors.newSingleThreadExecutor(new NameThreadFactory(r -> name));
+        this.name = name;
+        this.processor = processor;
+        running = true;
+        LOGGER.trace(String.format("AsyncProcessor %s has started", name));
+    }
+
+    public Future<RES> submit(@NotNull REQ task) {
+        if (!running) {
+            throw new IllegalStateException(String.format("AsyncProcessor %s has already been shutdown", name));
+        }
+        return executor.submit(() -> processor.apply(task));
+    }
+
+    @Override
+    public void close() {
+        running = false;
+        LOGGER.trace(String.format("AsyncProcessor %s is shutting down", name));
+        executor.shutdown();
+        try {
+            executor.awaitTermination(1, TimeUnit.MINUTES);
+        } catch (InterruptedException e) {
+            LOGGER.trace(String.format("AsyncProcessor %s was interrupted during shutdown", name));
+        }
+        LOGGER.trace(String.format("AsyncProcessor %s is shut down", name));
+    }
+
+}

common/src/main/java/xyz/gianlu/librespot/common/AsyncWorker.java

@@ -1,52 +1,33 @@
 package xyz.gianlu.librespot.common;
 
-import org.apache.log4j.Logger;
 import org.jetbrains.annotations.NotNull;
 
 import java.io.Closeable;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.Future;
 import java.util.function.Consumer;
 
-public class AsyncWorker<T> implements Closeable, Runnable {
-    private static final Logger LOGGER = Logger.getLogger(AsyncWorker.class);
-    private final Thread thread;
-    private final BlockingQueue<T> internalQueue;
-    private final String name;
-    private final Consumer<T> consumer;
-    private volatile boolean running = true;
+/**
+ * Wrapper around AsyncProcessor that deals with void methods and does not expect a response type
+ *
+ * @param <T> Task type for processor
+ */
+public class AsyncWorker<T> implements Closeable {
 
-    public AsyncWorker(@NotNull String name, @NotNull Consumer<T> consumer) {
-        this.name = name;
-        this.consumer = consumer;
+    private final AsyncProcessor<T, Void> underlyingProcessor;
 
-        internalQueue = new LinkedBlockingQueue<>();
-        thread = new Thread(this, name);
-        thread.start();
+    public AsyncWorker(@NotNull String name, @NotNull Consumer<T> consumer) {
+        this.underlyingProcessor = new AsyncProcessor<>(name, t -> {
+           consumer.accept(t);
+           return null;
+        });
     }
 
-    public void submit(@NotNull T task) {
-        internalQueue.add(task);
+    public Future<Void> submit(@NotNull T task) {
+        return underlyingProcessor.submit(task);
     }
 
     @Override
     public void close() {
-        running = false;
-        thread.interrupt();
-    }
-
-    @Override
-    public void run() {
-        LOGGER.trace(String.format("AsyncWorker{%s} is starting", name));
-
-        while (running) {
-            try {
-                T polled = internalQueue.take();
-                consumer.accept(polled);
-            } catch (InterruptedException ignored) {
-            }
-        }
-
-        LOGGER.trace(String.format("AsyncWorker{%s} is shutting down", name));
+        underlyingProcessor.close();
     }
 }

common/src/main/resources/log4j.properties

@@ -0,0 +1,5 @@
+log4j.rootLogger=TRACE, stdout
+log4j.appender.stdout=org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.Target=System.out
+log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=%d [%t] %-5p %c{1}:%L - %m%n

common/src/test/java/xyz/gianlu/librespot/common/AsyncProcessorTest.java

@@ -0,0 +1,56 @@
+package xyz.gianlu.librespot.common;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class AsyncProcessorTest {
+
+    @Test
+    void testAsyncProcessor() throws ExecutionException, InterruptedException {
+        AtomicInteger internalState = new AtomicInteger();
+        AsyncProcessor<Integer, Integer> asyncProcessor = new AsyncProcessor<>("test-processor-1", internalState::addAndGet);
+        asyncProcessor.submit(1);
+        asyncProcessor.submit(2);
+        asyncProcessor.submit(3);
+        Future<Integer> lastTask = asyncProcessor.submit(4);
+
+        Integer lastResult = lastTask.get(); // we only need to wait for the last one as tasks are executed in order
+        Assertions.assertEquals(10, internalState.get());
+        Assertions.assertEquals(10, lastResult);
+        asyncProcessor.close();
+    }
+
+    @Test
+    void testAsyncProcessorExceptionHandling() {
+        AsyncProcessor<Integer, Integer> asyncProcessor = new AsyncProcessor<>("test-processor-2", i -> {
+            throw new IllegalStateException();
+        });
+
+        Future<Integer> firstTask = asyncProcessor.submit(1);
+        Assertions.assertThrows(ExecutionException.class, firstTask::get);
+
+        // now we check our loop didn't break and we are able to submit more tasks to our queue
+        Future<Integer> secondTask = asyncProcessor.submit(1);
+        Assertions.assertThrows(ExecutionException.class, secondTask::get);
+        asyncProcessor.close();
+    }
+
+    @Test
+    void testAsyncProcessorFailAfterShutdown() throws ExecutionException, InterruptedException {
+        AtomicInteger internalState = new AtomicInteger();
+        AsyncProcessor<Integer, Integer> asyncProcessor = new AsyncProcessor<>("test-processor-3", internalState::addAndGet);
+
+        Future<Integer> taskBeforeShutdown = asyncProcessor.submit(1);
+        Assertions.assertEquals(1, taskBeforeShutdown.get());
+
+        asyncProcessor.close();
+
+        IllegalStateException underlyingException = Assertions.assertThrows(IllegalStateException.class, () -> asyncProcessor.submit(1));
+        Assertions.assertEquals("AsyncProcessor test-processor-3 has already been shutdown", underlyingException.getMessage());
+    }
+
+}

core/src/main/resources/log4j.properties

@@ -2,4 +2,4 @@ log4j.rootLogger=TRACE, stdout
 log4j.appender.stdout=org.apache.log4j.ConsoleAppender
 log4j.appender.stdout.Target=System.out
 log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
-log4j.appender.stdout.layout.ConversionPattern=%d %-5p %c{1}:%L - %m%n
+log4j.appender.stdout.layout.ConversionPattern=%d [%t] %-5p %c{1}:%L - %m%n