You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@cxf.apache.org by Sergey Beryozkin <sb...@gmail.com> on 2017/10/19 08:44:58 UTC
Re: [cxf] branch master updated: [CXF-7532]REST on JMS transport
can't handle the request message with text messageType
Hi Freeman
Can you move ReaderInputStream to the CXF core where CachedInputStream
and other IO utility code is located ?
Please also update the code which gets Reader.class and check it for
null first, and also drop a comment re the JMS transport, checking the
reader can make sense in all the cases...
Thanks, Sergey
On 19/10/17 04:16, ffang@apache.org wrote:
> This is an automated email from the ASF dual-hosted git repository.
>
> ffang pushed a commit to branch master
> in repository https://gitbox.apache.org/repos/asf/cxf.git
>
>
> The following commit(s) were added to refs/heads/master by this push:
> new 1d66475 [CXF-7532]REST on JMS transport can't handle the request message with text messageType
> 1d66475 is described below
>
> commit 1d66475b4dea2a0031dcb1765e8e9fd08430ec9b
> Author: Freeman Fang <fr...@gmail.com>
> AuthorDate: Thu Oct 19 11:15:50 2017 +0800
>
> [CXF-7532]REST on JMS transport can't handle the request message with text messageType
> ---
> .../org/apache/cxf/jaxrs/impl/ResponseImpl.java | 5 +
> .../org/apache/cxf/jaxrs/utils/JAXRSUtils.java | 6 +-
> .../apache/cxf/jaxrs/utils/ReaderInputStream.java | 296 +++++++++++++++++++++
> .../apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java | 90 +++++++
> 4 files changed, 396 insertions(+), 1 deletion(-)
>
> diff --git a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
> index e3a3acf..82f183a 100644
> --- a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
> +++ b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
> @@ -58,6 +58,7 @@ import org.apache.cxf.jaxrs.provider.ProviderFactory;
> import org.apache.cxf.jaxrs.utils.HttpUtils;
> import org.apache.cxf.jaxrs.utils.InjectionUtils;
> import org.apache.cxf.jaxrs.utils.JAXRSUtils;
> +import org.apache.cxf.jaxrs.utils.ReaderInputStream;
> import org.apache.cxf.message.Message;
> import org.apache.cxf.message.MessageUtils;
>
> @@ -342,6 +343,10 @@ public final class ResponseImpl extends Response {
> entityStreamAvailable = entityStream != null;
> } else if (entity instanceof InputStream) {
> entityStream = InputStream.class.cast(entity);
> + } else {
> + Message inMessage = getResponseMessage();
> + Reader reader = inMessage.getContent(Reader.class);
> + entityStream = InputStream.class.cast(new ReaderInputStream(reader));
> }
>
> // we need to check for readers even if no IS is set - the readers may still do it
> diff --git a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
> index 0ef6565..509f76d 100644
> --- a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
> +++ b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
> @@ -832,7 +832,11 @@ public final class JAXRSUtils {
> OperationResourceInfo ori)
> throws IOException, WebApplicationException {
> InputStream is = message.getContent(InputStream.class);
> -
> + if (is == null) {
> + //may use the jms transport so check the Reader;
> + Reader reader = message.getContent(Reader.class);
> + is = new ReaderInputStream(reader);
> + }
> if (parameter.getType() == ParameterType.REQUEST_BODY) {
>
> if (parameterClass == AsyncResponse.class) {
> diff --git a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
> new file mode 100644
> index 0000000..c7b142e
> --- /dev/null
> +++ b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
> @@ -0,0 +1,296 @@
> +/**
> + * Licensed to the Apache Software Foundation (ASF) under one
> + * or more contributor license agreements. See the NOTICE file
> + * distributed with this work for additional information
> + * regarding copyright ownership. The ASF licenses this file
> + * to you 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.apache.cxf.jaxrs.utils;
> +
> +import java.io.IOException;
> +import java.io.InputStream;
> +import java.io.Reader;
> +import java.nio.ByteBuffer;
> +import java.nio.CharBuffer;
> +import java.nio.charset.Charset;
> +import java.nio.charset.CharsetEncoder;
> +import java.nio.charset.CoderResult;
> +import java.nio.charset.CodingErrorAction;
> +
> +/**
> + * {@link InputStream} implementation that reads a character stream from a {@link Reader}
> + * and transforms it to a byte stream using a specified charset encoding. The stream
> + * is transformed using a {@link CharsetEncoder} object, guaranteeing that all charset
> + * encodings supported by the JRE are handled correctly. In particular for charsets such as
> + * UTF-16, the implementation ensures that one and only one byte order marker
> + * is produced.
> + * <p>
> + * Since in general it is not possible to predict the number of characters to be read from the
> + * {@link Reader} to satisfy a read request on the {@link ReaderInputStream}, all reads from
> + * the {@link Reader} are buffered. There is therefore no well defined correlation
> + * between the current position of the {@link Reader} and that of the {@link ReaderInputStream}.
> + * This also implies that in general there is no need to wrap the underlying {@link Reader}
> + * in a {@link java.io.BufferedReader}.
> + * <p>
> + * {@link ReaderInputStream} implements the inverse transformation of {@link java.io.InputStreamReader};
> + * in the following example, reading from <tt>in2</tt> would return the same byte
> + * sequence as reading from <tt>in</tt> (provided that the initial byte sequence is legal
> + * with respect to the charset encoding):
> + * <pre>
> + * InputStream in = ...
> + * Charset cs = ...
> + * InputStreamReader reader = new InputStreamReader(in, cs);
> + * ReaderInputStream in2 = new ReaderInputStream(reader, cs);</pre>
> + * {@link ReaderInputStream} implements the same transformation as {@link java.io.OutputStreamWriter},
> + * except that the control flow is reversed: both classes transform a character stream
> + * into a byte stream, but {@link java.io.OutputStreamWriter} pushes data to the underlying stream,
> + * while {@link ReaderInputStream} pulls it from the underlying stream.
> + * <p>
> + * Note that while there are use cases where there is no alternative to using
> + * this class, very often the need to use this class is an indication of a flaw
> + * in the design of the code. This class is typically used in situations where an existing
> + * API only accepts an {@link InputStream}, but where the most natural way to produce the data
> + * is as a character stream, i.e. by providing a {@link Reader} instance. An example of a situation
> + * where this problem may appear is when implementing the {@link javax.activation.DataSource}
> + * interface from the Java Activation Framework.
> + * <p>
> + * Given the fact that the {@link Reader} class doesn't provide any way to predict whether the next
> + * read operation will block or not, it is not possible to provide a meaningful
> + * implementation of the {@link InputStream#available()} method. A call to this method
> + * will always return 0. Also, this class doesn't support {@link InputStream#mark(int)}.
> + * <p>
> + * Instances of {@link ReaderInputStream} are not thread safe.
> + *
> + * @see org.apache.commons.io.output.WriterOutputStream
> + *
> + * @since 2.0
> + */
> +public class ReaderInputStream extends InputStream {
> + private static final int DEFAULT_BUFFER_SIZE = 1024;
> +
> + private final Reader reader;
> + private final CharsetEncoder encoder;
> +
> + /**
> + * CharBuffer used as input for the decoder. It should be reasonably
> + * large as we read data from the underlying Reader into this buffer.
> + */
> + private final CharBuffer encoderIn;
> +
> + /**
> + * ByteBuffer used as output for the decoder. This buffer can be small
> + * as it is only used to transfer data from the decoder to the
> + * buffer provided by the caller.
> + */
> + private final ByteBuffer encoderOut;
> +
> + private CoderResult lastCoderResult;
> + private boolean endOfInput;
> +
> + /**
> + * Construct a new {@link ReaderInputStream}.
> + *
> + * @param reader the target {@link Reader}
> + * @param encoder the charset encoder
> + * @since 2.1
> + */
> + public ReaderInputStream(Reader reader, CharsetEncoder encoder) {
> + this(reader, encoder, DEFAULT_BUFFER_SIZE);
> + }
> +
> + /**
> + * Construct a new {@link ReaderInputStream}.
> + *
> + * @param reader the target {@link Reader}
> + * @param encoder the charset encoder
> + * @param bufferSize the size of the input buffer in number of characters
> + * @since 2.1
> + */
> + public ReaderInputStream(Reader reader, CharsetEncoder encoder, int bufferSize) {
> + this.reader = reader;
> + this.encoder = encoder;
> + this.encoderIn = CharBuffer.allocate(bufferSize);
> + this.encoderIn.flip();
> + this.encoderOut = ByteBuffer.allocate(128);
> + this.encoderOut.flip();
> + }
> +
> + /**
> + * Construct a new {@link ReaderInputStream}.
> + *
> + * @param reader the target {@link Reader}
> + * @param charset the charset encoding
> + * @param bufferSize the size of the input buffer in number of characters
> + */
> + public ReaderInputStream(Reader reader, Charset charset, int bufferSize) {
> + this(reader,
> + charset.newEncoder()
> + .onMalformedInput(CodingErrorAction.REPLACE)
> + .onUnmappableCharacter(CodingErrorAction.REPLACE),
> + bufferSize);
> + }
> +
> + /**
> + * Construct a new {@link ReaderInputStream} with a default input buffer size of
> + * 1024 characters.
> + *
> + * @param reader the target {@link Reader}
> + * @param charset the charset encoding
> + */
> + public ReaderInputStream(Reader reader, Charset charset) {
> + this(reader, charset, DEFAULT_BUFFER_SIZE);
> + }
> +
> + /**
> + * Construct a new {@link ReaderInputStream}.
> + *
> + * @param reader the target {@link Reader}
> + * @param charsetName the name of the charset encoding
> + * @param bufferSize the size of the input buffer in number of characters
> + */
> + public ReaderInputStream(Reader reader, String charsetName, int bufferSize) {
> + this(reader, Charset.forName(charsetName), bufferSize);
> + }
> +
> + /**
> + * Construct a new {@link ReaderInputStream} with a default input buffer size of
> + * 1024 characters.
> + *
> + * @param reader the target {@link Reader}
> + * @param charsetName the name of the charset encoding
> + */
> + public ReaderInputStream(Reader reader, String charsetName) {
> + this(reader, charsetName, DEFAULT_BUFFER_SIZE);
> + }
> +
> + /**
> + * Construct a new {@link ReaderInputStream} that uses the default character encoding
> + * with a default input buffer size of 1024 characters.
> + *
> + * @param reader the target {@link Reader}
> + */
> + public ReaderInputStream(Reader reader) {
> + this(reader, Charset.defaultCharset());
> + }
> +
> + /**
> + * Fills the internal char buffer from the reader.
> + *
> + * @throws IOException
> + * If an I/O error occurs
> + */
> + private void fillBuffer() throws IOException {
> + if (!endOfInput && (lastCoderResult == null || lastCoderResult.isUnderflow())) {
> + encoderIn.compact();
> + int position = encoderIn.position();
> + // We don't use Reader#read(CharBuffer) here because it is more efficient
> + // to write directly to the underlying char array (the default implementation
> + // copies data to a temporary char array).
> + int c = reader.read(encoderIn.array(), position, encoderIn.remaining());
> + if (c == -1) {
> + endOfInput = true;
> + } else {
> + encoderIn.position(position + c);
> + }
> + encoderIn.flip();
> + }
> + encoderOut.compact();
> + lastCoderResult = encoder.encode(encoderIn, encoderOut, endOfInput);
> + encoderOut.flip();
> + }
> +
> + /**
> + * Read the specified number of bytes into an array.
> + *
> + * @param b the byte array to read into
> + * @param off the offset to start reading bytes into
> + * @param len the number of bytes to read
> + * @return the number of bytes read or <code>-1</code>
> + * if the end of the stream has been reached
> + * @throws IOException if an I/O error occurs
> + */
> + @Override
> + public int read(byte[] b, int off, int len) throws IOException {
> + if (b == null) {
> + throw new NullPointerException("Byte array must not be null");
> + }
> + if (len < 0 || off < 0 || (off + len) > b.length) {
> + throw new IndexOutOfBoundsException("Array Size=" + b.length
> + + ", offset=" + off + ", length=" + len);
> + }
> + int read = 0;
> + if (len == 0) {
> + return 0; // Always return 0 if len == 0
> + }
> + while (len > 0) {
> + if (encoderOut.hasRemaining()) {
> + int c = Math.min(encoderOut.remaining(), len);
> + encoderOut.get(b, off, c);
> + off += c;
> + len -= c;
> + read += c;
> + } else {
> + fillBuffer();
> + if (endOfInput && !encoderOut.hasRemaining()) {
> + break;
> + }
> + }
> + }
> + return read == 0 && endOfInput ? -1 : read;
> + }
> +
> + /**
> + * Read the specified number of bytes into an array.
> + *
> + * @param b the byte array to read into
> + * @return the number of bytes read or <code>-1</code>
> + * if the end of the stream has been reached
> + * @throws IOException if an I/O error occurs
> + */
> + @Override
> + public int read(byte[] b) throws IOException {
> + return read(b, 0, b.length);
> + }
> +
> + /**
> + * Read a single byte.
> + *
> + * @return either the byte read or <code>-1</code> if the end of the stream
> + * has been reached
> + * @throws IOException if an I/O error occurs
> + */
> + @Override
> + public int read() throws IOException {
> + for (;;) {
> + if (encoderOut.hasRemaining()) {
> + return encoderOut.get() & 0xFF;
> + } else {
> + fillBuffer();
> + if (endOfInput && !encoderOut.hasRemaining()) {
> + return -1;
> + }
> + }
> + }
> + }
> +
> + /**
> + * Close the stream. This method will cause the underlying {@link Reader}
> + * to be closed.
> + * @throws IOException if an I/O error occurs
> + */
> + @Override
> + public void close() throws IOException {
> + reader.close();
> + }
> +}
> diff --git a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
> index f77b5ed..b957f36 100644
> --- a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
> +++ b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
> @@ -96,6 +96,25 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
> assertEquals("Get a wrong response code.", 200, client.getResponse().getStatus());
> assertEquals("Get a wrong book id.", 123, book.getId());
> }
> +
> + @Test
> + public void testGetBookFromWebClientWithTextJMSMessage() throws Exception {
> + // setup the the client
> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
> + + "?replyToName=dynamicQueues/test.jmstransport.response"
> + + "&jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
> + + "&jndiURL=tcp://localhost:" + JMS_PORT
> + + "&messageType=text";
> +
> + WebClient client = WebClient.create(endpointAddressUrlEncoded);
> + WebClient.getConfig(client).getInInterceptors().add(new LoggingInInterceptor());
> + WebClient.getConfig(client).getRequestContext()
> + .put(org.apache.cxf.message.Message.REQUEST_URI, "/bookstore/books/123");
> +
> + Book book = client.get(Book.class);
> + assertEquals("Get a wrong response code.", 200, client.getResponse().getStatus());
> + assertEquals("Get a wrong book id.", 123, book.getId());
> + }
>
> @Test
> public void testPutBookOneWayWithWebClient() throws Exception {
> @@ -130,6 +149,7 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
> }
> }
>
> +
> @Test
> public void testGetBookFromWebClientWithPath() throws Exception {
> // setup the the client
> @@ -148,6 +168,25 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
> }
>
> @Test
> + public void testGetBookFromWebClientWithPathWithTextJMSMessage() throws Exception {
> + // setup the the client
> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
> + + "&replyToName=dynamicQueues/test.jmstransport.response"
> + + "&jndiURL=tcp://localhost:" + JMS_PORT
> + + "&jndiConnectionFactoryName=ConnectionFactory"
> + + "&messageType=text";
> +
> + WebClient client = WebClient.create(endpointAddressUrlEncoded);
> + client.path("bookstore").path("books").path("123");
> +
> + Book book = client.get(Book.class);
> + assertEquals("Get a wrong response code.", 200, client.getResponse().getStatus());
> + assertEquals("Get a wrong book id.", 123, book.getId());
> + }
> +
> +
> + @Test
> public void testGetBookFromProxyClient() throws Exception {
> // setup the the client
> String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
> @@ -161,6 +200,22 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
> assertEquals("Get a wrong response code.", 200, WebClient.client(client).getResponse().getStatus());
> assertEquals("Get a wrong book id.", 123, book.getId());
> }
> +
> + @Test
> + public void testGetBookFromProxyClientWithTextJMSMessage() throws Exception {
> + // setup the the client
> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
> + + "&replyToName=dynamicQueues/test.jmstransport.response"
> + + "&jndiURL=tcp://localhost:" + JMS_PORT
> + + "&jndiConnectionFactoryName=ConnectionFactory"
> + + "&messageType=text";
> +
> + JMSBookStore client = JAXRSClientFactory.create(endpointAddressUrlEncoded, JMSBookStore.class);
> + Book book = client.getBook("123");
> + assertEquals("Get a wrong response code.", 200, WebClient.client(client).getResponse().getStatus());
> + assertEquals("Get a wrong book id.", 123, book.getId());
> + }
>
> @Test
> public void testGetBookFromSubresourceProxyClient() throws Exception {
> @@ -177,6 +232,23 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
> assertEquals("Get a wrong response code.", 200, WebClient.client(bookProxy).getResponse().getStatus());
> assertEquals("Get a wrong book id.", 123, book.getId());
> }
> +
> + @Test
> + public void testGetBookFromSubresourceProxyClientWithTextJMSMessage() throws Exception {
> + // setup the the client
> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
> + + "&replyToName=dynamicQueues/test.jmstransport.response"
> + + "&jndiURL=tcp://localhost:" + JMS_PORT
> + + "&jndiConnectionFactoryName=ConnectionFactory"
> + + "&messageType=text";
> +
> + JMSBookStore client = JAXRSClientFactory.create(endpointAddressUrlEncoded, JMSBookStore.class);
> + Book bookProxy = client.getBookSubResource("123");
> + Book book = bookProxy.retrieveState();
> + assertEquals("Get a wrong response code.", 200, WebClient.client(bookProxy).getResponse().getStatus());
> + assertEquals("Get a wrong book id.", 123, book.getId());
> + }
>
> @Test
> public void testGetBookFromProxyClientWithQuery() throws Exception {
> @@ -193,6 +265,24 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
> assertEquals("Get a wrong book id.", 123, book.getId());
> }
>
> +
> + @Test
> + public void testGetBookFromProxyClientWithQueryWithTextJMSMessage() throws Exception {
> + // setup the the client
> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
> + + "&replyToName=dynamicQueues/test.jmstransport.response"
> + + "&jndiURL=tcp://localhost:" + JMS_PORT
> + + "&jndiConnectionFactoryName=ConnectionFactory"
> + + "&messageType=text";
> +
> + JMSBookStore client = JAXRSClientFactory.create(endpointAddressUrlEncoded, JMSBookStore.class);
> + Book book = client.getBookByURLQuery(new String[] {"1", "2", "3"});
> + assertEquals("Get a wrong response code.", 200, WebClient.client(client).getResponse().getStatus());
> + assertEquals("Get a wrong book id.", 123, book.getId());
> + }
> +
> +
> @Test
> public void testGetBook() throws Exception {
> Context ctx = getContext();
>
--
Sergey Beryozkin
Talend Community Coders
http://coders.talend.com/
Re: [cxf] branch master updated: [CXF-7532]REST on JMS transport can't handle the request message with text messageType
Posted by ffang <fr...@gmail.com>.
Sure, will do!
－－－－－－－－－－－－－
Freeman(Yue) Fang
Red Hat, Inc.
FuseSource is now part of Red Hat
> On Oct 19, 2017, at 5:05 PM, Sergey Beryozkin <sb...@gmail.com> wrote:
>
> Hi Freeman
>
> Reader may likely need to be set too for SOAP, and I recall checking Reader somewhere else too (will double check now), but I thought may be the transport can set both Reader and InputStream, hmm, that is an extra instance though, may be lets start with moving ReaderInputStream to the core first ?
>
> Sergey
> On 19/10/17 09:59, ffang wrote:
>> Yeah, set ReaderInputStream instead of Reader for the JMS transport is a better solution. If no objection I will go this way.
>> －－－－－－－－－－－－－
>> Freeman(Yue) Fang
>> Red Hat, Inc.
>> FuseSource is now part of Red Hat
>>> On Oct 19, 2017, at 4:50 PM, Sergey Beryozkin <sberyozkin@gmail.com <ma...@gmail.com>> wrote:
>>>
>>> The other thing is, should JMS transport itself set ReaderInputStream as InputStream on the message, as opposed to patching the JAX-RS code in various places ?
>>>
>>> Sergey
>>> On 19/10/17 09:44, Sergey Beryozkin wrote:
>>>> Hi Freeman
>>>> Can you move ReaderInputStream to the CXF core where CachedInputStream and other IO utility code is located ?
>>>> Please also update the code which gets Reader.class and check it for null first, and also drop a comment re the JMS transport, checking the reader can make sense in all the cases...
>>>> Thanks, Sergey
>>>> On 19/10/17 04:16, ffang@apache.org <ma...@apache.org> wrote:
>>>>> This is an automated email from the ASF dual-hosted git repository.
>>>>>
>>>>> ffang pushed a commit to branch master
>>>>> in repository https://gitbox.apache.org/repos/asf/cxf.git
>>>>>
>>>>>
>>>>> The following commit(s) were added to refs/heads/master by this push:
>>>>> new 1d66475 [CXF-7532]REST on JMS transport can't handle the request message with text messageType
>>>>> 1d66475 is described below
>>>>>
>>>>> commit 1d66475b4dea2a0031dcb1765e8e9fd08430ec9b
>>>>> Author: Freeman Fang <freeman.fang@gmail.com <ma...@gmail.com>>
>>>>> AuthorDate: Thu Oct 19 11:15:50 2017 +0800
>>>>>
>>>>> [CXF-7532]REST on JMS transport can't handle the request message with text messageType
>>>>> ---
>>>>> .../org/apache/cxf/jaxrs/impl/ResponseImpl.java | 5 +
>>>>> .../org/apache/cxf/jaxrs/utils/JAXRSUtils.java | 6 +-
>>>>> .../apache/cxf/jaxrs/utils/ReaderInputStream.java | 296 +++++++++++++++++++++
>>>>> .../apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java | 90 +++++++
>>>>> 4 files changed, 396 insertions(+), 1 deletion(-)
>>>>>
>>>>> diff --git a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>> index e3a3acf..82f183a 100644
>>>>> --- a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>> +++ b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>> @@ -58,6 +58,7 @@ import org.apache.cxf.jaxrs.provider.ProviderFactory;
>>>>> import org.apache.cxf.jaxrs.utils.HttpUtils;
>>>>> import org.apache.cxf.jaxrs.utils.InjectionUtils;
>>>>> import org.apache.cxf.jaxrs.utils.JAXRSUtils;
>>>>> +import org.apache.cxf.jaxrs.utils.ReaderInputStream;
>>>>> import org.apache.cxf.message.Message;
>>>>> import org.apache.cxf.message.MessageUtils;
>>>>> @@ -342,6 +343,10 @@ public final class ResponseImpl extends Response {
>>>>> entityStreamAvailable = entityStream != null;
>>>>> } else if (entity instanceof InputStream) {
>>>>> entityStream = InputStream.class.cast(entity);
>>>>> + } else {
>>>>> + Message inMessage = getResponseMessage();
>>>>> + Reader reader = inMessage.getContent(Reader.class);
>>>>> + entityStream = InputStream.class.cast(new ReaderInputStream(reader));
>>>>> }
>>>>> // we need to check for readers even if no IS is set - the readers may still do it
>>>>> diff --git a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>> index 0ef6565..509f76d 100644
>>>>> --- a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>> +++ b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>> @@ -832,7 +832,11 @@ public final class JAXRSUtils {
>>>>> OperationResourceInfo ori)
>>>>> throws IOException, WebApplicationException {
>>>>> InputStream is = message.getContent(InputStream.class);
>>>>> -
>>>>> + if (is == null) {
>>>>> + //may use the jms transport so check the Reader;
>>>>> + Reader reader = message.getContent(Reader.class);
>>>>> + is = new ReaderInputStream(reader);
>>>>> + }
>>>>> if (parameter.getType() == ParameterType.REQUEST_BODY) {
>>>>> if (parameterClass == AsyncResponse.class) {
>>>>> diff --git a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
>>>>> new file mode 100644
>>>>> index 0000000..c7b142e
>>>>> --- /dev/null
>>>>> +++ b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
>>>>> @@ -0,0 +1,296 @@
>>>>> +/**
>>>>> + * Licensed to the Apache Software Foundation (ASF) under one
>>>>> + * or more contributor license agreements. See the NOTICE file
>>>>> + * distributed with this work for additional information
>>>>> + * regarding copyright ownership. The ASF licenses this file
>>>>> + * to you 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.apache.cxf.jaxrs.utils;
>>>>> +
>>>>> +import java.io.IOException;
>>>>> +import java.io.InputStream;
>>>>> +import java.io.Reader;
>>>>> +import java.nio.ByteBuffer;
>>>>> +import java.nio.CharBuffer;
>>>>> +import java.nio.charset.Charset;
>>>>> +import java.nio.charset.CharsetEncoder;
>>>>> +import java.nio.charset.CoderResult;
>>>>> +import java.nio.charset.CodingErrorAction;
>>>>> +
>>>>> +/**
>>>>> + * {@link InputStream} implementation that reads a character stream from a {@link Reader}
>>>>> + * and transforms it to a byte stream using a specified charset encoding. The stream
>>>>> + * is transformed using a {@link CharsetEncoder} object, guaranteeing that all charset
>>>>> + * encodings supported by the JRE are handled correctly. In particular for charsets such as
>>>>> + * UTF-16, the implementation ensures that one and only one byte order marker
>>>>> + * is produced.
>>>>> + * <p>
>>>>> + * Since in general it is not possible to predict the number of characters to be read from the
>>>>> + * {@link Reader} to satisfy a read request on the {@link ReaderInputStream}, all reads from
>>>>> + * the {@link Reader} are buffered. There is therefore no well defined correlation
>>>>> + * between the current position of the {@link Reader} and that of the {@link ReaderInputStream}.
>>>>> + * This also implies that in general there is no need to wrap the underlying {@link Reader}
>>>>> + * in a {@link java.io.BufferedReader}.
>>>>> + * <p>
>>>>> + * {@link ReaderInputStream} implements the inverse transformation of {@link java.io.InputStreamReader};
>>>>> + * in the following example, reading from <tt>in2</tt> would return the same byte
>>>>> + * sequence as reading from <tt>in</tt> (provided that the initial byte sequence is legal
>>>>> + * with respect to the charset encoding):
>>>>> + * <pre>
>>>>> + * InputStream in = ...
>>>>> + * Charset cs = ...
>>>>> + * InputStreamReader reader = new InputStreamReader(in, cs);
>>>>> + * ReaderInputStream in2 = new ReaderInputStream(reader, cs);</pre>
>>>>> + * {@link ReaderInputStream} implements the same transformation as {@link java.io.OutputStreamWriter},
>>>>> + * except that the control flow is reversed: both classes transform a character stream
>>>>> + * into a byte stream, but {@link java.io.OutputStreamWriter} pushes data to the underlying stream,
>>>>> + * while {@link ReaderInputStream} pulls it from the underlying stream.
>>>>> + * <p>
>>>>> + * Note that while there are use cases where there is no alternative to using
>>>>> + * this class, very often the need to use this class is an indication of a flaw
>>>>> + * in the design of the code. This class is typically used in situations where an existing
>>>>> + * API only accepts an {@link InputStream}, but where the most natural way to produce the data
>>>>> + * is as a character stream, i.e. by providing a {@link Reader} instance. An example of a situation
>>>>> + * where this problem may appear is when implementing the {@link javax.activation.DataSource}
>>>>> + * interface from the Java Activation Framework.
>>>>> + * <p>
>>>>> + * Given the fact that the {@link Reader} class doesn't provide any way to predict whether the next
>>>>> + * read operation will block or not, it is not possible to provide a meaningful
>>>>> + * implementation of the {@link InputStream#available()} method. A call to this method
>>>>> + * will always return 0. Also, this class doesn't support {@link InputStream#mark(int)}.
>>>>> + * <p>
>>>>> + * Instances of {@link ReaderInputStream} are not thread safe.
>>>>> + *
>>>>> + * @see org.apache.commons.io.output.WriterOutputStream
>>>>> + *
>>>>> + * @since 2.0
>>>>> + */
>>>>> +public class ReaderInputStream extends InputStream {
>>>>> + private static final int DEFAULT_BUFFER_SIZE = 1024;
>>>>> +
>>>>> + private final Reader reader;
>>>>> + private final CharsetEncoder encoder;
>>>>> +
>>>>> + /**
>>>>> + * CharBuffer used as input for the decoder. It should be reasonably
>>>>> + * large as we read data from the underlying Reader into this buffer.
>>>>> + */
>>>>> + private final CharBuffer encoderIn;
>>>>> +
>>>>> + /**
>>>>> + * ByteBuffer used as output for the decoder. This buffer can be small
>>>>> + * as it is only used to transfer data from the decoder to the
>>>>> + * buffer provided by the caller.
>>>>> + */
>>>>> + private final ByteBuffer encoderOut;
>>>>> +
>>>>> + private CoderResult lastCoderResult;
>>>>> + private boolean endOfInput;
>>>>> +
>>>>> + /**
>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>> + *
>>>>> + * @param reader the target {@link Reader}
>>>>> + * @param encoder the charset encoder
>>>>> + * @since 2.1
>>>>> + */
>>>>> + public ReaderInputStream(Reader reader, CharsetEncoder encoder) {
>>>>> + this(reader, encoder, DEFAULT_BUFFER_SIZE);
>>>>> + }
>>>>> +
>>>>> + /**
>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>> + *
>>>>> + * @param reader the target {@link Reader}
>>>>> + * @param encoder the charset encoder
>>>>> + * @param bufferSize the size of the input buffer in number of characters
>>>>> + * @since 2.1
>>>>> + */
>>>>> + public ReaderInputStream(Reader reader, CharsetEncoder encoder, int bufferSize) {
>>>>> + this.reader = reader;
>>>>> + this.encoder = encoder;
>>>>> + this.encoderIn = CharBuffer.allocate(bufferSize);
>>>>> + this.encoderIn.flip();
>>>>> + this.encoderOut = ByteBuffer.allocate(128);
>>>>> + this.encoderOut.flip();
>>>>> + }
>>>>> +
>>>>> + /**
>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>> + *
>>>>> + * @param reader the target {@link Reader}
>>>>> + * @param charset the charset encoding
>>>>> + * @param bufferSize the size of the input buffer in number of characters
>>>>> + */
>>>>> + public ReaderInputStream(Reader reader, Charset charset, int bufferSize) {
>>>>> + this(reader,
>>>>> + charset.newEncoder()
>>>>> + .onMalformedInput(CodingErrorAction.REPLACE)
>>>>> + .onUnmappableCharacter(CodingErrorAction.REPLACE),
>>>>> + bufferSize);
>>>>> + }
>>>>> +
>>>>> + /**
>>>>> + * Construct a new {@link ReaderInputStream} with a default input buffer size of
>>>>> + * 1024 characters.
>>>>> + *
>>>>> + * @param reader the target {@link Reader}
>>>>> + * @param charset the charset encoding
>>>>> + */
>>>>> + public ReaderInputStream(Reader reader, Charset charset) {
>>>>> + this(reader, charset, DEFAULT_BUFFER_SIZE);
>>>>> + }
>>>>> +
>>>>> + /**
>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>> + *
>>>>> + * @param reader the target {@link Reader}
>>>>> + * @param charsetName the name of the charset encoding
>>>>> + * @param bufferSize the size of the input buffer in number of characters
>>>>> + */
>>>>> + public ReaderInputStream(Reader reader, String charsetName, int bufferSize) {
>>>>> + this(reader, Charset.forName(charsetName), bufferSize);
>>>>> + }
>>>>> +
>>>>> + /**
>>>>> + * Construct a new {@link ReaderInputStream} with a default input buffer size of
>>>>> + * 1024 characters.
>>>>> + *
>>>>> + * @param reader the target {@link Reader}
>>>>> + * @param charsetName the name of the charset encoding
>>>>> + */
>>>>> + public ReaderInputStream(Reader reader, String charsetName) {
>>>>> + this(reader, charsetName, DEFAULT_BUFFER_SIZE);
>>>>> + }
>>>>> +
>>>>> + /**
>>>>> + * Construct a new {@link ReaderInputStream} that uses the default character encoding
>>>>> + * with a default input buffer size of 1024 characters.
>>>>> + *
>>>>> + * @param reader the target {@link Reader}
>>>>> + */
>>>>> + public ReaderInputStream(Reader reader) {
>>>>> + this(reader, Charset.defaultCharset());
>>>>> + }
>>>>> +
>>>>> + /**
>>>>> + * Fills the internal char buffer from the reader.
>>>>> + *
>>>>> + * @throws IOException
>>>>> + * If an I/O error occurs
>>>>> + */
>>>>> + private void fillBuffer() throws IOException {
>>>>> + if (!endOfInput && (lastCoderResult == null || lastCoderResult.isUnderflow())) {
>>>>> + encoderIn.compact();
>>>>> + int position = encoderIn.position();
>>>>> + // We don't use Reader#read(CharBuffer) here because it is more efficient
>>>>> + // to write directly to the underlying char array (the default implementation
>>>>> + // copies data to a temporary char array).
>>>>> + int c = reader.read(encoderIn.array(), position, encoderIn.remaining());
>>>>> + if (c == -1) {
>>>>> + endOfInput = true;
>>>>> + } else {
>>>>> + encoderIn.position(position + c);
>>>>> + }
>>>>> + encoderIn.flip();
>>>>> + }
>>>>> + encoderOut.compact();
>>>>> + lastCoderResult = encoder.encode(encoderIn, encoderOut, endOfInput);
>>>>> + encoderOut.flip();
>>>>> + }
>>>>> +
>>>>> + /**
>>>>> + * Read the specified number of bytes into an array.
>>>>> + *
>>>>> + * @param b the byte array to read into
>>>>> + * @param off the offset to start reading bytes into
>>>>> + * @param len the number of bytes to read
>>>>> + * @return the number of bytes read or <code>-1</code>
>>>>> + * if the end of the stream has been reached
>>>>> + * @throws IOException if an I/O error occurs
>>>>> + */
>>>>> + @Override
>>>>> + public int read(byte[] b, int off, int len) throws IOException {
>>>>> + if (b == null) {
>>>>> + throw new NullPointerException("Byte array must not be null");
>>>>> + }
>>>>> + if (len < 0 || off < 0 || (off + len) > b.length) {
>>>>> + throw new IndexOutOfBoundsException("Array Size=" + b.length
>>>>> + + ", offset=" + off + ", length=" + len);
>>>>> + }
>>>>> + int read = 0;
>>>>> + if (len == 0) {
>>>>> + return 0; // Always return 0 if len == 0
>>>>> + }
>>>>> + while (len > 0) {
>>>>> + if (encoderOut.hasRemaining()) {
>>>>> + int c = Math.min(encoderOut.remaining(), len);
>>>>> + encoderOut.get(b, off, c);
>>>>> + off += c;
>>>>> + len -= c;
>>>>> + read += c;
>>>>> + } else {
>>>>> + fillBuffer();
>>>>> + if (endOfInput && !encoderOut.hasRemaining()) {
>>>>> + break;
>>>>> + }
>>>>> + }
>>>>> + }
>>>>> + return read == 0 && endOfInput ? -1 : read;
>>>>> + }
>>>>> +
>>>>> + /**
>>>>> + * Read the specified number of bytes into an array.
>>>>> + *
>>>>> + * @param b the byte array to read into
>>>>> + * @return the number of bytes read or <code>-1</code>
>>>>> + * if the end of the stream has been reached
>>>>> + * @throws IOException if an I/O error occurs
>>>>> + */
>>>>> + @Override
>>>>> + public int read(byte[] b) throws IOException {
>>>>> + return read(b, 0, b.length);
>>>>> + }
>>>>> +
>>>>> + /**
>>>>> + * Read a single byte.
>>>>> + *
>>>>> + * @return either the byte read or <code>-1</code> if the end of the stream
>>>>> + * has been reached
>>>>> + * @throws IOException if an I/O error occurs
>>>>> + */
>>>>> + @Override
>>>>> + public int read() throws IOException {
>>>>> + for (;;) {
>>>>> + if (encoderOut.hasRemaining()) {
>>>>> + return encoderOut.get() & 0xFF;
>>>>> + } else {
>>>>> + fillBuffer();
>>>>> + if (endOfInput && !encoderOut.hasRemaining()) {
>>>>> + return -1;
>>>>> + }
>>>>> + }
>>>>> + }
>>>>> + }
>>>>> +
>>>>> + /**
>>>>> + * Close the stream. This method will cause the underlying {@link Reader}
>>>>> + * to be closed.
>>>>> + * @throws IOException if an I/O error occurs
>>>>> + */
>>>>> + @Override
>>>>> + public void close() throws IOException {
>>>>> + reader.close();
>>>>> + }
>>>>> +}
>>>>> diff --git a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>> index f77b5ed..b957f36 100644
>>>>> --- a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>> +++ b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>> @@ -96,6 +96,25 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>> assertEquals("Get a wrong response code.", 200, client.getResponse().getStatus());
>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>> }
>>>>> +
>>>>> + @Test
>>>>> + public void testGetBookFromWebClientWithTextJMSMessage() throws Exception {
>>>>> + // setup the the client
>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>> + + "?replyToName=dynamicQueues/test.jmstransport.response"
>>>>> + + "&jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>> + + "&messageType=text";
>>>>> +
>>>>> + WebClient client = WebClient.create(endpointAddressUrlEncoded);
>>>>> + WebClient.getConfig(client).getInInterceptors().add(new LoggingInInterceptor());
>>>>> + WebClient.getConfig(client).getRequestContext()
>>>>> + .put(org.apache.cxf.message.Message.REQUEST_URI, "/bookstore/books/123");
>>>>> +
>>>>> + Book book = client.get(Book.class);
>>>>> + assertEquals("Get a wrong response code.", 200, client.getResponse().getStatus());
>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>> + }
>>>>> @Test
>>>>> public void testPutBookOneWayWithWebClient() throws Exception {
>>>>> @@ -130,6 +149,7 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>> }
>>>>> }
>>>>> +
>>>>> @Test
>>>>> public void testGetBookFromWebClientWithPath() throws Exception {
>>>>> // setup the the client
>>>>> @@ -148,6 +168,25 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>> }
>>>>> @Test
>>>>> + public void testGetBookFromWebClientWithPathWithTextJMSMessage() throws Exception {
>>>>> + // setup the the client
>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>> + + "&messageType=text";
>>>>> +
>>>>> + WebClient client = WebClient.create(endpointAddressUrlEncoded);
>>>>> + client.path("bookstore").path("books").path("123");
>>>>> +
>>>>> + Book book = client.get(Book.class);
>>>>> + assertEquals("Get a wrong response code.", 200, client.getResponse().getStatus());
>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>> + }
>>>>> +
>>>>> +
>>>>> + @Test
>>>>> public void testGetBookFromProxyClient() throws Exception {
>>>>> // setup the the client
>>>>> String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>> @@ -161,6 +200,22 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>> assertEquals("Get a wrong response code.", 200, WebClient.client(client).getResponse().getStatus());
>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>> }
>>>>> +
>>>>> + @Test
>>>>> + public void testGetBookFromProxyClientWithTextJMSMessage() throws Exception {
>>>>> + // setup the the client
>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>> + + "&messageType=text";
>>>>> +
>>>>> + JMSBookStore client = JAXRSClientFactory.create(endpointAddressUrlEncoded, JMSBookStore.class);
>>>>> + Book book = client.getBook("123");
>>>>> + assertEquals("Get a wrong response code.", 200, WebClient.client(client).getResponse().getStatus());
>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>> + }
>>>>> @Test
>>>>> public void testGetBookFromSubresourceProxyClient() throws Exception {
>>>>> @@ -177,6 +232,23 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>> assertEquals("Get a wrong response code.", 200, WebClient.client(bookProxy).getResponse().getStatus());
>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>> }
>>>>> +
>>>>> + @Test
>>>>> + public void testGetBookFromSubresourceProxyClientWithTextJMSMessage() throws Exception {
>>>>> + // setup the the client
>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>> + + "&messageType=text";
>>>>> +
>>>>> + JMSBookStore client = JAXRSClientFactory.create(endpointAddressUrlEncoded, JMSBookStore.class);
>>>>> + Book bookProxy = client.getBookSubResource("123");
>>>>> + Book book = bookProxy.retrieveState();
>>>>> + assertEquals("Get a wrong response code.", 200, WebClient.client(bookProxy).getResponse().getStatus());
>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>> + }
>>>>> @Test
>>>>> public void testGetBookFromProxyClientWithQuery() throws Exception {
>>>>> @@ -193,6 +265,24 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>> }
>>>>> +
>>>>> + @Test
>>>>> + public void testGetBookFromProxyClientWithQueryWithTextJMSMessage() throws Exception {
>>>>> + // setup the the client
>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>> + + "&messageType=text";
>>>>> +
>>>>> + JMSBookStore client = JAXRSClientFactory.create(endpointAddressUrlEncoded, JMSBookStore.class);
>>>>> + Book book = client.getBookByURLQuery(new String[] {"1", "2", "3"});
>>>>> + assertEquals("Get a wrong response code.", 200, WebClient.client(client).getResponse().getStatus());
>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>> + }
>>>>> +
>>>>> +
>>>>> @Test
>>>>> public void testGetBook() throws Exception {
>>>>> Context ctx = getContext();
>>>>>
>>>
Re: [cxf] branch master updated: [CXF-7532]REST on JMS transport
can't handle the request message with text messageType
Posted by Sergey Beryozkin <sb...@gmail.com>.
Thanks Freeman, it looks fine, yes, it's a bit tricky to make sure it
works across all of JAX-RS providers, especially with custom MBRs
expecting InputStream to be not null...
Sergey
On 20/10/17 03:08, Freeman Fang wrote:
> Hi Sergey,
>
> Cleaned up accordingly.
>
> Cheers
> －－－－－－－－－－－－－
> Freeman(Yue) Fang
>
> Red Hat, Inc.
> FuseSource is now part of Red Hat
>
>
>
>> On Oct 20, 2017, at 5:59 AM, Sergey Beryozkin <sberyozkin@gmail.com
>> <ma...@gmail.com>> wrote:
>>
>> Some JAX-RS providers check Reader or XMLStreamReader if InputStream
>> is null.
>> But at the moment, Freeman, please do a bit of cleanup as agreed, and
>> then we can check if the use ReaderInputStream can be avoided at all.
>>
>> Sergey
>>
>> On 19/10/17 19:47, Daniel Kulp wrote:
>>> If using the TEXT stuff for JMS, it should remain as a Reader on the
>>> message if at all possible. Converting from Reader to InputStream is
>>> not a trivial operation and does enact a performance penalty. All
>>> of the SOAP pathways for JMS should be OK for use of a Reader so that
>>> we can avoid the entire String->UTF-8 bytes->chars round tripping.
>>> Dan
>>> On Oct 19, 2017, at 5:21 AM, ffang <freeman.fang@gmail.com
>>> <ma...@gmail.com>> wrote:
>>>>
>>>> I modified the JMS transport part and set ReaderInputStream instead
>>>> of Reader, running all tests now, will see what I can get.
>>>>
>>>> If I run into any test failure, I will keep things as is, just clean up.
>>>> －－－－－－－－－－－－－
>>>> Freeman(Yue) Fang
>>>>
>>>> Red Hat, Inc.
>>>> FuseSource is now part of Red Hat
>>>>
>>>>
>>>>
>>>>> On Oct 19, 2017, at 5:16 PM, Sergey Beryozkin <sberyozkin@gmail.com
>>>>> <ma...@gmail.com>> wrote:
>>>>>
>>>>> If we assume that checking Reader can be useful in the general
>>>>> sense, then may be lets keep things as is, with the minor cleanup
>>>>> only...
>>>>>
>>>>> Sergey
>>>>> On 19/10/17 10:05, Sergey Beryozkin wrote:
>>>>>> Hi Freeman
>>>>>> Reader may likely need to be set too for SOAP, and I recall
>>>>>> checking Reader somewhere else too (will double check now), but I
>>>>>> thought may be the transport can set both Reader and InputStream,
>>>>>> hmm, that is an extra instance though, may be lets start with
>>>>>> moving ReaderInputStream to the core first ?
>>>>>> Sergey
>>>>>> On 19/10/17 09:59, ffang wrote:
>>>>>>> Yeah, set ReaderInputStream instead of Reader for the JMS
>>>>>>> transport is a better solution. If no objection I will go this way.
>>>>>>> －－－－－－－－－－－－－
>>>>>>> Freeman(Yue) Fang
>>>>>>>
>>>>>>> Red Hat, Inc.
>>>>>>> FuseSource is now part of Red Hat
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>>> On Oct 19, 2017, at 4:50 PM, Sergey Beryozkin
>>>>>>>> <sberyozkin@gmail.com <ma...@gmail.com>
>>>>>>>> <ma...@gmail.com>> wrote:
>>>>>>>>
>>>>>>>> The other thing is, should JMS transport itself set
>>>>>>>> ReaderInputStream as InputStream on the message, as opposed to
>>>>>>>> patching the JAX-RS code in various places ?
>>>>>>>>
>>>>>>>> Sergey
>>>>>>>> On 19/10/17 09:44, Sergey Beryozkin wrote:
>>>>>>>>> Hi Freeman
>>>>>>>>> Can you move ReaderInputStream to the CXF core where
>>>>>>>>> CachedInputStream and other IO utility code is located ?
>>>>>>>>> Please also update the code which gets Reader.class and check
>>>>>>>>> it for null first, and also drop a comment re the JMS
>>>>>>>>> transport, checking the reader can make sense in all the cases...
>>>>>>>>> Thanks, Sergey
>>>>>>>>> On 19/10/17 04:16, ffang@apache.org <ma...@apache.org>
>>>>>>>>> <ma...@apache.org> wrote:
>>>>>>>>>> This is an automated email from the ASF dual-hosted git
>>>>>>>>>> repository.
>>>>>>>>>>
>>>>>>>>>> ffang pushed a commit to branch master
>>>>>>>>>> in repository https://gitbox.apache.org/repos/asf/cxf.git
>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>> The following commit(s) were added to refs/heads/master by
>>>>>>>>>> this push:
>>>>>>>>>> new 1d66475 [CXF-7532]REST on JMS transport can't handle
>>>>>>>>>> the request message with text messageType
>>>>>>>>>> 1d66475 is described below
>>>>>>>>>>
>>>>>>>>>> commit 1d66475b4dea2a0031dcb1765e8e9fd08430ec9b
>>>>>>>>>> Author: Freeman Fang <freeman.fang@gmail.com
>>>>>>>>>> <ma...@gmail.com> <ma...@gmail.com>>
>>>>>>>>>> AuthorDate: Thu Oct 19 11:15:50 2017 +0800
>>>>>>>>>>
>>>>>>>>>> [CXF-7532]REST on JMS transport can't handle the request
>>>>>>>>>> message with text messageType
>>>>>>>>>> ---
>>>>>>>>>> .../org/apache/cxf/jaxrs/impl/ResponseImpl.java | 5 +
>>>>>>>>>> .../org/apache/cxf/jaxrs/utils/JAXRSUtils.java | 6 +-
>>>>>>>>>> .../apache/cxf/jaxrs/utils/ReaderInputStream.java | 296
>>>>>>>>>> +++++++++++++++++++++
>>>>>>>>>> .../apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java | 90 +++++++
>>>>>>>>>> 4 files changed, 396 insertions(+), 1 deletion(-)
>>>>>>>>>>
>>>>>>>>>> diff --git
>>>>>>>>>> a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>>>>>>> b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>>>>>>> index e3a3acf..82f183a 100644
>>>>>>>>>> ---
>>>>>>>>>> a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>>>>>>> +++
>>>>>>>>>> b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>>>>>>> @@ -58,6 +58,7 @@ import
>>>>>>>>>> org.apache.cxf.jaxrs.provider.ProviderFactory;
>>>>>>>>>> import org.apache.cxf.jaxrs.utils.HttpUtils;
>>>>>>>>>> import org.apache.cxf.jaxrs.utils.InjectionUtils;
>>>>>>>>>> import org.apache.cxf.jaxrs.utils.JAXRSUtils;
>>>>>>>>>> +import org.apache.cxf.jaxrs.utils.ReaderInputStream;
>>>>>>>>>> import org.apache.cxf.message.Message;
>>>>>>>>>> import org.apache.cxf.message.MessageUtils;
>>>>>>>>>> @@ -342,6 +343,10 @@ public final class ResponseImpl extends
>>>>>>>>>> Response {
>>>>>>>>>> entityStreamAvailable = entityStream != null;
>>>>>>>>>> } else if (entity instanceof InputStream) {
>>>>>>>>>> entityStream = InputStream.class.cast(entity);
>>>>>>>>>> + } else {
>>>>>>>>>> + Message inMessage = getResponseMessage();
>>>>>>>>>> + Reader reader = inMessage.getContent(Reader.class);
>>>>>>>>>> + entityStream = InputStream.class.cast(new
>>>>>>>>>> ReaderInputStream(reader));
>>>>>>>>>> }
>>>>>>>>>> // we need to check for readers even if no IS is set
>>>>>>>>>> - the readers may still do it
>>>>>>>>>> diff --git
>>>>>>>>>> a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>>>>>>> b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>>>>>>> index 0ef6565..509f76d 100644
>>>>>>>>>> ---
>>>>>>>>>> a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>>>>>>> +++
>>>>>>>>>> b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>>>>>>> @@ -832,7 +832,11 @@ public final class JAXRSUtils {
>>>>>>>>>> OperationResourceInfo
>>>>>>>>>> ori)
>>>>>>>>>> throws IOException, WebApplicationException {
>>>>>>>>>> InputStream is = message.getContent(InputStream.class);
>>>>>>>>>> -
>>>>>>>>>> + if (is == null) {
>>>>>>>>>> + //may use the jms transport so check the Reader;
>>>>>>>>>> + Reader reader = message.getContent(Reader.class);
>>>>>>>>>> + is = new ReaderInputStream(reader);
>>>>>>>>>> + }
>>>>>>>>>> if (parameter.getType() == ParameterType.REQUEST_BODY) {
>>>>>>>>>> if (parameterClass == AsyncResponse.class) {
>>>>>>>>>> diff --git
>>>>>>>>>> a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
>>>>>>>>>> b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
>>>>>>>>>> new file mode 100644
>>>>>>>>>> index 0000000..c7b142e
>>>>>>>>>> --- /dev/null
>>>>>>>>>> +++
>>>>>>>>>> b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
>>>>>>>>>> @@ -0,0 +1,296 @@
>>>>>>>>>> +/**
>>>>>>>>>> + * Licensed to the Apache Software Foundation (ASF) under one
>>>>>>>>>> + * or more contributor license agreements. See the NOTICE file
>>>>>>>>>> + * distributed with this work for additional information
>>>>>>>>>> + * regarding copyright ownership. The ASF licenses this file
>>>>>>>>>> + * to you 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.apache.cxf.jaxrs.utils;
>>>>>>>>>> +
>>>>>>>>>> +import java.io.IOException;
>>>>>>>>>> +import java.io.InputStream;
>>>>>>>>>> +import java.io.Reader;
>>>>>>>>>> +import java.nio.ByteBuffer;
>>>>>>>>>> +import java.nio.CharBuffer;
>>>>>>>>>> +import java.nio.charset.Charset;
>>>>>>>>>> +import java.nio.charset.CharsetEncoder;
>>>>>>>>>> +import java.nio.charset.CoderResult;
>>>>>>>>>> +import java.nio.charset.CodingErrorAction;
>>>>>>>>>> +
>>>>>>>>>> +/**
>>>>>>>>>> + * {@link InputStream} implementation that reads a character
>>>>>>>>>> stream from a {@link Reader}
>>>>>>>>>> + * and transforms it to a byte stream using a specified
>>>>>>>>>> charset encoding. The stream
>>>>>>>>>> + * is transformed using a {@link CharsetEncoder} object,
>>>>>>>>>> guaranteeing that all charset
>>>>>>>>>> + * encodings supported by the JRE are handled correctly. In
>>>>>>>>>> particular for charsets such as
>>>>>>>>>> + * UTF-16, the implementation ensures that one and only one
>>>>>>>>>> byte order marker
>>>>>>>>>> + * is produced.
>>>>>>>>>> + * <p>
>>>>>>>>>> + * Since in general it is not possible to predict the number
>>>>>>>>>> of characters to be read from the
>>>>>>>>>> + * {@link Reader} to satisfy a read request on the {@link
>>>>>>>>>> ReaderInputStream}, all reads from
>>>>>>>>>> + * the {@link Reader} are buffered. There is therefore no
>>>>>>>>>> well defined correlation
>>>>>>>>>> + * between the current position of the {@link Reader} and
>>>>>>>>>> that of the {@link ReaderInputStream}.
>>>>>>>>>> + * This also implies that in general there is no need to wrap
>>>>>>>>>> the underlying {@link Reader}
>>>>>>>>>> + * in a {@link java.io.BufferedReader}.
>>>>>>>>>> + * <p>
>>>>>>>>>> + * {@link ReaderInputStream} implements the inverse
>>>>>>>>>> transformation of {@link java.io.InputStreamReader};
>>>>>>>>>> + * in the following example, reading from <tt>in2</tt> would
>>>>>>>>>> return the same byte
>>>>>>>>>> + * sequence as reading from <tt>in</tt> (provided that the
>>>>>>>>>> initial byte sequence is legal
>>>>>>>>>> + * with respect to the charset encoding):
>>>>>>>>>> + * <pre>
>>>>>>>>>> + * InputStream in = ...
>>>>>>>>>> + * Charset cs = ...
>>>>>>>>>> + * InputStreamReader reader = new InputStreamReader(in, cs);
>>>>>>>>>> + * ReaderInputStream in2 = new ReaderInputStream(reader,
>>>>>>>>>> cs);</pre>
>>>>>>>>>> + * {@link ReaderInputStream} implements the same
>>>>>>>>>> transformation as {@link java.io.OutputStreamWriter},
>>>>>>>>>> + * except that the control flow is reversed: both classes
>>>>>>>>>> transform a character stream
>>>>>>>>>> + * into a byte stream, but {@link java.io.OutputStreamWriter}
>>>>>>>>>> pushes data to the underlying stream,
>>>>>>>>>> + * while {@link ReaderInputStream} pulls it from the
>>>>>>>>>> underlying stream.
>>>>>>>>>> + * <p>
>>>>>>>>>> + * Note that while there are use cases where there is no
>>>>>>>>>> alternative to using
>>>>>>>>>> + * this class, very often the need to use this class is an
>>>>>>>>>> indication of a flaw
>>>>>>>>>> + * in the design of the code. This class is typically used in
>>>>>>>>>> situations where an existing
>>>>>>>>>> + * API only accepts an {@link InputStream}, but where the
>>>>>>>>>> most natural way to produce the data
>>>>>>>>>> + * is as a character stream, i.e. by providing a {@link
>>>>>>>>>> Reader} instance. An example of a situation
>>>>>>>>>> + * where this problem may appear is when implementing the
>>>>>>>>>> {@link javax.activation.DataSource}
>>>>>>>>>> + * interface from the Java Activation Framework.
>>>>>>>>>> + * <p>
>>>>>>>>>> + * Given the fact that the {@link Reader} class doesn't
>>>>>>>>>> provide any way to predict whether the next
>>>>>>>>>> + * read operation will block or not, it is not possible to
>>>>>>>>>> provide a meaningful
>>>>>>>>>> + * implementation of the {@link InputStream#available()}
>>>>>>>>>> method. A call to this method
>>>>>>>>>> + * will always return 0. Also, this class doesn't support
>>>>>>>>>> {@link InputStream#mark(int)}.
>>>>>>>>>> + * <p>
>>>>>>>>>> + * Instances of {@link ReaderInputStream} are not thread safe.
>>>>>>>>>> + *
>>>>>>>>>> + * @see org.apache.commons.io.output.WriterOutputStream
>>>>>>>>>> + *
>>>>>>>>>> + * @since 2.0
>>>>>>>>>> + */
>>>>>>>>>> +public class ReaderInputStream extends InputStream {
>>>>>>>>>> + private static final int DEFAULT_BUFFER_SIZE = 1024;
>>>>>>>>>> +
>>>>>>>>>> + private final Reader reader;
>>>>>>>>>> + private final CharsetEncoder encoder;
>>>>>>>>>> +
>>>>>>>>>> + /**
>>>>>>>>>> + * CharBuffer used as input for the decoder. It should be
>>>>>>>>>> reasonably
>>>>>>>>>> + * large as we read data from the underlying Reader into
>>>>>>>>>> this buffer.
>>>>>>>>>> + */
>>>>>>>>>> + private final CharBuffer encoderIn;
>>>>>>>>>> +
>>>>>>>>>> + /**
>>>>>>>>>> + * ByteBuffer used as output for the decoder. This buffer
>>>>>>>>>> can be small
>>>>>>>>>> + * as it is only used to transfer data from the decoder
>>>>>>>>>> to the
>>>>>>>>>> + * buffer provided by the caller.
>>>>>>>>>> + */
>>>>>>>>>> + private final ByteBuffer encoderOut;
>>>>>>>>>> +
>>>>>>>>>> + private CoderResult lastCoderResult;
>>>>>>>>>> + private boolean endOfInput;
>>>>>>>>>> +
>>>>>>>>>> + /**
>>>>>>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>>>>>>> + *
>>>>>>>>>> + * @param reader the target {@link Reader}
>>>>>>>>>> + * @param encoder the charset encoder
>>>>>>>>>> + * @since 2.1
>>>>>>>>>> + */
>>>>>>>>>> + public ReaderInputStream(Reader reader, CharsetEncoder
>>>>>>>>>> encoder) {
>>>>>>>>>> + this(reader, encoder, DEFAULT_BUFFER_SIZE);
>>>>>>>>>> + }
>>>>>>>>>> +
>>>>>>>>>> + /**
>>>>>>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>>>>>>> + *
>>>>>>>>>> + * @param reader the target {@link Reader}
>>>>>>>>>> + * @param encoder the charset encoder
>>>>>>>>>> + * @param bufferSize the size of the input buffer in
>>>>>>>>>> number of characters
>>>>>>>>>> + * @since 2.1
>>>>>>>>>> + */
>>>>>>>>>> + public ReaderInputStream(Reader reader, CharsetEncoder
>>>>>>>>>> encoder, int bufferSize) {
>>>>>>>>>> + this.reader = reader;
>>>>>>>>>> + this.encoder = encoder;
>>>>>>>>>> + this.encoderIn = CharBuffer.allocate(bufferSize);
>>>>>>>>>> + this.encoderIn.flip();
>>>>>>>>>> + this.encoderOut = ByteBuffer.allocate(128);
>>>>>>>>>> + this.encoderOut.flip();
>>>>>>>>>> + }
>>>>>>>>>> +
>>>>>>>>>> + /**
>>>>>>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>>>>>>> + *
>>>>>>>>>> + * @param reader the target {@link Reader}
>>>>>>>>>> + * @param charset the charset encoding
>>>>>>>>>> + * @param bufferSize the size of the input buffer in
>>>>>>>>>> number of characters
>>>>>>>>>> + */
>>>>>>>>>> + public ReaderInputStream(Reader reader, Charset charset,
>>>>>>>>>> int bufferSize) {
>>>>>>>>>> + this(reader,
>>>>>>>>>> + charset.newEncoder()
>>>>>>>>>> + .onMalformedInput(CodingErrorAction.REPLACE)
>>>>>>>>>> +
>>>>>>>>>> .onUnmappableCharacter(CodingErrorAction.REPLACE),
>>>>>>>>>> + bufferSize);
>>>>>>>>>> + }
>>>>>>>>>> +
>>>>>>>>>> + /**
>>>>>>>>>> + * Construct a new {@link ReaderInputStream} with a
>>>>>>>>>> default input buffer size of
>>>>>>>>>> + * 1024 characters.
>>>>>>>>>> + *
>>>>>>>>>> + * @param reader the target {@link Reader}
>>>>>>>>>> + * @param charset the charset encoding
>>>>>>>>>> + */
>>>>>>>>>> + public ReaderInputStream(Reader reader, Charset charset) {
>>>>>>>>>> + this(reader, charset, DEFAULT_BUFFER_SIZE);
>>>>>>>>>> + }
>>>>>>>>>> +
>>>>>>>>>> + /**
>>>>>>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>>>>>>> + *
>>>>>>>>>> + * @param reader the target {@link Reader}
>>>>>>>>>> + * @param charsetName the name of the charset encoding
>>>>>>>>>> + * @param bufferSize the size of the input buffer in
>>>>>>>>>> number of characters
>>>>>>>>>> + */
>>>>>>>>>> + public ReaderInputStream(Reader reader, String
>>>>>>>>>> charsetName, int bufferSize) {
>>>>>>>>>> + this(reader, Charset.forName(charsetName), bufferSize);
>>>>>>>>>> + }
>>>>>>>>>> +
>>>>>>>>>> + /**
>>>>>>>>>> + * Construct a new {@link ReaderInputStream} with a
>>>>>>>>>> default input buffer size of
>>>>>>>>>> + * 1024 characters.
>>>>>>>>>> + *
>>>>>>>>>> + * @param reader the target {@link Reader}
>>>>>>>>>> + * @param charsetName the name of the charset encoding
>>>>>>>>>> + */
>>>>>>>>>> + public ReaderInputStream(Reader reader, String charsetName) {
>>>>>>>>>> + this(reader, charsetName, DEFAULT_BUFFER_SIZE);
>>>>>>>>>> + }
>>>>>>>>>> +
>>>>>>>>>> + /**
>>>>>>>>>> + * Construct a new {@link ReaderInputStream} that uses
>>>>>>>>>> the default character encoding
>>>>>>>>>> + * with a default input buffer size of 1024 characters.
>>>>>>>>>> + *
>>>>>>>>>> + * @param reader the target {@link Reader}
>>>>>>>>>> + */
>>>>>>>>>> + public ReaderInputStream(Reader reader) {
>>>>>>>>>> + this(reader, Charset.defaultCharset());
>>>>>>>>>> + }
>>>>>>>>>> +
>>>>>>>>>> + /**
>>>>>>>>>> + * Fills the internal char buffer from the reader.
>>>>>>>>>> + *
>>>>>>>>>> + * @throws IOException
>>>>>>>>>> + * If an I/O error occurs
>>>>>>>>>> + */
>>>>>>>>>> + private void fillBuffer() throws IOException {
>>>>>>>>>> + if (!endOfInput && (lastCoderResult == null ||
>>>>>>>>>> lastCoderResult.isUnderflow())) {
>>>>>>>>>> + encoderIn.compact();
>>>>>>>>>> + int position = encoderIn.position();
>>>>>>>>>> + // We don't use Reader#read(CharBuffer) here
>>>>>>>>>> because it is more efficient
>>>>>>>>>> + // to write directly to the underlying char array
>>>>>>>>>> (the default implementation
>>>>>>>>>> + // copies data to a temporary char array).
>>>>>>>>>> + int c = reader.read(encoderIn.array(), position,
>>>>>>>>>> encoderIn.remaining());
>>>>>>>>>> + if (c == -1) {
>>>>>>>>>> + endOfInput = true;
>>>>>>>>>> + } else {
>>>>>>>>>> + encoderIn.position(position + c);
>>>>>>>>>> + }
>>>>>>>>>> + encoderIn.flip();
>>>>>>>>>> + }
>>>>>>>>>> + encoderOut.compact();
>>>>>>>>>> + lastCoderResult = encoder.encode(encoderIn,
>>>>>>>>>> encoderOut, endOfInput);
>>>>>>>>>> + encoderOut.flip();
>>>>>>>>>> + }
>>>>>>>>>> +
>>>>>>>>>> + /**
>>>>>>>>>> + * Read the specified number of bytes into an array.
>>>>>>>>>> + *
>>>>>>>>>> + * @param b the byte array to read into
>>>>>>>>>> + * @param off the offset to start reading bytes into
>>>>>>>>>> + * @param len the number of bytes to read
>>>>>>>>>> + * @return the number of bytes read or <code>-1</code>
>>>>>>>>>> + * if the end of the stream has been reached
>>>>>>>>>> + * @throws IOException if an I/O error occurs
>>>>>>>>>> + */
>>>>>>>>>> + @Override
>>>>>>>>>> + public int read(byte[] b, int off, int len) throws
>>>>>>>>>> IOException {
>>>>>>>>>> + if (b == null) {
>>>>>>>>>> + throw new NullPointerException("Byte array must
>>>>>>>>>> not be null");
>>>>>>>>>> + }
>>>>>>>>>> + if (len < 0 || off < 0 || (off + len) > b.length) {
>>>>>>>>>> + throw new IndexOutOfBoundsException("Array Size="
>>>>>>>>>> + b.length
>>>>>>>>>> + + ", offset=" + off + ", length=" + len);
>>>>>>>>>> + }
>>>>>>>>>> + int read = 0;
>>>>>>>>>> + if (len == 0) {
>>>>>>>>>> + return 0; // Always return 0 if len == 0
>>>>>>>>>> + }
>>>>>>>>>> + while (len > 0) {
>>>>>>>>>> + if (encoderOut.hasRemaining()) {
>>>>>>>>>> + int c = Math.min(encoderOut.remaining(), len);
>>>>>>>>>> + encoderOut.get(b, off, c);
>>>>>>>>>> + off += c;
>>>>>>>>>> + len -= c;
>>>>>>>>>> + read += c;
>>>>>>>>>> + } else {
>>>>>>>>>> + fillBuffer();
>>>>>>>>>> + if (endOfInput && !encoderOut.hasRemaining()) {
>>>>>>>>>> + break;
>>>>>>>>>> + }
>>>>>>>>>> + }
>>>>>>>>>> + }
>>>>>>>>>> + return read == 0 && endOfInput ? -1 : read;
>>>>>>>>>> + }
>>>>>>>>>> +
>>>>>>>>>> + /**
>>>>>>>>>> + * Read the specified number of bytes into an array.
>>>>>>>>>> + *
>>>>>>>>>> + * @param b the byte array to read into
>>>>>>>>>> + * @return the number of bytes read or <code>-1</code>
>>>>>>>>>> + * if the end of the stream has been reached
>>>>>>>>>> + * @throws IOException if an I/O error occurs
>>>>>>>>>> + */
>>>>>>>>>> + @Override
>>>>>>>>>> + public int read(byte[] b) throws IOException {
>>>>>>>>>> + return read(b, 0, b.length);
>>>>>>>>>> + }
>>>>>>>>>> +
>>>>>>>>>> + /**
>>>>>>>>>> + * Read a single byte.
>>>>>>>>>> + *
>>>>>>>>>> + * @return either the byte read or <code>-1</code> if the
>>>>>>>>>> end of the stream
>>>>>>>>>> + * has been reached
>>>>>>>>>> + * @throws IOException if an I/O error occurs
>>>>>>>>>> + */
>>>>>>>>>> + @Override
>>>>>>>>>> + public int read() throws IOException {
>>>>>>>>>> + for (;;) {
>>>>>>>>>> + if (encoderOut.hasRemaining()) {
>>>>>>>>>> + return encoderOut.get() & 0xFF;
>>>>>>>>>> + } else {
>>>>>>>>>> + fillBuffer();
>>>>>>>>>> + if (endOfInput && !encoderOut.hasRemaining()) {
>>>>>>>>>> + return -1;
>>>>>>>>>> + }
>>>>>>>>>> + }
>>>>>>>>>> + }
>>>>>>>>>> + }
>>>>>>>>>> +
>>>>>>>>>> + /**
>>>>>>>>>> + * Close the stream. This method will cause the
>>>>>>>>>> underlying {@link Reader}
>>>>>>>>>> + * to be closed.
>>>>>>>>>> + * @throws IOException if an I/O error occurs
>>>>>>>>>> + */
>>>>>>>>>> + @Override
>>>>>>>>>> + public void close() throws IOException {
>>>>>>>>>> + reader.close();
>>>>>>>>>> + }
>>>>>>>>>> +}
>>>>>>>>>> diff --git
>>>>>>>>>> a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>>>>>>> b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>>>>>>> index f77b5ed..b957f36 100644
>>>>>>>>>> ---
>>>>>>>>>> a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>>>>>>> +++
>>>>>>>>>> b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>>>>>>> @@ -96,6 +96,25 @@ public class JAXRSJmsTest extends
>>>>>>>>>> AbstractBusClientServerTestBase {
>>>>>>>>>> assertEquals("Get a wrong response code.", 200,
>>>>>>>>>> client.getResponse().getStatus());
>>>>>>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>>>> }
>>>>>>>>>> +
>>>>>>>>>> + @Test
>>>>>>>>>> + public void testGetBookFromWebClientWithTextJMSMessage()
>>>>>>>>>> throws Exception {
>>>>>>>>>> + // setup the the client
>>>>>>>>>> + String endpointAddressUrlEncoded =
>>>>>>>>>> "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>>>>>> + +
>>>>>>>>>> "?replyToName=dynamicQueues/test.jmstransport.response"
>>>>>>>>>> + +
>>>>>>>>>> "&jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>>>>>> + + "&messageType=text";
>>>>>>>>>> +
>>>>>>>>>> + WebClient client =
>>>>>>>>>> WebClient.create(endpointAddressUrlEncoded);
>>>>>>>>>> +
>>>>>>>>>> WebClient.getConfig(client).getInInterceptors().add(new
>>>>>>>>>> LoggingInInterceptor());
>>>>>>>>>> + WebClient.getConfig(client).getRequestContext()
>>>>>>>>>> + .put(org.apache.cxf.message.Message.REQUEST_URI,
>>>>>>>>>> "/bookstore/books/123");
>>>>>>>>>> +
>>>>>>>>>> + Book book = client.get(Book.class);
>>>>>>>>>> + assertEquals("Get a wrong response code.", 200,
>>>>>>>>>> client.getResponse().getStatus());
>>>>>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>>>> + }
>>>>>>>>>> @Test
>>>>>>>>>> public void testPutBookOneWayWithWebClient() throws
>>>>>>>>>> Exception {
>>>>>>>>>> @@ -130,6 +149,7 @@ public class JAXRSJmsTest extends
>>>>>>>>>> AbstractBusClientServerTestBase {
>>>>>>>>>> }
>>>>>>>>>> }
>>>>>>>>>> +
>>>>>>>>>> @Test
>>>>>>>>>> public void testGetBookFromWebClientWithPath() throws
>>>>>>>>>> Exception {
>>>>>>>>>> // setup the the client
>>>>>>>>>> @@ -148,6 +168,25 @@ public class JAXRSJmsTest extends
>>>>>>>>>> AbstractBusClientServerTestBase {
>>>>>>>>>> }
>>>>>>>>>> @Test
>>>>>>>>>> + public void
>>>>>>>>>> testGetBookFromWebClientWithPathWithTextJMSMessage() throws
>>>>>>>>>> Exception {
>>>>>>>>>> + // setup the the client
>>>>>>>>>> + String endpointAddressUrlEncoded =
>>>>>>>>>> "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>>>>>> + +
>>>>>>>>>> "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>>>>>> + +
>>>>>>>>>> "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>>>>>>> + + "&messageType=text";
>>>>>>>>>> +
>>>>>>>>>> + WebClient client =
>>>>>>>>>> WebClient.create(endpointAddressUrlEncoded);
>>>>>>>>>> + client.path("bookstore").path("books").path("123");
>>>>>>>>>> +
>>>>>>>>>> + Book book = client.get(Book.class);
>>>>>>>>>> + assertEquals("Get a wrong response code.", 200,
>>>>>>>>>> client.getResponse().getStatus());
>>>>>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>>>> + }
>>>>>>>>>> +
>>>>>>>>>> +
>>>>>>>>>> + @Test
>>>>>>>>>> public void testGetBookFromProxyClient() throws Exception {
>>>>>>>>>> // setup the the client
>>>>>>>>>> String endpointAddressUrlEncoded =
>>>>>>>>>> "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>>>>>> @@ -161,6 +200,22 @@ public class JAXRSJmsTest extends
>>>>>>>>>> AbstractBusClientServerTestBase {
>>>>>>>>>> assertEquals("Get a wrong response code.", 200,
>>>>>>>>>> WebClient.client(client).getResponse().getStatus());
>>>>>>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>>>> }
>>>>>>>>>> +
>>>>>>>>>> + @Test
>>>>>>>>>> + public void
>>>>>>>>>> testGetBookFromProxyClientWithTextJMSMessage() throws Exception {
>>>>>>>>>> + // setup the the client
>>>>>>>>>> + String endpointAddressUrlEncoded =
>>>>>>>>>> "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>>>>>> + +
>>>>>>>>>> "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>>>>>> + +
>>>>>>>>>> "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>>>>>>> + + "&messageType=text";
>>>>>>>>>> +
>>>>>>>>>> + JMSBookStore client =
>>>>>>>>>> JAXRSClientFactory.create(endpointAddressUrlEncoded,
>>>>>>>>>> JMSBookStore.class);
>>>>>>>>>> + Book book = client.getBook("123");
>>>>>>>>>> + assertEquals("Get a wrong response code.", 200,
>>>>>>>>>> WebClient.client(client).getResponse().getStatus());
>>>>>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>>>> + }
>>>>>>>>>> @Test
>>>>>>>>>> public void testGetBookFromSubresourceProxyClient()
>>>>>>>>>> throws Exception {
>>>>>>>>>> @@ -177,6 +232,23 @@ public class JAXRSJmsTest extends
>>>>>>>>>> AbstractBusClientServerTestBase {
>>>>>>>>>> assertEquals("Get a wrong response code.", 200,
>>>>>>>>>> WebClient.client(bookProxy).getResponse().getStatus());
>>>>>>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>>>> }
>>>>>>>>>> +
>>>>>>>>>> + @Test
>>>>>>>>>> + public void
>>>>>>>>>> testGetBookFromSubresourceProxyClientWithTextJMSMessage()
>>>>>>>>>> throws Exception {
>>>>>>>>>> + // setup the the client
>>>>>>>>>> + String endpointAddressUrlEncoded =
>>>>>>>>>> "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>>>>>> + +
>>>>>>>>>> "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>>>>>> + +
>>>>>>>>>> "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>>>>>>> + + "&messageType=text";
>>>>>>>>>> +
>>>>>>>>>> + JMSBookStore client =
>>>>>>>>>> JAXRSClientFactory.create(endpointAddressUrlEncoded,
>>>>>>>>>> JMSBookStore.class);
>>>>>>>>>> + Book bookProxy = client.getBookSubResource("123");
>>>>>>>>>> + Book book = bookProxy.retrieveState();
>>>>>>>>>> + assertEquals("Get a wrong response code.", 200,
>>>>>>>>>> WebClient.client(bookProxy).getResponse().getStatus());
>>>>>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>>>> + }
>>>>>>>>>> @Test
>>>>>>>>>> public void testGetBookFromProxyClientWithQuery() throws
>>>>>>>>>> Exception {
>>>>>>>>>> @@ -193,6 +265,24 @@ public class JAXRSJmsTest extends
>>>>>>>>>> AbstractBusClientServerTestBase {
>>>>>>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>>>> }
>>>>>>>>>> +
>>>>>>>>>> + @Test
>>>>>>>>>> + public void
>>>>>>>>>> testGetBookFromProxyClientWithQueryWithTextJMSMessage() throws
>>>>>>>>>> Exception {
>>>>>>>>>> + // setup the the client
>>>>>>>>>> + String endpointAddressUrlEncoded =
>>>>>>>>>> "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>>>>>> + +
>>>>>>>>>> "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>>>>>> + +
>>>>>>>>>> "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>>>>>>> + + "&messageType=text";
>>>>>>>>>> +
>>>>>>>>>> + JMSBookStore client =
>>>>>>>>>> JAXRSClientFactory.create(endpointAddressUrlEncoded,
>>>>>>>>>> JMSBookStore.class);
>>>>>>>>>> + Book book = client.getBookByURLQuery(new String[]
>>>>>>>>>> {"1", "2", "3"});
>>>>>>>>>> + assertEquals("Get a wrong response code.", 200,
>>>>>>>>>> WebClient.client(client).getResponse().getStatus());
>>>>>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>>>> + }
>>>>>>>>>> +
>>>>>>>>>> +
>>>>>>>>>> @Test
>>>>>>>>>> public void testGetBook() throws Exception {
>>>>>>>>>> Context ctx = getContext();
>>>>>>>>>>
>>>>>>>>
>>>>>>>
>>>>>
>>>>>
>>>>> --
>>>>> Sergey Beryozkin
>>>>>
>>>>> Talend Community Coders
>>>>> http://coders.talend.com/
>>>>
>
Re: [cxf] branch master updated: [CXF-7532]REST on JMS transport can't handle the request message with text messageType
Posted by Freeman Fang <fr...@gmail.com>.
Hi Sergey,
Cleaned up accordingly.
Cheers
－－－－－－－－－－－－－
Freeman(Yue) Fang
Red Hat, Inc.
FuseSource is now part of Red Hat
> On Oct 20, 2017, at 5:59 AM, Sergey Beryozkin <sb...@gmail.com> wrote:
>
> Some JAX-RS providers check Reader or XMLStreamReader if InputStream is null.
> But at the moment, Freeman, please do a bit of cleanup as agreed, and then we can check if the use ReaderInputStream can be avoided at all.
>
> Sergey
>
> On 19/10/17 19:47, Daniel Kulp wrote:
>> If using the TEXT stuff for JMS, it should remain as a Reader on the message if at all possible. Converting from Reader to InputStream is not a trivial operation and does enact a performance penalty. All of the SOAP pathways for JMS should be OK for use of a Reader so that we can avoid the entire String->UTF-8 bytes->chars round tripping.
>> Dan
>> On Oct 19, 2017, at 5:21 AM, ffang <fr...@gmail.com> wrote:
>>>
>>> I modified the JMS transport part and set ReaderInputStream instead of Reader, running all tests now, will see what I can get.
>>>
>>> If I run into any test failure, I will keep things as is, just clean up.
>>> －－－－－－－－－－－－－
>>> Freeman(Yue) Fang
>>>
>>> Red Hat, Inc.
>>> FuseSource is now part of Red Hat
>>>
>>>
>>>
>>>> On Oct 19, 2017, at 5:16 PM, Sergey Beryozkin <sb...@gmail.com> wrote:
>>>>
>>>> If we assume that checking Reader can be useful in the general sense, then may be lets keep things as is, with the minor cleanup only...
>>>>
>>>> Sergey
>>>> On 19/10/17 10:05, Sergey Beryozkin wrote:
>>>>> Hi Freeman
>>>>> Reader may likely need to be set too for SOAP, and I recall checking Reader somewhere else too (will double check now), but I thought may be the transport can set both Reader and InputStream, hmm, that is an extra instance though, may be lets start with moving ReaderInputStream to the core first ?
>>>>> Sergey
>>>>> On 19/10/17 09:59, ffang wrote:
>>>>>> Yeah, set ReaderInputStream instead of Reader for the JMS transport is a better solution. If no objection I will go this way.
>>>>>> －－－－－－－－－－－－－
>>>>>> Freeman(Yue) Fang
>>>>>>
>>>>>> Red Hat, Inc.
>>>>>> FuseSource is now part of Red Hat
>>>>>>
>>>>>>
>>>>>>
>>>>>>> On Oct 19, 2017, at 4:50 PM, Sergey Beryozkin <sberyozkin@gmail.com <ma...@gmail.com>> wrote:
>>>>>>>
>>>>>>> The other thing is, should JMS transport itself set ReaderInputStream as InputStream on the message, as opposed to patching the JAX-RS code in various places ?
>>>>>>>
>>>>>>> Sergey
>>>>>>> On 19/10/17 09:44, Sergey Beryozkin wrote:
>>>>>>>> Hi Freeman
>>>>>>>> Can you move ReaderInputStream to the CXF core where CachedInputStream and other IO utility code is located ?
>>>>>>>> Please also update the code which gets Reader.class and check it for null first, and also drop a comment re the JMS transport, checking the reader can make sense in all the cases...
>>>>>>>> Thanks, Sergey
>>>>>>>> On 19/10/17 04:16, ffang@apache.org <ma...@apache.org> wrote:
>>>>>>>>> This is an automated email from the ASF dual-hosted git repository.
>>>>>>>>>
>>>>>>>>> ffang pushed a commit to branch master
>>>>>>>>> in repository https://gitbox.apache.org/repos/asf/cxf.git
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> The following commit(s) were added to refs/heads/master by this push:
>>>>>>>>> new 1d66475 [CXF-7532]REST on JMS transport can't handle the request message with text messageType
>>>>>>>>> 1d66475 is described below
>>>>>>>>>
>>>>>>>>> commit 1d66475b4dea2a0031dcb1765e8e9fd08430ec9b
>>>>>>>>> Author: Freeman Fang <freeman.fang@gmail.com <ma...@gmail.com>>
>>>>>>>>> AuthorDate: Thu Oct 19 11:15:50 2017 +0800
>>>>>>>>>
>>>>>>>>> [CXF-7532]REST on JMS transport can't handle the request message with text messageType
>>>>>>>>> ---
>>>>>>>>> .../org/apache/cxf/jaxrs/impl/ResponseImpl.java | 5 +
>>>>>>>>> .../org/apache/cxf/jaxrs/utils/JAXRSUtils.java | 6 +-
>>>>>>>>> .../apache/cxf/jaxrs/utils/ReaderInputStream.java | 296 +++++++++++++++++++++
>>>>>>>>> .../apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java | 90 +++++++
>>>>>>>>> 4 files changed, 396 insertions(+), 1 deletion(-)
>>>>>>>>>
>>>>>>>>> diff --git a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>>>>>> index e3a3acf..82f183a 100644
>>>>>>>>> --- a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>>>>>> +++ b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>>>>>> @@ -58,6 +58,7 @@ import org.apache.cxf.jaxrs.provider.ProviderFactory;
>>>>>>>>> import org.apache.cxf.jaxrs.utils.HttpUtils;
>>>>>>>>> import org.apache.cxf.jaxrs.utils.InjectionUtils;
>>>>>>>>> import org.apache.cxf.jaxrs.utils.JAXRSUtils;
>>>>>>>>> +import org.apache.cxf.jaxrs.utils.ReaderInputStream;
>>>>>>>>> import org.apache.cxf.message.Message;
>>>>>>>>> import org.apache.cxf.message.MessageUtils;
>>>>>>>>> @@ -342,6 +343,10 @@ public final class ResponseImpl extends Response {
>>>>>>>>> entityStreamAvailable = entityStream != null;
>>>>>>>>> } else if (entity instanceof InputStream) {
>>>>>>>>> entityStream = InputStream.class.cast(entity);
>>>>>>>>> + } else {
>>>>>>>>> + Message inMessage = getResponseMessage();
>>>>>>>>> + Reader reader = inMessage.getContent(Reader.class);
>>>>>>>>> + entityStream = InputStream.class.cast(new ReaderInputStream(reader));
>>>>>>>>> }
>>>>>>>>> // we need to check for readers even if no IS is set - the readers may still do it
>>>>>>>>> diff --git a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>>>>>> index 0ef6565..509f76d 100644
>>>>>>>>> --- a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>>>>>> +++ b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>>>>>> @@ -832,7 +832,11 @@ public final class JAXRSUtils {
>>>>>>>>> OperationResourceInfo ori)
>>>>>>>>> throws IOException, WebApplicationException {
>>>>>>>>> InputStream is = message.getContent(InputStream.class);
>>>>>>>>> -
>>>>>>>>> + if (is == null) {
>>>>>>>>> + //may use the jms transport so check the Reader;
>>>>>>>>> + Reader reader = message.getContent(Reader.class);
>>>>>>>>> + is = new ReaderInputStream(reader);
>>>>>>>>> + }
>>>>>>>>> if (parameter.getType() == ParameterType.REQUEST_BODY) {
>>>>>>>>> if (parameterClass == AsyncResponse.class) {
>>>>>>>>> diff --git a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
>>>>>>>>> new file mode 100644
>>>>>>>>> index 0000000..c7b142e
>>>>>>>>> --- /dev/null
>>>>>>>>> +++ b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
>>>>>>>>> @@ -0,0 +1,296 @@
>>>>>>>>> +/**
>>>>>>>>> + * Licensed to the Apache Software Foundation (ASF) under one
>>>>>>>>> + * or more contributor license agreements. See the NOTICE file
>>>>>>>>> + * distributed with this work for additional information
>>>>>>>>> + * regarding copyright ownership. The ASF licenses this file
>>>>>>>>> + * to you 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.apache.cxf.jaxrs.utils;
>>>>>>>>> +
>>>>>>>>> +import java.io.IOException;
>>>>>>>>> +import java.io.InputStream;
>>>>>>>>> +import java.io.Reader;
>>>>>>>>> +import java.nio.ByteBuffer;
>>>>>>>>> +import java.nio.CharBuffer;
>>>>>>>>> +import java.nio.charset.Charset;
>>>>>>>>> +import java.nio.charset.CharsetEncoder;
>>>>>>>>> +import java.nio.charset.CoderResult;
>>>>>>>>> +import java.nio.charset.CodingErrorAction;
>>>>>>>>> +
>>>>>>>>> +/**
>>>>>>>>> + * {@link InputStream} implementation that reads a character stream from a {@link Reader}
>>>>>>>>> + * and transforms it to a byte stream using a specified charset encoding. The stream
>>>>>>>>> + * is transformed using a {@link CharsetEncoder} object, guaranteeing that all charset
>>>>>>>>> + * encodings supported by the JRE are handled correctly. In particular for charsets such as
>>>>>>>>> + * UTF-16, the implementation ensures that one and only one byte order marker
>>>>>>>>> + * is produced.
>>>>>>>>> + * <p>
>>>>>>>>> + * Since in general it is not possible to predict the number of characters to be read from the
>>>>>>>>> + * {@link Reader} to satisfy a read request on the {@link ReaderInputStream}, all reads from
>>>>>>>>> + * the {@link Reader} are buffered. There is therefore no well defined correlation
>>>>>>>>> + * between the current position of the {@link Reader} and that of the {@link ReaderInputStream}.
>>>>>>>>> + * This also implies that in general there is no need to wrap the underlying {@link Reader}
>>>>>>>>> + * in a {@link java.io.BufferedReader}.
>>>>>>>>> + * <p>
>>>>>>>>> + * {@link ReaderInputStream} implements the inverse transformation of {@link java.io.InputStreamReader};
>>>>>>>>> + * in the following example, reading from <tt>in2</tt> would return the same byte
>>>>>>>>> + * sequence as reading from <tt>in</tt> (provided that the initial byte sequence is legal
>>>>>>>>> + * with respect to the charset encoding):
>>>>>>>>> + * <pre>
>>>>>>>>> + * InputStream in = ...
>>>>>>>>> + * Charset cs = ...
>>>>>>>>> + * InputStreamReader reader = new InputStreamReader(in, cs);
>>>>>>>>> + * ReaderInputStream in2 = new ReaderInputStream(reader, cs);</pre>
>>>>>>>>> + * {@link ReaderInputStream} implements the same transformation as {@link java.io.OutputStreamWriter},
>>>>>>>>> + * except that the control flow is reversed: both classes transform a character stream
>>>>>>>>> + * into a byte stream, but {@link java.io.OutputStreamWriter} pushes data to the underlying stream,
>>>>>>>>> + * while {@link ReaderInputStream} pulls it from the underlying stream.
>>>>>>>>> + * <p>
>>>>>>>>> + * Note that while there are use cases where there is no alternative to using
>>>>>>>>> + * this class, very often the need to use this class is an indication of a flaw
>>>>>>>>> + * in the design of the code. This class is typically used in situations where an existing
>>>>>>>>> + * API only accepts an {@link InputStream}, but where the most natural way to produce the data
>>>>>>>>> + * is as a character stream, i.e. by providing a {@link Reader} instance. An example of a situation
>>>>>>>>> + * where this problem may appear is when implementing the {@link javax.activation.DataSource}
>>>>>>>>> + * interface from the Java Activation Framework.
>>>>>>>>> + * <p>
>>>>>>>>> + * Given the fact that the {@link Reader} class doesn't provide any way to predict whether the next
>>>>>>>>> + * read operation will block or not, it is not possible to provide a meaningful
>>>>>>>>> + * implementation of the {@link InputStream#available()} method. A call to this method
>>>>>>>>> + * will always return 0. Also, this class doesn't support {@link InputStream#mark(int)}.
>>>>>>>>> + * <p>
>>>>>>>>> + * Instances of {@link ReaderInputStream} are not thread safe.
>>>>>>>>> + *
>>>>>>>>> + * @see org.apache.commons.io.output.WriterOutputStream
>>>>>>>>> + *
>>>>>>>>> + * @since 2.0
>>>>>>>>> + */
>>>>>>>>> +public class ReaderInputStream extends InputStream {
>>>>>>>>> + private static final int DEFAULT_BUFFER_SIZE = 1024;
>>>>>>>>> +
>>>>>>>>> + private final Reader reader;
>>>>>>>>> + private final CharsetEncoder encoder;
>>>>>>>>> +
>>>>>>>>> + /**
>>>>>>>>> + * CharBuffer used as input for the decoder. It should be reasonably
>>>>>>>>> + * large as we read data from the underlying Reader into this buffer.
>>>>>>>>> + */
>>>>>>>>> + private final CharBuffer encoderIn;
>>>>>>>>> +
>>>>>>>>> + /**
>>>>>>>>> + * ByteBuffer used as output for the decoder. This buffer can be small
>>>>>>>>> + * as it is only used to transfer data from the decoder to the
>>>>>>>>> + * buffer provided by the caller.
>>>>>>>>> + */
>>>>>>>>> + private final ByteBuffer encoderOut;
>>>>>>>>> +
>>>>>>>>> + private CoderResult lastCoderResult;
>>>>>>>>> + private boolean endOfInput;
>>>>>>>>> +
>>>>>>>>> + /**
>>>>>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>>>>>> + *
>>>>>>>>> + * @param reader the target {@link Reader}
>>>>>>>>> + * @param encoder the charset encoder
>>>>>>>>> + * @since 2.1
>>>>>>>>> + */
>>>>>>>>> + public ReaderInputStream(Reader reader, CharsetEncoder encoder) {
>>>>>>>>> + this(reader, encoder, DEFAULT_BUFFER_SIZE);
>>>>>>>>> + }
>>>>>>>>> +
>>>>>>>>> + /**
>>>>>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>>>>>> + *
>>>>>>>>> + * @param reader the target {@link Reader}
>>>>>>>>> + * @param encoder the charset encoder
>>>>>>>>> + * @param bufferSize the size of the input buffer in number of characters
>>>>>>>>> + * @since 2.1
>>>>>>>>> + */
>>>>>>>>> + public ReaderInputStream(Reader reader, CharsetEncoder encoder, int bufferSize) {
>>>>>>>>> + this.reader = reader;
>>>>>>>>> + this.encoder = encoder;
>>>>>>>>> + this.encoderIn = CharBuffer.allocate(bufferSize);
>>>>>>>>> + this.encoderIn.flip();
>>>>>>>>> + this.encoderOut = ByteBuffer.allocate(128);
>>>>>>>>> + this.encoderOut.flip();
>>>>>>>>> + }
>>>>>>>>> +
>>>>>>>>> + /**
>>>>>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>>>>>> + *
>>>>>>>>> + * @param reader the target {@link Reader}
>>>>>>>>> + * @param charset the charset encoding
>>>>>>>>> + * @param bufferSize the size of the input buffer in number of characters
>>>>>>>>> + */
>>>>>>>>> + public ReaderInputStream(Reader reader, Charset charset, int bufferSize) {
>>>>>>>>> + this(reader,
>>>>>>>>> + charset.newEncoder()
>>>>>>>>> + .onMalformedInput(CodingErrorAction.REPLACE)
>>>>>>>>> + .onUnmappableCharacter(CodingErrorAction.REPLACE),
>>>>>>>>> + bufferSize);
>>>>>>>>> + }
>>>>>>>>> +
>>>>>>>>> + /**
>>>>>>>>> + * Construct a new {@link ReaderInputStream} with a default input buffer size of
>>>>>>>>> + * 1024 characters.
>>>>>>>>> + *
>>>>>>>>> + * @param reader the target {@link Reader}
>>>>>>>>> + * @param charset the charset encoding
>>>>>>>>> + */
>>>>>>>>> + public ReaderInputStream(Reader reader, Charset charset) {
>>>>>>>>> + this(reader, charset, DEFAULT_BUFFER_SIZE);
>>>>>>>>> + }
>>>>>>>>> +
>>>>>>>>> + /**
>>>>>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>>>>>> + *
>>>>>>>>> + * @param reader the target {@link Reader}
>>>>>>>>> + * @param charsetName the name of the charset encoding
>>>>>>>>> + * @param bufferSize the size of the input buffer in number of characters
>>>>>>>>> + */
>>>>>>>>> + public ReaderInputStream(Reader reader, String charsetName, int bufferSize) {
>>>>>>>>> + this(reader, Charset.forName(charsetName), bufferSize);
>>>>>>>>> + }
>>>>>>>>> +
>>>>>>>>> + /**
>>>>>>>>> + * Construct a new {@link ReaderInputStream} with a default input buffer size of
>>>>>>>>> + * 1024 characters.
>>>>>>>>> + *
>>>>>>>>> + * @param reader the target {@link Reader}
>>>>>>>>> + * @param charsetName the name of the charset encoding
>>>>>>>>> + */
>>>>>>>>> + public ReaderInputStream(Reader reader, String charsetName) {
>>>>>>>>> + this(reader, charsetName, DEFAULT_BUFFER_SIZE);
>>>>>>>>> + }
>>>>>>>>> +
>>>>>>>>> + /**
>>>>>>>>> + * Construct a new {@link ReaderInputStream} that uses the default character encoding
>>>>>>>>> + * with a default input buffer size of 1024 characters.
>>>>>>>>> + *
>>>>>>>>> + * @param reader the target {@link Reader}
>>>>>>>>> + */
>>>>>>>>> + public ReaderInputStream(Reader reader) {
>>>>>>>>> + this(reader, Charset.defaultCharset());
>>>>>>>>> + }
>>>>>>>>> +
>>>>>>>>> + /**
>>>>>>>>> + * Fills the internal char buffer from the reader.
>>>>>>>>> + *
>>>>>>>>> + * @throws IOException
>>>>>>>>> + * If an I/O error occurs
>>>>>>>>> + */
>>>>>>>>> + private void fillBuffer() throws IOException {
>>>>>>>>> + if (!endOfInput && (lastCoderResult == null || lastCoderResult.isUnderflow())) {
>>>>>>>>> + encoderIn.compact();
>>>>>>>>> + int position = encoderIn.position();
>>>>>>>>> + // We don't use Reader#read(CharBuffer) here because it is more efficient
>>>>>>>>> + // to write directly to the underlying char array (the default implementation
>>>>>>>>> + // copies data to a temporary char array).
>>>>>>>>> + int c = reader.read(encoderIn.array(), position, encoderIn.remaining());
>>>>>>>>> + if (c == -1) {
>>>>>>>>> + endOfInput = true;
>>>>>>>>> + } else {
>>>>>>>>> + encoderIn.position(position + c);
>>>>>>>>> + }
>>>>>>>>> + encoderIn.flip();
>>>>>>>>> + }
>>>>>>>>> + encoderOut.compact();
>>>>>>>>> + lastCoderResult = encoder.encode(encoderIn, encoderOut, endOfInput);
>>>>>>>>> + encoderOut.flip();
>>>>>>>>> + }
>>>>>>>>> +
>>>>>>>>> + /**
>>>>>>>>> + * Read the specified number of bytes into an array.
>>>>>>>>> + *
>>>>>>>>> + * @param b the byte array to read into
>>>>>>>>> + * @param off the offset to start reading bytes into
>>>>>>>>> + * @param len the number of bytes to read
>>>>>>>>> + * @return the number of bytes read or <code>-1</code>
>>>>>>>>> + * if the end of the stream has been reached
>>>>>>>>> + * @throws IOException if an I/O error occurs
>>>>>>>>> + */
>>>>>>>>> + @Override
>>>>>>>>> + public int read(byte[] b, int off, int len) throws IOException {
>>>>>>>>> + if (b == null) {
>>>>>>>>> + throw new NullPointerException("Byte array must not be null");
>>>>>>>>> + }
>>>>>>>>> + if (len < 0 || off < 0 || (off + len) > b.length) {
>>>>>>>>> + throw new IndexOutOfBoundsException("Array Size=" + b.length
>>>>>>>>> + + ", offset=" + off + ", length=" + len);
>>>>>>>>> + }
>>>>>>>>> + int read = 0;
>>>>>>>>> + if (len == 0) {
>>>>>>>>> + return 0; // Always return 0 if len == 0
>>>>>>>>> + }
>>>>>>>>> + while (len > 0) {
>>>>>>>>> + if (encoderOut.hasRemaining()) {
>>>>>>>>> + int c = Math.min(encoderOut.remaining(), len);
>>>>>>>>> + encoderOut.get(b, off, c);
>>>>>>>>> + off += c;
>>>>>>>>> + len -= c;
>>>>>>>>> + read += c;
>>>>>>>>> + } else {
>>>>>>>>> + fillBuffer();
>>>>>>>>> + if (endOfInput && !encoderOut.hasRemaining()) {
>>>>>>>>> + break;
>>>>>>>>> + }
>>>>>>>>> + }
>>>>>>>>> + }
>>>>>>>>> + return read == 0 && endOfInput ? -1 : read;
>>>>>>>>> + }
>>>>>>>>> +
>>>>>>>>> + /**
>>>>>>>>> + * Read the specified number of bytes into an array.
>>>>>>>>> + *
>>>>>>>>> + * @param b the byte array to read into
>>>>>>>>> + * @return the number of bytes read or <code>-1</code>
>>>>>>>>> + * if the end of the stream has been reached
>>>>>>>>> + * @throws IOException if an I/O error occurs
>>>>>>>>> + */
>>>>>>>>> + @Override
>>>>>>>>> + public int read(byte[] b) throws IOException {
>>>>>>>>> + return read(b, 0, b.length);
>>>>>>>>> + }
>>>>>>>>> +
>>>>>>>>> + /**
>>>>>>>>> + * Read a single byte.
>>>>>>>>> + *
>>>>>>>>> + * @return either the byte read or <code>-1</code> if the end of the stream
>>>>>>>>> + * has been reached
>>>>>>>>> + * @throws IOException if an I/O error occurs
>>>>>>>>> + */
>>>>>>>>> + @Override
>>>>>>>>> + public int read() throws IOException {
>>>>>>>>> + for (;;) {
>>>>>>>>> + if (encoderOut.hasRemaining()) {
>>>>>>>>> + return encoderOut.get() & 0xFF;
>>>>>>>>> + } else {
>>>>>>>>> + fillBuffer();
>>>>>>>>> + if (endOfInput && !encoderOut.hasRemaining()) {
>>>>>>>>> + return -1;
>>>>>>>>> + }
>>>>>>>>> + }
>>>>>>>>> + }
>>>>>>>>> + }
>>>>>>>>> +
>>>>>>>>> + /**
>>>>>>>>> + * Close the stream. This method will cause the underlying {@link Reader}
>>>>>>>>> + * to be closed.
>>>>>>>>> + * @throws IOException if an I/O error occurs
>>>>>>>>> + */
>>>>>>>>> + @Override
>>>>>>>>> + public void close() throws IOException {
>>>>>>>>> + reader.close();
>>>>>>>>> + }
>>>>>>>>> +}
>>>>>>>>> diff --git a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>>>>>> index f77b5ed..b957f36 100644
>>>>>>>>> --- a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>>>>>> +++ b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>>>>>> @@ -96,6 +96,25 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>>>>>> assertEquals("Get a wrong response code.", 200, client.getResponse().getStatus());
>>>>>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>>> }
>>>>>>>>> +
>>>>>>>>> + @Test
>>>>>>>>> + public void testGetBookFromWebClientWithTextJMSMessage() throws Exception {
>>>>>>>>> + // setup the the client
>>>>>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>>>>> + + "?replyToName=dynamicQueues/test.jmstransport.response"
>>>>>>>>> + + "&jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>>>>> + + "&messageType=text";
>>>>>>>>> +
>>>>>>>>> + WebClient client = WebClient.create(endpointAddressUrlEncoded);
>>>>>>>>> + WebClient.getConfig(client).getInInterceptors().add(new LoggingInInterceptor());
>>>>>>>>> + WebClient.getConfig(client).getRequestContext()
>>>>>>>>> + .put(org.apache.cxf.message.Message.REQUEST_URI, "/bookstore/books/123");
>>>>>>>>> +
>>>>>>>>> + Book book = client.get(Book.class);
>>>>>>>>> + assertEquals("Get a wrong response code.", 200, client.getResponse().getStatus());
>>>>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>>> + }
>>>>>>>>> @Test
>>>>>>>>> public void testPutBookOneWayWithWebClient() throws Exception {
>>>>>>>>> @@ -130,6 +149,7 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>>>>>> }
>>>>>>>>> }
>>>>>>>>> +
>>>>>>>>> @Test
>>>>>>>>> public void testGetBookFromWebClientWithPath() throws Exception {
>>>>>>>>> // setup the the client
>>>>>>>>> @@ -148,6 +168,25 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>>>>>> }
>>>>>>>>> @Test
>>>>>>>>> + public void testGetBookFromWebClientWithPathWithTextJMSMessage() throws Exception {
>>>>>>>>> + // setup the the client
>>>>>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>>>>> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>>>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>>>>>> + + "&messageType=text";
>>>>>>>>> +
>>>>>>>>> + WebClient client = WebClient.create(endpointAddressUrlEncoded);
>>>>>>>>> + client.path("bookstore").path("books").path("123");
>>>>>>>>> +
>>>>>>>>> + Book book = client.get(Book.class);
>>>>>>>>> + assertEquals("Get a wrong response code.", 200, client.getResponse().getStatus());
>>>>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>>> + }
>>>>>>>>> +
>>>>>>>>> +
>>>>>>>>> + @Test
>>>>>>>>> public void testGetBookFromProxyClient() throws Exception {
>>>>>>>>> // setup the the client
>>>>>>>>> String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>>>>> @@ -161,6 +200,22 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>>>>>> assertEquals("Get a wrong response code.", 200, WebClient.client(client).getResponse().getStatus());
>>>>>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>>> }
>>>>>>>>> +
>>>>>>>>> + @Test
>>>>>>>>> + public void testGetBookFromProxyClientWithTextJMSMessage() throws Exception {
>>>>>>>>> + // setup the the client
>>>>>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>>>>> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>>>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>>>>>> + + "&messageType=text";
>>>>>>>>> +
>>>>>>>>> + JMSBookStore client = JAXRSClientFactory.create(endpointAddressUrlEncoded, JMSBookStore.class);
>>>>>>>>> + Book book = client.getBook("123");
>>>>>>>>> + assertEquals("Get a wrong response code.", 200, WebClient.client(client).getResponse().getStatus());
>>>>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>>> + }
>>>>>>>>> @Test
>>>>>>>>> public void testGetBookFromSubresourceProxyClient() throws Exception {
>>>>>>>>> @@ -177,6 +232,23 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>>>>>> assertEquals("Get a wrong response code.", 200, WebClient.client(bookProxy).getResponse().getStatus());
>>>>>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>>> }
>>>>>>>>> +
>>>>>>>>> + @Test
>>>>>>>>> + public void testGetBookFromSubresourceProxyClientWithTextJMSMessage() throws Exception {
>>>>>>>>> + // setup the the client
>>>>>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>>>>> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>>>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>>>>>> + + "&messageType=text";
>>>>>>>>> +
>>>>>>>>> + JMSBookStore client = JAXRSClientFactory.create(endpointAddressUrlEncoded, JMSBookStore.class);
>>>>>>>>> + Book bookProxy = client.getBookSubResource("123");
>>>>>>>>> + Book book = bookProxy.retrieveState();
>>>>>>>>> + assertEquals("Get a wrong response code.", 200, WebClient.client(bookProxy).getResponse().getStatus());
>>>>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>>> + }
>>>>>>>>> @Test
>>>>>>>>> public void testGetBookFromProxyClientWithQuery() throws Exception {
>>>>>>>>> @@ -193,6 +265,24 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>>> }
>>>>>>>>> +
>>>>>>>>> + @Test
>>>>>>>>> + public void testGetBookFromProxyClientWithQueryWithTextJMSMessage() throws Exception {
>>>>>>>>> + // setup the the client
>>>>>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>>>>> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>>>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>>>>>> + + "&messageType=text";
>>>>>>>>> +
>>>>>>>>> + JMSBookStore client = JAXRSClientFactory.create(endpointAddressUrlEncoded, JMSBookStore.class);
>>>>>>>>> + Book book = client.getBookByURLQuery(new String[] {"1", "2", "3"});
>>>>>>>>> + assertEquals("Get a wrong response code.", 200, WebClient.client(client).getResponse().getStatus());
>>>>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>>> + }
>>>>>>>>> +
>>>>>>>>> +
>>>>>>>>> @Test
>>>>>>>>> public void testGetBook() throws Exception {
>>>>>>>>> Context ctx = getContext();
>>>>>>>>>
>>>>>>>
>>>>>>
>>>>
>>>>
>>>> --
>>>> Sergey Beryozkin
>>>>
>>>> Talend Community Coders
>>>> http://coders.talend.com/
>>>
Re: [cxf] branch master updated: [CXF-7532]REST on JMS transport
can't handle the request message with text messageType
Posted by Sergey Beryozkin <sb...@gmail.com>.
Some JAX-RS providers check Reader or XMLStreamReader if InputStream is
null.
But at the moment, Freeman, please do a bit of cleanup as agreed, and
then we can check if the use ReaderInputStream can be avoided at all.
Sergey
On 19/10/17 19:47, Daniel Kulp wrote:
> If using the TEXT stuff for JMS, it should remain as a Reader on the message if at all possible. Converting from Reader to InputStream is not a trivial operation and does enact a performance penalty. All of the SOAP pathways for JMS should be OK for use of a Reader so that we can avoid the entire String->UTF-8 bytes->chars round tripping.
>
> Dan
>
>
> On Oct 19, 2017, at 5:21 AM, ffang <fr...@gmail.com> wrote:
>>
>> I modified the JMS transport part and set ReaderInputStream instead of Reader, running all tests now, will see what I can get.
>>
>> If I run into any test failure, I will keep things as is, just clean up.
>> －－－－－－－－－－－－－
>> Freeman(Yue) Fang
>>
>> Red Hat, Inc.
>> FuseSource is now part of Red Hat
>>
>>
>>
>>> On Oct 19, 2017, at 5:16 PM, Sergey Beryozkin <sb...@gmail.com> wrote:
>>>
>>> If we assume that checking Reader can be useful in the general sense, then may be lets keep things as is, with the minor cleanup only...
>>>
>>> Sergey
>>> On 19/10/17 10:05, Sergey Beryozkin wrote:
>>>> Hi Freeman
>>>> Reader may likely need to be set too for SOAP, and I recall checking Reader somewhere else too (will double check now), but I thought may be the transport can set both Reader and InputStream, hmm, that is an extra instance though, may be lets start with moving ReaderInputStream to the core first ?
>>>> Sergey
>>>> On 19/10/17 09:59, ffang wrote:
>>>>> Yeah, set ReaderInputStream instead of Reader for the JMS transport is a better solution. If no objection I will go this way.
>>>>> －－－－－－－－－－－－－
>>>>> Freeman(Yue) Fang
>>>>>
>>>>> Red Hat, Inc.
>>>>> FuseSource is now part of Red Hat
>>>>>
>>>>>
>>>>>
>>>>>> On Oct 19, 2017, at 4:50 PM, Sergey Beryozkin <sberyozkin@gmail.com <ma...@gmail.com>> wrote:
>>>>>>
>>>>>> The other thing is, should JMS transport itself set ReaderInputStream as InputStream on the message, as opposed to patching the JAX-RS code in various places ?
>>>>>>
>>>>>> Sergey
>>>>>> On 19/10/17 09:44, Sergey Beryozkin wrote:
>>>>>>> Hi Freeman
>>>>>>> Can you move ReaderInputStream to the CXF core where CachedInputStream and other IO utility code is located ?
>>>>>>> Please also update the code which gets Reader.class and check it for null first, and also drop a comment re the JMS transport, checking the reader can make sense in all the cases...
>>>>>>> Thanks, Sergey
>>>>>>> On 19/10/17 04:16, ffang@apache.org <ma...@apache.org> wrote:
>>>>>>>> This is an automated email from the ASF dual-hosted git repository.
>>>>>>>>
>>>>>>>> ffang pushed a commit to branch master
>>>>>>>> in repository https://gitbox.apache.org/repos/asf/cxf.git
>>>>>>>>
>>>>>>>>
>>>>>>>> The following commit(s) were added to refs/heads/master by this push:
>>>>>>>> new 1d66475 [CXF-7532]REST on JMS transport can't handle the request message with text messageType
>>>>>>>> 1d66475 is described below
>>>>>>>>
>>>>>>>> commit 1d66475b4dea2a0031dcb1765e8e9fd08430ec9b
>>>>>>>> Author: Freeman Fang <freeman.fang@gmail.com <ma...@gmail.com>>
>>>>>>>> AuthorDate: Thu Oct 19 11:15:50 2017 +0800
>>>>>>>>
>>>>>>>> [CXF-7532]REST on JMS transport can't handle the request message with text messageType
>>>>>>>> ---
>>>>>>>> .../org/apache/cxf/jaxrs/impl/ResponseImpl.java | 5 +
>>>>>>>> .../org/apache/cxf/jaxrs/utils/JAXRSUtils.java | 6 +-
>>>>>>>> .../apache/cxf/jaxrs/utils/ReaderInputStream.java | 296 +++++++++++++++++++++
>>>>>>>> .../apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java | 90 +++++++
>>>>>>>> 4 files changed, 396 insertions(+), 1 deletion(-)
>>>>>>>>
>>>>>>>> diff --git a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>>>>> index e3a3acf..82f183a 100644
>>>>>>>> --- a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>>>>> +++ b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>>>>> @@ -58,6 +58,7 @@ import org.apache.cxf.jaxrs.provider.ProviderFactory;
>>>>>>>> import org.apache.cxf.jaxrs.utils.HttpUtils;
>>>>>>>> import org.apache.cxf.jaxrs.utils.InjectionUtils;
>>>>>>>> import org.apache.cxf.jaxrs.utils.JAXRSUtils;
>>>>>>>> +import org.apache.cxf.jaxrs.utils.ReaderInputStream;
>>>>>>>> import org.apache.cxf.message.Message;
>>>>>>>> import org.apache.cxf.message.MessageUtils;
>>>>>>>> @@ -342,6 +343,10 @@ public final class ResponseImpl extends Response {
>>>>>>>> entityStreamAvailable = entityStream != null;
>>>>>>>> } else if (entity instanceof InputStream) {
>>>>>>>> entityStream = InputStream.class.cast(entity);
>>>>>>>> + } else {
>>>>>>>> + Message inMessage = getResponseMessage();
>>>>>>>> + Reader reader = inMessage.getContent(Reader.class);
>>>>>>>> + entityStream = InputStream.class.cast(new ReaderInputStream(reader));
>>>>>>>> }
>>>>>>>> // we need to check for readers even if no IS is set - the readers may still do it
>>>>>>>> diff --git a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>>>>> index 0ef6565..509f76d 100644
>>>>>>>> --- a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>>>>> +++ b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>>>>> @@ -832,7 +832,11 @@ public final class JAXRSUtils {
>>>>>>>> OperationResourceInfo ori)
>>>>>>>> throws IOException, WebApplicationException {
>>>>>>>> InputStream is = message.getContent(InputStream.class);
>>>>>>>> -
>>>>>>>> + if (is == null) {
>>>>>>>> + //may use the jms transport so check the Reader;
>>>>>>>> + Reader reader = message.getContent(Reader.class);
>>>>>>>> + is = new ReaderInputStream(reader);
>>>>>>>> + }
>>>>>>>> if (parameter.getType() == ParameterType.REQUEST_BODY) {
>>>>>>>> if (parameterClass == AsyncResponse.class) {
>>>>>>>> diff --git a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
>>>>>>>> new file mode 100644
>>>>>>>> index 0000000..c7b142e
>>>>>>>> --- /dev/null
>>>>>>>> +++ b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
>>>>>>>> @@ -0,0 +1,296 @@
>>>>>>>> +/**
>>>>>>>> + * Licensed to the Apache Software Foundation (ASF) under one
>>>>>>>> + * or more contributor license agreements. See the NOTICE file
>>>>>>>> + * distributed with this work for additional information
>>>>>>>> + * regarding copyright ownership. The ASF licenses this file
>>>>>>>> + * to you 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.apache.cxf.jaxrs.utils;
>>>>>>>> +
>>>>>>>> +import java.io.IOException;
>>>>>>>> +import java.io.InputStream;
>>>>>>>> +import java.io.Reader;
>>>>>>>> +import java.nio.ByteBuffer;
>>>>>>>> +import java.nio.CharBuffer;
>>>>>>>> +import java.nio.charset.Charset;
>>>>>>>> +import java.nio.charset.CharsetEncoder;
>>>>>>>> +import java.nio.charset.CoderResult;
>>>>>>>> +import java.nio.charset.CodingErrorAction;
>>>>>>>> +
>>>>>>>> +/**
>>>>>>>> + * {@link InputStream} implementation that reads a character stream from a {@link Reader}
>>>>>>>> + * and transforms it to a byte stream using a specified charset encoding. The stream
>>>>>>>> + * is transformed using a {@link CharsetEncoder} object, guaranteeing that all charset
>>>>>>>> + * encodings supported by the JRE are handled correctly. In particular for charsets such as
>>>>>>>> + * UTF-16, the implementation ensures that one and only one byte order marker
>>>>>>>> + * is produced.
>>>>>>>> + * <p>
>>>>>>>> + * Since in general it is not possible to predict the number of characters to be read from the
>>>>>>>> + * {@link Reader} to satisfy a read request on the {@link ReaderInputStream}, all reads from
>>>>>>>> + * the {@link Reader} are buffered. There is therefore no well defined correlation
>>>>>>>> + * between the current position of the {@link Reader} and that of the {@link ReaderInputStream}.
>>>>>>>> + * This also implies that in general there is no need to wrap the underlying {@link Reader}
>>>>>>>> + * in a {@link java.io.BufferedReader}.
>>>>>>>> + * <p>
>>>>>>>> + * {@link ReaderInputStream} implements the inverse transformation of {@link java.io.InputStreamReader};
>>>>>>>> + * in the following example, reading from <tt>in2</tt> would return the same byte
>>>>>>>> + * sequence as reading from <tt>in</tt> (provided that the initial byte sequence is legal
>>>>>>>> + * with respect to the charset encoding):
>>>>>>>> + * <pre>
>>>>>>>> + * InputStream in = ...
>>>>>>>> + * Charset cs = ...
>>>>>>>> + * InputStreamReader reader = new InputStreamReader(in, cs);
>>>>>>>> + * ReaderInputStream in2 = new ReaderInputStream(reader, cs);</pre>
>>>>>>>> + * {@link ReaderInputStream} implements the same transformation as {@link java.io.OutputStreamWriter},
>>>>>>>> + * except that the control flow is reversed: both classes transform a character stream
>>>>>>>> + * into a byte stream, but {@link java.io.OutputStreamWriter} pushes data to the underlying stream,
>>>>>>>> + * while {@link ReaderInputStream} pulls it from the underlying stream.
>>>>>>>> + * <p>
>>>>>>>> + * Note that while there are use cases where there is no alternative to using
>>>>>>>> + * this class, very often the need to use this class is an indication of a flaw
>>>>>>>> + * in the design of the code. This class is typically used in situations where an existing
>>>>>>>> + * API only accepts an {@link InputStream}, but where the most natural way to produce the data
>>>>>>>> + * is as a character stream, i.e. by providing a {@link Reader} instance. An example of a situation
>>>>>>>> + * where this problem may appear is when implementing the {@link javax.activation.DataSource}
>>>>>>>> + * interface from the Java Activation Framework.
>>>>>>>> + * <p>
>>>>>>>> + * Given the fact that the {@link Reader} class doesn't provide any way to predict whether the next
>>>>>>>> + * read operation will block or not, it is not possible to provide a meaningful
>>>>>>>> + * implementation of the {@link InputStream#available()} method. A call to this method
>>>>>>>> + * will always return 0. Also, this class doesn't support {@link InputStream#mark(int)}.
>>>>>>>> + * <p>
>>>>>>>> + * Instances of {@link ReaderInputStream} are not thread safe.
>>>>>>>> + *
>>>>>>>> + * @see org.apache.commons.io.output.WriterOutputStream
>>>>>>>> + *
>>>>>>>> + * @since 2.0
>>>>>>>> + */
>>>>>>>> +public class ReaderInputStream extends InputStream {
>>>>>>>> + private static final int DEFAULT_BUFFER_SIZE = 1024;
>>>>>>>> +
>>>>>>>> + private final Reader reader;
>>>>>>>> + private final CharsetEncoder encoder;
>>>>>>>> +
>>>>>>>> + /**
>>>>>>>> + * CharBuffer used as input for the decoder. It should be reasonably
>>>>>>>> + * large as we read data from the underlying Reader into this buffer.
>>>>>>>> + */
>>>>>>>> + private final CharBuffer encoderIn;
>>>>>>>> +
>>>>>>>> + /**
>>>>>>>> + * ByteBuffer used as output for the decoder. This buffer can be small
>>>>>>>> + * as it is only used to transfer data from the decoder to the
>>>>>>>> + * buffer provided by the caller.
>>>>>>>> + */
>>>>>>>> + private final ByteBuffer encoderOut;
>>>>>>>> +
>>>>>>>> + private CoderResult lastCoderResult;
>>>>>>>> + private boolean endOfInput;
>>>>>>>> +
>>>>>>>> + /**
>>>>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>>>>> + *
>>>>>>>> + * @param reader the target {@link Reader}
>>>>>>>> + * @param encoder the charset encoder
>>>>>>>> + * @since 2.1
>>>>>>>> + */
>>>>>>>> + public ReaderInputStream(Reader reader, CharsetEncoder encoder) {
>>>>>>>> + this(reader, encoder, DEFAULT_BUFFER_SIZE);
>>>>>>>> + }
>>>>>>>> +
>>>>>>>> + /**
>>>>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>>>>> + *
>>>>>>>> + * @param reader the target {@link Reader}
>>>>>>>> + * @param encoder the charset encoder
>>>>>>>> + * @param bufferSize the size of the input buffer in number of characters
>>>>>>>> + * @since 2.1
>>>>>>>> + */
>>>>>>>> + public ReaderInputStream(Reader reader, CharsetEncoder encoder, int bufferSize) {
>>>>>>>> + this.reader = reader;
>>>>>>>> + this.encoder = encoder;
>>>>>>>> + this.encoderIn = CharBuffer.allocate(bufferSize);
>>>>>>>> + this.encoderIn.flip();
>>>>>>>> + this.encoderOut = ByteBuffer.allocate(128);
>>>>>>>> + this.encoderOut.flip();
>>>>>>>> + }
>>>>>>>> +
>>>>>>>> + /**
>>>>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>>>>> + *
>>>>>>>> + * @param reader the target {@link Reader}
>>>>>>>> + * @param charset the charset encoding
>>>>>>>> + * @param bufferSize the size of the input buffer in number of characters
>>>>>>>> + */
>>>>>>>> + public ReaderInputStream(Reader reader, Charset charset, int bufferSize) {
>>>>>>>> + this(reader,
>>>>>>>> + charset.newEncoder()
>>>>>>>> + .onMalformedInput(CodingErrorAction.REPLACE)
>>>>>>>> + .onUnmappableCharacter(CodingErrorAction.REPLACE),
>>>>>>>> + bufferSize);
>>>>>>>> + }
>>>>>>>> +
>>>>>>>> + /**
>>>>>>>> + * Construct a new {@link ReaderInputStream} with a default input buffer size of
>>>>>>>> + * 1024 characters.
>>>>>>>> + *
>>>>>>>> + * @param reader the target {@link Reader}
>>>>>>>> + * @param charset the charset encoding
>>>>>>>> + */
>>>>>>>> + public ReaderInputStream(Reader reader, Charset charset) {
>>>>>>>> + this(reader, charset, DEFAULT_BUFFER_SIZE);
>>>>>>>> + }
>>>>>>>> +
>>>>>>>> + /**
>>>>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>>>>> + *
>>>>>>>> + * @param reader the target {@link Reader}
>>>>>>>> + * @param charsetName the name of the charset encoding
>>>>>>>> + * @param bufferSize the size of the input buffer in number of characters
>>>>>>>> + */
>>>>>>>> + public ReaderInputStream(Reader reader, String charsetName, int bufferSize) {
>>>>>>>> + this(reader, Charset.forName(charsetName), bufferSize);
>>>>>>>> + }
>>>>>>>> +
>>>>>>>> + /**
>>>>>>>> + * Construct a new {@link ReaderInputStream} with a default input buffer size of
>>>>>>>> + * 1024 characters.
>>>>>>>> + *
>>>>>>>> + * @param reader the target {@link Reader}
>>>>>>>> + * @param charsetName the name of the charset encoding
>>>>>>>> + */
>>>>>>>> + public ReaderInputStream(Reader reader, String charsetName) {
>>>>>>>> + this(reader, charsetName, DEFAULT_BUFFER_SIZE);
>>>>>>>> + }
>>>>>>>> +
>>>>>>>> + /**
>>>>>>>> + * Construct a new {@link ReaderInputStream} that uses the default character encoding
>>>>>>>> + * with a default input buffer size of 1024 characters.
>>>>>>>> + *
>>>>>>>> + * @param reader the target {@link Reader}
>>>>>>>> + */
>>>>>>>> + public ReaderInputStream(Reader reader) {
>>>>>>>> + this(reader, Charset.defaultCharset());
>>>>>>>> + }
>>>>>>>> +
>>>>>>>> + /**
>>>>>>>> + * Fills the internal char buffer from the reader.
>>>>>>>> + *
>>>>>>>> + * @throws IOException
>>>>>>>> + * If an I/O error occurs
>>>>>>>> + */
>>>>>>>> + private void fillBuffer() throws IOException {
>>>>>>>> + if (!endOfInput && (lastCoderResult == null || lastCoderResult.isUnderflow())) {
>>>>>>>> + encoderIn.compact();
>>>>>>>> + int position = encoderIn.position();
>>>>>>>> + // We don't use Reader#read(CharBuffer) here because it is more efficient
>>>>>>>> + // to write directly to the underlying char array (the default implementation
>>>>>>>> + // copies data to a temporary char array).
>>>>>>>> + int c = reader.read(encoderIn.array(), position, encoderIn.remaining());
>>>>>>>> + if (c == -1) {
>>>>>>>> + endOfInput = true;
>>>>>>>> + } else {
>>>>>>>> + encoderIn.position(position + c);
>>>>>>>> + }
>>>>>>>> + encoderIn.flip();
>>>>>>>> + }
>>>>>>>> + encoderOut.compact();
>>>>>>>> + lastCoderResult = encoder.encode(encoderIn, encoderOut, endOfInput);
>>>>>>>> + encoderOut.flip();
>>>>>>>> + }
>>>>>>>> +
>>>>>>>> + /**
>>>>>>>> + * Read the specified number of bytes into an array.
>>>>>>>> + *
>>>>>>>> + * @param b the byte array to read into
>>>>>>>> + * @param off the offset to start reading bytes into
>>>>>>>> + * @param len the number of bytes to read
>>>>>>>> + * @return the number of bytes read or <code>-1</code>
>>>>>>>> + * if the end of the stream has been reached
>>>>>>>> + * @throws IOException if an I/O error occurs
>>>>>>>> + */
>>>>>>>> + @Override
>>>>>>>> + public int read(byte[] b, int off, int len) throws IOException {
>>>>>>>> + if (b == null) {
>>>>>>>> + throw new NullPointerException("Byte array must not be null");
>>>>>>>> + }
>>>>>>>> + if (len < 0 || off < 0 || (off + len) > b.length) {
>>>>>>>> + throw new IndexOutOfBoundsException("Array Size=" + b.length
>>>>>>>> + + ", offset=" + off + ", length=" + len);
>>>>>>>> + }
>>>>>>>> + int read = 0;
>>>>>>>> + if (len == 0) {
>>>>>>>> + return 0; // Always return 0 if len == 0
>>>>>>>> + }
>>>>>>>> + while (len > 0) {
>>>>>>>> + if (encoderOut.hasRemaining()) {
>>>>>>>> + int c = Math.min(encoderOut.remaining(), len);
>>>>>>>> + encoderOut.get(b, off, c);
>>>>>>>> + off += c;
>>>>>>>> + len -= c;
>>>>>>>> + read += c;
>>>>>>>> + } else {
>>>>>>>> + fillBuffer();
>>>>>>>> + if (endOfInput && !encoderOut.hasRemaining()) {
>>>>>>>> + break;
>>>>>>>> + }
>>>>>>>> + }
>>>>>>>> + }
>>>>>>>> + return read == 0 && endOfInput ? -1 : read;
>>>>>>>> + }
>>>>>>>> +
>>>>>>>> + /**
>>>>>>>> + * Read the specified number of bytes into an array.
>>>>>>>> + *
>>>>>>>> + * @param b the byte array to read into
>>>>>>>> + * @return the number of bytes read or <code>-1</code>
>>>>>>>> + * if the end of the stream has been reached
>>>>>>>> + * @throws IOException if an I/O error occurs
>>>>>>>> + */
>>>>>>>> + @Override
>>>>>>>> + public int read(byte[] b) throws IOException {
>>>>>>>> + return read(b, 0, b.length);
>>>>>>>> + }
>>>>>>>> +
>>>>>>>> + /**
>>>>>>>> + * Read a single byte.
>>>>>>>> + *
>>>>>>>> + * @return either the byte read or <code>-1</code> if the end of the stream
>>>>>>>> + * has been reached
>>>>>>>> + * @throws IOException if an I/O error occurs
>>>>>>>> + */
>>>>>>>> + @Override
>>>>>>>> + public int read() throws IOException {
>>>>>>>> + for (;;) {
>>>>>>>> + if (encoderOut.hasRemaining()) {
>>>>>>>> + return encoderOut.get() & 0xFF;
>>>>>>>> + } else {
>>>>>>>> + fillBuffer();
>>>>>>>> + if (endOfInput && !encoderOut.hasRemaining()) {
>>>>>>>> + return -1;
>>>>>>>> + }
>>>>>>>> + }
>>>>>>>> + }
>>>>>>>> + }
>>>>>>>> +
>>>>>>>> + /**
>>>>>>>> + * Close the stream. This method will cause the underlying {@link Reader}
>>>>>>>> + * to be closed.
>>>>>>>> + * @throws IOException if an I/O error occurs
>>>>>>>> + */
>>>>>>>> + @Override
>>>>>>>> + public void close() throws IOException {
>>>>>>>> + reader.close();
>>>>>>>> + }
>>>>>>>> +}
>>>>>>>> diff --git a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>>>>> index f77b5ed..b957f36 100644
>>>>>>>> --- a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>>>>> +++ b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>>>>> @@ -96,6 +96,25 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>>>>> assertEquals("Get a wrong response code.", 200, client.getResponse().getStatus());
>>>>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>> }
>>>>>>>> +
>>>>>>>> + @Test
>>>>>>>> + public void testGetBookFromWebClientWithTextJMSMessage() throws Exception {
>>>>>>>> + // setup the the client
>>>>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>>>> + + "?replyToName=dynamicQueues/test.jmstransport.response"
>>>>>>>> + + "&jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>>>> + + "&messageType=text";
>>>>>>>> +
>>>>>>>> + WebClient client = WebClient.create(endpointAddressUrlEncoded);
>>>>>>>> + WebClient.getConfig(client).getInInterceptors().add(new LoggingInInterceptor());
>>>>>>>> + WebClient.getConfig(client).getRequestContext()
>>>>>>>> + .put(org.apache.cxf.message.Message.REQUEST_URI, "/bookstore/books/123");
>>>>>>>> +
>>>>>>>> + Book book = client.get(Book.class);
>>>>>>>> + assertEquals("Get a wrong response code.", 200, client.getResponse().getStatus());
>>>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>> + }
>>>>>>>> @Test
>>>>>>>> public void testPutBookOneWayWithWebClient() throws Exception {
>>>>>>>> @@ -130,6 +149,7 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>>>>> }
>>>>>>>> }
>>>>>>>> +
>>>>>>>> @Test
>>>>>>>> public void testGetBookFromWebClientWithPath() throws Exception {
>>>>>>>> // setup the the client
>>>>>>>> @@ -148,6 +168,25 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>>>>> }
>>>>>>>> @Test
>>>>>>>> + public void testGetBookFromWebClientWithPathWithTextJMSMessage() throws Exception {
>>>>>>>> + // setup the the client
>>>>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>>>> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>>>>> + + "&messageType=text";
>>>>>>>> +
>>>>>>>> + WebClient client = WebClient.create(endpointAddressUrlEncoded);
>>>>>>>> + client.path("bookstore").path("books").path("123");
>>>>>>>> +
>>>>>>>> + Book book = client.get(Book.class);
>>>>>>>> + assertEquals("Get a wrong response code.", 200, client.getResponse().getStatus());
>>>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>> + }
>>>>>>>> +
>>>>>>>> +
>>>>>>>> + @Test
>>>>>>>> public void testGetBookFromProxyClient() throws Exception {
>>>>>>>> // setup the the client
>>>>>>>> String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>>>> @@ -161,6 +200,22 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>>>>> assertEquals("Get a wrong response code.", 200, WebClient.client(client).getResponse().getStatus());
>>>>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>> }
>>>>>>>> +
>>>>>>>> + @Test
>>>>>>>> + public void testGetBookFromProxyClientWithTextJMSMessage() throws Exception {
>>>>>>>> + // setup the the client
>>>>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>>>> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>>>>> + + "&messageType=text";
>>>>>>>> +
>>>>>>>> + JMSBookStore client = JAXRSClientFactory.create(endpointAddressUrlEncoded, JMSBookStore.class);
>>>>>>>> + Book book = client.getBook("123");
>>>>>>>> + assertEquals("Get a wrong response code.", 200, WebClient.client(client).getResponse().getStatus());
>>>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>> + }
>>>>>>>> @Test
>>>>>>>> public void testGetBookFromSubresourceProxyClient() throws Exception {
>>>>>>>> @@ -177,6 +232,23 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>>>>> assertEquals("Get a wrong response code.", 200, WebClient.client(bookProxy).getResponse().getStatus());
>>>>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>> }
>>>>>>>> +
>>>>>>>> + @Test
>>>>>>>> + public void testGetBookFromSubresourceProxyClientWithTextJMSMessage() throws Exception {
>>>>>>>> + // setup the the client
>>>>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>>>> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>>>>> + + "&messageType=text";
>>>>>>>> +
>>>>>>>> + JMSBookStore client = JAXRSClientFactory.create(endpointAddressUrlEncoded, JMSBookStore.class);
>>>>>>>> + Book bookProxy = client.getBookSubResource("123");
>>>>>>>> + Book book = bookProxy.retrieveState();
>>>>>>>> + assertEquals("Get a wrong response code.", 200, WebClient.client(bookProxy).getResponse().getStatus());
>>>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>> + }
>>>>>>>> @Test
>>>>>>>> public void testGetBookFromProxyClientWithQuery() throws Exception {
>>>>>>>> @@ -193,6 +265,24 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>> }
>>>>>>>> +
>>>>>>>> + @Test
>>>>>>>> + public void testGetBookFromProxyClientWithQueryWithTextJMSMessage() throws Exception {
>>>>>>>> + // setup the the client
>>>>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>>>> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>>>>> + + "&messageType=text";
>>>>>>>> +
>>>>>>>> + JMSBookStore client = JAXRSClientFactory.create(endpointAddressUrlEncoded, JMSBookStore.class);
>>>>>>>> + Book book = client.getBookByURLQuery(new String[] {"1", "2", "3"});
>>>>>>>> + assertEquals("Get a wrong response code.", 200, WebClient.client(client).getResponse().getStatus());
>>>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>>> + }
>>>>>>>> +
>>>>>>>> +
>>>>>>>> @Test
>>>>>>>> public void testGetBook() throws Exception {
>>>>>>>> Context ctx = getContext();
>>>>>>>>
>>>>>>
>>>>>
>>>
>>>
>>> --
>>> Sergey Beryozkin
>>>
>>> Talend Community Coders
>>> http://coders.talend.com/
>>
>
Re: [cxf] branch master updated: [CXF-7532]REST on JMS transport
can't handle the request message with text messageType
Posted by Daniel Kulp <dk...@apache.org>.
If using the TEXT stuff for JMS, it should remain as a Reader on the message if at all possible. Converting from Reader to InputStream is not a trivial operation and does enact a performance penalty. All of the SOAP pathways for JMS should be OK for use of a Reader so that we can avoid the entire String->UTF-8 bytes->chars round tripping.
Dan
On Oct 19, 2017, at 5:21 AM, ffang <fr...@gmail.com> wrote:
>
> I modified the JMS transport part and set ReaderInputStream instead of Reader, running all tests now, will see what I can get.
>
> If I run into any test failure, I will keep things as is, just clean up.
> －－－－－－－－－－－－－
> Freeman(Yue) Fang
>
> Red Hat, Inc.
> FuseSource is now part of Red Hat
>
>
>
>> On Oct 19, 2017, at 5:16 PM, Sergey Beryozkin <sb...@gmail.com> wrote:
>>
>> If we assume that checking Reader can be useful in the general sense, then may be lets keep things as is, with the minor cleanup only...
>>
>> Sergey
>> On 19/10/17 10:05, Sergey Beryozkin wrote:
>>> Hi Freeman
>>> Reader may likely need to be set too for SOAP, and I recall checking Reader somewhere else too (will double check now), but I thought may be the transport can set both Reader and InputStream, hmm, that is an extra instance though, may be lets start with moving ReaderInputStream to the core first ?
>>> Sergey
>>> On 19/10/17 09:59, ffang wrote:
>>>> Yeah, set ReaderInputStream instead of Reader for the JMS transport is a better solution. If no objection I will go this way.
>>>> －－－－－－－－－－－－－
>>>> Freeman(Yue) Fang
>>>>
>>>> Red Hat, Inc.
>>>> FuseSource is now part of Red Hat
>>>>
>>>>
>>>>
>>>>> On Oct 19, 2017, at 4:50 PM, Sergey Beryozkin <sberyozkin@gmail.com <ma...@gmail.com>> wrote:
>>>>>
>>>>> The other thing is, should JMS transport itself set ReaderInputStream as InputStream on the message, as opposed to patching the JAX-RS code in various places ?
>>>>>
>>>>> Sergey
>>>>> On 19/10/17 09:44, Sergey Beryozkin wrote:
>>>>>> Hi Freeman
>>>>>> Can you move ReaderInputStream to the CXF core where CachedInputStream and other IO utility code is located ?
>>>>>> Please also update the code which gets Reader.class and check it for null first, and also drop a comment re the JMS transport, checking the reader can make sense in all the cases...
>>>>>> Thanks, Sergey
>>>>>> On 19/10/17 04:16, ffang@apache.org <ma...@apache.org> wrote:
>>>>>>> This is an automated email from the ASF dual-hosted git repository.
>>>>>>>
>>>>>>> ffang pushed a commit to branch master
>>>>>>> in repository https://gitbox.apache.org/repos/asf/cxf.git
>>>>>>>
>>>>>>>
>>>>>>> The following commit(s) were added to refs/heads/master by this push:
>>>>>>> new 1d66475 [CXF-7532]REST on JMS transport can't handle the request message with text messageType
>>>>>>> 1d66475 is described below
>>>>>>>
>>>>>>> commit 1d66475b4dea2a0031dcb1765e8e9fd08430ec9b
>>>>>>> Author: Freeman Fang <freeman.fang@gmail.com <ma...@gmail.com>>
>>>>>>> AuthorDate: Thu Oct 19 11:15:50 2017 +0800
>>>>>>>
>>>>>>> [CXF-7532]REST on JMS transport can't handle the request message with text messageType
>>>>>>> ---
>>>>>>> .../org/apache/cxf/jaxrs/impl/ResponseImpl.java | 5 +
>>>>>>> .../org/apache/cxf/jaxrs/utils/JAXRSUtils.java | 6 +-
>>>>>>> .../apache/cxf/jaxrs/utils/ReaderInputStream.java | 296 +++++++++++++++++++++
>>>>>>> .../apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java | 90 +++++++
>>>>>>> 4 files changed, 396 insertions(+), 1 deletion(-)
>>>>>>>
>>>>>>> diff --git a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>>>> index e3a3acf..82f183a 100644
>>>>>>> --- a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>>>> +++ b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>>>> @@ -58,6 +58,7 @@ import org.apache.cxf.jaxrs.provider.ProviderFactory;
>>>>>>> import org.apache.cxf.jaxrs.utils.HttpUtils;
>>>>>>> import org.apache.cxf.jaxrs.utils.InjectionUtils;
>>>>>>> import org.apache.cxf.jaxrs.utils.JAXRSUtils;
>>>>>>> +import org.apache.cxf.jaxrs.utils.ReaderInputStream;
>>>>>>> import org.apache.cxf.message.Message;
>>>>>>> import org.apache.cxf.message.MessageUtils;
>>>>>>> @@ -342,6 +343,10 @@ public final class ResponseImpl extends Response {
>>>>>>> entityStreamAvailable = entityStream != null;
>>>>>>> } else if (entity instanceof InputStream) {
>>>>>>> entityStream = InputStream.class.cast(entity);
>>>>>>> + } else {
>>>>>>> + Message inMessage = getResponseMessage();
>>>>>>> + Reader reader = inMessage.getContent(Reader.class);
>>>>>>> + entityStream = InputStream.class.cast(new ReaderInputStream(reader));
>>>>>>> }
>>>>>>> // we need to check for readers even if no IS is set - the readers may still do it
>>>>>>> diff --git a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>>>> index 0ef6565..509f76d 100644
>>>>>>> --- a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>>>> +++ b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>>>> @@ -832,7 +832,11 @@ public final class JAXRSUtils {
>>>>>>> OperationResourceInfo ori)
>>>>>>> throws IOException, WebApplicationException {
>>>>>>> InputStream is = message.getContent(InputStream.class);
>>>>>>> -
>>>>>>> + if (is == null) {
>>>>>>> + //may use the jms transport so check the Reader;
>>>>>>> + Reader reader = message.getContent(Reader.class);
>>>>>>> + is = new ReaderInputStream(reader);
>>>>>>> + }
>>>>>>> if (parameter.getType() == ParameterType.REQUEST_BODY) {
>>>>>>> if (parameterClass == AsyncResponse.class) {
>>>>>>> diff --git a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
>>>>>>> new file mode 100644
>>>>>>> index 0000000..c7b142e
>>>>>>> --- /dev/null
>>>>>>> +++ b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
>>>>>>> @@ -0,0 +1,296 @@
>>>>>>> +/**
>>>>>>> + * Licensed to the Apache Software Foundation (ASF) under one
>>>>>>> + * or more contributor license agreements. See the NOTICE file
>>>>>>> + * distributed with this work for additional information
>>>>>>> + * regarding copyright ownership. The ASF licenses this file
>>>>>>> + * to you 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.apache.cxf.jaxrs.utils;
>>>>>>> +
>>>>>>> +import java.io.IOException;
>>>>>>> +import java.io.InputStream;
>>>>>>> +import java.io.Reader;
>>>>>>> +import java.nio.ByteBuffer;
>>>>>>> +import java.nio.CharBuffer;
>>>>>>> +import java.nio.charset.Charset;
>>>>>>> +import java.nio.charset.CharsetEncoder;
>>>>>>> +import java.nio.charset.CoderResult;
>>>>>>> +import java.nio.charset.CodingErrorAction;
>>>>>>> +
>>>>>>> +/**
>>>>>>> + * {@link InputStream} implementation that reads a character stream from a {@link Reader}
>>>>>>> + * and transforms it to a byte stream using a specified charset encoding. The stream
>>>>>>> + * is transformed using a {@link CharsetEncoder} object, guaranteeing that all charset
>>>>>>> + * encodings supported by the JRE are handled correctly. In particular for charsets such as
>>>>>>> + * UTF-16, the implementation ensures that one and only one byte order marker
>>>>>>> + * is produced.
>>>>>>> + * <p>
>>>>>>> + * Since in general it is not possible to predict the number of characters to be read from the
>>>>>>> + * {@link Reader} to satisfy a read request on the {@link ReaderInputStream}, all reads from
>>>>>>> + * the {@link Reader} are buffered. There is therefore no well defined correlation
>>>>>>> + * between the current position of the {@link Reader} and that of the {@link ReaderInputStream}.
>>>>>>> + * This also implies that in general there is no need to wrap the underlying {@link Reader}
>>>>>>> + * in a {@link java.io.BufferedReader}.
>>>>>>> + * <p>
>>>>>>> + * {@link ReaderInputStream} implements the inverse transformation of {@link java.io.InputStreamReader};
>>>>>>> + * in the following example, reading from <tt>in2</tt> would return the same byte
>>>>>>> + * sequence as reading from <tt>in</tt> (provided that the initial byte sequence is legal
>>>>>>> + * with respect to the charset encoding):
>>>>>>> + * <pre>
>>>>>>> + * InputStream in = ...
>>>>>>> + * Charset cs = ...
>>>>>>> + * InputStreamReader reader = new InputStreamReader(in, cs);
>>>>>>> + * ReaderInputStream in2 = new ReaderInputStream(reader, cs);</pre>
>>>>>>> + * {@link ReaderInputStream} implements the same transformation as {@link java.io.OutputStreamWriter},
>>>>>>> + * except that the control flow is reversed: both classes transform a character stream
>>>>>>> + * into a byte stream, but {@link java.io.OutputStreamWriter} pushes data to the underlying stream,
>>>>>>> + * while {@link ReaderInputStream} pulls it from the underlying stream.
>>>>>>> + * <p>
>>>>>>> + * Note that while there are use cases where there is no alternative to using
>>>>>>> + * this class, very often the need to use this class is an indication of a flaw
>>>>>>> + * in the design of the code. This class is typically used in situations where an existing
>>>>>>> + * API only accepts an {@link InputStream}, but where the most natural way to produce the data
>>>>>>> + * is as a character stream, i.e. by providing a {@link Reader} instance. An example of a situation
>>>>>>> + * where this problem may appear is when implementing the {@link javax.activation.DataSource}
>>>>>>> + * interface from the Java Activation Framework.
>>>>>>> + * <p>
>>>>>>> + * Given the fact that the {@link Reader} class doesn't provide any way to predict whether the next
>>>>>>> + * read operation will block or not, it is not possible to provide a meaningful
>>>>>>> + * implementation of the {@link InputStream#available()} method. A call to this method
>>>>>>> + * will always return 0. Also, this class doesn't support {@link InputStream#mark(int)}.
>>>>>>> + * <p>
>>>>>>> + * Instances of {@link ReaderInputStream} are not thread safe.
>>>>>>> + *
>>>>>>> + * @see org.apache.commons.io.output.WriterOutputStream
>>>>>>> + *
>>>>>>> + * @since 2.0
>>>>>>> + */
>>>>>>> +public class ReaderInputStream extends InputStream {
>>>>>>> + private static final int DEFAULT_BUFFER_SIZE = 1024;
>>>>>>> +
>>>>>>> + private final Reader reader;
>>>>>>> + private final CharsetEncoder encoder;
>>>>>>> +
>>>>>>> + /**
>>>>>>> + * CharBuffer used as input for the decoder. It should be reasonably
>>>>>>> + * large as we read data from the underlying Reader into this buffer.
>>>>>>> + */
>>>>>>> + private final CharBuffer encoderIn;
>>>>>>> +
>>>>>>> + /**
>>>>>>> + * ByteBuffer used as output for the decoder. This buffer can be small
>>>>>>> + * as it is only used to transfer data from the decoder to the
>>>>>>> + * buffer provided by the caller.
>>>>>>> + */
>>>>>>> + private final ByteBuffer encoderOut;
>>>>>>> +
>>>>>>> + private CoderResult lastCoderResult;
>>>>>>> + private boolean endOfInput;
>>>>>>> +
>>>>>>> + /**
>>>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>>>> + *
>>>>>>> + * @param reader the target {@link Reader}
>>>>>>> + * @param encoder the charset encoder
>>>>>>> + * @since 2.1
>>>>>>> + */
>>>>>>> + public ReaderInputStream(Reader reader, CharsetEncoder encoder) {
>>>>>>> + this(reader, encoder, DEFAULT_BUFFER_SIZE);
>>>>>>> + }
>>>>>>> +
>>>>>>> + /**
>>>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>>>> + *
>>>>>>> + * @param reader the target {@link Reader}
>>>>>>> + * @param encoder the charset encoder
>>>>>>> + * @param bufferSize the size of the input buffer in number of characters
>>>>>>> + * @since 2.1
>>>>>>> + */
>>>>>>> + public ReaderInputStream(Reader reader, CharsetEncoder encoder, int bufferSize) {
>>>>>>> + this.reader = reader;
>>>>>>> + this.encoder = encoder;
>>>>>>> + this.encoderIn = CharBuffer.allocate(bufferSize);
>>>>>>> + this.encoderIn.flip();
>>>>>>> + this.encoderOut = ByteBuffer.allocate(128);
>>>>>>> + this.encoderOut.flip();
>>>>>>> + }
>>>>>>> +
>>>>>>> + /**
>>>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>>>> + *
>>>>>>> + * @param reader the target {@link Reader}
>>>>>>> + * @param charset the charset encoding
>>>>>>> + * @param bufferSize the size of the input buffer in number of characters
>>>>>>> + */
>>>>>>> + public ReaderInputStream(Reader reader, Charset charset, int bufferSize) {
>>>>>>> + this(reader,
>>>>>>> + charset.newEncoder()
>>>>>>> + .onMalformedInput(CodingErrorAction.REPLACE)
>>>>>>> + .onUnmappableCharacter(CodingErrorAction.REPLACE),
>>>>>>> + bufferSize);
>>>>>>> + }
>>>>>>> +
>>>>>>> + /**
>>>>>>> + * Construct a new {@link ReaderInputStream} with a default input buffer size of
>>>>>>> + * 1024 characters.
>>>>>>> + *
>>>>>>> + * @param reader the target {@link Reader}
>>>>>>> + * @param charset the charset encoding
>>>>>>> + */
>>>>>>> + public ReaderInputStream(Reader reader, Charset charset) {
>>>>>>> + this(reader, charset, DEFAULT_BUFFER_SIZE);
>>>>>>> + }
>>>>>>> +
>>>>>>> + /**
>>>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>>>> + *
>>>>>>> + * @param reader the target {@link Reader}
>>>>>>> + * @param charsetName the name of the charset encoding
>>>>>>> + * @param bufferSize the size of the input buffer in number of characters
>>>>>>> + */
>>>>>>> + public ReaderInputStream(Reader reader, String charsetName, int bufferSize) {
>>>>>>> + this(reader, Charset.forName(charsetName), bufferSize);
>>>>>>> + }
>>>>>>> +
>>>>>>> + /**
>>>>>>> + * Construct a new {@link ReaderInputStream} with a default input buffer size of
>>>>>>> + * 1024 characters.
>>>>>>> + *
>>>>>>> + * @param reader the target {@link Reader}
>>>>>>> + * @param charsetName the name of the charset encoding
>>>>>>> + */
>>>>>>> + public ReaderInputStream(Reader reader, String charsetName) {
>>>>>>> + this(reader, charsetName, DEFAULT_BUFFER_SIZE);
>>>>>>> + }
>>>>>>> +
>>>>>>> + /**
>>>>>>> + * Construct a new {@link ReaderInputStream} that uses the default character encoding
>>>>>>> + * with a default input buffer size of 1024 characters.
>>>>>>> + *
>>>>>>> + * @param reader the target {@link Reader}
>>>>>>> + */
>>>>>>> + public ReaderInputStream(Reader reader) {
>>>>>>> + this(reader, Charset.defaultCharset());
>>>>>>> + }
>>>>>>> +
>>>>>>> + /**
>>>>>>> + * Fills the internal char buffer from the reader.
>>>>>>> + *
>>>>>>> + * @throws IOException
>>>>>>> + * If an I/O error occurs
>>>>>>> + */
>>>>>>> + private void fillBuffer() throws IOException {
>>>>>>> + if (!endOfInput && (lastCoderResult == null || lastCoderResult.isUnderflow())) {
>>>>>>> + encoderIn.compact();
>>>>>>> + int position = encoderIn.position();
>>>>>>> + // We don't use Reader#read(CharBuffer) here because it is more efficient
>>>>>>> + // to write directly to the underlying char array (the default implementation
>>>>>>> + // copies data to a temporary char array).
>>>>>>> + int c = reader.read(encoderIn.array(), position, encoderIn.remaining());
>>>>>>> + if (c == -1) {
>>>>>>> + endOfInput = true;
>>>>>>> + } else {
>>>>>>> + encoderIn.position(position + c);
>>>>>>> + }
>>>>>>> + encoderIn.flip();
>>>>>>> + }
>>>>>>> + encoderOut.compact();
>>>>>>> + lastCoderResult = encoder.encode(encoderIn, encoderOut, endOfInput);
>>>>>>> + encoderOut.flip();
>>>>>>> + }
>>>>>>> +
>>>>>>> + /**
>>>>>>> + * Read the specified number of bytes into an array.
>>>>>>> + *
>>>>>>> + * @param b the byte array to read into
>>>>>>> + * @param off the offset to start reading bytes into
>>>>>>> + * @param len the number of bytes to read
>>>>>>> + * @return the number of bytes read or <code>-1</code>
>>>>>>> + * if the end of the stream has been reached
>>>>>>> + * @throws IOException if an I/O error occurs
>>>>>>> + */
>>>>>>> + @Override
>>>>>>> + public int read(byte[] b, int off, int len) throws IOException {
>>>>>>> + if (b == null) {
>>>>>>> + throw new NullPointerException("Byte array must not be null");
>>>>>>> + }
>>>>>>> + if (len < 0 || off < 0 || (off + len) > b.length) {
>>>>>>> + throw new IndexOutOfBoundsException("Array Size=" + b.length
>>>>>>> + + ", offset=" + off + ", length=" + len);
>>>>>>> + }
>>>>>>> + int read = 0;
>>>>>>> + if (len == 0) {
>>>>>>> + return 0; // Always return 0 if len == 0
>>>>>>> + }
>>>>>>> + while (len > 0) {
>>>>>>> + if (encoderOut.hasRemaining()) {
>>>>>>> + int c = Math.min(encoderOut.remaining(), len);
>>>>>>> + encoderOut.get(b, off, c);
>>>>>>> + off += c;
>>>>>>> + len -= c;
>>>>>>> + read += c;
>>>>>>> + } else {
>>>>>>> + fillBuffer();
>>>>>>> + if (endOfInput && !encoderOut.hasRemaining()) {
>>>>>>> + break;
>>>>>>> + }
>>>>>>> + }
>>>>>>> + }
>>>>>>> + return read == 0 && endOfInput ? -1 : read;
>>>>>>> + }
>>>>>>> +
>>>>>>> + /**
>>>>>>> + * Read the specified number of bytes into an array.
>>>>>>> + *
>>>>>>> + * @param b the byte array to read into
>>>>>>> + * @return the number of bytes read or <code>-1</code>
>>>>>>> + * if the end of the stream has been reached
>>>>>>> + * @throws IOException if an I/O error occurs
>>>>>>> + */
>>>>>>> + @Override
>>>>>>> + public int read(byte[] b) throws IOException {
>>>>>>> + return read(b, 0, b.length);
>>>>>>> + }
>>>>>>> +
>>>>>>> + /**
>>>>>>> + * Read a single byte.
>>>>>>> + *
>>>>>>> + * @return either the byte read or <code>-1</code> if the end of the stream
>>>>>>> + * has been reached
>>>>>>> + * @throws IOException if an I/O error occurs
>>>>>>> + */
>>>>>>> + @Override
>>>>>>> + public int read() throws IOException {
>>>>>>> + for (;;) {
>>>>>>> + if (encoderOut.hasRemaining()) {
>>>>>>> + return encoderOut.get() & 0xFF;
>>>>>>> + } else {
>>>>>>> + fillBuffer();
>>>>>>> + if (endOfInput && !encoderOut.hasRemaining()) {
>>>>>>> + return -1;
>>>>>>> + }
>>>>>>> + }
>>>>>>> + }
>>>>>>> + }
>>>>>>> +
>>>>>>> + /**
>>>>>>> + * Close the stream. This method will cause the underlying {@link Reader}
>>>>>>> + * to be closed.
>>>>>>> + * @throws IOException if an I/O error occurs
>>>>>>> + */
>>>>>>> + @Override
>>>>>>> + public void close() throws IOException {
>>>>>>> + reader.close();
>>>>>>> + }
>>>>>>> +}
>>>>>>> diff --git a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>>>> index f77b5ed..b957f36 100644
>>>>>>> --- a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>>>> +++ b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>>>> @@ -96,6 +96,25 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>>>> assertEquals("Get a wrong response code.", 200, client.getResponse().getStatus());
>>>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>> }
>>>>>>> +
>>>>>>> + @Test
>>>>>>> + public void testGetBookFromWebClientWithTextJMSMessage() throws Exception {
>>>>>>> + // setup the the client
>>>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>>> + + "?replyToName=dynamicQueues/test.jmstransport.response"
>>>>>>> + + "&jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>>> + + "&messageType=text";
>>>>>>> +
>>>>>>> + WebClient client = WebClient.create(endpointAddressUrlEncoded);
>>>>>>> + WebClient.getConfig(client).getInInterceptors().add(new LoggingInInterceptor());
>>>>>>> + WebClient.getConfig(client).getRequestContext()
>>>>>>> + .put(org.apache.cxf.message.Message.REQUEST_URI, "/bookstore/books/123");
>>>>>>> +
>>>>>>> + Book book = client.get(Book.class);
>>>>>>> + assertEquals("Get a wrong response code.", 200, client.getResponse().getStatus());
>>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>> + }
>>>>>>> @Test
>>>>>>> public void testPutBookOneWayWithWebClient() throws Exception {
>>>>>>> @@ -130,6 +149,7 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>>>> }
>>>>>>> }
>>>>>>> +
>>>>>>> @Test
>>>>>>> public void testGetBookFromWebClientWithPath() throws Exception {
>>>>>>> // setup the the client
>>>>>>> @@ -148,6 +168,25 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>>>> }
>>>>>>> @Test
>>>>>>> + public void testGetBookFromWebClientWithPathWithTextJMSMessage() throws Exception {
>>>>>>> + // setup the the client
>>>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>>> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>>>> + + "&messageType=text";
>>>>>>> +
>>>>>>> + WebClient client = WebClient.create(endpointAddressUrlEncoded);
>>>>>>> + client.path("bookstore").path("books").path("123");
>>>>>>> +
>>>>>>> + Book book = client.get(Book.class);
>>>>>>> + assertEquals("Get a wrong response code.", 200, client.getResponse().getStatus());
>>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>> + }
>>>>>>> +
>>>>>>> +
>>>>>>> + @Test
>>>>>>> public void testGetBookFromProxyClient() throws Exception {
>>>>>>> // setup the the client
>>>>>>> String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>>> @@ -161,6 +200,22 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>>>> assertEquals("Get a wrong response code.", 200, WebClient.client(client).getResponse().getStatus());
>>>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>> }
>>>>>>> +
>>>>>>> + @Test
>>>>>>> + public void testGetBookFromProxyClientWithTextJMSMessage() throws Exception {
>>>>>>> + // setup the the client
>>>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>>> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>>>> + + "&messageType=text";
>>>>>>> +
>>>>>>> + JMSBookStore client = JAXRSClientFactory.create(endpointAddressUrlEncoded, JMSBookStore.class);
>>>>>>> + Book book = client.getBook("123");
>>>>>>> + assertEquals("Get a wrong response code.", 200, WebClient.client(client).getResponse().getStatus());
>>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>> + }
>>>>>>> @Test
>>>>>>> public void testGetBookFromSubresourceProxyClient() throws Exception {
>>>>>>> @@ -177,6 +232,23 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>>>> assertEquals("Get a wrong response code.", 200, WebClient.client(bookProxy).getResponse().getStatus());
>>>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>> }
>>>>>>> +
>>>>>>> + @Test
>>>>>>> + public void testGetBookFromSubresourceProxyClientWithTextJMSMessage() throws Exception {
>>>>>>> + // setup the the client
>>>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>>> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>>>> + + "&messageType=text";
>>>>>>> +
>>>>>>> + JMSBookStore client = JAXRSClientFactory.create(endpointAddressUrlEncoded, JMSBookStore.class);
>>>>>>> + Book bookProxy = client.getBookSubResource("123");
>>>>>>> + Book book = bookProxy.retrieveState();
>>>>>>> + assertEquals("Get a wrong response code.", 200, WebClient.client(bookProxy).getResponse().getStatus());
>>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>> + }
>>>>>>> @Test
>>>>>>> public void testGetBookFromProxyClientWithQuery() throws Exception {
>>>>>>> @@ -193,6 +265,24 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>> }
>>>>>>> +
>>>>>>> + @Test
>>>>>>> + public void testGetBookFromProxyClientWithQueryWithTextJMSMessage() throws Exception {
>>>>>>> + // setup the the client
>>>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>>> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>>>> + + "&messageType=text";
>>>>>>> +
>>>>>>> + JMSBookStore client = JAXRSClientFactory.create(endpointAddressUrlEncoded, JMSBookStore.class);
>>>>>>> + Book book = client.getBookByURLQuery(new String[] {"1", "2", "3"});
>>>>>>> + assertEquals("Get a wrong response code.", 200, WebClient.client(client).getResponse().getStatus());
>>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>>> + }
>>>>>>> +
>>>>>>> +
>>>>>>> @Test
>>>>>>> public void testGetBook() throws Exception {
>>>>>>> Context ctx = getContext();
>>>>>>>
>>>>>
>>>>
>>
>>
>> --
>> Sergey Beryozkin
>>
>> Talend Community Coders
>> http://coders.talend.com/
>
--
Daniel Kulp
dkulp@apache.org - http://dankulp.com/blog
Talend Community Coder - http://coders.talend.com
Re: [cxf] branch master updated: [CXF-7532]REST on JMS transport can't handle the request message with text messageType
Posted by ffang <fr...@gmail.com>.
I modified the JMS transport part and set ReaderInputStream instead of Reader, running all tests now, will see what I can get.
If I run into any test failure, I will keep things as is, just clean up.
－－－－－－－－－－－－－
Freeman(Yue) Fang
Red Hat, Inc.
FuseSource is now part of Red Hat
> On Oct 19, 2017, at 5:16 PM, Sergey Beryozkin <sb...@gmail.com> wrote:
>
> If we assume that checking Reader can be useful in the general sense, then may be lets keep things as is, with the minor cleanup only...
>
> Sergey
> On 19/10/17 10:05, Sergey Beryozkin wrote:
>> Hi Freeman
>> Reader may likely need to be set too for SOAP, and I recall checking Reader somewhere else too (will double check now), but I thought may be the transport can set both Reader and InputStream, hmm, that is an extra instance though, may be lets start with moving ReaderInputStream to the core first ?
>> Sergey
>> On 19/10/17 09:59, ffang wrote:
>>> Yeah, set ReaderInputStream instead of Reader for the JMS transport is a better solution. If no objection I will go this way.
>>> －－－－－－－－－－－－－
>>> Freeman(Yue) Fang
>>>
>>> Red Hat, Inc.
>>> FuseSource is now part of Red Hat
>>>
>>>
>>>
>>>> On Oct 19, 2017, at 4:50 PM, Sergey Beryozkin <sberyozkin@gmail.com <ma...@gmail.com>> wrote:
>>>>
>>>> The other thing is, should JMS transport itself set ReaderInputStream as InputStream on the message, as opposed to patching the JAX-RS code in various places ?
>>>>
>>>> Sergey
>>>> On 19/10/17 09:44, Sergey Beryozkin wrote:
>>>>> Hi Freeman
>>>>> Can you move ReaderInputStream to the CXF core where CachedInputStream and other IO utility code is located ?
>>>>> Please also update the code which gets Reader.class and check it for null first, and also drop a comment re the JMS transport, checking the reader can make sense in all the cases...
>>>>> Thanks, Sergey
>>>>> On 19/10/17 04:16, ffang@apache.org <ma...@apache.org> wrote:
>>>>>> This is an automated email from the ASF dual-hosted git repository.
>>>>>>
>>>>>> ffang pushed a commit to branch master
>>>>>> in repository https://gitbox.apache.org/repos/asf/cxf.git
>>>>>>
>>>>>>
>>>>>> The following commit(s) were added to refs/heads/master by this push:
>>>>>> new 1d66475 [CXF-7532]REST on JMS transport can't handle the request message with text messageType
>>>>>> 1d66475 is described below
>>>>>>
>>>>>> commit 1d66475b4dea2a0031dcb1765e8e9fd08430ec9b
>>>>>> Author: Freeman Fang <freeman.fang@gmail.com <ma...@gmail.com>>
>>>>>> AuthorDate: Thu Oct 19 11:15:50 2017 +0800
>>>>>>
>>>>>> [CXF-7532]REST on JMS transport can't handle the request message with text messageType
>>>>>> ---
>>>>>> .../org/apache/cxf/jaxrs/impl/ResponseImpl.java | 5 +
>>>>>> .../org/apache/cxf/jaxrs/utils/JAXRSUtils.java | 6 +-
>>>>>> .../apache/cxf/jaxrs/utils/ReaderInputStream.java | 296 +++++++++++++++++++++
>>>>>> .../apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java | 90 +++++++
>>>>>> 4 files changed, 396 insertions(+), 1 deletion(-)
>>>>>>
>>>>>> diff --git a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>>> index e3a3acf..82f183a 100644
>>>>>> --- a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>>> +++ b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>>> @@ -58,6 +58,7 @@ import org.apache.cxf.jaxrs.provider.ProviderFactory;
>>>>>> import org.apache.cxf.jaxrs.utils.HttpUtils;
>>>>>> import org.apache.cxf.jaxrs.utils.InjectionUtils;
>>>>>> import org.apache.cxf.jaxrs.utils.JAXRSUtils;
>>>>>> +import org.apache.cxf.jaxrs.utils.ReaderInputStream;
>>>>>> import org.apache.cxf.message.Message;
>>>>>> import org.apache.cxf.message.MessageUtils;
>>>>>> @@ -342,6 +343,10 @@ public final class ResponseImpl extends Response {
>>>>>> entityStreamAvailable = entityStream != null;
>>>>>> } else if (entity instanceof InputStream) {
>>>>>> entityStream = InputStream.class.cast(entity);
>>>>>> + } else {
>>>>>> + Message inMessage = getResponseMessage();
>>>>>> + Reader reader = inMessage.getContent(Reader.class);
>>>>>> + entityStream = InputStream.class.cast(new ReaderInputStream(reader));
>>>>>> }
>>>>>> // we need to check for readers even if no IS is set - the readers may still do it
>>>>>> diff --git a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>>> index 0ef6565..509f76d 100644
>>>>>> --- a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>>> +++ b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>>> @@ -832,7 +832,11 @@ public final class JAXRSUtils {
>>>>>> OperationResourceInfo ori)
>>>>>> throws IOException, WebApplicationException {
>>>>>> InputStream is = message.getContent(InputStream.class);
>>>>>> -
>>>>>> + if (is == null) {
>>>>>> + //may use the jms transport so check the Reader;
>>>>>> + Reader reader = message.getContent(Reader.class);
>>>>>> + is = new ReaderInputStream(reader);
>>>>>> + }
>>>>>> if (parameter.getType() == ParameterType.REQUEST_BODY) {
>>>>>> if (parameterClass == AsyncResponse.class) {
>>>>>> diff --git a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
>>>>>> new file mode 100644
>>>>>> index 0000000..c7b142e
>>>>>> --- /dev/null
>>>>>> +++ b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
>>>>>> @@ -0,0 +1,296 @@
>>>>>> +/**
>>>>>> + * Licensed to the Apache Software Foundation (ASF) under one
>>>>>> + * or more contributor license agreements. See the NOTICE file
>>>>>> + * distributed with this work for additional information
>>>>>> + * regarding copyright ownership. The ASF licenses this file
>>>>>> + * to you 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.apache.cxf.jaxrs.utils;
>>>>>> +
>>>>>> +import java.io.IOException;
>>>>>> +import java.io.InputStream;
>>>>>> +import java.io.Reader;
>>>>>> +import java.nio.ByteBuffer;
>>>>>> +import java.nio.CharBuffer;
>>>>>> +import java.nio.charset.Charset;
>>>>>> +import java.nio.charset.CharsetEncoder;
>>>>>> +import java.nio.charset.CoderResult;
>>>>>> +import java.nio.charset.CodingErrorAction;
>>>>>> +
>>>>>> +/**
>>>>>> + * {@link InputStream} implementation that reads a character stream from a {@link Reader}
>>>>>> + * and transforms it to a byte stream using a specified charset encoding. The stream
>>>>>> + * is transformed using a {@link CharsetEncoder} object, guaranteeing that all charset
>>>>>> + * encodings supported by the JRE are handled correctly. In particular for charsets such as
>>>>>> + * UTF-16, the implementation ensures that one and only one byte order marker
>>>>>> + * is produced.
>>>>>> + * <p>
>>>>>> + * Since in general it is not possible to predict the number of characters to be read from the
>>>>>> + * {@link Reader} to satisfy a read request on the {@link ReaderInputStream}, all reads from
>>>>>> + * the {@link Reader} are buffered. There is therefore no well defined correlation
>>>>>> + * between the current position of the {@link Reader} and that of the {@link ReaderInputStream}.
>>>>>> + * This also implies that in general there is no need to wrap the underlying {@link Reader}
>>>>>> + * in a {@link java.io.BufferedReader}.
>>>>>> + * <p>
>>>>>> + * {@link ReaderInputStream} implements the inverse transformation of {@link java.io.InputStreamReader};
>>>>>> + * in the following example, reading from <tt>in2</tt> would return the same byte
>>>>>> + * sequence as reading from <tt>in</tt> (provided that the initial byte sequence is legal
>>>>>> + * with respect to the charset encoding):
>>>>>> + * <pre>
>>>>>> + * InputStream in = ...
>>>>>> + * Charset cs = ...
>>>>>> + * InputStreamReader reader = new InputStreamReader(in, cs);
>>>>>> + * ReaderInputStream in2 = new ReaderInputStream(reader, cs);</pre>
>>>>>> + * {@link ReaderInputStream} implements the same transformation as {@link java.io.OutputStreamWriter},
>>>>>> + * except that the control flow is reversed: both classes transform a character stream
>>>>>> + * into a byte stream, but {@link java.io.OutputStreamWriter} pushes data to the underlying stream,
>>>>>> + * while {@link ReaderInputStream} pulls it from the underlying stream.
>>>>>> + * <p>
>>>>>> + * Note that while there are use cases where there is no alternative to using
>>>>>> + * this class, very often the need to use this class is an indication of a flaw
>>>>>> + * in the design of the code. This class is typically used in situations where an existing
>>>>>> + * API only accepts an {@link InputStream}, but where the most natural way to produce the data
>>>>>> + * is as a character stream, i.e. by providing a {@link Reader} instance. An example of a situation
>>>>>> + * where this problem may appear is when implementing the {@link javax.activation.DataSource}
>>>>>> + * interface from the Java Activation Framework.
>>>>>> + * <p>
>>>>>> + * Given the fact that the {@link Reader} class doesn't provide any way to predict whether the next
>>>>>> + * read operation will block or not, it is not possible to provide a meaningful
>>>>>> + * implementation of the {@link InputStream#available()} method. A call to this method
>>>>>> + * will always return 0. Also, this class doesn't support {@link InputStream#mark(int)}.
>>>>>> + * <p>
>>>>>> + * Instances of {@link ReaderInputStream} are not thread safe.
>>>>>> + *
>>>>>> + * @see org.apache.commons.io.output.WriterOutputStream
>>>>>> + *
>>>>>> + * @since 2.0
>>>>>> + */
>>>>>> +public class ReaderInputStream extends InputStream {
>>>>>> + private static final int DEFAULT_BUFFER_SIZE = 1024;
>>>>>> +
>>>>>> + private final Reader reader;
>>>>>> + private final CharsetEncoder encoder;
>>>>>> +
>>>>>> + /**
>>>>>> + * CharBuffer used as input for the decoder. It should be reasonably
>>>>>> + * large as we read data from the underlying Reader into this buffer.
>>>>>> + */
>>>>>> + private final CharBuffer encoderIn;
>>>>>> +
>>>>>> + /**
>>>>>> + * ByteBuffer used as output for the decoder. This buffer can be small
>>>>>> + * as it is only used to transfer data from the decoder to the
>>>>>> + * buffer provided by the caller.
>>>>>> + */
>>>>>> + private final ByteBuffer encoderOut;
>>>>>> +
>>>>>> + private CoderResult lastCoderResult;
>>>>>> + private boolean endOfInput;
>>>>>> +
>>>>>> + /**
>>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>>> + *
>>>>>> + * @param reader the target {@link Reader}
>>>>>> + * @param encoder the charset encoder
>>>>>> + * @since 2.1
>>>>>> + */
>>>>>> + public ReaderInputStream(Reader reader, CharsetEncoder encoder) {
>>>>>> + this(reader, encoder, DEFAULT_BUFFER_SIZE);
>>>>>> + }
>>>>>> +
>>>>>> + /**
>>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>>> + *
>>>>>> + * @param reader the target {@link Reader}
>>>>>> + * @param encoder the charset encoder
>>>>>> + * @param bufferSize the size of the input buffer in number of characters
>>>>>> + * @since 2.1
>>>>>> + */
>>>>>> + public ReaderInputStream(Reader reader, CharsetEncoder encoder, int bufferSize) {
>>>>>> + this.reader = reader;
>>>>>> + this.encoder = encoder;
>>>>>> + this.encoderIn = CharBuffer.allocate(bufferSize);
>>>>>> + this.encoderIn.flip();
>>>>>> + this.encoderOut = ByteBuffer.allocate(128);
>>>>>> + this.encoderOut.flip();
>>>>>> + }
>>>>>> +
>>>>>> + /**
>>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>>> + *
>>>>>> + * @param reader the target {@link Reader}
>>>>>> + * @param charset the charset encoding
>>>>>> + * @param bufferSize the size of the input buffer in number of characters
>>>>>> + */
>>>>>> + public ReaderInputStream(Reader reader, Charset charset, int bufferSize) {
>>>>>> + this(reader,
>>>>>> + charset.newEncoder()
>>>>>> + .onMalformedInput(CodingErrorAction.REPLACE)
>>>>>> + .onUnmappableCharacter(CodingErrorAction.REPLACE),
>>>>>> + bufferSize);
>>>>>> + }
>>>>>> +
>>>>>> + /**
>>>>>> + * Construct a new {@link ReaderInputStream} with a default input buffer size of
>>>>>> + * 1024 characters.
>>>>>> + *
>>>>>> + * @param reader the target {@link Reader}
>>>>>> + * @param charset the charset encoding
>>>>>> + */
>>>>>> + public ReaderInputStream(Reader reader, Charset charset) {
>>>>>> + this(reader, charset, DEFAULT_BUFFER_SIZE);
>>>>>> + }
>>>>>> +
>>>>>> + /**
>>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>>> + *
>>>>>> + * @param reader the target {@link Reader}
>>>>>> + * @param charsetName the name of the charset encoding
>>>>>> + * @param bufferSize the size of the input buffer in number of characters
>>>>>> + */
>>>>>> + public ReaderInputStream(Reader reader, String charsetName, int bufferSize) {
>>>>>> + this(reader, Charset.forName(charsetName), bufferSize);
>>>>>> + }
>>>>>> +
>>>>>> + /**
>>>>>> + * Construct a new {@link ReaderInputStream} with a default input buffer size of
>>>>>> + * 1024 characters.
>>>>>> + *
>>>>>> + * @param reader the target {@link Reader}
>>>>>> + * @param charsetName the name of the charset encoding
>>>>>> + */
>>>>>> + public ReaderInputStream(Reader reader, String charsetName) {
>>>>>> + this(reader, charsetName, DEFAULT_BUFFER_SIZE);
>>>>>> + }
>>>>>> +
>>>>>> + /**
>>>>>> + * Construct a new {@link ReaderInputStream} that uses the default character encoding
>>>>>> + * with a default input buffer size of 1024 characters.
>>>>>> + *
>>>>>> + * @param reader the target {@link Reader}
>>>>>> + */
>>>>>> + public ReaderInputStream(Reader reader) {
>>>>>> + this(reader, Charset.defaultCharset());
>>>>>> + }
>>>>>> +
>>>>>> + /**
>>>>>> + * Fills the internal char buffer from the reader.
>>>>>> + *
>>>>>> + * @throws IOException
>>>>>> + * If an I/O error occurs
>>>>>> + */
>>>>>> + private void fillBuffer() throws IOException {
>>>>>> + if (!endOfInput && (lastCoderResult == null || lastCoderResult.isUnderflow())) {
>>>>>> + encoderIn.compact();
>>>>>> + int position = encoderIn.position();
>>>>>> + // We don't use Reader#read(CharBuffer) here because it is more efficient
>>>>>> + // to write directly to the underlying char array (the default implementation
>>>>>> + // copies data to a temporary char array).
>>>>>> + int c = reader.read(encoderIn.array(), position, encoderIn.remaining());
>>>>>> + if (c == -1) {
>>>>>> + endOfInput = true;
>>>>>> + } else {
>>>>>> + encoderIn.position(position + c);
>>>>>> + }
>>>>>> + encoderIn.flip();
>>>>>> + }
>>>>>> + encoderOut.compact();
>>>>>> + lastCoderResult = encoder.encode(encoderIn, encoderOut, endOfInput);
>>>>>> + encoderOut.flip();
>>>>>> + }
>>>>>> +
>>>>>> + /**
>>>>>> + * Read the specified number of bytes into an array.
>>>>>> + *
>>>>>> + * @param b the byte array to read into
>>>>>> + * @param off the offset to start reading bytes into
>>>>>> + * @param len the number of bytes to read
>>>>>> + * @return the number of bytes read or <code>-1</code>
>>>>>> + * if the end of the stream has been reached
>>>>>> + * @throws IOException if an I/O error occurs
>>>>>> + */
>>>>>> + @Override
>>>>>> + public int read(byte[] b, int off, int len) throws IOException {
>>>>>> + if (b == null) {
>>>>>> + throw new NullPointerException("Byte array must not be null");
>>>>>> + }
>>>>>> + if (len < 0 || off < 0 || (off + len) > b.length) {
>>>>>> + throw new IndexOutOfBoundsException("Array Size=" + b.length
>>>>>> + + ", offset=" + off + ", length=" + len);
>>>>>> + }
>>>>>> + int read = 0;
>>>>>> + if (len == 0) {
>>>>>> + return 0; // Always return 0 if len == 0
>>>>>> + }
>>>>>> + while (len > 0) {
>>>>>> + if (encoderOut.hasRemaining()) {
>>>>>> + int c = Math.min(encoderOut.remaining(), len);
>>>>>> + encoderOut.get(b, off, c);
>>>>>> + off += c;
>>>>>> + len -= c;
>>>>>> + read += c;
>>>>>> + } else {
>>>>>> + fillBuffer();
>>>>>> + if (endOfInput && !encoderOut.hasRemaining()) {
>>>>>> + break;
>>>>>> + }
>>>>>> + }
>>>>>> + }
>>>>>> + return read == 0 && endOfInput ? -1 : read;
>>>>>> + }
>>>>>> +
>>>>>> + /**
>>>>>> + * Read the specified number of bytes into an array.
>>>>>> + *
>>>>>> + * @param b the byte array to read into
>>>>>> + * @return the number of bytes read or <code>-1</code>
>>>>>> + * if the end of the stream has been reached
>>>>>> + * @throws IOException if an I/O error occurs
>>>>>> + */
>>>>>> + @Override
>>>>>> + public int read(byte[] b) throws IOException {
>>>>>> + return read(b, 0, b.length);
>>>>>> + }
>>>>>> +
>>>>>> + /**
>>>>>> + * Read a single byte.
>>>>>> + *
>>>>>> + * @return either the byte read or <code>-1</code> if the end of the stream
>>>>>> + * has been reached
>>>>>> + * @throws IOException if an I/O error occurs
>>>>>> + */
>>>>>> + @Override
>>>>>> + public int read() throws IOException {
>>>>>> + for (;;) {
>>>>>> + if (encoderOut.hasRemaining()) {
>>>>>> + return encoderOut.get() & 0xFF;
>>>>>> + } else {
>>>>>> + fillBuffer();
>>>>>> + if (endOfInput && !encoderOut.hasRemaining()) {
>>>>>> + return -1;
>>>>>> + }
>>>>>> + }
>>>>>> + }
>>>>>> + }
>>>>>> +
>>>>>> + /**
>>>>>> + * Close the stream. This method will cause the underlying {@link Reader}
>>>>>> + * to be closed.
>>>>>> + * @throws IOException if an I/O error occurs
>>>>>> + */
>>>>>> + @Override
>>>>>> + public void close() throws IOException {
>>>>>> + reader.close();
>>>>>> + }
>>>>>> +}
>>>>>> diff --git a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>>> index f77b5ed..b957f36 100644
>>>>>> --- a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>>> +++ b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>>> @@ -96,6 +96,25 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>>> assertEquals("Get a wrong response code.", 200, client.getResponse().getStatus());
>>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>> }
>>>>>> +
>>>>>> + @Test
>>>>>> + public void testGetBookFromWebClientWithTextJMSMessage() throws Exception {
>>>>>> + // setup the the client
>>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>> + + "?replyToName=dynamicQueues/test.jmstransport.response"
>>>>>> + + "&jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>> + + "&messageType=text";
>>>>>> +
>>>>>> + WebClient client = WebClient.create(endpointAddressUrlEncoded);
>>>>>> + WebClient.getConfig(client).getInInterceptors().add(new LoggingInInterceptor());
>>>>>> + WebClient.getConfig(client).getRequestContext()
>>>>>> + .put(org.apache.cxf.message.Message.REQUEST_URI, "/bookstore/books/123");
>>>>>> +
>>>>>> + Book book = client.get(Book.class);
>>>>>> + assertEquals("Get a wrong response code.", 200, client.getResponse().getStatus());
>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>> + }
>>>>>> @Test
>>>>>> public void testPutBookOneWayWithWebClient() throws Exception {
>>>>>> @@ -130,6 +149,7 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>>> }
>>>>>> }
>>>>>> +
>>>>>> @Test
>>>>>> public void testGetBookFromWebClientWithPath() throws Exception {
>>>>>> // setup the the client
>>>>>> @@ -148,6 +168,25 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>>> }
>>>>>> @Test
>>>>>> + public void testGetBookFromWebClientWithPathWithTextJMSMessage() throws Exception {
>>>>>> + // setup the the client
>>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>>> + + "&messageType=text";
>>>>>> +
>>>>>> + WebClient client = WebClient.create(endpointAddressUrlEncoded);
>>>>>> + client.path("bookstore").path("books").path("123");
>>>>>> +
>>>>>> + Book book = client.get(Book.class);
>>>>>> + assertEquals("Get a wrong response code.", 200, client.getResponse().getStatus());
>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>> + }
>>>>>> +
>>>>>> +
>>>>>> + @Test
>>>>>> public void testGetBookFromProxyClient() throws Exception {
>>>>>> // setup the the client
>>>>>> String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>> @@ -161,6 +200,22 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>>> assertEquals("Get a wrong response code.", 200, WebClient.client(client).getResponse().getStatus());
>>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>> }
>>>>>> +
>>>>>> + @Test
>>>>>> + public void testGetBookFromProxyClientWithTextJMSMessage() throws Exception {
>>>>>> + // setup the the client
>>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>>> + + "&messageType=text";
>>>>>> +
>>>>>> + JMSBookStore client = JAXRSClientFactory.create(endpointAddressUrlEncoded, JMSBookStore.class);
>>>>>> + Book book = client.getBook("123");
>>>>>> + assertEquals("Get a wrong response code.", 200, WebClient.client(client).getResponse().getStatus());
>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>> + }
>>>>>> @Test
>>>>>> public void testGetBookFromSubresourceProxyClient() throws Exception {
>>>>>> @@ -177,6 +232,23 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>>> assertEquals("Get a wrong response code.", 200, WebClient.client(bookProxy).getResponse().getStatus());
>>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>> }
>>>>>> +
>>>>>> + @Test
>>>>>> + public void testGetBookFromSubresourceProxyClientWithTextJMSMessage() throws Exception {
>>>>>> + // setup the the client
>>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>>> + + "&messageType=text";
>>>>>> +
>>>>>> + JMSBookStore client = JAXRSClientFactory.create(endpointAddressUrlEncoded, JMSBookStore.class);
>>>>>> + Book bookProxy = client.getBookSubResource("123");
>>>>>> + Book book = bookProxy.retrieveState();
>>>>>> + assertEquals("Get a wrong response code.", 200, WebClient.client(bookProxy).getResponse().getStatus());
>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>> + }
>>>>>> @Test
>>>>>> public void testGetBookFromProxyClientWithQuery() throws Exception {
>>>>>> @@ -193,6 +265,24 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>> }
>>>>>> +
>>>>>> + @Test
>>>>>> + public void testGetBookFromProxyClientWithQueryWithTextJMSMessage() throws Exception {
>>>>>> + // setup the the client
>>>>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>>> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>>> + + "&messageType=text";
>>>>>> +
>>>>>> + JMSBookStore client = JAXRSClientFactory.create(endpointAddressUrlEncoded, JMSBookStore.class);
>>>>>> + Book book = client.getBookByURLQuery(new String[] {"1", "2", "3"});
>>>>>> + assertEquals("Get a wrong response code.", 200, WebClient.client(client).getResponse().getStatus());
>>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>>> + }
>>>>>> +
>>>>>> +
>>>>>> @Test
>>>>>> public void testGetBook() throws Exception {
>>>>>> Context ctx = getContext();
>>>>>>
>>>>
>>>
>
>
> --
> Sergey Beryozkin
>
> Talend Community Coders
> http://coders.talend.com/
Re: [cxf] branch master updated: [CXF-7532]REST on JMS transport
can't handle the request message with text messageType
Posted by Sergey Beryozkin <sb...@gmail.com>.
If we assume that checking Reader can be useful in the general sense,
then may be lets keep things as is, with the minor cleanup only...
Sergey
On 19/10/17 10:05, Sergey Beryozkin wrote:
> Hi Freeman
>
> Reader may likely need to be set too for SOAP, and I recall checking
> Reader somewhere else too (will double check now), but I thought may be
> the transport can set both Reader and InputStream, hmm, that is an extra
> instance though, may be lets start with moving ReaderInputStream to the
> core first ?
>
> Sergey
> On 19/10/17 09:59, ffang wrote:
>> Yeah, set ReaderInputStream instead of Reader for the JMS transport is
>> a better solution. If no objection I will go this way.
>> －－－－－－－－－－－－－
>> Freeman(Yue) Fang
>>
>> Red Hat, Inc.
>> FuseSource is now part of Red Hat
>>
>>
>>
>>> On Oct 19, 2017, at 4:50 PM, Sergey Beryozkin <sberyozkin@gmail.com
>>> <ma...@gmail.com>> wrote:
>>>
>>> The other thing is, should JMS transport itself set ReaderInputStream
>>> as InputStream on the message, as opposed to patching the JAX-RS code
>>> in various places ?
>>>
>>> Sergey
>>> On 19/10/17 09:44, Sergey Beryozkin wrote:
>>>> Hi Freeman
>>>> Can you move ReaderInputStream to the CXF core where
>>>> CachedInputStream and other IO utility code is located ?
>>>> Please also update the code which gets Reader.class and check it for
>>>> null first, and also drop a comment re the JMS transport, checking
>>>> the reader can make sense in all the cases...
>>>> Thanks, Sergey
>>>> On 19/10/17 04:16, ffang@apache.org <ma...@apache.org> wrote:
>>>>> This is an automated email from the ASF dual-hosted git repository.
>>>>>
>>>>> ffang pushed a commit to branch master
>>>>> in repository https://gitbox.apache.org/repos/asf/cxf.git
>>>>>
>>>>>
>>>>> The following commit(s) were added to refs/heads/master by this push:
>>>>> new 1d66475 [CXF-7532]REST on JMS transport can't handle the
>>>>> request message with text messageType
>>>>> 1d66475 is described below
>>>>>
>>>>> commit 1d66475b4dea2a0031dcb1765e8e9fd08430ec9b
>>>>> Author: Freeman Fang <freeman.fang@gmail.com
>>>>> <ma...@gmail.com>>
>>>>> AuthorDate: Thu Oct 19 11:15:50 2017 +0800
>>>>>
>>>>> [CXF-7532]REST on JMS transport can't handle the request
>>>>> message with text messageType
>>>>> ---
>>>>> .../org/apache/cxf/jaxrs/impl/ResponseImpl.java | 5 +
>>>>> .../org/apache/cxf/jaxrs/utils/JAXRSUtils.java | 6 +-
>>>>> .../apache/cxf/jaxrs/utils/ReaderInputStream.java | 296
>>>>> +++++++++++++++++++++
>>>>> .../apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java | 90 +++++++
>>>>> 4 files changed, 396 insertions(+), 1 deletion(-)
>>>>>
>>>>> diff --git
>>>>> a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>> b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>>
>>>>> index e3a3acf..82f183a 100644
>>>>> ---
>>>>> a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>>
>>>>> +++
>>>>> b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>>
>>>>> @@ -58,6 +58,7 @@ import
>>>>> org.apache.cxf.jaxrs.provider.ProviderFactory;
>>>>> import org.apache.cxf.jaxrs.utils.HttpUtils;
>>>>> import org.apache.cxf.jaxrs.utils.InjectionUtils;
>>>>> import org.apache.cxf.jaxrs.utils.JAXRSUtils;
>>>>> +import org.apache.cxf.jaxrs.utils.ReaderInputStream;
>>>>> import org.apache.cxf.message.Message;
>>>>> import org.apache.cxf.message.MessageUtils;
>>>>> @@ -342,6 +343,10 @@ public final class ResponseImpl extends
>>>>> Response {
>>>>> entityStreamAvailable = entityStream != null;
>>>>> } else if (entity instanceof InputStream) {
>>>>> entityStream = InputStream.class.cast(entity);
>>>>> + } else {
>>>>> + Message inMessage = getResponseMessage();
>>>>> + Reader reader = inMessage.getContent(Reader.class);
>>>>> + entityStream = InputStream.class.cast(new
>>>>> ReaderInputStream(reader));
>>>>> }
>>>>> // we need to check for readers even if no IS is set -
>>>>> the readers may still do it
>>>>> diff --git
>>>>> a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>> b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>>
>>>>> index 0ef6565..509f76d 100644
>>>>> ---
>>>>> a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>>
>>>>> +++
>>>>> b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>>
>>>>> @@ -832,7 +832,11 @@ public final class JAXRSUtils {
>>>>> OperationResourceInfo
>>>>> ori)
>>>>> throws IOException, WebApplicationException {
>>>>> InputStream is = message.getContent(InputStream.class);
>>>>> -
>>>>> + if (is == null) {
>>>>> + //may use the jms transport so check the Reader;
>>>>> + Reader reader = message.getContent(Reader.class);
>>>>> + is = new ReaderInputStream(reader);
>>>>> + }
>>>>> if (parameter.getType() == ParameterType.REQUEST_BODY) {
>>>>> if (parameterClass == AsyncResponse.class) {
>>>>> diff --git
>>>>> a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
>>>>> b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
>>>>>
>>>>> new file mode 100644
>>>>> index 0000000..c7b142e
>>>>> --- /dev/null
>>>>> +++
>>>>> b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
>>>>>
>>>>> @@ -0,0 +1,296 @@
>>>>> +/**
>>>>> + * Licensed to the Apache Software Foundation (ASF) under one
>>>>> + * or more contributor license agreements. See the NOTICE file
>>>>> + * distributed with this work for additional information
>>>>> + * regarding copyright ownership. The ASF licenses this file
>>>>> + * to you 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.apache.cxf.jaxrs.utils;
>>>>> +
>>>>> +import java.io.IOException;
>>>>> +import java.io.InputStream;
>>>>> +import java.io.Reader;
>>>>> +import java.nio.ByteBuffer;
>>>>> +import java.nio.CharBuffer;
>>>>> +import java.nio.charset.Charset;
>>>>> +import java.nio.charset.CharsetEncoder;
>>>>> +import java.nio.charset.CoderResult;
>>>>> +import java.nio.charset.CodingErrorAction;
>>>>> +
>>>>> +/**
>>>>> + * {@link InputStream} implementation that reads a character
>>>>> stream from a {@link Reader}
>>>>> + * and transforms it to a byte stream using a specified charset
>>>>> encoding. The stream
>>>>> + * is transformed using a {@link CharsetEncoder} object,
>>>>> guaranteeing that all charset
>>>>> + * encodings supported by the JRE are handled correctly. In
>>>>> particular for charsets such as
>>>>> + * UTF-16, the implementation ensures that one and only one byte
>>>>> order marker
>>>>> + * is produced.
>>>>> + * <p>
>>>>> + * Since in general it is not possible to predict the number of
>>>>> characters to be read from the
>>>>> + * {@link Reader} to satisfy a read request on the {@link
>>>>> ReaderInputStream}, all reads from
>>>>> + * the {@link Reader} are buffered. There is therefore no well
>>>>> defined correlation
>>>>> + * between the current position of the {@link Reader} and that of
>>>>> the {@link ReaderInputStream}.
>>>>> + * This also implies that in general there is no need to wrap the
>>>>> underlying {@link Reader}
>>>>> + * in a {@link java.io.BufferedReader}.
>>>>> + * <p>
>>>>> + * {@link ReaderInputStream} implements the inverse transformation
>>>>> of {@link java.io.InputStreamReader};
>>>>> + * in the following example, reading from <tt>in2</tt> would
>>>>> return the same byte
>>>>> + * sequence as reading from <tt>in</tt> (provided that the initial
>>>>> byte sequence is legal
>>>>> + * with respect to the charset encoding):
>>>>> + * <pre>
>>>>> + * InputStream in = ...
>>>>> + * Charset cs = ...
>>>>> + * InputStreamReader reader = new InputStreamReader(in, cs);
>>>>> + * ReaderInputStream in2 = new ReaderInputStream(reader, cs);</pre>
>>>>> + * {@link ReaderInputStream} implements the same transformation as
>>>>> {@link java.io.OutputStreamWriter},
>>>>> + * except that the control flow is reversed: both classes
>>>>> transform a character stream
>>>>> + * into a byte stream, but {@link java.io.OutputStreamWriter}
>>>>> pushes data to the underlying stream,
>>>>> + * while {@link ReaderInputStream} pulls it from the underlying
>>>>> stream.
>>>>> + * <p>
>>>>> + * Note that while there are use cases where there is no
>>>>> alternative to using
>>>>> + * this class, very often the need to use this class is an
>>>>> indication of a flaw
>>>>> + * in the design of the code. This class is typically used in
>>>>> situations where an existing
>>>>> + * API only accepts an {@link InputStream}, but where the most
>>>>> natural way to produce the data
>>>>> + * is as a character stream, i.e. by providing a {@link Reader}
>>>>> instance. An example of a situation
>>>>> + * where this problem may appear is when implementing the {@link
>>>>> javax.activation.DataSource}
>>>>> + * interface from the Java Activation Framework.
>>>>> + * <p>
>>>>> + * Given the fact that the {@link Reader} class doesn't provide
>>>>> any way to predict whether the next
>>>>> + * read operation will block or not, it is not possible to provide
>>>>> a meaningful
>>>>> + * implementation of the {@link InputStream#available()} method. A
>>>>> call to this method
>>>>> + * will always return 0. Also, this class doesn't support {@link
>>>>> InputStream#mark(int)}.
>>>>> + * <p>
>>>>> + * Instances of {@link ReaderInputStream} are not thread safe.
>>>>> + *
>>>>> + * @see org.apache.commons.io.output.WriterOutputStream
>>>>> + *
>>>>> + * @since 2.0
>>>>> + */
>>>>> +public class ReaderInputStream extends InputStream {
>>>>> + private static final int DEFAULT_BUFFER_SIZE = 1024;
>>>>> +
>>>>> + private final Reader reader;
>>>>> + private final CharsetEncoder encoder;
>>>>> +
>>>>> + /**
>>>>> + * CharBuffer used as input for the decoder. It should be
>>>>> reasonably
>>>>> + * large as we read data from the underlying Reader into this
>>>>> buffer.
>>>>> + */
>>>>> + private final CharBuffer encoderIn;
>>>>> +
>>>>> + /**
>>>>> + * ByteBuffer used as output for the decoder. This buffer can
>>>>> be small
>>>>> + * as it is only used to transfer data from the decoder to the
>>>>> + * buffer provided by the caller.
>>>>> + */
>>>>> + private final ByteBuffer encoderOut;
>>>>> +
>>>>> + private CoderResult lastCoderResult;
>>>>> + private boolean endOfInput;
>>>>> +
>>>>> + /**
>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>> + *
>>>>> + * @param reader the target {@link Reader}
>>>>> + * @param encoder the charset encoder
>>>>> + * @since 2.1
>>>>> + */
>>>>> + public ReaderInputStream(Reader reader, CharsetEncoder encoder) {
>>>>> + this(reader, encoder, DEFAULT_BUFFER_SIZE);
>>>>> + }
>>>>> +
>>>>> + /**
>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>> + *
>>>>> + * @param reader the target {@link Reader}
>>>>> + * @param encoder the charset encoder
>>>>> + * @param bufferSize the size of the input buffer in number of
>>>>> characters
>>>>> + * @since 2.1
>>>>> + */
>>>>> + public ReaderInputStream(Reader reader, CharsetEncoder
>>>>> encoder, int bufferSize) {
>>>>> + this.reader = reader;
>>>>> + this.encoder = encoder;
>>>>> + this.encoderIn = CharBuffer.allocate(bufferSize);
>>>>> + this.encoderIn.flip();
>>>>> + this.encoderOut = ByteBuffer.allocate(128);
>>>>> + this.encoderOut.flip();
>>>>> + }
>>>>> +
>>>>> + /**
>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>> + *
>>>>> + * @param reader the target {@link Reader}
>>>>> + * @param charset the charset encoding
>>>>> + * @param bufferSize the size of the input buffer in number of
>>>>> characters
>>>>> + */
>>>>> + public ReaderInputStream(Reader reader, Charset charset, int
>>>>> bufferSize) {
>>>>> + this(reader,
>>>>> + charset.newEncoder()
>>>>> + .onMalformedInput(CodingErrorAction.REPLACE)
>>>>> +
>>>>> .onUnmappableCharacter(CodingErrorAction.REPLACE),
>>>>> + bufferSize);
>>>>> + }
>>>>> +
>>>>> + /**
>>>>> + * Construct a new {@link ReaderInputStream} with a default
>>>>> input buffer size of
>>>>> + * 1024 characters.
>>>>> + *
>>>>> + * @param reader the target {@link Reader}
>>>>> + * @param charset the charset encoding
>>>>> + */
>>>>> + public ReaderInputStream(Reader reader, Charset charset) {
>>>>> + this(reader, charset, DEFAULT_BUFFER_SIZE);
>>>>> + }
>>>>> +
>>>>> + /**
>>>>> + * Construct a new {@link ReaderInputStream}.
>>>>> + *
>>>>> + * @param reader the target {@link Reader}
>>>>> + * @param charsetName the name of the charset encoding
>>>>> + * @param bufferSize the size of the input buffer in number of
>>>>> characters
>>>>> + */
>>>>> + public ReaderInputStream(Reader reader, String charsetName,
>>>>> int bufferSize) {
>>>>> + this(reader, Charset.forName(charsetName), bufferSize);
>>>>> + }
>>>>> +
>>>>> + /**
>>>>> + * Construct a new {@link ReaderInputStream} with a default
>>>>> input buffer size of
>>>>> + * 1024 characters.
>>>>> + *
>>>>> + * @param reader the target {@link Reader}
>>>>> + * @param charsetName the name of the charset encoding
>>>>> + */
>>>>> + public ReaderInputStream(Reader reader, String charsetName) {
>>>>> + this(reader, charsetName, DEFAULT_BUFFER_SIZE);
>>>>> + }
>>>>> +
>>>>> + /**
>>>>> + * Construct a new {@link ReaderInputStream} that uses the
>>>>> default character encoding
>>>>> + * with a default input buffer size of 1024 characters.
>>>>> + *
>>>>> + * @param reader the target {@link Reader}
>>>>> + */
>>>>> + public ReaderInputStream(Reader reader) {
>>>>> + this(reader, Charset.defaultCharset());
>>>>> + }
>>>>> +
>>>>> + /**
>>>>> + * Fills the internal char buffer from the reader.
>>>>> + *
>>>>> + * @throws IOException
>>>>> + * If an I/O error occurs
>>>>> + */
>>>>> + private void fillBuffer() throws IOException {
>>>>> + if (!endOfInput && (lastCoderResult == null ||
>>>>> lastCoderResult.isUnderflow())) {
>>>>> + encoderIn.compact();
>>>>> + int position = encoderIn.position();
>>>>> + // We don't use Reader#read(CharBuffer) here because
>>>>> it is more efficient
>>>>> + // to write directly to the underlying char array (the
>>>>> default implementation
>>>>> + // copies data to a temporary char array).
>>>>> + int c = reader.read(encoderIn.array(), position,
>>>>> encoderIn.remaining());
>>>>> + if (c == -1) {
>>>>> + endOfInput = true;
>>>>> + } else {
>>>>> + encoderIn.position(position + c);
>>>>> + }
>>>>> + encoderIn.flip();
>>>>> + }
>>>>> + encoderOut.compact();
>>>>> + lastCoderResult = encoder.encode(encoderIn, encoderOut,
>>>>> endOfInput);
>>>>> + encoderOut.flip();
>>>>> + }
>>>>> +
>>>>> + /**
>>>>> + * Read the specified number of bytes into an array.
>>>>> + *
>>>>> + * @param b the byte array to read into
>>>>> + * @param off the offset to start reading bytes into
>>>>> + * @param len the number of bytes to read
>>>>> + * @return the number of bytes read or <code>-1</code>
>>>>> + * if the end of the stream has been reached
>>>>> + * @throws IOException if an I/O error occurs
>>>>> + */
>>>>> + @Override
>>>>> + public int read(byte[] b, int off, int len) throws IOException {
>>>>> + if (b == null) {
>>>>> + throw new NullPointerException("Byte array must not be
>>>>> null");
>>>>> + }
>>>>> + if (len < 0 || off < 0 || (off + len) > b.length) {
>>>>> + throw new IndexOutOfBoundsException("Array Size=" +
>>>>> b.length
>>>>> + + ", offset=" + off + ", length=" + len);
>>>>> + }
>>>>> + int read = 0;
>>>>> + if (len == 0) {
>>>>> + return 0; // Always return 0 if len == 0
>>>>> + }
>>>>> + while (len > 0) {
>>>>> + if (encoderOut.hasRemaining()) {
>>>>> + int c = Math.min(encoderOut.remaining(), len);
>>>>> + encoderOut.get(b, off, c);
>>>>> + off += c;
>>>>> + len -= c;
>>>>> + read += c;
>>>>> + } else {
>>>>> + fillBuffer();
>>>>> + if (endOfInput && !encoderOut.hasRemaining()) {
>>>>> + break;
>>>>> + }
>>>>> + }
>>>>> + }
>>>>> + return read == 0 && endOfInput ? -1 : read;
>>>>> + }
>>>>> +
>>>>> + /**
>>>>> + * Read the specified number of bytes into an array.
>>>>> + *
>>>>> + * @param b the byte array to read into
>>>>> + * @return the number of bytes read or <code>-1</code>
>>>>> + * if the end of the stream has been reached
>>>>> + * @throws IOException if an I/O error occurs
>>>>> + */
>>>>> + @Override
>>>>> + public int read(byte[] b) throws IOException {
>>>>> + return read(b, 0, b.length);
>>>>> + }
>>>>> +
>>>>> + /**
>>>>> + * Read a single byte.
>>>>> + *
>>>>> + * @return either the byte read or <code>-1</code> if the end
>>>>> of the stream
>>>>> + * has been reached
>>>>> + * @throws IOException if an I/O error occurs
>>>>> + */
>>>>> + @Override
>>>>> + public int read() throws IOException {
>>>>> + for (;;) {
>>>>> + if (encoderOut.hasRemaining()) {
>>>>> + return encoderOut.get() & 0xFF;
>>>>> + } else {
>>>>> + fillBuffer();
>>>>> + if (endOfInput && !encoderOut.hasRemaining()) {
>>>>> + return -1;
>>>>> + }
>>>>> + }
>>>>> + }
>>>>> + }
>>>>> +
>>>>> + /**
>>>>> + * Close the stream. This method will cause the underlying
>>>>> {@link Reader}
>>>>> + * to be closed.
>>>>> + * @throws IOException if an I/O error occurs
>>>>> + */
>>>>> + @Override
>>>>> + public void close() throws IOException {
>>>>> + reader.close();
>>>>> + }
>>>>> +}
>>>>> diff --git
>>>>> a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>> b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>>
>>>>> index f77b5ed..b957f36 100644
>>>>> ---
>>>>> a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>>
>>>>> +++
>>>>> b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>>
>>>>> @@ -96,6 +96,25 @@ public class JAXRSJmsTest extends
>>>>> AbstractBusClientServerTestBase {
>>>>> assertEquals("Get a wrong response code.", 200,
>>>>> client.getResponse().getStatus());
>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>> }
>>>>> +
>>>>> + @Test
>>>>> + public void testGetBookFromWebClientWithTextJMSMessage()
>>>>> throws Exception {
>>>>> + // setup the the client
>>>>> + String endpointAddressUrlEncoded =
>>>>> "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>> + +
>>>>> "?replyToName=dynamicQueues/test.jmstransport.response"
>>>>> + +
>>>>> "&jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>
>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>> + + "&messageType=text";
>>>>> +
>>>>> + WebClient client =
>>>>> WebClient.create(endpointAddressUrlEncoded);
>>>>> + WebClient.getConfig(client).getInInterceptors().add(new
>>>>> LoggingInInterceptor());
>>>>> + WebClient.getConfig(client).getRequestContext()
>>>>> + .put(org.apache.cxf.message.Message.REQUEST_URI,
>>>>> "/bookstore/books/123");
>>>>> +
>>>>> + Book book = client.get(Book.class);
>>>>> + assertEquals("Get a wrong response code.", 200,
>>>>> client.getResponse().getStatus());
>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>> + }
>>>>> @Test
>>>>> public void testPutBookOneWayWithWebClient() throws Exception {
>>>>> @@ -130,6 +149,7 @@ public class JAXRSJmsTest extends
>>>>> AbstractBusClientServerTestBase {
>>>>> }
>>>>> }
>>>>> +
>>>>> @Test
>>>>> public void testGetBookFromWebClientWithPath() throws
>>>>> Exception {
>>>>> // setup the the client
>>>>> @@ -148,6 +168,25 @@ public class JAXRSJmsTest extends
>>>>> AbstractBusClientServerTestBase {
>>>>> }
>>>>> @Test
>>>>> + public void
>>>>> testGetBookFromWebClientWithPathWithTextJMSMessage() throws
>>>>> Exception {
>>>>> + // setup the the client
>>>>> + String endpointAddressUrlEncoded =
>>>>> "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>> + +
>>>>> "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>
>>>>> + +
>>>>> "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>> + + "&messageType=text";
>>>>> +
>>>>> + WebClient client =
>>>>> WebClient.create(endpointAddressUrlEncoded);
>>>>> + client.path("bookstore").path("books").path("123");
>>>>> +
>>>>> + Book book = client.get(Book.class);
>>>>> + assertEquals("Get a wrong response code.", 200,
>>>>> client.getResponse().getStatus());
>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>> + }
>>>>> +
>>>>> +
>>>>> + @Test
>>>>> public void testGetBookFromProxyClient() throws Exception {
>>>>> // setup the the client
>>>>> String endpointAddressUrlEncoded =
>>>>> "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>> @@ -161,6 +200,22 @@ public class JAXRSJmsTest extends
>>>>> AbstractBusClientServerTestBase {
>>>>> assertEquals("Get a wrong response code.", 200,
>>>>> WebClient.client(client).getResponse().getStatus());
>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>> }
>>>>> +
>>>>> + @Test
>>>>> + public void testGetBookFromProxyClientWithTextJMSMessage()
>>>>> throws Exception {
>>>>> + // setup the the client
>>>>> + String endpointAddressUrlEncoded =
>>>>> "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>> + +
>>>>> "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>
>>>>> + +
>>>>> "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>> + + "&messageType=text";
>>>>> +
>>>>> + JMSBookStore client =
>>>>> JAXRSClientFactory.create(endpointAddressUrlEncoded,
>>>>> JMSBookStore.class);
>>>>> + Book book = client.getBook("123");
>>>>> + assertEquals("Get a wrong response code.", 200,
>>>>> WebClient.client(client).getResponse().getStatus());
>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>> + }
>>>>> @Test
>>>>> public void testGetBookFromSubresourceProxyClient() throws
>>>>> Exception {
>>>>> @@ -177,6 +232,23 @@ public class JAXRSJmsTest extends
>>>>> AbstractBusClientServerTestBase {
>>>>> assertEquals("Get a wrong response code.", 200,
>>>>> WebClient.client(bookProxy).getResponse().getStatus());
>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>> }
>>>>> +
>>>>> + @Test
>>>>> + public void
>>>>> testGetBookFromSubresourceProxyClientWithTextJMSMessage() throws
>>>>> Exception {
>>>>> + // setup the the client
>>>>> + String endpointAddressUrlEncoded =
>>>>> "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>> + +
>>>>> "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>
>>>>> + +
>>>>> "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>> + + "&messageType=text";
>>>>> +
>>>>> + JMSBookStore client =
>>>>> JAXRSClientFactory.create(endpointAddressUrlEncoded,
>>>>> JMSBookStore.class);
>>>>> + Book bookProxy = client.getBookSubResource("123");
>>>>> + Book book = bookProxy.retrieveState();
>>>>> + assertEquals("Get a wrong response code.", 200,
>>>>> WebClient.client(bookProxy).getResponse().getStatus());
>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>> + }
>>>>> @Test
>>>>> public void testGetBookFromProxyClientWithQuery() throws
>>>>> Exception {
>>>>> @@ -193,6 +265,24 @@ public class JAXRSJmsTest extends
>>>>> AbstractBusClientServerTestBase {
>>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>>> }
>>>>> +
>>>>> + @Test
>>>>> + public void
>>>>> testGetBookFromProxyClientWithQueryWithTextJMSMessage() throws
>>>>> Exception {
>>>>> + // setup the the client
>>>>> + String endpointAddressUrlEncoded =
>>>>> "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>>> + +
>>>>> "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>>
>>>>> + +
>>>>> "&replyToName=dynamicQueues/test.jmstransport.response"
>>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>>> + + "&messageType=text";
>>>>> +
>>>>> + JMSBookStore client =
>>>>> JAXRSClientFactory.create(endpointAddressUrlEncoded,
>>>>> JMSBookStore.class);
>>>>> + Book book = client.getBookByURLQuery(new String[] {"1",
>>>>> "2", "3"});
>>>>> + assertEquals("Get a wrong response code.", 200,
>>>>> WebClient.client(client).getResponse().getStatus());
>>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>>> + }
>>>>> +
>>>>> +
>>>>> @Test
>>>>> public void testGetBook() throws Exception {
>>>>> Context ctx = getContext();
>>>>>
>>>
>>
--
Sergey Beryozkin
Talend Community Coders
http://coders.talend.com/
Re: [cxf] branch master updated: [CXF-7532]REST on JMS transport
can't handle the request message with text messageType
Posted by Sergey Beryozkin <sb...@gmail.com>.
Hi Freeman
Reader may likely need to be set too for SOAP, and I recall checking
Reader somewhere else too (will double check now), but I thought may be
the transport can set both Reader and InputStream, hmm, that is an extra
instance though, may be lets start with moving ReaderInputStream to the
core first ?
Sergey
On 19/10/17 09:59, ffang wrote:
> Yeah, set ReaderInputStream instead of Reader for the JMS transport is a
> better solution. If no objection I will go this way.
> －－－－－－－－－－－－－
> Freeman(Yue) Fang
>
> Red Hat, Inc.
> FuseSource is now part of Red Hat
>
>
>
>> On Oct 19, 2017, at 4:50 PM, Sergey Beryozkin <sberyozkin@gmail.com
>> <ma...@gmail.com>> wrote:
>>
>> The other thing is, should JMS transport itself set ReaderInputStream
>> as InputStream on the message, as opposed to patching the JAX-RS code
>> in various places ?
>>
>> Sergey
>> On 19/10/17 09:44, Sergey Beryozkin wrote:
>>> Hi Freeman
>>> Can you move ReaderInputStream to the CXF core where
>>> CachedInputStream and other IO utility code is located ?
>>> Please also update the code which gets Reader.class and check it for
>>> null first, and also drop a comment re the JMS transport, checking
>>> the reader can make sense in all the cases...
>>> Thanks, Sergey
>>> On 19/10/17 04:16, ffang@apache.org <ma...@apache.org> wrote:
>>>> This is an automated email from the ASF dual-hosted git repository.
>>>>
>>>> ffang pushed a commit to branch master
>>>> in repository https://gitbox.apache.org/repos/asf/cxf.git
>>>>
>>>>
>>>> The following commit(s) were added to refs/heads/master by this push:
>>>> new 1d66475 [CXF-7532]REST on JMS transport can't handle the
>>>> request message with text messageType
>>>> 1d66475 is described below
>>>>
>>>> commit 1d66475b4dea2a0031dcb1765e8e9fd08430ec9b
>>>> Author: Freeman Fang <freeman.fang@gmail.com
>>>> <ma...@gmail.com>>
>>>> AuthorDate: Thu Oct 19 11:15:50 2017 +0800
>>>>
>>>> [CXF-7532]REST on JMS transport can't handle the request
>>>> message with text messageType
>>>> ---
>>>> .../org/apache/cxf/jaxrs/impl/ResponseImpl.java | 5 +
>>>> .../org/apache/cxf/jaxrs/utils/JAXRSUtils.java | 6 +-
>>>> .../apache/cxf/jaxrs/utils/ReaderInputStream.java | 296
>>>> +++++++++++++++++++++
>>>> .../apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java | 90 +++++++
>>>> 4 files changed, 396 insertions(+), 1 deletion(-)
>>>>
>>>> diff --git
>>>> a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>> b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>
>>>> index e3a3acf..82f183a 100644
>>>> ---
>>>> a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>
>>>> +++
>>>> b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>>>
>>>> @@ -58,6 +58,7 @@ import org.apache.cxf.jaxrs.provider.ProviderFactory;
>>>> import org.apache.cxf.jaxrs.utils.HttpUtils;
>>>> import org.apache.cxf.jaxrs.utils.InjectionUtils;
>>>> import org.apache.cxf.jaxrs.utils.JAXRSUtils;
>>>> +import org.apache.cxf.jaxrs.utils.ReaderInputStream;
>>>> import org.apache.cxf.message.Message;
>>>> import org.apache.cxf.message.MessageUtils;
>>>> @@ -342,6 +343,10 @@ public final class ResponseImpl extends Response {
>>>> entityStreamAvailable = entityStream != null;
>>>> } else if (entity instanceof InputStream) {
>>>> entityStream = InputStream.class.cast(entity);
>>>> + } else {
>>>> + Message inMessage = getResponseMessage();
>>>> + Reader reader = inMessage.getContent(Reader.class);
>>>> + entityStream = InputStream.class.cast(new
>>>> ReaderInputStream(reader));
>>>> }
>>>> // we need to check for readers even if no IS is set - the
>>>> readers may still do it
>>>> diff --git
>>>> a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>> b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>
>>>> index 0ef6565..509f76d 100644
>>>> ---
>>>> a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>
>>>> +++
>>>> b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>>>
>>>> @@ -832,7 +832,11 @@ public final class JAXRSUtils {
>>>> OperationResourceInfo ori)
>>>> throws IOException, WebApplicationException {
>>>> InputStream is = message.getContent(InputStream.class);
>>>> -
>>>> + if (is == null) {
>>>> + //may use the jms transport so check the Reader;
>>>> + Reader reader = message.getContent(Reader.class);
>>>> + is = new ReaderInputStream(reader);
>>>> + }
>>>> if (parameter.getType() == ParameterType.REQUEST_BODY) {
>>>> if (parameterClass == AsyncResponse.class) {
>>>> diff --git
>>>> a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
>>>> b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
>>>>
>>>> new file mode 100644
>>>> index 0000000..c7b142e
>>>> --- /dev/null
>>>> +++
>>>> b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
>>>>
>>>> @@ -0,0 +1,296 @@
>>>> +/**
>>>> + * Licensed to the Apache Software Foundation (ASF) under one
>>>> + * or more contributor license agreements. See the NOTICE file
>>>> + * distributed with this work for additional information
>>>> + * regarding copyright ownership. The ASF licenses this file
>>>> + * to you 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.apache.cxf.jaxrs.utils;
>>>> +
>>>> +import java.io.IOException;
>>>> +import java.io.InputStream;
>>>> +import java.io.Reader;
>>>> +import java.nio.ByteBuffer;
>>>> +import java.nio.CharBuffer;
>>>> +import java.nio.charset.Charset;
>>>> +import java.nio.charset.CharsetEncoder;
>>>> +import java.nio.charset.CoderResult;
>>>> +import java.nio.charset.CodingErrorAction;
>>>> +
>>>> +/**
>>>> + * {@link InputStream} implementation that reads a character stream
>>>> from a {@link Reader}
>>>> + * and transforms it to a byte stream using a specified charset
>>>> encoding. The stream
>>>> + * is transformed using a {@link CharsetEncoder} object,
>>>> guaranteeing that all charset
>>>> + * encodings supported by the JRE are handled correctly. In
>>>> particular for charsets such as
>>>> + * UTF-16, the implementation ensures that one and only one byte
>>>> order marker
>>>> + * is produced.
>>>> + * <p>
>>>> + * Since in general it is not possible to predict the number of
>>>> characters to be read from the
>>>> + * {@link Reader} to satisfy a read request on the {@link
>>>> ReaderInputStream}, all reads from
>>>> + * the {@link Reader} are buffered. There is therefore no well
>>>> defined correlation
>>>> + * between the current position of the {@link Reader} and that of
>>>> the {@link ReaderInputStream}.
>>>> + * This also implies that in general there is no need to wrap the
>>>> underlying {@link Reader}
>>>> + * in a {@link java.io.BufferedReader}.
>>>> + * <p>
>>>> + * {@link ReaderInputStream} implements the inverse transformation
>>>> of {@link java.io.InputStreamReader};
>>>> + * in the following example, reading from <tt>in2</tt> would return
>>>> the same byte
>>>> + * sequence as reading from <tt>in</tt> (provided that the initial
>>>> byte sequence is legal
>>>> + * with respect to the charset encoding):
>>>> + * <pre>
>>>> + * InputStream in = ...
>>>> + * Charset cs = ...
>>>> + * InputStreamReader reader = new InputStreamReader(in, cs);
>>>> + * ReaderInputStream in2 = new ReaderInputStream(reader, cs);</pre>
>>>> + * {@link ReaderInputStream} implements the same transformation as
>>>> {@link java.io.OutputStreamWriter},
>>>> + * except that the control flow is reversed: both classes transform
>>>> a character stream
>>>> + * into a byte stream, but {@link java.io.OutputStreamWriter}
>>>> pushes data to the underlying stream,
>>>> + * while {@link ReaderInputStream} pulls it from the underlying stream.
>>>> + * <p>
>>>> + * Note that while there are use cases where there is no
>>>> alternative to using
>>>> + * this class, very often the need to use this class is an
>>>> indication of a flaw
>>>> + * in the design of the code. This class is typically used in
>>>> situations where an existing
>>>> + * API only accepts an {@link InputStream}, but where the most
>>>> natural way to produce the data
>>>> + * is as a character stream, i.e. by providing a {@link Reader}
>>>> instance. An example of a situation
>>>> + * where this problem may appear is when implementing the {@link
>>>> javax.activation.DataSource}
>>>> + * interface from the Java Activation Framework.
>>>> + * <p>
>>>> + * Given the fact that the {@link Reader} class doesn't provide any
>>>> way to predict whether the next
>>>> + * read operation will block or not, it is not possible to provide
>>>> a meaningful
>>>> + * implementation of the {@link InputStream#available()} method. A
>>>> call to this method
>>>> + * will always return 0. Also, this class doesn't support {@link
>>>> InputStream#mark(int)}.
>>>> + * <p>
>>>> + * Instances of {@link ReaderInputStream} are not thread safe.
>>>> + *
>>>> + * @see org.apache.commons.io.output.WriterOutputStream
>>>> + *
>>>> + * @since 2.0
>>>> + */
>>>> +public class ReaderInputStream extends InputStream {
>>>> + private static final int DEFAULT_BUFFER_SIZE = 1024;
>>>> +
>>>> + private final Reader reader;
>>>> + private final CharsetEncoder encoder;
>>>> +
>>>> + /**
>>>> + * CharBuffer used as input for the decoder. It should be
>>>> reasonably
>>>> + * large as we read data from the underlying Reader into this
>>>> buffer.
>>>> + */
>>>> + private final CharBuffer encoderIn;
>>>> +
>>>> + /**
>>>> + * ByteBuffer used as output for the decoder. This buffer can
>>>> be small
>>>> + * as it is only used to transfer data from the decoder to the
>>>> + * buffer provided by the caller.
>>>> + */
>>>> + private final ByteBuffer encoderOut;
>>>> +
>>>> + private CoderResult lastCoderResult;
>>>> + private boolean endOfInput;
>>>> +
>>>> + /**
>>>> + * Construct a new {@link ReaderInputStream}.
>>>> + *
>>>> + * @param reader the target {@link Reader}
>>>> + * @param encoder the charset encoder
>>>> + * @since 2.1
>>>> + */
>>>> + public ReaderInputStream(Reader reader, CharsetEncoder encoder) {
>>>> + this(reader, encoder, DEFAULT_BUFFER_SIZE);
>>>> + }
>>>> +
>>>> + /**
>>>> + * Construct a new {@link ReaderInputStream}.
>>>> + *
>>>> + * @param reader the target {@link Reader}
>>>> + * @param encoder the charset encoder
>>>> + * @param bufferSize the size of the input buffer in number of
>>>> characters
>>>> + * @since 2.1
>>>> + */
>>>> + public ReaderInputStream(Reader reader, CharsetEncoder encoder,
>>>> int bufferSize) {
>>>> + this.reader = reader;
>>>> + this.encoder = encoder;
>>>> + this.encoderIn = CharBuffer.allocate(bufferSize);
>>>> + this.encoderIn.flip();
>>>> + this.encoderOut = ByteBuffer.allocate(128);
>>>> + this.encoderOut.flip();
>>>> + }
>>>> +
>>>> + /**
>>>> + * Construct a new {@link ReaderInputStream}.
>>>> + *
>>>> + * @param reader the target {@link Reader}
>>>> + * @param charset the charset encoding
>>>> + * @param bufferSize the size of the input buffer in number of
>>>> characters
>>>> + */
>>>> + public ReaderInputStream(Reader reader, Charset charset, int
>>>> bufferSize) {
>>>> + this(reader,
>>>> + charset.newEncoder()
>>>> + .onMalformedInput(CodingErrorAction.REPLACE)
>>>> + .onUnmappableCharacter(CodingErrorAction.REPLACE),
>>>> + bufferSize);
>>>> + }
>>>> +
>>>> + /**
>>>> + * Construct a new {@link ReaderInputStream} with a default
>>>> input buffer size of
>>>> + * 1024 characters.
>>>> + *
>>>> + * @param reader the target {@link Reader}
>>>> + * @param charset the charset encoding
>>>> + */
>>>> + public ReaderInputStream(Reader reader, Charset charset) {
>>>> + this(reader, charset, DEFAULT_BUFFER_SIZE);
>>>> + }
>>>> +
>>>> + /**
>>>> + * Construct a new {@link ReaderInputStream}.
>>>> + *
>>>> + * @param reader the target {@link Reader}
>>>> + * @param charsetName the name of the charset encoding
>>>> + * @param bufferSize the size of the input buffer in number of
>>>> characters
>>>> + */
>>>> + public ReaderInputStream(Reader reader, String charsetName, int
>>>> bufferSize) {
>>>> + this(reader, Charset.forName(charsetName), bufferSize);
>>>> + }
>>>> +
>>>> + /**
>>>> + * Construct a new {@link ReaderInputStream} with a default
>>>> input buffer size of
>>>> + * 1024 characters.
>>>> + *
>>>> + * @param reader the target {@link Reader}
>>>> + * @param charsetName the name of the charset encoding
>>>> + */
>>>> + public ReaderInputStream(Reader reader, String charsetName) {
>>>> + this(reader, charsetName, DEFAULT_BUFFER_SIZE);
>>>> + }
>>>> +
>>>> + /**
>>>> + * Construct a new {@link ReaderInputStream} that uses the
>>>> default character encoding
>>>> + * with a default input buffer size of 1024 characters.
>>>> + *
>>>> + * @param reader the target {@link Reader}
>>>> + */
>>>> + public ReaderInputStream(Reader reader) {
>>>> + this(reader, Charset.defaultCharset());
>>>> + }
>>>> +
>>>> + /**
>>>> + * Fills the internal char buffer from the reader.
>>>> + *
>>>> + * @throws IOException
>>>> + * If an I/O error occurs
>>>> + */
>>>> + private void fillBuffer() throws IOException {
>>>> + if (!endOfInput && (lastCoderResult == null ||
>>>> lastCoderResult.isUnderflow())) {
>>>> + encoderIn.compact();
>>>> + int position = encoderIn.position();
>>>> + // We don't use Reader#read(CharBuffer) here because it
>>>> is more efficient
>>>> + // to write directly to the underlying char array (the
>>>> default implementation
>>>> + // copies data to a temporary char array).
>>>> + int c = reader.read(encoderIn.array(), position,
>>>> encoderIn.remaining());
>>>> + if (c == -1) {
>>>> + endOfInput = true;
>>>> + } else {
>>>> + encoderIn.position(position + c);
>>>> + }
>>>> + encoderIn.flip();
>>>> + }
>>>> + encoderOut.compact();
>>>> + lastCoderResult = encoder.encode(encoderIn, encoderOut,
>>>> endOfInput);
>>>> + encoderOut.flip();
>>>> + }
>>>> +
>>>> + /**
>>>> + * Read the specified number of bytes into an array.
>>>> + *
>>>> + * @param b the byte array to read into
>>>> + * @param off the offset to start reading bytes into
>>>> + * @param len the number of bytes to read
>>>> + * @return the number of bytes read or <code>-1</code>
>>>> + * if the end of the stream has been reached
>>>> + * @throws IOException if an I/O error occurs
>>>> + */
>>>> + @Override
>>>> + public int read(byte[] b, int off, int len) throws IOException {
>>>> + if (b == null) {
>>>> + throw new NullPointerException("Byte array must not be
>>>> null");
>>>> + }
>>>> + if (len < 0 || off < 0 || (off + len) > b.length) {
>>>> + throw new IndexOutOfBoundsException("Array Size=" +
>>>> b.length
>>>> + + ", offset=" + off + ", length=" + len);
>>>> + }
>>>> + int read = 0;
>>>> + if (len == 0) {
>>>> + return 0; // Always return 0 if len == 0
>>>> + }
>>>> + while (len > 0) {
>>>> + if (encoderOut.hasRemaining()) {
>>>> + int c = Math.min(encoderOut.remaining(), len);
>>>> + encoderOut.get(b, off, c);
>>>> + off += c;
>>>> + len -= c;
>>>> + read += c;
>>>> + } else {
>>>> + fillBuffer();
>>>> + if (endOfInput && !encoderOut.hasRemaining()) {
>>>> + break;
>>>> + }
>>>> + }
>>>> + }
>>>> + return read == 0 && endOfInput ? -1 : read;
>>>> + }
>>>> +
>>>> + /**
>>>> + * Read the specified number of bytes into an array.
>>>> + *
>>>> + * @param b the byte array to read into
>>>> + * @return the number of bytes read or <code>-1</code>
>>>> + * if the end of the stream has been reached
>>>> + * @throws IOException if an I/O error occurs
>>>> + */
>>>> + @Override
>>>> + public int read(byte[] b) throws IOException {
>>>> + return read(b, 0, b.length);
>>>> + }
>>>> +
>>>> + /**
>>>> + * Read a single byte.
>>>> + *
>>>> + * @return either the byte read or <code>-1</code> if the end
>>>> of the stream
>>>> + * has been reached
>>>> + * @throws IOException if an I/O error occurs
>>>> + */
>>>> + @Override
>>>> + public int read() throws IOException {
>>>> + for (;;) {
>>>> + if (encoderOut.hasRemaining()) {
>>>> + return encoderOut.get() & 0xFF;
>>>> + } else {
>>>> + fillBuffer();
>>>> + if (endOfInput && !encoderOut.hasRemaining()) {
>>>> + return -1;
>>>> + }
>>>> + }
>>>> + }
>>>> + }
>>>> +
>>>> + /**
>>>> + * Close the stream. This method will cause the underlying
>>>> {@link Reader}
>>>> + * to be closed.
>>>> + * @throws IOException if an I/O error occurs
>>>> + */
>>>> + @Override
>>>> + public void close() throws IOException {
>>>> + reader.close();
>>>> + }
>>>> +}
>>>> diff --git
>>>> a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>> b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>
>>>> index f77b5ed..b957f36 100644
>>>> ---
>>>> a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>
>>>> +++
>>>> b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>>>
>>>> @@ -96,6 +96,25 @@ public class JAXRSJmsTest extends
>>>> AbstractBusClientServerTestBase {
>>>> assertEquals("Get a wrong response code.", 200,
>>>> client.getResponse().getStatus());
>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>> }
>>>> +
>>>> + @Test
>>>> + public void testGetBookFromWebClientWithTextJMSMessage() throws
>>>> Exception {
>>>> + // setup the the client
>>>> + String endpointAddressUrlEncoded =
>>>> "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>> + + "?replyToName=dynamicQueues/test.jmstransport.response"
>>>> + +
>>>> "&jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>
>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>> + + "&messageType=text";
>>>> +
>>>> + WebClient client = WebClient.create(endpointAddressUrlEncoded);
>>>> + WebClient.getConfig(client).getInInterceptors().add(new
>>>> LoggingInInterceptor());
>>>> + WebClient.getConfig(client).getRequestContext()
>>>> + .put(org.apache.cxf.message.Message.REQUEST_URI,
>>>> "/bookstore/books/123");
>>>> +
>>>> + Book book = client.get(Book.class);
>>>> + assertEquals("Get a wrong response code.", 200,
>>>> client.getResponse().getStatus());
>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>> + }
>>>> @Test
>>>> public void testPutBookOneWayWithWebClient() throws Exception {
>>>> @@ -130,6 +149,7 @@ public class JAXRSJmsTest extends
>>>> AbstractBusClientServerTestBase {
>>>> }
>>>> }
>>>> +
>>>> @Test
>>>> public void testGetBookFromWebClientWithPath() throws Exception {
>>>> // setup the the client
>>>> @@ -148,6 +168,25 @@ public class JAXRSJmsTest extends
>>>> AbstractBusClientServerTestBase {
>>>> }
>>>> @Test
>>>> + public void
>>>> testGetBookFromWebClientWithPathWithTextJMSMessage() throws Exception {
>>>> + // setup the the client
>>>> + String endpointAddressUrlEncoded =
>>>> "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>> + +
>>>> "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>
>>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>> + + "&messageType=text";
>>>> +
>>>> + WebClient client = WebClient.create(endpointAddressUrlEncoded);
>>>> + client.path("bookstore").path("books").path("123");
>>>> +
>>>> + Book book = client.get(Book.class);
>>>> + assertEquals("Get a wrong response code.", 200,
>>>> client.getResponse().getStatus());
>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>> + }
>>>> +
>>>> +
>>>> + @Test
>>>> public void testGetBookFromProxyClient() throws Exception {
>>>> // setup the the client
>>>> String endpointAddressUrlEncoded =
>>>> "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>> @@ -161,6 +200,22 @@ public class JAXRSJmsTest extends
>>>> AbstractBusClientServerTestBase {
>>>> assertEquals("Get a wrong response code.", 200,
>>>> WebClient.client(client).getResponse().getStatus());
>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>> }
>>>> +
>>>> + @Test
>>>> + public void testGetBookFromProxyClientWithTextJMSMessage()
>>>> throws Exception {
>>>> + // setup the the client
>>>> + String endpointAddressUrlEncoded =
>>>> "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>> + +
>>>> "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>
>>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>> + + "&messageType=text";
>>>> +
>>>> + JMSBookStore client =
>>>> JAXRSClientFactory.create(endpointAddressUrlEncoded,
>>>> JMSBookStore.class);
>>>> + Book book = client.getBook("123");
>>>> + assertEquals("Get a wrong response code.", 200,
>>>> WebClient.client(client).getResponse().getStatus());
>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>> + }
>>>> @Test
>>>> public void testGetBookFromSubresourceProxyClient() throws
>>>> Exception {
>>>> @@ -177,6 +232,23 @@ public class JAXRSJmsTest extends
>>>> AbstractBusClientServerTestBase {
>>>> assertEquals("Get a wrong response code.", 200,
>>>> WebClient.client(bookProxy).getResponse().getStatus());
>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>> }
>>>> +
>>>> + @Test
>>>> + public void
>>>> testGetBookFromSubresourceProxyClientWithTextJMSMessage() throws
>>>> Exception {
>>>> + // setup the the client
>>>> + String endpointAddressUrlEncoded =
>>>> "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>> + +
>>>> "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>
>>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>> + + "&messageType=text";
>>>> +
>>>> + JMSBookStore client =
>>>> JAXRSClientFactory.create(endpointAddressUrlEncoded,
>>>> JMSBookStore.class);
>>>> + Book bookProxy = client.getBookSubResource("123");
>>>> + Book book = bookProxy.retrieveState();
>>>> + assertEquals("Get a wrong response code.", 200,
>>>> WebClient.client(bookProxy).getResponse().getStatus());
>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>> + }
>>>> @Test
>>>> public void testGetBookFromProxyClientWithQuery() throws
>>>> Exception {
>>>> @@ -193,6 +265,24 @@ public class JAXRSJmsTest extends
>>>> AbstractBusClientServerTestBase {
>>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>>> }
>>>> +
>>>> + @Test
>>>> + public void
>>>> testGetBookFromProxyClientWithQueryWithTextJMSMessage() throws
>>>> Exception {
>>>> + // setup the the client
>>>> + String endpointAddressUrlEncoded =
>>>> "jms:jndi:dynamicQueues/test.jmstransport.text"
>>>> + +
>>>> "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>>>
>>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>>> + + "&messageType=text";
>>>> +
>>>> + JMSBookStore client =
>>>> JAXRSClientFactory.create(endpointAddressUrlEncoded,
>>>> JMSBookStore.class);
>>>> + Book book = client.getBookByURLQuery(new String[] {"1",
>>>> "2", "3"});
>>>> + assertEquals("Get a wrong response code.", 200,
>>>> WebClient.client(client).getResponse().getStatus());
>>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>>> + }
>>>> +
>>>> +
>>>> @Test
>>>> public void testGetBook() throws Exception {
>>>> Context ctx = getContext();
>>>>
>>
>
Re: [cxf] branch master updated: [CXF-7532]REST on JMS transport can't handle the request message with text messageType
Posted by ffang <fr...@gmail.com>.
Yeah, set ReaderInputStream instead of Reader for the JMS transport is a better solution. If no objection I will go this way.
－－－－－－－－－－－－－
Freeman(Yue) Fang
Red Hat, Inc.
FuseSource is now part of Red Hat
> On Oct 19, 2017, at 4:50 PM, Sergey Beryozkin <sb...@gmail.com> wrote:
>
> The other thing is, should JMS transport itself set ReaderInputStream as InputStream on the message, as opposed to patching the JAX-RS code in various places ?
>
> Sergey
> On 19/10/17 09:44, Sergey Beryozkin wrote:
>> Hi Freeman
>> Can you move ReaderInputStream to the CXF core where CachedInputStream and other IO utility code is located ?
>> Please also update the code which gets Reader.class and check it for null first, and also drop a comment re the JMS transport, checking the reader can make sense in all the cases...
>> Thanks, Sergey
>> On 19/10/17 04:16, ffang@apache.org wrote:
>>> This is an automated email from the ASF dual-hosted git repository.
>>>
>>> ffang pushed a commit to branch master
>>> in repository https://gitbox.apache.org/repos/asf/cxf.git
>>>
>>>
>>> The following commit(s) were added to refs/heads/master by this push:
>>> new 1d66475 [CXF-7532]REST on JMS transport can't handle the request message with text messageType
>>> 1d66475 is described below
>>>
>>> commit 1d66475b4dea2a0031dcb1765e8e9fd08430ec9b
>>> Author: Freeman Fang <fr...@gmail.com>
>>> AuthorDate: Thu Oct 19 11:15:50 2017 +0800
>>>
>>> [CXF-7532]REST on JMS transport can't handle the request message with text messageType
>>> ---
>>> .../org/apache/cxf/jaxrs/impl/ResponseImpl.java | 5 +
>>> .../org/apache/cxf/jaxrs/utils/JAXRSUtils.java | 6 +-
>>> .../apache/cxf/jaxrs/utils/ReaderInputStream.java | 296 +++++++++++++++++++++
>>> .../apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java | 90 +++++++
>>> 4 files changed, 396 insertions(+), 1 deletion(-)
>>>
>>> diff --git a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>> index e3a3acf..82f183a 100644
>>> --- a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>> +++ b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>> @@ -58,6 +58,7 @@ import org.apache.cxf.jaxrs.provider.ProviderFactory;
>>> import org.apache.cxf.jaxrs.utils.HttpUtils;
>>> import org.apache.cxf.jaxrs.utils.InjectionUtils;
>>> import org.apache.cxf.jaxrs.utils.JAXRSUtils;
>>> +import org.apache.cxf.jaxrs.utils.ReaderInputStream;
>>> import org.apache.cxf.message.Message;
>>> import org.apache.cxf.message.MessageUtils;
>>> @@ -342,6 +343,10 @@ public final class ResponseImpl extends Response {
>>> entityStreamAvailable = entityStream != null;
>>> } else if (entity instanceof InputStream) {
>>> entityStream = InputStream.class.cast(entity);
>>> + } else {
>>> + Message inMessage = getResponseMessage();
>>> + Reader reader = inMessage.getContent(Reader.class);
>>> + entityStream = InputStream.class.cast(new ReaderInputStream(reader));
>>> }
>>> // we need to check for readers even if no IS is set - the readers may still do it
>>> diff --git a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>> index 0ef6565..509f76d 100644
>>> --- a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>> +++ b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>> @@ -832,7 +832,11 @@ public final class JAXRSUtils {
>>> OperationResourceInfo ori)
>>> throws IOException, WebApplicationException {
>>> InputStream is = message.getContent(InputStream.class);
>>> -
>>> + if (is == null) {
>>> + //may use the jms transport so check the Reader;
>>> + Reader reader = message.getContent(Reader.class);
>>> + is = new ReaderInputStream(reader);
>>> + }
>>> if (parameter.getType() == ParameterType.REQUEST_BODY) {
>>> if (parameterClass == AsyncResponse.class) {
>>> diff --git a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
>>> new file mode 100644
>>> index 0000000..c7b142e
>>> --- /dev/null
>>> +++ b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
>>> @@ -0,0 +1,296 @@
>>> +/**
>>> + * Licensed to the Apache Software Foundation (ASF) under one
>>> + * or more contributor license agreements. See the NOTICE file
>>> + * distributed with this work for additional information
>>> + * regarding copyright ownership. The ASF licenses this file
>>> + * to you 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.apache.cxf.jaxrs.utils;
>>> +
>>> +import java.io.IOException;
>>> +import java.io.InputStream;
>>> +import java.io.Reader;
>>> +import java.nio.ByteBuffer;
>>> +import java.nio.CharBuffer;
>>> +import java.nio.charset.Charset;
>>> +import java.nio.charset.CharsetEncoder;
>>> +import java.nio.charset.CoderResult;
>>> +import java.nio.charset.CodingErrorAction;
>>> +
>>> +/**
>>> + * {@link InputStream} implementation that reads a character stream from a {@link Reader}
>>> + * and transforms it to a byte stream using a specified charset encoding. The stream
>>> + * is transformed using a {@link CharsetEncoder} object, guaranteeing that all charset
>>> + * encodings supported by the JRE are handled correctly. In particular for charsets such as
>>> + * UTF-16, the implementation ensures that one and only one byte order marker
>>> + * is produced.
>>> + * <p>
>>> + * Since in general it is not possible to predict the number of characters to be read from the
>>> + * {@link Reader} to satisfy a read request on the {@link ReaderInputStream}, all reads from
>>> + * the {@link Reader} are buffered. There is therefore no well defined correlation
>>> + * between the current position of the {@link Reader} and that of the {@link ReaderInputStream}.
>>> + * This also implies that in general there is no need to wrap the underlying {@link Reader}
>>> + * in a {@link java.io.BufferedReader}.
>>> + * <p>
>>> + * {@link ReaderInputStream} implements the inverse transformation of {@link java.io.InputStreamReader};
>>> + * in the following example, reading from <tt>in2</tt> would return the same byte
>>> + * sequence as reading from <tt>in</tt> (provided that the initial byte sequence is legal
>>> + * with respect to the charset encoding):
>>> + * <pre>
>>> + * InputStream in = ...
>>> + * Charset cs = ...
>>> + * InputStreamReader reader = new InputStreamReader(in, cs);
>>> + * ReaderInputStream in2 = new ReaderInputStream(reader, cs);</pre>
>>> + * {@link ReaderInputStream} implements the same transformation as {@link java.io.OutputStreamWriter},
>>> + * except that the control flow is reversed: both classes transform a character stream
>>> + * into a byte stream, but {@link java.io.OutputStreamWriter} pushes data to the underlying stream,
>>> + * while {@link ReaderInputStream} pulls it from the underlying stream.
>>> + * <p>
>>> + * Note that while there are use cases where there is no alternative to using
>>> + * this class, very often the need to use this class is an indication of a flaw
>>> + * in the design of the code. This class is typically used in situations where an existing
>>> + * API only accepts an {@link InputStream}, but where the most natural way to produce the data
>>> + * is as a character stream, i.e. by providing a {@link Reader} instance. An example of a situation
>>> + * where this problem may appear is when implementing the {@link javax.activation.DataSource}
>>> + * interface from the Java Activation Framework.
>>> + * <p>
>>> + * Given the fact that the {@link Reader} class doesn't provide any way to predict whether the next
>>> + * read operation will block or not, it is not possible to provide a meaningful
>>> + * implementation of the {@link InputStream#available()} method. A call to this method
>>> + * will always return 0. Also, this class doesn't support {@link InputStream#mark(int)}.
>>> + * <p>
>>> + * Instances of {@link ReaderInputStream} are not thread safe.
>>> + *
>>> + * @see org.apache.commons.io.output.WriterOutputStream
>>> + *
>>> + * @since 2.0
>>> + */
>>> +public class ReaderInputStream extends InputStream {
>>> + private static final int DEFAULT_BUFFER_SIZE = 1024;
>>> +
>>> + private final Reader reader;
>>> + private final CharsetEncoder encoder;
>>> +
>>> + /**
>>> + * CharBuffer used as input for the decoder. It should be reasonably
>>> + * large as we read data from the underlying Reader into this buffer.
>>> + */
>>> + private final CharBuffer encoderIn;
>>> +
>>> + /**
>>> + * ByteBuffer used as output for the decoder. This buffer can be small
>>> + * as it is only used to transfer data from the decoder to the
>>> + * buffer provided by the caller.
>>> + */
>>> + private final ByteBuffer encoderOut;
>>> +
>>> + private CoderResult lastCoderResult;
>>> + private boolean endOfInput;
>>> +
>>> + /**
>>> + * Construct a new {@link ReaderInputStream}.
>>> + *
>>> + * @param reader the target {@link Reader}
>>> + * @param encoder the charset encoder
>>> + * @since 2.1
>>> + */
>>> + public ReaderInputStream(Reader reader, CharsetEncoder encoder) {
>>> + this(reader, encoder, DEFAULT_BUFFER_SIZE);
>>> + }
>>> +
>>> + /**
>>> + * Construct a new {@link ReaderInputStream}.
>>> + *
>>> + * @param reader the target {@link Reader}
>>> + * @param encoder the charset encoder
>>> + * @param bufferSize the size of the input buffer in number of characters
>>> + * @since 2.1
>>> + */
>>> + public ReaderInputStream(Reader reader, CharsetEncoder encoder, int bufferSize) {
>>> + this.reader = reader;
>>> + this.encoder = encoder;
>>> + this.encoderIn = CharBuffer.allocate(bufferSize);
>>> + this.encoderIn.flip();
>>> + this.encoderOut = ByteBuffer.allocate(128);
>>> + this.encoderOut.flip();
>>> + }
>>> +
>>> + /**
>>> + * Construct a new {@link ReaderInputStream}.
>>> + *
>>> + * @param reader the target {@link Reader}
>>> + * @param charset the charset encoding
>>> + * @param bufferSize the size of the input buffer in number of characters
>>> + */
>>> + public ReaderInputStream(Reader reader, Charset charset, int bufferSize) {
>>> + this(reader,
>>> + charset.newEncoder()
>>> + .onMalformedInput(CodingErrorAction.REPLACE)
>>> + .onUnmappableCharacter(CodingErrorAction.REPLACE),
>>> + bufferSize);
>>> + }
>>> +
>>> + /**
>>> + * Construct a new {@link ReaderInputStream} with a default input buffer size of
>>> + * 1024 characters.
>>> + *
>>> + * @param reader the target {@link Reader}
>>> + * @param charset the charset encoding
>>> + */
>>> + public ReaderInputStream(Reader reader, Charset charset) {
>>> + this(reader, charset, DEFAULT_BUFFER_SIZE);
>>> + }
>>> +
>>> + /**
>>> + * Construct a new {@link ReaderInputStream}.
>>> + *
>>> + * @param reader the target {@link Reader}
>>> + * @param charsetName the name of the charset encoding
>>> + * @param bufferSize the size of the input buffer in number of characters
>>> + */
>>> + public ReaderInputStream(Reader reader, String charsetName, int bufferSize) {
>>> + this(reader, Charset.forName(charsetName), bufferSize);
>>> + }
>>> +
>>> + /**
>>> + * Construct a new {@link ReaderInputStream} with a default input buffer size of
>>> + * 1024 characters.
>>> + *
>>> + * @param reader the target {@link Reader}
>>> + * @param charsetName the name of the charset encoding
>>> + */
>>> + public ReaderInputStream(Reader reader, String charsetName) {
>>> + this(reader, charsetName, DEFAULT_BUFFER_SIZE);
>>> + }
>>> +
>>> + /**
>>> + * Construct a new {@link ReaderInputStream} that uses the default character encoding
>>> + * with a default input buffer size of 1024 characters.
>>> + *
>>> + * @param reader the target {@link Reader}
>>> + */
>>> + public ReaderInputStream(Reader reader) {
>>> + this(reader, Charset.defaultCharset());
>>> + }
>>> +
>>> + /**
>>> + * Fills the internal char buffer from the reader.
>>> + *
>>> + * @throws IOException
>>> + * If an I/O error occurs
>>> + */
>>> + private void fillBuffer() throws IOException {
>>> + if (!endOfInput && (lastCoderResult == null || lastCoderResult.isUnderflow())) {
>>> + encoderIn.compact();
>>> + int position = encoderIn.position();
>>> + // We don't use Reader#read(CharBuffer) here because it is more efficient
>>> + // to write directly to the underlying char array (the default implementation
>>> + // copies data to a temporary char array).
>>> + int c = reader.read(encoderIn.array(), position, encoderIn.remaining());
>>> + if (c == -1) {
>>> + endOfInput = true;
>>> + } else {
>>> + encoderIn.position(position + c);
>>> + }
>>> + encoderIn.flip();
>>> + }
>>> + encoderOut.compact();
>>> + lastCoderResult = encoder.encode(encoderIn, encoderOut, endOfInput);
>>> + encoderOut.flip();
>>> + }
>>> +
>>> + /**
>>> + * Read the specified number of bytes into an array.
>>> + *
>>> + * @param b the byte array to read into
>>> + * @param off the offset to start reading bytes into
>>> + * @param len the number of bytes to read
>>> + * @return the number of bytes read or <code>-1</code>
>>> + * if the end of the stream has been reached
>>> + * @throws IOException if an I/O error occurs
>>> + */
>>> + @Override
>>> + public int read(byte[] b, int off, int len) throws IOException {
>>> + if (b == null) {
>>> + throw new NullPointerException("Byte array must not be null");
>>> + }
>>> + if (len < 0 || off < 0 || (off + len) > b.length) {
>>> + throw new IndexOutOfBoundsException("Array Size=" + b.length
>>> + + ", offset=" + off + ", length=" + len);
>>> + }
>>> + int read = 0;
>>> + if (len == 0) {
>>> + return 0; // Always return 0 if len == 0
>>> + }
>>> + while (len > 0) {
>>> + if (encoderOut.hasRemaining()) {
>>> + int c = Math.min(encoderOut.remaining(), len);
>>> + encoderOut.get(b, off, c);
>>> + off += c;
>>> + len -= c;
>>> + read += c;
>>> + } else {
>>> + fillBuffer();
>>> + if (endOfInput && !encoderOut.hasRemaining()) {
>>> + break;
>>> + }
>>> + }
>>> + }
>>> + return read == 0 && endOfInput ? -1 : read;
>>> + }
>>> +
>>> + /**
>>> + * Read the specified number of bytes into an array.
>>> + *
>>> + * @param b the byte array to read into
>>> + * @return the number of bytes read or <code>-1</code>
>>> + * if the end of the stream has been reached
>>> + * @throws IOException if an I/O error occurs
>>> + */
>>> + @Override
>>> + public int read(byte[] b) throws IOException {
>>> + return read(b, 0, b.length);
>>> + }
>>> +
>>> + /**
>>> + * Read a single byte.
>>> + *
>>> + * @return either the byte read or <code>-1</code> if the end of the stream
>>> + * has been reached
>>> + * @throws IOException if an I/O error occurs
>>> + */
>>> + @Override
>>> + public int read() throws IOException {
>>> + for (;;) {
>>> + if (encoderOut.hasRemaining()) {
>>> + return encoderOut.get() & 0xFF;
>>> + } else {
>>> + fillBuffer();
>>> + if (endOfInput && !encoderOut.hasRemaining()) {
>>> + return -1;
>>> + }
>>> + }
>>> + }
>>> + }
>>> +
>>> + /**
>>> + * Close the stream. This method will cause the underlying {@link Reader}
>>> + * to be closed.
>>> + * @throws IOException if an I/O error occurs
>>> + */
>>> + @Override
>>> + public void close() throws IOException {
>>> + reader.close();
>>> + }
>>> +}
>>> diff --git a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>> index f77b5ed..b957f36 100644
>>> --- a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>> +++ b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>> @@ -96,6 +96,25 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>> assertEquals("Get a wrong response code.", 200, client.getResponse().getStatus());
>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>> }
>>> +
>>> + @Test
>>> + public void testGetBookFromWebClientWithTextJMSMessage() throws Exception {
>>> + // setup the the client
>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>> + + "?replyToName=dynamicQueues/test.jmstransport.response"
>>> + + "&jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>> + + "&messageType=text";
>>> +
>>> + WebClient client = WebClient.create(endpointAddressUrlEncoded);
>>> + WebClient.getConfig(client).getInInterceptors().add(new LoggingInInterceptor());
>>> + WebClient.getConfig(client).getRequestContext()
>>> + .put(org.apache.cxf.message.Message.REQUEST_URI, "/bookstore/books/123");
>>> +
>>> + Book book = client.get(Book.class);
>>> + assertEquals("Get a wrong response code.", 200, client.getResponse().getStatus());
>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>> + }
>>> @Test
>>> public void testPutBookOneWayWithWebClient() throws Exception {
>>> @@ -130,6 +149,7 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>> }
>>> }
>>> +
>>> @Test
>>> public void testGetBookFromWebClientWithPath() throws Exception {
>>> // setup the the client
>>> @@ -148,6 +168,25 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>> }
>>> @Test
>>> + public void testGetBookFromWebClientWithPathWithTextJMSMessage() throws Exception {
>>> + // setup the the client
>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>> + + "&messageType=text";
>>> +
>>> + WebClient client = WebClient.create(endpointAddressUrlEncoded);
>>> + client.path("bookstore").path("books").path("123");
>>> +
>>> + Book book = client.get(Book.class);
>>> + assertEquals("Get a wrong response code.", 200, client.getResponse().getStatus());
>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>> + }
>>> +
>>> +
>>> + @Test
>>> public void testGetBookFromProxyClient() throws Exception {
>>> // setup the the client
>>> String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>> @@ -161,6 +200,22 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>> assertEquals("Get a wrong response code.", 200, WebClient.client(client).getResponse().getStatus());
>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>> }
>>> +
>>> + @Test
>>> + public void testGetBookFromProxyClientWithTextJMSMessage() throws Exception {
>>> + // setup the the client
>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>> + + "&messageType=text";
>>> +
>>> + JMSBookStore client = JAXRSClientFactory.create(endpointAddressUrlEncoded, JMSBookStore.class);
>>> + Book book = client.getBook("123");
>>> + assertEquals("Get a wrong response code.", 200, WebClient.client(client).getResponse().getStatus());
>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>> + }
>>> @Test
>>> public void testGetBookFromSubresourceProxyClient() throws Exception {
>>> @@ -177,6 +232,23 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>> assertEquals("Get a wrong response code.", 200, WebClient.client(bookProxy).getResponse().getStatus());
>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>> }
>>> +
>>> + @Test
>>> + public void testGetBookFromSubresourceProxyClientWithTextJMSMessage() throws Exception {
>>> + // setup the the client
>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>> + + "&messageType=text";
>>> +
>>> + JMSBookStore client = JAXRSClientFactory.create(endpointAddressUrlEncoded, JMSBookStore.class);
>>> + Book bookProxy = client.getBookSubResource("123");
>>> + Book book = bookProxy.retrieveState();
>>> + assertEquals("Get a wrong response code.", 200, WebClient.client(bookProxy).getResponse().getStatus());
>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>> + }
>>> @Test
>>> public void testGetBookFromProxyClientWithQuery() throws Exception {
>>> @@ -193,6 +265,24 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
>>> assertEquals("Get a wrong book id.", 123, book.getId());
>>> }
>>> +
>>> + @Test
>>> + public void testGetBookFromProxyClientWithQueryWithTextJMSMessage() throws Exception {
>>> + // setup the the client
>>> + String endpointAddressUrlEncoded = "jms:jndi:dynamicQueues/test.jmstransport.text"
>>> + + "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>>> + + "&messageType=text";
>>> +
>>> + JMSBookStore client = JAXRSClientFactory.create(endpointAddressUrlEncoded, JMSBookStore.class);
>>> + Book book = client.getBookByURLQuery(new String[] {"1", "2", "3"});
>>> + assertEquals("Get a wrong response code.", 200, WebClient.client(client).getResponse().getStatus());
>>> + assertEquals("Get a wrong book id.", 123, book.getId());
>>> + }
>>> +
>>> +
>>> @Test
>>> public void testGetBook() throws Exception {
>>> Context ctx = getContext();
>>>
>
Re: [cxf] branch master updated: [CXF-7532]REST on JMS transport
can't handle the request message with text messageType
Posted by Sergey Beryozkin <sb...@gmail.com>.
The other thing is, should JMS transport itself set ReaderInputStream as
InputStream on the message, as opposed to patching the JAX-RS code in
various places ?
Sergey
On 19/10/17 09:44, Sergey Beryozkin wrote:
> Hi Freeman
>
> Can you move ReaderInputStream to the CXF core where CachedInputStream
> and other IO utility code is located ?
>
> Please also update the code which gets Reader.class and check it for
> null first, and also drop a comment re the JMS transport, checking the
> reader can make sense in all the cases...
>
> Thanks, Sergey
> On 19/10/17 04:16, ffang@apache.org wrote:
>> This is an automated email from the ASF dual-hosted git repository.
>>
>> ffang pushed a commit to branch master
>> in repository https://gitbox.apache.org/repos/asf/cxf.git
>>
>>
>> The following commit(s) were added to refs/heads/master by this push:
>> new 1d66475 [CXF-7532]REST on JMS transport can't handle the
>> request message with text messageType
>> 1d66475 is described below
>>
>> commit 1d66475b4dea2a0031dcb1765e8e9fd08430ec9b
>> Author: Freeman Fang <fr...@gmail.com>
>> AuthorDate: Thu Oct 19 11:15:50 2017 +0800
>>
>> [CXF-7532]REST on JMS transport can't handle the request message
>> with text messageType
>> ---
>> .../org/apache/cxf/jaxrs/impl/ResponseImpl.java | 5 +
>> .../org/apache/cxf/jaxrs/utils/JAXRSUtils.java | 6 +-
>> .../apache/cxf/jaxrs/utils/ReaderInputStream.java | 296
>> +++++++++++++++++++++
>> .../apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java | 90 +++++++
>> 4 files changed, 396 insertions(+), 1 deletion(-)
>>
>> diff --git
>> a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>> b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>
>> index e3a3acf..82f183a 100644
>> ---
>> a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>
>> +++
>> b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/ResponseImpl.java
>>
>> @@ -58,6 +58,7 @@ import org.apache.cxf.jaxrs.provider.ProviderFactory;
>> import org.apache.cxf.jaxrs.utils.HttpUtils;
>> import org.apache.cxf.jaxrs.utils.InjectionUtils;
>> import org.apache.cxf.jaxrs.utils.JAXRSUtils;
>> +import org.apache.cxf.jaxrs.utils.ReaderInputStream;
>> import org.apache.cxf.message.Message;
>> import org.apache.cxf.message.MessageUtils;
>> @@ -342,6 +343,10 @@ public final class ResponseImpl extends Response {
>> entityStreamAvailable = entityStream != null;
>> } else if (entity instanceof InputStream) {
>> entityStream = InputStream.class.cast(entity);
>> + } else {
>> + Message inMessage = getResponseMessage();
>> + Reader reader = inMessage.getContent(Reader.class);
>> + entityStream = InputStream.class.cast(new
>> ReaderInputStream(reader));
>> }
>> // we need to check for readers even if no IS is set - the
>> readers may still do it
>> diff --git
>> a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>> b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>
>> index 0ef6565..509f76d 100644
>> ---
>> a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>
>> +++
>> b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/JAXRSUtils.java
>>
>> @@ -832,7 +832,11 @@ public final class JAXRSUtils {
>> OperationResourceInfo ori)
>> throws IOException, WebApplicationException {
>> InputStream is = message.getContent(InputStream.class);
>> -
>> + if (is == null) {
>> + //may use the jms transport so check the Reader;
>> + Reader reader = message.getContent(Reader.class);
>> + is = new ReaderInputStream(reader);
>> + }
>> if (parameter.getType() == ParameterType.REQUEST_BODY) {
>> if (parameterClass == AsyncResponse.class) {
>> diff --git
>> a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
>> b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
>>
>> new file mode 100644
>> index 0000000..c7b142e
>> --- /dev/null
>> +++
>> b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/utils/ReaderInputStream.java
>>
>> @@ -0,0 +1,296 @@
>> +/**
>> + * Licensed to the Apache Software Foundation (ASF) under one
>> + * or more contributor license agreements. See the NOTICE file
>> + * distributed with this work for additional information
>> + * regarding copyright ownership. The ASF licenses this file
>> + * to you 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.apache.cxf.jaxrs.utils;
>> +
>> +import java.io.IOException;
>> +import java.io.InputStream;
>> +import java.io.Reader;
>> +import java.nio.ByteBuffer;
>> +import java.nio.CharBuffer;
>> +import java.nio.charset.Charset;
>> +import java.nio.charset.CharsetEncoder;
>> +import java.nio.charset.CoderResult;
>> +import java.nio.charset.CodingErrorAction;
>> +
>> +/**
>> + * {@link InputStream} implementation that reads a character stream
>> from a {@link Reader}
>> + * and transforms it to a byte stream using a specified charset
>> encoding. The stream
>> + * is transformed using a {@link CharsetEncoder} object, guaranteeing
>> that all charset
>> + * encodings supported by the JRE are handled correctly. In
>> particular for charsets such as
>> + * UTF-16, the implementation ensures that one and only one byte
>> order marker
>> + * is produced.
>> + * <p>
>> + * Since in general it is not possible to predict the number of
>> characters to be read from the
>> + * {@link Reader} to satisfy a read request on the {@link
>> ReaderInputStream}, all reads from
>> + * the {@link Reader} are buffered. There is therefore no well
>> defined correlation
>> + * between the current position of the {@link Reader} and that of the
>> {@link ReaderInputStream}.
>> + * This also implies that in general there is no need to wrap the
>> underlying {@link Reader}
>> + * in a {@link java.io.BufferedReader}.
>> + * <p>
>> + * {@link ReaderInputStream} implements the inverse transformation of
>> {@link java.io.InputStreamReader};
>> + * in the following example, reading from <tt>in2</tt> would return
>> the same byte
>> + * sequence as reading from <tt>in</tt> (provided that the initial
>> byte sequence is legal
>> + * with respect to the charset encoding):
>> + * <pre>
>> + * InputStream in = ...
>> + * Charset cs = ...
>> + * InputStreamReader reader = new InputStreamReader(in, cs);
>> + * ReaderInputStream in2 = new ReaderInputStream(reader, cs);</pre>
>> + * {@link ReaderInputStream} implements the same transformation as
>> {@link java.io.OutputStreamWriter},
>> + * except that the control flow is reversed: both classes transform a
>> character stream
>> + * into a byte stream, but {@link java.io.OutputStreamWriter} pushes
>> data to the underlying stream,
>> + * while {@link ReaderInputStream} pulls it from the underlying stream.
>> + * <p>
>> + * Note that while there are use cases where there is no alternative
>> to using
>> + * this class, very often the need to use this class is an indication
>> of a flaw
>> + * in the design of the code. This class is typically used in
>> situations where an existing
>> + * API only accepts an {@link InputStream}, but where the most
>> natural way to produce the data
>> + * is as a character stream, i.e. by providing a {@link Reader}
>> instance. An example of a situation
>> + * where this problem may appear is when implementing the {@link
>> javax.activation.DataSource}
>> + * interface from the Java Activation Framework.
>> + * <p>
>> + * Given the fact that the {@link Reader} class doesn't provide any
>> way to predict whether the next
>> + * read operation will block or not, it is not possible to provide a
>> meaningful
>> + * implementation of the {@link InputStream#available()} method. A
>> call to this method
>> + * will always return 0. Also, this class doesn't support {@link
>> InputStream#mark(int)}.
>> + * <p>
>> + * Instances of {@link ReaderInputStream} are not thread safe.
>> + *
>> + * @see org.apache.commons.io.output.WriterOutputStream
>> + *
>> + * @since 2.0
>> + */
>> +public class ReaderInputStream extends InputStream {
>> + private static final int DEFAULT_BUFFER_SIZE = 1024;
>> +
>> + private final Reader reader;
>> + private final CharsetEncoder encoder;
>> +
>> + /**
>> + * CharBuffer used as input for the decoder. It should be reasonably
>> + * large as we read data from the underlying Reader into this
>> buffer.
>> + */
>> + private final CharBuffer encoderIn;
>> +
>> + /**
>> + * ByteBuffer used as output for the decoder. This buffer can be
>> small
>> + * as it is only used to transfer data from the decoder to the
>> + * buffer provided by the caller.
>> + */
>> + private final ByteBuffer encoderOut;
>> +
>> + private CoderResult lastCoderResult;
>> + private boolean endOfInput;
>> +
>> + /**
>> + * Construct a new {@link ReaderInputStream}.
>> + *
>> + * @param reader the target {@link Reader}
>> + * @param encoder the charset encoder
>> + * @since 2.1
>> + */
>> + public ReaderInputStream(Reader reader, CharsetEncoder encoder) {
>> + this(reader, encoder, DEFAULT_BUFFER_SIZE);
>> + }
>> +
>> + /**
>> + * Construct a new {@link ReaderInputStream}.
>> + *
>> + * @param reader the target {@link Reader}
>> + * @param encoder the charset encoder
>> + * @param bufferSize the size of the input buffer in number of
>> characters
>> + * @since 2.1
>> + */
>> + public ReaderInputStream(Reader reader, CharsetEncoder encoder,
>> int bufferSize) {
>> + this.reader = reader;
>> + this.encoder = encoder;
>> + this.encoderIn = CharBuffer.allocate(bufferSize);
>> + this.encoderIn.flip();
>> + this.encoderOut = ByteBuffer.allocate(128);
>> + this.encoderOut.flip();
>> + }
>> +
>> + /**
>> + * Construct a new {@link ReaderInputStream}.
>> + *
>> + * @param reader the target {@link Reader}
>> + * @param charset the charset encoding
>> + * @param bufferSize the size of the input buffer in number of
>> characters
>> + */
>> + public ReaderInputStream(Reader reader, Charset charset, int
>> bufferSize) {
>> + this(reader,
>> + charset.newEncoder()
>> + .onMalformedInput(CodingErrorAction.REPLACE)
>> + .onUnmappableCharacter(CodingErrorAction.REPLACE),
>> + bufferSize);
>> + }
>> +
>> + /**
>> + * Construct a new {@link ReaderInputStream} with a default input
>> buffer size of
>> + * 1024 characters.
>> + *
>> + * @param reader the target {@link Reader}
>> + * @param charset the charset encoding
>> + */
>> + public ReaderInputStream(Reader reader, Charset charset) {
>> + this(reader, charset, DEFAULT_BUFFER_SIZE);
>> + }
>> +
>> + /**
>> + * Construct a new {@link ReaderInputStream}.
>> + *
>> + * @param reader the target {@link Reader}
>> + * @param charsetName the name of the charset encoding
>> + * @param bufferSize the size of the input buffer in number of
>> characters
>> + */
>> + public ReaderInputStream(Reader reader, String charsetName, int
>> bufferSize) {
>> + this(reader, Charset.forName(charsetName), bufferSize);
>> + }
>> +
>> + /**
>> + * Construct a new {@link ReaderInputStream} with a default input
>> buffer size of
>> + * 1024 characters.
>> + *
>> + * @param reader the target {@link Reader}
>> + * @param charsetName the name of the charset encoding
>> + */
>> + public ReaderInputStream(Reader reader, String charsetName) {
>> + this(reader, charsetName, DEFAULT_BUFFER_SIZE);
>> + }
>> +
>> + /**
>> + * Construct a new {@link ReaderInputStream} that uses the
>> default character encoding
>> + * with a default input buffer size of 1024 characters.
>> + *
>> + * @param reader the target {@link Reader}
>> + */
>> + public ReaderInputStream(Reader reader) {
>> + this(reader, Charset.defaultCharset());
>> + }
>> +
>> + /**
>> + * Fills the internal char buffer from the reader.
>> + *
>> + * @throws IOException
>> + * If an I/O error occurs
>> + */
>> + private void fillBuffer() throws IOException {
>> + if (!endOfInput && (lastCoderResult == null ||
>> lastCoderResult.isUnderflow())) {
>> + encoderIn.compact();
>> + int position = encoderIn.position();
>> + // We don't use Reader#read(CharBuffer) here because it
>> is more efficient
>> + // to write directly to the underlying char array (the
>> default implementation
>> + // copies data to a temporary char array).
>> + int c = reader.read(encoderIn.array(), position,
>> encoderIn.remaining());
>> + if (c == -1) {
>> + endOfInput = true;
>> + } else {
>> + encoderIn.position(position + c);
>> + }
>> + encoderIn.flip();
>> + }
>> + encoderOut.compact();
>> + lastCoderResult = encoder.encode(encoderIn, encoderOut,
>> endOfInput);
>> + encoderOut.flip();
>> + }
>> +
>> + /**
>> + * Read the specified number of bytes into an array.
>> + *
>> + * @param b the byte array to read into
>> + * @param off the offset to start reading bytes into
>> + * @param len the number of bytes to read
>> + * @return the number of bytes read or <code>-1</code>
>> + * if the end of the stream has been reached
>> + * @throws IOException if an I/O error occurs
>> + */
>> + @Override
>> + public int read(byte[] b, int off, int len) throws IOException {
>> + if (b == null) {
>> + throw new NullPointerException("Byte array must not be
>> null");
>> + }
>> + if (len < 0 || off < 0 || (off + len) > b.length) {
>> + throw new IndexOutOfBoundsException("Array Size=" + b.length
>> + + ", offset=" + off + ", length=" + len);
>> + }
>> + int read = 0;
>> + if (len == 0) {
>> + return 0; // Always return 0 if len == 0
>> + }
>> + while (len > 0) {
>> + if (encoderOut.hasRemaining()) {
>> + int c = Math.min(encoderOut.remaining(), len);
>> + encoderOut.get(b, off, c);
>> + off += c;
>> + len -= c;
>> + read += c;
>> + } else {
>> + fillBuffer();
>> + if (endOfInput && !encoderOut.hasRemaining()) {
>> + break;
>> + }
>> + }
>> + }
>> + return read == 0 && endOfInput ? -1 : read;
>> + }
>> +
>> + /**
>> + * Read the specified number of bytes into an array.
>> + *
>> + * @param b the byte array to read into
>> + * @return the number of bytes read or <code>-1</code>
>> + * if the end of the stream has been reached
>> + * @throws IOException if an I/O error occurs
>> + */
>> + @Override
>> + public int read(byte[] b) throws IOException {
>> + return read(b, 0, b.length);
>> + }
>> +
>> + /**
>> + * Read a single byte.
>> + *
>> + * @return either the byte read or <code>-1</code> if the end of
>> the stream
>> + * has been reached
>> + * @throws IOException if an I/O error occurs
>> + */
>> + @Override
>> + public int read() throws IOException {
>> + for (;;) {
>> + if (encoderOut.hasRemaining()) {
>> + return encoderOut.get() & 0xFF;
>> + } else {
>> + fillBuffer();
>> + if (endOfInput && !encoderOut.hasRemaining()) {
>> + return -1;
>> + }
>> + }
>> + }
>> + }
>> +
>> + /**
>> + * Close the stream. This method will cause the underlying {@link
>> Reader}
>> + * to be closed.
>> + * @throws IOException if an I/O error occurs
>> + */
>> + @Override
>> + public void close() throws IOException {
>> + reader.close();
>> + }
>> +}
>> diff --git
>> a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>> b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>
>> index f77b5ed..b957f36 100644
>> ---
>> a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>
>> +++
>> b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
>>
>> @@ -96,6 +96,25 @@ public class JAXRSJmsTest extends
>> AbstractBusClientServerTestBase {
>> assertEquals("Get a wrong response code.", 200,
>> client.getResponse().getStatus());
>> assertEquals("Get a wrong book id.", 123, book.getId());
>> }
>> +
>> + @Test
>> + public void testGetBookFromWebClientWithTextJMSMessage() throws
>> Exception {
>> + // setup the the client
>> + String endpointAddressUrlEncoded =
>> "jms:jndi:dynamicQueues/test.jmstransport.text"
>> + + "?replyToName=dynamicQueues/test.jmstransport.response"
>> + +
>> "&jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>
>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>> + + "&messageType=text";
>> +
>> + WebClient client = WebClient.create(endpointAddressUrlEncoded);
>> + WebClient.getConfig(client).getInInterceptors().add(new
>> LoggingInInterceptor());
>> + WebClient.getConfig(client).getRequestContext()
>> + .put(org.apache.cxf.message.Message.REQUEST_URI,
>> "/bookstore/books/123");
>> +
>> + Book book = client.get(Book.class);
>> + assertEquals("Get a wrong response code.", 200,
>> client.getResponse().getStatus());
>> + assertEquals("Get a wrong book id.", 123, book.getId());
>> + }
>> @Test
>> public void testPutBookOneWayWithWebClient() throws Exception {
>> @@ -130,6 +149,7 @@ public class JAXRSJmsTest extends
>> AbstractBusClientServerTestBase {
>> }
>> }
>> +
>> @Test
>> public void testGetBookFromWebClientWithPath() throws Exception {
>> // setup the the client
>> @@ -148,6 +168,25 @@ public class JAXRSJmsTest extends
>> AbstractBusClientServerTestBase {
>> }
>> @Test
>> + public void testGetBookFromWebClientWithPathWithTextJMSMessage()
>> throws Exception {
>> + // setup the the client
>> + String endpointAddressUrlEncoded =
>> "jms:jndi:dynamicQueues/test.jmstransport.text"
>> + +
>> "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>
>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>> + + "&messageType=text";
>> +
>> + WebClient client = WebClient.create(endpointAddressUrlEncoded);
>> + client.path("bookstore").path("books").path("123");
>> +
>> + Book book = client.get(Book.class);
>> + assertEquals("Get a wrong response code.", 200,
>> client.getResponse().getStatus());
>> + assertEquals("Get a wrong book id.", 123, book.getId());
>> + }
>> +
>> +
>> + @Test
>> public void testGetBookFromProxyClient() throws Exception {
>> // setup the the client
>> String endpointAddressUrlEncoded =
>> "jms:jndi:dynamicQueues/test.jmstransport.text"
>> @@ -161,6 +200,22 @@ public class JAXRSJmsTest extends
>> AbstractBusClientServerTestBase {
>> assertEquals("Get a wrong response code.", 200,
>> WebClient.client(client).getResponse().getStatus());
>> assertEquals("Get a wrong book id.", 123, book.getId());
>> }
>> +
>> + @Test
>> + public void testGetBookFromProxyClientWithTextJMSMessage() throws
>> Exception {
>> + // setup the the client
>> + String endpointAddressUrlEncoded =
>> "jms:jndi:dynamicQueues/test.jmstransport.text"
>> + +
>> "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>
>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>> + + "&messageType=text";
>> +
>> + JMSBookStore client =
>> JAXRSClientFactory.create(endpointAddressUrlEncoded, JMSBookStore.class);
>> + Book book = client.getBook("123");
>> + assertEquals("Get a wrong response code.", 200,
>> WebClient.client(client).getResponse().getStatus());
>> + assertEquals("Get a wrong book id.", 123, book.getId());
>> + }
>> @Test
>> public void testGetBookFromSubresourceProxyClient() throws
>> Exception {
>> @@ -177,6 +232,23 @@ public class JAXRSJmsTest extends
>> AbstractBusClientServerTestBase {
>> assertEquals("Get a wrong response code.", 200,
>> WebClient.client(bookProxy).getResponse().getStatus());
>> assertEquals("Get a wrong book id.", 123, book.getId());
>> }
>> +
>> + @Test
>> + public void
>> testGetBookFromSubresourceProxyClientWithTextJMSMessage() throws
>> Exception {
>> + // setup the the client
>> + String endpointAddressUrlEncoded =
>> "jms:jndi:dynamicQueues/test.jmstransport.text"
>> + +
>> "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>
>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>> + + "&messageType=text";
>> +
>> + JMSBookStore client =
>> JAXRSClientFactory.create(endpointAddressUrlEncoded, JMSBookStore.class);
>> + Book bookProxy = client.getBookSubResource("123");
>> + Book book = bookProxy.retrieveState();
>> + assertEquals("Get a wrong response code.", 200,
>> WebClient.client(bookProxy).getResponse().getStatus());
>> + assertEquals("Get a wrong book id.", 123, book.getId());
>> + }
>> @Test
>> public void testGetBookFromProxyClientWithQuery() throws
>> Exception {
>> @@ -193,6 +265,24 @@ public class JAXRSJmsTest extends
>> AbstractBusClientServerTestBase {
>> assertEquals("Get a wrong book id.", 123, book.getId());
>> }
>> +
>> + @Test
>> + public void
>> testGetBookFromProxyClientWithQueryWithTextJMSMessage() throws
>> Exception {
>> + // setup the the client
>> + String endpointAddressUrlEncoded =
>> "jms:jndi:dynamicQueues/test.jmstransport.text"
>> + +
>> "?jndiInitialContextFactory=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
>>
>> + + "&replyToName=dynamicQueues/test.jmstransport.response"
>> + + "&jndiURL=tcp://localhost:" + JMS_PORT
>> + + "&jndiConnectionFactoryName=ConnectionFactory"
>> + + "&messageType=text";
>> +
>> + JMSBookStore client =
>> JAXRSClientFactory.create(endpointAddressUrlEncoded, JMSBookStore.class);
>> + Book book = client.getBookByURLQuery(new String[] {"1", "2",
>> "3"});
>> + assertEquals("Get a wrong response code.", 200,
>> WebClient.client(client).getResponse().getStatus());
>> + assertEquals("Get a wrong book id.", 123, book.getId());
>> + }
>> +
>> +
>> @Test
>> public void testGetBook() throws Exception {
>> Context ctx = getContext();
>>
>
>