Add rawWrite and rawRead API to the SnappyOutputStream and SnappyInputStream to support primitive array compression/decompression

src/main/java/org/xerial/snappy/Snappy.java

@@ -42,6 +42,25 @@ public class Snappy
         impl = SnappyLoader.load();
     }
 
+    /**
+     * Copy bytes from source to destination
+     * 
+     * @param src
+     *            pointer to the source array
+     * @param offset
+     *            byte offset in the source array
+     * @param byteLength
+     *            the number of bytes to copy
+     * @param dest
+     *            pointer to the destination array
+     * @param dest_offset
+     *            byte offset in the destination array
+     * @throws IOException
+     */
+    static void arrayCopy(Object src, int offset, int byteLength, Object dest, int dest_offset) throws IOException {
+        impl.arrayCopy(src, offset, byteLength, dest, dest_offset);
+    }
+
     /**
      * High-level API for compressing the input byte array. This method performs
      * array copy to generate the result. If you want to reduce the memory copy

src/main/java/org/xerial/snappy/SnappyInputStream.java

@@ -105,8 +105,12 @@ public class SnappyInputStream extends InputStream
 
     @Override
     public int read(byte[] b, int off, int len) throws IOException {
+        return rawRead(b, off, len);
+    }
+
+    public int rawRead(Object array, int byteOffset, int byteLength) throws IOException {
         int writtenBytes = 0;
-        for (; writtenBytes < len;) {
+        for (; writtenBytes < byteLength;) {
             if (uncompressedCursor >= uncompressedLimit) {
                 if (hasNextChunk())
                     continue;
@@ -114,8 +118,8 @@ public class SnappyInputStream extends InputStream
                     return writtenBytes == 0 ? -1 : writtenBytes;
                 }
             }
-            int bytesToWrite = Math.min(uncompressedLimit - uncompressedCursor, len - writtenBytes);
-            System.arraycopy(uncompressed, uncompressedCursor, b, off + writtenBytes, bytesToWrite);
+            int bytesToWrite = Math.min(uncompressedLimit - uncompressedCursor, byteLength - writtenBytes);
+            Snappy.arrayCopy(uncompressed, uncompressedCursor, bytesToWrite, array, byteOffset + writtenBytes);
             writtenBytes += bytesToWrite;
             uncompressedCursor += bytesToWrite;
         }
@@ -123,6 +127,24 @@ public class SnappyInputStream extends InputStream
         return writtenBytes;
     }
 
+    /**
+     * @param d
+     *            input
+     * @param off
+     *            offset
+     * @param len
+     *            the number of long elements to read
+     * @return written bytes
+     * @throws IOException
+     */
+    public int read(long[] d, int off, int len) throws IOException {
+        return rawRead(d, off * 8, len * 8);
+    }
+
+    public int read(long[] d) throws IOException {
+        return read(d, 0, d.length);
+    }
+
     protected boolean hasNextChunk() throws IOException {
         if (finishedReading)
             return false;

src/main/java/org/xerial/snappy/SnappyNative.cpp

@@ -14,6 +14,7 @@
  *  limitations under the License.
  *--------------------------------------------------------------------------*/
 #include <string>
+#include <cstring>
 #include <snappy.h>
 #include "SnappyNative.h"
 
@@ -216,3 +217,21 @@ JNIEXPORT jboolean JNICALL Java_org_xerial_snappy_SnappyNative_isValidCompressed
 	return ret;
 }
 
+JNIEXPORT void JNICALL Java_org_xerial_snappy_SnappyNative_arrayCopy
+  (JNIEnv * env, jobject self, jobject input, jint offset, jint length, jobject output, jint output_offset)
+{
+	char* src = (char*) env->GetPrimitiveArrayCritical((jarray) input, 0);
+	char* dest = (char*) env->GetPrimitiveArrayCritical((jarray) output, 0);
+	if(src == 0 || dest == 0) {
+		// out of memory
+		throw_exception(env, self, 4);
+		return;
+	}
+
+	memcpy(dest+output_offset, src+offset, (size_t) length);
+
+	env->ReleasePrimitiveArrayCritical((jarray) input, src, 0);
+	env->ReleasePrimitiveArrayCritical((jarray) output, dest, 0);
+}
+
+

src/main/java/org/xerial/snappy/SnappyNative.h

@@ -87,6 +87,14 @@ JNIEXPORT jboolean JNICALL Java_org_xerial_snappy_SnappyNative_isValidCompressed
 JNIEXPORT jboolean JNICALL Java_org_xerial_snappy_SnappyNative_isValidCompressedBuffer__Ljava_lang_Object_2II
   (JNIEnv *, jobject, jobject, jint, jint);
 
+/*
+ * Class:     org_xerial_snappy_SnappyNative
+ * Method:    arrayCopy
+ * Signature: (Ljava/lang/Object;IILjava/lang/Object;I)V
+ */
+JNIEXPORT void JNICALL Java_org_xerial_snappy_SnappyNative_arrayCopy
+  (JNIEnv *, jobject, jobject, jint, jint, jobject, jint);
+
 #ifdef __cplusplus
 }
 #endif

src/main/java/org/xerial/snappy/SnappyNative.java

@@ -68,6 +68,8 @@ public class SnappyNative implements SnappyNativeAPI
 
     public native boolean isValidCompressedBuffer(Object input, int offset, int len) throws IOException;
 
+    public native void arrayCopy(Object src, int offset, int byteLength, Object dest, int dOffset) throws IOException;
+
     public void throw_error(int errorCode) throws IOException {
         throw new IOException(String.format("%s(%d)", SnappyErrorCode.getErrorMessage(errorCode), errorCode));
     }

src/main/java/org/xerial/snappy/SnappyNativeAPI.java

@@ -66,6 +66,8 @@ public interface SnappyNativeAPI
 
     public boolean isValidCompressedBuffer(Object input, int offset, int len) throws IOException;
 
+    public void arrayCopy(Object src, int offset, int byteLength, Object dest, int dOffset) throws IOException;
+
     public void throw_error(int errorCode) throws IOException;
 
 }

src/main/java/org/xerial/snappy/SnappyOutputStream.java

@@ -34,8 +34,13 @@ import java.io.OutputStream;
  * The input data is blocked into 32kb size (in default), and each block is
  * compressed and then passed to the given {@link OutputStream}.
  * 
- * The output data format is a sequence of (compressed data size, compressed
- * data...) pair.
+ * The output data format is:
+ * <ol>
+ * <li>snappy codec header defined in {@link SnappyCodec}
+ * <li>a pair of (compressed data size, compressed data...)
+ * <li>a pair of (compressed data size, compressed data...)
+ * <li>...
+ * </ol>
  * 
  * Note that the compressed data created by {@link SnappyOutputStream} cannot be
  * uncompressed by {@link Snappy#uncompress(byte[])} since the output formats of
@@ -74,10 +79,49 @@ public class SnappyOutputStream extends OutputStream
 
     @Override
     public void write(byte[] b, int off, int len) throws IOException {
+        rawWrite(b, off, len);
+    }
 
-        for (int readBytes = 0; readBytes < len;) {
-            int copyLen = Math.min(uncompressed.length - cursor, len - readBytes);
-            System.arraycopy(b, off + readBytes, uncompressed, cursor, copyLen);
+    /**
+     * Compress the input long array data
+     * 
+     * @param d
+     *            input array
+     * @param off
+     *            offset in the array
+     * @param len
+     *            the number of elements in the array to copy
+     * @throws IOException
+     */
+    public void write(long[] d, int off, int len) throws IOException {
+        rawWrite(d, off * 8, len * 8);
+    }
+
+    public void write(float[] f, int off, int len) throws IOException {
+        rawWrite(f, off * 4, len * 4);
+    }
+
+    public void write(long[] d) throws IOException {
+        write(d, 0, d.length);
+    }
+
+    public void write(float[] f) throws IOException {
+        write(f, 0, f.length);
+    }
+
+    /**
+     * Compress the raw byte array data.
+     * 
+     * @param array
+     *            array data of any type (e.g., byte[], float[], long[], ...)
+     * @param byteOffset
+     * @param byteLength
+     * @throws IOException
+     */
+    public void rawWrite(Object array, int byteOffset, int byteLength) throws IOException {
+        for (int readBytes = 0; readBytes < byteLength;) {
+            int copyLen = Math.min(uncompressed.length - cursor, byteLength - readBytes);
+            Snappy.arrayCopy(array, byteOffset + readBytes, copyLen, uncompressed, cursor);
             readBytes += copyLen;
             cursor += copyLen;
 
@@ -101,14 +145,14 @@ public class SnappyOutputStream extends OutputStream
         out.flush();
     }
 
-    public static void writeInt(OutputStream out, int value) throws IOException {
+    static void writeInt(OutputStream out, int value) throws IOException {
         out.write((value >> 24) & 0xFF);
         out.write((value >> 16) & 0xFF);
         out.write((value >> 8) & 0xFF);
         out.write((value >> 0) & 0xFF);
     }
 
-    public static int readInt(byte[] buffer, int pos) {
+    static int readInt(byte[] buffer, int pos) {
         int b1 = (buffer[pos] & 0xFF) << 24;
         int b2 = (buffer[pos + 1] & 0xFF) << 16;
         int b3 = (buffer[pos + 2] & 0xFF) << 8;

src/test/java/org/xerial/snappy/SnappyOutputStreamTest.java

@@ -95,4 +95,26 @@ public class SnappyOutputStreamTest
         is.close();
     }
 
+    @Test
+    public void longArrayCompress() throws Exception {
+        long[] l = new long[10];
+        for (int i = 0; i < l.length; ++i) {
+            l[i] = i % 3 + i * 11;
+        }
+
+        ByteArrayOutputStream b = new ByteArrayOutputStream();
+        SnappyOutputStream os = new SnappyOutputStream(b);
+
+        os.write(l);
+        os.close();
+        SnappyInputStream is = new SnappyInputStream(new ByteArrayInputStream(b.toByteArray()));
+        long[] l2 = new long[10];
+        int readBytes = is.read(l2);
+        is.close();
+
+        assertEquals(10 * 8, readBytes);
+        assertArrayEquals(l, l2);
+
+    }
+
 }