Merge pull request #378 from msgpack/directbuf

Add ByteBuffer input support again

Author: frsyuki

msgpack-core/src/main/java/org/msgpack/core/MessagePack.java

@@ -16,6 +16,7 @@
 package org.msgpack.core;
 
 import org.msgpack.core.buffer.ArrayBufferInput;
+import org.msgpack.core.buffer.ByteBufferInput;
 import org.msgpack.core.buffer.ChannelBufferInput;
 import org.msgpack.core.buffer.ChannelBufferOutput;
 import org.msgpack.core.buffer.InputStreamBufferInput;
@@ -25,6 +26,7 @@
 
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.nio.ByteBuffer;
 import java.nio.channels.ReadableByteChannel;
 import java.nio.channels.WritableByteChannel;
 import java.nio.charset.Charset;
@@ -236,6 +238,17 @@ public static MessageUnpacker newDefaultUnpacker(byte[] contents, int offset, in
         return DEFAULT_UNPACKER_CONFIG.newUnpacker(contents, offset, length);
     }
 
+    /**
+     * Create an unpacker that reads the data from a given ByteBuffer
+     *
+     * @param contents
+     * @return
+     */
+    public static MessageUnpacker newDefaultUnpacker(ByteBuffer contents)
+    {
+        return DEFAULT_UNPACKER_CONFIG.newUnpacker(contents);
+    }
+
     /**
      * MessagePacker configuration.
      */
@@ -524,6 +537,17 @@ public MessageUnpacker newUnpacker(byte[] contents, int offset, int length)
             return newUnpacker(new ArrayBufferInput(contents, offset, length));
         }
 
+        /**
+         * Create an unpacker that reads the data from a given ByteBuffer
+         *
+         * @param contents
+         * @return
+         */
+        public MessageUnpacker newUnpacker(ByteBuffer contents)
+        {
+            return newUnpacker(new ByteBufferInput(contents));
+        }
+
         /**
          * Allow unpackBinaryHeader to read str format family  (default: true)
          */

msgpack-core/src/main/java/org/msgpack/core/MessagePacker.java

@@ -532,8 +532,7 @@ else if (s.length() < (1 << 8)) {
                         throw new IllegalArgumentException("Unexpected UTF-8 encoder state");
                     }
                     // move 1 byte backward to expand 3-byte header region to 3 bytes
-                    buffer.putBytes(position + 3,
-                            buffer.array(), buffer.arrayOffset() + position + 2, written);
+                    buffer.putMessageBuffer(position + 3, buffer, position + 2, written);
                     // write 3-byte header
                     buffer.putByte(position++, STR16);
                     buffer.putShort(position, (short) written);
@@ -561,8 +560,7 @@ else if (s.length() < (1 << 16)) {
                         throw new IllegalArgumentException("Unexpected UTF-8 encoder state");
                     }
                     // move 2 bytes backward to expand 3-byte header region to 5 bytes
-                    buffer.putBytes(position + 5,
-                            buffer.array(), buffer.arrayOffset() + position + 3, written);
+                    buffer.putMessageBuffer(position + 5, buffer, position + 3, written);
                     // write 3-byte header header
                     buffer.putByte(position++, STR32);
                     buffer.putInt(position, written);

msgpack-core/src/main/java/org/msgpack/core/MessageUnpacker.java

@@ -213,14 +213,9 @@ private MessageBuffer prepareNumberBuffer(int readLength)
             // fill the temporary buffer from the current data fragment and
             // next fragment(s).
 
-            // TODO buffer.array() doesn't work if MessageBuffer is allocated by
-            //      newDirectBuffer. dd copy method to MessageBuffer to solve this issue.
-
             int off = 0;
             if (remaining > 0) {
-                numberBuffer.putBytes(0,
-                        buffer.array(), buffer.arrayOffset() + position,
-                        remaining);
+                numberBuffer.putMessageBuffer(0, buffer, position, remaining);
                 readLength -= remaining;
                 off += remaining;
             }
@@ -229,16 +224,12 @@ private MessageBuffer prepareNumberBuffer(int readLength)
                 nextBuffer();
                 int nextSize = buffer.size();
                 if (nextSize >= readLength) {
-                    numberBuffer.putBytes(off,
-                            buffer.array(), buffer.arrayOffset(),
-                            readLength);
+                    numberBuffer.putMessageBuffer(off, buffer, 0, readLength);
                     position = readLength;
                     break;
                 }
                 else {
-                    numberBuffer.putBytes(off,
-                            buffer.array(), buffer.arrayOffset(),
-                            nextSize);
+                    numberBuffer.putMessageBuffer(off, buffer, 0, nextSize);
                     readLength -= nextSize;
                     off += nextSize;
                 }
@@ -1041,7 +1032,8 @@ private void handleCoderError(CoderResult cr)
     private String decodeStringFastPath(int length)
     {
         if (actionOnMalformedString == CodingErrorAction.REPLACE &&
-                actionOnUnmappableString == CodingErrorAction.REPLACE) {
+                actionOnUnmappableString == CodingErrorAction.REPLACE &&
+                buffer.hasArray()) {
             String s = new String(buffer.array(), buffer.arrayOffset() + position, length, MessagePack.UTF8);
             position += length;
             return s;

msgpack-core/src/main/java/org/msgpack/core/buffer/ArrayBufferInput.java

@@ -15,8 +15,6 @@
 //
 package org.msgpack.core.buffer;
 
-import java.io.IOException;
-
 import static org.msgpack.core.Preconditions.checkNotNull;
 
 /**
@@ -80,7 +78,6 @@ public void reset(byte[] arr, int offset, int len)
 
     @Override
     public MessageBuffer next()
-            throws IOException
     {
         if (isEmpty) {
             return null;
@@ -91,7 +88,6 @@ public MessageBuffer next()
 
     @Override
     public void close()
-            throws IOException
     {
         buffer = null;
         isEmpty = true;

msgpack-core/src/main/java/org/msgpack/core/buffer/ByteBufferInput.java

@@ -0,0 +1,67 @@
+//
+// MessagePack for Java
+//
+//    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.
+//
+package org.msgpack.core.buffer;
+
+import java.nio.ByteBuffer;
+
+import static org.msgpack.core.Preconditions.checkNotNull;
+
+/**
+ * {@link MessageBufferInput} adapter for {@link java.nio.ByteBuffer}
+ */
+public class ByteBufferInput
+        implements MessageBufferInput
+{
+    private ByteBuffer input;
+    private boolean isRead = false;
+
+    public ByteBufferInput(ByteBuffer input)
+    {
+        this.input = checkNotNull(input, "input ByteBuffer is null").slice();
+    }
+
+    /**
+     * Reset buffer.
+     *
+     * @param input new buffer
+     * @return the old buffer
+     */
+    public ByteBuffer reset(ByteBuffer input)
+    {
+        ByteBuffer old = this.input;
+        this.input = checkNotNull(input, "input ByteBuffer is null").slice();
+        isRead = false;
+        return old;
+    }
+
+    @Override
+    public MessageBuffer next()
+    {
+        if (isRead) {
+            return null;
+        }
+
+        MessageBuffer b = MessageBuffer.wrap(input);
+        isRead = true;
+        return b;
+    }
+
+    @Override
+    public void close()
+    {
+        // Nothing to do
+    }
+}