CAMEL-10883: replace complex code with standard blocking channel
Project: http://git-wip-us.apache.org/repos/asf/camel/repo Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/603467ef Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/603467ef Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/603467ef Branch: refs/heads/master Commit: 603467efbf45a3968631598d47b8f0ab5d7d57b0 Parents: 4196654 Author: rohan <[email protected]> Authored: Tue Feb 21 15:47:24 2017 +1300 Committer: Andrea Cosentino <[email protected]> Committed: Thu Feb 23 10:00:26 2017 +0100 ---------------------------------------------------------------------- .../undertow/DefaultUndertowHttpBinding.java | 45 +++----------------- 1 file changed, 6 insertions(+), 39 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/camel/blob/603467ef/components/camel-undertow/src/main/java/org/apache/camel/component/undertow/DefaultUndertowHttpBinding.java ---------------------------------------------------------------------- diff --git a/components/camel-undertow/src/main/java/org/apache/camel/component/undertow/DefaultUndertowHttpBinding.java b/components/camel-undertow/src/main/java/org/apache/camel/component/undertow/DefaultUndertowHttpBinding.java index 5aa774a..153279b 100644 --- a/components/camel-undertow/src/main/java/org/apache/camel/component/undertow/DefaultUndertowHttpBinding.java +++ b/components/camel-undertow/src/main/java/org/apache/camel/component/undertow/DefaultUndertowHttpBinding.java @@ -22,13 +22,12 @@ import java.io.ObjectOutputStream; import java.io.PrintWriter; import java.io.StringWriter; import java.nio.ByteBuffer; +import java.nio.channels.ReadableByteChannel; import java.util.Deque; import java.util.HashMap; import java.util.Iterator; import java.util.Locale; import java.util.Map; -import java.util.concurrent.ArrayBlockingQueue; -import java.util.concurrent.BlockingQueue; import javax.activation.FileDataSource; @@ -56,7 +55,7 @@ import org.apache.camel.util.MessageHelper; import org.apache.camel.util.ObjectHelper; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.xnio.ChannelListener; +import org.xnio.channels.BlockingReadableByteChannel; import org.xnio.channels.StreamSourceChannel; /** @@ -384,46 +383,14 @@ public class DefaultUndertowHttpBinding implements UndertowHttpBinding { final ByteArrayOutputStream out = new ByteArrayOutputStream(); final ByteBuffer buffer = ByteBuffer.wrap(new byte[1024]); + ReadableByteChannel blockingSource = new BlockingReadableByteChannel(source); + for (;;) { - int res = source.read(buffer); + int res = blockingSource.read(buffer); if (res == -1) { return out.toByteArray(); } else if (res == 0) { - BlockingQueue<Integer> ping = new ArrayBlockingQueue<Integer>(1); - source.getReadSetter().set(new ChannelListener<StreamSourceChannel>() { - @Override - public void handleEvent(StreamSourceChannel channel) { - for (;;) { - try { - int res = channel.read(buffer); - switch (res) { - case -1: - ping.put(res); - return; - case 0: - // await next chunk - source.getReadSetter().set(this); - source.resumeReads(); - return; - default: - buffer.flip(); - out.write(buffer.array(), buffer.arrayOffset() + buffer.position(), buffer.arrayOffset() + buffer.limit()); - buffer.clear(); - } - } catch (IOException | InterruptedException e) { - LOG.error("Exception reading from channel {}", e); - } - } - } - }); - source.resumeReads(); - try { - // wait for the listener to complete - ping.take(); - } catch (InterruptedException e) { - LOG.error("Exception reading from channel {}", e); - } - return out.toByteArray(); + LOG.error("Channel did not block"); } else { buffer.flip(); out.write(buffer.array(), buffer.arrayOffset() + buffer.position(), buffer.arrayOffset() + buffer.limit());
