ConditionDemo: Using Condition locks

ConditionDemo/src/ConditionDemo.java

@@ -0,0 +1,22 @@
+// Running the world
+//
+
+public class ConditionDemo {
+
+	public static void main(String[] args) throws InterruptedException {
+        SharedFiFoQueue sharedQueue = new SharedFiFoQueue(10);
+         
+        //Create a producer and a consumer.
+        Thread producer = new Producer(sharedQueue);
+        Thread consumer = new Consumer(sharedQueue);
+         
+        //Start both threads.
+        producer.start();
+        consumer.start();
+         
+        //Wait for both threads to terminate.
+        producer.join();
+        consumer.join();
+    }
+
+}

ConditionDemo/src/Consumer.java

@@ -0,0 +1,42 @@
+// Consuming the words from the queue.
+// Counting total number of words and keeping track of unique words
+//
+import java.util.HashSet;
+import java.util.Set;
+ 
+public class Consumer extends Thread {
+    private final Set knownObjects = new HashSet();
+    private int total = 0;
+    private int unique = 0;
+    private final SharedFiFoQueue queue;
+ 
+    public Consumer(SharedFiFoQueue queue) {
+        this.queue = queue;
+    }
+     
+    @Override
+    public void run() {
+        try {
+            do {
+                Object obj = queue.remove();
+                if(obj == null)
+                    break;
+                total++;
+                 
+                if(!knownObjects.contains(obj)) {
+                    unique++;
+                    knownObjects.add(obj);
+                }
+                 
+                System.out.println("[Consumer] Read the element: " + obj.toString());
+                 
+            } while(true);
+        }
+        catch (InterruptedException ex) {
+            System.err.println("An InterruptedException was caught: " + ex.getMessage());
+            ex.printStackTrace();
+        }
+         
+        System.out.println("\n[Consumer] " + total + " words read counting " + unique + " unique words");
+    }
+}

ConditionDemo/src/Producer.java

@@ -0,0 +1,57 @@
+// The file "input.txt" is read on a line basis.
+// Words are separated and fed to the queue
+// Finally a null object is fed to the queue to signal the consumer the end of operation
+
+
+import java.io.BufferedReader;
+import java.io.FileReader;
+import java.io.IOException;
+ 
+public class Producer extends Thread {
+    private final static String FILENAME = "input.txt";
+    private final SharedFiFoQueue queue;
+     
+    public Producer(SharedFiFoQueue queue) {
+        this.queue = queue;
+    }
+     
+    @Override
+    public void run() {
+        BufferedReader rd = null;
+         
+        try {
+            rd = new BufferedReader(new FileReader(FILENAME));
+             
+            String inputLine = null;
+            while((inputLine = rd.readLine()) != null) {
+                String[] inputWords = inputLine.split(" ");
+                 
+                for(String inputWord: inputWords) {
+                    queue.add(inputWord);
+                    System.out.println("[Producer] Wrote the element: " + inputWord);
+                }
+            }
+             
+            // End of the world
+            queue.add(null);
+        }
+        catch (InterruptedException ex) {
+            System.err.println("An InterruptedException was caught: " + ex.getMessage());
+            ex.printStackTrace();
+        }
+        catch (IOException ex) {
+            System.err.println("An IOException was caught: " + ex.getMessage());
+            ex.printStackTrace();
+        }
+        finally {
+            try {
+                if(rd != null)
+                    rd.close();
+            }
+            catch (IOException ex) {
+                System.err.println("An IOException was caught: " + ex.getMessage());
+                ex.printStackTrace();
+            }
+        }
+    }
+}

ConditionDemo/src/SharedFiFoQueue.java

@@ -0,0 +1,57 @@
+// Similar to the Circular buffer example, but with objects
+
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+ 
+public class SharedFiFoQueue {
+ 
+    private Object[] elems = null;
+    private int PosW = 0;
+    private int PosR = 0;
+     
+    private final Lock lock = new ReentrantLock();
+    private final Condition isEmpty = lock.newCondition();
+    private final Condition isFull = lock.newCondition();
+     
+    public SharedFiFoQueue(int capacity) {
+        this.elems = new Object[capacity];
+    }
+     
+    public void add(Object elem) throws InterruptedException {
+        lock.lock();
+        while( ((PosW + 1) % elems.length) == PosR)
+            isFull.await();
+     
+        elems[PosW] = elem;
+         
+        //We need the modulo, in order to avoid going out of bounds.
+        PosW = (PosW + 1) % elems.length;
+                  
+        //Notify the consumer that there is data available.
+        isEmpty.signal();
+         
+        lock.unlock();
+    }
+ 
+    public Object remove() throws InterruptedException {
+        Object elem = null;
+         
+        lock.lock();
+        while( PosR == PosW )
+            isEmpty.await();
+     
+        elem = elems[PosR];
+ 
+        //We need the modulo, in order to avoid going out of bounds.
+        PosR = (PosR + 1) % elems.length;
+         
+         
+        //Notify the producer that there is space available.
+        isFull.signal();
+         
+        lock.unlock();
+         
+        return elem;
+    }
+}