Fixes issue 8

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

@@ -0,0 +1,95 @@
+/*--------------------------------------------------------------------------
+ *  Copyright 2011 Taro L. Saito
+ *
+ *  Licensed 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.
+ *--------------------------------------------------------------------------*/
+//--------------------------------------
+// XerialJ
+//
+// SnappyCodec.java
+// Since: 2011/04/03 14:50:20
+//
+// $URL$
+// $Author$
+//--------------------------------------
+package org.xerial.snappy;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Arrays;
+
+/**
+ * Preamble header for {@link SnappyOutputStream}
+ * 
+ * @author leo
+ * 
+ */
+public class SnappyCodec
+{
+    public static final byte[] MAGIC_HEADER               = new byte[] { -126, 'S', 'N', 'A', 'P', 'P', 'Y', 0 };
+    public static final int    MAGIC_LEN                  = 8;
+
+    public static final int    DEFAULT_VERSION            = 1;
+    public static final int    MINIMUM_COMPATIBLE_VERSION = 1;
+
+    public final byte[]        magic;
+    public final int           version;
+    public final int           compatibleVersion;
+
+    private SnappyCodec(byte[] magic, int version, int compatibleVersion) {
+        this.magic = magic;
+        this.version = version;
+        this.compatibleVersion = compatibleVersion;
+    }
+
+    @Override
+    public String toString() {
+        return String.format("version:%d, compatible version:%d", version, compatibleVersion);
+    }
+
+    public static int headerSize() {
+        return MAGIC_LEN + 4 * 2;
+    }
+
+    public void writeHeader(OutputStream out) throws IOException {
+        ByteArrayOutputStream header = new ByteArrayOutputStream();
+        DataOutputStream d = new DataOutputStream(header);
+        d.write(magic, 0, MAGIC_LEN);
+        d.writeInt(version);
+        d.writeInt(compatibleVersion);
+        d.close();
+        out.write(header.toByteArray(), 0, header.size());
+    }
+
+    public boolean isValidMagicHeader() {
+        return Arrays.equals(MAGIC_HEADER, magic);
+    }
+
+    public static SnappyCodec readHeader(InputStream in) throws IOException {
+        DataInputStream d = new DataInputStream(in);
+        byte[] magic = new byte[MAGIC_LEN];
+        d.read(magic, 0, MAGIC_LEN);
+        int version = d.readInt();
+        int compatibleVersion = d.readInt();
+        return new SnappyCodec(magic, version, compatibleVersion);
+    }
+
+    public static SnappyCodec currentHeader() {
+        return new SnappyCodec(MAGIC_HEADER, DEFAULT_VERSION, MINIMUM_COMPATIBLE_VERSION);
+    }
+
+}

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

@@ -24,6 +24,7 @@
 //--------------------------------------
 package org.xerial.snappy;
 
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 
@@ -39,9 +40,8 @@ public class SnappyInputStream extends InputStream
     private boolean             finishedReading    = false;
     private int                 blockSize          = SnappyOutputStream.DEFAULT_BLOCK_SIZE;
 
-    private byte[]              compressed         = new byte[blockSize];
-
-    private byte[]              uncompressed       = new byte[blockSize];
+    private byte[]              compressed;
+    private byte[]              uncompressed;
     private int                 uncompressedCursor = 0;
     private int                 uncompressedLimit  = 0;
 
@@ -49,6 +49,64 @@ public class SnappyInputStream extends InputStream
 
     public SnappyInputStream(InputStream input) throws IOException {
         this.in = input;
+        readHeader();
+
+        if (compressed == null)
+            compressed = new byte[blockSize];
+        if (uncompressed == null)
+            uncompressed = new byte[blockSize];
+    }
+
+    protected void readHeader() throws IOException {
+        byte[] header = new byte[SnappyCodec.headerSize()];
+        int readBytes = in.read(header, 0, header.length);
+        if (readBytes < header.length) {
+            // do the default uncompression
+            readFully(header, readBytes);
+            return;
+        }
+        SnappyCodec codec = SnappyCodec.readHeader(new ByteArrayInputStream(header));
+        if (codec.isValidMagicHeader()) {
+            // compressed by SnappyOutputStream
+            if (codec.version < SnappyCodec.MINIMUM_COMPATIBLE_VERSION) {
+                throw new IOException(String.format(
+                        "compressed with imcompatible codec version %d. At least version %d is required",
+                        codec.version, SnappyCodec.MINIMUM_COMPATIBLE_VERSION));
+            }
+        }
+        else {
+            // (probably) compressed by Snappy.compress(byte[])
+            readFully(header, readBytes);
+            return;
+        }
+    }
+
+    protected void readFully(byte[] fragment, int fragmentLength) throws IOException {
+        // read the entire input data to the buffer 
+        compressed = new byte[Math.max(blockSize, fragmentLength)];
+        System.arraycopy(fragment, 0, compressed, 0, fragmentLength);
+        int cursor = fragmentLength;
+        for (int readBytes = 0; (readBytes = in.read(compressed, cursor, compressed.length - cursor)) != -1;) {
+            cursor += readBytes;
+            if (cursor >= compressed.length) {
+                byte[] newBuf = new byte[(compressed.length * 2)];
+                System.arraycopy(compressed, 0, newBuf, 0, compressed.length);
+                compressed = newBuf;
+            }
+        }
+
+        // Uncompress
+        try {
+            int uncompressedLength = Snappy.uncompressedLength(compressed, 0, cursor);
+            uncompressed = new byte[uncompressedLength];
+            Snappy.uncompress(compressed, 0, cursor, uncompressed, 0);
+            this.uncompressedCursor = 0;
+            this.uncompressedLimit = uncompressedLength;
+        }
+        catch (SnappyException e) {
+            throw new IOException(e);
+        }
+
     }
 
     @Override

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

@@ -59,6 +59,11 @@ public class SnappyOutputStream extends OutputStream
         this.blockSize = blockSize;
         uncompressed = new byte[blockSize];
         compressed = new byte[Snappy.maxCompressedLength(blockSize)];
+        writeHeader();
+    }
+
+    protected void writeHeader() throws IOException {
+        SnappyCodec.currentHeader().writeHeader(out);
     }
 
     @Override

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

@@ -73,4 +73,16 @@ public class SnappyInputStreamTest
         assertArrayEquals(orig, uncompressed);
 
     }
+
+    @Test
+    public void readBlockCompressedData() throws Exception {
+        byte[] orig = readResourceFile("alice29.txt");
+        byte[] compressed = Snappy.compress(orig);
+
+        SnappyInputStream in = new SnappyInputStream(new ByteArrayInputStream(compressed));
+        byte[] uncompressed = readFully(in);
+
+        assertEquals(orig.length, uncompressed.length);
+        assertArrayEquals(orig, uncompressed);
+    }
 }

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

@@ -61,7 +61,7 @@ public class SnappyOutputStreamTest
         ByteArrayOutputStream decompressed = new ByteArrayOutputStream();
         byte[] compressed = buf.toByteArray();
         // decompress
-        for (int cursor = 0; cursor < compressed.length;) {
+        for (int cursor = SnappyCodec.headerSize(); cursor < compressed.length;) {
             int chunkSize = SnappyOutputStream.readInt(compressed, cursor);
             cursor += 4;
             byte[] tmpOut = new byte[Snappy.uncompressedLength(compressed, cursor, chunkSize)];