fix

Author: Pyrokines17

training/topic8/src/main/java/ru/nsu/gunko/topic8/task1/CommonRes.java

@@ -0,0 +1,17 @@
+package ru.nsu.gunko.topic8.task1;
+
+public class CommonRes {
+    private boolean flag;
+
+    public CommonRes() {
+        flag = true;
+    }
+
+    public boolean getFlag() {
+        return flag;
+    }
+
+    public void setFlag(boolean flag) {
+        this.flag = flag;
+    }
+}

training/topic8/src/main/java/ru/nsu/gunko/topic8/task1/Consumer.java

@@ -4,15 +4,20 @@
 
 public class Consumer implements Runnable {
     private final BlockingDeque<String> storage;
+    private final CommonRes commonRes;
 
-    Consumer(BlockingDeque<String> s) {
+    public Consumer(BlockingDeque<String> s, CommonRes resC) {
         storage = s;
+        commonRes = resC;
     }
 
     public void run() {
         try {
-            while (true) {
-                consume(storage.take());
+            while (commonRes.getFlag()) {
+                if (!storage.isEmpty()) {
+                    consume(storage.take());
+                }
+
                 synchronized (Thread.currentThread()) {
                     try {
                         Thread.currentThread().wait(75);

training/topic8/src/main/java/ru/nsu/gunko/topic8/task1/Main.java

@@ -11,30 +11,83 @@ public static void main(String[] args) {
 
         int countP = Integer.parseInt(args[1]);
         int countC = Integer.parseInt(args[2]);
+        int T = Integer.parseInt(args[3]);
 
-        Producer p = new Producer(storage);
-        Consumer c = new Consumer(storage);
+        ExecutorService serviceP = Executors.newFixedThreadPool(countP);
+        ExecutorService serviceC = Executors.newFixedThreadPool(countC);
 
-        List<Future<?>> futures = new ArrayList<>();
+        List<Future<?>> futuresP = new ArrayList<>();
+        List<Future<?>> futuresC = new ArrayList<>();
 
-        try (ExecutorService service = Executors.newCachedThreadPool()) {
-            for (int i = 0; i < countP; ++i) {
-                futures.add(service.submit(p));
+        CommonRes resP = new CommonRes();
+        CommonRes resC = new CommonRes();
+
+        for (int i = 0; i < countP; ++i) {
+            futuresP.add(serviceP.submit(new Producer(storage, resP)));
+        }
+
+        for (int i = 0; i < countC; ++i) {
+            futuresC.add(serviceC.submit(new Consumer(storage, resC)));
+        }
+
+        synchronized (Thread.currentThread()) {
+            try {
+                Thread.currentThread().wait(T * 1000L);
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
             }
+        }
 
-            for (int i = 0; i < countC; ++i) {
-                futures.add(service.submit(c));
+        serviceP.shutdown();
+        resP.setFlag(false);
+
+        while (true) {
+            if (Check.start(futuresP)) {
+                synchronized (Thread.currentThread()) {
+                    try {
+                        Thread.currentThread().wait(100);
+                    } catch (InterruptedException e) {
+                        throw new RuntimeException(e);
+                    }
+                }
+            } else {
+                System.out.println("Producer's are finished");
+                break;
             }
+        }
 
-            synchronized (Thread.currentThread()) {
-                Thread.currentThread().wait(1000);
+        while (true) {
+            if (!storage.isEmpty()) {
+                synchronized (Thread.currentThread()) {
+                    try {
+                        Thread.currentThread().wait(100);
+                    } catch (InterruptedException e) {
+                        throw new RuntimeException(e);
+                    }
+                }
+            } else {
+                System.out.println("Storage is empty");
+                break;
             }
+        }
 
-            if (Check.start(futures)) {
-                service.shutdownNow();
+        serviceC.shutdown();
+        resC.setFlag(false);
+
+        if (Check.start(futuresC)) {
+            try {
+                if (!serviceC.awaitTermination(10, TimeUnit.SECONDS)) {
+                    serviceC.shutdownNow();
+                    if (!serviceC.awaitTermination(10, TimeUnit.SECONDS)) {
+                        System.err.println("Pool did not terminate");
+                    }
+                }
+            } catch (InterruptedException ex) {
+                serviceC.shutdownNow();
+                Thread.currentThread().interrupt();
             }
-        } catch (InterruptedException ex) {
-            System.err.println(ex.getLocalizedMessage());
         }
+
+        System.out.println("Consumer's are finished");
     }
-}
+}

training/topic8/src/main/java/ru/nsu/gunko/topic8/task1/Producer.java

@@ -5,15 +5,17 @@
 public class Producer implements Runnable {
     private final BlockingDeque<String> storage;
     private int count;
+    private final CommonRes commonRes;
 
-    Producer(BlockingDeque<String> s) {
+    Producer(BlockingDeque<String> s, CommonRes resP) {
         storage = s;
         count = 1;
+        commonRes = resP;
     }
 
     public void run() {
         try {
-            while (true) {
+            while (commonRes.getFlag()) {
                 storage.put(produce());
                 ++count;
             }