Merge branch 'master' of github.com:eclipse/rdf4j into develop

Author: James Leigh

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

@@ -7,14 +7,11 @@
  *******************************************************************************/
 package org.eclipse.rdf4j.query.impl;
 
-import java.io.IOException;
 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;
 
@@ -46,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;
@@ -66,118 +65,47 @@ public BackgroundGraphResult(QueueCursor<Statement> queue, RDFParser parser, Inp
 	}
 
 	@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()
+	protected void handleClose()
 		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;
+			super.handleClose();
 		}
-	}
-
-	@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.");
+		finally {
+			queue.done();
 		}
-
 		try {
-			queue.remove();
+			finishedParsing.await();
 		}
-		catch (IllegalStateException e) {
-			close();
-			throw e;
+		catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+		} finally {
+			queue.checkException();
 		}
 	}
 
 	@Override
-	protected void handleClose()
-		throws QueryEvaluationException
-	{
+	public void run() {
 		try {
 			try {
-				super.handleClose();
-			}
-			finally {
-				try {
-					// After checking that we ourselves cannot possibly be generating an NPE ourselves, 
-					// attempt to close the input stream we were given
-					InputStream toClose = in;
-					if (toClose != null) {
-						toClose.close();
-					}
+				parser.setRDFHandler(this);
+				if (charset == null) {
+					parser.parse(in, baseURI);
 				}
-				catch (NullPointerException e) {
-					// Swallow NullPointerException that Apache HTTPClient is hiding behind a NotThreadSafe annotation
+				else {
+					parser.parse(new InputStreamReader(in, charset), baseURI);
 				}
+			} finally {
+				in.close();
 			}
 		}
-		catch (IOException e) {
-			throw new QueryEvaluationException(e);
-		}
-		finally {
-			queue.close();
-		}
-	}
-
-	@Override
-	public void run() {
-		try {
-			parser.setRDFHandler(this);
-			if (charset == null) {
-				parser.parse(in, baseURI);
-			}
-			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();
 		}
 	}
 
@@ -197,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();
 		}
 	}
 
@@ -226,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

@@ -7,9 +7,7 @@
  *******************************************************************************/
 package org.eclipse.rdf4j.query.resultio.helpers;
 
-import java.io.IOException;
 import java.io.InputStream;
-import java.lang.reflect.UndeclaredThrowableException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -43,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);
 	}
@@ -61,66 +61,52 @@ protected void handleClose()
 		throws QueryEvaluationException
 	{
 		try {
-			try {
-				super.handleClose();
-			}
-			finally {
-				try {
-					// After checking that we ourselves cannot possibly be generating an NPE ourselves, 
-					// attempt to close the input stream we were given
-					InputStream toClose = in;
-					if (toClose != null) {
-						toClose.close();
-					}
-				}
-				catch (NullPointerException e) {
-					// Swallow NullPointerException that Apache HTTPClient is hiding behind a NotThreadSafe annotation
-				}
-			}
-		}
-		catch (IOException e) {
-			throw new QueryEvaluationException(e);
+			super.handleClose();
 		}
 		finally {
-			queue.close();
+			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();
 		}
 	}
 
@@ -141,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();