1 files changed, 123 insertions, 0 deletions
diff --git a/src/main/java/com/it_jaros/jscanner/scan/engine/ScanHostTask.java b/src/main/java/com/it_jaros/jscanner/scan/engine/ScanHostTask.java
new file mode 100644
index 0000000..7a01d54
--- /dev/null
+++ b/src/main/java/com/it_jaros/jscanner/scan/engine/ScanHostTask.java
@@ -0,0 +1,123 @@
+package com.it_jaros.jscanner.scan.engine;
+
+import com.it_jaros.jscanner.scan.Scan;
+import com.it_jaros.jscanner.scan.domain.PortRange;
+import com.it_jaros.jscanner.scan.domain.PortResult;
+import com.it_jaros.jscanner.scan.domain.ScanResult;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.LockSupport;
+
+public class ScanHostTask implements Callable<ScanResult> {
+
+    private final Scan scan;
+    private final String host; // input parameter
+    private final ScanExecutionContext context;
+    private final CancelledToken cancelledToken;
+    private final boolean disableOnlineCheck;
+    private final int maxWorkersPerHost;
+    private final int timeoutInMillis;
+    private final long delayInNanos;
+
+    public ScanHostTask(Scan scan, String host, ScanExecutionContext context) {
+        this.scan = scan;
+        this.host = host;
+        this.context = context;
+        this.cancelledToken = context.cancelledToken();
+        this.delayInNanos = TimeUnit.MILLISECONDS.toNanos(Math.max(0, context.scanOptions().delayInMillis()));
+        this.disableOnlineCheck = context.scanOptions().disableOnlineCheck();
+        this.maxWorkersPerHost = context.scanOptions().maxWorkersPerHost();
+        this.timeoutInMillis = context.scanOptions().timeoutInMillis();
+    }
+
+    @Override
+    public ScanResult call() {
+        try {
+            scan.hostStart();
+            if (context.cancelledToken().isCancelled()) {
+                return ScanResult.empty(host);
+            }
+            return scanHostPorts();
+        } finally {
+            scan.hostFinish();
+        }
+    }
+
+    private ScanResult scanHostPorts() {
+        if (!disableOnlineCheck) {
+            boolean isHostOnline = checkHostOnline();
+            if (!isHostOnline) {
+                // Unreachable host
+                return ScanResult.empty(host);
+            }
+            // online check also sends packets to the target system.
+            // in order not to violate set delay time
+            // we wait here too
+            LockSupport.parkNanos(delayInNanos);
+        }
+
+        final PortRange portRange = new PortRange(scan.getPorts());
+        // producer thread
+        scan.producerStart();
+        ProducerState<PortResult> state = new ProducerThread(context).startProducer(
+                portRange.iterator(),
+                maxWorkersPerHost,
+                port -> new ScanPortTask(host, port, context)
+        );
+
+        // 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
+        final PortResultAccumulator accumulator = new PortResultAccumulator(host);
+        while (!cancelledToken.isCancelled() && (state.running().get() || state.inPipeline().get() > 0)) {
+            try {
+                PollState<PortResult> poll = getPortResult(state);
+                if (poll instanceof PollState.Success<PortResult>(PortResult value)) {
+                    accumulator.add(value);
+                } else if (poll instanceof PollState.Failure(Throwable error)) {
+                    System.err.printf("scanHostPorts(%s): ScanPortTask() failed for with error %s -> %s%n", host, error.getClass().getSimpleName(), error.getMessage());
+                }
+            } catch (InterruptedException ignored) {
+                Thread.currentThread().interrupt();
+            }
+        }
+        scan.producerStop();
+
+        return accumulator.build();
+    }
+
+    private boolean checkHostOnline() {
+        try {
+            return InetAddress.getByName(host).isReachable(timeoutInMillis);
+        } catch (IOException e) {
+            // we ignore this error because it means that the host is probably not online
+        }
+
+        return false;
+    }
+
+    private PollState<PortResult> getPortResult(ProducerState<PortResult> state) throws InterruptedException {
+        Future<PortResult> portResultFuture = state.completionService().poll(ProducerThread.pollInterval.toMillis(), TimeUnit.MILLISECONDS);
+        if (portResultFuture == null) {
+            return new PollState.Unavailable<>();
+        }
+
+        PortResult portResult;
+        try {
+            portResult = portResultFuture.get();
+            return new PollState.Success<>(portResult);
+        } catch (ExecutionException e) {
+            Throwable cause = e.getCause();
+            return new PollState.Failure<>(cause);
+        } finally {
+            state.activeWorkers().release();
+            state.inPipeline().decrementAndGet();
+        }
+    }
+}