You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cxf.apache.org by ff...@apache.org on 2017/10/19 03:16:10 UTC
[cxf] branch master updated: [CXF-7532]REST on JMS transport can't
handle the request message with text messageType
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();
--
To stop receiving notification emails like this one, please contact
['"commits@cxf.apache.org" <co...@cxf.apache.org>'].
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();
>>
>
>
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
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/