Added test cases for reading various types of MessageBufferInputs, and fixed some errors

xerial · xerial · commit c96303782553 · 2014-05-30T16:53:23.000+09:00
diff --git a/msgpack-core/src/main/java/org/msgpack/core/buffer/ByteBufferInput.java b/msgpack-core/src/main/java/org/msgpack/core/buffer/ByteBufferInput.java
@@ -0,0 +1,34 @@
+package org.msgpack.core.buffer;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import static org.msgpack.core.Preconditions.*;
+
+/**
+ * {@link MessageBufferInput} adapter for {@link java.nio.ByteBuffer}
+ */
+public class ByteBufferInput implements MessageBufferInput {
+
+    private final ByteBuffer input;
+    private boolean isRead = false;
+
+    public ByteBufferInput(ByteBuffer input) {
+        this.input = checkNotNull(input, "input ByteBuffer is null");
+    }
+
+
+    @Override
+    public MessageBuffer next() throws IOException {
+        if(isRead)
+            return null;
+
+        isRead = true;
+        return MessageBuffer.wrap(input);
+    }
+
+
+    @Override
+    public void close() throws IOException {
+        // Nothing to do
+    }
+}
diff --git a/msgpack-core/src/main/java/org/msgpack/core/buffer/ChannelBufferInput.java b/msgpack-core/src/main/java/org/msgpack/core/buffer/ChannelBufferInput.java
@@ -32,25 +32,15 @@ public MessageBuffer next() throws IOException {
             return null;
 
         MessageBuffer m = MessageBuffer.newBuffer(bufferSize);
-        ByteBuffer b = m.toByteBuffer(0, m.size);
-        while(b.remaining() > 0) {
-            int ret = 0;
-            ret = channel.read(b);
+        ByteBuffer b = m.toByteBuffer();
+        while(!reachedEOF && b.remaining() > 0) {
+            int ret = channel.read(b);
             if(ret == -1) {
                 reachedEOF = true;
-                break;
             }
         }
         b.flip();
-        if(b.remaining() == 0) {
-            return null;
-        }
-        else if(b.limit() < m.size) {
-            return m.slice(0, b.limit());
-        }
-        else {
-            return m;
-        }
+        return b.remaining() == 0 ? null : m.slice(0, b.limit());
     }
 
     @Override
diff --git a/msgpack-core/src/main/java/org/msgpack/core/buffer/InputStreamBufferInput.java b/msgpack-core/src/main/java/org/msgpack/core/buffer/InputStreamBufferInput.java
@@ -13,8 +13,6 @@
  */
 public class InputStreamBufferInput implements MessageBufferInput {
 
-    private final InputStream in;
-    private final int bufferSize;
 
     private static Field bufField;
     private static Field bufPosField;
@@ -38,6 +36,10 @@ private static Field getField(String name) {
         bufCountField = getField("count");
     }
 
+    private final InputStream in;
+    private final int bufferSize;
+    private boolean reachedEOF = false;
+
     public static MessageBufferInput newBufferInput(InputStream in) {
         if(in instanceof ByteArrayInputStream) {
             ByteArrayInputStream b = (ByteArrayInputStream) in;
@@ -69,23 +71,24 @@ public InputStreamBufferInput(InputStream in, int bufferSize) {
 
     @Override
     public MessageBuffer next() throws IOException {
+        if(reachedEOF)
+            return null;
+
         byte[] buffer = null;
         int cursor = 0;
-        while(cursor < bufferSize) {
+        while(!reachedEOF && cursor < bufferSize) {
             if(buffer == null)
                 buffer = new byte[bufferSize];
+
             int readLen = in.read(buffer, cursor, bufferSize - cursor);
             if(readLen == -1) {
+                reachedEOF = true;
                 break;
             }
             cursor += readLen;
         }
-        if(buffer == null)
-            return null;
-        else {
-            MessageBuffer m = MessageBuffer.wrap(buffer);
-            return cursor == bufferSize ? m : m.slice(0, cursor);
-        }
+
+        return buffer == null ? null : MessageBuffer.wrap(buffer).slice(0, cursor);
     }
 
     @Override
diff --git a/msgpack-core/src/main/java/org/msgpack/core/buffer/MessageBuffer.java b/msgpack-core/src/main/java/org/msgpack/core/buffer/MessageBuffer.java
@@ -259,7 +259,10 @@ private MessageBuffer(Object base, long address, int length, ByteBuffer referenc
 
     public MessageBuffer slice(int offset, int length) {
         // TODO ensure deleting this slice does not collapse this MessageBuffer
-        return new MessageBuffer(base, address + offset, length, reference);
+        if(offset == 0 && length == size())
+            return this;
+        else
+            return new MessageBuffer(base, address + offset, length, reference);
     }
 
     public byte getByte(int index) {
@@ -388,6 +391,17 @@ public ByteBuffer toByteBuffer(int index, int length) {
         }
     }
 
+    public ByteBuffer toByteBuffer() {
+        return toByteBuffer(0, size());
+    }
+
+    public byte[] toByteArray() {
+        byte[] b = new byte[size()];
+        unsafe.copyMemory(base, address, b, ARRAY_BYTE_BASE_OFFSET, size());
+        return b;
+    }
+
+
     public void relocate(int offset, int length, int dst) {
         unsafe.copyMemory(base, address + offset, base, address+dst, length);
     }
diff --git a/msgpack-core/src/test/scala/org/msgpack/core/MessagePackSpec.scala b/msgpack-core/src/test/scala/org/msgpack/core/MessagePackSpec.scala
@@ -4,8 +4,17 @@ import org.scalatest._
 import xerial.core.log.{LogLevel, Logger}
 import xerial.core.util.{TimeReport, Timer}
 import scala.language.implicitConversions
-
-trait MessagePackSpec extends WordSpec with Matchers with GivenWhenThen with OptionValues with BeforeAndAfter with Benchmark with Logger {
+import org.scalatest.prop.PropertyChecks
+
+trait MessagePackSpec
+  extends WordSpec
+  with Matchers
+  with GivenWhenThen
+  with OptionValues
+  with BeforeAndAfter
+  with PropertyChecks
+  with Benchmark
+  with Logger {
 
   implicit def toTag(s:String) : Tag = Tag(s)
 
diff --git a/msgpack-core/src/test/scala/org/msgpack/core/MessagePackTest.scala b/msgpack-core/src/test/scala/org/msgpack/core/MessagePackTest.scala
@@ -9,7 +9,7 @@ import java.math.BigInteger
 /**
  * Created on 2014/05/07.
  */
-class MessagePackTest extends MessagePackSpec with PropertyChecks {
+class MessagePackTest extends MessagePackSpec  {
 
   "MessagePack" should {
     "detect fixint values" in {
diff --git a/msgpack-core/src/test/scala/org/msgpack/core/buffer/MessageBufferInputTest.scala b/msgpack-core/src/test/scala/org/msgpack/core/buffer/MessageBufferInputTest.scala
@@ -0,0 +1,101 @@
+package org.msgpack.core.buffer
+
+import org.msgpack.core.MessagePackSpec
+import java.io._
+import xerial.core.io.IOUtil
+import scala.util.Random
+import java.util.zip.{GZIPOutputStream, GZIPInputStream}
+import java.nio.ByteBuffer
+
+/**
+ * Created on 5/30/14.
+ */
+class MessageBufferInputTest extends MessagePackSpec {
+
+  val targetInputSize = Seq(0, 10, 500, 1000, 2000, 4000, 8000, 10000, 30000, 50000, 100000)
+
+  def testData(size:Int) = {
+    //debug(s"test data size: ${size}")
+    val b = new Array[Byte](size)
+    Random.nextBytes(b)
+    b
+  }
+
+  def testDataSet = {
+    targetInputSize.map(testData)
+  }
+  
+  def runTest(factory:Array[Byte] => MessageBufferInput) {
+    for(b <- testDataSet) {
+      checkInputData(b, factory(b))
+    }
+  }
+
+  implicit class InputData(b:Array[Byte]) {
+    def compress = {
+      val compressed = new ByteArrayOutputStream()
+      val out = new GZIPOutputStream(compressed)
+      out.write(b)
+      out.close()
+      compressed.toByteArray
+    }
+
+    def toByteBuffer = {
+      ByteBuffer.wrap(b)
+    }
+
+    def saveToTmpFile : File = {
+      val tmp = File.createTempFile("testbuf", ".dat", new File("target"))
+      tmp.getParentFile.mkdirs()
+      tmp.deleteOnExit()
+      IOUtil.withResource(new FileOutputStream(tmp)) { out =>
+        out.write(b)
+      }
+      tmp
+    }
+  }
+
+
+
+  def checkInputData(inputData:Array[Byte], in:MessageBufferInput) {
+    When(s"input data size = ${inputData.length}")
+    var cursor = 0
+    for(m <- Iterator.continually(in.next).takeWhile(_ != null)) {
+      m.toByteArray() shouldBe inputData.slice(cursor, cursor + m.size())
+      cursor += m.size()
+    }
+    cursor shouldBe inputData.length
+
+  }
+
+
+  "MessageBufferInput" should {
+    "support byte arrays" in {
+      runTest(new ArrayBufferInput(_))
+    }
+
+    "support ByteBuffers" in {
+      runTest(b => new ByteBufferInput(b.toByteBuffer))
+    }
+
+    "support InputStreams" taggedAs("is") in {
+      runTest(b => 
+        new InputStreamBufferInput(
+          new GZIPInputStream(new ByteArrayInputStream(b.compress)))
+      )
+    }
+
+    "support file input channel" taggedAs("fc") in {
+      runTest { b =>
+        val tmp = b.saveToTmpFile
+        try {
+          InputStreamBufferInput.newBufferInput(new FileInputStream(tmp))
+        }
+        finally {
+          tmp.delete()
+        }
+      }
+    }
+
+  }
+}
