RYA-466 Update the Rya Streams Client to stream results to file.
diff --git a/common/rya.api/src/main/java/org/apache/rya/api/utils/QueryInvestigator.java b/common/rya.api/src/main/java/org/apache/rya/api/utils/QueryInvestigator.java
new file mode 100644
index 0000000..2fbd09b
--- /dev/null
+++ b/common/rya.api/src/main/java/org/apache/rya/api/utils/QueryInvestigator.java
@@ -0,0 +1,103 @@
+/**
+ * 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.rya.api.utils;
+
+import static java.util.Objects.requireNonNull;
+
+import org.openrdf.query.MalformedQueryException;
+import org.openrdf.query.parser.ParsedGraphQuery;
+import org.openrdf.query.parser.ParsedQuery;
+import org.openrdf.query.parser.QueryParserUtil;
+import org.openrdf.query.parser.sparql.SPARQLParser;
+
+import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+/**
+ * A utility class that is used to glean insight into the structure of SPARQL queries.
+ */
+@DefaultAnnotation(NonNull.class)
+public class QueryInvestigator {
+
+ private static final SPARQLParser PARSER = new SPARQLParser();
+
+ private QueryInvestigator() { }
+
+ /**
+ * Determines whether a SPARQL command is a CONSTRUCT or not.
+ *
+ * @param sparql - The SPARQL to evaluate. (not null)
+ * @return {@code true} if the provided SPARQL is a CONSTRUCT query; otherwise {@code false}.
+ * @throws MalformedQueryException The SPARQL is neither a well formed query or update.
+ */
+ public static boolean isConstruct(final String sparql) throws MalformedQueryException {
+ requireNonNull(sparql);
+
+ try {
+ // Constructs are queries, so try to create a ParsedQuery.
+ final ParsedQuery parsedQuery = PARSER.parseQuery(sparql, null);
+
+ // Check to see if the SPARQL looks like a CONSTRUCT query.
+ return parsedQuery instanceof ParsedGraphQuery;
+
+ } catch(final MalformedQueryException queryE) {
+ try {
+ // Maybe it's an update.
+ PARSER.parseUpdate(sparql, null);
+
+ // It was, so return false.
+ return false;
+
+ } catch(final MalformedQueryException updateE) {
+ // It's not. Actually malformed.
+ throw queryE;
+ }
+ }
+ }
+
+ /**
+ * Determines whether a SPARQL command is an INSERT with a WHERE clause or not.
+ *
+ * @param sparql - The SPARQL to evaluate. (not null)
+ * @return {@code true} if the provided SPARQL is an INSERT update; otherwise {@code false}.
+ * @throws MalformedQueryException The SPARQL is neither a well formed query or update.
+ */
+ public static boolean isInsertWhere(final String sparql) throws MalformedQueryException {
+ requireNonNull(sparql);
+
+ try {
+ // Inserts are updated, so try to create a ParsedUpdate.
+ PARSER.parseUpdate(sparql, null);
+ final String strippedOperation = QueryParserUtil.removeSPARQLQueryProlog(sparql.toLowerCase());
+ return strippedOperation.startsWith("insert");
+ } catch(final MalformedQueryException updateE) {
+ try {
+ // Maybe it's a query.
+ PARSER.parseQuery(sparql, null);
+
+ // It was, so return false.
+ return false;
+
+ } catch(final MalformedQueryException queryE) {
+ // It's not. Actually malformed.
+ throw updateE;
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/common/rya.api/src/test/java/org/apache/rya/api/utils/QueryInvestigatorTest.java b/common/rya.api/src/test/java/org/apache/rya/api/utils/QueryInvestigatorTest.java
new file mode 100644
index 0000000..fbe2ebe
--- /dev/null
+++ b/common/rya.api/src/test/java/org/apache/rya/api/utils/QueryInvestigatorTest.java
@@ -0,0 +1,150 @@
+/**
+ * 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.rya.api.utils;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Test;
+import org.openrdf.query.MalformedQueryException;
+
+/**
+ * Unit tests the methods of {@link QueryInvestigator}.
+ */
+public class QueryInvestigatorTest {
+
+ @Test
+ public void isConstruct_true() throws Exception {
+ final String sparql =
+ "PREFIX vCard: <http://www.w3.org/2001/vcard-rdf/3.0#> " +
+ "PREFIX foaf: <http://xmlns.com/foaf/0.1/> " +
+ "CONSTRUCT { " +
+ "?X vCard:FN ?name . " +
+ "?X vCard:URL ?url . " +
+ "?X vCard:TITLE ?title . " +
+ "} " +
+ "FROM <http://www.w3.org/People/Berners-Lee/card> " +
+ "WHERE { " +
+ "OPTIONAL { ?X foaf:name ?name . FILTER isLiteral(?name) . } " +
+ "OPTIONAL { ?X foaf:homepage ?url . FILTER isURI(?url) . } " +
+ "OPTIONAL { ?X foaf:title ?title . FILTER isLiteral(?title) . } " +
+ "}";
+
+ assertTrue( QueryInvestigator.isConstruct(sparql) );
+ }
+
+ @Test
+ public void isConstruct_false_notAConstruct() throws Exception {
+ final String sparql = "SELECT * WHERE { ?a ?b ?c . }";
+ assertFalse( QueryInvestigator.isConstruct(sparql) );
+ }
+
+ @Test
+ public void isConstruct_false_notAConstructWithKeywords() throws Exception {
+ final String sparql =
+ "SELECT ?construct " +
+ "WHERE {" +
+ " ?construct <urn:built> <urn:skyscraper> ." +
+ "}";
+ assertFalse( QueryInvestigator.isConstruct(sparql) );
+ }
+
+ @Test
+ public void isConstruct_false_notAQuery() throws Exception {
+ final String sparql =
+ "PREFIX Sensor: <http://example.com/Equipment.owl#> " +
+ "INSERT { " +
+ "?subject Sensor:test2 ?newValue " +
+ "} WHERE {" +
+ "values (?oldValue ?newValue) {" +
+ "('testValue1' 'newValue1')" +
+ "('testValue2' 'newValue2')" +
+ "}" +
+ "?subject Sensor:test1 ?oldValue" +
+ "}";
+
+ assertFalse( QueryInvestigator.isConstruct(sparql) );
+ }
+
+ @Test(expected = MalformedQueryException.class)
+ public void isConstruct_false_malformed() throws MalformedQueryException {
+ assertFalse( QueryInvestigator.isConstruct("not sparql") );
+ }
+
+ @Test
+ public void isInsert_true() throws Exception {
+ final String sparql =
+ "PREFIX Sensor: <http://example.com/Equipment.owl#> " +
+ "INSERT { " +
+ "?subject Sensor:test2 ?newValue " +
+ "} WHERE {" +
+ "values (?oldValue ?newValue) {" +
+ "('testValue1' 'newValue1')" +
+ "('testValue2' 'newValue2')" +
+ "}" +
+ "?subject Sensor:test1 ?oldValue" +
+ "}";
+
+ assertTrue( QueryInvestigator.isInsertWhere(sparql) );
+ }
+
+ @Test
+ public void isInsert_false_notAnInsert() throws Exception {
+ final String sparql =
+ "PREFIX dc: <http://purl.org/dc/elements/1.1/> " +
+ "PREFIX xsd: <http://www.w3.org/2001/XMLSchema#> " +
+ "DELETE { " +
+ "?book ?p ?v " +
+ "} WHERE { " +
+ "?book dc:date ?date . " +
+ "FILTER ( ?date < \"2000-01-01T00:00:00\"^^xsd:dateTime ) " +
+ "?book ?p ?v " +
+ "}";
+
+ assertFalse( QueryInvestigator.isInsertWhere(sparql) );
+ }
+
+ @Test
+ public void isInsert_false_notAnInsertWithKeywords() throws Exception {
+ final String sparql =
+ "DELETE" +
+ "{ " +
+ " ?bookInsert ?p ?o" +
+ "}" +
+ "WHERE" +
+ "{ " +
+ " ?bookInsert <urn:datePrinted> ?datePrinted ." +
+ " FILTER ( ?datePrinted < \"2018-01-01T00:00:00\"^^xsd:dateTime )" +
+ " ?bookInsert ?p ?o" +
+ "}";
+
+ assertFalse( QueryInvestigator.isInsertWhere(sparql) );
+ }
+
+ @Test
+ public void isInsert_false_notAnUpdate() throws Exception {
+ final String sparql = "SELECT * WHERE { ?a ?b ?c . }";
+ assertFalse( QueryInvestigator.isInsertWhere(sparql) );
+ }
+
+ @Test(expected = MalformedQueryException.class)
+ public void isInsert_false_malformed() throws MalformedQueryException {
+ assertFalse( QueryInvestigator.isInsertWhere("not sparql") );
+ }
+}
\ No newline at end of file
diff --git a/extras/rya.streams/client/pom.xml b/extras/rya.streams/client/pom.xml
index 5bd7933..38c3c86 100644
--- a/extras/rya.streams/client/pom.xml
+++ b/extras/rya.streams/client/pom.xml
@@ -69,6 +69,10 @@
<groupId>org.openrdf.sesame</groupId>
<artifactId>sesame-rio-turtle</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.openrdf.sesame</groupId>
+ <artifactId>sesame-queryresultio-sparqljson</artifactId>
+ </dependency>
<!-- Third Party dependencies -->
<dependency>
@@ -92,6 +96,11 @@
<scope>test</scope>
</dependency>
<dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-all</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
<groupId>org.apache.rya</groupId>
<artifactId>rya.test.kafka</artifactId>
<scope>test</scope>
diff --git a/extras/rya.streams/client/src/main/java/org/apache/rya/streams/client/command/StreamResultsCommand.java b/extras/rya.streams/client/src/main/java/org/apache/rya/streams/client/command/StreamResultsCommand.java
index 3612dd0..8ae4e08 100644
--- a/extras/rya.streams/client/src/main/java/org/apache/rya/streams/client/command/StreamResultsCommand.java
+++ b/extras/rya.streams/client/src/main/java/org/apache/rya/streams/client/command/StreamResultsCommand.java
@@ -1,4 +1,4 @@
-/*
+/**
* 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
@@ -20,11 +20,18 @@
import static java.util.Objects.requireNonNull;
+import java.io.OutputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.rya.api.model.VisibilityBindingSet;
+import org.apache.rya.api.model.VisibilityStatement;
+import org.apache.rya.api.utils.QueryInvestigator;
import org.apache.rya.streams.api.entity.QueryResultStream;
import org.apache.rya.streams.api.entity.StreamsQuery;
import org.apache.rya.streams.api.interactor.GetQueryResultStream;
@@ -32,13 +39,13 @@
import org.apache.rya.streams.api.queries.QueryChangeLog;
import org.apache.rya.streams.api.queries.QueryRepository;
import org.apache.rya.streams.client.RyaStreamsCommand;
+import org.apache.rya.streams.client.util.QueryResultsOutputUtil;
import org.apache.rya.streams.kafka.KafkaTopics;
import org.apache.rya.streams.kafka.interactor.KafkaGetQueryResultStream;
import org.apache.rya.streams.kafka.queries.KafkaQueryChangeLogFactory;
import org.apache.rya.streams.kafka.serialization.VisibilityBindingSetDeserializer;
import org.apache.rya.streams.kafka.serialization.VisibilityStatementDeserializer;
import org.openrdf.query.MalformedQueryException;
-import org.openrdf.query.algebra.Reduced;
import org.openrdf.query.algebra.TupleExpr;
import org.openrdf.query.parser.sparql.SPARQLParser;
@@ -65,14 +72,20 @@
@Parameter(names = {"--queryId", "-q"}, required = true, description = "The query whose results will be streamed to the console.")
private String queryId;
+ @Parameter(names = {"--file", "-f"}, required = false, description = "If provided, the output file the results will stream into.")
+ private String outputPath;
+
@Override
public String toString() {
final StringBuilder parameters = new StringBuilder();
parameters.append(super.toString());
if (!Strings.isNullOrEmpty(queryId)) {
- parameters.append("\tQuery ID: " + queryId);
- parameters.append("\n");
+ parameters.append("\tQuery ID: " + queryId + "\n");
+ }
+
+ if(!Strings.isNullOrEmpty(outputPath)) {
+ parameters.append("\tOutput Path: " + outputPath + "\n");
}
return parameters.toString();
@@ -160,10 +173,12 @@
});
// Build the interactor based on the type of result the query produces.
+ boolean isStatementResults = false;
+
final GetQueryResultStream<?> getQueryResultStream;
try {
- final TupleExpr tupleExpr = new SPARQLParser().parseQuery(sparql, null).getTupleExpr();
- if(tupleExpr instanceof Reduced) {
+ isStatementResults = QueryInvestigator.isConstruct(sparql) | QueryInvestigator.isInsertWhere(sparql);
+ if(isStatementResults) {
getQueryResultStream = new KafkaGetQueryResultStream<>(params.kafkaIP, params.kafkaPort, VisibilityStatementDeserializer.class);
} else {
getQueryResultStream = new KafkaGetQueryResultStream<>(params.kafkaIP, params.kafkaPort, VisibilityBindingSetDeserializer.class);
@@ -172,12 +187,22 @@
throw new ExecutionException("Could not parse the SPARQL for the query: " + sparql, e);
}
- // Iterate through the results and print them to the console until the program or the stream ends.
- try (final QueryResultStream<?> stream = getQueryResultStream.fromStart(queryId)) {
- while(!finished.get()) {
- for(final Object result : stream.poll(1000)) {
- System.out.println(result);
+ // Iterate through the results and print them to the configured output mechanism.
+ try (final QueryResultStream<?> resultsStream = getQueryResultStream.fromStart(queryId)) {
+ final TupleExpr tupleExpr = new SPARQLParser().parseQuery(sparql, null).getTupleExpr();
+ if(params.outputPath != null) {
+ final Path file = Paths.get(params.outputPath);
+ try (final OutputStream out = Files.newOutputStream(file)) {
+ if(isStatementResults) {
+ final QueryResultStream<VisibilityStatement> stmtStream = (QueryResultStream<VisibilityStatement>) resultsStream;
+ QueryResultsOutputUtil.toNtriplesFile(out, stmtStream, finished);
+ } else {
+ final QueryResultStream<VisibilityBindingSet> bsStream = (QueryResultStream<VisibilityBindingSet>) resultsStream;
+ QueryResultsOutputUtil.toBindingSetJSONFile(out, tupleExpr, bsStream, finished);
+ }
}
+ } else {
+ streamToSystemOut(resultsStream, finished);
}
} catch (final Exception e) {
System.err.println("Error while reading the results from the stream.");
@@ -185,4 +210,15 @@
System.exit(1);
}
}
+
+ private static void streamToSystemOut(final QueryResultStream<?> stream, final AtomicBoolean shutdownSignal) throws Exception {
+ requireNonNull(stream);
+ requireNonNull(shutdownSignal);
+
+ while(!shutdownSignal.get()) {
+ for(final Object result : stream.poll(1000)) {
+ System.out.println(result);
+ }
+ }
+ }
}
\ No newline at end of file
diff --git a/extras/rya.streams/client/src/main/java/org/apache/rya/streams/client/util/QueryResultsOutputUtil.java b/extras/rya.streams/client/src/main/java/org/apache/rya/streams/client/util/QueryResultsOutputUtil.java
new file mode 100644
index 0000000..114a6fe
--- /dev/null
+++ b/extras/rya.streams/client/src/main/java/org/apache/rya/streams/client/util/QueryResultsOutputUtil.java
@@ -0,0 +1,133 @@
+/**
+ * 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.rya.streams.client.util;
+
+import static java.util.Objects.requireNonNull;
+
+import java.io.OutputStream;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.rya.api.model.VisibilityBindingSet;
+import org.apache.rya.api.model.VisibilityStatement;
+import org.apache.rya.streams.api.entity.QueryResultStream;
+import org.apache.rya.streams.api.exception.RyaStreamsException;
+import org.openrdf.query.TupleQueryResultHandlerException;
+import org.openrdf.query.algebra.TupleExpr;
+import org.openrdf.query.resultio.sparqljson.SPARQLResultsJSONWriter;
+import org.openrdf.rio.RDFFormat;
+import org.openrdf.rio.RDFHandlerException;
+import org.openrdf.rio.RDFWriter;
+import org.openrdf.rio.Rio;
+import org.openrdf.rio.WriterConfig;
+import org.openrdf.rio.helpers.BasicWriterSettings;
+
+import com.google.common.collect.Lists;
+
+import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+/**
+ * A utility that writes {@link QueryResultStream} results to an {@link OutputStream}.
+ */
+@DefaultAnnotation(NonNull.class)
+public class QueryResultsOutputUtil {
+
+ /**
+ * Private constructor to prevent instantiation.
+ */
+ private QueryResultsOutputUtil() { }
+
+ /**
+ * Writes the results of a {@link QueryResultStream} to the output stream as NTriples until the
+ * shutdown signal is set.
+ *
+ * @param out - The stream the NTriples data will be written to. (not null)
+ * @param resultsStream - The results stream that will be polled for results to
+ * write to {@code out}. (not null)
+ * @param shutdownSignal - Setting this signal will cause the thread that
+ * is processing this function to finish and leave. (not null)
+ * @throws RDFHandlerException A problem was encountered while
+ * writing the NTriples to the output stream.
+ * @throws IllegalStateException The {@code resultsStream} is closed.
+ * @throws RyaStreamsException Could not fetch the next set of results.
+ */
+ public static void toNtriplesFile(
+ final OutputStream out,
+ final QueryResultStream<VisibilityStatement> resultsStream,
+ final AtomicBoolean shutdownSignal) throws RDFHandlerException, IllegalStateException, RyaStreamsException {
+ requireNonNull(out);
+ requireNonNull(resultsStream);
+ requireNonNull(shutdownSignal);
+
+ final RDFWriter writer = Rio.createWriter(RDFFormat.NTRIPLES, out);
+ writer.startRDF();
+
+ while(!shutdownSignal.get()) {
+ final Iterable<VisibilityStatement> it = resultsStream.poll(1000);
+ for(final VisibilityStatement result : it) {
+ writer.handleStatement(result);
+ }
+ }
+
+ writer.endRDF();
+ }
+
+ /**
+ * Writes the results of a {@link QueryResultStream} to the output stream as JSON until the
+ * shutdown signal is set.
+ *
+ * @param out - The stream the JSON will be written to. (not null)
+ * @param query - The parsed SPARQL Query whose results are being output. This
+ * object is used to figure out which bindings may appear. (not null)
+ * @param resultsStream - The results stream that will be polled for results to
+ * write to {@code out}. (not null)
+ * @param shutdownSignal - Setting this signal will cause the thread that
+ * is processing this function to finish and leave. (not null)
+ * @throws TupleQueryResultHandlerException A problem was encountered while
+ * writing the JSON to the output stream.
+ * @throws IllegalStateException The {@code resultsStream} is closed.
+ * @throws RyaStreamsException Could not fetch the next set of results.
+ */
+ public static void toBindingSetJSONFile(
+ final OutputStream out,
+ final TupleExpr query,
+ final QueryResultStream<VisibilityBindingSet> resultsStream,
+ final AtomicBoolean shutdownSignal) throws TupleQueryResultHandlerException, IllegalStateException, RyaStreamsException {
+ requireNonNull(out);
+ requireNonNull(query);
+ requireNonNull(resultsStream);
+ requireNonNull(shutdownSignal);
+
+ // Create a writer that does not pretty print.
+ final SPARQLResultsJSONWriter writer = new SPARQLResultsJSONWriter(out);
+ final WriterConfig config = writer.getWriterConfig();
+ config.set(BasicWriterSettings.PRETTY_PRINT, false);
+
+ // Start the JSON and enumerate the possible binding names.
+ writer.startQueryResult( Lists.newArrayList(query.getBindingNames()) );
+
+ while(!shutdownSignal.get()) {
+ for(final VisibilityBindingSet result : resultsStream.poll(1000)) {
+ writer.handleSolution(result);
+ }
+ }
+
+ writer.endQueryResult();
+ }
+}
\ No newline at end of file
diff --git a/extras/rya.streams/client/src/test/java/org/apache/rya/streams/client/util/QueryResultsOutputUtilTest.java b/extras/rya.streams/client/src/test/java/org/apache/rya/streams/client/util/QueryResultsOutputUtilTest.java
new file mode 100644
index 0000000..b82e671
--- /dev/null
+++ b/extras/rya.streams/client/src/test/java/org/apache/rya/streams/client/util/QueryResultsOutputUtilTest.java
@@ -0,0 +1,135 @@
+/**
+ * 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.rya.streams.client.util;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.rya.api.model.VisibilityBindingSet;
+import org.apache.rya.api.model.VisibilityStatement;
+import org.apache.rya.streams.api.entity.QueryResultStream;
+import org.junit.Test;
+import org.openrdf.model.Statement;
+import org.openrdf.model.ValueFactory;
+import org.openrdf.model.impl.ValueFactoryImpl;
+import org.openrdf.query.algebra.TupleExpr;
+import org.openrdf.query.impl.MapBindingSet;
+import org.openrdf.query.parser.sparql.SPARQLParser;
+
+import com.google.common.base.Charsets;
+
+/**
+ * Unit tests the methods of {@link QueryResultsOutputUtil}.
+ */
+public class QueryResultsOutputUtilTest {
+
+ private static final ValueFactory VF = new ValueFactoryImpl();
+
+ @Test
+ public void toNtriplesFile() throws Exception {
+ // Mock a result stream that signals shutdown when it returns a set of results.
+ final AtomicBoolean shutdownSignal = new AtomicBoolean(false);
+ final QueryResultStream<VisibilityStatement> resultsStream = mock(QueryResultStream.class);
+ when(resultsStream.poll(anyLong())).thenAnswer(invocation -> {
+ shutdownSignal.set(true);
+
+ final List<VisibilityStatement> results = new ArrayList<>();
+
+ Statement stmt = VF.createStatement(VF.createURI("urn:alice"), VF.createURI("urn:age"), VF.createLiteral(23));
+ results.add( new VisibilityStatement(stmt) );
+
+ stmt = VF.createStatement(VF.createURI("urn:bob"), VF.createURI("urn:worksAt"), VF.createLiteral("Taco Shop"));
+ results.add( new VisibilityStatement(stmt) );
+ return results;
+ });
+
+ // The stream the JSON will be written to.
+ final ByteArrayOutputStream out = new ByteArrayOutputStream();
+
+ // Invoke the test method. This will write the NTriples.
+ QueryResultsOutputUtil.toNtriplesFile(out, resultsStream, shutdownSignal);
+
+ // Show the produced NTriples matches the expected NTriples.
+ final String expected =
+ "<urn:alice> <urn:age> \"23\"^^<http://www.w3.org/2001/XMLSchema#int> .\n" +
+ "<urn:bob> <urn:worksAt> \"Taco Shop\" .\n";
+
+ final String nTriples = new String(out.toByteArray(), Charsets.UTF_8);
+ assertEquals(expected, nTriples);
+ }
+
+ @Test
+ public void toBindingSetJSONFile() throws Exception {
+ // A SPARQL query that uses OPTIONAL values.
+ final String sparql =
+ "SELECT * WHERE { " +
+ "?name <urn:worksAt> ?company . " +
+ "OPTIONAL{ ?name <urn:ssn> ?ssn} " +
+ "}";
+ final TupleExpr query = new SPARQLParser().parseQuery(sparql, null).getTupleExpr();
+
+ // Mock a results stream that signals shutdown when it returns a set of results.
+ final AtomicBoolean shutdownSignal = new AtomicBoolean(false);
+ final QueryResultStream<VisibilityBindingSet> resultsStream = mock(QueryResultStream.class);
+ when(resultsStream.poll(anyLong())).thenAnswer(invocation -> {
+ shutdownSignal.set(true);
+
+ final List<VisibilityBindingSet> results = new ArrayList<>();
+
+ // A result with the optional value.
+ MapBindingSet bs = new MapBindingSet();
+ bs.addBinding("name", VF.createLiteral("alice"));
+ bs.addBinding("company", VF.createLiteral("Taco Shop"));
+ bs.addBinding("ssn", VF.createURI("urn:111-11-1111"));
+ results.add(new VisibilityBindingSet(bs, ""));
+
+
+ // A result without the optional value.
+ bs = new MapBindingSet();
+ bs.addBinding("name", VF.createLiteral("bob"));
+ bs.addBinding("company", VF.createLiteral("Cafe"));
+ results.add(new VisibilityBindingSet(bs, ""));
+
+ return results;
+ });
+
+ // The stream the JSON will be written to.
+ final ByteArrayOutputStream out = new ByteArrayOutputStream();
+
+ // Invoke the test method. This will write the json.
+ QueryResultsOutputUtil.toBindingSetJSONFile(out, query, resultsStream, shutdownSignal);
+
+ // Show the produced JSON matches the expected JSON.
+ final String expected = "{\"head\":{\"vars\":[\"name\",\"company\",\"ssn\"]},\"results\":{" +
+ "\"bindings\":[{\"name\":{\"type\":\"literal\",\"value\":\"alice\"},\"company\":{" +
+ "\"type\":\"literal\",\"value\":\"Taco Shop\"},\"ssn\":{\"type\":\"uri\",\"value\":" +
+ "\"urn:111-11-1111\"}},{\"name\":{\"type\":\"literal\",\"value\":\"bob\"},\"company\"" +
+ ":{\"type\":\"literal\",\"value\":\"Cafe\"}}]}}";
+ final String json = new String(out.toByteArray(), Charsets.UTF_8);
+ System.out.println(json);
+ assertEquals(expected, json);
+ }
+}
\ No newline at end of file
