From 1cbb5b6ffd5afdf55ea6727ee23550796a57809b Mon Sep 17 00:00:00 2001 From: Matthias Jaros Date: Wed, 12 Aug 2026 16:36:28 +0200 Subject: Renamed jscanner package to jns --- .../jscanner/scan/engine/ProducerThread.java | 83 ---------------------- 1 file changed, 83 deletions(-) delete mode 100644 src/main/java/com/it_jaros/jscanner/scan/engine/ProducerThread.java (limited to 'src/main/java/com/it_jaros/jscanner/scan/engine/ProducerThread.java') diff --git a/src/main/java/com/it_jaros/jscanner/scan/engine/ProducerThread.java b/src/main/java/com/it_jaros/jscanner/scan/engine/ProducerThread.java deleted file mode 100644 index e7bcbcc..0000000 --- a/src/main/java/com/it_jaros/jscanner/scan/engine/ProducerThread.java +++ /dev/null @@ -1,83 +0,0 @@ -package com.it_jaros.jscanner.scan.engine; - -import java.time.Duration; -import java.util.Iterator; -import java.util.concurrent.*; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.function.Function; - -public class ProducerThread { - - public static final Duration pollInterval = Duration.ofSeconds(1); - - private final ExecutorService executor; - private final CancelledToken cancelledToken; - - public ProducerThread(ScanExecutionContext context) { - this.executor = context.executorService(); - this.cancelledToken = context.cancelledToken(); - } - - /** - * This method helps to cleanup the code a bit and remove redundancy - * The producer for providing hosts and the one for providing ports - * are similar and the small differences can be handled using a function - * - * @param queue - * @param maxWorkers - * @param taskFactory - * @param - * @param - * @return - */ - public ProducerState startProducer( - Iterator queue, - int maxWorkers, - Function> taskFactory - ) { - final AtomicInteger inPipeline = new AtomicInteger(0); - final AtomicBoolean running = new AtomicBoolean(true); - final Semaphore activeWorkers = new Semaphore(maxWorkers); - CompletionService completionService = new ExecutorCompletionService<>(executor); - executor.submit(() -> { - try { - while (!cancelledToken.isCancelled() && queue.hasNext()) { - // get semaphore and remember if task got submitted - // so in case we fail to submit we release the semaphore - activeWorkers.acquire(); - boolean isTaskSubmitted = false; - try { - // just in case something - // changed while waiting - if (cancelledToken.isCancelled()) { - break; - } - - // get next item and create callable - // using lambda expression - final INPUT item = queue.next(); - Callable task = taskFactory.apply(item); - inPipeline.incrementAndGet(); - try { - completionService.submit(task); - isTaskSubmitted = true; - } catch (Throwable e) { - inPipeline.decrementAndGet(); - throw e; - } - } finally { - if (!isTaskSubmitted) { - activeWorkers.release(); - } - } - } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } finally { - running.set(false); - } - }); - return new ProducerState<>(running, inPipeline, activeWorkers, completionService); - } -} -- cgit v1.3.1