msgpack#117: Add next(buffer size) method to enable future optimization, such as bulk buffer outputs

xerial · xerial · commit a74d16358f35 · 2014-07-14T23:37:48.000+09:00
diff --git a/msgpack-core/src/main/java/org/msgpack/core/MessagePacker.java b/msgpack-core/src/main/java/org/msgpack/core/MessagePacker.java
@@ -57,7 +57,7 @@ public class MessagePacker implements Closeable {
     private final MessagePack.Config config;
 
     private final MessageBufferOutput out;
-    private final MessageBuffer buffer;
+    private MessageBuffer buffer;
     private int position;
 
 
@@ -96,14 +96,16 @@ private void prepareEncoder() {
     }
 
     public void flush() throws IOException {
-        out.flush(buffer, 0, position);
+        if(position == buffer.size()) {
+            out.flush(buffer);
+        }
+        else {
+            out.flush(buffer.slice(0, position));
+        }
+        buffer = out.next(config.getPackerBufferSize());
         position = 0;
     }
 
-    private void flushBuffer(MessageBuffer b) throws IOException {
-        out.flush(b, 0, b.size());
-    }
-
     public void close() throws IOException {
         try {
             flush();
@@ -296,12 +298,12 @@ public MessagePacker packBigInteger(BigInteger bi) throws IOException {
         }
         return this;
     }
-    
+
     public MessagePacker packFloat(float v) throws IOException {
         writeByteAndFloat(FLOAT32, v);
         return this;
     }
- 
+
     public MessagePacker packDouble(double v) throws IOException {
         writeByteAndDouble(FLOAT64, v);
         return this;
@@ -493,7 +495,7 @@ public MessagePacker writePayload(ByteBuffer src) throws IOException {
             // Wrap the input source as a MessageBuffer
             MessageBuffer wrapped = MessageBuffer.wrap(src).slice(src.position(), src.remaining());
             // Then, dump the source data to the output
-            flushBuffer(wrapped);
+            out.flush(wrapped);
             src.position(src.limit());
         }
         else {
@@ -523,7 +525,7 @@ public MessagePacker writePayload(byte[] src, int off, int len) throws IOExcepti
             // Wrap the input array as a MessageBuffer
             MessageBuffer wrapped = MessageBuffer.wrap(src).slice(off, len);
             // Dump the source data to the output
-            flushBuffer(wrapped);
+            out.flush(wrapped);
         }
         else {
             int cursor = 0;
diff --git a/msgpack-core/src/main/java/org/msgpack/core/buffer/ArrayBufferInput.java b/msgpack-core/src/main/java/org/msgpack/core/buffer/ArrayBufferInput.java
@@ -20,7 +20,6 @@ public ArrayBufferInput(byte[] arr, int offset, int length) {
         this.buffer = MessageBuffer.wrap(checkNotNull(arr, "input array is null")).slice(offset, length);
     }
 
-
     @Override
     public MessageBuffer next() throws IOException {
         if(isRead)
diff --git a/msgpack-core/src/main/java/org/msgpack/core/buffer/ChannelBufferOutput.java b/msgpack-core/src/main/java/org/msgpack/core/buffer/ChannelBufferOutput.java
@@ -12,15 +12,23 @@
 public class ChannelBufferOutput implements MessageBufferOutput {
 
     private final WritableByteChannel channel;
+    private MessageBuffer buffer;
 
     public ChannelBufferOutput(WritableByteChannel channel) {
         this.channel = checkNotNull(channel, "output channel is null");
     }
 
     @Override
-    public void flush(MessageBuffer buf, int offset, int len) throws IOException {
-        assert(offset + len <= buf.size());
-        ByteBuffer bb = buf.toByteBuffer(offset, len);
+    public MessageBuffer next(int bufferSize) throws IOException {
+        if(buffer == null || buffer.size() != bufferSize) {
+            buffer = MessageBuffer.newBuffer(bufferSize);
+        }
+        return buffer;
+    }
+
+    @Override
+    public void flush(MessageBuffer buf) throws IOException {
+        ByteBuffer bb = buf.toByteBuffer();
         channel.write(bb);
     }
 
diff --git a/msgpack-core/src/main/java/org/msgpack/core/buffer/MessageBufferOutput.java b/msgpack-core/src/main/java/org/msgpack/core/buffer/MessageBufferOutput.java
@@ -9,15 +9,22 @@
  */
 public interface MessageBufferOutput extends Closeable {
 
-//    /**
-//     * Retrieves the next buffer for writing message packed data
-//     * @return
-//     * @throws IOException
-//     */
-//    public MessageBuffer next() throws IOException;
-
-
-    public void flush(MessageBuffer buf, int offset, int len) throws IOException;
+    /**
+     * Retrieves the next buffer for writing message packed data
+     *
+     * @param bufferSize the buffer size to retrieve
+     * @return
+     * @throws IOException
+     */
+    public MessageBuffer next(int bufferSize) throws IOException;
+
+    /**
+     * Output the buffer contents. If you need to output a part of the
+     * buffer use {@link MessageBuffer#slice(int, int)}
+     * @param buf
+     * @throws IOException
+     */
+    public void flush(MessageBuffer buf) throws IOException;
 
 }
 
diff --git a/msgpack-core/src/main/java/org/msgpack/core/buffer/OutputStreamBufferOutput.java b/msgpack-core/src/main/java/org/msgpack/core/buffer/OutputStreamBufferOutput.java
@@ -11,19 +11,36 @@
 public class OutputStreamBufferOutput implements MessageBufferOutput {
 
     private final OutputStream out;
+    private MessageBuffer buffer;
+    private byte[] tmpBuf;
 
     public OutputStreamBufferOutput(OutputStream out) {
         this.out = checkNotNull(out, "output is null");
     }
 
     @Override
-    public void flush(MessageBuffer buf, int offset, int len) throws IOException {
-        assert(offset + len <= buf.size());
+    public MessageBuffer next(int bufferSize) throws IOException {
+        if(buffer == null || buffer.size != bufferSize) {
+            return buffer = MessageBuffer.newBuffer(bufferSize);
+        }
+        else {
+            return buffer;
+        }
+    }
 
-        // TODO reuse the allocated buffer
-        byte[] in = new byte[len];
-        buf.getBytes(offset, in, 0, len);
-        out.write(in, 0, len);
+    @Override
+    public void flush(MessageBuffer buf) throws IOException {
+        int writeLen = buf.size();
+        if(buf.hasArray()) {
+            out.write(buf.getArray(), buf.offset(), writeLen);
+        }
+        else {
+            if(tmpBuf == null || tmpBuf.length < writeLen) {
+                tmpBuf = new byte[writeLen];
+            }
+            buf.getBytes(0, tmpBuf, 0, writeLen);
+            out.write(tmpBuf, 0, writeLen);
+        }
     }
 
     @Override

Original file line number	Diff line number	Diff line change
`@@ -20,7 +20,6 @@ public ArrayBufferInput(byte[] arr, int offset, int length) {`
`20`	`20`	`this.buffer = MessageBuffer.wrap(checkNotNull(arr, "input array is null")).slice(offset, length);`
`21`	`21`	`}`
`22`	`22`
`23`		`-`
`24`	`23`	`@Override`
`25`	`24`	`public MessageBuffer next() throws IOException {`
`26`	`25`	`if(isRead)`