finished coding this shit that was fun

src/main/java/com/producerConsumerLab/App.java

 package com.producerConsumerLab;
 
-/**
- * Hello world!
- *
- */
-public class App 
-{
-    public static void main( String[] args )
-    {
-        System.out.println( "Hello World!" );
+import java.io.FileNotFoundException;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+public class App {
+
+    public static void main(String[] args) {
+
+        String filePath = "/home/drnull/UNI/course_p/lab4/producerConsumerLab/src/main/java/resources/Poem.txt";
+        String outputPath = "/home/drnull/UNI/course_p/lab4/producerConsumerLab/src/main/java/resources/output/Poem_out.txt";
+
+        
+        BlockingQueue<Message> fileContentQueue = new LinkedBlockingQueue<>();
+        BlockingQueue<Message> postProcessingQueue = new LinkedBlockingQueue<>();
+
+        int numCapConsumers = 3;
+
+        
+        try {
+            Producer producer = new Producer(filePath, fileContentQueue);
+            producer.start();
+
+            
+            for (int i = 0; i < numCapConsumers; i++) {
+                ConsumerCap capConsumer = new ConsumerCap(fileContentQueue, postProcessingQueue);
+                capConsumer.setName("CapConsumer-" + i);
+                capConsumer.start();
+            }
+
+            
+            ConsumerWrite writerConsumer = new ConsumerWrite(postProcessingQueue, outputPath);
+            writerConsumer.setName("WriterConsumer");
+            writerConsumer.start();
+
+            
+            producer.join();
+
+            
+            for (int i = 0; i < numCapConsumers; i++) {
+                fileContentQueue.put(Message.poisonPill());
+            }
+
+            
+            writerConsumer.join();
+
+            System.out.println("Pipeline processing complete.");
+
+        } catch (FileNotFoundException e) {
+            System.err.println("File not found: " + e.getMessage());
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
     }
 }

src/main/java/com/producerConsumerLab/ConsumerCap.java

+package com.producerConsumerLab;
+
+import java.io.FileNotFoundException;
+import java.lang.Thread;
+import java.util.concurrent.BlockingQueue;
+
+public class ConsumerCap extends Thread {
+    private final BlockingQueue<Message> fileContentQueue;
+    private final BlockingQueue<Message> postProcessingQueue;
+
+    public ConsumerCap(BlockingQueue<Message> fileContentQueue, BlockingQueue<Message> postProcessingQueue)
+            throws FileNotFoundException {
+        // Step 1 initialize the queue refrence
+        this.fileContentQueue = fileContentQueue;
+        this.postProcessingQueue = postProcessingQueue;
+    }
+
+    @Override
+    public void run() {
+        try {
+            while (true) {
+                Message msg = fileContentQueue.take();
+                if (msg.poison) {
+                    // forward one poison pill downstream and exit
+                    postProcessingQueue.put(Message.poisonPill());
+                    break;
+                }
+
+                // process normal message
+                Message out = new Message(msg.payload.toUpperCase());
+                postProcessingQueue.put(out);
+            }
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+    }
+}
\ No newline at end of file

src/main/java/com/producerConsumerLab/ConsumerWrite.java

+package com.producerConsumerLab;
+
+import java.io.BufferedWriter;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+public class ConsumerWrite extends Thread {
+
+    private final BlockingQueue<Message> postProcessingQueue;
+    private final String filePath;
+
+    public ConsumerWrite(BlockingQueue<Message> postProcessingQueue, String filePath) {
+        this.postProcessingQueue = postProcessingQueue;
+        this.filePath = filePath;
+    }
+
+    @Override
+    public void run() {
+
+        try (BufferedWriter writer = new BufferedWriter(new FileWriter(filePath))) {
+
+            while (true) {
+
+                // Step 1 
+                Message msg = postProcessingQueue.poll(300, TimeUnit.MILLISECONDS);
+
+                if (msg == null) {
+                    
+                    break;
+                }
+
+                if (msg.poison) {
+                    // posisons may be out of sync so we don't stop upon first poistion
+                    continue;
+                }
+
+                // Write output line
+                writer.write(msg.payload);
+                writer.newLine();
+            }
+
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+    }
+}

src/main/java/com/producerConsumerLab/Message.java

+// Message.java
+package com.producerConsumerLab;
+
+public class Message {
+    public final String payload;
+    public final boolean poison; 
+
+    public Message(String payload) {
+        this.payload = payload;
+        this.poison = false;
+    }
+
+    private Message(boolean poison) {
+        this.payload = null;
+        this.poison = poison;
+    }
+
+    public static Message poisonPill() { return new Message(true); }
+}

src/main/java/com/producerConsumerLab/Producer.java

+package com.producerConsumerLab;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.util.Scanner;
+import java.lang.Thread;
+import java.util.concurrent.BlockingQueue;
+//import java.util.concurrent.LinkedBlockingQueue;
+// damn didn't reliaze that # is alot better for comments then //
+
+public class Producer extends Thread {
+
+    private final BlockingQueue<Message> fileContentQueue;
+    private final Scanner scanner;
+
+    public Producer(String filePath, BlockingQueue<Message> fileContentQueue) throws FileNotFoundException {
+        // Step 1 check if file exist
+
+        File file = new File(filePath);
+
+        // Step 2 making an instace of the scanner
+
+        scanner = new Scanner(file);
+
+        // Step 3 assign this queue
+        this.fileContentQueue = fileContentQueue;
+    }
+
+    @Override
+    public void run() {
+        // Step 3 start the scanner and put the data in a common queue
+        // as advised in the homework title i should use concurrent data structure
+        // how do i make this data structure shared through i don't remember
+
+        try {
+            while (scanner.hasNextLine()) {
+                fileContentQueue.put(new Message(scanner.nextLine()));
+            }
+
+            // Signal to consumers that we are done
+            //moved this logic to main thread 
+            //because main thread knows how many second lvl consumers are there
+            //fileContentQueue.put("EOF");
+
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt(); 
+        } finally {
+            scanner.close();
+        }
+    }
+}

src/main/java/resources/Poem.txt

+No man is an island,
+Entire of itself;
+Every man is a piece of the continent,
+A part of the main.
+If a clod be washed away by the sea,
+Europe is the less,
+As well as if a promontory were:
+As well as if a manor of thy friend's
+Or of thine own were.
+Any man's death diminishes me,
+Because I am involved in mankind.
+And therefore never send to know for whom the bell tolls;
+It tolls for thee.
+Lorem ipsum dolor sit amet, consectetuer adipiscing elit. Aenean commodo ligula eget dolor.
+Aenean massa. 
+Cum sociis natoque penatibus et magnis dis parturient montes,
+nascetur ridiculus mus. Donec quam felis, ultricies nec, pellentesque eu, pretium quis, 
+em.
+Nulla consequat massa quis enim.
+Donec pede justo, fringilla vel, aliquet nec, vulputate eget, arcu.
+In enim justo, rhoncus ut, imperdiet a, venenatis vitae, justo. Nullam dictum felis eu pede mollis pretium.
+Integer tincidunt. Cras dapibus.
+Vivamus elementum semper nisi.
+Aenean vulputate eleifend tellus.
+Aenean leo ligula, porttitor eu, consequat vitae, eleifend ac, enim. Aliquam lorem ante,
+dapibus in, viverra quis, feugiat a, tellus. Phasellus viverra nulla ut metus varius laoreet.
+Quisque rutrum. Aenean imperdiet. Etiam ultricies nisi vel augue.
+Curabitur ullamcorper ultricies nisi. Nam eget dui. Etiam rhoncus.
+Maecenas tempus, tellus eget condimentum rhoncus, sem quam semper libero,
+sit amet adipiscing sem neque sed ipsum. Nam quam nunc, blandit vel,
+luctus pulvinar, hendrerit id, lorem. Maecenas nec odio et ante tincidunt tempus.
+Donec vitae sapien ut libero venenatis faucibus.
+Nullam quis ante. Etiam sit amet orci eget eros faucibus tincidunt.
+Duis leo. Sed fringilla mauris sit amet nibh.
+Donec sodales sagittis magna. Sed consequat,
+leo eget bibendum sodales, augue velit cursus nunc,
\ No newline at end of file

src/main/java/resources/output/Poem_out.txt

+NO MAN IS AN ISLAND,
+ENTIRE OF ITSELF;
+EVERY MAN IS A PIECE OF THE CONTINENT,
+A PART OF THE MAIN.
+IF A CLOD BE WASHED AWAY BY THE SEA,
+EUROPE IS THE LESS,
+AS WELL AS IF A PROMONTORY WERE:
+AS WELL AS IF A MANOR OF THY FRIEND'S
+OR OF THINE OWN WERE.
+ANY MAN'S DEATH DIMINISHES ME,
+BECAUSE I AM INVOLVED IN MANKIND.
+AND THEREFORE NEVER SEND TO KNOW FOR WHOM THE BELL TOLLS;
+IT TOLLS FOR THEE.
+LOREM IPSUM DOLOR SIT AMET, CONSECTETUER ADIPISCING ELIT. AENEAN COMMODO LIGULA EGET DOLOR.
+AENEAN MASSA. 
+CUM SOCIIS NATOQUE PENATIBUS ET MAGNIS DIS PARTURIENT MONTES,
+NASCETUR RIDICULUS MUS. DONEC QUAM FELIS, ULTRICIES NEC, PELLENTESQUE EU, PRETIUM QUIS, 
+EM.
+NULLA CONSEQUAT MASSA QUIS ENIM.
+DONEC PEDE JUSTO, FRINGILLA VEL, ALIQUET NEC, VULPUTATE EGET, ARCU.
+IN ENIM JUSTO, RHONCUS UT, IMPERDIET A, VENENATIS VITAE, JUSTO. NULLAM DICTUM FELIS EU PEDE MOLLIS PRETIUM.
+INTEGER TINCIDUNT. CRAS DAPIBUS.
+VIVAMUS ELEMENTUM SEMPER NISI.
+AENEAN VULPUTATE ELEIFEND TELLUS.
+AENEAN LEO LIGULA, PORTTITOR EU, CONSEQUAT VITAE, ELEIFEND AC, ENIM. ALIQUAM LOREM ANTE,
+DAPIBUS IN, VIVERRA QUIS, FEUGIAT A, TELLUS. PHASELLUS VIVERRA NULLA UT METUS VARIUS LAOREET.
+QUISQUE RUTRUM. AENEAN IMPERDIET. ETIAM ULTRICIES NISI VEL AUGUE.
+CURABITUR ULLAMCORPER ULTRICIES NISI. NAM EGET DUI. ETIAM RHONCUS.
+MAECENAS TEMPUS, TELLUS EGET CONDIMENTUM RHONCUS, SEM QUAM SEMPER LIBERO,
+SIT AMET ADIPISCING SEM NEQUE SED IPSUM. NAM QUAM NUNC, BLANDIT VEL,
+LUCTUS PULVINAR, HENDRERIT ID, LOREM. MAECENAS NEC ODIO ET ANTE TINCIDUNT TEMPUS.
+DONEC VITAE SAPIEN UT LIBERO VENENATIS FAUCIBUS.
+NULLAM QUIS ANTE. ETIAM SIT AMET ORCI EGET EROS FAUCIBUS TINCIDUNT.
+DUIS LEO. SED FRINGILLA MAURIS SIT AMET NIBH.
+DONEC SODALES SAGITTIS MAGNA. SED CONSEQUAT,
+LEO EGET BIBENDUM SODALES, AUGUE VELIT CURSUS NUNC,