fix iterators may not be closed

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.baidu.hugegraph</groupId>
     <artifactId>hugegraph-common</artifactId>
-    <version>1.7.1</version>
+    <version>1.7.2</version>
 
     <name>hugegraph-common</name>
     <url>https://github.com/hugegraph/hugegraph-common</url>
@@ -218,7 +218,7 @@
                         <manifestEntries>
                             <!-- Must be on one line, otherwise the automatic
                                  upgrade script cannot replace the version number -->
-                            <Implementation-Version>1.7.1.0</Implementation-Version>
+                            <Implementation-Version>1.7.2.0</Implementation-Version>
                         </manifestEntries>
                     </archive>
                 </configuration>

src/main/java/com/baidu/hugegraph/iterator/BatchMapperIterator.java

@@ -0,0 +1,103 @@
+/*
+ * Copyright 2017 HugeGraph Authors
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.baidu.hugegraph.iterator;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.function.Function;
+
+import com.baidu.hugegraph.util.E;
+import com.baidu.hugegraph.util.InsertionOrderUtil;
+import com.google.common.collect.ImmutableList;
+
+public class BatchMapperIterator<T, R> extends WrappedIterator<R> {
+
+    private final int batch;
+    private final Iterator<T> originIterator;
+    private final Function<List<T>, Iterator<R>> mapperCallback;
+
+    private Iterator<R> batchIterator;
+
+    public BatchMapperIterator(int batch, Iterator<T> origin,
+                               Function<List<T>, Iterator<R>> mapper) {
+        E.checkArgument(batch > 0, "Expect batch > 0, but got %s", batch);
+        this.batch = batch;
+        this.originIterator = origin;
+        this.mapperCallback = mapper;
+        this.batchIterator = null;
+    }
+
+    @Override
+    protected Iterator<T> originIterator() {
+        return this.originIterator;
+    }
+
+    @Override
+    protected final boolean fetch() {
+        if (this.batchIterator != null && this.fetchFromBatch()) {
+            return true;
+        }
+
+        List<T> list = this.fetchBatch();
+        if (!list.isEmpty()) {
+            assert this.batchIterator == null;
+            // Do fetch
+            this.batchIterator = this.mapperCallback.apply(list);
+            if (this.batchIterator != null && this.fetchFromBatch()) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    protected List<T> fetchBatch() {
+        if (!this.originIterator.hasNext()) {
+            return ImmutableList.of();
+        }
+        List<T> list = InsertionOrderUtil.newList();
+        for (int i = 0; i < this.batch && this.originIterator.hasNext(); i++) {
+            T next = this.originIterator.next();
+            list.add(next);
+        }
+        return list;
+    }
+
+    protected boolean fetchFromBatch() {
+        E.checkNotNull(this.batchIterator, "mapper results");
+        while (this.batchIterator.hasNext()) {
+            R result = this.batchIterator.next();
+            if (result != null) {
+                assert this.current == none();
+                this.current = result;
+                return true;
+            }
+        }
+        this.resetResults();
+        return false;
+    }
+
+    protected final void resetResults() {
+        if (this.batchIterator == null) {
+            return;
+        }
+        close(this.batchIterator);
+        this.batchIterator = null;
+    }
+}

src/main/java/com/baidu/hugegraph/iterator/IterableIterator.java → src/main/java/com/baidu/hugegraph/iterator/CIter.java

@@ -21,32 +21,5 @@
 
 import java.util.Iterator;
 
-public class IterableIterator<T> implements Iterator<T> {
-
-    private final Iterable<T> iterable;
-    private final Iterator<T> iterator;
-
-    public IterableIterator(Iterable<T> iterable) {
-        this.iterable = iterable;
-        this.iterator = iterable.iterator();
-    }
-
-    @Override
-    public boolean hasNext() {
-        return this.iterator.hasNext();
-    }
-
-    @Override
-    public T next() {
-        return this.iterator.next();
-    }
-
-    @Override
-    public void remove() {
-        this.iterator.remove();
-    }
-
-    public Iterable<T> iterable() {
-        return iterable;
-    }
+public interface CIter<R> extends Iterator<R>, AutoCloseable, Metadatable {
 }

src/main/java/com/baidu/hugegraph/iterator/ExtendableIterator.java

@@ -19,24 +19,20 @@
 
 package com.baidu.hugegraph.iterator;
 
-import java.util.ArrayList;
 import java.util.Deque;
 import java.util.Iterator;
-import java.util.List;
 import java.util.concurrent.ConcurrentLinkedDeque;
 
 import com.baidu.hugegraph.util.E;
 
 public class ExtendableIterator<T> extends WrappedIterator<T> {
 
     private final Deque<Iterator<T>> itors;
-    private final List<Iterator<T>> removedItors;
 
     private Iterator<T> currentIterator;
 
     public ExtendableIterator() {
         this.itors = new ConcurrentLinkedDeque<>();
-        this.removedItors = new ArrayList<>();
         this.currentIterator = null;
     }
 
@@ -62,11 +58,6 @@ public ExtendableIterator<T> extend(Iterator<T> iter) {
 
     @Override
     public void close() throws Exception {
-        for (Iterator<T> iter : this.removedItors) {
-            if (iter instanceof AutoCloseable) {
-                ((AutoCloseable) iter).close();
-            }
-        }
         for (Iterator<T> iter : this.itors) {
             if (iter instanceof AutoCloseable) {
                 ((AutoCloseable) iter).close();
@@ -75,7 +66,7 @@ public void close() throws Exception {
     }
 
     @Override
-    protected Iterator<?> originIterator() {
+    protected Iterator<T> originIterator() {
         return this.currentIterator;
     }
 
@@ -98,7 +89,7 @@ protected boolean fetch() {
                 // The last one
                 return false;
             }
-            this.removedItors.add(this.itors.removeFirst());
+            close(this.itors.removeFirst());
         }
 
         assert first != null && first.hasNext();

src/main/java/com/baidu/hugegraph/iterator/FlatMapperFilterIterator.java

@@ -36,17 +36,17 @@ public FlatMapperFilterIterator(Iterator<T> origin,
     }
 
     @Override
-    protected final boolean fetchMapped() {
-        E.checkNotNull(this.results, "mapper results");
-        while (this.results.hasNext()) {
-            R result = this.results.next();
+    protected final boolean fetchFromMap() {
+        E.checkNotNull(this.batchIterator, "mapper results");
+        while (this.batchIterator.hasNext()) {
+            R result = this.batchIterator.next();
             if (result != null && this.filterCallback.apply(result)) {
                 assert this.current == none();
                 this.current = result;
                 return true;
             }
         }
-        this.results = null;
+        this.resetResults();
         return false;
     }
 }

src/main/java/com/baidu/hugegraph/iterator/FlatMapperIterator.java

@@ -29,48 +29,63 @@ public class FlatMapperIterator<T, R> extends WrappedIterator<R> {
     private final Iterator<T> originIterator;
     private final Function<T, Iterator<R>> mapperCallback;
 
-    protected Iterator<R> results;
+    protected Iterator<R> batchIterator;
 
     public FlatMapperIterator(Iterator<T> origin,
                               Function<T, Iterator<R>> mapper) {
         this.originIterator = origin;
         this.mapperCallback = mapper;
-        this.results = null;
+        this.batchIterator = null;
     }
 
     @Override
-    protected Iterator<?> originIterator() {
+    public void close() throws Exception {
+        this.resetResults();
+        super.close();
+    }
+
+    @Override
+    protected Iterator<T> originIterator() {
         return this.originIterator;
     }
 
     @Override
     protected final boolean fetch() {
-        if (this.results != null && this.fetchMapped()) {
+        if (this.batchIterator != null && this.fetchFromMap()) {
             return true;
         }
 
         while (this.originIterator.hasNext()) {
             T next = this.originIterator.next();
-            assert this.results == null;
-            this.results = this.mapperCallback.apply(next);
-            if (this.results != null && this.fetchMapped()) {
+            assert this.batchIterator == null;
+            // Do fetch
+            this.batchIterator = this.mapperCallback.apply(next);
+            if (this.batchIterator != null && this.fetchFromMap()) {
                 return true;
             }
         }
         return false;
     }
 
-    protected boolean fetchMapped() {
-        E.checkNotNull(this.results, "mapper results");
-        while (this.results.hasNext()) {
-            R result = this.results.next();
+    protected boolean fetchFromMap() {
+        E.checkNotNull(this.batchIterator, "mapper results");
+        while (this.batchIterator.hasNext()) {
+            R result = this.batchIterator.next();
             if (result != null) {
                 assert this.current == none();
                 this.current = result;
                 return true;
             }
         }
-        this.results = null;
+        this.resetResults();
         return false;
     }
+
+    protected final void resetResults() {
+        if (this.batchIterator == null) {
+            return;
+        }
+        close(this.batchIterator);
+        this.batchIterator = null;
+    }
 }

src/main/java/com/baidu/hugegraph/iterator/ListIterator.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright 2017 HugeGraph Authors
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.baidu.hugegraph.iterator;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+
+import com.baidu.hugegraph.util.InsertionOrderUtil;
+
+public class ListIterator<T> extends WrappedIterator<T> {
+
+    private final Iterator<T> originIterator;
+    private final Iterator<T> resultsIterator;
+    private final Collection<T> results;
+
+    public ListIterator(long capacity, Iterator<T> origin) {
+        List<T> results = InsertionOrderUtil.newList();
+        while (origin.hasNext()) {
+            if (capacity >= 0L && results.size() >= capacity) {
+                throw new IllegalArgumentException(
+                          "The iterator exceeded capacity " + capacity);
+            }
+            results.add(origin.next());
+        }
+        this.originIterator = origin;
+        this.results = Collections.unmodifiableList(results);
+        this.resultsIterator = this.results.iterator();
+    }
+
+    public ListIterator(Collection<T> origin) {
+        this.originIterator = origin.iterator();
+        this.results = origin instanceof List ?
+                       Collections.unmodifiableList((List<T>) origin) :
+                       Collections.unmodifiableCollection(origin);
+        this.resultsIterator = this.results.iterator();
+    }
+
+    @Override
+    public void remove() {
+        this.resultsIterator.remove();
+    }
+
+    public Collection<T> list() {
+        return this.results;
+    }
+
+    @Override
+    protected boolean fetch() {
+        assert this.current == none();
+        if (!this.resultsIterator.hasNext()) {
+            return false;
+        }
+        this.current = this.resultsIterator.next();
+        return true;
+    }
+
+    @Override
+    protected Iterator<T> originIterator() {
+        return this.originIterator;
+    }
+}

src/main/java/com/baidu/hugegraph/iterator/MapperIterator.java

@@ -33,7 +33,7 @@ public MapperIterator(Iterator<T> origin, Function<T, R> mapper) {
     }
 
     @Override
-    protected Iterator<?> originIterator() {
+    protected Iterator<T> originIterator() {
         return this.originIterator;
     }