Update Pipe.

Author: gk-brown

kilo-client/src/main/java/org/httprpc/kilo/util/concurrent/Pipe.java

@@ -28,7 +28,8 @@
  * The element type.
  */
 public class Pipe<E> implements Iterable<E> {
-    private BlockingQueue<Object> queue;
+    private volatile BlockingQueue<Object> queue;
+
     private int timeout;
 
     private static final Object TERMINATOR = new Object();
@@ -81,6 +82,10 @@ public void submit(Iterable<? extends E> elements) {
             throw new IllegalArgumentException();
         }
 
+        if (queue == null) {
+            throw new IllegalStateException();
+        }
+
         for (var element : elements) {
             submit(element);
         }
@@ -112,6 +117,10 @@ public Iterator<E> iterator() {
             @Override
             @SuppressWarnings("unchecked")
             public boolean hasNext() {
+                if (queue == null) {
+                    return false;
+                }
+
                 if (next == null) {
                     Object value;
                     try {
@@ -130,6 +139,8 @@ public boolean hasNext() {
 
                     if (value != TERMINATOR) {
                         next = (E)value;
+                    } else {
+                        queue = null;
                     }
                 }