diff options
Diffstat (limited to 'src/main')
3 files changed, 48 insertions, 43 deletions
diff --git a/src/main/java/com/it_jaros/jns/scan/Scanner.java b/src/main/java/com/it_jaros/jns/scan/Scanner.java index d9dfe47..e76d804 100644 --- a/src/main/java/com/it_jaros/jns/scan/Scanner.java +++ b/src/main/java/com/it_jaros/jns/scan/Scanner.java @@ -46,33 +46,21 @@ public class Scanner implements AutoCloseable { ); try { - runConsumerLoop(state, consumer); + new ConsumerThread<>(cancelledToken, state).runConsumerLoop(ignored -> { + PollState<ScanResult> poll = getPollStateOfScanResult(state); + switch (poll) { + case PollState.Success<ScanResult>(ScanResult value) -> consumer.accept(value); + case PollState.Failure<ScanResult>(Throwable error) -> mapToScanResult(error).ifPresent(consumer); + case PollState.Unavailable<ScanResult> ignore -> { + // we do nothing + } + } + }); } finally { scan.stop(); } } - /** - * the main thread is the consumer - * Let the consumer run as long as the producer runs - * or if still tasks are pending in pipeline - * we do not listen to canceled here because we want - * all results (also partial) collected for the consumer - * with whatever is there already - */ - private void runConsumerLoop(ProducerState<ScanResult> state, Consumer<ScanResult> consumer) { - while (!cancelledToken.isCancelled() && (state.running().get() || state.inPipeline().get() > 0)) { - PollState<ScanResult> poll = getPollStateOfScanResult(state); - switch (poll) { - case PollState.Success<ScanResult>(ScanResult value) -> consumer.accept(value); - case PollState.Failure<ScanResult>(Throwable error) -> mapToScanResult(error).ifPresent(consumer); - case PollState.Unavailable<ScanResult> ignore -> { - // we do nothing - } - } - } - } - private Optional<ScanResult> mapToScanResult(Throwable error) { if (error instanceof RejectedExecutionException) { // happens when Ctrl-C is pressed diff --git a/src/main/java/com/it_jaros/jns/scan/engine/ConsumerThread.java b/src/main/java/com/it_jaros/jns/scan/engine/ConsumerThread.java new file mode 100644 index 0000000..8cbfcde --- /dev/null +++ b/src/main/java/com/it_jaros/jns/scan/engine/ConsumerThread.java @@ -0,0 +1,28 @@ +package com.it_jaros.jns.scan.engine; + +import java.util.function.Consumer; + +public class ConsumerThread<T> { + + private final CancelledToken cancelledToken; + private final ProducerState<T> state; + + public ConsumerThread(CancelledToken cancelledToken, ProducerState<T> state) { + this.cancelledToken = cancelledToken; + this.state = state; + } + + /** + * the main thread is the consumer + * Let the consumer run as long as the producer runs + * or if still tasks are pending in pipeline + * we do not listen to canceled here because we want + * all results (also partial) collected for the consumer + * with whatever is there already + */ + public void runConsumerLoop(Consumer<Void> callable) { + while (!cancelledToken.isCancelled() && (state.running().get() || state.inPipeline().get() > 0)) { + callable.accept(null); + } + } +} diff --git a/src/main/java/com/it_jaros/jns/scan/engine/ScanHostTask.java b/src/main/java/com/it_jaros/jns/scan/engine/ScanHostTask.java index 086a864..87b5f95 100644 --- a/src/main/java/com/it_jaros/jns/scan/engine/ScanHostTask.java +++ b/src/main/java/com/it_jaros/jns/scan/engine/ScanHostTask.java @@ -73,7 +73,16 @@ public class ScanHostTask implements Callable<ScanResult> { port -> new ScanPortTask(port, context, hostContext) ); - runConsumerLoop(state); + new ConsumerThread<>(cancelledToken, state).runConsumerLoop(ignored -> { + PollState<PortResult> poll = getPortResult(state); + switch (poll) { + case PollState.Success<PortResult>(PortResult value) -> accumulator.add(value); + case PollState.Failure(Throwable error) -> accumulator.recordFailure(error, -1); + case PollState.Unavailable<PortResult> ignore -> { + // we do nothing + } + } + }); return accumulator.build(); } @@ -91,26 +100,6 @@ public class ScanHostTask implements Callable<ScanResult> { return false; } - /** - * consumer is the main thread - * we run as long as the producer is running OR - * as long as things are in pipeline waiting to be processed - * ONLY exception is when cancelled is set - */ - private void runConsumerLoop(ProducerState<PortResult> state) { - while (!context.cancelledToken().isCancelled() && (state.running().get() || state.inPipeline().get() > 0)) { - PollState<PortResult> poll = getPortResult(state); - switch (poll) { - case PollState.Success<PortResult>(PortResult value) -> accumulator.add(value); - case PollState.Failure(Throwable error) -> accumulator.recordFailure(error, -1); - case PollState.Unavailable<PortResult> ignore -> { - // we do nothing - } - } - - } - } - private PollState<PortResult> getPortResult(ProducerState<PortResult> state) { try { return pollPortResult(state); |