1 | package com.renomad.minum.queue; | |
2 | ||
3 | import com.renomad.minum.state.Context; | |
4 | import com.renomad.minum.logging.ILogger; | |
5 | import com.renomad.minum.utils.*; | |
6 | ||
7 | import java.util.concurrent.*; | |
8 | ||
9 | /** | |
10 | * This class provides the ability to pop items into | |
11 | * a queue thread-safely and know they'll happen later. | |
12 | * <p> | |
13 | * For example, this is helpful for minum.logging, or passing | |
14 | * functions to a minum.database. It lets us run a bit faster, | |
15 | * since the I/O actions are happening on a separate | |
16 | * thread and the only time required is passing the | |
17 | * function of what we want to run later. | |
18 | * </p> | |
19 | * <h3>Example:</h3> | |
20 | * <p> | |
21 | * This example shows where an {@link java.io.InputStream} representing the bytes | |
22 | * of an image file are sent to the {@link ActionQueue} for processing. The call | |
23 | * to {@link ActionQueue#enqueue(String, ThrowingRunnable)} returns immediately, | |
24 | * and processing continues on another thread. | |
25 | * </p> | |
26 | * <pre> | |
27 | * {@code | |
28 | * ActionQueue photoResizingQueue; | |
29 | * InputStream photoInputStream; | |
30 | * photoResizingQueue = new ActionQueue("photo_resizing", context).initialize(); | |
31 | * photoResizingQueue.enqueue("resize an image", () -> resizeImage(photoInputStream)); | |
32 | * } | |
33 | * </pre> | |
34 | */ | |
35 | public final class ActionQueue implements AbstractActionQueue { | |
36 | private final String name; | |
37 | private final ExecutorService queueExecutor; | |
38 | private final LinkedBlockingQueue<RunnableWithDescription> queue; | |
39 | private final ILogger logger; | |
40 | private boolean stop = false; | |
41 | private Thread queueThread; | |
42 | private boolean isStoppedStatus; | |
43 | ||
44 | /** | |
45 | * See the {@link ActionQueue} description for more detail. This | |
46 | * constructor will build your new action queue and handle registering | |
47 | * it with a list of other action queues in the {@link Context} object. | |
48 | * @param name give this object a unique, explanatory name. | |
49 | */ | |
50 | public ActionQueue(String name, Context context) { | |
51 | this.name = name; | |
52 | this.queueExecutor = context.getExecutorService(); | |
53 | this.queue = new LinkedBlockingQueue<>(); | |
54 |
1
1. <init> : removed call to com/renomad/minum/queue/ActionQueueState::offerToQueue → KILLED |
context.getActionQueueState().offerToQueue(this); |
55 | this.logger = context.getLogger(); | |
56 | } | |
57 | ||
58 | // Regarding the InfiniteLoopStatement - indeed, we expect that the while loop | |
59 | // below is an infinite loop unless there's an exception thrown, that's what it is. | |
60 | @SuppressWarnings("InfiniteLoopStatement") | |
61 | @Override | |
62 | public ActionQueue initialize() { | |
63 | Runnable centralLoop = () -> { | |
64 | Thread.currentThread().setName(name); | |
65 | this.queueThread = Thread.currentThread(); | |
66 | try { | |
67 | while (true) { | |
68 |
1
1. lambda$initialize$1 : removed call to com/renomad/minum/queue/ActionQueue::runAction → KILLED |
runAction(); |
69 | } | |
70 | } catch (InterruptedException ex) { | |
71 | /* | |
72 | this is what we expect to happen. | |
73 | once this happens, we just continue on. | |
74 | this only gets called when we are trying to shut everything | |
75 | down cleanly | |
76 | */ | |
77 | logger.logDebug(() -> String.format("%s ActionQueue for %s is stopped.%n", TimeUtils.getTimestampIsoInstant(), name)); | |
78 | Thread.currentThread().interrupt(); | |
79 | } | |
80 | }; | |
81 | queueExecutor.submit(centralLoop); | |
82 |
1
1. initialize : replaced return value with null for com/renomad/minum/queue/ActionQueue::initialize → KILLED |
return this; |
83 | } | |
84 | ||
85 | private void runAction() throws InterruptedException { | |
86 | RunnableWithDescription action = queue.take(); | |
87 | try { | |
88 |
1
1. runAction : removed call to com/renomad/minum/utils/RunnableWithDescription::run → KILLED |
action.run(); |
89 | } catch (Exception e) { | |
90 | logger.logAsyncError(() -> StacktraceUtils.stackTraceToString(e)); | |
91 | } | |
92 | } | |
93 | ||
94 | /** | |
95 | * Adds something to the queue to be processed. | |
96 | * <p> | |
97 | * Here is an example use of .enqueue: | |
98 | * </p> | |
99 | * <p> | |
100 | * <pre> | |
101 | * {@code actionQueue.enqueue("Write person file to disk at " + filePath, () -> { | |
102 | * Files.writeString(filePath, pf.serialize()); | |
103 | * });} | |
104 | * </pre> | |
105 | * </p> | |
106 | */ | |
107 | @Override | |
108 | public void enqueue(String description, ThrowingRunnable action) { | |
109 |
1
1. enqueue : negated conditional → KILLED |
if (! stop) { |
110 | queue.add(new RunnableWithDescription(action, description)); | |
111 | } else { | |
112 | throw new UtilsException(String.format("failed to enqueue %s - ActionQueue \"%s\" is stopped", description, this.name)); | |
113 | } | |
114 | } | |
115 | ||
116 | /** | |
117 | * Stops the action queue | |
118 | * @param count how many loops to wait before we crash it closed | |
119 | * @param sleepTime how long to wait in milliseconds between loops | |
120 | */ | |
121 | @Override | |
122 | public void stop(int count, int sleepTime) { | |
123 | String timestamp = TimeUtils.getTimestampIsoInstant(); | |
124 | logger.logDebug(() -> String.format("%s Stopping queue %s", timestamp, this)); | |
125 | stop = true; | |
126 |
2
1. stop : negated conditional → KILLED 2. stop : changed conditional boundary → KILLED |
for (int i = 0; i < count; i++) { |
127 |
1
1. stop : negated conditional → KILLED |
if (queue.isEmpty()) return; |
128 | logger.logDebug(() -> String.format("%s Queue not yet empty, has %d elements. waiting...%n",timestamp, queue.size())); | |
129 | MyThread.sleep(sleepTime); | |
130 | } | |
131 | isStoppedStatus = true; | |
132 | logger.logDebug(() -> String.format("%s Queue %s has %d elements left but we're done waiting. Queue toString: %s", timestamp, this, queue.size(), queue)); | |
133 | } | |
134 | ||
135 | /** | |
136 | * This will prevent any new actions being | |
137 | * queued (by setting the stop flag to true and thus | |
138 | * causing an exception to be thrown | |
139 | * when a call is made to [enqueue]) and will | |
140 | * block until the queue is empty. | |
141 | */ | |
142 | @Override | |
143 | public void stop() { | |
144 |
1
1. stop : removed call to com/renomad/minum/queue/ActionQueue::stop → KILLED |
stop(5, 20); |
145 | } | |
146 | ||
147 | @Override | |
148 | public String toString() { | |
149 |
1
1. toString : replaced return value with "" for com/renomad/minum/queue/ActionQueue::toString → KILLED |
return this.name; |
150 | } | |
151 | ||
152 | Thread getQueueThread() { | |
153 |
1
1. getQueueThread : replaced return value with null for com/renomad/minum/queue/ActionQueue::getQueueThread → KILLED |
return queueThread; |
154 | } | |
155 | ||
156 | @Override | |
157 | public LinkedBlockingQueue<RunnableWithDescription> getQueue() { | |
158 |
1
1. getQueue : replaced return value with null for com/renomad/minum/queue/ActionQueue::getQueue → KILLED |
return new LinkedBlockingQueue<>(queue); |
159 | } | |
160 | ||
161 | @Override | |
162 | public boolean isStopped() { | |
163 |
2
1. isStopped : replaced boolean return with true for com/renomad/minum/queue/ActionQueue::isStopped → KILLED 2. isStopped : replaced boolean return with false for com/renomad/minum/queue/ActionQueue::isStopped → KILLED |
return isStoppedStatus; |
164 | } | |
165 | } | |
Mutations | ||
54 |
1.1 |
|
68 |
1.1 |
|
82 |
1.1 |
|
88 |
1.1 |
|
109 |
1.1 |
|
126 |
1.1 2.2 |
|
127 |
1.1 |
|
144 |
1.1 |
|
149 |
1.1 |
|
153 |
1.1 |
|
158 |
1.1 |
|
163 |
1.1 2.2 |