Issue #896: Close input in same thread as parser

Signed-off-by: James Leigh <james.leigh@ontotext.com>

Author: James Leigh

core/query/src/main/java/org/eclipse/rdf4j/query/impl/BackgroundGraphResult.java

@@ -9,11 +9,9 @@
 
 import java.io.InputStream;
 import java.io.InputStreamReader;
-import java.lang.reflect.UndeclaredThrowableException;
 import java.nio.charset.Charset;
 import java.util.Collections;
 import java.util.Map;
-import java.util.NoSuchElementException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 
@@ -45,6 +43,8 @@ public class BackgroundGraphResult extends IterationWrapper<Statement, QueryEval
 
 	private final CountDownLatch namespacesReady = new CountDownLatch(1);
 
+	private final CountDownLatch finishedParsing = new CountDownLatch(1);
+
 	private final Map<String, String> namespaces = new ConcurrentHashMap<String, String>();
 
 	private final QueueCursor<Statement> queue;
@@ -64,67 +64,6 @@ public BackgroundGraphResult(QueueCursor<Statement> queue, RDFParser parser, Inp
 		this.baseURI = baseURI;
 	}
 
-	@Override
-	public boolean hasNext()
-		throws QueryEvaluationException
-	{
-		if (isClosed()) {
-			return false;
-		}
-		if (Thread.currentThread().isInterrupted()) {
-			close();
-			return false;
-		}
-
-		boolean result = queue.hasNext();
-		if (!result) {
-			close();
-		}
-		return result;
-	}
-
-	@Override
-	public Statement next()
-		throws QueryEvaluationException
-	{
-		if (isClosed()) {
-			throw new NoSuchElementException("The iteration has been closed.");
-		}
-		if (Thread.currentThread().isInterrupted()) {
-			close();
-			throw new NoSuchElementException("The iteration has been closed.");
-		}
-
-		try {
-			return queue.next();
-		}
-		catch (NoSuchElementException e) {
-			close();
-			throw e;
-		}
-	}
-
-	@Override
-	public void remove()
-		throws QueryEvaluationException
-	{
-		if (isClosed()) {
-			throw new IllegalStateException("The iteration has been closed.");
-		}
-		if (Thread.currentThread().isInterrupted()) {
-			close();
-			throw new IllegalStateException("The iteration has been closed.");
-		}
-
-		try {
-			queue.remove();
-		}
-		catch (IllegalStateException e) {
-			close();
-			throw e;
-		}
-	}
-
 	@Override
 	protected void handleClose()
 		throws QueryEvaluationException
@@ -135,30 +74,38 @@ protected void handleClose()
 		finally {
 			queue.done();
 		}
+		try {
+			finishedParsing.await();
+		}
+		catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+		} finally {
+			queue.checkException();
+		}
 	}
 
 	@Override
 	public void run() {
 		try {
-			parser.setRDFHandler(this);
-			if (charset == null) {
-				parser.parse(in, baseURI);
+			try {
+				parser.setRDFHandler(this);
+				if (charset == null) {
+					parser.parse(in, baseURI);
+				}
+				else {
+					parser.parse(new InputStreamReader(in, charset), baseURI);
+				}
+			} finally {
+				in.close();
 			}
-			else {
-				parser.parse(new InputStreamReader(in, charset), baseURI);
-			}
-		}
-		catch (RDFHandlerException e) {
-			// parsing was cancelled or interrupted
-			close();
 		}
 		catch (Exception e) {
 			queue.toss(e);
-			close();
 		}
 		finally {
 			queue.done();
 			namespacesReady.countDown();
+			finishedParsing.countDown();
 		}
 	}
 
@@ -178,8 +125,9 @@ public Map<String, String> getNamespaces() {
 		}
 		catch (InterruptedException e) {
 			Thread.currentThread().interrupt();
-			close();
-			throw new UndeclaredThrowableException(e);
+			return Collections.emptyMap();
+		} finally {
+			queue.checkException();
 		}
 	}
 
@@ -207,11 +155,8 @@ public void handleStatement(Statement st)
 		}
 		catch (InterruptedException e) {
 			Thread.currentThread().interrupt();
-			close();
-			throw new RDFHandlerException(e);
-		}
-		if (isClosed()) {
-			throw new RDFHandlerException("Result closed");
+			queue.toss(e);
+			queue.done();
 		}
 	}
 

core/queryresultio/api/src/main/java/org/eclipse/rdf4j/query/resultio/helpers/BackgroundTupleResult.java

@@ -8,7 +8,6 @@
 package org.eclipse.rdf4j.query.resultio.helpers;
 
 import java.io.InputStream;
-import java.lang.reflect.UndeclaredThrowableException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -42,6 +41,8 @@ public class BackgroundTupleResult extends IteratingTupleQueryResult
 
 	private final CountDownLatch bindingNamesReady = new CountDownLatch(1);
 
+	private final CountDownLatch finishedParsing = new CountDownLatch(1);
+
 	public BackgroundTupleResult(TupleQueryResultParser parser, InputStream in) {
 		this(new QueueCursor<BindingSet>(10), parser, in);
 	}
@@ -65,43 +66,47 @@ protected void handleClose()
 		finally {
 			queue.done();
 		}
+		try {
+			finishedParsing.await();
+		}
+		catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+		} finally {
+			queue.checkException();
+		}
 	}
 
 	@Override
 	public List<String> getBindingNames() {
 		try {
 			bindingNamesReady.await();
-			queue.checkException();
 			return bindingNames;
 		}
 		catch (InterruptedException e) {
 			Thread.currentThread().interrupt();
-			close();
-			throw new UndeclaredThrowableException(e);
-		}
-		catch (QueryEvaluationException e) {
-			close();
-			throw new UndeclaredThrowableException(e);
+			return Collections.emptyList();
+		} finally {
+			queue.checkException();
 		}
 	}
 
 	@Override
 	public void run() {
 		try {
-			parser.setQueryResultHandler(this);
-			parser.parseQueryResult(in);
-		}
-		catch (QueryResultHandlerException e) {
-			// parsing cancelled or interrupted
-			close();
+			try {
+				parser.setQueryResultHandler(this);
+				parser.parseQueryResult(in);
+			} finally {
+				in.close();
+			}
 		}
 		catch (Exception e) {
 			queue.toss(e);
-			close();
 		}
 		finally {
 			queue.done();
 			bindingNamesReady.countDown();
+			finishedParsing.countDown();
 		}
 	}
 
@@ -122,11 +127,8 @@ public void handleSolution(BindingSet bindingSet)
 		}
 		catch (InterruptedException e) {
 			Thread.currentThread().interrupt();
-			close();
-			throw new TupleQueryResultHandlerException(e);
-		}
-		if (isClosed()) {
-			throw new TupleQueryResultHandlerException("Result closed");
+			queue.toss(e);
+			queue.done();
 		}
 	}
 

core/util/src/main/java/org/eclipse/rdf4j/common/iteration/IterationWrapper.java

@@ -60,6 +60,10 @@ public boolean hasNext()
 		if (isClosed()) {
 			return false;
 		}
+		else if (Thread.currentThread().isInterrupted()) {
+			close();
+			return false;
+		}
 		boolean result = wrappedIter.hasNext();
 		if (!result) {
 			close();
@@ -79,6 +83,10 @@ public E next()
 		if (isClosed()) {
 			throw new NoSuchElementException("The iteration has been closed.");
 		}
+		else if (Thread.currentThread().isInterrupted()) {
+			close();
+			throw new NoSuchElementException("The iteration has been interrupted.");
+		}
 		try {
 			return wrappedIter.next();
 		}
@@ -103,6 +111,10 @@ public void remove()
 		if (isClosed()) {
 			throw new IllegalStateException("The iteration has been closed.");
 		}
+		else if (Thread.currentThread().isInterrupted()) {
+			close();
+			throw new IllegalStateException("The iteration has been interrupted.");
+		}
 		try {
 			wrappedIter.remove();
 		}

core/util/src/main/java/org/eclipse/rdf4j/common/iteration/QueueIteration.java

@@ -142,8 +142,8 @@ public E getNextElement()
 				}
 			}
 			if (isAfterLast(take)) {
-				checkException();
 				done(); // put afterLast back for others
+				checkException();
 				return null;
 			}
 			checkException();