This is an automated email from the ASF dual-hosted git repository.
afs 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 f3d2f33fa1 GH-3955: StreamRDF.flush
f3d2f33fa1 is described below
commit f3d2f33fa19640f439fd347d4582818d5800fdc9
Author: Andy Seaborne <[email protected]>
AuthorDate: Thu May 28 11:32:51 2026 +0100
GH-3955: StreamRDF.flush
---
.../java/org/apache/jena/riot/system/BatchedStreamRDF.java | 3 ++-
.../java/org/apache/jena/riot/system/PrintingStreamRDF.java | 8 ++------
.../src/main/java/org/apache/jena/riot/system/StreamRDF.java | 10 ++++++----
.../main/java/org/apache/jena/riot/system/StreamRDF2.java | 6 ++++++
.../main/java/org/apache/jena/riot/system/StreamRDFBase.java | 4 ++++
.../main/java/org/apache/jena/riot/system/StreamRDFLib.java | 12 ++++++++++--
.../java/org/apache/jena/riot/system/StreamRDFReject.java | 7 ++++++-
.../java/org/apache/jena/riot/system/StreamRDFWrapper.java | 4 ++++
.../java/org/apache/jena/riot/thrift/StreamRDF2Thrift.java | 7 ++++++-
.../java/org/apache/jena/riot/writer/StreamWriterTriX.java | 2 ++
.../org/apache/jena/riot/writer/WriterStreamRDFBatched.java | 8 +++++---
.../org/apache/jena/riot/writer/WriterStreamRDFPlain.java | 6 +++++-
.../apache/jena/example/streaming/StreamRDFToConnection.java | 4 ++--
.../src/main/java/org/apache/jena/rdfpatch/StreamPatch.java | 1 -
.../org/apache/jena/tdb2/xloader/ProcBuildNodeTableX.java | 7 ++++++-
.../java/org/apache/jena/tdb2/xloader/ProcIngestDataX.java | 11 ++++++++---
16 files changed, 74 insertions(+), 26 deletions(-)
diff --git
a/jena-arq/src/main/java/org/apache/jena/riot/system/BatchedStreamRDF.java
b/jena-arq/src/main/java/org/apache/jena/riot/system/BatchedStreamRDF.java
index f6c1558593..26c953b366 100644
--- a/jena-arq/src/main/java/org/apache/jena/riot/system/BatchedStreamRDF.java
+++ b/jena-arq/src/main/java/org/apache/jena/riot/system/BatchedStreamRDF.java
@@ -123,7 +123,8 @@ public class BatchedStreamRDF implements StreamRDF {
batchQuads.add(Quad);
}
- private void flush() {
+ @Override
+ public void flush() {
finishBatchTriple(currentSubject);
finishBatchQuad(currentGraph, currentSubject);
}
diff --git
a/jena-arq/src/main/java/org/apache/jena/riot/system/PrintingStreamRDF.java
b/jena-arq/src/main/java/org/apache/jena/riot/system/PrintingStreamRDF.java
index 0e6020b44f..5b919ecda1 100644
--- a/jena-arq/src/main/java/org/apache/jena/riot/system/PrintingStreamRDF.java
+++ b/jena-arq/src/main/java/org/apache/jena/riot/system/PrintingStreamRDF.java
@@ -129,16 +129,12 @@ public class PrintingStreamRDF extends
WriterStreamRDFPlain
@Override
public void triple(Triple triple) {
super.triple(triple);
- flush();
+ flushOutput();
}
@Override
public void quad(Quad quad) {
super.quad(quad);
- flush();
- }
-
- public void flush() {
- IO.flush(out) ;
+ flushOutput();
}
}
diff --git a/jena-arq/src/main/java/org/apache/jena/riot/system/StreamRDF.java
b/jena-arq/src/main/java/org/apache/jena/riot/system/StreamRDF.java
index 0f66dcdb2e..057a03d157 100644
--- a/jena-arq/src/main/java/org/apache/jena/riot/system/StreamRDF.java
+++ b/jena-arq/src/main/java/org/apache/jena/riot/system/StreamRDF.java
@@ -44,16 +44,18 @@ public interface StreamRDF
// /** Generalized emitted */
// public void tuple(Tuple<Node> tuple) ;
//
- /** base declaration seen */
+ /** Base declaration seen */
public void base(String base) ;
- /** prefix declaration seen */
+ /** Prefix declaration seen */
public void prefix(String prefix, String iri) ;
- /** version declaration seen */
-
+ /** Version declaration seen */
public void version(String version);
+ /** Flush the stream */
+ public default void flush() {}
+
/** Finish processing */
public void finish() ;
diff --git a/jena-arq/src/main/java/org/apache/jena/riot/system/StreamRDF2.java
b/jena-arq/src/main/java/org/apache/jena/riot/system/StreamRDF2.java
index f5313743fa..56ffa6d31b 100644
--- a/jena-arq/src/main/java/org/apache/jena/riot/system/StreamRDF2.java
+++ b/jena-arq/src/main/java/org/apache/jena/riot/system/StreamRDF2.java
@@ -78,6 +78,12 @@ public class StreamRDF2 implements StreamRDF {
sink2.version(version);
}
+ @Override
+ public void flush() {
+ sink1.flush();
+ sink2.flush();
+ }
+
@Override
public void finish() {
sink1.finish();
diff --git
a/jena-arq/src/main/java/org/apache/jena/riot/system/StreamRDFBase.java
b/jena-arq/src/main/java/org/apache/jena/riot/system/StreamRDFBase.java
index 08ccbaec4c..32830638e4 100644
--- a/jena-arq/src/main/java/org/apache/jena/riot/system/StreamRDFBase.java
+++ b/jena-arq/src/main/java/org/apache/jena/riot/system/StreamRDFBase.java
@@ -43,6 +43,10 @@ public class StreamRDFBase implements StreamRDF {
@Override
public void version(String version) {}
+ @Override
+ public void flush() {}
+
@Override
public void finish() {}
+
}
diff --git
a/jena-arq/src/main/java/org/apache/jena/riot/system/StreamRDFLib.java
b/jena-arq/src/main/java/org/apache/jena/riot/system/StreamRDFLib.java
index cab99589b9..89494b0229 100644
--- a/jena-arq/src/main/java/org/apache/jena/riot/system/StreamRDFLib.java
+++ b/jena-arq/src/main/java/org/apache/jena/riot/system/StreamRDFLib.java
@@ -165,8 +165,12 @@ public class StreamRDFLib
{ sink.send(triple); }
@Override
- public void finish()
+ public void flush()
{ sink.flush(); }
+
+ @Override
+ public void finish()
+ { flush(); }
}
private static class ParserOutputSinkQuads extends StreamRDFBase
@@ -181,8 +185,12 @@ public class StreamRDFLib
{ sink.send(quad); }
@Override
- public void finish()
+ public void flush()
{ sink.flush(); }
+
+ @Override
+ public void finish()
+ { flush(); }
}
private static class ParserOutputGraph extends StreamRDFBase {
diff --git
a/jena-arq/src/main/java/org/apache/jena/riot/system/StreamRDFReject.java
b/jena-arq/src/main/java/org/apache/jena/riot/system/StreamRDFReject.java
index 9b521465c0..ae85e9d6df 100644
--- a/jena-arq/src/main/java/org/apache/jena/riot/system/StreamRDFReject.java
+++ b/jena-arq/src/main/java/org/apache/jena/riot/system/StreamRDFReject.java
@@ -25,7 +25,9 @@ import org.apache.jena.graph.Triple ;
import org.apache.jena.sparql.core.Quad ;
/**
- * {@link StreamRDF} that throws an exception on each operation except {@code
start()} and {@code finish()}.
+ * {@link StreamRDF} that throws an exception on each operation except
+ * {@code start()}, {@code flush()}, and {@code finish()}.
+ *
* @see StreamRDFBase
*/
public class StreamRDFReject implements StreamRDF
@@ -33,6 +35,9 @@ public class StreamRDFReject implements StreamRDF
@Override
public void start() {}
+ @Override
+ public void flush() {}
+
@Override
public void finish() {}
diff --git
a/jena-arq/src/main/java/org/apache/jena/riot/system/StreamRDFWrapper.java
b/jena-arq/src/main/java/org/apache/jena/riot/system/StreamRDFWrapper.java
index 62f74cce3d..c9b47562c9 100644
--- a/jena-arq/src/main/java/org/apache/jena/riot/system/StreamRDFWrapper.java
+++ b/jena-arq/src/main/java/org/apache/jena/riot/system/StreamRDFWrapper.java
@@ -59,6 +59,10 @@ public class StreamRDFWrapper implements StreamRDF
public void version(String versionString)
{ other.version(versionString); }
+ @Override
+ public void flush()
+ { other.flush(); }
+
@Override
public void finish()
{ other.finish() ; }
diff --git
a/jena-arq/src/main/java/org/apache/jena/riot/thrift/StreamRDF2Thrift.java
b/jena-arq/src/main/java/org/apache/jena/riot/thrift/StreamRDF2Thrift.java
index cb60a0539a..7bcd2701ad 100644
--- a/jena-arq/src/main/java/org/apache/jena/riot/thrift/StreamRDF2Thrift.java
+++ b/jena-arq/src/main/java/org/apache/jena/riot/thrift/StreamRDF2Thrift.java
@@ -161,7 +161,12 @@ public class StreamRDF2Thrift implements StreamRDF,
AutoCloseable
}
@Override
- public void finish() {
+ public void flush() {
TRDF.flush(protocol);
}
+
+ @Override
+ public void finish() {
+ flush();
+ }
}
diff --git
a/jena-arq/src/main/java/org/apache/jena/riot/writer/StreamWriterTriX.java
b/jena-arq/src/main/java/org/apache/jena/riot/writer/StreamWriterTriX.java
index 5764474ebf..39e60a60fb 100644
--- a/jena-arq/src/main/java/org/apache/jena/riot/writer/StreamWriterTriX.java
+++ b/jena-arq/src/main/java/org/apache/jena/riot/writer/StreamWriterTriX.java
@@ -86,6 +86,8 @@ public class StreamWriterTriX implements StreamRDF {
depth ++;
}
+ @Override public void flush() { }
+
@Override public void finish() {
depth--;
if ( depth != 0 )
diff --git
a/jena-arq/src/main/java/org/apache/jena/riot/writer/WriterStreamRDFBatched.java
b/jena-arq/src/main/java/org/apache/jena/riot/writer/WriterStreamRDFBatched.java
index cbe7a6275a..5c226e639b 100644
---
a/jena-arq/src/main/java/org/apache/jena/riot/writer/WriterStreamRDFBatched.java
+++
b/jena-arq/src/main/java/org/apache/jena/riot/writer/WriterStreamRDFBatched.java
@@ -56,12 +56,14 @@ abstract class WriterStreamRDFBatched extends
WriterStreamRDFBase
protected final void startData() { reset(); }
@Override
- protected final void endData() { flush(); }
+ protected final void endData() {
+ flushBatch();
+ finalizeRun();
+ }
- private void flush() {
+ private void flushBatch() {
finishBatchTriples(currentSubject);
finishBatchQuad(currentGraph, currentSubject);
- finalizeRun();
}
@Override
diff --git
a/jena-arq/src/main/java/org/apache/jena/riot/writer/WriterStreamRDFPlain.java
b/jena-arq/src/main/java/org/apache/jena/riot/writer/WriterStreamRDFPlain.java
index fb978c81ac..9778ec267b 100644
---
a/jena-arq/src/main/java/org/apache/jena/riot/writer/WriterStreamRDFPlain.java
+++
b/jena-arq/src/main/java/org/apache/jena/riot/writer/WriterStreamRDFPlain.java
@@ -72,9 +72,13 @@ public class WriterStreamRDFPlain implements StreamRDF {
@Override
public void start() {}
+ public void flushOutput() {
+ IO.flush(out);
+ }
+
@Override
public void finish() {
- IO.flush(out);
+ flushOutput();
}
@Override
diff --git
a/jena-examples/src/main/java/org/apache/jena/example/streaming/StreamRDFToConnection.java
b/jena-examples/src/main/java/org/apache/jena/example/streaming/StreamRDFToConnection.java
index 3a437e3f4b..726430678c 100644
---
a/jena-examples/src/main/java/org/apache/jena/example/streaming/StreamRDFToConnection.java
+++
b/jena-examples/src/main/java/org/apache/jena/example/streaming/StreamRDFToConnection.java
@@ -78,7 +78,7 @@ public class StreamRDFToConnection implements StreamRDF {
/**
* Flushes the buffer to the connection.
*/
- private void flush() {
+ private void flushInternal() {
UpdateBuilder builder = new UpdateBuilder();
builder.addPrefixes(model);
builder.addInsert(model);
@@ -122,7 +122,7 @@ public class StreamRDFToConnection implements StreamRDF {
@Override
public void finish() {
- flush();
+ flushInternal();
}
public static void main(String[] args) {
diff --git
a/jena-rdfpatch/src/main/java/org/apache/jena/rdfpatch/StreamPatch.java
b/jena-rdfpatch/src/main/java/org/apache/jena/rdfpatch/StreamPatch.java
index 3f427dc258..114138ab3e 100644
--- a/jena-rdfpatch/src/main/java/org/apache/jena/rdfpatch/StreamPatch.java
+++ b/jena-rdfpatch/src/main/java/org/apache/jena/rdfpatch/StreamPatch.java
@@ -56,7 +56,6 @@ public class StreamPatch implements StreamRDF {
}
}
-
@Override
public void finish() {
if ( depth == 1 ) {
diff --git
a/jena-tdb2/src/main/java/org/apache/jena/tdb2/xloader/ProcBuildNodeTableX.java
b/jena-tdb2/src/main/java/org/apache/jena/tdb2/xloader/ProcBuildNodeTableX.java
index cd16f9951b..5a710fef5b 100644
---
a/jena-tdb2/src/main/java/org/apache/jena/tdb2/xloader/ProcBuildNodeTableX.java
+++
b/jena-tdb2/src/main/java/org/apache/jena/tdb2/xloader/ProcBuildNodeTableX.java
@@ -455,8 +455,13 @@ public class ProcBuildNodeTableX {
public void version(String version) {}
@Override
- public void finish() {
+ public void flush() {
IO.flush(outputData);
}
+
+ @Override
+ public void finish() {
+ flush();
+ }
}
}
diff --git
a/jena-tdb2/src/main/java/org/apache/jena/tdb2/xloader/ProcIngestDataX.java
b/jena-tdb2/src/main/java/org/apache/jena/tdb2/xloader/ProcIngestDataX.java
index 832f0cf1da..bb986a251a 100644
--- a/jena-tdb2/src/main/java/org/apache/jena/tdb2/xloader/ProcIngestDataX.java
+++ b/jena-tdb2/src/main/java/org/apache/jena/tdb2/xloader/ProcIngestDataX.java
@@ -212,15 +212,20 @@ public class ProcIngestDataX {
@Override
public void start() {}
+ @Override
+ public void flush() {
+ writerTriples.flush();
+ writerQuads.flush();
+
+ }
+
@Override
public void finish() {}
// @Override
public void finishBulk() {
- writerTriples.flush();
- writerQuads.flush();
+ flush();
nodeTable.sync();
-
// dsg.getStoragePrefixes().sync();
}
