You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@jena.apache.org by an...@apache.org on 2022/09/18 15:32:08 UTC
[jena] branch main updated: GH-1477: AsyncParserBuilder and improved resource management
This is an automated email from the ASF dual-hosted git repository.
andy pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/jena.git
The following commit(s) were added to refs/heads/main by this push:
new 96e379e2c4 GH-1477: AsyncParserBuilder and improved resource management
new d24b9c6eb2 Merge pull request #1478 from Aklakan/gh-1477
96e379e2c4 is described below
commit 96e379e2c40a8b43fffa611f072d72bb3a59bd57
Author: Claus Stadler <Ra...@googlemail.com>
AuthorDate: Fri Aug 12 17:37:22 2022 +0200
GH-1477: AsyncParserBuilder and improved resource management
---
.../org/apache/jena/riot/system/AsyncParser.java | 567 +++++++++++++++------
.../jena/riot/system/AsyncParserBuilder.java | 175 +++++++
.../org/apache/jena/riot/system/EltStreamRDF.java | 111 ++++
.../apache/jena/riot/system/EltStreamRDFType.java | 29 ++
.../apache/jena/riot/system/TestAsyncParser.java | 240 ++++++++-
.../jena/atlas/iterator/IteratorFlatMap.java | 4 +-
6 files changed, 953 insertions(+), 173 deletions(-)
diff --git a/jena-arq/src/main/java/org/apache/jena/riot/system/AsyncParser.java b/jena-arq/src/main/java/org/apache/jena/riot/system/AsyncParser.java
index 2b65b11e87..20ec9d9466 100644
--- a/jena-arq/src/main/java/org/apache/jena/riot/system/AsyncParser.java
+++ b/jena-arq/src/main/java/org/apache/jena/riot/system/AsyncParser.java
@@ -20,25 +20,30 @@ package org.apache.jena.riot.system;
import java.io.InputStream;
import java.util.ArrayList;
-import java.util.Iterator;
+import java.util.Arrays;
import java.util.List;
import java.util.Objects;
-import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;
-import org.apache.jena.atlas.iterator.Iter;
+import org.apache.jena.atlas.iterator.IteratorCloseable;
import org.apache.jena.atlas.iterator.IteratorSlotted;
import org.apache.jena.atlas.lib.InternalErrorException;
import org.apache.jena.atlas.logging.FmtLog;
import org.apache.jena.graph.Node;
import org.apache.jena.graph.Triple;
-import org.apache.jena.riot.*;
+import org.apache.jena.riot.Lang;
+import org.apache.jena.riot.RDFParser;
+import org.apache.jena.riot.RDFParserBuilder;
+import org.apache.jena.riot.RiotException;
+import org.apache.jena.riot.SysRIOT;
import org.apache.jena.riot.lang.RiotParsers;
import org.apache.jena.sparql.core.Quad;
+import org.apache.jena.util.iterator.ClosableIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -51,14 +56,31 @@ import org.slf4j.LoggerFactory;
* <p>
* There are overheads, so this is only beneficial in some situations. Delivery to
* the StreamRDF has an initial latency while the first batch of work is accumulated.
+ * Using the {@link AsyncParserBuilder} gives control over the chunk size such that initial latency
+ * can be reduced at the cost of possibly decreasing the overall throughput.
+ * <p>
+ * Closing the returned {@link ClosableIterator}s terminates the parsing
+ * process and closes the involved resources.
*/
public class AsyncParser {
- private static Logger LOG = LoggerFactory.getLogger(AsyncParser.class);
- private static int chunkSize = 100_000;
+ static final Logger LOG = LoggerFactory.getLogger(AsyncParser.class);
+ static final ErrorHandler dftErrorHandler = createDefaultErrorhandler(LOG);
+
+ static final int dftChunkSize = 100_000;
// There is no point letting the parser get a long way ahead.
- private static int queueSize = 10;
- private static List<EltStreamRDF> END = List.of();
+ static final int dftQueueSize = 10;
+ static final List<EltStreamRDF> END = List.of();
+
+ private static final StreamRDF alwaysFailingStreamRdf = new StreamToElements(elt -> { throw new RuntimeException(new InterruptedException()); });
+
+ /* Destination states for handling concurrent abort */
+ /** Normal operation: Accept incoming data */
+ private static final int RUNNING = 0;
+ /** Abort requested: The next chunk will trigger placing the poison on the queue and transition to STOPPED state */
+ private static final int ABORTING = 1;
+ /** Stopped (regardless whether due to normal or exceptional cause): Any further incoming data raises an exception */
+ private static final int STOPPED = 2;
private AsyncParser() {}
@@ -66,8 +88,8 @@ public class AsyncParser {
* Function that reads a file or GETs a URL, parses the content on a separate
* thread and sends the output to a StreamRDF on the callers thread.
*/
- public static void asyncParse(String filesOrURL, StreamRDF output) {
- asyncParse(List.of(filesOrURL), output);
+ public static void asyncParse(String fileOrURL, StreamRDF output) {
+ AsyncParser.of(fileOrURL).asyncParseSources(output);
}
/**
@@ -75,10 +97,7 @@ public class AsyncParser {
* and sends the output to a StreamRDF on the callers thread.
*/
public static void asyncParse(List<String> filesOrURLs, StreamRDF output) {
- Objects.requireNonNull(filesOrURLs);
- Objects.requireNonNull(output);
- LOG.debug("Parse: "+filesOrURLs);
- asyncParseSources(urlsToSource(filesOrURLs), output);
+ AsyncParser.ofLocations(filesOrURLs).asyncParseSources(output);
}
/**
@@ -86,10 +105,7 @@ public class AsyncParser {
* and sends the output to a StreamRDF on the callers thread.
*/
public static void asyncParse(InputStream input, Lang lang, String baseURI, StreamRDF output) {
- Objects.requireNonNull(input);
- Objects.requireNonNull(lang);
- Objects.requireNonNull(output);
- asyncParseSources(inputStreamToSource(input, lang, baseURI), output);
+ AsyncParser.of(input, lang, baseURI).asyncParseSources(output);
}
/**
@@ -97,52 +113,17 @@ public class AsyncParser {
* send the output to a StreamRDF on the callers thread.
*/
public static void asyncParseSources(List<RDFParserBuilder> sources, StreamRDF output) {
- // Async thread
- Logger LOG1 = LOG;
- // Receiver
- Logger LOG2 = LOG;
- BlockingQueue<List<EltStreamRDF>> queue = new ArrayBlockingQueue<>(queueSize);
- startParserThread(LOG1, sources, queue);
- receiver(LOG2, queue, output);
+ AsyncParser.ofSources(sources).asyncParseSources(output);
}
- private static List<RDFParserBuilder> urlsToSource(List<String> filesOrURLs) {
- return filesOrURLs.stream().map(uriOrFile-> RDFParser.source(uriOrFile)).collect(Collectors.toList());
- }
-
- private static List<RDFParserBuilder> inputStreamToSource(InputStream input, Lang lang, String baseURI) {
- return List.of(RDFParser.source(input).lang(lang));
- }
-
- private static Function<EltStreamRDF, Triple> elt2Triple = x-> {
- if ( x.exception != null )
- throw x.exception;
- if ( x.quad != null ) {
- Node g = x.quad.getGraph();
- if ( g == Quad.tripleInQuad || Quad.isDefaultGraph(g) )
- return x.quad.asTriple();
- return null;
- }
- return x.triple;
- };
-
- private static Function<EltStreamRDF, Quad> elt2Quad = x-> {
- if ( x.exception != null )
- throw x.exception;
- if ( x.triple != null )
- return Quad.create(Quad.defaultGraphIRI, x.triple);
- return x.quad;
- };
-
/** Pull parser - triples */
- public static Iterator<Triple> asyncParseTriples(String fileOrURL) {
- return asyncParseTriples(List.of(fileOrURL));
+ public static IteratorCloseable<Triple> asyncParseTriples(String fileOrURL) {
+ return AsyncParser.of(fileOrURL).asyncParseTriples();
}
/** Pull parser - triples */
- public static Iterator<Triple> asyncParseTriples(List<String> filesOrURLs) {
- Iterator<EltStreamRDF> source = asyncParseIterator(urlsToSource(filesOrURLs));
- return Iter.iter(source).map(elt2Triple).removeNulls();
+ public static IteratorCloseable<Triple> asyncParseTriples(List<String> filesOrURLs) {
+ return AsyncParser.ofLocations(filesOrURLs).asyncParseTriples();
}
/**
@@ -150,20 +131,18 @@ public class AsyncParser {
* <p>
* See also {@link RiotParsers#createIteratorNTriples}.
*/
- public static Iterator<Triple> asyncParseTriples(InputStream input, Lang lang, String baseURI) {
- Iterator<EltStreamRDF> source = asyncParseIterator(inputStreamToSource(input, lang, baseURI));
- return Iter.iter(source).map(elt2Triple).removeNulls();
+ public static IteratorCloseable<Triple> asyncParseTriples(InputStream input, Lang lang, String baseURI) {
+ return AsyncParser.of(input, lang, baseURI).asyncParseTriples();
}
/** Pull parser - quads */
- public static Iterator<Quad> asyncParseQuads(String fileOrURL) {
- return asyncParseQuads(List.of(fileOrURL));
+ public static IteratorCloseable<Quad> asyncParseQuads(String fileOrURL) {
+ return AsyncParser.of(fileOrURL).asyncParseQuads();
}
/** Pull parser - quads */
- public static Iterator<Quad> asyncParseQuads(List<String> filesOrURLs) {
- Iterator<EltStreamRDF> source = asyncParseIterator(urlsToSource(filesOrURLs));
- return Iter.iter(source).map(elt2Quad).removeNulls();
+ public static IteratorCloseable<Quad> asyncParseQuads(List<String> filesOrURLs) {
+ return AsyncParser.ofLocations(filesOrURLs).asyncParseQuads();
}
/**
@@ -171,29 +150,109 @@ public class AsyncParser {
* <p>
* See also {@link RiotParsers#createIteratorNQuads}.
*/
- public static Iterator<Quad> asyncParseQuads(InputStream input, Lang lang, String baseURI) {
- Iterator<EltStreamRDF> source = asyncParseIterator(inputStreamToSource(input, lang, baseURI));
- return Iter.iter(source).map(elt2Quad).removeNulls();
+ public static IteratorCloseable<Quad> asyncParseQuads(InputStream input, Lang lang, String baseURI) {
+ return AsyncParser.of(input, lang, baseURI).asyncParseQuads();
+ }
+
+ /** Create an {@link AsyncParserBuilder} from a file or URL. */
+ public static AsyncParserBuilder of(String fileOrURL) {
+ Objects.requireNonNull(fileOrURL);
+ return ofLocations(List.of(fileOrURL));
+ }
+
+ /** Create an {@link AsyncParserBuilder} from a set of files and/or URLs. */
+ public static AsyncParserBuilder ofLocations(List<String> filesOrURLs) {
+ Objects.requireNonNull(filesOrURLs);
+ LOG.debug("Parse: "+filesOrURLs);
+ return new AsyncParserBuilder(urlsToSource(filesOrURLs));
+ }
+
+ /** Create an {@link AsyncParserBuilder} from an input stream. */
+ public static AsyncParserBuilder of(InputStream input, Lang lang, String baseURI) {
+ Objects.requireNonNull(input);
+ Objects.requireNonNull(lang);
+ return ofSources(inputStreamToSource(input, lang, baseURI));
+ }
+
+ /** Create an {@link AsyncParserBuilder} from an {@link RDFParserBuilder}. */
+ public static AsyncParserBuilder of(RDFParserBuilder source) {
+ Objects.requireNonNull(source);
+ return ofSources(Arrays.asList(source));
+ }
+
+ /** Create an {@link AsyncParserBuilder} from a set of {@link RDFParserBuilder} instances. */
+ public static AsyncParserBuilder ofSources(List<RDFParserBuilder> sources) {
+ Objects.requireNonNull(sources);
+ return new AsyncParserBuilder(sources);
+ }
+
+ /*
+ * Internals
+ */
+
+ private static List<RDFParserBuilder> urlsToSource(List<String> filesOrURLs) {
+ return filesOrURLs.stream().map(
+ uriOrFile -> RDFParser.source(uriOrFile).errorHandler(dftErrorHandler))
+ .collect(Collectors.toList());
+ }
+
+ /** Create a source object from the given arguments that is suitable for use with
+ * {@link #asyncParseIterator(List)} which return*/
+ private static List<RDFParserBuilder> inputStreamToSource(InputStream input, Lang lang, String baseURI) {
+ return List.of(RDFParser.source(input).lang(lang).base(baseURI).errorHandler(dftErrorHandler));
}
- /** Pull parser */
- private static Iterator<EltStreamRDF> asyncParseIterator(List<RDFParserBuilder> sources) {
- BlockingQueue<List<EltStreamRDF>> queue = new ArrayBlockingQueue<>(queueSize);
- startParserThread(LOG, sources, queue);
- Iterator<List<EltStreamRDF>> blocks = blockingIterator(queue, x->x==END);
- Iterator<EltStreamRDF> elements = Iter.flatMap(blocks, x->x.iterator());
- return elements;
+ /**
+ * This method raises a caller-thread-side exception for a throwable returned by the parser thread.
+ * If that throwable is a RuntimeException then the caller side's stacktrace is added
+ * as a suppressed exception in order to retain the original exception type.
+ * Any other type of throwable raises a generic RuntimeException with it as the cause.
+ */
+ private static void raiseException(Throwable throwable) {
+ if (throwable instanceof RuntimeException) {
+ RuntimeException e = (RuntimeException)throwable;
+ e.addSuppressed(new RuntimeException("Encountered error element from parse thread"));
+ throw e;
+ }
+
+ throw new RuntimeException("Encountered error element from parse thread", throwable);
}
+ static Function<EltStreamRDF, Triple> elt2Triple = x-> {
+ if ( x.isException() ) {
+ raiseException(x.exception());
+ }
+ if ( x.isQuad() ) {
+ Quad quad = x.quad();
+ Node g = quad.getGraph();
+ if ( g == Quad.tripleInQuad || Quad.isDefaultGraph(g) )
+ return quad.asTriple();
+ return null;
+ }
+ return x.triple();
+ };
+
+ static Function<EltStreamRDF, Quad> elt2Quad = x-> {
+ if ( x.isException() ) {
+ raiseException(x.exception());
+ }
+ if ( x.isTriple() )
+ return Quad.create(Quad.defaultGraphIRI, x.triple());
+ return x.quad();
+ };
+
// Iterator that stops when an end marker is seen.
- private static <X> Iterator<X> blockingIterator(BlockingQueue<X> queue, Predicate<X> endTest) {
- return new IteratorSlotted<X>() {
+ static <X> IteratorCloseable<X> blockingIterator(Runnable closeAction, BlockingQueue<X> queue, Predicate<X> endTest) {
+ return new IteratorSlotted<>() {
boolean ended = false;
@Override
protected X moveToNext() {
try {
- X x = queue.take();
+ X x = null;
+ if (!ended) {
+ x = queue.take();
+ }
if ( endTest.test(x) ) {
ended = true;
return null;
@@ -209,22 +268,16 @@ public class AsyncParser {
protected boolean hasMore() {
return !ended;
}
- };
- }
- private static void startParserThread(Logger LOG1, List<RDFParserBuilder> parserBuilders, BlockingQueue<List<EltStreamRDF>> queue) {
- // -- Parser thread setup
- Consumer<List<EltStreamRDF>> destination = batch->{
- try {
- queue.put(batch);
- } catch (InterruptedException ex) {
- FmtLog.error(LOG, ex, "Error: %s", ex.getMessage());
+ @Override
+ protected void closeIterator() {
+ closeAction.run();
}
};
- EltStreamBatcher batcher = new EltStreamBatcher(destination, chunkSize);
- StreamRDF generatorStream = new StreamToElements(batcher);
+ }
- ErrorHandler errhandler = new ErrorHandler() {
+ static ErrorHandler createDefaultErrorhandler(Logger LOG1) {
+ return new ErrorHandler() {
@Override
public void warning(String message, long line, long col)
{ LOG1.warn(SysRIOT.fmtMessage(message, line, col)); }
@@ -239,74 +292,256 @@ public class AsyncParser {
throw new RiotException(SysRIOT.fmtMessage(message, line, col)) ;
}
};
+ }
- // Parser thread
- Runnable task = ()->{
- batcher.startBatching();
- if ( LOG1.isDebugEnabled() )
- LOG1.debug("Start parsing");
+ /** This class's purpose is to run a set of RDF parsers and place their generated elements on a queue.
+ * Thereby errors and abort requests are handled. */
+ private static class Task implements Runnable {
+ private Logger logger;
+ private List<RDFParserBuilder> sources;
+ private BlockingQueue<List<EltStreamRDF>> queue;
+ private int chunkSize;
+ private Predicate<EltStreamRDF> prematureDispatch;
+
+ // Destination resources are initialized upon calling run();
+ private EltStreamBatcher<EltStreamRDF> batcher;
+ private StreamRDF generatorStreamRdf;
+ private AtomicInteger destinationState = new AtomicInteger(RUNNING);
+ private boolean errorEncountered = false;
+
+ public Task(List<RDFParserBuilder> sources, BlockingQueue<List<EltStreamRDF>> queue, int chunkSize, Predicate<EltStreamRDF> prematureDispatch, Logger logger) {
+ this.sources = sources;
+ this.queue = queue;
+ this.chunkSize = chunkSize;
+ this.prematureDispatch = prematureDispatch;
+ this.logger = logger;
+ }
+
+ @Override
+ public void run() {
+ initDestination();
try {
- for ( RDFParserBuilder parser : parserBuilders ) {
- parser.errorHandler(errhandler).parse(generatorStream);
+ start();
+ int n = sources.size();
+ for (int i = 0; i < n; ++i) {
+ RDFParserBuilder parser = sources.get(i);
+ parse(parser);
+ }
+ } finally {
+ finish();
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("Finish parsing");
+ }
+ }
+
+ /** Sets the destination into ABORTING state (if it was running).
+ * This triggers transition into ABORT state on the next chunk. */
+ public void abort() {
+ destinationState.compareAndSet(RUNNING, ABORTING);
+ }
+
+ private void initDestination() {
+ // -- Parser thread setup
+ Consumer<List<EltStreamRDF>> destination = batch -> {
+ if (destinationState.get() == RUNNING) {
+ try {
+ queue.put(batch);
+ } catch (InterruptedException ex) {
+ // After interrupt we may be in ABORTING state - therefore
+ // check whether to transition into ABORT state
+ handleAbortingState();
+ // FmtLog.error(LOG, ex, "Error: %s", ex.getMessage());
+ throw new RuntimeException(ex);
+ }
+ } else {
+ handleAbortingState();
+ throw new RuntimeException(new InterruptedException());
}
+ };
+
+ this.batcher = new EltStreamBatcher<>(destination, END, chunkSize);
+ Consumer<EltStreamRDF> eltSink = batcher;
+
+ // Take care of a custom dispatch policy
+ if (prematureDispatch != null) {
+ eltSink = elt -> {
+ boolean dispatchImmediately = prematureDispatch.test(elt);
+ batcher.accept(elt);
+ if (dispatchImmediately) {
+ batcher.flush();
+ }
+ };
+ }
+
+ this.generatorStreamRdf = new StreamToElements(eltSink);
+ }
+
+ private void start() {
+ batcher.startBatching();
+ if (logger.isDebugEnabled()) {
+ logger.debug("Start parsing");
+ }
+ }
+
+ private void parse(RDFParserBuilder parser) {
+ try {
+ // If an error occured then all parser are invoked anyway because any
+ // resources they own need yet to be closed.
+ // At this point, however, the parser's sink will always fail and
+ // any further errors will be suppressed.
+ StreamRDF sink = errorEncountered
+ ? alwaysFailingStreamRdf
+ : generatorStreamRdf;
+ parser.parse(sink);
} catch (RuntimeException ex) {
- // Parse error.
- EltStreamRDF elt = new EltStreamRDF();
- elt.exception = ex;
- batcher.accept(elt);
- } catch (Throwable cause) {
- // Very bad!
- EltStreamRDF elt = new EltStreamRDF();
- elt.exception = new RuntimeException(cause);
- batcher.accept(elt);
+ Throwable cause = ex.getCause();
+ if (errorEncountered) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Suppressed exception", ex);
+ }
+ } else {
+ if (cause instanceof InterruptedException) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Parsing was interrupted");
+ }
+ } else {
+ // Parse error.
+ EltStreamRDF elt = EltStreamRDF.exception(ex);
+ batcher.accept(elt);
+ }
+ errorEncountered = true;
+ }
+ } catch (Throwable throwable) {
+ if (errorEncountered) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Suppressed exception", throwable);
+ }
+ } else {
+ // Very bad! Wrapping with runtime exception to track this location
+ EltStreamRDF elt = EltStreamRDF.exception(new RuntimeException(throwable));
+ batcher.accept(elt);
+ errorEncountered = true;
+ }
}
- batcher.finishBatching();
- if ( LOG1.isDebugEnabled() )
- LOG1.debug("Finish parsing");
- };
+ }
+
+ private void finish() {
+ try {
+ // If we are already in ABORTED state then finishBatching will do nothing but raise
+ // an exception
+ // If we are in ABORTING state we still need to dispatch another batch in order to transition into ABORT state
+ if (destinationState.get() != STOPPED) {
+ batcher.finishBatching();
+ }
+ } catch(Throwable ex) {
+ if (errorEncountered) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Suppressed exception", ex);
+ }
+ } else {
+ Throwable cause = ex.getCause();
+ if (cause instanceof InterruptedException) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Parsing was interrupted");
+ }
+ } else {
+ // Rethrow if something else went wrong
+ throw new RuntimeException(ex);
+ }
+ }
+ } finally {
+ // We are done; dispatching any further chunks is now an error and raises an exception
+ destinationState.set(STOPPED);
+ }
+ }
+
+ /** Check */
+ private void handleAbortingState() {
+ if (destinationState.compareAndSet(ABORTING, STOPPED)) {
+ // If we are transitioning into ABORT state then force putting
+ // the poison on the queue - regardless of further interruptions!
+ while (true) {
+ try {
+ queue.clear();
+ queue.put(END);
+ break;
+ } catch (InterruptedException e) {
+ continue;
+ }
+ }
+ }
+ }
+ }
+
+ /** Returns a runnable for stopping the parse process */
+ static Runnable startParserThread(
+ Logger logger, List<RDFParserBuilder> sources,
+ BlockingQueue<List<EltStreamRDF>> queue,
+ int chunkSize,
+ Predicate<EltStreamRDF> prematureDispatch,
+ boolean daemonMode) {
+
+ Task task = new Task(sources, queue, chunkSize, prematureDispatch, logger);
// Ensures runnable has started.
- //ThreadLib.async(task);
- // Does not ensure runnable has started.
- Thread th = new Thread(task, "AsyncParser");
- th.setDaemon(true);
- th.start();
+ // ThreadLib.async(task) does not ensure runnable has started.
+ Thread parserThread = new Thread(task, "AsyncParser");
+ parserThread.setDaemon(daemonMode);
+ parserThread.start();
+ return () -> {
+ task.abort();
+
+ // Interrupt the parsing thread
+ // Note that InputStreams and ByteChannels may close themselves
+ // when interrupted while reading
+ parserThread.interrupt();
+
+ try {
+ // Wait for the thread to terminate so that all is clean when we return from close()
+ // Note: Some AsyncParser unit tests assert that the number of threads
+ // before and after parsing match up (within tolerance) - so joining is essential.
+ parserThread.join();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ };
}
private static void dispatch(EltStreamRDF elt, StreamRDF stream) {
- if ( elt.triple != null )
- stream.triple(elt.triple);
- else if ( elt.quad != null )
- stream.quad(elt.quad);
- else if ( elt.prefix != null )
- stream.prefix( elt.prefix, elt.iri);
- else if ( elt.iri != null )
- stream.base(elt.iri);
- else if ( elt.exception != null )
- throw elt.exception;
- else
- throw new InternalErrorException("Bad EltStreamRDF");
+ switch (elt.getType()) {
+ case TRIPLE: stream.triple(elt.triple()); break;
+ case QUAD: stream.quad(elt.quad()); break;
+ case PREFIX: stream.prefix(elt.prefix(), elt.iri()); break;
+ case BASE: stream.base(elt.iri()); break;
+ case EXCEPTION: raiseException(elt.exception()); break;
+ default: throw new InternalErrorException("Bad EltStreamRDF");
+ }
}
/** Receiver. Take chunks off the queue and send to the output StreamRDF. */
- private static void receiver(Logger LOG2, BlockingQueue<List<EltStreamRDF>> queue, StreamRDF output) {
+ static void receiver(Runnable closeAction, Logger LOG2, BlockingQueue<List<EltStreamRDF>> queue, StreamRDF output) {
// -- Receiver thread
int count = 0;
// Receive.
- for(;;) {
- try {
+ try {
+ for(;;) {
List<EltStreamRDF> batch = queue.take();
if ( batch == END ) {
FmtLog.debug(LOG2, "Receive: END (%,d)", count);
break;
}
- //List<EltStreamRDF> batch = queue.poll(1, TimeUnit.SECONDS);
count += batch.size();
if ( LOG.isDebugEnabled() )
FmtLog.debug(LOG2, "Receive: Batch : %,d (%,d)", batch.size(), count);
dispatch(batch, output);
- } catch (InterruptedException e) {
- FmtLog.error(LOG2, e, "Interrupted");
}
+ } catch (InterruptedException e) {
+ FmtLog.error(LOG2, e, "Interrupted");
+ } finally {
+ // The close action not only stops the parser but also waits
+ // for its thread to finish
+ closeAction.run();
}
}
@@ -315,17 +550,8 @@ public class AsyncParser {
dispatch(elt, stream);
}
- /** An item of a StreamRDF, including exceptions. */
- private static class EltStreamRDF {
- Triple triple = null;
- Quad quad = null;
- String prefix = null; // Null implies "base".
- String iri = null;
- RuntimeException exception = null;
- }
-
/** Convert a Stream into EltStreamRDF */
- private static class StreamToElements implements StreamRDF {
+ private static class StreamToElements implements StreamRDF {
private final Consumer<EltStreamRDF> destination;
@@ -334,38 +560,32 @@ public class AsyncParser {
}
@Override
- public void start() {}
+ public void start() { /* nothing to do */ }
@Override
- public void finish() {}
+ public void finish() { /* nothing to do */ }
@Override
public void triple(Triple triple) {
- EltStreamRDF elt = new EltStreamRDF();
- elt.triple = triple;
+ EltStreamRDF elt = EltStreamRDF.triple(triple);
deliver(elt);
}
@Override
public void quad(Quad quad) {
- EltStreamRDF elt = new EltStreamRDF();
- elt.quad = quad;
+ EltStreamRDF elt = EltStreamRDF.quad(quad);
deliver(elt);
}
@Override
public void base(String base) {
- EltStreamRDF elt = new EltStreamRDF();
- //elt.prefix = null;
- elt.iri = base;
+ EltStreamRDF elt = EltStreamRDF.base(base);
deliver(elt);
}
@Override
public void prefix(String prefix, String iri) {
- EltStreamRDF elt = new EltStreamRDF();
- elt.prefix = prefix;
- elt.iri = iri;
+ EltStreamRDF elt = EltStreamRDF.prefix(prefix, iri);
deliver(elt);
}
@@ -375,34 +595,44 @@ public class AsyncParser {
}
/** Batch items from a StreamRDF */
- private static class EltStreamBatcher implements Consumer<EltStreamRDF> {
+ private static class EltStreamBatcher<T> implements Consumer<T> {
private final int batchSize;
- private List<EltStreamRDF> elements = null;
- private final Consumer<List<EltStreamRDF>> batchDestination;
+ private List<T> elements = null;
+ private final Consumer<List<T>> batchDestination;
+ private final List<T> endMarker;
- public EltStreamBatcher(Consumer<List<EltStreamRDF>> batchDestination, int batchSize) {
+ public EltStreamBatcher(Consumer<List<T>> batchDestination, List<T> endMarker, int batchSize) {
this.batchDestination = batchDestination;
this.batchSize = batchSize;
+ this.endMarker = endMarker;
}
- public void startBatching() {}
+ public void startBatching() { /* nothing to do */ }
- public void finishBatching() {
- // Flush.
- if ( elements != null ) {
+ /** Immediately dispatch any available data. */
+ public void flush() {
+ if (elements != null) {
dispatch(elements);
elements = null;
}
- dispatch(END);
+ }
+
+ public void finishBatching() {
+ try {
+ flush();
+ } finally {
+ dispatch(endMarker);
+ }
}
private <X> boolean isEmpty(List<X> list) {
return list == null || list.isEmpty();
}
+ /** The added element is included in the dispatched batch when triggered */
@Override
- public void accept(EltStreamRDF elt) {
+ public void accept(T elt) {
if ( elements == null )
elements = allocChunk();
elements.add(elt);
@@ -418,14 +648,13 @@ public class AsyncParser {
}
private int count = 0 ;
- private void dispatch(List<EltStreamRDF> batch) {
+ private void dispatch(List<T> batch) {
count += batch.size();
batchDestination.accept(batch);
}
- private List<EltStreamRDF> allocChunk() {
+ private List<T> allocChunk() {
return new ArrayList<>(batchSize);
}
}
}
-
diff --git a/jena-arq/src/main/java/org/apache/jena/riot/system/AsyncParserBuilder.java b/jena-arq/src/main/java/org/apache/jena/riot/system/AsyncParserBuilder.java
new file mode 100644
index 0000000000..1da4f27c45
--- /dev/null
+++ b/jena-arq/src/main/java/org/apache/jena/riot/system/AsyncParserBuilder.java
@@ -0,0 +1,175 @@
+/*
+ * 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 org.apache.jena.riot.system;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.stream.Stream;
+
+import org.apache.jena.atlas.iterator.Iter;
+import org.apache.jena.atlas.iterator.IteratorCloseable;
+import org.apache.jena.graph.Triple;
+import org.apache.jena.riot.RDFParserBuilder;
+import org.apache.jena.sparql.core.Quad;
+import org.slf4j.Logger;
+
+public class AsyncParserBuilder {
+ /* These following two attributes only exist for javadoc references */
+ private static final int dftChunkSize = AsyncParser.dftChunkSize;
+ private static final int dftQueueSize = AsyncParser.dftQueueSize;
+
+ private int chunkSize;
+ private int queueSize;
+ private boolean daemonMode;
+ private Predicate<EltStreamRDF> prematureDispatch;
+ private List<RDFParserBuilder> sources;
+
+ public AsyncParserBuilder() {
+ this.chunkSize = dftChunkSize;
+ this.queueSize = dftQueueSize;
+ this.daemonMode = true;
+ this.prematureDispatch = null;
+ this.sources = Collections.emptyList();
+ }
+
+ public AsyncParserBuilder(List<RDFParserBuilder> sources) {
+ this();
+ this.sources = List.copyOf(sources);
+ }
+
+ public int getChunkSize() {
+ return chunkSize;
+ }
+
+ /** The chunk size controls the maxmimum number of elements the parser thread transfers at once
+ * to the client iterator's queue. The default value ({@value #dftChunkSize})
+ * is optimized for throughput however it introduces a delay until the first item arrives.
+ * Data can be dispatched prematurely by setting {@link #setPrematureDispatch(Predicate)}.
+ */
+ public AsyncParserBuilder setChunkSize(int chunkSize) {
+ this.chunkSize = chunkSize;
+ return this;
+ }
+
+ public int getQueueSize() {
+ return queueSize;
+ }
+
+ /**
+ * The queue size controls the number of chunks the parser thread can read ahead.
+ * A full queue blocks that thread.
+ * The default value is {@value #dftQueueSize}.
+ */
+ public AsyncParserBuilder setQueueSize(int queueSize) {
+ this.queueSize = queueSize;
+ return this;
+ }
+
+ public boolean isDaemonMode() {
+ return daemonMode;
+ }
+
+ /** Whether the parser thread should run as a daemon thread.
+ * A JVM will terminate as soon as there are no more non-daemon threads. */
+ public AsyncParserBuilder setDaemonMode(boolean daemonMode) {
+ this.daemonMode = daemonMode;
+ return this;
+ }
+
+ /** Set a custom dispatch (flush) policy: When the predicate returns true for an element then that element and
+ * any gathered data is dispatched immediately. Before dispatch a check is made whether parsing has been
+ * (concurrently) aborted. */
+ public AsyncParserBuilder setPrematureDispatch(Predicate<EltStreamRDF> prematureDispatch) {
+ this.prematureDispatch = prematureDispatch;
+ return this;
+ }
+
+ public Predicate<EltStreamRDF> getPrematureDispatch() {
+ return prematureDispatch;
+ }
+
+ public List<RDFParserBuilder> getSources() {
+ return sources;
+ }
+
+ public AsyncParserBuilder setSources(List<RDFParserBuilder> sources) {
+ this.sources = List.copyOf(sources);
+ return this;
+ }
+
+ /** Run an operation on all RDFParserBuilder sources in this builder.
+ * Allows for e.g. configuring prefixes, the error handler or the label-to-node strategy */
+ public AsyncParserBuilder mutateSources(Consumer<RDFParserBuilder> mutator) {
+ for (RDFParserBuilder source : sources) {
+ mutator.accept(source);
+ }
+ return this;
+ }
+
+ private IteratorCloseable<EltStreamRDF> asyncParseElements() {
+ Objects.requireNonNull(sources);
+
+ BlockingQueue<List<EltStreamRDF>> queue = new ArrayBlockingQueue<>(queueSize);
+ Runnable closeAction = AsyncParser.startParserThread(AsyncParser.LOG, sources, queue, chunkSize, prematureDispatch, daemonMode);
+ IteratorCloseable<List<EltStreamRDF>> blocks = AsyncParser.blockingIterator(closeAction, queue, x -> x == AsyncParser.END);
+
+ IteratorCloseable<EltStreamRDF> elements = (IteratorCloseable<EltStreamRDF>)Iter.flatMap(blocks, x->x.iterator());
+ return elements;
+ }
+
+ public IteratorCloseable<Triple> asyncParseTriples() {
+ return Iter.iter(asyncParseElements()).map(AsyncParser.elt2Triple).removeNulls();
+ }
+
+ public IteratorCloseable<Quad> asyncParseQuads() {
+ return Iter.iter(asyncParseElements()).map(AsyncParser.elt2Quad).removeNulls();
+ }
+
+ public Stream<EltStreamRDF> streamElements() {
+ return Iter.asStream(asyncParseElements());
+ }
+
+ public Stream<Triple> streamTriples() {
+ return Iter.asStream(asyncParseTriples());
+ }
+
+ public Stream<Quad> streamQuads() {
+ return Iter.asStream(asyncParseQuads());
+ }
+
+ /** Calling the returned runnable stops parsing.
+ * The sink will see no further data. */
+ public Runnable asyncParseSources(StreamRDF output) {
+ // Async thread
+ Logger LOG1 = AsyncParser.LOG;
+ // Receiver
+ Logger LOG2 = AsyncParser.LOG;
+
+ BlockingQueue<List<EltStreamRDF>> queue = new ArrayBlockingQueue<>(queueSize);
+ Runnable closeAction = AsyncParser.startParserThread(LOG1, sources, queue, chunkSize, prematureDispatch, daemonMode);
+ AsyncParser.receiver(closeAction, LOG2, queue, output);
+
+ return closeAction;
+ }
+}
diff --git a/jena-arq/src/main/java/org/apache/jena/riot/system/EltStreamRDF.java b/jena-arq/src/main/java/org/apache/jena/riot/system/EltStreamRDF.java
new file mode 100644
index 0000000000..895d2286d1
--- /dev/null
+++ b/jena-arq/src/main/java/org/apache/jena/riot/system/EltStreamRDF.java
@@ -0,0 +1,111 @@
+/*
+ * 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 org.apache.jena.riot.system;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+import org.apache.jena.graph.Triple;
+import org.apache.jena.riot.out.NodeFmtLib;
+import org.apache.jena.sparql.core.Quad;
+
+/** An item of a StreamRDF, including exceptions. */
+public class EltStreamRDF
+ implements Serializable
+{
+ private static final long serialVersionUID = 1L;
+
+ private final EltStreamRDFType type;
+ private final Triple triple;
+ private final Quad quad;
+ private final String prefix; // Null implies "base".
+ private final String iri;
+ private final Throwable exception;
+
+ /* Prefer static constructors in EltStreamRDF */
+
+ EltStreamRDF(Triple triple) { this(EltStreamRDFType.TRIPLE, triple, null, null, null, null); }
+ EltStreamRDF(Quad quad) { this(EltStreamRDFType.QUAD, null, quad, null, null, null); }
+ EltStreamRDF(String prefix, String iri) { this(EltStreamRDFType.PREFIX, null, null, prefix, iri, null); }
+ EltStreamRDF(String iri) { this(EltStreamRDFType.BASE, null, null, null, iri, null); }
+ EltStreamRDF(Throwable exception) { this(EltStreamRDFType.EXCEPTION, null, null, null, null, exception); }
+
+ EltStreamRDF(EltStreamRDFType eltType, Triple triple, Quad quad, String prefix, String iri, Throwable exception) {
+ this.type = eltType;
+ this.triple = triple;
+ this.quad = quad;
+ this.prefix = prefix;
+ this.iri = iri;
+ this.exception = exception;
+ }
+
+ public boolean isTriple() { return EltStreamRDFType.TRIPLE.equals(type); }
+ public Triple triple() { return triple; }
+ public boolean isQuad() { return EltStreamRDFType.QUAD.equals(type); }
+ public Quad quad() { return quad; }
+ public boolean isPrefix() { return EltStreamRDFType.PREFIX.equals(type); }
+ public String prefix() { return prefix; }
+ public boolean isBase() { return EltStreamRDFType.BASE.equals(type); }
+ public String iri() { return iri; }
+ public boolean isException() { return EltStreamRDFType.EXCEPTION.equals(type); }
+ public Throwable exception() { return exception; }
+
+ public EltStreamRDFType getType() {
+ return type;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(type, triple, quad, prefix, iri, exception);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj)
+ return true;
+ if (obj == null)
+ return false;
+ if (getClass() != obj.getClass())
+ return false;
+ EltStreamRDF other = (EltStreamRDF) obj;
+ return Objects.equals(type, other.type) && Objects.equals(triple, other.triple) &&
+ Objects.equals(quad, other.quad) && Objects.equals(prefix, other.prefix) &&
+ Objects.equals(iri, other.iri) && Objects.equals(exception, other.exception);
+ }
+
+ @Override
+ public String toString() {
+ String str;
+ switch (getType()) {
+ case TRIPLE: str = "triple(" + NodeFmtLib.str(triple()) + ")"; break;
+ case QUAD: str = "quad(" + NodeFmtLib.str(quad()) + ")"; break;
+ case PREFIX: str = "prefix(" + Objects.toString(prefix() + ", " + iri()) + ")"; break;
+ case BASE: str = "base(" + Objects.toString(iri()) + ")"; break;
+ case EXCEPTION: str = "exception(" + Objects.toString(exception()) + ")"; break;
+ default: str = "unknown"; break;
+ }
+ return str;
+ }
+
+ public static EltStreamRDF triple(Triple triple) { return new EltStreamRDF(Objects.requireNonNull(triple)); }
+ public static EltStreamRDF quad(Quad quad) { return new EltStreamRDF(Objects.requireNonNull(quad)); }
+ public static EltStreamRDF base(String iri) { return new EltStreamRDF(Objects.requireNonNull(iri)); }
+ public static EltStreamRDF prefix(String prefix, String iri) { return new EltStreamRDF(Objects.requireNonNull(prefix), Objects.requireNonNull(iri)); }
+ public static EltStreamRDF exception(Throwable exception) { return new EltStreamRDF(Objects.requireNonNull(exception)); }
+}
diff --git a/jena-arq/src/main/java/org/apache/jena/riot/system/EltStreamRDFType.java b/jena-arq/src/main/java/org/apache/jena/riot/system/EltStreamRDFType.java
new file mode 100644
index 0000000000..a60570f846
--- /dev/null
+++ b/jena-arq/src/main/java/org/apache/jena/riot/system/EltStreamRDFType.java
@@ -0,0 +1,29 @@
+/*
+ * 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 org.apache.jena.riot.system;
+
+/** Payload types of {@link EltStreamRDF} elements */
+public enum EltStreamRDFType {
+ UNKNOWN,
+ TRIPLE,
+ QUAD,
+ PREFIX,
+ BASE,
+ EXCEPTION
+}
diff --git a/jena-arq/src/test/java/org/apache/jena/riot/system/TestAsyncParser.java b/jena-arq/src/test/java/org/apache/jena/riot/system/TestAsyncParser.java
index 14cb36756b..930ffb008e 100644
--- a/jena-arq/src/test/java/org/apache/jena/riot/system/TestAsyncParser.java
+++ b/jena-arq/src/test/java/org/apache/jena/riot/system/TestAsyncParser.java
@@ -22,14 +22,40 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.lang.management.ManagementFactory;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.jena.atlas.iterator.IteratorCloseable;
+import org.apache.jena.ext.com.google.common.base.Preconditions;
+import org.apache.jena.ext.com.google.common.base.Stopwatch;
import org.apache.jena.graph.Graph;
import org.apache.jena.graph.Triple;
-import org.apache.jena.riot.*;
+import org.apache.jena.riot.Lang;
+import org.apache.jena.riot.RDFFormat;
+import org.apache.jena.riot.RDFParser;
+import org.apache.jena.riot.RDFParserBuilder;
+import org.apache.jena.riot.RiotException;
+import org.apache.jena.riot.RiotNotFoundException;
import org.apache.jena.sparql.graph.GraphFactory;
import org.apache.jena.sparql.util.IsoMatcher;
+import org.junit.Assert;
import org.junit.Test;
public class TestAsyncParser {
@@ -91,4 +117,216 @@ public class TestAsyncParser {
assertTrue( IsoMatcher.isomorphic(graph1, graph2));
}
+ /** Tests the parser on an infinite amount of data with a failing client.
+ * This should stop the parser. */
+ @Test
+ public void failingSink() {
+ int beforeThreadCount = ManagementFactory.getThreadMXBean().getThreadCount();
+ for (int i = 0; i < 100; ++i) {
+ try (InputStream in = openInfiniteNtStream();
+ OutputStream out = new FailingOutputStream()) {
+ StreamRDF sink = StreamRDFWriter.getWriterStream(out, RDFFormat.NT);
+ sink.start();
+ // Use small read ahead amount due to the infinite input and always failing sink
+ AsyncParser.of(in, Lang.NT, null).setChunkSize(10).setQueueSize(3).asyncParseSources(sink);
+ sink.finish();
+ } catch (Exception e) {
+ // Expected to fail
+ continue;
+ }
+ throw new RuntimeException("Parsing unexpectedly succeeded");
+ }
+ int afterThreadCount = ManagementFactory.getThreadMXBean().getThreadCount();
+ int threadCountDifference = Math.abs(afterThreadCount - beforeThreadCount);
+ int maxAllowedThreadCountDifference = 5;
+
+ Assert.assertTrue("Cancelling RDF parsing resulted in too many dangling threads ("
+ + threadCountDifference + ")",
+ threadCountDifference <= maxAllowedThreadCountDifference);
+ }
+
+ /** Tests to ensure threads are not piling up when repeatedly canceling parsers */
+ @Test
+ public void repeatedParsingCancellation_1() throws Exception {
+ int beforeThreadCount = ManagementFactory.getThreadMXBean().getThreadCount();
+ for (int i = 0; i < 1000; ++i) {
+ IteratorCloseable<Triple> iter = null;
+ try {
+ iter = AsyncParser.of(DIR+"data.ttl").setDaemonMode(false).asyncParseTriples();
+
+ // Ensure parsing is triggered
+ iter.hasNext();
+ } finally {
+ if (iter != null) {
+ iter.close();
+ }
+ }
+ }
+ int afterThreadCount = ManagementFactory.getThreadMXBean().getThreadCount();
+ int threadCountDifference = Math.abs(afterThreadCount - beforeThreadCount);
+
+ // Give some room in how much the number of threads may differ
+ // The main point is that the acceptable margin is significantly lower than
+ // the number of started parser threads
+ int maxAllowedThreadCountDifference = 5;
+
+ Assert.assertTrue("Cancelling RDF parsing resulted in too many dangling threads ("
+ + threadCountDifference + ")",
+ threadCountDifference <= maxAllowedThreadCountDifference);
+ }
+
+ /** A test that checks that only a limited number of bytes is read when using a small chunk size*/
+ @Test
+ public void lowLatencyParse() {
+ long expectedLimit = 10;
+ RepeatingReadableByteChannel channel = openInfiniteNtChannel();
+ try (Stream<Triple> s = AsyncParser.of(Channels.newInputStream(channel), Lang.TURTLE, null)
+ .setChunkSize(100).streamTriples().limit(expectedLimit)) {
+ // s.forEach(t -> System.out.println("Triple: " + t));
+ long actualLimit = s.count();
+ Assert.assertEquals(expectedLimit, actualLimit);
+ }
+
+ long pos = channel.position();
+ // The exact number of bytes consumed from the channel when writing the test was 10050 bytes
+ // However to give room for implementation changes the value tested against here is about twice as large
+ Assert.assertTrue("Too many bytes consumed from input stream (" + pos + ")", pos < 20000);
+ }
+
+ /** This test first creates some 'good' data for reference and then appends some 'bad' data in order
+ * to provoke a parse error.
+ * Parsing the bad data should then yield all the good data followed by an error event.
+ * Conversely, errors must not cause loss of any events. */
+ @Test
+ public void testNoEventsAreLost() {
+ // Generate some "good" sample data - good means: no parse errors
+ StringBuilder sb = new StringBuilder();
+ sb.append("PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>");
+ for (int i = 0; i < 4; ++i) {
+ sb.append("<urn:example:s> rdfs:label \"item # " + i + "\" .\n");
+ }
+ String goodInputStr = sb.toString();
+
+ List<EltStreamRDF> expected;
+ try (Stream<EltStreamRDF> s = AsyncParser.of(new ByteArrayInputStream(goodInputStr.getBytes()), Lang.TURTLE, null).streamElements()) {
+ expected = s.collect(Collectors.toList());
+ }
+
+ // Validate parser on the good data
+ int expectedGoodEventCount = 5;
+ Assert.assertEquals(expected.size(), expectedGoodEventCount);
+
+ // Append some bad data
+ String badInputStr = sb.append("parse error").toString();
+
+ // Now test that we get back all the good data plus the error event
+ for (int chunkSize = 1; chunkSize < 10; ++chunkSize) {
+ try (Stream<EltStreamRDF> s = AsyncParser.of(new ByteArrayInputStream(badInputStr.getBytes()), Lang.TURTLE, null)
+ .setChunkSize(chunkSize)
+ .streamElements()) {
+ List<EltStreamRDF> actual = s.collect(Collectors.toList());
+ Assert.assertEquals(expectedGoodEventCount + 1, actual.size());
+ Assert.assertEquals(expected, actual.subList(0, expectedGoodEventCount));
+ Assert.assertTrue(actual.get(actual.size() - 1).isException());
+ }
+ }
+ }
+
+ /** Test an AsyncParser setup with a custom dispatch (flush) policy */
+ @Test
+ public void testPrematureDispatch() {
+ final int expectedSize = 10;
+
+ AtomicInteger counter = new AtomicInteger(1);
+ String expectedErrorMsg = "Forced abort";
+ List<EltStreamRDF> actual;
+ try (Stream<EltStreamRDF> stream = AsyncParser.of(openInfiniteNtStream(), Lang.TURTLE, null)
+ .setPrematureDispatch(elt -> {
+ if (counter.getAndIncrement() < expectedSize) {
+ return true;
+ }
+ throw new RuntimeException(expectedErrorMsg);
+ })
+ .streamElements()
+ .limit(1000)) { // The limit is just a safety net to prevent infinite parsing in case of malfunction
+ actual = stream.collect(Collectors.toList());
+ }
+
+ // Check that only the expected number of elements were dispatched
+ Assert.assertEquals(expectedSize, actual.size());
+
+ // The payload of the last element must be an exception, all others must be triples
+ for (int i = 0; i < expectedSize; ++i) {
+ boolean isLastItem = i + 1 == expectedSize;
+ EltStreamRDF elt = actual.get(i);
+ if (isLastItem) {
+ Assert.assertTrue("Last element expected to be an exception", elt.isException());
+ Assert.assertEquals(expectedErrorMsg, elt.exception().getMessage());
+ } else {
+ Assert.assertTrue("Non-last element expected to be a triple", elt.isTriple());
+ }
+ }
+
+ }
+
+ /** An infinite stream of ntriple data */
+ private static InputStream openInfiniteNtStream() {
+ return Channels.newInputStream(openInfiniteNtChannel());
+ }
+
+ private static RepeatingReadableByteChannel openInfiniteNtChannel() {
+ return new RepeatingReadableByteChannel(
+ "<urn:example:s> <urn:example:p> <urn:example:o> .\n".getBytes(StandardCharsets.UTF_8));
+ }
+
+ /** An always failing output stream */
+ private static class FailingOutputStream
+ extends OutputStream
+ {
+ @Override
+ public void write(int b) throws IOException {
+ throw new IOException("Mocked IO Error");
+ }
+ }
+
+ /** A byte channel that infinitely repeats a given chunk of data */
+ private static class RepeatingReadableByteChannel
+ implements ReadableByteChannel
+ {
+ protected byte[] data;
+ protected boolean isOpen;
+ protected long pos;
+
+ public RepeatingReadableByteChannel(byte[] data) {
+ this(data, 0l);
+ }
+
+ public RepeatingReadableByteChannel(byte[] data, long pos) {
+ super();
+ Objects.requireNonNull(data);
+ Preconditions.checkArgument(data.length > 0, "Provided data array must have at least 1 item");
+ this.data = data;
+ this.pos = pos;
+ this.isOpen = true;
+ }
+ @Override public boolean isOpen() { return isOpen; }
+ @Override public void close() throws IOException { isOpen = false; }
+ @Override
+ public int read(ByteBuffer dst) throws IOException {
+ int l = data.length;
+ int arrOffset = (int)(pos % l);
+ int n = Math.min(dst.remaining(), l - arrOffset);
+ dst.put(data, arrOffset, n);
+ pos += n;
+ return n;
+ }
+
+ public long position() {
+ return pos;
+ }
+
+ public void position(long pos) {
+ this.pos = pos;
+ }
+ }
}
diff --git a/jena-base/src/main/java/org/apache/jena/atlas/iterator/IteratorFlatMap.java b/jena-base/src/main/java/org/apache/jena/atlas/iterator/IteratorFlatMap.java
index b4340c23d3..84c82914e3 100644
--- a/jena-base/src/main/java/org/apache/jena/atlas/iterator/IteratorFlatMap.java
+++ b/jena-base/src/main/java/org/apache/jena/atlas/iterator/IteratorFlatMap.java
@@ -23,12 +23,10 @@ import java.util.NoSuchElementException;
import java.util.function.Function;
import java.util.stream.Stream;
-import org.apache.jena.atlas.lib.Closeable;
-
/** flatMap iterator.
* See {@link Stream#flatMap}
*/
-/*package*/ class IteratorFlatMap<IN,OUT> implements Iterator<OUT>, Closeable {
+/*package*/ class IteratorFlatMap<IN,OUT> implements IteratorCloseable<OUT> {
private boolean finished = false;
private Iterator<OUT> current = null;
private Iterator<IN> input;