From 0ea9fb845f4462c62010faff6719c76040f66ae8 Mon Sep 17 00:00:00 2001
From: asy <asy>
Date: Thu, 16 Feb 2017 20:19:21 -0500
Subject: [PATCH 1/2] OrderedLinkifyObject and SequentialOutputBuffer

---
 .gitignore                                    |  2 +
 DesignNotes.txt                               | 12 +++
 pom.xml                                       | 94 ++++++++++---------
 .../java/sy/andrew/linkifier/Linkifier.java   |  2 +-
 .../linkifier/SequentialOutputBuffer.java     | 77 +++++++++++++++
 .../linkifier/model/OrderedLinkifyObject.java | 29 ++++++
 .../model/OrderedLinkifyObjectComparator.java | 15 +++
 .../linkifier/SequentialOutputBufferTest.java | 76 +++++++++++++++
 .../OrderedLinkifyObjectComparatorTest.java   | 50 ++++++++++
 9 files changed, 312 insertions(+), 45 deletions(-)
 create mode 100644 DesignNotes.txt
 create mode 100644 src/main/java/sy/andrew/linkifier/SequentialOutputBuffer.java
 create mode 100644 src/main/java/sy/andrew/linkifier/model/OrderedLinkifyObject.java
 create mode 100644 src/main/java/sy/andrew/linkifier/model/OrderedLinkifyObjectComparator.java
 create mode 100644 src/test/java/sy/andrew/linkifier/SequentialOutputBufferTest.java
 create mode 100644 src/test/java/sy/andrew/linkifier/model/OrderedLinkifyObjectComparatorTest.java

diff --git a/.gitignore b/.gitignore
index b3f8f69..db378e4 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,4 +1,6 @@
 target/
+*tmp
+tmp*
 
 ##############
 # Eclipse IDE
diff --git a/DesignNotes.txt b/DesignNotes.txt
new file mode 100644
index 0000000..8252a13
--- /dev/null
+++ b/DesignNotes.txt
@@ -0,0 +1,12 @@
+Main Code Features
+- uses guava: great library (collections, etc.), cleaner code
+- fluent api?
+- TreeRangeSet queries is O(log n + k) time and O(n) space, where n is number of interval points and k is number of intervals returned ?
+- RangeSetUtil 
+  - use of generics (template method)
+  - 
+
+
+Test Code Features
+- use hamcrest
+- use mock ? 
\ No newline at end of file
diff --git a/pom.xml b/pom.xml
index 96586f4..2aff456 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1,44 +1,50 @@
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-	<modelVersion>4.0.0</modelVersion>
-
-	<groupId>sy.andrew</groupId>
-	<artifactId>linkifier</artifactId>
-	<version>0.0.1-SNAPSHOT</version>
-	<packaging>jar</packaging>
-
-	<name>linkifier</name>
-	<url>http://maven.apache.org</url>
-
-	<properties>
-		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-	</properties>
-
-	<build>
-		<plugins>
-			<plugin>
-				<groupId>org.apache.maven.plugins</groupId>
-				<artifactId>maven-compiler-plugin</artifactId>
-				<version>3.6.1</version>
-				<configuration>
-					<source>1.8</source>
-					<target>1.8</target>
-				</configuration>
-			</plugin>
-		</plugins>
-	</build>
-
-	<dependencies>
-		<dependency>
-			<groupId>junit</groupId>
-			<artifactId>junit</artifactId>
-			<version>4.12</version>
-			<scope>test</scope>
-		</dependency>
-		<dependency>
-			<groupId>org.hamcrest</groupId>
-			<artifactId>hamcrest-junit</artifactId>
-			<version>2.0.0.0</version>
-		</dependency>
-	</dependencies>
-</project>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+
+  <groupId>sy.andrew</groupId>
+  <artifactId>linkifier</artifactId>
+  <version>0.0.1-SNAPSHOT</version>
+  <packaging>jar</packaging>
+
+  <name>linkifier</name>
+  <url>http://maven.apache.org</url>
+
+  <properties>
+    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+  </properties>
+
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-compiler-plugin</artifactId>
+        <version>3.6.1</version>
+        <configuration>
+          <source>1.8</source>
+          <target>1.8</target>
+        </configuration>
+      </plugin>
+    </plugins>
+  </build>
+
+  <dependencies>
+    <dependency>
+      <groupId>com.google.guava</groupId>
+      <artifactId>guava</artifactId>
+      <version>21.0</version>
+    </dependency>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <version>4.12</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.hamcrest</groupId>
+      <artifactId>hamcrest-junit</artifactId>
+      <version>2.0.0.0</version>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+</project>
diff --git a/src/main/java/sy/andrew/linkifier/Linkifier.java b/src/main/java/sy/andrew/linkifier/Linkifier.java
index 59ac6d4..040962a 100644
--- a/src/main/java/sy/andrew/linkifier/Linkifier.java
+++ b/src/main/java/sy/andrew/linkifier/Linkifier.java
@@ -1,4 +1,4 @@
-package sy.andrew.linkifier;
+ package sy.andrew.linkifier;
 
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
diff --git a/src/main/java/sy/andrew/linkifier/SequentialOutputBuffer.java b/src/main/java/sy/andrew/linkifier/SequentialOutputBuffer.java
new file mode 100644
index 0000000..f235965
--- /dev/null
+++ b/src/main/java/sy/andrew/linkifier/SequentialOutputBuffer.java
@@ -0,0 +1,77 @@
+package sy.andrew.linkifier;
+
+import java.util.PriorityQueue;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import sy.andrew.linkifier.model.OrderedLinkifyObject;
+import sy.andrew.linkifier.model.OrderedLinkifyObjectComparator;
+
+/** 
+ * Holds OrderedLinkifyObjects until they are ready to be 
+ * moved to the finalOutputQueue for consumers to pickup. Each  
+ * OrderedLinkifyObject should be made available to consumers in strict 
+ * sequential order based on sequence number. This buffer maintains
+ * information about the sequence number that is next in line to be
+ * added to the finalOutputQueue.  
+ * Any OrderedLinkifyObject whose sequence number is greater will be 
+ * held in this buffer until their turn comes up, at which point 
+ * it is automatically moved over to the finalOutputQueue.
+ *  
+ * @author asy
+ *
+ */
+public class SequentialOutputBuffer {
+    private int sequenceNumberToWaitFor = 0;
+    private final ConcurrentLinkedQueue<OrderedLinkifyObject> finalOutputQueue; 
+    
+    //access to this non-synchronized queue should be protected by using "this" as monitor
+    private final PriorityQueue<OrderedLinkifyObject> holdingHeap = new PriorityQueue(200, new OrderedLinkifyObjectComparator());
+    
+    public static SequentialOutputBuffer create(int firstSequenceNumber, 
+            ConcurrentLinkedQueue<OrderedLinkifyObject> finalOutputQueue) {
+        //TODO: do we want to return singleton instead? 
+        return new SequentialOutputBuffer(firstSequenceNumber, finalOutputQueue);
+    }
+
+    private SequentialOutputBuffer(int firstSequenceNumber, 
+            ConcurrentLinkedQueue<OrderedLinkifyObject> finalOutputQueue) {
+        this.sequenceNumberToWaitFor = firstSequenceNumber;
+        this.finalOutputQueue = finalOutputQueue;
+    }
+    
+    public void put(OrderedLinkifyObject incomingOlo) {
+        synchronized(this) {
+            if (incomingOlo.getSequenceNumber() > sequenceNumberToWaitFor) {
+                holdingHeap.add(incomingOlo);
+            } else {
+                promote(incomingOlo);  
+                promoteAllQualifiedOlos();
+            }                     
+        }
+    }
+    
+    protected void promote(OrderedLinkifyObject olo) {
+        synchronized(this) {
+            finalOutputQueue.offer(olo);
+            sequenceNumberToWaitFor++;
+        }    
+    }
+    
+    protected void promoteAllQualifiedOlos() {
+        synchronized(this) {
+            while(qualifiedOloExists()) {
+                OrderedLinkifyObject minOlo = holdingHeap.poll();
+                promote(minOlo);
+            }
+        }
+    }
+    
+    protected boolean qualifiedOloExists() {
+        synchronized(this) {
+            OrderedLinkifyObject minOlo = holdingHeap.peek();
+            return (minOlo != null && minOlo.getSequenceNumber() <= sequenceNumberToWaitFor);            
+        }
+    }
+    
+}
diff --git a/src/main/java/sy/andrew/linkifier/model/OrderedLinkifyObject.java b/src/main/java/sy/andrew/linkifier/model/OrderedLinkifyObject.java
new file mode 100644
index 0000000..a8fe6b3
--- /dev/null
+++ b/src/main/java/sy/andrew/linkifier/model/OrderedLinkifyObject.java
@@ -0,0 +1,29 @@
+package sy.andrew.linkifier.model;
+
+public class OrderedLinkifyObject {
+    private final int sequenceNumber;
+    private final String originalString;
+    private String linkifiedString;
+    
+    public OrderedLinkifyObject(int sequenceNumber, String originalString) {
+        this.sequenceNumber = sequenceNumber;
+        this.originalString = originalString;
+    }
+
+    public String getLinkifiedString() {
+        return this.linkifiedString;
+    }
+
+    public void setLinkifiedString(String linkifiedString) {
+        this.linkifiedString = linkifiedString;
+    }
+
+    public int getSequenceNumber() {
+        return this.sequenceNumber;
+    }
+
+    public String getOriginalString() {
+        return this.originalString;
+    }
+    
+}
diff --git a/src/main/java/sy/andrew/linkifier/model/OrderedLinkifyObjectComparator.java b/src/main/java/sy/andrew/linkifier/model/OrderedLinkifyObjectComparator.java
new file mode 100644
index 0000000..e483174
--- /dev/null
+++ b/src/main/java/sy/andrew/linkifier/model/OrderedLinkifyObjectComparator.java
@@ -0,0 +1,15 @@
+package sy.andrew.linkifier.model;
+
+import java.util.Comparator;
+
+public class OrderedLinkifyObjectComparator implements Comparator<OrderedLinkifyObject> {
+    @Override
+    public int compare(OrderedLinkifyObject a, OrderedLinkifyObject b) {
+        if (a.getSequenceNumber() == b.getSequenceNumber()) {
+            return 0;
+        } else {
+            return a.getSequenceNumber() > b.getSequenceNumber() ? 1 : -1;
+        }   
+    }
+
+}
diff --git a/src/test/java/sy/andrew/linkifier/SequentialOutputBufferTest.java b/src/test/java/sy/andrew/linkifier/SequentialOutputBufferTest.java
new file mode 100644
index 0000000..f5580b2
--- /dev/null
+++ b/src/test/java/sy/andrew/linkifier/SequentialOutputBufferTest.java
@@ -0,0 +1,76 @@
+package sy.andrew.linkifier;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.junit.Test;
+
+import sy.andrew.linkifier.model.OrderedLinkifyObject;
+
+public class SequentialOutputBufferTest {
+
+    @Test
+    public void givenInputOutOfOrder_shouldProduceOutputInOrder() {
+        // ARRANGE
+        final ConcurrentLinkedQueue<OrderedLinkifyObject> outputQueue = new ConcurrentLinkedQueue<>();
+        SequentialOutputBuffer sut = SequentialOutputBuffer.create(1, outputQueue);
+
+        // ACT
+        // Out of order inputs
+        sut.put(new OrderedLinkifyObject(6, ""));
+        sut.put(new OrderedLinkifyObject(5, ""));
+        sut.put(new OrderedLinkifyObject(4, ""));
+        sut.put(new OrderedLinkifyObject(1, ""));
+        sut.put(new OrderedLinkifyObject(2, ""));
+        sut.put(new OrderedLinkifyObject(3, ""));
+        sut.put(new OrderedLinkifyObject(7, ""));
+        sut.put(new OrderedLinkifyObject(8, ""));
+
+        // ASSERT
+        assertEquals("Output should be in order.", outputQueue.poll().getSequenceNumber(), 1);
+        assertEquals("Output should be in order.", outputQueue.poll().getSequenceNumber(), 2);
+        assertEquals("Output should be in order.", outputQueue.poll().getSequenceNumber(), 3);
+        assertEquals("Output should be in order.", outputQueue.poll().getSequenceNumber(), 4);
+        assertEquals("Output should be in order.", outputQueue.poll().getSequenceNumber(), 5);
+        assertEquals("Output should be in order.", outputQueue.poll().getSequenceNumber(), 6);
+        assertEquals("Output should be in order.", outputQueue.poll().getSequenceNumber(), 7);
+        assertEquals("Output should be in order.", outputQueue.poll().getSequenceNumber(), 8);
+        assertTrue("outputQueue should be empty.", outputQueue.isEmpty());
+    }
+
+    @Test
+    public void givenInputWithSkippedSequenceNumber_thenOutputShouldExcludeSequenceNumbersGreaterThanOrEqual() {
+        // ARRANGE
+        final ConcurrentLinkedQueue<OrderedLinkifyObject> outputQueue = new ConcurrentLinkedQueue<>();
+        SequentialOutputBuffer sut = SequentialOutputBuffer.create(1, outputQueue);
+
+        // ACT
+        // Out of order inputs
+        sut.put(new OrderedLinkifyObject(1, ""));
+        sut.put(new OrderedLinkifyObject(2, ""));
+        sut.put(new OrderedLinkifyObject(3, ""));
+        //skip 4
+        sut.put(new OrderedLinkifyObject(5, "")); 
+
+        // ASSERT
+        assertEquals("Output should be in order.", outputQueue.poll().getSequenceNumber(), 1);
+        assertEquals("Output should be in order.", outputQueue.poll().getSequenceNumber(), 2);
+        assertEquals("Output should be in order.", outputQueue.poll().getSequenceNumber(), 3);
+        assertTrue("outputQueue should be empty.", outputQueue.isEmpty());        
+    }
+    
+    @Test
+    public void givenNoInput_thenOutputShouldBeEmpty() {
+        //ARRANGE 
+        final ConcurrentLinkedQueue<OrderedLinkifyObject> outputQueue = 
+                new ConcurrentLinkedQueue<>();
+        SequentialOutputBuffer sut = SequentialOutputBuffer.create(1, outputQueue);
+
+        //ACT - no inputs
+
+        //ASSERT
+        assertTrue("outputQueue should be empty.", outputQueue.isEmpty());        
+    }
+}
\ No newline at end of file
diff --git a/src/test/java/sy/andrew/linkifier/model/OrderedLinkifyObjectComparatorTest.java b/src/test/java/sy/andrew/linkifier/model/OrderedLinkifyObjectComparatorTest.java
new file mode 100644
index 0000000..bf08ef4
--- /dev/null
+++ b/src/test/java/sy/andrew/linkifier/model/OrderedLinkifyObjectComparatorTest.java
@@ -0,0 +1,50 @@
+package sy.andrew.linkifier.model;
+
+import static org.junit.Assert.assertEquals;
+import org.junit.Test;
+
+import sy.andrew.linkifier.model.OrderedLinkifyObject;
+import sy.andrew.linkifier.model.OrderedLinkifyObjectComparator;
+
+public class OrderedLinkifyObjectComparatorTest {
+
+    @Test
+    public void ifAEqualsB_thenReturn0() {
+        //ARRANGE
+        final OrderedLinkifyObject a = new OrderedLinkifyObject(1, "a");
+        final OrderedLinkifyObject b = new OrderedLinkifyObject(1, "b");
+        
+        //ACT
+        int actual = new OrderedLinkifyObjectComparator().compare(a, b);
+        
+        //ASSERT
+        assertEquals("Given a = b, ", actual, 0);
+    }
+
+    @Test
+    public void ifAGreaterB_thenReturn1() {
+        //ARRANGE
+        final OrderedLinkifyObject a = new OrderedLinkifyObject(2, "a");
+        final OrderedLinkifyObject b = new OrderedLinkifyObject(1, "b");
+        
+        //ACT
+        int actual = new OrderedLinkifyObjectComparator().compare(a, b);
+        
+        //ASSERT
+        assertEquals("Given a = b, ", actual, 1);
+    }
+
+    @Test
+    public void ifALessThanB_thenReturnMinus1() {
+        //ARRANGE
+        final OrderedLinkifyObject a = new OrderedLinkifyObject(1, "a");
+        final OrderedLinkifyObject b = new OrderedLinkifyObject(2, "b");
+        
+        //ACT
+        int actual = new OrderedLinkifyObjectComparator().compare(a, b);
+        
+        //ASSERT
+        assertEquals("Given a = b, ", actual, -1);
+    }
+}
+

From e8ee64f59b374a2b2a6cc1a8a457a8c36a966053 Mon Sep 17 00:00:00 2001
From: asy <asy>
Date: Fri, 17 Feb 2017 02:32:25 -0500
Subject: [PATCH 2/2] completely functional first-cut with an internal
 SequentialOutputBuffer

---
 DesignNotes.txt                               |   1 -
 .../linkifier/LinkifyProcessingQueue.java     |  29 ----
 .../sy/andrew/linkifier/MultithreadDemo.java  | 103 -------------
 .../sy/andrew/linkifier/ProcessingQueue.java  |   6 +-
 .../sy/andrew/linkifier/SingleThreadDemo.java |  27 ----
 .../linkifier/demo/MultithreadDemo.java       | 141 ++++++++++++++++++
 .../linkifier/demo/SingleThreadDemo.java      |  62 ++++++++
 .../linkifier/model/OrderedLinkifyObject.java |  27 ++++
 .../processors/LinkifyProcessingQueue.java    |  66 ++++++++
 .../linkifier/processors/LinkifyTask.java     |  34 +++++
 .../SequentialOutputBuffer.java               |  10 +-
 .../andrew/linkifier/util/OLOLinkifier.java   |  13 ++
 .../StringLinkifier.java}                     |   4 +-
 .../linkifier/LinkifyProcessingQueueTest.java |  58 -------
 .../LinkifyProcessingQueueTest.java           |  86 +++++++++++
 .../SequentialOutputBufferTest.java           |  19 +--
 .../linkifier/util/OLOLinkifierTest.java      |  33 ++++
 .../StringLinkifierTest.java}                 |  44 +++---
 18 files changed, 499 insertions(+), 264 deletions(-)
 delete mode 100644 src/main/java/sy/andrew/linkifier/LinkifyProcessingQueue.java
 delete mode 100644 src/main/java/sy/andrew/linkifier/MultithreadDemo.java
 delete mode 100644 src/main/java/sy/andrew/linkifier/SingleThreadDemo.java
 create mode 100644 src/main/java/sy/andrew/linkifier/demo/MultithreadDemo.java
 create mode 100644 src/main/java/sy/andrew/linkifier/demo/SingleThreadDemo.java
 create mode 100644 src/main/java/sy/andrew/linkifier/processors/LinkifyProcessingQueue.java
 create mode 100644 src/main/java/sy/andrew/linkifier/processors/LinkifyTask.java
 rename src/main/java/sy/andrew/linkifier/{ => processors}/SequentialOutputBuffer.java (86%)
 create mode 100644 src/main/java/sy/andrew/linkifier/util/OLOLinkifier.java
 rename src/main/java/sy/andrew/linkifier/{Linkifier.java => util/StringLinkifier.java} (94%)
 delete mode 100644 src/test/java/sy/andrew/linkifier/LinkifyProcessingQueueTest.java
 create mode 100644 src/test/java/sy/andrew/linkifier/processors/LinkifyProcessingQueueTest.java
 rename src/test/java/sy/andrew/linkifier/{ => processors}/SequentialOutputBufferTest.java (78%)
 create mode 100644 src/test/java/sy/andrew/linkifier/util/OLOLinkifierTest.java
 rename src/test/java/sy/andrew/linkifier/{LinkifierTest.java => util/StringLinkifierTest.java} (78%)

diff --git a/DesignNotes.txt b/DesignNotes.txt
index 8252a13..d3279ff 100644
--- a/DesignNotes.txt
+++ b/DesignNotes.txt
@@ -1,7 +1,6 @@
 Main Code Features
 - uses guava: great library (collections, etc.), cleaner code
 - fluent api?
-- TreeRangeSet queries is O(log n + k) time and O(n) space, where n is number of interval points and k is number of intervals returned ?
 - RangeSetUtil 
   - use of generics (template method)
   - 
diff --git a/src/main/java/sy/andrew/linkifier/LinkifyProcessingQueue.java b/src/main/java/sy/andrew/linkifier/LinkifyProcessingQueue.java
deleted file mode 100644
index 5b16c3c..0000000
--- a/src/main/java/sy/andrew/linkifier/LinkifyProcessingQueue.java
+++ /dev/null
@@ -1,29 +0,0 @@
-package sy.andrew.linkifier;
-
-import java.util.concurrent.ConcurrentLinkedQueue;
-
-public class LinkifyProcessingQueue implements ProcessingQueue {
-
-    private final ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<>();
-    private final Linkifier linky = new Linkifier();
-
-    @Override
-    public boolean offer(final String input) {
-        final String linkifiedResult = linky.linkify(input);
-        return queue.offer(linkifiedResult);
-    }
-
-    @Override
-    public String poll() {
-        return queue.poll();
-    }
-
-    protected boolean contains(final String element) {
-        return queue.contains(element);
-    }
-
-    protected int size() {
-        return queue.size();
-    }
-
-}
diff --git a/src/main/java/sy/andrew/linkifier/MultithreadDemo.java b/src/main/java/sy/andrew/linkifier/MultithreadDemo.java
deleted file mode 100644
index 1b6311c..0000000
--- a/src/main/java/sy/andrew/linkifier/MultithreadDemo.java
+++ /dev/null
@@ -1,103 +0,0 @@
-package sy.andrew.linkifier;
-
-import java.util.Arrays;
-import java.util.List;
-
-public class MultithreadDemo {
-
-    private static class Producer implements Runnable {
-        private final String name;
-        private final LinkifyProcessingQueue queue;
-        private final List<String> inputTextList;
-
-        private Producer(final String name, final LinkifyProcessingQueue queue,
-                final List<String> inputTextList) {
-            this.name = name;
-            this.queue = queue;
-            this.inputTextList = inputTextList;
-        }
-
-        @Override
-        public void run() {
-            int successfulOffersCount = 0;
-            for (final String inputText : inputTextList) {
-                if (queue.offer(inputText)) {
-                    successfulOffersCount++;
-                    System.out.println();
-                    System.out.println(name + ": successfulOffersCount=" + successfulOffersCount
-                            + ", inputText=[" + inputText + "]");
-                } else {
-                    System.out.println();
-                    System.out.println(name + ": offer failed. inputText=[" + inputText + "]");
-                }
-                try {
-                    Thread.sleep((long) Math.random() * 5000);
-                } catch (final InterruptedException e) {
-                    // Just ignore it.
-                }
-            }
-            System.out.println();
-            System.out.println(name + ": done!");
-        }
-    }
-
-    private static class Consumer implements Runnable {
-        private final String name;
-        private static final int MAX_POLL_ATTEMPTS = 30;
-        private final LinkifyProcessingQueue queue;
-
-        private Consumer(final String name, final LinkifyProcessingQueue queue) {
-            this.name = name;
-            this.queue = queue;
-        }
-
-        @Override
-        public void run() {
-            int successfulPollsCount = 0;
-            for (int pollAttempts = 0; pollAttempts <= MAX_POLL_ATTEMPTS; pollAttempts++) {
-                final String polledText = queue.poll();
-                if (polledText != null) {
-                    successfulPollsCount++;
-                    System.out.println();
-                    System.out.println(name + ": successfulPollsCount=" + successfulPollsCount
-                            + ", polledText=[" + polledText + "]");
-                } else {
-                    System.out.println();
-                    System.out.println(name + ": polledText is null. ");
-                }
-
-                try {
-                    Thread.sleep((long) Math.random() * 50000);
-                } catch (final InterruptedException e) {
-                    // Just ignore it.
-                }
-            }
-            System.out.println();
-            System.out.println(name + ": done!");
-        }
-    }
-
-    /**
-     * @param args
-     */
-    public static void main(final String[] args) {
-        final String[] arrInputText = { "<html><body>http://foo</body></html>",
-                "<name>Andy</name>", " http://example.com/query?name=Bill", "hi", "ho", "boo",
-                "dog", "cat" };
-        final LinkifyProcessingQueue queue = new LinkifyProcessingQueue();
-
-        final Consumer consumer1 = new Consumer("Consumer1", queue);
-        new Thread(consumer1).start();
-
-        final Producer producer1 = new Producer("Producer1", queue, Arrays.asList(arrInputText));
-        new Thread(producer1).start();
-
-        final Producer producer2 = new Producer("Producer2", queue, Arrays.asList(arrInputText));
-        new Thread(producer2).start();
-
-        final Consumer consumer2 = new Consumer("Consumer2", queue);
-        new Thread(consumer2).start();
-
-    }
-
-}
diff --git a/src/main/java/sy/andrew/linkifier/ProcessingQueue.java b/src/main/java/sy/andrew/linkifier/ProcessingQueue.java
index becdb0d..736dba1 100644
--- a/src/main/java/sy/andrew/linkifier/ProcessingQueue.java
+++ b/src/main/java/sy/andrew/linkifier/ProcessingQueue.java
@@ -3,7 +3,7 @@
  * Interface for a FIFO queue which performs some processing on its inputs
  * before making them available to consumers.
  */
-public interface ProcessingQueue
+public interface ProcessingQueue<T>
 {
     /**
      * Inserts the specified element into this queue, if possible.
@@ -11,12 +11,12 @@ public interface ProcessingQueue
      * @param input the element to insert.
      * @return true if it was possible to add the element to this queue, else false
      */
-    boolean offer(String input);
+    boolean offer(T input);
 
     /**
      * Retrieves and removes the head of this queue, or null if this queue is empty.
      * 
      * @return the head of this queue, or null if this queue is empty.
      */
-    String poll();
+    T poll();
 }
diff --git a/src/main/java/sy/andrew/linkifier/SingleThreadDemo.java b/src/main/java/sy/andrew/linkifier/SingleThreadDemo.java
deleted file mode 100644
index abb557b..0000000
--- a/src/main/java/sy/andrew/linkifier/SingleThreadDemo.java
+++ /dev/null
@@ -1,27 +0,0 @@
-package sy.andrew.linkifier;
-
-public class SingleThreadDemo {
-
-    /**
-     * @param args
-     */
-    public static void main(final String[] args) {
-        final String[] arrInputText = {
-                "<html><body><p>http://linkify</p><img src=\"http://do-not-linkify\"/></body></html>",
-                "<name>Andy</name><age>21</age>",
-                " http://example.com/query?name=Bill http://linkify-me-too" };
-        final LinkifyProcessingQueue queue = new LinkifyProcessingQueue();
-        for (final String inputText : arrInputText) {
-            if (!queue.offer(inputText)) {
-                System.out.println("queue.offer() failed for inputText=[" + inputText + "]");
-            } else {
-                System.out.println();
-                System.out.println("inputText=[" + inputText + "]");
-                final String textPolledFromLinkifyQueue = queue.poll();
-                System.out.println("textPolledFromLinkifyQueue=[" + textPolledFromLinkifyQueue
-                        + "]");
-            }
-        }
-
-    }
-}
diff --git a/src/main/java/sy/andrew/linkifier/demo/MultithreadDemo.java b/src/main/java/sy/andrew/linkifier/demo/MultithreadDemo.java
new file mode 100644
index 0000000..8d227bd
--- /dev/null
+++ b/src/main/java/sy/andrew/linkifier/demo/MultithreadDemo.java
@@ -0,0 +1,141 @@
+package sy.andrew.linkifier.demo;
+
+import java.util.Arrays;
+import java.util.List;
+
+import sy.andrew.linkifier.model.OrderedLinkifyObject;
+import sy.andrew.linkifier.processors.LinkifyProcessingQueue;
+
+public class MultithreadDemo {
+    /**
+     * @param args
+     */
+    public static void main(final String[] args) {
+        final List<OrderedLinkifyObject> inputOlos_0 = Arrays.asList(
+                new OrderedLinkifyObject(0, "<html><body>http://foo</body></html>"),
+                new OrderedLinkifyObject(2, "<name>Andy</name><age>21</age>"),
+                new OrderedLinkifyObject(4, " http://example.com/query?name=Bill http://linkify-me-too"),
+                new OrderedLinkifyObject(6, "hi"),
+                new OrderedLinkifyObject(8, "ho"),
+                new OrderedLinkifyObject(10, "boo"),
+                new OrderedLinkifyObject(12, "dog"),
+                new OrderedLinkifyObject(14, "cat")
+                );
+
+        final List<OrderedLinkifyObject> inputOlos_1 = Arrays.asList(
+                new OrderedLinkifyObject(1, "<html><body>http://foo</body></html>"),
+                new OrderedLinkifyObject(3, "<name>Andy</name><age>21</age>"),
+                new OrderedLinkifyObject(5, " http://example.com/query?name=Bill http://linkify-me-too"),
+                new OrderedLinkifyObject(7, "hi"),
+                new OrderedLinkifyObject(9, "ho"),
+                new OrderedLinkifyObject(11, "boo"),
+                new OrderedLinkifyObject(13, "dog"),
+                new OrderedLinkifyObject(15, "cat")
+                );
+
+        try {
+            final LinkifyProcessingQueue queue = createLinkifyProcessingQueue();
+
+            final Consumer consumer1 = new Consumer("Consumer0", queue);
+            new Thread(consumer1).start();
+
+            final Producer producer1 = new Producer("Producer0", queue, inputOlos_0);
+            new Thread(producer1).start();
+
+            final Producer producer2 = new Producer("Producer1", queue, inputOlos_1);
+            new Thread(producer2).start();
+
+            final Consumer consumer2 = new Consumer("Consumer1", queue);
+            new Thread(consumer2).start();
+
+            Thread.sleep(5000);
+            queue.shutdown();
+            
+        } catch(Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    private static LinkifyProcessingQueue createLinkifyProcessingQueue() {
+        int firstSequenceNumber = 0;
+        int capacity = 100;
+        int executorsThreadPoolSize = 4;
+        return new LinkifyProcessingQueue(firstSequenceNumber, capacity, 
+                executorsThreadPoolSize);
+   
+    }
+    
+
+    private static class Producer implements Runnable {
+        private final String name;
+        private final LinkifyProcessingQueue queue;
+        private final List<OrderedLinkifyObject> inputOloList;
+
+        private Producer(final String name, final LinkifyProcessingQueue queue,
+                final List<OrderedLinkifyObject> inputOloList) {
+            this.name = name;
+            this.queue = queue;
+            this.inputOloList = inputOloList;
+        }
+
+        @Override
+        public void run() {
+            int successfulOffersCount = 0;
+            for (final OrderedLinkifyObject inputOlo : inputOloList) {
+                if (queue.offer(inputOlo)) {
+                    successfulOffersCount++;
+                    System.out.println();
+                    System.out.println(name + ": successfulOffersCount=" + successfulOffersCount
+                            + ", inputOlo=[" + inputOlo + "]");
+                } else {
+                    System.out.println();
+                    System.out.println(name + ": offer failed. inputOlo=[" + inputOlo + "]");
+                }
+                try {
+                    Thread.sleep((long) Math.random() * 5000);
+                } catch (final InterruptedException e) {
+                    // Just ignore it.
+                }
+            }
+            System.out.println();
+            System.out.println(name + ": done!");
+        }
+    }
+
+    private static class Consumer implements Runnable {
+        private final String name;
+        private static final int MAX_POLL_ATTEMPTS = 30;
+        private final LinkifyProcessingQueue queue;
+
+        private Consumer(final String name, final LinkifyProcessingQueue queue) {
+            this.name = name;
+            this.queue = queue;
+        }
+
+        @Override
+        public void run() {
+            int successfulPollsCount = 0;
+            for (int pollAttempts = 0; pollAttempts <= MAX_POLL_ATTEMPTS; pollAttempts++) {
+                final OrderedLinkifyObject polledOlo = queue.poll();
+                if (polledOlo != null) {
+                    successfulPollsCount++;
+                    System.out.println();
+                    System.out.println(name + ": successfulPollsCount=" + successfulPollsCount
+                            + ", polledOlo=[" + polledOlo + "]");
+                } else {
+                    System.out.println();
+                    System.out.println(name + ": polledOlo is null. ");
+                }
+
+                try {
+                    Thread.sleep((long) Math.random() * 50000);
+                } catch (final InterruptedException e) {
+                    // Just ignore it.
+                }
+            }
+            System.out.println();
+            System.out.println(name + ": done!");
+        }
+    }
+
+}
diff --git a/src/main/java/sy/andrew/linkifier/demo/SingleThreadDemo.java b/src/main/java/sy/andrew/linkifier/demo/SingleThreadDemo.java
new file mode 100644
index 0000000..096c2b3
--- /dev/null
+++ b/src/main/java/sy/andrew/linkifier/demo/SingleThreadDemo.java
@@ -0,0 +1,62 @@
+package sy.andrew.linkifier.demo;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import sy.andrew.linkifier.model.OrderedLinkifyObject;
+import sy.andrew.linkifier.processors.LinkifyProcessingQueue;
+
+public class SingleThreadDemo {
+
+    /**
+     * @param args
+     */
+    public static void main(final String[] args) {
+        final LinkifyProcessingQueue queue = createLinkifyProcessingQueue();
+        
+        final List<OrderedLinkifyObject> inputOlos = Arrays.asList(
+                new OrderedLinkifyObject(0, "<html><body><p>http://linkify</p><img src=\"http://do-not-linkify\"/></body></html>"),
+                new OrderedLinkifyObject(1, "<name>Andy</name><age>21</age>"),
+                new OrderedLinkifyObject(2, " http://example.com/query?name=Bill http://linkify-me-too")
+                );
+
+        for (final OrderedLinkifyObject inputOlo : inputOlos) {
+            if (!queue.offer(inputOlo)) {
+                System.out.println("queue.offer() failed for olo=[" + inputOlo + "]");
+            } else {
+                System.out.println();
+                System.out.println("input=[" + inputOlo + "]");
+                OrderedLinkifyObject output = pollTillMaxTry(queue, 500);
+                System.out.println("output=[" + output
+                        + "]");
+            }
+        }
+
+    }
+    
+    private static LinkifyProcessingQueue createLinkifyProcessingQueue() {
+        int firstSequenceNumber = 0;
+        int capacity = 100;
+        int executorsThreadPoolSize = 4;
+        return new LinkifyProcessingQueue(firstSequenceNumber, capacity, 
+                executorsThreadPoolSize);
+   
+    }
+    
+    private static OrderedLinkifyObject pollTillMaxTry(final LinkifyProcessingQueue linkifierQueue, 
+            final int MAX_TRY) {
+        
+        OrderedLinkifyObject output = null;
+        for(int attempts = 0; 
+                output == null && attempts < MAX_TRY; 
+                attempts++) {
+            
+            output = linkifierQueue.poll();
+        }
+        
+        return output;
+    }
+
+    
+}
diff --git a/src/main/java/sy/andrew/linkifier/model/OrderedLinkifyObject.java b/src/main/java/sy/andrew/linkifier/model/OrderedLinkifyObject.java
index a8fe6b3..d9f98dd 100644
--- a/src/main/java/sy/andrew/linkifier/model/OrderedLinkifyObject.java
+++ b/src/main/java/sy/andrew/linkifier/model/OrderedLinkifyObject.java
@@ -1,5 +1,7 @@
 package sy.andrew.linkifier.model;
 
+import java.util.Objects;
+
 public class OrderedLinkifyObject {
     private final int sequenceNumber;
     private final String originalString;
@@ -25,5 +27,30 @@ public int getSequenceNumber() {
     public String getOriginalString() {
         return this.originalString;
     }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(this.sequenceNumber, this.originalString, this.linkifiedString);
+    }
+
+    @Override
+    public boolean equals(Object other) {
+        if (other == null) { return false; }
+        if (this.getClass() != other.getClass()) { return false; }
+        
+        final OrderedLinkifyObject that = (OrderedLinkifyObject) other;
+        return Objects.equals(this.sequenceNumber, that.sequenceNumber) 
+                && Objects.equals(this.originalString, that.originalString)
+                && Objects.equals(this.linkifiedString, that.linkifiedString)
+                ;
+    }
+
+    @Override
+    public String toString() {
+        return "OrderedLinkifyObject [sequenceNumber=" + this.sequenceNumber + ", originalString=" + this.originalString
+                + ", linkifiedString=" + this.linkifiedString + "]";
+    }
+    
+    
     
 }
diff --git a/src/main/java/sy/andrew/linkifier/processors/LinkifyProcessingQueue.java b/src/main/java/sy/andrew/linkifier/processors/LinkifyProcessingQueue.java
new file mode 100644
index 0000000..5fafa29
--- /dev/null
+++ b/src/main/java/sy/andrew/linkifier/processors/LinkifyProcessingQueue.java
@@ -0,0 +1,66 @@
+package sy.andrew.linkifier.processors;
+
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import sy.andrew.linkifier.ProcessingQueue;
+import sy.andrew.linkifier.model.OrderedLinkifyObject;
+
+public class LinkifyProcessingQueue implements ProcessingQueue<OrderedLinkifyObject> {
+
+    private final ArrayBlockingQueue<OrderedLinkifyObject> inputQueue;  
+    private final ArrayBlockingQueue<OrderedLinkifyObject> outputQueue;
+    private final SequentialOutputBuffer sob;
+    private final ExecutorService executor;
+    
+    public LinkifyProcessingQueue(int firstSequenceNumber, 
+            int capacity, int executorsThreadPoolSize) {
+        
+        if (capacity <= 0) { 
+            throw new IllegalArgumentException("queueCapacity must be positive. queueCapacity=" + capacity); 
+        }
+        this.inputQueue = new ArrayBlockingQueue<>(capacity);        
+        this.outputQueue = new ArrayBlockingQueue<>(capacity);
+        
+        this.sob = SequentialOutputBuffer.create(firstSequenceNumber, outputQueue);
+
+        if (executorsThreadPoolSize <= 0) { 
+            throw new IllegalArgumentException("executorsThreadPoolSize must be positive. executorsThreadPoolSize=" + executorsThreadPoolSize); 
+        }
+        this.executor = Executors.newFixedThreadPool(executorsThreadPoolSize);
+        
+        executor.submit(new LinkifyTask(inputQueue, sob));
+    }
+
+    @Override
+    public boolean offer(OrderedLinkifyObject input) {
+        return inputQueue.offer(input);
+    }
+
+    @Override
+    public OrderedLinkifyObject poll() {
+        return outputQueue.poll();
+    }
+    
+    public void shutdown() {
+        try {
+            System.out.println("attempt to shutdown executor");
+            executor.shutdown();
+            executor.awaitTermination(5, TimeUnit.SECONDS);
+        }
+        catch (InterruptedException e) {
+            System.err.println("tasks interrupted");
+        }
+        finally {
+            if (!executor.isTerminated()) {
+                System.err.println("cancel non-finished tasks");
+            }
+            executor.shutdownNow();
+            System.out.println("shutdown finished");
+        }        
+    }
+
+}
diff --git a/src/main/java/sy/andrew/linkifier/processors/LinkifyTask.java b/src/main/java/sy/andrew/linkifier/processors/LinkifyTask.java
new file mode 100644
index 0000000..1bd3a15
--- /dev/null
+++ b/src/main/java/sy/andrew/linkifier/processors/LinkifyTask.java
@@ -0,0 +1,34 @@
+package sy.andrew.linkifier.processors;
+
+import java.util.concurrent.ArrayBlockingQueue;
+
+import sy.andrew.linkifier.model.OrderedLinkifyObject;
+import sy.andrew.linkifier.util.OLOLinkifier;
+
+public class LinkifyTask implements Runnable {
+
+    private final ArrayBlockingQueue<OrderedLinkifyObject> inputQueue;  
+    private final SequentialOutputBuffer sob;
+    private final OLOLinkifier linky = new OLOLinkifier();
+
+    public LinkifyTask(ArrayBlockingQueue<OrderedLinkifyObject> inputQueue, 
+            SequentialOutputBuffer sob) {
+        this.inputQueue = inputQueue;
+        this.sob = sob;   
+    }
+    
+    @Override
+    public void run() {
+        try {
+            while(true) {
+                OrderedLinkifyObject olo = inputQueue.take();
+                linky.linkify(olo);
+                sob.put(olo);
+            }
+        } catch (InterruptedException e) {
+            //just ignore it. 
+        }
+    }
+    
+
+}
diff --git a/src/main/java/sy/andrew/linkifier/SequentialOutputBuffer.java b/src/main/java/sy/andrew/linkifier/processors/SequentialOutputBuffer.java
similarity index 86%
rename from src/main/java/sy/andrew/linkifier/SequentialOutputBuffer.java
rename to src/main/java/sy/andrew/linkifier/processors/SequentialOutputBuffer.java
index f235965..6e3e615 100644
--- a/src/main/java/sy/andrew/linkifier/SequentialOutputBuffer.java
+++ b/src/main/java/sy/andrew/linkifier/processors/SequentialOutputBuffer.java
@@ -1,7 +1,7 @@
-package sy.andrew.linkifier;
+package sy.andrew.linkifier.processors;
 
 import java.util.PriorityQueue;
-import java.util.Queue;
+import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.ConcurrentLinkedQueue;
 
 import sy.andrew.linkifier.model.OrderedLinkifyObject;
@@ -23,19 +23,19 @@
  */
 public class SequentialOutputBuffer {
     private int sequenceNumberToWaitFor = 0;
-    private final ConcurrentLinkedQueue<OrderedLinkifyObject> finalOutputQueue; 
+    private final ArrayBlockingQueue<OrderedLinkifyObject> finalOutputQueue; 
     
     //access to this non-synchronized queue should be protected by using "this" as monitor
     private final PriorityQueue<OrderedLinkifyObject> holdingHeap = new PriorityQueue(200, new OrderedLinkifyObjectComparator());
     
     public static SequentialOutputBuffer create(int firstSequenceNumber, 
-            ConcurrentLinkedQueue<OrderedLinkifyObject> finalOutputQueue) {
+            ArrayBlockingQueue<OrderedLinkifyObject> finalOutputQueue) {
         //TODO: do we want to return singleton instead? 
         return new SequentialOutputBuffer(firstSequenceNumber, finalOutputQueue);
     }
 
     private SequentialOutputBuffer(int firstSequenceNumber, 
-            ConcurrentLinkedQueue<OrderedLinkifyObject> finalOutputQueue) {
+            ArrayBlockingQueue<OrderedLinkifyObject> finalOutputQueue) {
         this.sequenceNumberToWaitFor = firstSequenceNumber;
         this.finalOutputQueue = finalOutputQueue;
     }
diff --git a/src/main/java/sy/andrew/linkifier/util/OLOLinkifier.java b/src/main/java/sy/andrew/linkifier/util/OLOLinkifier.java
new file mode 100644
index 0000000..a8dde61
--- /dev/null
+++ b/src/main/java/sy/andrew/linkifier/util/OLOLinkifier.java
@@ -0,0 +1,13 @@
+package sy.andrew.linkifier.util;
+
+import sy.andrew.linkifier.model.OrderedLinkifyObject;
+
+public class OLOLinkifier {
+    private final StringLinkifier linky = new StringLinkifier();
+  
+    public OrderedLinkifyObject linkify(OrderedLinkifyObject olo) {
+        String linkifiedString = linky.linkify(olo.getOriginalString());
+        olo.setLinkifiedString(linkifiedString);
+        return olo;
+    }
+}
diff --git a/src/main/java/sy/andrew/linkifier/Linkifier.java b/src/main/java/sy/andrew/linkifier/util/StringLinkifier.java
similarity index 94%
rename from src/main/java/sy/andrew/linkifier/Linkifier.java
rename to src/main/java/sy/andrew/linkifier/util/StringLinkifier.java
index 040962a..2a7d80e 100644
--- a/src/main/java/sy/andrew/linkifier/Linkifier.java
+++ b/src/main/java/sy/andrew/linkifier/util/StringLinkifier.java
@@ -1,4 +1,4 @@
- package sy.andrew.linkifier;
+ package sy.andrew.linkifier.util;
 
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -11,7 +11,7 @@
  * @author asy
  * 
  */
-public class Linkifier {
+public class StringLinkifier {
 
     /**
      * Matches input string till first occurrence of one of the following is
diff --git a/src/test/java/sy/andrew/linkifier/LinkifyProcessingQueueTest.java b/src/test/java/sy/andrew/linkifier/LinkifyProcessingQueueTest.java
deleted file mode 100644
index e8f8d37..0000000
--- a/src/test/java/sy/andrew/linkifier/LinkifyProcessingQueueTest.java
+++ /dev/null
@@ -1,58 +0,0 @@
-package sy.andrew.linkifier;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-
-import java.util.ArrayList;
-import java.util.List;
-
-import org.junit.Before;
-import org.junit.Test;
-
-public class LinkifyProcessingQueueTest {
-    private Linkifier linky;
-    private LinkifyProcessingQueue queue;
-    private final String[] arrInputText = { "<html><body>http://foo</body></html>",
-            "<name>Andy</name>" };
-
-    @Before
-    public void setUp() {
-        linky = new Linkifier();
-        queue = new LinkifyProcessingQueue();
-    }
-
-    @Test
-    public final void testOffer() {
-        int expectedSizeOfQueue = 0;
-        for (final String inputText : arrInputText) {
-            final String linkifiedText = linky.linkify(inputText);
-            if (queue.offer(inputText)) {
-                expectedSizeOfQueue++;
-                assertTrue("queue should contain linkifiedText=[" + linkifiedText + "]",
-                        queue.contains(linkifiedText));
-            }
-        }
-        assertEquals("size of queue", expectedSizeOfQueue, queue.size());
-    }
-
-    @Test
-    public final void testFifo() {
-        final List<String> linkifiedTextList = new ArrayList<>(arrInputText.length);
-        for (final String inputText : arrInputText) {
-            if (queue.offer(inputText)) {
-                linkifiedTextList.add(linky.linkify(inputText));
-            }
-        }
-
-        for (final String expectedLinkifiedText : linkifiedTextList) {
-            final String textPolledFromQueue = queue.poll();
-            assertEquals("expectedLinkifiedText should match textPolledFromQueue.",
-                    expectedLinkifiedText, textPolledFromQueue);
-        }
-    }
-
-    @Test
-    public final void testEmptyQueuePoll() {
-        assertEquals("polling empty queue should return null.", null, queue.poll());
-    }
-}
diff --git a/src/test/java/sy/andrew/linkifier/processors/LinkifyProcessingQueueTest.java b/src/test/java/sy/andrew/linkifier/processors/LinkifyProcessingQueueTest.java
new file mode 100644
index 0000000..34b96e9
--- /dev/null
+++ b/src/test/java/sy/andrew/linkifier/processors/LinkifyProcessingQueueTest.java
@@ -0,0 +1,86 @@
+package sy.andrew.linkifier.processors;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import sy.andrew.linkifier.model.OrderedLinkifyObject;
+import sy.andrew.linkifier.processors.LinkifyProcessingQueue;
+import sy.andrew.linkifier.util.OLOLinkifier;
+
+public class LinkifyProcessingQueueTest {
+    private final OLOLinkifier linky = new OLOLinkifier();
+    private LinkifyProcessingQueue linkifierQueue;
+    private final List<OrderedLinkifyObject> orderedInput = Arrays.asList(
+            new OrderedLinkifyObject(0, "<html><body>http://foo</body></html>"),
+            new OrderedLinkifyObject(1, "<name>Andy</name>"),
+            new OrderedLinkifyObject(2, "<name>Mom</name>"),
+            new OrderedLinkifyObject(3, "<name>Pop</name>"),
+            new OrderedLinkifyObject(4, "<name>Baby</name>"),
+            new OrderedLinkifyObject(5, "<name>Jane</name>"),
+            new OrderedLinkifyObject(6, "<name>Mom</name>"),
+            new OrderedLinkifyObject(7, "<name>Mom</name>"),
+            new OrderedLinkifyObject(8, "<name>Mom</name>"),
+            new OrderedLinkifyObject(9, "<name>Mom</name>"),
+            new OrderedLinkifyObject(10, "<name>Mom</name>"),
+            new OrderedLinkifyObject(11, "<name>Mom</name>"),
+            new OrderedLinkifyObject(12, "<name>Mom</name>"),
+            new OrderedLinkifyObject(13, "<name>Mom</name>"),
+            new OrderedLinkifyObject(14, "<name>Mom</name>"),
+            new OrderedLinkifyObject(15, "<name>Mom</name>"),
+            new OrderedLinkifyObject(16, "<name>Mom</name>"),
+            new OrderedLinkifyObject(17, "<name>Mom</name>"),
+            new OrderedLinkifyObject(18, "<name>Mom</name>"),
+            new OrderedLinkifyObject(19, "<name>Mom</name>")
+            );
+
+    @Before
+    public void setUp() {
+        int firstSequenceNumber = 0;
+        int inputCapacity = 100;
+        int executorsThreadPoolSize = 4;
+        linkifierQueue = new LinkifyProcessingQueue(firstSequenceNumber, inputCapacity, 
+                executorsThreadPoolSize);
+    }
+
+    @Test
+    public final void testOrderedOutput() {
+        List<OrderedLinkifyObject> expectedOutput = orderedInput;
+        
+        orderedInput.stream().forEach(olo -> linkifierQueue.offer(olo));
+        
+        final int MAX_TRY = expectedOutput.size() * 256000;
+        List<OrderedLinkifyObject> actualOutput = pollTillMaxTry(MAX_TRY, 
+                expectedOutput.size(), linkifierQueue);
+            
+        assertEquals("Sizes should match.", expectedOutput.size(), actualOutput.size());
+        assertEquals("Checking poll output.", expectedOutput, actualOutput);
+    }
+    
+    private List<OrderedLinkifyObject> pollTillMaxTry(final int MAX_TRY, 
+            final int expectedOutputSize, LinkifyProcessingQueue linkifierQueue) {
+        
+        List<OrderedLinkifyObject> actualOutput = new ArrayList();
+        for(int attempts = 0; 
+                actualOutput.size() < expectedOutputSize && attempts < MAX_TRY; 
+                attempts++) {
+            
+            OrderedLinkifyObject output = linkifierQueue.poll();
+            if(output != null) { actualOutput.add(output); }
+        }
+        
+        return actualOutput;
+    }
+
+    @Test
+    public final void testEmptyQueuePoll() {
+        assertEquals("polling empty queue should return null.", null, linkifierQueue.poll());
+    }
+}
diff --git a/src/test/java/sy/andrew/linkifier/SequentialOutputBufferTest.java b/src/test/java/sy/andrew/linkifier/processors/SequentialOutputBufferTest.java
similarity index 78%
rename from src/test/java/sy/andrew/linkifier/SequentialOutputBufferTest.java
rename to src/test/java/sy/andrew/linkifier/processors/SequentialOutputBufferTest.java
index f5580b2..2131dd1 100644
--- a/src/test/java/sy/andrew/linkifier/SequentialOutputBufferTest.java
+++ b/src/test/java/sy/andrew/linkifier/processors/SequentialOutputBufferTest.java
@@ -1,8 +1,9 @@
-package sy.andrew.linkifier;
+package sy.andrew.linkifier.processors;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
+import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.ConcurrentLinkedQueue;
 
 import org.junit.Test;
@@ -14,7 +15,7 @@ public class SequentialOutputBufferTest {
     @Test
     public void givenInputOutOfOrder_shouldProduceOutputInOrder() {
         // ARRANGE
-        final ConcurrentLinkedQueue<OrderedLinkifyObject> outputQueue = new ConcurrentLinkedQueue<>();
+        final ArrayBlockingQueue<OrderedLinkifyObject> outputQueue = new ArrayBlockingQueue<>(100);
         SequentialOutputBuffer sut = SequentialOutputBuffer.create(1, outputQueue);
 
         // ACT
@@ -43,7 +44,7 @@ public void givenInputOutOfOrder_shouldProduceOutputInOrder() {
     @Test
     public void givenInputWithSkippedSequenceNumber_thenOutputShouldExcludeSequenceNumbersGreaterThanOrEqual() {
         // ARRANGE
-        final ConcurrentLinkedQueue<OrderedLinkifyObject> outputQueue = new ConcurrentLinkedQueue<>();
+        final ArrayBlockingQueue<OrderedLinkifyObject> outputQueue = new ArrayBlockingQueue<>(100);
         SequentialOutputBuffer sut = SequentialOutputBuffer.create(1, outputQueue);
 
         // ACT
@@ -61,16 +62,4 @@ public void givenInputWithSkippedSequenceNumber_thenOutputShouldExcludeSequenceN
         assertTrue("outputQueue should be empty.", outputQueue.isEmpty());        
     }
     
-    @Test
-    public void givenNoInput_thenOutputShouldBeEmpty() {
-        //ARRANGE 
-        final ConcurrentLinkedQueue<OrderedLinkifyObject> outputQueue = 
-                new ConcurrentLinkedQueue<>();
-        SequentialOutputBuffer sut = SequentialOutputBuffer.create(1, outputQueue);
-
-        //ACT - no inputs
-
-        //ASSERT
-        assertTrue("outputQueue should be empty.", outputQueue.isEmpty());        
-    }
 }
\ No newline at end of file
diff --git a/src/test/java/sy/andrew/linkifier/util/OLOLinkifierTest.java b/src/test/java/sy/andrew/linkifier/util/OLOLinkifierTest.java
new file mode 100644
index 0000000..76490e1
--- /dev/null
+++ b/src/test/java/sy/andrew/linkifier/util/OLOLinkifierTest.java
@@ -0,0 +1,33 @@
+package sy.andrew.linkifier.util;
+
+import static org.junit.Assert.assertEquals;
+
+import org.junit.Test;
+
+import sy.andrew.linkifier.model.OrderedLinkifyObject;
+
+public class OLOLinkifierTest {
+
+    @Test
+    public void testLinkify() {
+        //ARRANGE
+        final OLOLinkifier sut = new OLOLinkifier();
+
+        final String stringToLinkify = "http://boo";
+        
+        final OrderedLinkifyObject inputOlo =  new OrderedLinkifyObject(2, stringToLinkify);
+
+        //expectedOutput
+        final OrderedLinkifyObject expectedOutput =  new OrderedLinkifyObject(2, "http://boo");
+        final String linkifiedString = new StringLinkifier().linkify(stringToLinkify);
+        expectedOutput.setLinkifiedString(linkifiedString);
+        
+        //ACT
+        OrderedLinkifyObject actualOutput = sut.linkify(inputOlo);
+        
+        //ASSERT
+        assertEquals("After linkifying, ", actualOutput, expectedOutput);
+        
+    }
+
+}
diff --git a/src/test/java/sy/andrew/linkifier/LinkifierTest.java b/src/test/java/sy/andrew/linkifier/util/StringLinkifierTest.java
similarity index 78%
rename from src/test/java/sy/andrew/linkifier/LinkifierTest.java
rename to src/test/java/sy/andrew/linkifier/util/StringLinkifierTest.java
index 5c41f5f..6a3c5c3 100644
--- a/src/test/java/sy/andrew/linkifier/LinkifierTest.java
+++ b/src/test/java/sy/andrew/linkifier/util/StringLinkifierTest.java
@@ -1,20 +1,22 @@
-package sy.andrew.linkifier;
+package sy.andrew.linkifier.util;
 
 import static org.junit.Assert.assertEquals;
 
 import org.junit.Test;
 
-public class LinkifierTest {
+import sy.andrew.linkifier.util.StringLinkifier;
+
+public class StringLinkifierTest {
 
     @Test
     public final void whenInputTextIsNull_outputIsEmptyString() {
-        final String output = new Linkifier().linkify(null);
+        final String output = new StringLinkifier().linkify(null);
         assertEquals("Given null input, result should be empty string", "", output);
     }
 
     @Test
     public final void whenInputIsEmptyString_outputIsEmptyString() {
-        final String output = new Linkifier().linkify("");
+        final String output = new StringLinkifier().linkify("");
         assertEquals("Given empty string input, result should be empty string", "", output);
     }
 
@@ -22,7 +24,7 @@ public final void whenInputIsEmptyString_outputIsEmptyString() {
     public final void whenInputHasNoUrlNorTags_outputIsSameAsInput() {
         final String inputText = "##hi there bob! ";
         final String expectedOutput = inputText;
-        final String output = new Linkifier().linkify(inputText);
+        final String output = new StringLinkifier().linkify(inputText);
         assertEquals("inputText=[" + inputText + "]", expectedOutput, output);
     }
 
@@ -30,7 +32,7 @@ public final void whenInputHasNoUrlNorTags_outputIsSameAsInput() {
     public final void whenInputHasSchemeHttp_outputIsLinkified() {
         final String inputText = "http://boo";
         final String expectedOutput = "<a href=\"http://boo\">boo</a>";
-        final String output = new Linkifier().linkify(inputText);
+        final String output = new StringLinkifier().linkify(inputText);
         assertEquals("inputText=[" + inputText + "]", expectedOutput, output);
     }
 
@@ -38,7 +40,7 @@ public final void whenInputHasSchemeHttp_outputIsLinkified() {
     public final void whenInputHasSchemeSecureHttps_outputIsLinkified() {
         final String inputText = "https://boo";
         final String expectedOutput = "<a href=\"https://boo\">boo</a>";
-        final String output = new Linkifier().linkify(inputText);
+        final String output = new StringLinkifier().linkify(inputText);
         assertEquals("inputText=[" + inputText + "]", expectedOutput, output);
     }
 
@@ -46,7 +48,7 @@ public final void whenInputHasSchemeSecureHttps_outputIsLinkified() {
     public final void whenInputHasUnrecognizedSchemeFtp_outputIsSameAsInput() {
         final String inputText = "ftp://boo";
         final String expectedOutput = inputText;
-        final String output = new Linkifier().linkify(inputText);
+        final String output = new StringLinkifier().linkify(inputText);
         assertEquals("inputText=[" + inputText + "]", expectedOutput, output);
     }
 
@@ -54,7 +56,7 @@ public final void whenInputHasUnrecognizedSchemeFtp_outputIsSameAsInput() {
     public final void whenInputHasSchemeHttpWithoutBody_outputIsSameAsInput() {
         final String inputText = "http://";
         final String expectedOutput = inputText;
-        final String output = new Linkifier().linkify(inputText);
+        final String output = new StringLinkifier().linkify(inputText);
         assertEquals("inputText=[" + inputText + "]", expectedOutput, output);
     }
 
@@ -62,7 +64,7 @@ public final void whenInputHasSchemeHttpWithoutBody_outputIsSameAsInput() {
     public final void testLinkifyIsCaseInsensitive() {
         final String inputText = "<tiTLe>hTTPs://millenialmedia.COM</Title>";
         final String expectedOutput = "<tiTLe><a href=\"hTTPs://millenialmedia.COM\">millenialmedia.COM</a></Title>";
-        final String output = new Linkifier().linkify(inputText);
+        final String output = new StringLinkifier().linkify(inputText);
         assertEquals("inputText=[" + inputText + "]", expectedOutput, output);
     }
 
@@ -70,7 +72,7 @@ public final void testLinkifyIsCaseInsensitive() {
     public final void testHttpUrlWithQueryParams() {
         final String inputText = "https://nih.gov/array/query?a=1&b=2";
         final String expectedOutput = "<a href=\"https://nih.gov/array/query?a=1&b=2\">nih.gov/array/query?a=1&b=2</a>";
-        final String output = new Linkifier().linkify(inputText);
+        final String output = new StringLinkifier().linkify(inputText);
         assertEquals("inputText=[" + inputText + "]", expectedOutput, output);
     }
 
@@ -78,7 +80,7 @@ public final void testHttpUrlWithQueryParams() {
     public final void testHttpUrlWithHashAndParentheses() {
         final String inputText = "http://docs.oracle.com/javase/6/docs/api/java/util/regex/Matcher.html#appendTail(java.lang.StringBuffer)";
         final String expectedOutput = "<a href=\"http://docs.oracle.com/javase/6/docs/api/java/util/regex/Matcher.html#appendTail(java.lang.StringBuffer)\">docs.oracle.com/javase/6/docs/api/java/util/regex/Matcher.html#appendTail(java.lang.StringBuffer)</a>";
-        final String output = new Linkifier().linkify(inputText);
+        final String output = new StringLinkifier().linkify(inputText);
         assertEquals("inputText=[" + inputText + "]", expectedOutput, output);
     }
 
@@ -86,7 +88,7 @@ public final void testHttpUrlWithHashAndParentheses() {
     public final void whenInputIsEmptyTag_outputIsSameAsInput() {
         final String inputText = "<>< >";
         final String expectedOutput = inputText;
-        final String output = new Linkifier().linkify(inputText);
+        final String output = new StringLinkifier().linkify(inputText);
         assertEquals("inputText=[" + inputText + "]", expectedOutput, output);
     }
 
@@ -94,7 +96,7 @@ public final void whenInputIsEmptyTag_outputIsSameAsInput() {
     public final void testNestedHtmlElements() {
         final String inputText = "<html><title>The Title</title></html>";
         final String expectedOutput = inputText;
-        final String output = new Linkifier().linkify(inputText);
+        final String output = new StringLinkifier().linkify(inputText);
         assertEquals("inputText=[" + inputText + "]", expectedOutput, output);
     }
 
@@ -102,7 +104,7 @@ public final void testNestedHtmlElements() {
     public final void testHttpSchemeInsideHtmlTag() {
         final String inputText = "<img src=\"http://boo\"/>";
         final String expectedOutput = inputText;
-        final String output = new Linkifier().linkify(inputText);
+        final String output = new StringLinkifier().linkify(inputText);
         assertEquals("inputText=[" + inputText + "]", expectedOutput, output);
     }
 
@@ -110,15 +112,15 @@ public final void testHttpSchemeInsideHtmlTag() {
     public final void testMultipleHttpUrlsAndMultipleHtmlTags() {
         final String inputText = "<html><body><div>https://millenialmedia.com/boo http://amazon.com</div><http://foo/></body></html>";
         final String expectedOutput = "<html><body><div><a href=\"https://millenialmedia.com/boo\">millenialmedia.com/boo</a> <a href=\"http://amazon.com\">amazon.com</a></div><http://foo/></body></html>";
-        final String output = new Linkifier().linkify(inputText);
+        final String output = new StringLinkifier().linkify(inputText);
         assertEquals("inputText=[" + inputText + "]", expectedOutput, output);
     }
 
     @Test
     public final void testRelinkifyingIsIdempotent() {
         final String inputText = "<html><body><div>https://millenialmedia.com/boo http://amazon.com</div><http://foo/></body></html>";
-        final String linkifiedText = new Linkifier().linkify(inputText);
-        final String relinkifiedText = new Linkifier().linkify(linkifiedText);
+        final String linkifiedText = new StringLinkifier().linkify(inputText);
+        final String relinkifiedText = new StringLinkifier().linkify(linkifiedText);
         assertEquals("relinkifiedText should equal linkifiedText", linkifiedText, relinkifiedText);
     }
 
@@ -126,7 +128,7 @@ public final void testRelinkifyingIsIdempotent() {
     public final void testRandomSpaces1() {
         final String inputText = "  \t<img  src=\"http://boo\" />  < link /> ";
         final String expectedOutput = inputText;
-        final String output = new Linkifier().linkify(inputText);
+        final String output = new StringLinkifier().linkify(inputText);
         assertEquals("inputText=[" + inputText + "]", expectedOutput, output);
     }
 
@@ -134,7 +136,7 @@ public final void testRandomSpaces1() {
     public final void testRandomSpaces2() {
         final String inputText = " http://foo/boo hi there \t mo&bo http://goo  ";
         final String expectedOutput = " <a href=\"http://foo/boo\">foo/boo</a> hi there \t mo&bo <a href=\"http://goo\">goo</a>  ";
-        final String output = new Linkifier().linkify(inputText);
+        final String output = new StringLinkifier().linkify(inputText);
         assertEquals("inputText=[" + inputText + "]", expectedOutput, output);
     }
 
@@ -146,7 +148,7 @@ public final void testMultipleLineInput() {
         final String expectedOutput = "<html>\n" + "<body> \n" + "<div>\n"
                 + "<a href=\"https://millenialmedia.com/boo\">millenialmedia.com/boo</a>\n"
                 + " <a href=\"http://amazon.com\">amazon.com</a></div><http://foo/></body></html>";
-        final String output = new Linkifier().linkify(inputText);
+        final String output = new StringLinkifier().linkify(inputText);
         assertEquals("inputText=[" + inputText + "]", expectedOutput, output);
     }
 }