This class provides a streaming API for jq processing, allowing:
+ *Example usage:
+ *{@code
+ * JqLibrary lib = ImmutableJqLibrary.of();
+ * try (JqFilter filter = JqFilter.compile(lib, ".[] | select(.active)")) {
+ * // Callback-based processing
+ * filter.process(jsonInput, output -> System.out.println(output));
+ *
+ * // Iterator-based processing
+ * Iterator results = filter.processIterator(anotherInput);
+ * while (results.hasNext()) {
+ * System.out.println(results.next());
+ * }
+ *
+ * // Chunked input processing
+ * try (JqStreamProcessor processor = filter.createStreamProcessor(output -> ...)) {
+ * processor.feedChunk(chunk1, false);
+ * processor.feedChunk(chunk2, true); // finished=true on last chunk
+ * }
+ * }
+ * }
+ *
+ * Thread Safety: jq is not thread-safe. All operations on this filter + * are synchronized with a global lock to ensure thread safety.
+ */ +public class JqFilter implements AutoCloseable { + private static final Logger LOGGER = Logger.getLogger(JqFilter.class.getName()); + + /** + * Global lock for thread safety - jq is not thread-safe. + * Shared with JqRequest to ensure consistency. + */ + private static final ReentrantLock SYNC = new ReentrantLock(); + + private final JqLibrary lib; + private final String filter; + private final Pointer jq; + private final int dumpFlags; + private volatile boolean closed = false; + + private JqFilter(JqLibrary lib, String filter, Pointer jq, int dumpFlags) { + this.lib = lib; + this.filter = filter; + this.jq = jq; + this.dumpFlags = dumpFlags; + } + + /** + * Compiles a jq filter expression for reuse. + * + * @param lib the jq library instance + * @param filter the jq filter expression to compile + * @return a compiled filter that can be reused + * @throws JqFilterException if the filter cannot be compiled + */ + public static JqFilter compile(JqLibrary lib, String filter) { + return builder() + .lib(lib) + .filter(filter) + .build(); + } + + /** + * Creates a new builder for configuring a JqFilter. + * + * @return a new builder instance + */ + public static Builder builder() { + return new Builder(); + } + + /** + * Builder for creating JqFilter instances with custom configuration. + */ + public static class Builder { + private JqLibrary lib; + private String filter = "."; + private boolean pretty = false; + private Indent indent = Indent.NONE; + private boolean sortKeys = false; + private ListNote: The iterator must be fully consumed before processing more input + * with this filter, as jq maintains internal state during iteration.
+ * + * @param input the JSON input to process + * @return iterator over output values + * @throws JqFilterException if processing fails + */ + public IteratorUse this when you need to process very large inputs that should be + * fed incrementally rather than loaded entirely into memory.
+ * + *Example:
+ *{@code
+ * try (JqStreamProcessor processor = filter.createStreamProcessor(output -> ...)) {
+ * processor.feedChunk(chunk1, false);
+ * processor.feedChunk(chunk2, false);
+ * processor.feedChunk(chunk3, true); // finished=true on last chunk
+ * }
+ * }
+ *
+ * @param callback called for each output value produced
+ * @return a new stream processor
+ */
+ public JqStreamProcessor createStreamProcessor(JqOutputCallback callback) {
+ ensureOpen();
+ return new JqStreamProcessor(this, callback);
+ }
+
+ private void processInternal(String input, JqOutputCallback callback) {
+ ListAfter closing, this filter cannot be used for processing.
+ */ + @Override + public void close() { + if (!closed) { + SYNC.lock(); + try { + if (!closed) { + LOGGER.log(FINE, "Releasing JQ state"); + lib.jq_teardown(jq); + closed = true; + LOGGER.log(FINE, "JQ state released"); + } + } finally { + SYNC.unlock(); + } + } + } +} diff --git a/src/main/java/com/arakelian/jq/JqFilterException.java b/src/main/java/com/arakelian/jq/JqFilterException.java new file mode 100644 index 0000000..17917e1 --- /dev/null +++ b/src/main/java/com/arakelian/jq/JqFilterException.java @@ -0,0 +1,75 @@ +/* + * 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 com.arakelian.jq; + +import java.util.List; + +import com.google.common.collect.ImmutableList; + +/** + * Exception thrown when a jq filter compilation or execution fails. + * + *This exception is used by the streaming API ({@link JqFilter}) to report + * errors during filter compilation or JSON processing.
+ */ +public class JqFilterException extends RuntimeException { + private static final long serialVersionUID = 1L; + + private final ListEach output value from the jq filter is passed to this callback as it is produced, + * allowing for immediate processing without buffering all results in memory.
+ * + *Example usage:
+ *{@code
+ * try (JqFilter filter = JqFilter.compile(library, ".[] | select(.active)")) {
+ * filter.process(jsonInput, output -> {
+ * System.out.println(output);
+ * });
+ * }
+ * }
+ */
+@FunctionalInterface
+public interface JqOutputCallback {
+ /**
+ * Called for each output value produced by the jq filter.
+ *
+ * @param output the serialized JSON output value
+ */
+ void onOutput(String output);
+}
diff --git a/src/main/java/com/arakelian/jq/JqRequest.java b/src/main/java/com/arakelian/jq/JqRequest.java
index 50b2b97..9657875 100644
--- a/src/main/java/com/arakelian/jq/JqRequest.java
+++ b/src/main/java/com/arakelian/jq/JqRequest.java
@@ -17,24 +17,13 @@
package com.arakelian.jq;
-import static java.util.logging.Level.FINE;
-
import java.io.File;
-import java.io.IOException;
-import java.io.UncheckedIOException;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.locks.ReentrantLock;
-import java.util.logging.Logger;
import org.immutables.value.Value;
-import com.arakelian.jq.JqLibrary.Jv;
-import com.google.common.base.Charsets;
-import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
-import com.sun.jna.Memory;
-import com.sun.jna.Pointer;
@Value.Immutable
public abstract class JqRequest {
@@ -45,19 +34,25 @@ public enum Indent {
TWO_SPACES;
}
- private static final Logger LOGGER = Logger.getLogger(JqRequest.class.getName());
-
- /**
- * JQ is not thread-safe - https://github.com/stedolan/jq/issues/120
- */
- private static final ReentrantLock SYNC = new ReentrantLock();
-
public final JqResponse execute() {
- SYNC.lock();
- try {
- return jq();
- } finally {
- SYNC.unlock();
+ try (JqFilter filter = JqFilter.builder()
+ .lib(getLib())
+ .filter(getFilter())
+ .pretty(isPretty())
+ .indent(getIndent())
+ .sortKeys(isSortKeys())
+ .modulePaths(getModulePaths())
+ .argJson(getArgJson())
+ .build()) {
+
+ ListThis class allows processing very large JSON inputs incrementally, + * without loading the entire input into memory at once. Input chunks are + * fed to the parser, and output values are emitted via callback as they + * become available.
+ * + *Example usage:
+ *{@code
+ * try (JqFilter filter = JqFilter.compile(lib, ".[] | select(.active)")) {
+ * try (JqStreamProcessor processor = filter.createStreamProcessor(output -> {
+ * System.out.println(output);
+ * })) {
+ * // Feed chunks of JSON data
+ * processor.feedChunk("{\"items\": [", false);
+ * processor.feedChunk("{\"name\": \"a\", \"active\": true},", false);
+ * processor.feedChunk("{\"name\": \"b\", \"active\": false}", false);
+ * processor.feedChunk("]}", true); // finished=true on last chunk
+ * }
+ * }
+ * }
+ *
+ * Thread Safety: Operations on this processor are synchronized + * with the global jq lock to ensure thread safety.
+ * + *Important: You must call {@link #feedChunk(String, boolean)} + * with {@code finished=true} at least once to signal end of input, or call + * {@link #finish()} before closing.
+ */ +public class JqStreamProcessor implements AutoCloseable { + private static final Logger LOGGER = Logger.getLogger(JqStreamProcessor.class.getName()); + + /** + * Global lock for thread safety - shared with JqFilter and JqRequest. + */ + private static final ReentrantLock SYNC = new ReentrantLock(); + + private final JqFilter filter; + private final JqOutputCallback callback; + private final JqLibrary lib; + private final Pointer parser; + private final ListChunks can be any portion of valid JSON - they don't need to be + * complete JSON values. The parser will buffer incomplete values until + * more data arrives.
+ * + *When feeding the last chunk, set {@code finished} to {@code true} + * to signal end of input. This allows the parser to finalize any + * buffered partial values.
+ * + * @param chunk the chunk of JSON text to process + * @param finished true if this is the last chunk + * @throws JqFilterException if a processing error occurs + * @throws IllegalStateException if the processor is closed or already finished + */ + public void feedChunk(String chunk, boolean finished) { + feedChunk(chunk.getBytes(Charsets.UTF_8), finished); + } + + /** + * Feeds a chunk of JSON input as raw bytes. + * + *This method is useful when working with byte streams or when + * the input encoding is already known to be UTF-8.
+ * + * @param bytes the chunk of UTF-8 encoded JSON + * @param finished true if this is the last chunk + * @throws JqFilterException if a processing error occurs + * @throws IllegalStateException if the processor is closed or already finished + */ + public void feedChunk(byte[] bytes, boolean finished) { + ensureOpen(); + if (this.finished) { + throw new IllegalStateException("Stream processing already finished"); + } + + SYNC.lock(); + try { + LOGGER.log(FINE, "Feeding chunk to parser, finished=" + finished); + JqFilter.feedParserBuffer(lib, parser, bytes, finished); + + // Process any complete values that are now available + filter.processParserOutput(parser, callback, errors); + + if (finished) { + this.finished = true; + } + } finally { + SYNC.unlock(); + } + + // Throw any accumulated errors + if (!errors.isEmpty()) { + throw new JqFilterException(errors.get(0), errors); + } + } + + /** + * Signals that all input has been provided. + * + *This is equivalent to calling {@code feedChunk("", true)} and is + * useful when you've already fed all chunks but haven't signaled + * end of input.
+ * + * @throws JqFilterException if a processing error occurs + */ + public void finish() { + if (!finished) { + feedChunk("", true); + } + } + + /** + * Returns whether the processor has finished receiving input. + * + * @return true if finish() was called or feedChunk was called with finished=true + */ + public boolean isFinished() { + return finished; + } + + private void ensureOpen() { + if (closed) { + throw new IllegalStateException("JqStreamProcessor has been closed"); + } + } + + /** + * Releases the stream processor resources. + * + *If input processing wasn't finished, this will attempt to finish it first.
+ */ + @Override + public void close() { + if (!closed) { + SYNC.lock(); + try { + if (!closed) { + // Clear error callback + lib.jq_set_error_cb(filter.getJq(), null, null); + + LOGGER.log(FINE, "Releasing stream parser"); + lib.jv_parser_free(parser); + closed = true; + LOGGER.log(FINE, "Stream parser released"); + } + } finally { + SYNC.unlock(); + } + } + } +} diff --git a/src/test/java/com/arakelian/jq/JqFilterTest.java b/src/test/java/com/arakelian/jq/JqFilterTest.java new file mode 100644 index 0000000..06aea4b --- /dev/null +++ b/src/test/java/com/arakelian/jq/JqFilterTest.java @@ -0,0 +1,449 @@ +/* + * 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 com.arakelian.jq; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.File; +import java.net.URL; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; + +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +import com.arakelian.jq.JqRequest.Indent; + +/** + * Tests for the streaming API (JqFilter, JqStreamProcessor). + */ +public class JqFilterTest { + private static JqLibrary library; + private static File modulePath; + + @BeforeAll + public static void setup() { + library = ImmutableJqLibrary.of(); + + // Find module path for tests that need imports + URL a_jq = JqFilterTest.class.getClassLoader().getResource("modules/a.jq"); + if (a_jq != null) { + modulePath = new File(a_jq.getFile()).getParentFile(); + } + } + + @Test + public void testSimpleFilter() { + try (JqFilter filter = JqFilter.compile(library, ".foo")) { + List