ugh

src/com/alexisdrai/Listener.java

@@ -0,0 +1,25 @@
+package com.alexisdrai;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+public class Listener extends Thread {
+    private final BlockingQueue<Integer> out;
+
+    public Listener(BlockingQueue<Integer> out) {
+        this.out = out;
+    }
+
+    @Override
+    public  void run() {
+        while (!(out.isEmpty())) {
+            System.out.println("resultat " + out.peek());
+//            try {
+//                wait(500);
+//            } catch (InterruptedException e) {
+//                Logger.getLogger(Listener.class.getName()).log(Level.SEVERE, null, e);
+//            }
+        }
+    }
+}

src/com/alexisdrai/Main.java

@@ -1,8 +0,0 @@
-package com.alexisdrai;
-
-public class Main {
-
-    public static void main(String[] args) {
-	// write your code here
-    }
-}

src/com/alexisdrai/Pipeline.java

@@ -0,0 +1,51 @@
+package com.alexisdrai;
+
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+/**
+ * @author sasa
+ */
+public class Pipeline {
+    public static final int PIPE_CAPACITY = 10;
+    public static final int STAGES_ABOVE_FIRST = 3;
+
+    public static void main(String[] argv) {
+
+        BlockingQueue<Integer> first, in, out = null;
+        first = in = new LinkedBlockingQueue<>(PIPE_CAPACITY);
+        for (int i = 0; i < STAGES_ABOVE_FIRST; i++) {
+            out = new LinkedBlockingQueue<>(PIPE_CAPACITY);
+            (new Stage(in, out, -1)).start();
+            in = out;
+        }
+
+        (new Listener(out)).start();
+
+        System.out.println(first.isEmpty());
+
+        try {
+            Thread.sleep(1000);
+        } catch (InterruptedException ex) {
+            Logger.getLogger(Pipeline.class.getName()).log(Level.SEVERE, null, ex);
+        }
+        first.add(2); // 6
+
+        try {
+            Thread.sleep(1000);
+        } catch (InterruptedException ex) {
+            Logger.getLogger(Pipeline.class.getName()).log(Level.SEVERE, null, ex);
+        }
+        first.add(6); // 10
+
+        try {
+            Thread.sleep(1000);
+        } catch (InterruptedException ex) {
+            Logger.getLogger(Pipeline.class.getName()).log(Level.SEVERE, null, ex);
+        }
+        first.add(3); // 7
+    }
+}

src/com/alexisdrai/Stage.java

@@ -0,0 +1,48 @@
+package com.alexisdrai;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+public class Stage extends Thread {
+    private final BlockingQueue<Integer> in;
+    private final BlockingQueue<Integer> out;
+    private int number;
+
+    public Stage(BlockingQueue<Integer> in, BlockingQueue<Integer> out, int number) {
+        this.in = in;
+        this.out = out;
+        this.number = number;
+    }
+
+    @Override
+    public void run() {
+        while (true) {
+//            while (in.isEmpty()) {
+//                try {
+//                    wait();
+//                } catch (InterruptedException e) {
+//                    Logger.getLogger(Stage.class.getName()).log(Level.SEVERE, null, e);
+//                }
+//            }
+            try {
+
+                number = in.take();
+
+            } catch (InterruptedException e) {
+                Logger.getLogger(Stage.class.getName()).log(Level.SEVERE, null, e);
+            }
+
+            number++;
+
+            try {
+
+                out.put(number);
+
+            } catch (InterruptedException e) {
+                Logger.getLogger(Stage.class.getName()).log(Level.SEVERE, null, e);
+            }
+//            out.notifyAll();
+        }
+    }
+}