Closes #198: Implement the BlockingQueue as a wrapper around the Queu…

…eFile Switched to an ObjectQueue<T> as the backing store
square · Oct 25, 2018 · 7fb0b22 · 7fb0b22
1 parent db76821
commit 7fb0b22
Show file tree

Hide file tree

Showing 3 changed files with 120 additions and 93 deletions.
diff --git a/...com/squareup/tape2/BlockingFileQueue.java → ...m/squareup/tape2/BlockingObjectQueue.java b/...com/squareup/tape2/BlockingFileQueue.java → ...m/squareup/tape2/BlockingObjectQueue.java
@@ -1,38 +1,48 @@
 package com.squareup.tape2;
 
 import java.io.IOException;
-import java.util.Arrays;
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.NoSuchElementException;
+import java.util.Objects;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
 /**
- * This is a simple and unbounded @{@link BlockingQueue} implementation as a wrapper around a @{@link QueueFile}.
- * Thread safety is implemented using a single lock around all operations to the backing @{@link QueueFile}.
+ * This is a simple and unbounded {@link BlockingQueue} implementation as a wrapper
+ * around a {@link QueueFile}. Thread safety is implemented using a single lock around all
+ * operations to the backing {@link QueueFile}.
  */
-public class BlockingFileQueue implements BlockingQueue<byte[]> {
+public class BlockingObjectQueue<E> implements BlockingQueue<E> {
 
   private final Lock lock = new ReentrantLock();
 
   private final Condition nonEmpty = lock.newCondition();
 
-  private final QueueFile queue;
+  private final ObjectQueue<E> queue;
 
   /**
-   * Creates a new @{@link BlockingQueue} of type {@code byte[]} backed by the given @{@link QueueFile}.
+   * Creates a new {@link BlockingQueue} of type {@code E} backed by the given
+   * {@link ObjectQueue} of type {@code E}.
    *
    * @param queue the queue file which should not be shared to other places
    * */
-  public BlockingFileQueue(QueueFile queue) {
+  public BlockingObjectQueue(ObjectQueue<E> queue) {
     this.queue = queue;
   }
 
-  @Override public boolean add(byte[] bytes) {
+  public static <T> BlockingObjectQueue<T> create(QueueFile qf, ObjectQueue.Converter<T> conv) {
+    return new BlockingObjectQueue<>(new FileObjectQueue<>(qf, conv));
+  }
+
+  public static BlockingObjectQueue<byte[]> create(QueueFile qf) {
+    return create(qf, ByteArrayConverter.INSTANCE);
+  }
+
+  @Override public boolean add(E bytes) {
     lock.lock();
     try {
       queue.add(bytes);
@@ -45,25 +55,25 @@ public BlockingFileQueue(QueueFile queue) {
     }
   }
 
-  @Override public boolean offer(byte[] bytes) {
+  @Override public boolean offer(E bytes) {
     return add(bytes);
   }
 
-  @Override public void put(byte[] bytes) {
+  @Override public void put(E bytes) {
     add(bytes);
   }
 
-  @Override public boolean offer(byte[] bytes, long timeout, TimeUnit unit) {
+  @Override public boolean offer(E bytes, long timeout, TimeUnit unit) {
     return add(bytes);
   }
 
-  @Override public byte[] take() throws InterruptedException {
+  @Override public E take() throws InterruptedException {
     lock.lock();
     try {
       while (queue.isEmpty()) {
         nonEmpty.await();
       }
-      byte[] peek = queue.peek();
+      E peek = queue.peek();
       if (peek == null) {
         throw new IllegalStateException("Queue empty!");
       }
@@ -76,14 +86,14 @@ public BlockingFileQueue(QueueFile queue) {
     }
   }
 
-  @Override public byte[] poll(long timeout, TimeUnit unit) throws InterruptedException {
+  @Override public E poll(long timeout, TimeUnit unit) throws InterruptedException {
     lock.lock();
     try {
       long timeoutNanos = unit.toNanos(timeout);
       while (queue.isEmpty() && timeoutNanos > 0) {
         timeoutNanos = nonEmpty.awaitNanos(timeoutNanos);
       }
-      byte[] peek = queue.peek();
+      E peek = queue.peek();
       if (peek == null) {
         throw new NoSuchElementException();
       }
@@ -100,23 +110,22 @@ public BlockingFileQueue(QueueFile queue) {
     return Integer.MAX_VALUE; // as per BlockingQueue javadoc for unbounded queues
   }
 
-  /** The backing @{@link QueueFile} only supports removing the head, so this will only work if head matches. */
+  /**
+   * The underlying {@link QueueFile} only supports removing the head, so this will only work if
+   * head matches.
+   * */
   @Override public boolean remove(Object o) {
     if (o == null) {
       return false;
     }
-    if (o.getClass() != byte[].class && o.getClass() != Byte[].class) {
-      return false;
-    }
-    byte[] remove = (byte[])o;
     lock.lock();
     try {
       if (queue.isEmpty()) {
         return false;
       }
-      Iterator<byte[]> it = queue.iterator();
+      Iterator<E> it = queue.iterator();
       while (it.hasNext()) {
-        if (Arrays.equals(it.next(), remove)) {
+        if (Objects.deepEquals(it.next(), o)) {
           it.remove();
           return true;
         }
@@ -128,17 +137,10 @@ public BlockingFileQueue(QueueFile queue) {
   }
 
   @Override public boolean contains(Object o) {
-    if (o == null) {
-      return false;
-    }
-    if (o.getClass() != byte[].class && o.getClass() != Byte[].class) {
-      return false;
-    }
-    byte[] check = (byte[])o;
     lock.lock();
     try {
-      for (byte[] entry : queue) {
-        if (Arrays.equals(entry, check)) {
+      for (E entry : queue) {
+        if (Objects.deepEquals(entry, o)) {
           return true;
         }
       }
@@ -148,11 +150,11 @@ public BlockingFileQueue(QueueFile queue) {
     }
   }
 
-  @Override public int drainTo(Collection<? super byte[]> c) {
+  @Override public int drainTo(Collection<? super E> c) {
     lock.lock();
     try {
       int size = queue.size();
-      Iterator<byte[]> it = queue.iterator();
+      Iterator<E> it = queue.iterator();
       while (it.hasNext()) {
         c.add(it.next());
         it.remove();
@@ -163,13 +165,13 @@ public BlockingFileQueue(QueueFile queue) {
     }
   }
 
-  @Override public int drainTo(Collection<? super byte[]> c, int maxElements) {
+  @Override public int drainTo(Collection<? super E> c, int maxElements) {
     if (maxElements == 0) {
       return 0;
     }
     lock.lock();
     try {
-      Iterator<byte[]> it = queue.iterator();
+      Iterator<E> it = queue.iterator();
       int i = 0;
       while (it.hasNext() && i < maxElements) {
         c.add(it.next());
@@ -182,13 +184,13 @@ public BlockingFileQueue(QueueFile queue) {
     }
   }
 
-  @Override public byte[] remove() {
+  @Override public E remove() {
     lock.lock();
     try {
       if (queue.isEmpty()) {
         throw new NoSuchElementException();
       }
-      byte[] peek = queue.peek();
+      E peek = queue.peek();
       queue.remove();
       return peek;
     } catch (IOException e) {
@@ -198,13 +200,13 @@ public BlockingFileQueue(QueueFile queue) {
     }
   }
 
-  @Override public byte[] poll() {
+  @Override public E poll() {
     lock.lock();
     try {
       if (queue.isEmpty()) {
         return null;
       }
-      byte[] peek = queue.peek();
+      E peek = queue.peek();
       queue.remove();
       return peek;
     } catch (IOException e) {
@@ -214,7 +216,7 @@ public BlockingFileQueue(QueueFile queue) {
     }
   }
 
-  @Override public byte[] element() {
+  @Override public E element() {
     lock.lock();
     try {
       if (queue.isEmpty()) {
@@ -228,7 +230,7 @@ public BlockingFileQueue(QueueFile queue) {
     }
   }
 
-  @Override public byte[] peek() {
+  @Override public E peek() {
     lock.lock();
     try {
       if (queue.isEmpty()) {
@@ -243,23 +245,23 @@ public BlockingFileQueue(QueueFile queue) {
   }
 
   /**
-   * This overload is an addition to the @{@link BlockingQueue} interface similar to the {@link #poll(long, TimeUnit)}
-   * method, a blocking peek operation with a timeout.
+   * This overload is an addition to the {@link BlockingQueue} interface similar to the
+   * {@link #poll(long, TimeUnit)} method, a blocking peek operation with a timeout.
    *
    * @param timeout the timeout
    * @param unit the time unit of the timeout
    * @return the head of this queue, or {@code null} if this queue is empty
    * @see #peek() for more information
    * @throws InterruptedException if interrupted while waiting
    */
-  public byte[] peek(long timeout, TimeUnit unit) throws InterruptedException {
+  public E peek(long timeout, TimeUnit unit) throws InterruptedException {
     lock.lock();
     try {
       long timeoutNanos = unit.toNanos(timeout);
       while (queue.isEmpty() && timeoutNanos > 0) {
         timeoutNanos = nonEmpty.awaitNanos(timeoutNanos);
       }
-      byte[] peek = queue.peek();
+      E peek = queue.peek();
       if (peek == null) {
         throw new NoSuchElementException();
       }
@@ -289,7 +291,7 @@ public byte[] peek(long timeout, TimeUnit unit) throws InterruptedException {
     }
   }
 
-  @Override public Iterator<byte[]> iterator() {
+  @Override public Iterator<E> iterator() {
     lock.lock();
     try {
       return queue.iterator();
@@ -303,7 +305,7 @@ public byte[] peek(long timeout, TimeUnit unit) throws InterruptedException {
     try {
       Object[] out = new Object[queue.size()];
       int i = 0;
-      for (byte[] e : queue) {
+      for (E e : queue) {
         out[i++] = e;
       }
       return out;
@@ -330,13 +332,13 @@ public byte[] peek(long timeout, TimeUnit unit) throws InterruptedException {
     }
   }
 
-  @Override public boolean addAll(Collection<? extends byte[]> c) {
+  @Override public boolean addAll(Collection<? extends E> c) {
     if (c.isEmpty()) {
       return false;
     }
     lock.lock();
     try {
-      for (byte[] e : c) {
+      for (E e : c) {
         queue.add(e);
       }
       nonEmpty.signal();
@@ -348,7 +350,10 @@ public byte[] peek(long timeout, TimeUnit unit) throws InterruptedException {
     }
   }
 
-  /** The backing @{@link QueueFile} only supports removing the head, so this will only work if head matches. */
+  /**
+   * The underlying {@link QueueFile} only supports removing the head, so this will only work if
+   * head matches.
+   * */
   @Override public boolean removeAll(Collection<?> c) {
     lock.lock();
     boolean changed = false;
@@ -364,17 +369,19 @@ public byte[] peek(long timeout, TimeUnit unit) throws InterruptedException {
     }
   }
 
-  private static boolean arrayContained(Collection<?> haystack, byte[] needle) {
+  private static boolean contains(Iterable<?> haystack, Object needle) {
     for (Object o : haystack) {
-      byte[] byteArray = (byte[])o;
-      if (Arrays.equals(byteArray, needle)) {
+      if (Objects.deepEquals(o, needle)) {
         return true;
       }
     }
     return false;
   }
 
-  /** The backing @{@link QueueFile} only supports removing the head, so this will only work if head matches. */
+  /**
+   * The underlying {@link QueueFile} only supports removing the head, so this will only work if
+   * head matches.
+   * */
   @Override public boolean retainAll(Collection<?> c) {
     lock.lock();
     if (c.isEmpty()) {
@@ -387,9 +394,9 @@ private static boolean arrayContained(Collection<?> haystack, byte[] needle) {
     }
     try {
       boolean changed = false;
-      Iterator<byte[]> it = queue.iterator();
+      Iterator<E> it = queue.iterator();
       while (it.hasNext()) {
-        if (!arrayContained(c, it.next())) {
+        if (!contains(c, it.next())) {
           it.remove();
           changed = true;
         }

diff --git a/tape/src/main/java/com/squareup/tape2/ByteArrayConverter.java b/tape/src/main/java/com/squareup/tape2/ByteArrayConverter.java
@@ -0,0 +1,20 @@
+package com.squareup.tape2;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+/**
+ * This is a simple "NO-OP" converter for byte[] queues which don't actually need a conversion.
+ */
+public class ByteArrayConverter implements ObjectQueue.Converter<byte[]> {
+
+  public static final ByteArrayConverter INSTANCE = new ByteArrayConverter();
+
+  @Override public byte[] from(byte[] source) {
+    return source;
+  }
+
+  @Override public void toStream(byte[] value, OutputStream sink) throws IOException {
+    sink.write(value);
+  }
+}