Extracted AsyncWorker we can use for multiple consumers in code base. (#206)

* Extracted AsyncWorker we can use for multiple consumers in code base.

* Changed another consumer to use AsyncWorker pattern.

* Reformatted code + annotations + do not catch RuntimeExceptions (to match old behaviour)

* Using AsyncWorker in PacketsManager and DealerClient

Co-authored-by: Gianlu <[email protected]>

Author: josephcarroll

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

@@ -0,0 +1,50 @@
+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.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;
+
+    public AsyncWorker(@NotNull String name, @NotNull Consumer<T> consumer) {
+        this.name = name;
+        this.consumer = consumer;
+
+        internalQueue = new LinkedBlockingQueue<>();
+        thread = new Thread(this, name);
+        thread.start();
+    }
+
+    public void submit(@NotNull T task) {
+        internalQueue.add(task);
+    }
+
+    @Override
+    public void close() {
+        running = false;
+        thread.interrupt();
+    }
+
+    @Override
+    public void run() {
+        while (running) {
+            try {
+                T polled = internalQueue.take();
+                consumer.accept(polled);
+            } catch (InterruptedException ignored) {
+            }
+        }
+
+        LOGGER.trace(String.format("AsyncWorker{%s} is shutting down", name));
+    }
+}

core/src/main/java/xyz/gianlu/librespot/connectstate/DeviceStateHandler.java

@@ -11,6 +11,7 @@
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 import xyz.gianlu.librespot.Version;
+import xyz.gianlu.librespot.common.AsyncWorker;
 import xyz.gianlu.librespot.common.ProtoUtils;
 import xyz.gianlu.librespot.core.Session;
 import xyz.gianlu.librespot.core.TimeProvider;
@@ -23,9 +24,6 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.*;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
 
 /**
  * @author Gianlu
@@ -45,21 +43,19 @@ public final class DeviceStateHandler implements Closeable, DealerClient.Message
     private final Connect.DeviceInfo.Builder deviceInfo;
     private final List<Listener> listeners = Collections.synchronizedList(new ArrayList<>());
     private final Connect.PutStateRequest.Builder putState;
-    private final Worker worker;
+    private final AsyncWorker<Connect.PutStateRequest> putStateWorker;
     private volatile String connectionId = null;
 
     public DeviceStateHandler(@NotNull Session session) {
         this.session = session;
         this.deviceInfo = initializeDeviceInfo(session);
-        this.worker = new Worker();
+        this.putStateWorker = new AsyncWorker<>("put-state-worker", this::putConnectState);
         this.putState = Connect.PutStateRequest.newBuilder()
                 .setMemberType(Connect.MemberType.CONNECT_STATE)
                 .setDevice(Connect.Device.newBuilder()
                         .setDeviceInfo(deviceInfo)
                         .build());
 
-        new Thread(worker, "put-state-worker").start();
-
         session.dealer().addMessageListener(this, "hm://pusher/v1/connections/", "hm://connect-state/v1/connect/volume", "hm://connect-state/v1/cluster");
         session.dealer().addRequestListener(this, "hm://connect-state/v1/");
         session.mercury().interestedIn("hm://pusher/v1/connections/", this);
@@ -211,7 +207,7 @@ public synchronized void updateState(@NotNull Connect.PutStateReason reason, @No
                 .setClientSideTimestamp(TimeProvider.currentTimeMillis())
                 .getDeviceBuilder().setDeviceInfo(deviceInfo).setPlayerState(state);
 
-        worker.submit(putState.build());
+        putStateWorker.submit(putState.build());
     }
 
     public synchronized int getVolume() {
@@ -233,10 +229,25 @@ public void close() {
         session.dealer().removeRequestListener(this);
         session.mercury().notInterested(this);
 
-        worker.stop();
+        putStateWorker.close();
         listeners.clear();
     }
 
+    /**
+     * Performs the network request related to {@link Connect.PutStateRequest}. This MUST be called only from {@link DeviceStateHandler#putStateWorker}.
+     *
+     * @param req The {@link Connect.PutStateRequest}
+     */
+    private void putConnectState(@NotNull Connect.PutStateRequest req) {
+        try {
+            session.api().putConnectState(connectionId, req);
+            LOGGER.info(String.format("Put state. {ts: %d, connId: %s[truncated], reason: %s, request: %s}", req.getClientSideTimestamp(), connectionId.substring(0, 6),
+                    req.getPutStateReason(), ProtoUtils.toLogString(putState, LOGGER)));
+        } catch (IOException | MercuryClient.MercuryException ex) {
+            LOGGER.error("Failed updating state.", ex);
+        }
+    }
+
     public enum Endpoint {
         Play("play"), Pause("pause"), Resume("resume"), SeekTo("seek_to"), SkipNext("skip_next"),
         SkipPrev("skip_prev"), SetShufflingContext("set_shuffling_context"), SetRepeatingContext("set_repeating_context"),
@@ -429,38 +440,4 @@ public Boolean valueBool() {
         }
     }
 
-    private class Worker implements Runnable {
-        private final BlockingQueue<Connect.PutStateRequest> queue = new LinkedBlockingQueue<>();
-        private volatile boolean shouldStop = false;
-
-        void stop() {
-            shouldStop = true;
-        }
-
-        void submit(@NotNull Connect.PutStateRequest request) {
-            queue.add(request);
-        }
-
-        @Override
-        public void run() {
-            while (!shouldStop) {
-                Connect.PutStateRequest req;
-                try {
-                    req = queue.poll(1000, TimeUnit.MILLISECONDS);
-                    if (shouldStop) return;
-                    if (req == null) continue;
-                } catch (InterruptedException ignored) {
-                    continue;
-                }
-
-                try {
-                    session.api().putConnectState(connectionId, req);
-                    LOGGER.info(String.format("Put state. {ts: %d, connId: %s[truncated], reason: %s, request: %s}", req.getClientSideTimestamp(), connectionId.substring(0, 6),
-                            req.getPutStateReason(), ProtoUtils.toLogString(putState, LOGGER)));
-                } catch (IOException | MercuryClient.MercuryException ex) {
-                    LOGGER.error("Failed updating state.", ex);
-                }
-            }
-        }
-    }
 }

core/src/main/java/xyz/gianlu/librespot/core/PacketsManager.java

@@ -1,28 +1,30 @@
 package xyz.gianlu.librespot.core;
 
 import org.jetbrains.annotations.NotNull;
+import xyz.gianlu.librespot.common.AsyncWorker;
 import xyz.gianlu.librespot.crypto.Packet;
 
 import java.io.IOException;
-import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.LinkedBlockingQueue;
 
 /**
  * @author Gianlu
  */
 public abstract class PacketsManager implements AutoCloseable {
     protected final Session session;
-    private final BlockingQueue<Packet> queue;
-    private final Looper looper;
     private final ExecutorService executorService;
+    private final AsyncWorker<Packet> asyncWorker;
 
-    public PacketsManager(@NotNull Session session) {
+    public PacketsManager(@NotNull Session session, @NotNull String name) {
         this.session = session;
         this.executorService = session.executor();
-        this.queue = new LinkedBlockingQueue<>();
-        this.looper = new Looper();
-        new Thread(looper, "packets-manager-" + looper.hashCode()).start();
+        this.asyncWorker = new AsyncWorker<>("pm-" + name, packet -> executorService.execute(() -> {
+            try {
+                handle(packet);
+            } catch (IOException ex) {
+                exception(ex);
+            }
+        }));
     }
 
     public final void dispatch(@NotNull Packet packet) {
@@ -31,42 +33,17 @@ public final void dispatch(@NotNull Packet packet) {
 
     @Override
     public void close() {
-        looper.stop();
+        asyncWorker.close();
     }
 
     /**
      * This method can be overridden to process packet synchronously. This MUST not block for a long period of time.
      */
     protected void appendToQueue(@NotNull Packet packet) {
-        queue.add(packet);
+        asyncWorker.submit(packet);
     }
 
     protected abstract void handle(@NotNull Packet packet) throws IOException;
 
     protected abstract void exception(@NotNull Exception ex);
-
-    private final class Looper implements Runnable {
-        private volatile boolean shouldStop = false;
-
-        @Override
-        public void run() {
-            while (!shouldStop) {
-                try {
-                    Packet packet = queue.take();
-                    executorService.execute(() -> {
-                        try {
-                            handle(packet);
-                        } catch (IOException ex) {
-                            exception(ex);
-                        }
-                    });
-                } catch (InterruptedException ignored) {
-                }
-            }
-        }
-
-        void stop() {
-            shouldStop = true;
-        }
-    }
 }

core/src/main/java/xyz/gianlu/librespot/dealer/DealerClient.java

@@ -11,14 +11,18 @@
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 import xyz.gianlu.librespot.BytesArrayList;
+import xyz.gianlu.librespot.common.AsyncWorker;
 import xyz.gianlu.librespot.common.NameThreadFactory;
 import xyz.gianlu.librespot.core.ApResolver;
 import xyz.gianlu.librespot.core.Session;
 import xyz.gianlu.librespot.mercury.MercuryClient;
 
 import java.io.*;
 import java.util.*;
-import java.util.concurrent.*;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.zip.GZIPInputStream;
 
@@ -27,7 +31,7 @@
  */
 public class DealerClient implements Closeable {
     private static final Logger LOGGER = Logger.getLogger(DealerClient.class);
-    private final Looper looper = new Looper();
+    private final AsyncWorker<Runnable> asyncWorker;
     private final Session session;
     private final Map<String, RequestListener> reqListeners = new HashMap<>();
     private final Map<MessageListener, List<String>> msgListeners = new HashMap<>();
@@ -37,7 +41,7 @@ public class DealerClient implements Closeable {
 
     public DealerClient(@NotNull Session session) {
         this.session = session;
-        new Thread(looper, "dealer-looper").start();
+        this.asyncWorker = new AsyncWorker<>("dealer-worker", Runnable::run);
     }
 
     @NotNull
@@ -98,7 +102,7 @@ private void handleRequest(@NotNull JsonObject obj) {
                 if (mid.startsWith(midPrefix)) {
                     RequestListener listener = reqListeners.get(midPrefix);
                     interesting = true;
-                    looper.submit(() -> {
+                    asyncWorker.submit(() -> {
                         RequestResult result = listener.onRequest(mid, pid, sender, command);
                         conn.get().sendReply(key, result);
                         LOGGER.debug(String.format("Handled request. {key: %s, result: %s}", key, result));
@@ -156,7 +160,7 @@ private void handleMessage(@NotNull JsonObject obj) {
                 for (String key : keys) {
                     if (uri.startsWith(key) && !dispatched) {
                         interesting = true;
-                        looper.submit(() -> {
+                        asyncWorker.submit(() -> {
                             try {
                                 listener.onMessage(uri, headers, decodedPayload);
                             } catch (IOException ex) {
@@ -217,6 +221,7 @@ public void close() {
             lastScheduledReconnection = null;
         }
 
+        asyncWorker.close();
         scheduler.shutdown();
         msgListeners.clear();
     }
@@ -259,31 +264,6 @@ public interface MessageListener {
         void onMessage(@NotNull String uri, @NotNull Map<String, String> headers, @NotNull byte[] payload) throws IOException;
     }
 
-    private static class Looper implements Runnable, Closeable {
-        private final BlockingQueue<Runnable> tasks = new LinkedBlockingQueue<>();
-        private boolean shouldStop = false;
-
-        void submit(@NotNull Runnable task) {
-            tasks.add(task);
-        }
-
-        @Override
-        public void run() {
-            while (!shouldStop) {
-                try {
-                    tasks.take().run();
-                } catch (InterruptedException ex) {
-                    break;
-                }
-            }
-        }
-
-        @Override
-        public void close() {
-            shouldStop = true;
-        }
-    }
-
     private class ConnectionHolder implements Closeable {
         private final WebSocket ws;
         private boolean closed = false;

core/src/main/java/xyz/gianlu/librespot/mercury/MercuryClient.java

@@ -37,7 +37,7 @@ public final class MercuryClient extends PacketsManager {
     private final Map<Long, BytesArrayList> partials = new HashMap<>();
 
     public MercuryClient(@NotNull Session session) {
-        super(session);
+        super(session, "mercury");
     }
 
     public void subscribe(@NotNull String uri, @NotNull SubListener listener) throws IOException, PubSubException {

core/src/main/java/xyz/gianlu/librespot/player/AudioKeyManager.java

@@ -29,7 +29,7 @@ public final class AudioKeyManager extends PacketsManager {
     private final Map<Integer, Callback> callbacks = Collections.synchronizedMap(new HashMap<>());
 
     public AudioKeyManager(@NotNull Session session) {
-        super(session);
+        super(session, "audio-keys");
     }
 
     @NotNull