pipeline futures

src/main/java/ru/nbch/credit_tracker/service/task/Pipeline.java

@@ -0,0 +1,31 @@
+package ru.nbch.credit_tracker.service.task;
+
+import java.util.List;
+
+public class Pipeline {
+    private Runnable starter;
+    private List<Stage<?, ?>> stages;
+
+    public Pipeline() {
+    }
+
+    public Stage<?, ?> getLast() {
+        return stages.getLast();
+    }
+
+    public Runnable getStarter() {
+        return starter;
+    }
+
+    public void setStarter(Runnable starter) {
+        this.starter = starter;
+    }
+
+    public List<Stage<?, ?>> getStages() {
+        return stages;
+    }
+
+    public void setStages(List<Stage<?, ?>> stages) {
+        this.stages = stages;
+    }
+}

src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageManager.java

@@ -9,6 +9,8 @@ import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
 import java.util.function.BiFunction;
 import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
@@ -25,6 +27,7 @@ import ru.nbch.credit_tracker.service.task.impl.dprctd.FidIdEnricher;
 import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
 import ru.nbch.credit_tracker.service.task.setup.PipeLineBuilderV2;
 import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
+import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
 
 @Component
 @Slf4j
@@ -38,7 +41,9 @@ public class PipelinePackageManager {
     private final FidIdEnricher fidIdEnricherHandler;
     private final BiFunction<Long, String, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory;
     private final ExecutorService executorService = Executors.newFixedThreadPool(10);
+    private final Semaphore semaphore;
     private final static int BATCH_SIZE = 10_000;
+    private final static int MAX_PARALLEL_REQUESTS = 3;
 
     public PipelinePackageManager(ApplicationContext applicationContext,
                                   StaxXmlProcessor xmlProcessor,
@@ -55,6 +60,7 @@ public class PipelinePackageManager {
         this.saveSubjectHandler = saveSubjectHandler;
         this.fidIdEnricherHandler = fidIdEnricherHandler;
         this.pipelineStarterFactory = pipelineStarterFactory;
+        this.semaphore = new Semaphore(MAX_PARALLEL_REQUESTS);
     }
 
     public void buildAndRun(Long packageId, String pkgPath) throws Exception {
@@ -100,16 +106,43 @@ public class PipelinePackageManager {
     }
 
     public void buildAndRun2(Long packageId, String pkgPath) throws Exception {
-        var startPipeLine = PipeLineBuilderV2.first(saveSubjectQueueFactory.getObject(), idEnricherQueueFactory.getObject())
+        if (!semaphore.tryAcquire()) {
-                .handler(saveSubjectHandler)
+            throw new RuntimeException("Too many concurrent requests");
-                .poison(SubjectPayload.POISON)
+        }
-                .andThen()
+        Pipeline pipeline = null;
-            .stage(sendSignalQueueFactory.getObject())
+        try {
-                .handler(fidIdEnricherHandler)
+            pipeline = PipeLineBuilderV2.first(saveSubjectQueueFactory.getObject(), idEnricherQueueFactory.getObject())
-                .poison(SendSignalsPayload.POISON)
+                    .handler(saveSubjectHandler)
-                .andThen()
+                    .poison(SubjectPayload.POISON)
-            .fillInitialQueue(pipelineStarterFactory.apply(packageId, pkgPath));
+                    .andThen()
-        executorService.submit(startPipeLine);
+                .stage(sendSignalQueueFactory.getObject())
-        log.info("Subjects were registered. PackageId: {}", packageId);
+                    .handler(fidIdEnricherHandler)
+                    .poison(SendSignalsPayload.POISON)
+                    .andThen()
+                .fillInitialQueue(pipelineStarterFactory.apply(packageId, pkgPath));
+            executorService.submit(pipeline.getStarter());
+            pipeline.getLast()
+                .doneFuture()
+                .orTimeout(1, TimeUnit.MINUTES)
+                .whenComplete((v, e) -> {
+                    if (e != null) {
+                        log.error(ExceptionUtils.getStackTrace(e));
+                    }
+                    semaphore.release();
+            });
+        } catch (Exception e) {
+            semaphore.release();
+            throw e;
+        }
+
+//        try {
+//            pipeline.getLast().doneFuture().get(5, TimeUnit.SECONDS);
+//            log.info("Subjects were registered. PackageId: {}", packageId);
+//        } catch (InterruptedException e) {
+//            Thread.currentThread().interrupt();
+//            return;
+//        } catch (ExecutionException|TimeoutException e) {
+//            log.error(ExceptionUtils.getStackTrace(e));
+//        }
     }
 }

src/main/java/ru/nbch/credit_tracker/service/task/Stage.java

@@ -5,6 +5,7 @@ import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
@@ -21,6 +22,7 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
     private final BlockingQueue<Z> out;
     private final Z POISON;
     private final ExecutorService ioPool;
+    private final CompletableFuture<Void> done = new CompletableFuture<>();
 
     public Stage(Handler<X, Z> dispatcher,
                  BlockingQueue<X> in,
@@ -33,6 +35,10 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
         this.POISON = POISON;
     }
 
+    public CompletableFuture<Void> doneFuture() {
+        return done;
+    }
+
     public void startRunning() {
         ioPool.submit(this);
     }
@@ -51,6 +57,7 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
                         f.get();
                     }
                     putQuiet(out, POISON);
+                    done.complete(null);
                     return;
                 }
 
@@ -70,6 +77,7 @@ public class Stage<X extends StagePayload, Z extends StagePayload> implements Ru
         } catch (Exception e) {
             log.error("Error: {}", ExceptionUtils.getStackTrace(e));
             putQuiet(out, POISON);
+            done.complete(null);
         } finally {
             drainAll(inflight, out);
         }

src/main/java/ru/nbch/credit_tracker/service/task/setup/PipeLineBuilderV2.java

@@ -5,6 +5,7 @@ import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.BlockingQueue;
 import ru.nbch.credit_tracker.model.StagePayload;
+import ru.nbch.credit_tracker.service.task.Pipeline;
 import ru.nbch.credit_tracker.service.task.Stage;
 import ru.nbch.credit_tracker.service.task.setup.step.HandlerStep;
 
@@ -34,15 +35,18 @@ public class PipeLineBuilderV2<FIRST extends StagePayload, LAST extends StagePay
         return step;
     }
 
-    public Runnable fillInitialQueue(IPipeLineStarter<FIRST> pipeLineStarter) {
+    public Pipeline fillInitialQueue(IPipeLineStarter<FIRST> pipeLineStarter) {
         Objects.requireNonNull(pipeLineStarter, "pipeLineStarter is required");
         if (firstStage == null) {
             throw new IllegalStateException("Pipeline has no stages");
         }
-        return () -> {
+        Pipeline p = new Pipeline();
+        p.setStarter(() -> {
             stages.forEach(Stage::startRunning);
             pipeLineStarter.startFillingQueue(firstStage.getIn());
-        };
+        });
+        p.setStages(stages);
+        return p;
     }
 
     @SuppressWarnings("unchecked")

src/main/java/ru/nbch/credit_tracker/utils/error/ExceptionUtils.java

@@ -1,5 +1,7 @@
 package ru.nbch.credit_tracker.utils.error;
 
+import java.io.PrintWriter;
+import java.io.StringWriter;
 import ru.nbch.credit_tracker.enums.EnumKeyMessage;
 import ru.nbch.credit_tracker.enums.IEnumKeyCode;
 import ru.nbch.credit_tracker.enums.XmlErrorFormat;
@@ -8,6 +10,13 @@ import ru.nbch.credit_tracker.exceptions.SignalServerException;
 
 public class ExceptionUtils {
 
+    public static String getStackTrace(Throwable throwable) {
+        StringWriter sw = new StringWriter();
+        PrintWriter pw = new PrintWriter(sw, true);
+        throwable.printStackTrace(pw);
+        return sw.getBuffer().toString();
+    }
+
     public static SignalClientException clientExcp(IEnumKeyCode enumKeyCode, XmlErrorFormat xmlErrorFormat, Object... args) {
         return clientExcp(enumKeyCode, null, xmlErrorFormat, args);
     }