From 60ddba3cf6ec2f14bf2555f2f122892c644ad38c Mon Sep 17 00:00:00 2001 From: Mitsunori Komatsu Date: Fri, 23 Jun 2017 00:45:06 +0900 Subject: [PATCH 1/4] Fix a bug ChannelBufferInput#next blocks until the buffer is filled --- .../java/org/msgpack/core/buffer/ChannelBufferInput.java | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) 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 index f00cb0c30..922e5cf08 100644 --- a/msgpack-core/src/main/java/org/msgpack/core/buffer/ChannelBufferInput.java +++ b/msgpack-core/src/main/java/org/msgpack/core/buffer/ChannelBufferInput.java @@ -62,11 +62,9 @@ public MessageBuffer next() throws IOException { ByteBuffer b = buffer.sliceAsByteBuffer(); - while (b.remaining() > 0) { - int ret = channel.read(b); - if (ret == -1) { - break; - } + int ret = channel.read(b); + if (ret == -1) { + return null; } b.flip(); return b.remaining() == 0 ? null : buffer.slice(0, b.limit()); From c52d80652cc9e10c832a463db4a4298bec5f56b8 Mon Sep 17 00:00:00 2001 From: Mitsunori Komatsu Date: Sat, 24 Jun 2017 18:26:26 +0900 Subject: [PATCH 2/4] Don't need to consider the case ReadableByteChannel#read returns 0 since the buffer size must be more than 0 --- .../main/java/org/msgpack/core/buffer/ChannelBufferInput.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 index 922e5cf08..e8d7c1de8 100644 --- a/msgpack-core/src/main/java/org/msgpack/core/buffer/ChannelBufferInput.java +++ b/msgpack-core/src/main/java/org/msgpack/core/buffer/ChannelBufferInput.java @@ -67,7 +67,7 @@ public MessageBuffer next() return null; } b.flip(); - return b.remaining() == 0 ? null : buffer.slice(0, b.limit()); + return buffer.slice(0, b.limit()); } @Override From 323a5c3cf268a2de07657dc99a66ce5c6d0c4b77 Mon Sep 17 00:00:00 2001 From: Mitsunori Komatsu Date: Sat, 24 Jun 2017 21:30:14 +0900 Subject: [PATCH 3/4] Add a unit test --- .../core/buffer/MessageBufferInputTest.scala | 44 +++++++++++++++++++ 1 file changed, 44 insertions(+) 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 index 6b1c0da48..324448343 100644 --- a/msgpack-core/src/test/scala/org/msgpack/core/buffer/MessageBufferInputTest.scala +++ b/msgpack-core/src/test/scala/org/msgpack/core/buffer/MessageBufferInputTest.scala @@ -16,12 +16,17 @@ package org.msgpack.core.buffer import java.io._ +import java.net.{InetSocketAddress, ServerSocket, Socket} import java.nio.ByteBuffer +import java.nio.channels.{ServerSocketChannel, SocketChannel} +import java.util.concurrent +import java.util.concurrent.{Callable, Executors, TimeUnit} import java.util.zip.{GZIPInputStream, GZIPOutputStream} import org.msgpack.core.{MessagePack, MessagePackSpec, MessageUnpacker} import xerial.core.io.IOUtil._ +import scala.concurrent.Future import scala.util.Random class MessageBufferInputTest @@ -201,5 +206,44 @@ class MessageBufferInputTest buf.reset(in1) readInt(buf) shouldBe 42 } + + "unpack without blocking" in { + val server = ServerSocketChannel.open.bind(new InetSocketAddress("localhost", 0)) + val executorService = Executors.newCachedThreadPool + + try { + executorService.execute(new Runnable { + override def run { + val server_ch = server.accept + val packer = MessagePack.newDefaultPacker(server_ch) + packer.packString("0123456789") + packer.flush + // Keep the connection open + while (!executorService.isShutdown) { + TimeUnit.SECONDS.sleep(1) + } + packer.close + } + }) + + val future = executorService.submit(new Callable[String] { + override def call: String = { + val conn_ch = SocketChannel.open(new InetSocketAddress("localhost", server.socket.getLocalPort)) + val unpacker = MessagePack.newDefaultUnpacker(conn_ch) + val s = unpacker.unpackString + unpacker.close + s + } + }) + + future.get(5, TimeUnit.SECONDS) shouldBe "0123456789" + } + finally { + executorService.shutdown + if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) { + executorService.shutdownNow + } + } + } } } From fdddf4902f8734905cc97a935d3649f08cd9c3f0 Mon Sep 17 00:00:00 2001 From: Mitsunori Komatsu Date: Sat, 24 Jun 2017 22:21:08 +0900 Subject: [PATCH 4/4] Remove unused `import`s --- .../org/msgpack/core/buffer/MessageBufferInputTest.scala | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) 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 index 324448343..a7653797e 100644 --- a/msgpack-core/src/test/scala/org/msgpack/core/buffer/MessageBufferInputTest.scala +++ b/msgpack-core/src/test/scala/org/msgpack/core/buffer/MessageBufferInputTest.scala @@ -16,17 +16,15 @@ package org.msgpack.core.buffer import java.io._ -import java.net.{InetSocketAddress, ServerSocket, Socket} +import java.net.{InetSocketAddress} import java.nio.ByteBuffer import java.nio.channels.{ServerSocketChannel, SocketChannel} -import java.util.concurrent import java.util.concurrent.{Callable, Executors, TimeUnit} import java.util.zip.{GZIPInputStream, GZIPOutputStream} -import org.msgpack.core.{MessagePack, MessagePackSpec, MessageUnpacker} +import org.msgpack.core.{MessagePack, MessagePackSpec} import xerial.core.io.IOUtil._ -import scala.concurrent.Future import scala.util.Random class MessageBufferInputTest