Revert "Change vulnerability statement storage to include all stateme…

…nts, even"

This reverts commit 736a80c.

src/main/java/eu/fasten/vulnerabilityproducer/utils/parsers/ParserManager.java

@@ -40,6 +40,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
@@ -69,6 +70,7 @@ public class ParserManager {
     private static final Map<String, String> supportedForges =
             Map.of("mvn", "pkg:maven", "PyPI", "pkg:pypi", "debian", "pkg:deb/debian");
     private final int noParseWorkers = 5;
+    private volatile boolean isPatchFinderDone = false;
 
     public ParserManager(JavaHttpClient client,
                          MongoDatabase mongoDatabase,
@@ -104,6 +106,21 @@ public void getVulnerabilitiesFromParsers(boolean updatesOnly, int numThreadsToU
         }
 
         final var vulnerabilities = parseVulnerabilities(updatesOnly);
+        logger.info("Filtering out vulnerabilities without a forge and storing them to the disk");
+        final var smtQueue = new ConcurrentLinkedQueue<>(vulnerabilities.stream().filter((v) -> !vulnerabilityForgeIsSupported(v)).collect(Collectors.toList()));
+
+        // For writing vuln. statements without blocking other threads
+        Thread smtWriter = new Thread(() -> {
+            while (true) {
+                if (!smtQueue.isEmpty()) {
+                    storeStatement(smtQueue.remove());
+                }
+                else if (isPatchFinderDone) {
+                    break;
+                }
+            }
+        });
+        smtWriter.start();
 
         logger.info("Injecting information about Patches  using " + numThreadsToUse  + " threads");
         var vulnerabilitiesWithForge = vulnerabilities.stream().filter(this::vulnerabilityForgeIsSupported).collect(Collectors.toList());
@@ -112,25 +129,22 @@ public void getVulnerabilitiesFromParsers(boolean updatesOnly, int numThreadsToU
         List<Callable<Object>> patchFindingTasks = new ArrayList<>();
         try(ProgressBar pb = new ProgressBar("InjectPatches", vulnerabilitiesWithForge.size())) {
             for (var v: vulnerabilitiesWithForge) {
-                patchFindingTasks.add(() -> {parsePatchesAndPublish(v, pb);
+                patchFindingTasks.add(() -> {parsePatchesAndPublish(smtQueue, v, pb);
                     return null;
                 });
             }
             try {
                 executor.invokeAll(patchFindingTasks);
                 executor.shutdown();
+                isPatchFinderDone = true;
+                smtWriter.join();
             } catch (InterruptedException e) {
                 throw new RuntimeException(e);
             }
         }
-
-        logger.info("Storing vulnerability statements to disk.");
-        for(var v : vulnerabilities) {
-            storeStatement(v);
-        }
     }
 
-    private void parsePatchesAndPublish(Vulnerability v, ProgressBar pb) {
+    private void parsePatchesAndPublish(ConcurrentLinkedQueue<Vulnerability> smtQueue, Vulnerability v, ProgressBar pb) {
         try {
             var patchFinder = patchFinderFactory.getNewPatchFinder();
             logger.info("Inferring purls for " + v.getId());
@@ -139,6 +153,7 @@ private void parsePatchesAndPublish(Vulnerability v, ProgressBar pb) {
             patchFinder.parseReferences(v, nitriteController);
             logger.info("Publishing " + v.getId() + " to Kafka");
             emitKafkaMsg(v.toJson());
+            smtQueue.add(v);
             pb.step();
         } catch (Exception e) {
             logger.error("Could NOT process " + v.getId() + "\n" + e);