Added HostScanTask and PortScanTask inner class to reduce mental load of

lambda expressions

1 files changed, 209 insertions, 179 deletions

diff --git a/src/main/java/com/it_jaros/jscanner/Scanner.java b/src/main/java/com/it_jaros/jscanner/Scanner.java
index a0a8c00..a90fc55 100644
--- a/src/main/java/com/it_jaros/jscanner/Scanner.java
+++ b/src/main/java/com/it_jaros/jscanner/Scanner.java
@@ -50,7 +50,6 @@ public class Scanner implements AutoCloseable {
         this(options.socketLimit(), options.timeoutInMillis(), options.delayInMillis(), options.maxWorkersPerHost(), options.maxHostsLimit(), options.disableOnlineCheck(), options.bannerRecognition());
     }
 
-
     /**
      * Starts a given scan.
      *
@@ -67,20 +66,7 @@ public class Scanner implements AutoCloseable {
         final ProducerState<ScanResult> state = startProducer(
                 scan.getHosts().iterator(),
                 maxHostsLimit,
-                host -> () -> {
-                    try {
-                        scan.getThreadCounter().inc();
-                        scan.getHostCounter().inc();
-                        scan.getHostTotalCounter().inc();
-                        if (cancelled) {
-                            return null;
-                        }
-                        return scanHostPorts(host, scan);
-                    } finally {
-                        scan.getThreadCounter().dec();
-                        scan.getHostCounter().dec();
-                    }
-                },
+                host -> new ScanHostTask(scan, host),
                 scan.getThreadCounter()
         );
 
@@ -127,104 +113,6 @@ public class Scanner implements AutoCloseable {
     }
 
     /**
-     * Scans the ports of a given host
-     *
-     * @param host
-     * @param scan
-     * @return
-     */
-    private ScanResult scanHostPorts(final String host, final Scan scan) {
-        PortRange portRange = new PortRange(scan.getPorts());
-        BitSet openPorts = new BitSet(PortRange.MAX_PORT);
-        BitSet filteredPorts = new BitSet(PortRange.MAX_PORT);
-        HashMap<Integer, ServiceType> serviceTypes = new HashMap<>();
-
-        if (!disableOnlineCheck && !checkHostOnline(host)) {
-            // Unreachable host
-            return new ScanResult(host, openPorts, filteredPorts, serviceTypes, List.of());
-        }
-
-        List<ScanFailure> scanFailures = new ArrayList<>();
-        final AtomicLong portSlotFactory = new AtomicLong(System.nanoTime());
-        // producer thread
-        ProducerState<PortResult> state = startProducer(
-                portRange.iterator(),
-                maxWorkersPerHost,
-                port -> () -> {
-                    try {
-                        scan.getThreadCounter().inc();
-                        waitForSlot(portSlotFactory);
-                        if (cancelled) {
-                            return null;
-                        }
-                        return checkPort(host, port, scan);
-                    } finally {
-                        scan.getThreadCounter().dec();
-                    }
-                },
-                scan.getThreadCounter()
-        );
-
-        // consumer is the main thread
-        // we run as long as the producer is running or as long as things are in pipeline to be processed
-        // only exception is when cancelled is set
-        while (!cancelled && (state.running().get() || state.inPipeline().get() > 0)) {
-            try {
-                PortResult portResult = getPortResult(state);
-                if (portResult == null) {
-                    continue;
-                }
-
-                handlePortResult(portResult, openPorts, serviceTypes, filteredPorts, scanFailures);
-            } catch (InterruptedException ignored) {
-                Thread.currentThread().interrupt();
-            }
-        }
-
-        return new ScanResult(host, openPorts, filteredPorts, serviceTypes, scanFailures);
-    }
-
-    private PortResult getPortResult(ProducerState<PortResult> state) throws InterruptedException {
-        Future<PortResult> portResultFuture = state.completionService().poll(10, TimeUnit.MILLISECONDS);
-        if (portResultFuture == null) {
-            return null;
-        }
-
-        PortResult portResult = null;
-        try {
-            portResult = portResultFuture.get();
-        } catch (ExecutionException e) {
-            // something more serious did not work
-            System.err.printf("%s -> %s%n", e.getClass().getSimpleName(), e.getMessage());
-        } finally {
-            state.activeWorkers().release();
-            state.inPipeline().decrementAndGet();
-        }
-        return portResult;
-    }
-
-    private void handlePortResult(PortResult portResult, BitSet openPorts, HashMap<Integer, ServiceType> serviceTypes, BitSet filteredPorts, List<ScanFailure> scanFailures) {
-        switch (portResult.getState()) {
-            case OPEN -> {
-                // bitset is not thread-safe, so it is set
-                // outside the other virtual threads that update progress
-                openPorts.set(portResult.getPort());
-                serviceTypes.put(portResult.getPort(), ServiceDetector.detect(portResult.getBanner()));
-            }
-            case FILTERED -> {
-                filteredPorts.set(portResult.getPort());
-            }
-            default -> {
-                // sonarcube glücklich machen
-            }
-        }
-        Exception e = portResult.getException();
-        if (e != null) {
-            scanFailures.add(new ScanFailure(portResult.getPort(), ExceptionInfo.from(e)));
-        }
-    }
-
-    /**
      * 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
@@ -287,72 +175,6 @@ public class Scanner implements AutoCloseable {
         return new ProducerState<>(running, inPipeline, activeWorkers, completionService);
     }
 
-    private boolean checkHostOnline(String host) {
-        try {
-            return InetAddress.getByName(host).isReachable(timeoutInMillis);
-        } catch (IOException e) {
-            // something went wrong
-        }
-
-        return false;
-    }
-
-    private void waitForSlot(AtomicLong scanSlotFactory) throws InterruptedException {
-        if (delayInNanos > 0) {
-            long slot = scanSlotFactory.getAndAdd(delayInNanos);
-            long wait = slot - System.nanoTime();
-            if (wait > 0)
-                Thread.sleep(Duration.ofNanos(wait));
-        }
-    }
-
-    private PortResult checkPort(String host, int port, Scan scan) throws InterruptedException {
-        PortResult result = new PortResult();
-        result.setPort(port);
-        result.setState(PortState.UNKNOWN);
-
-        // don't allow more sockets then specified
-        socketLimit.acquire();
-        // count how many ports are concurrently checked
-        scan.getSocketCounter().inc();
-        try (Socket socket = new Socket()) {
-            socket.connect(new InetSocketAddress(host, port), timeoutInMillis);
-            result.setState(PortState.OPEN);
-            if (bannerRecognition) {
-                result.setBanner(getBanner(socket));
-            }
-        } catch (SocketTimeoutException ignored) {
-            result.setState(PortState.FILTERED);
-        } catch (ConnectException ignored) {
-            result.setState(PortState.CLOSED);
-        } catch (NoRouteToHostException ignored) {
-            // NoRouteToHostException: this can be safely ignored because the port is closed if a host is unreachable
-            // Will happen a lot when scanning for open ports, so not needed
-        } catch (IOException e) {
-            result.setException(e);
-        } finally {
-            scan.getSocketCounter().dec();
-            socketLimit.release();
-        }
-
-        return result;
-    }
-
-    private String getBanner(Socket socket) {
-        byte[] buffer = new byte[READ_BUFFER_SIZE];
-        try (InputStream input = socket.getInputStream()) {
-            socket.setSoTimeout(timeoutInMillis);
-            int bytesRead = input.read(buffer);
-            if (bytesRead <= 0) {
-                return null;
-            }
-            return new String(buffer, 0, bytesRead, StandardCharsets.UTF_8).trim();
-        } catch (IOException e) {
-            // we ignore this failure
-        }
-        return null;
-    }
-
     public boolean awaitTermination(Duration duration) throws InterruptedException {
         return executor.awaitTermination(duration.toMillis(), TimeUnit.MILLISECONDS);
     }
@@ -375,4 +197,212 @@ public class Scanner implements AutoCloseable {
     public void close() throws Exception {
         cancel();
     }
+
+    private final class ScanHostTask implements Callable<ScanResult> {
+        private final Scan scan;
+        private final String host; // input parameter
+
+        ScanHostTask(Scan scan, String host) {
+            this.scan = scan;
+            this.host = host;
+        }
+
+        @Override
+        public ScanResult call() {
+            try {
+                scan.getThreadCounter().inc();
+                scan.getHostCounter().inc();
+                scan.getHostTotalCounter().inc();
+                if (cancelled) {
+                    return null;
+                }
+                return scanHostPorts(host, scan);
+            } finally {
+                scan.getThreadCounter().dec();
+                scan.getHostCounter().dec();
+            }
+        }
+
+        /**
+         * Scans the ports of a given host
+         *
+         * @param host
+         * @param scan
+         * @return
+         */
+        private ScanResult scanHostPorts(final String host, final Scan scan) {
+            PortRange portRange = new PortRange(scan.getPorts());
+            BitSet openPorts = new BitSet(PortRange.MAX_PORT);
+            BitSet filteredPorts = new BitSet(PortRange.MAX_PORT);
+            HashMap<Integer, ServiceType> serviceTypes = new HashMap<>();
+
+            if (!disableOnlineCheck && !checkHostOnline(host)) {
+                // Unreachable host
+                return new ScanResult(host, openPorts, filteredPorts, serviceTypes, List.of());
+            }
+
+            final AtomicLong portSlotFactory = new AtomicLong(System.nanoTime());
+            // producer thread
+            ProducerState<PortResult> state = startProducer(
+                    portRange.iterator(),
+                    maxWorkersPerHost,
+                    port -> new ScanPortTask(host, port, scan, portSlotFactory),
+                    scan.getThreadCounter()
+            );
+
+            List<ScanFailure> scanFailures = new ArrayList<>();
+            // consumer is the main thread
+            // we run as long as the producer is running or as long as things are in pipeline to be processed
+            // only exception is when cancelled is set
+            while (!cancelled && (state.running().get() || state.inPipeline().get() > 0)) {
+                try {
+                    PortResult portResult = getPortResult(state);
+                    if (portResult == null) {
+                        continue;
+                    }
+
+                    handlePortResult(portResult, openPorts, serviceTypes, filteredPorts, scanFailures);
+                } catch (InterruptedException ignored) {
+                    Thread.currentThread().interrupt();
+                }
+            }
+
+            return new ScanResult(host, openPorts, filteredPorts, serviceTypes, scanFailures);
+        }
+
+        private boolean checkHostOnline(String host) {
+            try {
+                return InetAddress.getByName(host).isReachable(timeoutInMillis);
+            } catch (IOException e) {
+                // something went wrong
+            }
+
+            return false;
+        }
+
+        private PortResult getPortResult(ProducerState<PortResult> state) throws InterruptedException {
+            Future<PortResult> portResultFuture = state.completionService().poll(10, TimeUnit.MILLISECONDS);
+            if (portResultFuture == null) {
+                return null;
+            }
+
+            PortResult portResult = null;
+            try {
+                portResult = portResultFuture.get();
+            } catch (ExecutionException e) {
+                // something more serious did not work
+                System.err.printf("%s -> %s%n", e.getClass().getSimpleName(), e.getMessage());
+            } finally {
+                state.activeWorkers().release();
+                state.inPipeline().decrementAndGet();
+            }
+            return portResult;
+        }
+
+        private void handlePortResult(PortResult portResult, BitSet openPorts, HashMap<Integer, ServiceType> serviceTypes, BitSet filteredPorts, List<ScanFailure> scanFailures) {
+            switch (portResult.getState()) {
+                case OPEN -> {
+                    // bitset is not thread-safe, so it is set
+                    // outside the other virtual threads that update progress
+                    openPorts.set(portResult.getPort());
+                    serviceTypes.put(portResult.getPort(), ServiceDetector.detect(portResult.getBanner()));
+                }
+                case FILTERED -> {
+                    filteredPorts.set(portResult.getPort());
+                }
+                default -> {
+                    // sonarcube glücklich machen
+                }
+            }
+            Exception e = portResult.getException();
+            if (e != null) {
+                scanFailures.add(new ScanFailure(portResult.getPort(), ExceptionInfo.from(e)));
+            }
+        }
+    }
+
+    private final class ScanPortTask implements Callable<PortResult> {
+        private final Scan scan;
+        private final String host;
+        private final int port;
+        private final AtomicLong portSlotFactory;
+
+        private ScanPortTask(String host, int port, Scan scan, AtomicLong portSlotFactory) {
+            this.scan = scan;
+            this.host = host;
+            this.port = port;
+            this.portSlotFactory = portSlotFactory;
+        }
+
+        @Override
+        public PortResult call() throws Exception {
+            try {
+                scan.getThreadCounter().inc();
+                waitForSlot(portSlotFactory);
+                if (cancelled) {
+                    return null;
+                }
+                return checkPort(host, port, scan);
+            } finally {
+                scan.getThreadCounter().dec();
+            }
+        }
+
+        private void waitForSlot(AtomicLong scanSlotFactory) throws InterruptedException {
+            if (delayInNanos > 0) {
+                long slot = scanSlotFactory.getAndAdd(delayInNanos);
+                long wait = slot - System.nanoTime();
+                if (wait > 0)
+                    Thread.sleep(Duration.ofNanos(wait));
+            }
+        }
+
+        private PortResult checkPort(String host, int port, Scan scan) throws InterruptedException {
+            PortResult result = new PortResult();
+            result.setPort(port);
+            result.setState(PortState.UNKNOWN);
+
+            // don't allow more sockets then specified
+            socketLimit.acquire();
+            // count how many ports are concurrently checked
+            // todo object/data asymmetry
+            scan.getSocketCounter().inc();
+            try (Socket socket = new Socket()) {
+                socket.connect(new InetSocketAddress(host, port), timeoutInMillis);
+                result.setState(PortState.OPEN);
+                if (bannerRecognition) {
+                    result.setBanner(getBanner(socket));
+                }
+            } catch (SocketTimeoutException ignored) {
+                result.setState(PortState.FILTERED);
+            } catch (ConnectException ignored) {
+                result.setState(PortState.CLOSED);
+            } catch (NoRouteToHostException ignored) {
+                // NoRouteToHostException: this can be safely ignored because the port is closed if a host is unreachable
+                // Will happen a lot when scanning for open ports, so not needed
+            } catch (IOException e) {
+                result.setException(e);
+            } finally {
+                scan.getSocketCounter().dec();
+                socketLimit.release();
+            }
+
+            return result;
+        }
+
+        private String getBanner(Socket socket) {
+            byte[] buffer = new byte[READ_BUFFER_SIZE];
+            try (InputStream input = socket.getInputStream()) {
+                socket.setSoTimeout(timeoutInMillis);
+                int bytesRead = input.read(buffer);
+                if (bytesRead <= 0) {
+                    return null;
+                }
+                return new String(buffer, 0, bytesRead, StandardCharsets.UTF_8).trim();
+            } catch (IOException e) {
+                // we ignore this failure
+            }
+            return null;
+        }
+    }
 }