use AmqpJmsMessageBuilder to create incoming messages, remove transformer usage
Project: http://git-wip-us.apache.org/repos/asf/qpid-jms/repo Commit: http://git-wip-us.apache.org/repos/asf/qpid-jms/commit/9e3b5ecc Tree: http://git-wip-us.apache.org/repos/asf/qpid-jms/tree/9e3b5ecc Diff: http://git-wip-us.apache.org/repos/asf/qpid-jms/diff/9e3b5ecc Branch: refs/heads/master Commit: 9e3b5ecc8cedd4c93dde00a0ae282404e31c4fcb Parents: 05b73e4 Author: Robert Gemmell <rob...@apache.org> Authored: Mon Sep 29 12:10:24 2014 +0100 Committer: Robert Gemmell <rob...@apache.org> Committed: Mon Sep 29 17:16:24 2014 +0100 ---------------------------------------------------------------------- .../org/apache/qpid/jms/provider/amqp/AmqpConsumer.java | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/qpid-jms/blob/9e3b5ecc/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpConsumer.java ---------------------------------------------------------------------- diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpConsumer.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpConsumer.java index d949b70..9c04a5a 100644 --- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpConsumer.java +++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpConsumer.java @@ -32,6 +32,7 @@ import org.apache.qpid.jms.meta.JmsConsumerId; import org.apache.qpid.jms.meta.JmsConsumerInfo; import org.apache.qpid.jms.provider.AsyncResult; import org.apache.qpid.jms.provider.ProviderConstants.ACK_TYPE; +import org.apache.qpid.jms.provider.amqp.message.AmqpJmsMessageBuilder; import org.apache.qpid.jms.provider.ProviderListener; import org.apache.qpid.jms.util.IOExceptionSupport; import org.apache.qpid.proton.amqp.Binary; @@ -49,8 +50,7 @@ import org.apache.qpid.proton.amqp.transport.SenderSettleMode; import org.apache.qpid.proton.engine.Delivery; import org.apache.qpid.proton.engine.Receiver; import org.apache.qpid.proton.jms.EncodedMessage; -import org.apache.qpid.proton.jms.InboundTransformer; -import org.apache.qpid.proton.jms.JMSMappingInboundTransformer; +import org.apache.qpid.proton.message.Message; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -66,8 +66,6 @@ public class AmqpConsumer extends AbstractAmqpResource<JmsConsumerInfo, Receiver protected static final Symbol JMS_SELECTOR_SYMBOL = Symbol.valueOf("jms-selector"); protected final AmqpSession session; - protected final InboundTransformer inboundTransformer = - new JMSMappingInboundTransformer(AmqpJMSVendor.INSTANCE);; protected final Map<JmsInboundMessageDispatch, Delivery> delivered = new LinkedHashMap<JmsInboundMessageDispatch, Delivery>(); protected boolean presettle; @@ -310,7 +308,10 @@ public class AmqpConsumer extends AbstractAmqpResource<JmsConsumerInfo, Receiver EncodedMessage encoded = readIncomingMessage(incoming); JmsMessage message = null; try { - message = (JmsMessage) inboundTransformer.transform(encoded); + Message protonMessage = Message.Factory.create(); + protonMessage.decode(encoded.getArray(), encoded.getArrayOffset(), encoded.getLength()); + + message = (JmsMessage) AmqpJmsMessageBuilder.createJmsMessage(this, protonMessage); } catch (Exception e) { LOG.warn("Error on transform: {}", e.getMessage()); // TODO - We could signal provider error but not sure we want to fail @@ -431,6 +432,7 @@ public class AmqpConsumer extends AbstractAmqpResource<JmsConsumerInfo, Receiver buffer = streamBuffer.toByteArray(); try { + //TODO: get rid of EncodedMessage usage return new EncodedMessage(incoming.getMessageFormat(), buffer, 0, buffer.length); } finally { streamBuffer.reset(); --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@qpid.apache.org For additional commands, e-mail: commits-h...@qpid.apache.org
