updating tribble to support path wrappers

src/main/java/htsjdk/samtools/SamReaderFactory.java

@@ -77,8 +77,6 @@
 
     private static ValidationStringency defaultValidationStringency = ValidationStringency.DEFAULT_STRINGENCY;
 
-    private Function<SeekableByteChannel, SeekableByteChannel> pathWrapper = Function.identity();
-
     abstract public SamReader open(final File file);
 
     /**

src/main/java/htsjdk/samtools/seekablestream/ISeekableStreamFactory.java

@@ -2,6 +2,8 @@
 
 import java.io.IOException;
 import java.net.URL;
+import java.nio.channels.SeekableByteChannel;
+import java.util.function.Function;
 
 /**
  * Factory for creating {@link SeekableStream}s based on URLs/paths.
@@ -30,4 +32,24 @@
      * @return
      */
     public SeekableStream getBufferedStream(SeekableStream stream, int bufferSize);
+
+    /**
+     * Open a stream from the input path, applying the wrapper to the stream.
+     *
+     * The wrapper allows applying operations directly to the byte stream so that things like caching, prefetching, or decryption
+     * can be done at the raw byte level.
+     *
+     * The default implementation throws if wrapper != null, but implementations may support this wrapping operation
+     *
+     * @param path    a uri like String representing a resource to open
+     * @param wrapper a wrapper to apply to the stream
+     * @return a stream opened path
+     */
+    default SeekableStream getStreamFor(String path, Function<SeekableByteChannel, SeekableByteChannel> wrapper) throws IOException {
+        if(wrapper != null) {
+            throw new UnsupportedOperationException("This factory doesn't support adding wrappers");
+        } else {
+            return this.getStreamFor(path);
+        }
+    }
 }

src/main/java/htsjdk/samtools/seekablestream/SeekableStreamFactory.java

@@ -27,6 +27,8 @@
 import java.io.File;
 import java.io.IOException;
 import java.net.URL;
+import java.nio.channels.SeekableByteChannel;
+import java.util.function.Function;
 
 /**
  * Singleton class for getting {@link SeekableStream}s from URL/paths
@@ -65,11 +67,27 @@ public static boolean isFilePath(final String path) {
 
     private static class DefaultSeekableStreamFactory implements ISeekableStreamFactory {
 
+        @Override
         public SeekableStream getStreamFor(final URL url) throws IOException {
             return getStreamFor(url.toExternalForm());
         }
 
+        @Override
         public SeekableStream getStreamFor(final String path) throws IOException {
+            return getStreamFor(path, null);
+        }
+
+        /**
+         * The wrapper will only be applied to the stream if the stream is treated as a {@link java.nio.file.Path}
+         *
+         * This currently means any uri with a scheme that is not http, https, ftp, or file will have the wrapper applied to it
+         *
+         * @param path    a uri like String representing a resource to open
+         * @param wrapper a wrapper to apply to the stream allowing direct transformations on the byte stream to be applied
+         */
+        @Override
+        public SeekableStream getStreamFor(final String path,
+                                           Function<SeekableByteChannel, SeekableByteChannel> wrapper) throws IOException {
             // todo -- add support for SeekableBlockInputStream
 
             if (path.startsWith("http:") || path.startsWith("https:")) {
@@ -80,16 +98,18 @@ public SeekableStream getStreamFor(final String path) throws IOException {
             } else if (path.startsWith("file:")) {
                 return new SeekableFileStream(new File(new URL(path).getPath()));
             } else if (IOUtil.hasScheme(path)) {
-                return new SeekablePathStream(IOUtil.getPath(path));
+                return new SeekablePathStream(IOUtil.getPath(path), wrapper);
             } else {
                 return new SeekableFileStream(new File(path));
             }
         }
 
+        @Override
         public SeekableStream getBufferedStream(SeekableStream stream){
             return getBufferedStream(stream, SeekableBufferedStream.DEFAULT_BUFFER_SIZE);
         }
 
+        @Override
         public SeekableStream getBufferedStream(SeekableStream stream, int bufferSize){
             if (bufferSize == 0) return stream;
             else return new SeekableBufferedStream(stream, bufferSize);

src/main/java/htsjdk/samtools/util/BlockCompressedInputStream.java

@@ -407,7 +407,7 @@ private void checkAndRethrowDecompressionException() throws IOException {
     /**
      * Attempt to reuse the buffer of the given block 
      * @param block owning block
-     * @return null decompressiong buffer to resuse, null if no buffer is available
+     * @return null decompressing buffer to reuse, null if no buffer is available
      */
     private byte[] getBufferForReuse(DecompressedBlock block) {
         if (block == null) return null;

src/main/java/htsjdk/tribble/AbstractFeatureReader.java

@@ -25,11 +25,13 @@
 import java.io.File;
 import java.io.IOException;
 import java.net.URI;
+import java.nio.channels.SeekableByteChannel;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.Set;
+import java.util.function.Function;
 
 /**
  * jrobinso
@@ -43,6 +45,11 @@
     // the path to underlying data source
     String path;
 
+    // a wrapper to apply to the raw stream of the Feature file to allow features like prefetching and caching to be injected
+    final Function<SeekableByteChannel, SeekableByteChannel> wrapper;
+    // a wrapper to apply to the raw stream of the index file
+    final Function<SeekableByteChannel, SeekableByteChannel> indexWrapper;
+
     // the query source, codec, and header
     // protected final QuerySource querySource;
     protected final FeatureCodec<T, SOURCE> codec;
@@ -60,38 +67,51 @@
     }
 
     /**
-     * {@link #getFeatureReader(String, String, FeatureCodec, boolean)} with {@code null} for indexResource
+     * {@link #getFeatureReader(String, String, FeatureCodec, boolean, Function, Function)} with {@code null} for indexResource, wrapper, and indexWrapper
      * @throws TribbleException
      */
     public static <FEATURE extends Feature, SOURCE> AbstractFeatureReader<FEATURE, SOURCE> getFeatureReader(final String featureResource, final FeatureCodec<FEATURE, SOURCE> codec, final boolean requireIndex) throws TribbleException {
-        return getFeatureReader(featureResource, null, codec, requireIndex);
+        return getFeatureReader(featureResource, null, codec, requireIndex, null, null);
+    }
+
+
+    /**
+     * {@link #getFeatureReader(String, String, FeatureCodec, boolean, Function, Function)} with {@code null} for wrapper, and indexWrapper
+     * @throws TribbleException
+     */
+    public static <FEATURE extends Feature, SOURCE> AbstractFeatureReader<FEATURE, SOURCE> getFeatureReader(final String featureResource, String indexResource, final FeatureCodec<FEATURE, SOURCE> codec, final boolean requireIndex) throws TribbleException {
+        return getFeatureReader(featureResource, indexResource, codec, requireIndex, null, null);
     }
 
     /**
      *
      * @param featureResource the feature file to create from
      * @param indexResource   the index for the feature file. If null, will auto-generate (if necessary)
-     * @param codec
+     * @param codec           the codec to use to decode the individual features
      * @param requireIndex    whether an index is required for this file
-     * @return
+     * @param wrapper         a wrapper to apply to the byte stream from the featureResource allowing injecting features
+     *                        like caching and prefetching of the stream, may be null, will only be applied if featureResource
+     *                        is a uri representing a {@link java.nio.file.Path}
+     * @param indexWrapper    a wrapper to apply to the byte stream from the indexResource, may be null, will only be
+     *                        applied if indexResource is a uri representing a {@link java.nio.file.Path}
+     *
      * @throws TribbleException
      */
-    public static <FEATURE extends Feature, SOURCE> AbstractFeatureReader<FEATURE, SOURCE> getFeatureReader(final String featureResource, String indexResource, final FeatureCodec<FEATURE, SOURCE> codec, final boolean requireIndex) throws TribbleException {
-
+    public static <FEATURE extends Feature, SOURCE> AbstractFeatureReader<FEATURE, SOURCE> getFeatureReader(final String featureResource, String indexResource, final FeatureCodec<FEATURE, SOURCE> codec, final boolean requireIndex, Function<SeekableByteChannel, SeekableByteChannel> wrapper, Function<SeekableByteChannel, SeekableByteChannel> indexWrapper) throws TribbleException {
         try {
             // Test for tabix index
             if (methods.isTabix(featureResource, indexResource)) {
                 if ( ! (codec instanceof AsciiFeatureCodec) )
                     throw new TribbleException("Tabix indexed files only work with ASCII codecs, but received non-Ascii codec " + codec.getClass().getSimpleName());
-                return new TabixFeatureReader<FEATURE, SOURCE>(featureResource, indexResource, (AsciiFeatureCodec) codec);
+                return new TabixFeatureReader<>(featureResource, indexResource, (AsciiFeatureCodec) codec, wrapper, indexWrapper);
             }
             // Not tabix => tribble index file (might be gzipped, but not block gzipped)
             else {
-                return new TribbleIndexedFeatureReader<FEATURE, SOURCE>(featureResource, indexResource, codec, requireIndex);
+                return new TribbleIndexedFeatureReader<>(featureResource, indexResource, codec, requireIndex, wrapper, indexWrapper);
             }
-        } catch (IOException e) {
+        } catch (final IOException e) {
             throw new TribbleException.MalformedFeatureFile("Unable to create BasicFeatureReader using feature file ", featureResource, e);
-        } catch (TribbleException e) {
+        } catch (final TribbleException e) {
             e.setSource(featureResource);
             throw e;
         }
@@ -108,16 +128,24 @@
      */
     public static <FEATURE extends Feature, SOURCE> AbstractFeatureReader<FEATURE, SOURCE> getFeatureReader(final String featureResource, final FeatureCodec<FEATURE, SOURCE>  codec, final Index index) throws TribbleException {
         try {
-            return new TribbleIndexedFeatureReader<FEATURE, SOURCE>(featureResource, codec, index);
-        } catch (IOException e) {
+            return new TribbleIndexedFeatureReader<>(featureResource, codec, index);
+        } catch (final IOException e) {
             throw new TribbleException.MalformedFeatureFile("Unable to create AbstractFeatureReader using feature file ", featureResource, e);
         }
 
     }
 
     protected AbstractFeatureReader(final String path, final FeatureCodec<T, SOURCE> codec) {
+        this(path, codec, null, null);
+    }
+
+    protected AbstractFeatureReader(final String path, final FeatureCodec<T, SOURCE> codec,
+                                    final Function<SeekableByteChannel, SeekableByteChannel> wrapper,
+                                    final Function<SeekableByteChannel, SeekableByteChannel> indexWrapper) {
         this.path = path;
         this.codec = codec;
+        this.wrapper = wrapper;
+        this.indexWrapper = indexWrapper;
     }
 
     /**
@@ -169,25 +197,30 @@ public static boolean hasBlockCompressedExtension (final URI uri) {
      *
      * @return the header object we've read-in
      */
+    @Override
     public Object getHeader() {
         return header.getHeaderValue();
     }
 
     static class EmptyIterator<T extends Feature> implements CloseableTribbleIterator<T> {
-        public Iterator iterator() { return this; }
-        public boolean hasNext() { return false; }
-        public T next() { return null; }
-        public void remove() { }
+        @Override public Iterator<T> iterator() { return this; }
+        @Override public boolean hasNext() { return false; }
+        @Override public T next() { return null; }
+        @Override public void remove() { }
         @Override public void close() { }
     }
 
+    public static boolean isTabix(String resourcePath, String indexPath) throws IOException {
+        if(indexPath == null){
+            indexPath = ParsingUtils.appendToPath(resourcePath, TabixUtils.STANDARD_INDEX_EXTENSION);
+        }
+        return hasBlockCompressedExtension(resourcePath) && ParsingUtils.resourceExists(indexPath);
+    }
+
     public static class ComponentMethods{
 
         public boolean isTabix(String resourcePath, String indexPath) throws IOException{
-            if(indexPath == null){
-                indexPath = ParsingUtils.appendToPath(resourcePath, TabixUtils.STANDARD_INDEX_EXTENSION);
-            }
-            return hasBlockCompressedExtension(resourcePath) && ParsingUtils.resourceExists(indexPath);
+            return AbstractFeatureReader.isTabix(resourcePath, indexPath);
         }
     }
 }

src/main/java/htsjdk/tribble/FeatureCodec.java

@@ -125,7 +125,7 @@
      * Define the tabix format for the feature, used for indexing. Default implementation throws an exception.
      *
      * Note that only {@link AsciiFeatureCodec} could read tabix files as defined in
-     * {@link AbstractFeatureReader#getFeatureReader(String, String, FeatureCodec, boolean)}
+     * {@link AbstractFeatureReader#getFeatureReader(String, String, FeatureCodec, boolean, java.util.function.Function, java.util.function.Function)}
      *
      * @return the format to use with tabix
      * @throws TribbleException if the format is not defined