Apache
/
activemq-artemis
mirror of https://github.com/apache/activemq-artemis.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

ARTEMIS-3113 - Artemis AMQP shouldn't depend on JMS. 

* removing the  JMS dependency on AMQP module
* fixing destinations usage.
* refactoring to remove some JMS usage and make exceptions a bit better

Jira: https://issues.apache.org/jira/browse/ARTEMIS-3113
This commit is contained in:
Clebert Suconic 2021-02-10 18:37:13 -05:00
parent 3cd5efd7ae
commit e7e3c71511
24 changed files with 1380 additions and 1522 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

74
artemis-commons/src/main/java/org/apache/activemq/artemis/utils/DestinationUtil.java Normal file
View File

@ -0,0 +1,74 @@
/*
* 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.activemq.artemis.utils;
import org.apache.activemq.artemis.api.core.SimpleString;
public class DestinationUtil {
public static final String QUEUE_QUALIFIED_PREFIX = "queue://";
public static final String TOPIC_QUALIFIED_PREFIX = "topic://";
public static final String TEMP_QUEUE_QUALIFED_PREFIX = "temp-queue://";
public static final String TEMP_TOPIC_QUALIFED_PREFIX = "temp-topic://";
public static final char SEPARATOR = '.';
private static String escape(final String input) {
if (input == null) {
return "";
}
return input.replace("\\", "\\\\").replace(".", "\\.");
}
public static String createQueueNameForSharedSubscription(final boolean isDurable,
final String clientID,
final String subscriptionName) {
if (clientID != null) {
return (isDurable ? "Durable" : "nonDurable") + SEPARATOR +
escape(clientID) + SEPARATOR +
escape(subscriptionName);
} else {
return (isDurable ? "Durable" : "nonDurable") + SEPARATOR +
escape(subscriptionName);
}
}
public static SimpleString createQueueNameForSubscription(final boolean isDurable,
final String clientID,
final String subscriptionName) {
final String queueName;
if (clientID != null) {
if (isDurable) {
queueName = escape(clientID) + SEPARATOR +
escape(subscriptionName);
} else {
queueName = "nonDurable" + SEPARATOR +
escape(clientID) + SEPARATOR +
escape(subscriptionName);
}
} else {
if (isDurable) {
queueName = escape(subscriptionName);
} else {
queueName = "nonDurable" + SEPARATOR +
escape(subscriptionName);
}
}
return SimpleString.toSimpleString(queueName);
}
}

49
artemis-jms-client/src/main/java/org/apache/activemq/artemis/jms/client/ActiveMQDestination.java
View File

@ -31,6 +31,7 @@ import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl;
import org.apache.activemq.artemis.jndi.JNDIStorable;
import org.apache.activemq.artemis.utils.DestinationUtil;
/**
* ActiveMQ Artemis implementation of a JMS Destination.
@ -42,19 +43,10 @@ public class ActiveMQDestination extends JNDIStorable implements Destination, Se
private static final long serialVersionUID = 5027962425462382883L;
public static final String QUEUE_QUALIFIED_PREFIX = "queue://";
public static final String TOPIC_QUALIFIED_PREFIX = "topic://";
public static final String TEMP_QUEUE_QUALIFED_PREFIX = "temp-queue://";
public static final String TEMP_TOPIC_QUALIFED_PREFIX = "temp-topic://";
private static final char SEPARATOR = '.';
private static String escape(final String input) {
if (input == null) {
return "";
}
return input.replace("\\", "\\\\").replace(".", "\\.");
}
public static final String QUEUE_QUALIFIED_PREFIX = DestinationUtil.QUEUE_QUALIFIED_PREFIX;
public static final String TOPIC_QUALIFIED_PREFIX = DestinationUtil.TOPIC_QUALIFIED_PREFIX;
public static final String TEMP_QUEUE_QUALIFED_PREFIX = DestinationUtil.TEMP_QUEUE_QUALIFED_PREFIX;
public static final String TEMP_TOPIC_QUALIFED_PREFIX = DestinationUtil.TEMP_TOPIC_QUALIFED_PREFIX;
/** createQueue and createTopic from {@link ActiveMQSession} may change the name
* in case Prefix usage */
@ -167,38 +159,13 @@ public class ActiveMQDestination extends JNDIStorable implements Destination, Se
public static SimpleString createQueueNameForSubscription(final boolean isDurable,
final String clientID,
final String subscriptionName) {
final String queueName;
if (clientID != null) {
if (isDurable) {
queueName = ActiveMQDestination.escape(clientID) + SEPARATOR +
ActiveMQDestination.escape(subscriptionName);
} else {
queueName = "nonDurable" + SEPARATOR +
ActiveMQDestination.escape(clientID) + SEPARATOR +
ActiveMQDestination.escape(subscriptionName);
}
} else {
if (isDurable) {
queueName = ActiveMQDestination.escape(subscriptionName);
} else {
queueName = "nonDurable" + SEPARATOR +
ActiveMQDestination.escape(subscriptionName);
}
}
return SimpleString.toSimpleString(queueName);
return DestinationUtil.createQueueNameForSubscription(isDurable, clientID, subscriptionName);
}
public static String createQueueNameForSharedSubscription(final boolean isDurable,
final String clientID,
final String subscriptionName) {
if (clientID != null) {
return (isDurable ? "Durable" : "nonDurable") + SEPARATOR +
ActiveMQDestination.escape(clientID) + SEPARATOR +
ActiveMQDestination.escape(subscriptionName);
} else {
return (isDurable ? "Durable" : "nonDurable") + SEPARATOR +
ActiveMQDestination.escape(subscriptionName);
}
return DestinationUtil.createQueueNameForSharedSubscription(isDurable, clientID, subscriptionName);
}
public static Pair<String, String> decomposeQueueNameForDurableSubscription(final String queueName) {
@ -213,7 +180,7 @@ public class ActiveMQDestination extends JNDIStorable implements Destination, Se
char ch = queueName.charAt(pos);
pos++;
if (ch == SEPARATOR) {
if (ch == DestinationUtil.SEPARATOR) {
currentPart++;
if (currentPart >= parts.length) {
throw new JMSRuntimeException("Invalid message queue name: " + queueName);

16
artemis-protocols/artemis-amqp-protocol/pom.xml
View File

@ -32,12 +32,6 @@
</properties>
<dependencies>
<!-- JMS Client because of some conversions that are done -->
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>artemis-jms-client</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>artemis-selector</artifactId>
@ -48,6 +42,11 @@
<artifactId>artemis-core-client</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>artemis-commons</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.jboss.logging</groupId>
<artifactId>jboss-logging-processor</artifactId>
@ -110,11 +109,6 @@
<artifactId>commons-collections</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-jms_2.0_spec</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>

5
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/ProtonProtocolManager.java
View File

@ -32,7 +32,6 @@ import org.apache.activemq.artemis.core.remoting.impl.netty.NettyServerConnectio
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.management.Notification;
import org.apache.activemq.artemis.core.server.management.NotificationListener;
import org.apache.activemq.artemis.jms.client.ActiveMQDestination;
import org.apache.activemq.artemis.protocol.amqp.client.ProtonClientProtocolManager;
import org.apache.activemq.artemis.protocol.amqp.proton.AMQPConnectionContext;
import org.apache.activemq.artemis.protocol.amqp.proton.AMQPConstants;
@ -47,6 +46,7 @@ import org.apache.activemq.artemis.spi.core.remoting.Acceptor;
import org.apache.activemq.artemis.spi.core.remoting.Connection;
import io.netty.channel.ChannelPipeline;
import org.apache.activemq.artemis.utils.DestinationUtil;
import org.jboss.logging.Logger;
/**
@ -101,8 +101,7 @@ public class ProtonProtocolManager extends AbstractProtocolManager<AMQPMessage,
* used when you want to treat senders as a subscription on an address rather than consuming from the actual queue for
* the address. This can be changed on the acceptor.
* */
// TODO fix this
private String pubSubPrefix = ActiveMQDestination.TOPIC_QUALIFIED_PREFIX;
private String pubSubPrefix = DestinationUtil.TOPIC_QUALIFIED_PREFIX;
private int maxFrameSize = AmqpSupport.MAX_FRAME_SIZE_DEFAULT;

173
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/AMQPMessageSupport.java
View File

@ -28,42 +28,40 @@ import java.util.Arrays;
import java.util.Map;
import java.util.Set;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Queue;
import javax.jms.TemporaryQueue;
import javax.jms.TemporaryTopic;
import javax.jms.Topic;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.message.impl.CoreMessage;
import org.apache.activemq.artemis.core.persistence.CoreMessageObjectPools;
import org.apache.activemq.artemis.jms.client.ActiveMQDestination;
import org.apache.activemq.artemis.jms.client.ActiveMQQueue;
import org.apache.activemq.artemis.jms.client.ActiveMQTemporaryQueue;
import org.apache.activemq.artemis.jms.client.ActiveMQTemporaryTopic;
import org.apache.activemq.artemis.jms.client.ActiveMQTopic;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSBytesMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSMapMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSObjectMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSStreamMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSTextMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreBytesMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreMapMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreObjectMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreStreamMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreTextMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.exceptions.ActiveMQAMQPInvalidContentTypeException;
import org.apache.activemq.artemis.utils.DestinationUtil;
import org.apache.qpid.proton.amqp.Binary;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.messaging.Data;
import org.apache.qpid.proton.message.Message;
import org.jboss.logging.Logger;
import static org.apache.activemq.artemis.utils.DestinationUtil.QUEUE_QUALIFIED_PREFIX;
import static org.apache.activemq.artemis.utils.DestinationUtil.TEMP_QUEUE_QUALIFED_PREFIX;
import static org.apache.activemq.artemis.utils.DestinationUtil.TEMP_TOPIC_QUALIFED_PREFIX;
import static org.apache.activemq.artemis.utils.DestinationUtil.TOPIC_QUALIFIED_PREFIX;
/**
* Support class containing constant values and static methods that are used to map to / from
* AMQP Message types being sent or received.
*/
public final class AMQPMessageSupport {
public static final int NON_PERSISTENT = 1;
private static final Logger logger = Logger.getLogger(AMQPMessageSupport.class);
public static final int PERSISTENT = 2;
public static final long MESSAGE_DEFAULT_TIME_TO_LIVE = 0;
public static final int MESSAGE_DEFAULT_PRIORITY = 4;
public static SimpleString HDR_ORIGINAL_ADDRESS_ANNOTATION = SimpleString.toSimpleString("x-opt-ORIG-ADDRESS");
@ -303,79 +301,56 @@ public final class AMQPMessageSupport {
return key;
}
public static String toAddress(Destination destination) {
try {
if (destination instanceof ActiveMQDestination) {
return ((ActiveMQDestination) destination).getAddress();
} else {
if (destination instanceof Queue) {
return ((Queue) destination).getQueueName();
} else if (destination instanceof Topic) {
return ((Topic) destination).getTopicName();
}
}
} catch (JMSException e) {
// ActiveMQDestination (and most JMS implementations I know) will never throw an Exception here
// this is here for compilation support (as JMS declares it), and I don't want to propagate exceptions into
// the converter...
// and for the possibility of who knows in the future!!!
logger.warn(e.getMessage(), e);
}
return null;
public static CoreBytesMessageWrapper createBytesMessage(long id, CoreMessageObjectPools coreMessageObjectPools) {
return new CoreBytesMessageWrapper(newMessage(id, BYTES_TYPE, coreMessageObjectPools));
}
public static ServerJMSBytesMessage createBytesMessage(long id, CoreMessageObjectPools coreMessageObjectPools) {
return new ServerJMSBytesMessage(newMessage(id, BYTES_TYPE, coreMessageObjectPools));
}
public static ServerJMSBytesMessage createBytesMessage(long id, byte[] array, int arrayOffset, int length, CoreMessageObjectPools coreMessageObjectPools) throws JMSException {
ServerJMSBytesMessage message = createBytesMessage(id, coreMessageObjectPools);
public static CoreBytesMessageWrapper createBytesMessage(long id, byte[] array, int arrayOffset, int length, CoreMessageObjectPools coreMessageObjectPools) {
CoreBytesMessageWrapper message = createBytesMessage(id, coreMessageObjectPools);
message.writeBytes(array, arrayOffset, length);
return message;
}
public static ServerJMSStreamMessage createStreamMessage(long id, CoreMessageObjectPools coreMessageObjectPools) {
return new ServerJMSStreamMessage(newMessage(id, STREAM_TYPE, coreMessageObjectPools));
public static CoreStreamMessageWrapper createStreamMessage(long id, CoreMessageObjectPools coreMessageObjectPools) {
return new CoreStreamMessageWrapper(newMessage(id, STREAM_TYPE, coreMessageObjectPools));
}
public static ServerJMSMessage createMessage(long id, CoreMessageObjectPools coreMessageObjectPools) {
return new ServerJMSMessage(newMessage(id, DEFAULT_TYPE, coreMessageObjectPools));
public static CoreMessageWrapper createMessage(long id, CoreMessageObjectPools coreMessageObjectPools) {
return new CoreMessageWrapper(newMessage(id, DEFAULT_TYPE, coreMessageObjectPools));
}
public static ServerJMSTextMessage createTextMessage(long id, CoreMessageObjectPools coreMessageObjectPools) {
return new ServerJMSTextMessage(newMessage(id, TEXT_TYPE, coreMessageObjectPools));
public static CoreTextMessageWrapper createTextMessage(long id, CoreMessageObjectPools coreMessageObjectPools) {
return new CoreTextMessageWrapper(newMessage(id, TEXT_TYPE, coreMessageObjectPools));
}
public static ServerJMSTextMessage createTextMessage(long id, String text, CoreMessageObjectPools coreMessageObjectPools) throws JMSException {
ServerJMSTextMessage message = createTextMessage(id, coreMessageObjectPools);
public static CoreTextMessageWrapper createTextMessage(long id, String text, CoreMessageObjectPools coreMessageObjectPools) {
CoreTextMessageWrapper message = createTextMessage(id, coreMessageObjectPools);
message.setText(text);
return message;
}
public static ServerJMSObjectMessage createObjectMessage(long id, CoreMessageObjectPools coreMessageObjectPools) {
return new ServerJMSObjectMessage(newMessage(id, OBJECT_TYPE, coreMessageObjectPools));
public static CoreObjectMessageWrapper createObjectMessage(long id, CoreMessageObjectPools coreMessageObjectPools) {
return new CoreObjectMessageWrapper(newMessage(id, OBJECT_TYPE, coreMessageObjectPools));
}
public static ServerJMSMessage createObjectMessage(long id, Binary serializedForm, CoreMessageObjectPools coreMessageObjectPools) throws JMSException {
ServerJMSObjectMessage message = createObjectMessage(id, coreMessageObjectPools);
public static CoreMessageWrapper createObjectMessage(long id, Binary serializedForm, CoreMessageObjectPools coreMessageObjectPools) {
CoreObjectMessageWrapper message = createObjectMessage(id, coreMessageObjectPools);
message.setSerializedForm(serializedForm);
return message;
}
public static ServerJMSMessage createObjectMessage(long id, byte[] array, int offset, int length, CoreMessageObjectPools coreMessageObjectPools) throws JMSException {
ServerJMSObjectMessage message = createObjectMessage(id, coreMessageObjectPools);
public static CoreMessageWrapper createObjectMessage(long id, byte[] array, int offset, int length, CoreMessageObjectPools coreMessageObjectPools) {
CoreObjectMessageWrapper message = createObjectMessage(id, coreMessageObjectPools);
message.setSerializedForm(new Binary(array, offset, length));
return message;
}
public static ServerJMSMapMessage createMapMessage(long id, CoreMessageObjectPools coreMessageObjectPools) {
return new ServerJMSMapMessage(newMessage(id, MAP_TYPE, coreMessageObjectPools));
public static CoreMapMessageWrapper createMapMessage(long id, CoreMessageObjectPools coreMessageObjectPools) {
return new CoreMapMessageWrapper(newMessage(id, MAP_TYPE, coreMessageObjectPools));
}
public static ServerJMSMapMessage createMapMessage(long id, Map<String, Object> content, CoreMessageObjectPools coreMessageObjectPools) throws JMSException {
ServerJMSMapMessage message = createMapMessage(id, coreMessageObjectPools);
public static CoreMapMessageWrapper createMapMessage(long id, Map<String, Object> content, CoreMessageObjectPools coreMessageObjectPools) {
CoreMapMessageWrapper message = createMapMessage(id, coreMessageObjectPools);
final Set<Map.Entry<String, Object>> set = content.entrySet();
for (Map.Entry<String, Object> entry : set) {
Object value = entry.getValue();
@ -396,36 +371,54 @@ public final class AMQPMessageSupport {
}
public static byte destinationType(Destination destination) {
if (destination instanceof Queue) {
if (destination instanceof TemporaryQueue) {
return TEMP_QUEUE_TYPE;
} else {
return QUEUE_TYPE;
}
} else if (destination instanceof Topic) {
if (destination instanceof TemporaryTopic) {
return TEMP_TOPIC_TYPE;
} else {
return TOPIC_TYPE;
}
}
public static String toAddress(String destination) {
if (destination.startsWith(DestinationUtil.QUEUE_QUALIFIED_PREFIX)) {
return destination.substring(DestinationUtil.QUEUE_QUALIFIED_PREFIX.length());
} else if (destination.startsWith(DestinationUtil.TOPIC_QUALIFIED_PREFIX)) {
return destination.substring(DestinationUtil.TOPIC_QUALIFIED_PREFIX.length());
} else if (destination.startsWith(DestinationUtil.TEMP_QUEUE_QUALIFED_PREFIX)) {
return destination.substring(DestinationUtil.TEMP_QUEUE_QUALIFED_PREFIX.length());
} else if (destination.startsWith(DestinationUtil.TEMP_TOPIC_QUALIFED_PREFIX)) {
return destination.substring(DestinationUtil.TEMP_TOPIC_QUALIFED_PREFIX.length());
} else {
return destination;
}
}
public static byte destinationType(String destination) {
if (destination.startsWith(QUEUE_QUALIFIED_PREFIX)) {
return QUEUE_TYPE;
}
if (destination.startsWith(TOPIC_QUALIFIED_PREFIX)) {
return TOPIC_TYPE;
}
if (destination.startsWith(TEMP_QUEUE_QUALIFED_PREFIX)) {
return TEMP_QUEUE_TYPE;
}
if (destination.startsWith(TEMP_TOPIC_QUALIFED_PREFIX)) {
return TEMP_TOPIC_TYPE;
}
return QUEUE_TYPE;
}
public static Destination destination(byte destinationType, String address) {
public static String destination(RoutingType destinationType, String address) {
String prefix;
if (destinationType == null) {
destinationType = RoutingType.ANYCAST;
}
switch (destinationType) {
case TEMP_QUEUE_TYPE:
return new ActiveMQTemporaryQueue(address, null);
case TEMP_TOPIC_TYPE:
return new ActiveMQTemporaryTopic(address, null);
case TOPIC_TYPE:
return new ActiveMQTopic(address);
case QUEUE_TYPE:
return new ActiveMQQueue(address);
default:
return new ActiveMQQueue(address);
case ANYCAST: prefix = QUEUE_QUALIFIED_PREFIX; break;
case MULTICAST: prefix = TOPIC_QUALIFIED_PREFIX; break;
default: prefix = QUEUE_QUALIFIED_PREFIX; break;
}
if (!address.startsWith(prefix)) {
return prefix + address;
} else {
return address;
}
}

69
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/AmqpCoreConverter.java
View File

@ -60,16 +60,13 @@ import java.util.Map;
import java.util.Set;
import java.util.UUID;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import org.apache.activemq.artemis.api.core.ActiveMQPropertyConversionException;
import org.apache.activemq.artemis.api.core.ICoreMessage;
import org.apache.activemq.artemis.core.persistence.CoreMessageObjectPools;
import org.apache.activemq.artemis.jms.client.ActiveMQDestination;
import org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSStreamMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.ConversionException;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreStreamMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.util.NettyWritable;
import org.apache.activemq.artemis.protocol.amqp.util.TLSEncode;
import org.apache.activemq.artemis.utils.UUIDGenerator;
@ -98,23 +95,31 @@ import org.apache.qpid.proton.codec.WritableBuffer;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.NON_PERSISTENT;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.PERSISTENT;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.MESSAGE_DEFAULT_PRIORITY;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.MESSAGE_DEFAULT_TIME_TO_LIVE;
import static org.apache.activemq.artemis.utils.DestinationUtil.QUEUE_QUALIFIED_PREFIX;
import static org.apache.activemq.artemis.utils.DestinationUtil.TEMP_QUEUE_QUALIFED_PREFIX;
import static org.apache.activemq.artemis.utils.DestinationUtil.TEMP_TOPIC_QUALIFED_PREFIX;
import static org.apache.activemq.artemis.utils.DestinationUtil.TOPIC_QUALIFIED_PREFIX;
/**
* This class was created just to separate concerns on AMQPConverter.
* For better organization of the code.
* */
public class AmqpCoreConverter {
public static ICoreMessage toCore(AMQPMessage message, CoreMessageObjectPools coreMessageObjectPools) throws Exception {
return message.toCore(coreMessageObjectPools);
}
@SuppressWarnings("unchecked")
public static ICoreMessage toCore(AMQPMessage message, CoreMessageObjectPools coreMessageObjectPools, Header header, MessageAnnotations annotations, Properties properties, ApplicationProperties applicationProperties, Section body, Footer footer) throws Exception {
public static ICoreMessage toCore(AMQPMessage message, CoreMessageObjectPools coreMessageObjectPools, Header header, MessageAnnotations annotations, Properties properties, ApplicationProperties applicationProperties, Section body, Footer footer) throws ConversionException {
final long messageId = message.getMessageID();
final Symbol contentType = properties != null ? properties.getContentType() : null;
final String contentTypeString = contentType != null ? contentType.toString() : null;
ServerJMSMessage result;
CoreMessageWrapper result;
if (body == null) {
if (isContentType(SERIALIZED_JAVA_OBJECT_CONTENT_TYPE.toString(), contentType)) {
@ -157,7 +162,7 @@ public class AmqpCoreConverter {
result.setShortProperty(JMS_AMQP_ORIGINAL_ENCODING, AMQP_DATA);
} else if (body instanceof AmqpSequence) {
AmqpSequence sequence = (AmqpSequence) body;
ServerJMSStreamMessage m = createStreamMessage(messageId, coreMessageObjectPools);
CoreStreamMessageWrapper m = createStreamMessage(messageId, coreMessageObjectPools);
for (Object item : sequence.getValue()) {
m.writeObject(item);
}
@ -181,7 +186,7 @@ public class AmqpCoreConverter {
result.setShortProperty(JMS_AMQP_ORIGINAL_ENCODING, AMQP_VALUE_BINARY);
} else if (value instanceof List) {
ServerJMSStreamMessage m = createStreamMessage(messageId, coreMessageObjectPools);
CoreStreamMessageWrapper m = createStreamMessage(messageId, coreMessageObjectPools);
for (Object item : (List<Object>) value) {
m.writeObject(item);
}
@ -213,9 +218,9 @@ public class AmqpCoreConverter {
processExtraProperties(result, message.getExtraProperties());
// If the JMS expiration has not yet been set...
if (header != null && result.getJMSExpiration() == 0) {
if (header != null && result.getExpiration() == 0) {
// Then lets try to set it based on the message TTL.
long ttl = javax.jms.Message.DEFAULT_TIME_TO_LIVE;
long ttl = MESSAGE_DEFAULT_TIME_TO_LIVE;
if (header.getTtl() != null) {
ttl = header.getTtl().longValue();
}
@ -236,22 +241,22 @@ public class AmqpCoreConverter {
return result.getInnerMessage();
}
protected static ServerJMSMessage processHeader(ServerJMSMessage jms, Header header) throws Exception {
protected static CoreMessageWrapper processHeader(CoreMessageWrapper jms, Header header) throws ConversionException {
if (header != null) {
jms.setBooleanProperty(JMS_AMQP_HEADER, true);
if (header.getDurable() != null) {
jms.setBooleanProperty(JMS_AMQP_HEADER_DURABLE, true);
jms.setJMSDeliveryMode(header.getDurable().booleanValue() ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT);
jms.setDeliveryMode(header.getDurable().booleanValue() ? PERSISTENT : NON_PERSISTENT);
} else {
jms.setJMSDeliveryMode(DeliveryMode.NON_PERSISTENT);
jms.setDeliveryMode(NON_PERSISTENT);
}
if (header.getPriority() != null) {
jms.setBooleanProperty(JMS_AMQP_HEADER_PRIORITY, true);
jms.setJMSPriority(header.getPriority().intValue());
} else {
jms.setJMSPriority(javax.jms.Message.DEFAULT_PRIORITY);
jms.setJMSPriority(MESSAGE_DEFAULT_PRIORITY);
}
if (header.getFirstAcquirer() != null) {
@ -264,14 +269,14 @@ public class AmqpCoreConverter {
jms.setLongProperty("JMSXDeliveryCount", header.getDeliveryCount().longValue() + 1);
}
} else {
jms.setJMSPriority((byte) javax.jms.Message.DEFAULT_PRIORITY);
jms.setJMSDeliveryMode(DeliveryMode.NON_PERSISTENT);
jms.setJMSPriority((byte) MESSAGE_DEFAULT_PRIORITY);
jms.setDeliveryMode(NON_PERSISTENT);
}
return jms;
}
protected static ServerJMSMessage processMessageAnnotations(ServerJMSMessage jms, MessageAnnotations annotations) throws Exception {
protected static CoreMessageWrapper processMessageAnnotations(CoreMessageWrapper jms, MessageAnnotations annotations) {
if (annotations != null && annotations.getValue() != null) {
for (Map.Entry<?, ?> entry : annotations.getValue().entrySet()) {
String key = entry.getKey().toString();
@ -296,7 +301,7 @@ public class AmqpCoreConverter {
return jms;
}
private static ServerJMSMessage processApplicationProperties(ServerJMSMessage jms, ApplicationProperties properties) throws Exception {
private static CoreMessageWrapper processApplicationProperties(CoreMessageWrapper jms, ApplicationProperties properties) {
if (properties != null && properties.getValue() != null) {
for (Map.Entry<String, Object> entry : (Set<Map.Entry<String, Object>>) properties.getValue().entrySet()) {
setProperty(jms, entry.getKey(), entry.getValue());
@ -306,7 +311,7 @@ public class AmqpCoreConverter {
return jms;
}
private static ServerJMSMessage processExtraProperties(ServerJMSMessage jms, TypedProperties properties) {
private static CoreMessageWrapper processExtraProperties(CoreMessageWrapper jms, TypedProperties properties) {
if (properties != null) {
properties.forEach((k, v) -> {
if (!k.equals(AMQPMessage.ADDRESS_PROPERTY)) {
@ -318,7 +323,7 @@ public class AmqpCoreConverter {
return jms;
}
private static ServerJMSMessage processProperties(ServerJMSMessage jms, Properties properties, MessageAnnotations annotations) throws Exception {
private static CoreMessageWrapper processProperties(CoreMessageWrapper jms, Properties properties, MessageAnnotations annotations) {
if (properties != null) {
if (properties.getMessageId() != null) {
jms.setJMSMessageID(AMQPMessageIdHelper.INSTANCE.toMessageIdString(properties.getMessageId()));
@ -337,7 +342,7 @@ public class AmqpCoreConverter {
}
if (properties.getTo() != null) {
byte queueType = parseQueueAnnotation(annotations, AMQPMessageSupport.JMS_DEST_TYPE_MSG_ANNOTATION);
jms.setJMSDestination(AMQPMessageSupport.destination(queueType, properties.getTo()));
jms.setDestination(properties.getTo());
}
if (properties.getSubject() != null) {
jms.setJMSType(properties.getSubject());
@ -347,19 +352,19 @@ public class AmqpCoreConverter {
switch (value) {
case AMQPMessageSupport.QUEUE_TYPE:
org.apache.activemq.artemis.reader.MessageUtil.setJMSReplyTo(jms.getInnerMessage(), ActiveMQDestination.QUEUE_QUALIFIED_PREFIX + properties.getReplyTo());
org.apache.activemq.artemis.reader.MessageUtil.setJMSReplyTo(jms.getInnerMessage(), QUEUE_QUALIFIED_PREFIX + properties.getReplyTo());
break;
case AMQPMessageSupport.TEMP_QUEUE_TYPE:
org.apache.activemq.artemis.reader.MessageUtil.setJMSReplyTo(jms.getInnerMessage(), ActiveMQDestination.TEMP_QUEUE_QUALIFED_PREFIX + properties.getReplyTo());
org.apache.activemq.artemis.reader.MessageUtil.setJMSReplyTo(jms.getInnerMessage(), TEMP_QUEUE_QUALIFED_PREFIX + properties.getReplyTo());
break;
case AMQPMessageSupport.TOPIC_TYPE:
org.apache.activemq.artemis.reader.MessageUtil.setJMSReplyTo(jms.getInnerMessage(), ActiveMQDestination.TOPIC_QUALIFIED_PREFIX + properties.getReplyTo());
org.apache.activemq.artemis.reader.MessageUtil.setJMSReplyTo(jms.getInnerMessage(), TOPIC_QUALIFIED_PREFIX + properties.getReplyTo());
break;
case AMQPMessageSupport.TEMP_TOPIC_TYPE:
org.apache.activemq.artemis.reader.MessageUtil.setJMSReplyTo(jms.getInnerMessage(), ActiveMQDestination.TEMP_TOPIC_QUALIFED_PREFIX + properties.getReplyTo());
org.apache.activemq.artemis.reader.MessageUtil.setJMSReplyTo(jms.getInnerMessage(), TEMP_TOPIC_QUALIFED_PREFIX + properties.getReplyTo());
break;
default:
org.apache.activemq.artemis.reader.MessageUtil.setJMSReplyTo(jms.getInnerMessage(), ActiveMQDestination.QUEUE_QUALIFIED_PREFIX + properties.getReplyTo());
org.apache.activemq.artemis.reader.MessageUtil.setJMSReplyTo(jms.getInnerMessage(), QUEUE_QUALIFIED_PREFIX + properties.getReplyTo());
break;
}
}
@ -410,7 +415,7 @@ public class AmqpCoreConverter {
}
@SuppressWarnings("unchecked")
private static ServerJMSMessage processFooter(ServerJMSMessage jms, Footer footer) throws Exception {
private static CoreMessageWrapper processFooter(CoreMessageWrapper jms, Footer footer) {
if (footer != null && footer.getValue() != null) {
for (Map.Entry<Symbol, Object> entry : (Set<Map.Entry<Symbol, Object>>) footer.getValue().entrySet()) {
String key = entry.getKey().toString();
@ -425,7 +430,7 @@ public class AmqpCoreConverter {
return jms;
}
private static void encodeUnsupportedMessagePropertyType(ServerJMSMessage jms, String key, Object value) throws JMSException {
private static void encodeUnsupportedMessagePropertyType(CoreMessageWrapper jms, String key, Object value) {
final ByteBuf buffer = PooledByteBufAllocator.DEFAULT.heapBuffer();
final EncoderImpl encoder = TLSEncode.getEncoder();
@ -444,7 +449,7 @@ public class AmqpCoreConverter {
}
}
private static void setProperty(javax.jms.Message msg, String key, Object value) throws JMSException {
private static void setProperty(CoreMessageWrapper msg, String key, Object value) {
if (value instanceof UnsignedLong) {
long v = ((UnsignedLong) value).longValue();
msg.setLongProperty(key, v);

210
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/CoreAmqpConverter.java
View File

@ -20,14 +20,6 @@ package org.apache.activemq.artemis.protocol.amqp.converter;
import static org.apache.activemq.artemis.api.core.FilterConstants.NATIVE_MESSAGE_ID;
import static org.apache.activemq.artemis.api.core.Message.EMBEDDED_TYPE;
import static org.apache.activemq.artemis.api.core.Message.HDR_SCHEDULED_DELIVERY_TIME;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.AMQP_DATA;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.AMQP_NULL;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.AMQP_SEQUENCE;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.AMQP_UNKNOWN;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.AMQP_VALUE_BINARY;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.AMQP_VALUE_LIST;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.AMQP_VALUE_STRING;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.EMPTY_BINARY;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.JMS_AMQP_CONTENT_ENCODING;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.JMS_AMQP_CONTENT_TYPE;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.JMS_AMQP_DELIVERY_ANNOTATION_PREFIX;
@ -47,36 +39,23 @@ import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSup
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.JMS_AMQP_REPLYTO_GROUP_ID;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.JMS_DEST_TYPE_MSG_ANNOTATION;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.JMS_REPLY_TO_TYPE_MSG_ANNOTATION;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.SERIALIZED_JAVA_OBJECT_CONTENT_TYPE;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.MESSAGE_DEFAULT_PRIORITY;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.toAddress;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Date;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageEOFException;
import javax.jms.TextMessage;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ICoreMessage;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.persistence.StorageManager;
import org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessage;
import org.apache.activemq.artemis.protocol.amqp.broker.AMQPStandardMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSBytesMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSMapMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSObjectMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSStreamMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSTextMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.ConversionException;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.exceptions.ActiveMQAMQPIllegalStateException;
import org.apache.activemq.artemis.protocol.amqp.util.NettyWritable;
import org.apache.activemq.artemis.protocol.amqp.util.TLSEncode;
@ -86,10 +65,7 @@ import org.apache.qpid.proton.amqp.Binary;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.UnsignedByte;
import org.apache.qpid.proton.amqp.UnsignedInteger;
import org.apache.qpid.proton.amqp.messaging.AmqpSequence;
import org.apache.qpid.proton.amqp.messaging.AmqpValue;
import org.apache.qpid.proton.amqp.messaging.ApplicationProperties;
import org.apache.qpid.proton.amqp.messaging.Data;
import org.apache.qpid.proton.amqp.messaging.DeliveryAnnotations;
import org.apache.qpid.proton.amqp.messaging.Footer;
import org.apache.qpid.proton.amqp.messaging.Header;
@ -118,7 +94,7 @@ public class CoreAmqpConverter {
}
}
public static AMQPMessage fromCore(ICoreMessage coreMessage, StorageManager storageManager) throws Exception {
public static AMQPMessage fromCore(ICoreMessage coreMessage, StorageManager storageManager) throws ConversionException {
if (coreMessage == null) {
return null;
}
@ -129,7 +105,7 @@ public class CoreAmqpConverter {
return (AMQPMessage) message;
}
}
ServerJMSMessage message = ServerJMSMessage.wrapCoreMessage(coreMessage);
CoreMessageWrapper message = CoreMessageWrapper.wrap(coreMessage);
message.decode();
long messageFormat = 0;
@ -140,7 +116,7 @@ public class CoreAmqpConverter {
Map<String, Object> apMap = null;
Map<Symbol, Object> footerMap = null;
Section body = convertBody(message, maMap, properties);
Section body = message.createAMQPSection(maMap, properties);
if (message.getInnerMessage().isDurable()) {
if (header == null) {
@ -149,7 +125,7 @@ public class CoreAmqpConverter {
header.setDurable(true);
}
byte priority = (byte) message.getJMSPriority();
if (priority != javax.jms.Message.DEFAULT_PRIORITY) {
if (priority != MESSAGE_DEFAULT_PRIORITY) {
if (header == null) {
header = new Header();
}
@ -171,15 +147,15 @@ public class CoreAmqpConverter {
properties.setMessageId("ID:" + message.getInnerMessage().getUserID().toString());
}
}
Destination destination = message.getJMSDestination();
SimpleString destination = message.getDestination();
if (destination != null) {
properties.setTo(toAddress(destination));
maMap.put(JMS_DEST_TYPE_MSG_ANNOTATION, AMQPMessageSupport.destinationType(destination));
properties.setTo(toAddress(destination.toString()));
maMap.put(JMS_DEST_TYPE_MSG_ANNOTATION, AMQPMessageSupport.destinationType(destination.toString()));
}
Destination replyTo = message.getJMSReplyTo();
SimpleString replyTo = message.getJMSReplyTo();
if (replyTo != null) {
properties.setReplyTo(toAddress(replyTo));
maMap.put(JMS_REPLY_TO_TYPE_MSG_ANNOTATION, AMQPMessageSupport.destinationType(replyTo));
properties.setReplyTo(toAddress(replyTo.toString()));
maMap.put(JMS_REPLY_TO_TYPE_MSG_ANNOTATION, AMQPMessageSupport.destinationType(replyTo.toString()));
}
Object correlationID = message.getInnerMessage().getCorrelationID();
@ -194,7 +170,7 @@ public class CoreAmqpConverter {
properties.setCorrelationId(correlationID);
}
long expiration = message.getJMSExpiration();
long expiration = message.getExpiration();
if (expiration != 0) {
long ttl = expiration - System.currentTimeMillis();
if (ttl < 0) {
@ -393,162 +369,4 @@ public class CoreAmqpConverter {
return decodedType;
}
private static Section convertBody(ServerJMSMessage message, Map<Symbol, Object> maMap, Properties properties) throws JMSException {
Section body = null;
short orignalEncoding = AMQP_UNKNOWN;
try {
orignalEncoding = message.getShortProperty(JMS_AMQP_ORIGINAL_ENCODING);
} catch (Exception ex) {
// Ignore and stick with UNKNOWN
}
if (message instanceof ServerJMSBytesMessage) {
Binary payload = getBinaryFromMessageBody((ServerJMSBytesMessage) message);
maMap.put(AMQPMessageSupport.JMS_MSG_TYPE, AMQPMessageSupport.JMS_BYTES_MESSAGE);
if (payload == null) {
payload = EMPTY_BINARY;
}
switch (orignalEncoding) {
case AMQP_NULL:
break;
case AMQP_VALUE_BINARY:
body = new AmqpValue(payload);
break;
case AMQP_DATA:
case AMQP_UNKNOWN:
default:
body = new Data(payload);
break;
}
} else if (message instanceof ServerJMSTextMessage) {
String text = (((TextMessage) message).getText());
switch (orignalEncoding) {
case AMQP_NULL:
break;
case AMQP_DATA:
if (text == null) {
body = new Data(EMPTY_BINARY);
} else {
body = new Data(new Binary(text.getBytes(StandardCharsets.UTF_8)));
}
break;
case AMQP_VALUE_STRING:
case AMQP_UNKNOWN:
default:
body = new AmqpValue(text);
break;
}
maMap.put(AMQPMessageSupport.JMS_MSG_TYPE, AMQPMessageSupport.JMS_TEXT_MESSAGE);
} else if (message instanceof ServerJMSMapMessage) {
body = new AmqpValue(getMapFromMessageBody((ServerJMSMapMessage) message));
maMap.put(AMQPMessageSupport.JMS_MSG_TYPE, AMQPMessageSupport.JMS_MAP_MESSAGE);
} else if (message instanceof ServerJMSStreamMessage) {
maMap.put(AMQPMessageSupport.JMS_MSG_TYPE, AMQPMessageSupport.JMS_STREAM_MESSAGE);
ArrayList<Object> list = new ArrayList<>();
final ServerJMSStreamMessage m = (ServerJMSStreamMessage) message;
try {
while (true) {
list.add(m.readObject());
}
} catch (MessageEOFException e) {
}
switch (orignalEncoding) {
case AMQP_SEQUENCE:
body = new AmqpSequence(list);
break;
case AMQP_VALUE_LIST:
case AMQP_UNKNOWN:
default:
body = new AmqpValue(list);
break;
}
} else if (message instanceof ServerJMSObjectMessage) {
properties.setContentType(AMQPMessageSupport.SERIALIZED_JAVA_OBJECT_CONTENT_TYPE);
maMap.put(AMQPMessageSupport.JMS_MSG_TYPE, AMQPMessageSupport.JMS_OBJECT_MESSAGE);
Binary payload = getBinaryFromMessageBody((ServerJMSObjectMessage) message);
if (payload == null) {
payload = EMPTY_BINARY;
}
switch (orignalEncoding) {
case AMQP_VALUE_BINARY:
body = new AmqpValue(payload);
break;
case AMQP_DATA:
case AMQP_UNKNOWN:
default:
body = new Data(payload);
break;
}
// For a non-AMQP message we tag the outbound content type as containing
// a serialized Java object so that an AMQP client has a hint as to what
// we are sending it.
if (!message.propertyExists(JMS_AMQP_CONTENT_TYPE)) {
message.setStringProperty(JMS_AMQP_CONTENT_TYPE, SERIALIZED_JAVA_OBJECT_CONTENT_TYPE.toString());
}
} else {
maMap.put(AMQPMessageSupport.JMS_MSG_TYPE, AMQPMessageSupport.JMS_MESSAGE);
// If this is not an AMQP message that was converted then the original encoding
// will be unknown so we check for special cases of messages with special data
// encoded into the server message body.
ICoreMessage internalMessage = message.getInnerMessage();
// this will represent a readOnly buffer for the message
ActiveMQBuffer buffer = internalMessage.getDataBuffer();
try {
// the buffer may be completely empty (e.g. if the original AMQP message had a null body)
if (buffer.readableBytes() > 0) {
Object s = buffer.readNullableSimpleString();
if (s != null) {
body = new AmqpValue(s.toString());
}
}
} catch (Throwable e) {
logger.debug("Exception ignored during conversion", e.getMessage(), e);
body = new AmqpValue("Conversion to AMQP error: " + e.getMessage());
}
}
return body;
}
private static Binary getBinaryFromMessageBody(ServerJMSBytesMessage message) throws JMSException {
byte[] data = new byte[(int) message.getBodyLength()];
message.readBytes(data);
message.reset(); // Need to reset after readBytes or future readBytes
return new Binary(data);
}
private static Binary getBinaryFromMessageBody(ServerJMSObjectMessage message) throws JMSException {
return message.getSerializedForm();
}
private static Map<String, Object> getMapFromMessageBody(ServerJMSMapMessage message) throws JMSException {
final HashMap<String, Object> map = new LinkedHashMap<>();
@SuppressWarnings("unchecked")
final Enumeration<String> names = message.getMapNames();
while (names.hasMoreElements()) {
String key = names.nextElement();
Object value = message.getObject(key);
if (value instanceof byte[]) {
value = new Binary((byte[]) value);
}
map.put(key, value);
}
return map;
}
}

28
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/coreWrapper/ConversionException.java Normal file
View File

@ -0,0 +1,28 @@
/*
* 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.activemq.artemis.protocol.amqp.converter.coreWrapper;
public class ConversionException extends Exception {
public ConversionException(String message) {
super(message);
}
public ConversionException(String message, Throwable cause) {
super(message, cause);
}
}

137
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/jms/ServerJMSBytesMessage.java → artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/coreWrapper/CoreBytesMessageWrapper.java
View File

@ -14,13 +14,24 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.protocol.amqp.converter.jms;
package org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper;
import javax.jms.BytesMessage;
import javax.jms.JMSException;
import java.util.Map;
import org.apache.activemq.artemis.api.core.ICoreMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport;
import org.apache.qpid.proton.amqp.Binary;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.messaging.AmqpValue;
import org.apache.qpid.proton.amqp.messaging.Data;
import org.apache.qpid.proton.amqp.messaging.Properties;
import org.apache.qpid.proton.amqp.messaging.Section;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.AMQP_DATA;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.AMQP_NULL;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.AMQP_UNKNOWN;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.AMQP_VALUE_BINARY;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.EMPTY_BINARY;
import static org.apache.activemq.artemis.reader.BytesMessageUtil.bytesMessageReset;
import static org.apache.activemq.artemis.reader.BytesMessageUtil.bytesReadBoolean;
import static org.apache.activemq.artemis.reader.BytesMessageUtil.bytesReadByte;
@ -46,160 +57,164 @@ import static org.apache.activemq.artemis.reader.BytesMessageUtil.bytesWriteObje
import static org.apache.activemq.artemis.reader.BytesMessageUtil.bytesWriteShort;
import static org.apache.activemq.artemis.reader.BytesMessageUtil.bytesWriteUTF;
public class ServerJMSBytesMessage extends ServerJMSMessage implements BytesMessage {
public class CoreBytesMessageWrapper extends CoreMessageWrapper {
public ServerJMSBytesMessage(ICoreMessage message) {
public CoreBytesMessageWrapper(ICoreMessage message) {
super(message);
}
protected static Binary getBinaryFromMessageBody(CoreBytesMessageWrapper message) {
byte[] data = new byte[(int) message.getBodyLength()];
message.readBytes(data);
message.reset(); // Need to reset after readBytes or future readBytes
return new Binary(data);
}
@Override
public long getBodyLength() throws JMSException {
public Section createAMQPSection(Map<Symbol, Object> maMap, Properties properties) {
short orignalEncoding = getOrignalEncoding();
Binary payload = getBinaryFromMessageBody(this);
maMap.put(AMQPMessageSupport.JMS_MSG_TYPE, AMQPMessageSupport.JMS_BYTES_MESSAGE);
if (payload == null) {
payload = EMPTY_BINARY;
}
switch (orignalEncoding) {
case AMQP_NULL:
return null;
case AMQP_VALUE_BINARY:
return new AmqpValue(payload);
case AMQP_DATA:
case AMQP_UNKNOWN:
default:
return new Data(payload);
}
}
public long getBodyLength() {
return message.getBodyBufferSize();
}
@Override
public boolean readBoolean() throws JMSException {
public boolean readBoolean() {
return bytesReadBoolean(getReadBodyBuffer());
}
@Override
public byte readByte() throws JMSException {
public byte readByte() {
return bytesReadByte(getReadBodyBuffer());
}
@Override
public int readUnsignedByte() throws JMSException {
public int readUnsignedByte() {
return bytesReadUnsignedByte(getReadBodyBuffer());
}
@Override
public short readShort() throws JMSException {
public short readShort() {
return bytesReadShort(getReadBodyBuffer());
}
@Override
public int readUnsignedShort() throws JMSException {
public int readUnsignedShort() {
return bytesReadUnsignedShort(getReadBodyBuffer());
}
@Override
public char readChar() throws JMSException {
public char readChar() {
return bytesReadChar(getReadBodyBuffer());
}
@Override
public int readInt() throws JMSException {
public int readInt() {
return bytesReadInt(getReadBodyBuffer());
}
@Override
public long readLong() throws JMSException {
public long readLong() {
return bytesReadLong(getReadBodyBuffer());
}
@Override
public float readFloat() throws JMSException {
public float readFloat() {
return bytesReadFloat(getReadBodyBuffer());
}
@Override
public double readDouble() throws JMSException {
public double readDouble() {
return bytesReadDouble(getReadBodyBuffer());
}
@Override
public String readUTF() throws JMSException {
public String readUTF() {
return bytesReadUTF(getReadBodyBuffer());
}
@Override
public int readBytes(byte[] value) throws JMSException {
public int readBytes(byte[] value) {
return bytesReadBytes(getReadBodyBuffer(), value);
}
@Override
public int readBytes(byte[] value, int length) throws JMSException {
public int readBytes(byte[] value, int length) {
return bytesReadBytes(getReadBodyBuffer(), value, length);
}
@Override
public void writeBoolean(boolean value) throws JMSException {
public void writeBoolean(boolean value) {
bytesWriteBoolean(getWriteBodyBuffer(), value);
}
@Override
public void writeByte(byte value) throws JMSException {
public void writeByte(byte value) {
bytesWriteByte(getWriteBodyBuffer(), value);
}
@Override
public void writeShort(short value) throws JMSException {
public void writeShort(short value) {
bytesWriteShort(getWriteBodyBuffer(), value);
}
@Override
public void writeChar(char value) throws JMSException {
public void writeChar(char value) {
bytesWriteChar(getWriteBodyBuffer(), value);
}
@Override
public void writeInt(int value) throws JMSException {
public void writeInt(int value) {
bytesWriteInt(getWriteBodyBuffer(), value);
}
@Override
public void writeLong(long value) throws JMSException {
public void writeLong(long value) {
bytesWriteLong(getWriteBodyBuffer(), value);
}
@Override
public void writeFloat(float value) throws JMSException {
public void writeFloat(float value) {
bytesWriteFloat(getWriteBodyBuffer(), value);
}
@Override
public void writeDouble(double value) throws JMSException {
public void writeDouble(double value) {
bytesWriteDouble(getWriteBodyBuffer(), value);
}
@Override
public void writeUTF(String value) throws JMSException {
public void writeUTF(String value) {
bytesWriteUTF(getWriteBodyBuffer(), value);
}
@Override
public void writeBytes(byte[] value) throws JMSException {
public void writeBytes(byte[] value) {
bytesWriteBytes(getWriteBodyBuffer(), value);
}
@Override
public void writeBytes(byte[] value, int offset, int length) throws JMSException {
public void writeBytes(byte[] value, int offset, int length) {
bytesWriteBytes(getWriteBodyBuffer(), value, offset, length);
}
@Override
public void writeObject(Object value) throws JMSException {
public void writeObject(Object value) throws ConversionException {
if (!bytesWriteObject(getWriteBodyBuffer(), value)) {
throw new JMSException("Can't make conversion of " + value + " to any known type");
throw new ConversionException("Can't make conversion of " + value + " to any known type");
}
}
@Override
public void encode() throws Exception {
public void encode() {
super.encode();
// this is to make sure we encode the body-length before it's persisted
getBodyLength();
}
@Override
public void decode() throws Exception {
public void decode() {
super.decode();
}
@Override
public void reset() throws JMSException {
public void reset() {
if (!message.isLargeMessage()) {
bytesMessageReset(getReadBodyBuffer());
bytesMessageReset(getWriteBodyBuffer());

225
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/coreWrapper/CoreMapMessageWrapper.java Normal file
View File

@ -0,0 +1,225 @@
/*
* 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.activemq.artemis.protocol.amqp.converter.coreWrapper;
import java.util.Collections;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import org.apache.activemq.artemis.api.core.ActiveMQPropertyConversionException;
import org.apache.activemq.artemis.api.core.ICoreMessage;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport;
import org.apache.activemq.artemis.utils.collections.TypedProperties;
import org.apache.qpid.proton.amqp.Binary;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.UnsignedByte;
import org.apache.qpid.proton.amqp.UnsignedInteger;
import org.apache.qpid.proton.amqp.UnsignedLong;
import org.apache.qpid.proton.amqp.UnsignedShort;
import org.apache.qpid.proton.amqp.messaging.AmqpValue;
import org.apache.qpid.proton.amqp.messaging.Properties;
import org.apache.qpid.proton.amqp.messaging.Section;
import static org.apache.activemq.artemis.reader.MapMessageUtil.readBodyMap;
import static org.apache.activemq.artemis.reader.MapMessageUtil.writeBodyMap;
public final class CoreMapMessageWrapper extends CoreMessageWrapper {
public static final byte TYPE = Message.MAP_TYPE;
private final TypedProperties map = new TypedProperties();
/*
* This constructor is used to construct messages prior to sending
*/
public CoreMapMessageWrapper(ICoreMessage message) {
super(message);
}
private static Map<String, Object> getMapFromMessageBody(CoreMapMessageWrapper message) {
final HashMap<String, Object> map = new LinkedHashMap<>();
@SuppressWarnings("unchecked")
final Enumeration<String> names = message.getMapNames();
while (names.hasMoreElements()) {
String key = names.nextElement();
Object value = message.getObject(key);
if (value instanceof byte[]) {
value = new Binary((byte[]) value);
}
map.put(key, value);
}
return map;
}
@Override
public Section createAMQPSection(Map<Symbol, Object> maMap, Properties properties) {
maMap.put(AMQPMessageSupport.JMS_MSG_TYPE, AMQPMessageSupport.JMS_MAP_MESSAGE);
return new AmqpValue(getMapFromMessageBody(this));
}
public void setBoolean(final String name, final boolean value) {
map.putBooleanProperty(new SimpleString(name), value);
}
public void setByte(final String name, final byte value) {
map.putByteProperty(new SimpleString(name), value);
}
public void setShort(final String name, final short value) {
map.putShortProperty(new SimpleString(name), value);
}
public void setChar(final String name, final char value) {
map.putCharProperty(new SimpleString(name), value);
}
public void setInt(final String name, final int value) {
map.putIntProperty(new SimpleString(name), value);
}
public void setLong(final String name, final long value) {
map.putLongProperty(new SimpleString(name), value);
}
public void setFloat(final String name, final float value) {
map.putFloatProperty(new SimpleString(name), value);
}
public void setDouble(final String name, final double value) {
map.putDoubleProperty(new SimpleString(name), value);
}
public void setString(final String name, final String value) {
map.putSimpleStringProperty(new SimpleString(name), SimpleString.toSimpleString(value));
}
public void setBytes(final String name, final byte[] value) {
map.putBytesProperty(new SimpleString(name), value);
}
public void setBytes(final String name, final byte[] value, final int offset, final int length) {
byte[] newBytes = new byte[length];
System.arraycopy(value, offset, newBytes, 0, length);
map.putBytesProperty(new SimpleString(name), newBytes);
}
public void setObject(final String name, final Object value) throws ActiveMQPropertyConversionException {
// primitives and String
Object val = value;
if (value instanceof UnsignedInteger) {
val = ((UnsignedInteger) value).intValue();
} else if (value instanceof UnsignedShort) {
val = ((UnsignedShort) value).shortValue();
} else if (value instanceof UnsignedByte) {
val = ((UnsignedByte) value).byteValue();
} else if (value instanceof UnsignedLong) {
val = ((UnsignedLong) value).longValue();
}
TypedProperties.setObjectProperty(new SimpleString(name), val, map);
}
public boolean getBoolean(final String name) throws ActiveMQPropertyConversionException {
return map.getBooleanProperty(new SimpleString(name));
}
public byte getByte(final String name) throws ActiveMQPropertyConversionException {
return map.getByteProperty(new SimpleString(name));
}
public short getShort(final String name) throws ActiveMQPropertyConversionException {
return map.getShortProperty(new SimpleString(name));
}
public char getChar(final String name) throws ActiveMQPropertyConversionException {
return map.getCharProperty(new SimpleString(name));
}
public int getInt(final String name) throws ActiveMQPropertyConversionException {
return map.getIntProperty(new SimpleString(name));
}
public long getLong(final String name) throws ActiveMQPropertyConversionException {
return map.getLongProperty(new SimpleString(name));
}
public float getFloat(final String name) throws ActiveMQPropertyConversionException {
return map.getFloatProperty(new SimpleString(name));
}
public double getDouble(final String name) throws ActiveMQPropertyConversionException {
return map.getDoubleProperty(new SimpleString(name));
}
public String getString(final String name) throws ActiveMQPropertyConversionException {
SimpleString str = map.getSimpleStringProperty(new SimpleString(name));
if (str == null) {
return null;
} else {
return str.toString();
}
}
public byte[] getBytes(final String name) throws ActiveMQPropertyConversionException {
return map.getBytesProperty(new SimpleString(name));
}
public Object getObject(final String name) {
Object val = map.getProperty(new SimpleString(name));
if (val instanceof SimpleString) {
val = ((SimpleString) val).toString();
}
return val;
}
public Enumeration getMapNames() {
return Collections.enumeration(map.getMapNames());
}
public boolean itemExists(final String name) {
return map.containsProperty(new SimpleString(name));
}
@Override
public void clearBody() {
super.clearBody();
map.clear();
}
@Override
public void encode() {
super.encode();
writeBodyMap(getWriteBodyBuffer(), map);
}
@Override
public void decode() {
super.decode();
readBodyMap(getReadBodyBuffer(), map);
}
}

351
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/coreWrapper/CoreMessageWrapper.java Normal file
View File

@ -0,0 +1,351 @@
/*
* 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.activemq.artemis.protocol.amqp.converter.coreWrapper;
import java.util.Collections;
import java.util.Enumeration;
import java.util.Map;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ICoreMessage;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport;
import org.apache.activemq.artemis.reader.MessageUtil;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.messaging.AmqpValue;
import org.apache.qpid.proton.amqp.messaging.Properties;
import org.apache.qpid.proton.amqp.messaging.Section;
import org.jboss.logging.Logger;
import static org.apache.activemq.artemis.api.core.FilterConstants.NATIVE_MESSAGE_ID;
import static org.apache.activemq.artemis.api.core.Message.BYTES_TYPE;
import static org.apache.activemq.artemis.api.core.Message.MAP_TYPE;
import static org.apache.activemq.artemis.api.core.Message.OBJECT_TYPE;
import static org.apache.activemq.artemis.api.core.Message.STREAM_TYPE;
import static org.apache.activemq.artemis.api.core.Message.TEXT_TYPE;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.AMQP_UNKNOWN;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.JMS_AMQP_ORIGINAL_ENCODING;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.NON_PERSISTENT;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.PERSISTENT;
public class CoreMessageWrapper {
private static final Logger logger = Logger.getLogger(CoreMessageWrapper.class);
protected final ICoreMessage message;
private ActiveMQBuffer readBodyBuffer;
public CoreMessageWrapper(ICoreMessage message) {
this.message = message;
}
public static CoreMessageWrapper wrap(ICoreMessage wrapped) {
switch (wrapped.getType()) {
case STREAM_TYPE:
return new CoreStreamMessageWrapper(wrapped);
case BYTES_TYPE:
return new CoreBytesMessageWrapper(wrapped);
case MAP_TYPE:
return new CoreMapMessageWrapper(wrapped);
case TEXT_TYPE:
return new CoreTextMessageWrapper(wrapped);
case OBJECT_TYPE:
return new CoreObjectMessageWrapper(wrapped);
default:
return new CoreMessageWrapper(wrapped);
}
}
protected short getOrignalEncoding() {
short orignalEncoding = AMQP_UNKNOWN;
try {
orignalEncoding = message.getShortProperty(JMS_AMQP_ORIGINAL_ENCODING);
} catch (Exception ex) {
logger.debug(ex.getMessage(), ex);
// Ignore and stick with UNKNOWN
}
return orignalEncoding;
}
// returns the AMQP Section for the message type
public Section createAMQPSection(Map<Symbol, Object> maMap, Properties properties) throws ConversionException {
maMap.put(AMQPMessageSupport.JMS_MSG_TYPE, AMQPMessageSupport.JMS_MESSAGE);
// If this is not an AMQP message that was converted then the original encoding
// will be unknown so we check for special cases of messages with special data
// encoded into the server message body.
ICoreMessage internalMessage = this.getInnerMessage();
// this will represent a readOnly buffer for the message
ActiveMQBuffer buffer = internalMessage.getDataBuffer();
try {
// the buffer may be completely empty (e.g. if the original AMQP message had a null body)
if (buffer.readableBytes() > 0) {
Object s = buffer.readNullableSimpleString();
if (s != null) {
return new AmqpValue(s.toString());
}
}
return null;
} catch (Throwable e) {
logger.debug("Exception ignored during conversion", e.getMessage(), e);
return new AmqpValue("Conversion to AMQP error: " + e.getMessage());
}
}
public ICoreMessage getInnerMessage() {
return message;
}
/**
* When reading we use a protected copy so multi-threads can work fine
*/
protected ActiveMQBuffer getReadBodyBuffer() {
if (readBodyBuffer == null) {
// to avoid clashes between multiple threads
readBodyBuffer = message.getDataBuffer();
}
return readBodyBuffer;
}
/**
* When writing on the conversion we use the buffer directly
*/
protected ActiveMQBuffer getWriteBodyBuffer() {
readBodyBuffer = null; // it invalidates this buffer if anything is written
return message.getBodyBuffer();
}
public final String getJMSMessageID() {
if (message.containsProperty(NATIVE_MESSAGE_ID)) {
return getStringProperty(NATIVE_MESSAGE_ID);
}
return null;
}
public final void setJMSMessageID(String id) {
if (id != null) {
message.putStringProperty(NATIVE_MESSAGE_ID, id);
}
}
public final long getJMSTimestamp() {
return message.getTimestamp();
}
public final void setJMSTimestamp(long timestamp) {
message.setTimestamp(timestamp);
}
public final byte[] getJMSCorrelationIDAsBytes() {
return MessageUtil.getJMSCorrelationIDAsBytes(message);
}
public final void setJMSCorrelationIDAsBytes(byte[] correlationID) {
message.setCorrelationID(correlationID);
}
public final String getJMSCorrelationID() {
Object correlationID = message.getCorrelationID();
if (correlationID instanceof String) {
return ((String) correlationID);
} else if (correlationID != null) {
return String.valueOf(correlationID);
} else {
return null;
}
}
public final void setJMSCorrelationID(String correlationID) {
message.setCorrelationID(correlationID);
}
public final SimpleString getJMSReplyTo() {
return MessageUtil.getJMSReplyTo(message);
}
public final void setJMSReplyTo(String replyTo) {
MessageUtil.setJMSReplyTo(message, SimpleString.toSimpleString(replyTo));
}
public SimpleString getDestination() {
if (message.getAddress() == null || message.getAddress().isEmpty()) {
return null;
}
return SimpleString.toSimpleString(AMQPMessageSupport.destination(message.getRoutingType(), message.getAddress()));
}
public final void setDestination(String destination) {
message.setAddress(destination);
}
public final int getJMSDeliveryMode() {
return message.isDurable() ? PERSISTENT : NON_PERSISTENT;
}
public final void setDeliveryMode(int deliveryMode) throws ConversionException {
switch (deliveryMode) {
case PERSISTENT:
message.setDurable(true);
break;
case NON_PERSISTENT:
message.setDurable(false);
break;
default:
// There shouldn't be any other values used
throw new ConversionException("Invalid mode " + deliveryMode);
}
}
public final String getJMSType() {
return MessageUtil.getJMSType(message);
}
public final void setJMSType(String type) {
MessageUtil.setJMSType(message, type);
}
public final long getExpiration() {
return message.getExpiration();
}
public final void setJMSExpiration(long expiration) {
message.setExpiration(expiration);
}
public final int getJMSPriority() {
return message.getPriority();
}
public final void setJMSPriority(int priority) {
message.setPriority((byte) priority);
}
public final void clearProperties() {
MessageUtil.clearProperties(message);
}
public final boolean propertyExists(String name) {
return MessageUtil.propertyExists(message, name);
}
public final boolean getBooleanProperty(String name) {
return message.getBooleanProperty(name);
}
public final byte getByteProperty(String name) {
return message.getByteProperty(name);
}
public final short getShortProperty(String name) {
return message.getShortProperty(name);
}
public final int getIntProperty(String name) {
return MessageUtil.getIntProperty(message, name);
}
public final long getLongProperty(String name) {
return MessageUtil.getLongProperty(message, name);
}
public final float getFloatProperty(String name) {
return message.getFloatProperty(name);
}
public final double getDoubleProperty(String name) {
return message.getDoubleProperty(name);
}
public final String getStringProperty(String name) {
return MessageUtil.getStringProperty(message, name);
}
public final Object getObjectProperty(String name) {
return MessageUtil.getObjectProperty(message, name);
}
public final Enumeration getPropertyNames() {
return Collections.enumeration(MessageUtil.getPropertyNames(message));
}
public final void setBooleanProperty(String name, boolean value) {
message.putBooleanProperty(name, value);
}
public final void setByteProperty(String name, byte value) {
message.putByteProperty(name, value);
}
public final void setShortProperty(String name, short value) {
message.putShortProperty(name, value);
}
public final void setIntProperty(String name, int value) {
MessageUtil.setIntProperty(message, name, value);
}
public final void setLongProperty(String name, long value) {
MessageUtil.setLongProperty(message, name, value);
}
public final void setFloatProperty(String name, float value) {
message.putFloatProperty(name, value);
}
public final void setDoubleProperty(String name, double value) {
message.putDoubleProperty(name, value);
}
public final void setStringProperty(String name, String value) {
MessageUtil.setStringProperty(message, name, value);
}
public final void setObjectProperty(String name, Object value) {
MessageUtil.setObjectProperty(message, name, value);
}
public void clearBody() {
message.getBodyBuffer().clear();
}
/**
* Encode the body into the internal message
*/
public void encode() {
if (!message.isLargeMessage()) {
message.getBodyBuffer().resetReaderIndex();
}
}
public void decode() {
if (!message.isLargeMessage()) {
message.getBodyBuffer().resetReaderIndex();
}
}
}

106
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/coreWrapper/CoreObjectMessageWrapper.java Normal file
View File

@ -0,0 +1,106 @@
/**
* 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
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* 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.activemq.artemis.protocol.amqp.converter.coreWrapper;
import java.util.Map;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ICoreMessage;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport;
import org.apache.qpid.proton.amqp.Binary;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.messaging.AmqpValue;
import org.apache.qpid.proton.amqp.messaging.Data;
import org.apache.qpid.proton.amqp.messaging.Properties;
import org.apache.qpid.proton.amqp.messaging.Section;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.AMQP_DATA;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.AMQP_UNKNOWN;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.AMQP_VALUE_BINARY;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.EMPTY_BINARY;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.JMS_AMQP_CONTENT_TYPE;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.SERIALIZED_JAVA_OBJECT_CONTENT_TYPE;
public class CoreObjectMessageWrapper extends CoreMessageWrapper {
public static final byte TYPE = Message.OBJECT_TYPE;
private Binary payload;
private static Binary getBinaryFromMessageBody(CoreObjectMessageWrapper message) {
return message.getSerializedForm();
}
@Override
public Section createAMQPSection(Map<Symbol, Object> maMap, Properties properties) throws ConversionException {
properties.setContentType(AMQPMessageSupport.SERIALIZED_JAVA_OBJECT_CONTENT_TYPE);
maMap.put(AMQPMessageSupport.JMS_MSG_TYPE, AMQPMessageSupport.JMS_OBJECT_MESSAGE);
Binary payload = getBinaryFromMessageBody(this);
if (payload == null) {
payload = EMPTY_BINARY;
}
// For a non-AMQP message we tag the outbound content type as containing
// a serialized Java object so that an AMQP client has a hint as to what
// we are sending it.
if (!this.propertyExists(JMS_AMQP_CONTENT_TYPE)) {
this.setStringProperty(JMS_AMQP_CONTENT_TYPE, SERIALIZED_JAVA_OBJECT_CONTENT_TYPE.toString());
}
switch (getOrignalEncoding()) {
case AMQP_VALUE_BINARY:
return new AmqpValue(payload);
case AMQP_DATA:
case AMQP_UNKNOWN:
default:
return new Data(payload);
}
}
public CoreObjectMessageWrapper(ICoreMessage message) {
super(message);
}
public void setSerializedForm(Binary payload) {
this.payload = payload;
}
public Binary getSerializedForm() {
return payload;
}
@Override
public void encode() {
super.encode();
getInnerMessage().getBodyBuffer().writeInt(payload.getLength());
getInnerMessage().getBodyBuffer().writeBytes(payload.getArray(), payload.getArrayOffset(), payload.getLength());
}
@Override
public void decode() {
super.decode();
ActiveMQBuffer buffer = getInnerMessage().getDataBuffer();
int size = buffer.readInt();
byte[] bytes = new byte[size];
buffer.readBytes(bytes);
payload = new Binary(bytes);
}
}

321
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/jms/ServerJMSStreamMessage.java → artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/coreWrapper/CoreStreamMessageWrapper.java
View File

@ -14,18 +14,25 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.protocol.amqp.converter.jms;
package org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper;
import javax.jms.JMSException;
import javax.jms.MessageEOFException;
import javax.jms.MessageFormatException;
import javax.jms.StreamMessage;
import java.util.ArrayList;
import java.util.Map;
import org.apache.activemq.artemis.api.core.ICoreMessage;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.Pair;
import org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport;
import org.apache.activemq.artemis.utils.DataConstants;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.messaging.AmqpSequence;
import org.apache.qpid.proton.amqp.messaging.AmqpValue;
import org.apache.qpid.proton.amqp.messaging.Properties;
import org.apache.qpid.proton.amqp.messaging.Section;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.AMQP_SEQUENCE;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.AMQP_UNKNOWN;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.AMQP_VALUE_LIST;
import static org.apache.activemq.artemis.reader.StreamMessageUtil.streamReadBoolean;
import static org.apache.activemq.artemis.reader.StreamMessageUtil.streamReadByte;
import static org.apache.activemq.artemis.reader.StreamMessageUtil.streamReadBytes;
@ -38,132 +45,140 @@ import static org.apache.activemq.artemis.reader.StreamMessageUtil.streamReadObj
import static org.apache.activemq.artemis.reader.StreamMessageUtil.streamReadShort;
import static org.apache.activemq.artemis.reader.StreamMessageUtil.streamReadString;
public final class ServerJMSStreamMessage extends ServerJMSMessage implements StreamMessage {
public final class CoreStreamMessageWrapper extends CoreMessageWrapper {
public static final byte TYPE = Message.STREAM_TYPE;
private int bodyLength = 0;
public ServerJMSStreamMessage(ICoreMessage message) {
super(message);
}
// StreamMessage implementation ----------------------------------
@Override
public boolean readBoolean() throws JMSException {
try {
return streamReadBoolean(getReadBodyBuffer());
} catch (IllegalStateException e) {
throw new MessageFormatException(e.getMessage());
} catch (IndexOutOfBoundsException e) {
throw new MessageEOFException("");
}
}
@Override
public byte readByte() throws JMSException {
try {
return streamReadByte(getReadBodyBuffer());
} catch (IllegalStateException e) {
throw new MessageFormatException(e.getMessage());
} catch (IndexOutOfBoundsException e) {
throw new MessageEOFException("");
}
}
@Override
public short readShort() throws JMSException {
try {
return streamReadShort(getReadBodyBuffer());
} catch (IllegalStateException e) {
throw new MessageFormatException(e.getMessage());
} catch (IndexOutOfBoundsException e) {
throw new MessageEOFException("");
}
}
@Override
public char readChar() throws JMSException {
try {
return streamReadChar(getReadBodyBuffer());
} catch (IllegalStateException e) {
throw new MessageFormatException(e.getMessage());
} catch (IndexOutOfBoundsException e) {
throw new MessageEOFException("");
}
}
@Override
public int readInt() throws JMSException {
try {
return streamReadInteger(getReadBodyBuffer());
} catch (IllegalStateException e) {
throw new MessageFormatException(e.getMessage());
} catch (IndexOutOfBoundsException e) {
throw new MessageEOFException("");
}
}
@Override
public long readLong() throws JMSException {
try {
return streamReadLong(getReadBodyBuffer());
} catch (IllegalStateException e) {
throw new MessageFormatException(e.getMessage());
} catch (IndexOutOfBoundsException e) {
throw new MessageEOFException("");
}
}
@Override
public float readFloat() throws JMSException {
try {
return streamReadFloat(getReadBodyBuffer());
} catch (IllegalStateException e) {
throw new MessageFormatException(e.getMessage());
} catch (IndexOutOfBoundsException e) {
throw new MessageEOFException("");
}
}
@Override
public double readDouble() throws JMSException {
try {
return streamReadDouble(getReadBodyBuffer());
} catch (IllegalStateException e) {
throw new MessageFormatException(e.getMessage());
} catch (IndexOutOfBoundsException e) {
throw new MessageEOFException("");
}
}
@Override
public String readString() throws JMSException {
try {
return streamReadString(getReadBodyBuffer());
} catch (IllegalStateException e) {
throw new MessageFormatException(e.getMessage());
} catch (IndexOutOfBoundsException e) {
throw new MessageEOFException("");
}
}
/**
* len here is used to control how many more bytes to read
*/
private int len = 0;
public CoreStreamMessageWrapper(ICoreMessage message) {
super(message);
}
@Override
public int readBytes(final byte[] value) throws JMSException {
public Section createAMQPSection(Map<Symbol, Object> maMap, Properties properties) throws ConversionException {
maMap.put(AMQPMessageSupport.JMS_MSG_TYPE, AMQPMessageSupport.JMS_STREAM_MESSAGE);
ArrayList<Object> list = new ArrayList<>();
try {
while (true) {
list.add(readObject());
}
} catch (MessageEOFException e) {
}
switch (getOrignalEncoding()) {
case AMQP_SEQUENCE:
return new AmqpSequence(list);
case AMQP_VALUE_LIST:
case AMQP_UNKNOWN:
default:
return new AmqpValue(list);
}
}
public boolean readBoolean() throws MessageEOFException, ConversionException {
try {
return streamReadBoolean(getReadBodyBuffer());
} catch (IllegalStateException e) {
throw new ConversionException(e.getMessage(), e);
} catch (IndexOutOfBoundsException e) {
throw new MessageEOFException();
}
}
public byte readByte() throws MessageEOFException, ConversionException {
try {
return streamReadByte(getReadBodyBuffer());
} catch (IllegalStateException e) {
throw new ConversionException(e.getMessage(), e);
} catch (IndexOutOfBoundsException e) {
throw new MessageEOFException();
}
}
public short readShort() throws MessageEOFException, ConversionException {
try {
return streamReadShort(getReadBodyBuffer());
} catch (IllegalStateException e) {
throw new ConversionException(e.getMessage(), e);
} catch (IndexOutOfBoundsException e) {
throw new MessageEOFException();
}
}
public char readChar() throws MessageEOFException, ConversionException {
try {
return streamReadChar(getReadBodyBuffer());
} catch (IllegalStateException e) {
throw new ConversionException(e.getMessage(), e);
} catch (IndexOutOfBoundsException e) {
throw new MessageEOFException();
}
}
public int readInt() throws MessageEOFException, ConversionException {
try {
return streamReadInteger(getReadBodyBuffer());
} catch (IllegalStateException e) {
throw new ConversionException(e.getMessage(), e);
} catch (IndexOutOfBoundsException e) {
throw new MessageEOFException();
}
}
public long readLong() throws ConversionException, MessageEOFException {
try {
return streamReadLong(getReadBodyBuffer());
} catch (IllegalStateException e) {
throw new ConversionException(e.getMessage(), e);
} catch (IndexOutOfBoundsException e) {
throw new MessageEOFException();
}
}
public float readFloat() throws ConversionException, MessageEOFException {
try {
return streamReadFloat(getReadBodyBuffer());
} catch (IllegalStateException e) {
throw new ConversionException(e.getMessage(), e);
} catch (IndexOutOfBoundsException e) {
throw new MessageEOFException();
}
}
public double readDouble() throws ConversionException, MessageEOFException {
try {
return streamReadDouble(getReadBodyBuffer());
} catch (IllegalStateException e) {
throw new ConversionException(e.getMessage(), e);
} catch (IndexOutOfBoundsException e) {
throw new MessageEOFException();
}
}
public String readString() throws ConversionException, MessageEOFException {
try {
return streamReadString(getReadBodyBuffer());
} catch (IllegalStateException e) {
throw new ConversionException(e.getMessage(), e);
} catch (IndexOutOfBoundsException e) {
throw new MessageEOFException();
}
}
public int readBytes(final byte[] value) throws ConversionException, MessageEOFException {
try {
Pair<Integer, Integer> pairRead = streamReadBytes(getReadBodyBuffer(), len, value);
@ -171,108 +186,94 @@ public final class ServerJMSStreamMessage extends ServerJMSMessage implements St
len = pairRead.getA();
return pairRead.getB();
} catch (IllegalStateException e) {
throw new MessageFormatException(e.getMessage());
throw new ConversionException(e.getMessage(), e);
} catch (IndexOutOfBoundsException e) {
throw new MessageEOFException("");
throw new MessageEOFException();
}
}
@Override
public Object readObject() throws JMSException {
public Object readObject() throws ConversionException, MessageEOFException {
if (getReadBodyBuffer().readerIndex() >= getReadBodyBuffer().writerIndex()) {
throw new MessageEOFException("");
throw new MessageEOFException();
}
try {
return streamReadObject(getReadBodyBuffer());
} catch (IllegalStateException e) {
throw new MessageFormatException(e.getMessage());
throw new ConversionException(e.getMessage(), e);
} catch (IndexOutOfBoundsException e) {
throw new MessageEOFException("");
throw new MessageEOFException();
}
}
@Override
public void writeBoolean(final boolean value) throws JMSException {
public void writeBoolean(final boolean value) {
getWriteBodyBuffer().writeByte(DataConstants.BOOLEAN);
getWriteBodyBuffer().writeBoolean(value);
}
@Override
public void writeByte(final byte value) throws JMSException {
public void writeByte(final byte value) {
getWriteBodyBuffer().writeByte(DataConstants.BYTE);
getWriteBodyBuffer().writeByte(value);
}
@Override
public void writeShort(final short value) throws JMSException {
public void writeShort(final short value) {
getWriteBodyBuffer().writeByte(DataConstants.SHORT);
getWriteBodyBuffer().writeShort(value);
}
@Override
public void writeChar(final char value) throws JMSException {
public void writeChar(final char value) {
getWriteBodyBuffer().writeByte(DataConstants.CHAR);
getWriteBodyBuffer().writeShort((short) value);
}
@Override
public void writeInt(final int value) throws JMSException {
public void writeInt(final int value) {
getWriteBodyBuffer().writeByte(DataConstants.INT);
getWriteBodyBuffer().writeInt(value);
}
@Override
public void writeLong(final long value) throws JMSException {
public void writeLong(final long value) {
getWriteBodyBuffer().writeByte(DataConstants.LONG);
getWriteBodyBuffer().writeLong(value);
}
@Override
public void writeFloat(final float value) throws JMSException {
public void writeFloat(final float value) {
getWriteBodyBuffer().writeByte(DataConstants.FLOAT);
getWriteBodyBuffer().writeInt(Float.floatToIntBits(value));
}
@Override
public void writeDouble(final double value) throws JMSException {
public void writeDouble(final double value) {
getWriteBodyBuffer().writeByte(DataConstants.DOUBLE);
getWriteBodyBuffer().writeLong(Double.doubleToLongBits(value));
}
@Override
public void writeString(final String value) throws JMSException {
public void writeString(final String value) {
getWriteBodyBuffer().writeByte(DataConstants.STRING);
getWriteBodyBuffer().writeNullableString(value);
}
@Override
public void writeBytes(final byte[] value) throws JMSException {
public void writeBytes(final byte[] value) {
getWriteBodyBuffer().writeByte(DataConstants.BYTES);
getWriteBodyBuffer().writeInt(value.length);
getWriteBodyBuffer().writeBytes(value);
}
@Override
public void writeBytes(final byte[] value, final int offset, final int length) throws JMSException {
public void writeBytes(final byte[] value, final int offset, final int length) {
getWriteBodyBuffer().writeByte(DataConstants.BYTES);
getWriteBodyBuffer().writeInt(length);
getWriteBodyBuffer().writeBytes(value, offset, length);
}
@Override
public void writeObject(final Object value) throws JMSException {
public void writeObject(final Object value) throws ConversionException {
if (value instanceof String) {
writeString((String) value);
} else if (value instanceof Boolean) {
@ -296,34 +297,30 @@ public final class ServerJMSStreamMessage extends ServerJMSMessage implements St
} else if (value == null) {
writeString(null);
} else {
throw new MessageFormatException("Invalid object type: " + value.getClass());
throw new ConversionException("Invalid object type: " + value.getClass());
}
}
@Override
public void reset() throws JMSException {
public void reset() {
getWriteBodyBuffer().resetReaderIndex();
}
// ActiveMQRAMessage overrides ----------------------------------------
@Override
public void clearBody() throws JMSException {
public void clearBody() {
super.clearBody();
getWriteBodyBuffer().clear();
}
@Override
public void decode() throws Exception {
public void decode() {
super.decode();
}
/**
* Encode the body into the internal message
*/
@Override
public void encode() throws Exception {
public void encode() {
super.encode();
bodyLength = message.getEndOfBodyPosition();
}

66
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/jms/ServerJMSTextMessage.java → artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/coreWrapper/CoreTextMessageWrapper.java
View File

@ -14,25 +14,32 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.protocol.amqp.converter.jms;
package org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper;
import javax.jms.JMSException;
import javax.jms.TextMessage;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import org.apache.activemq.artemis.api.core.ICoreMessage;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport;
import org.apache.qpid.proton.amqp.Binary;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.messaging.AmqpValue;
import org.apache.qpid.proton.amqp.messaging.Data;
import org.apache.qpid.proton.amqp.messaging.Properties;
import org.apache.qpid.proton.amqp.messaging.Section;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.AMQP_DATA;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.AMQP_NULL;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.AMQP_UNKNOWN;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.AMQP_VALUE_STRING;
import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.EMPTY_BINARY;
import static org.apache.activemq.artemis.reader.TextMessageUtil.readBodyText;
import static org.apache.activemq.artemis.reader.TextMessageUtil.writeBodyText;
/**
* ActiveMQ Artemis implementation of a JMS TextMessage.
* <br>
* This class was ported from SpyTextMessage in JBossMQ.
*/
public class ServerJMSTextMessage extends ServerJMSMessage implements TextMessage {
// Constants -----------------------------------------------------
public class CoreTextMessageWrapper extends CoreMessageWrapper {
public static final byte TYPE = Message.TEXT_TYPE;
@ -49,14 +56,40 @@ public class ServerJMSTextMessage extends ServerJMSMessage implements TextMessag
/*
* This constructor is used to construct messages prior to sending
*/
public ServerJMSTextMessage(ICoreMessage message) {
public CoreTextMessageWrapper(ICoreMessage message) {
super(message);
}
// TextMessage implementation ------------------------------------
@Override
public void setText(final String text) throws JMSException {
public Section createAMQPSection(Map<Symbol, Object> maMap, Properties properties) {
Section body = null;
String text = getText();
switch (getOrignalEncoding()) {
case AMQP_NULL:
break;
case AMQP_DATA:
if (text == null) {
body = new Data(EMPTY_BINARY);
} else {
body = new Data(new Binary(text.getBytes(StandardCharsets.UTF_8)));
}
break;
case AMQP_VALUE_STRING:
case AMQP_UNKNOWN:
default:
body = new AmqpValue(text);
break;
}
maMap.put(AMQPMessageSupport.JMS_MSG_TYPE, AMQPMessageSupport.JMS_TEXT_MESSAGE);
return body;
}
public void setText(final String text) {
if (text != null) {
this.text = new SimpleString(text);
} else {
@ -66,7 +99,6 @@ public class ServerJMSTextMessage extends ServerJMSMessage implements TextMessag
writeBodyText(getWriteBodyBuffer(), this.text);
}
@Override
public String getText() {
if (text != null) {
return text.toString();
@ -76,20 +108,20 @@ public class ServerJMSTextMessage extends ServerJMSMessage implements TextMessag
}
@Override
public void clearBody() throws JMSException {
public void clearBody() {
super.clearBody();
text = null;
}
@Override
public void encode() throws Exception {
public void encode() {
super.encode();
writeBodyText(getWriteBodyBuffer(), text);
}
@Override
public void decode() throws Exception {
public void decode() {
super.decode();
text = readBodyText(getReadBodyBuffer());
}

24
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/coreWrapper/MessageEOFException.java Normal file
View File

@ -0,0 +1,24 @@
/*
* 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.activemq.artemis.protocol.amqp.converter.coreWrapper;
public class MessageEOFException extends Exception {
public MessageEOFException() {
}
}

285
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/jms/ServerJMSMapMessage.java
View File

@ -1,285 +0,0 @@
/*
* 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.activemq.artemis.protocol.amqp.converter.jms;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.MessageFormatException;
import java.util.Collections;
import java.util.Enumeration;
import org.apache.activemq.artemis.api.core.ActiveMQPropertyConversionException;
import org.apache.activemq.artemis.api.core.ICoreMessage;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.utils.collections.TypedProperties;
import org.apache.qpid.proton.amqp.UnsignedByte;
import org.apache.qpid.proton.amqp.UnsignedInteger;
import org.apache.qpid.proton.amqp.UnsignedLong;
import org.apache.qpid.proton.amqp.UnsignedShort;
import static org.apache.activemq.artemis.reader.MapMessageUtil.readBodyMap;
import static org.apache.activemq.artemis.reader.MapMessageUtil.writeBodyMap;
/**
* ActiveMQ Artemis implementation of a JMS MapMessage.
*/
public final class ServerJMSMapMessage extends ServerJMSMessage implements MapMessage {
// Constants -----------------------------------------------------
public static final byte TYPE = Message.MAP_TYPE;
// Attributes ----------------------------------------------------
private final TypedProperties map = new TypedProperties();
// Static --------------------------------------------------------
// Constructors --------------------------------------------------
/*
* This constructor is used to construct messages prior to sending
*/
public ServerJMSMapMessage(ICoreMessage message) {
super(message);
}
// MapMessage implementation -------------------------------------
@Override
public void setBoolean(final String name, final boolean value) throws JMSException {
map.putBooleanProperty(new SimpleString(name), value);
}
@Override
public void setByte(final String name, final byte value) throws JMSException {
map.putByteProperty(new SimpleString(name), value);
}
@Override
public void setShort(final String name, final short value) throws JMSException {
map.putShortProperty(new SimpleString(name), value);
}
@Override
public void setChar(final String name, final char value) throws JMSException {
map.putCharProperty(new SimpleString(name), value);
}
@Override
public void setInt(final String name, final int value) throws JMSException {
map.putIntProperty(new SimpleString(name), value);
}
@Override
public void setLong(final String name, final long value) throws JMSException {
map.putLongProperty(new SimpleString(name), value);
}
@Override
public void setFloat(final String name, final float value) throws JMSException {
map.putFloatProperty(new SimpleString(name), value);
}
@Override
public void setDouble(final String name, final double value) throws JMSException {
map.putDoubleProperty(new SimpleString(name), value);
}
@Override
public void setString(final String name, final String value) throws JMSException {
map.putSimpleStringProperty(new SimpleString(name), value == null ? null : new SimpleString(value));
}
@Override
public void setBytes(final String name, final byte[] value) throws JMSException {
map.putBytesProperty(new SimpleString(name), value);
}
@Override
public void setBytes(final String name, final byte[] value, final int offset, final int length) throws JMSException {
if (offset + length > value.length) {
throw new JMSException("Invalid offset/length");
}
byte[] newBytes = new byte[length];
System.arraycopy(value, offset, newBytes, 0, length);
map.putBytesProperty(new SimpleString(name), newBytes);
}
@Override
public void setObject(final String name, final Object value) throws JMSException {
try {
// primitives and String
Object val = value;
if (value instanceof UnsignedInteger) {
val = ((UnsignedInteger) value).intValue();
} else if (value instanceof UnsignedShort) {
val = ((UnsignedShort) value).shortValue();
} else if (value instanceof UnsignedByte) {
val = ((UnsignedByte) value).byteValue();
} else if (value instanceof UnsignedLong) {
val = ((UnsignedLong) value).longValue();
}
TypedProperties.setObjectProperty(new SimpleString(name), val, map);
} catch (ActiveMQPropertyConversionException e) {
throw new MessageFormatException(e.getMessage());
}
}
@Override
public boolean getBoolean(final String name) throws JMSException {
try {
return map.getBooleanProperty(new SimpleString(name));
} catch (ActiveMQPropertyConversionException e) {
throw new MessageFormatException(e.getMessage());
}
}
@Override
public byte getByte(final String name) throws JMSException {
try {
return map.getByteProperty(new SimpleString(name));
} catch (ActiveMQPropertyConversionException e) {
throw new MessageFormatException(e.getMessage());
}
}
@Override
public short getShort(final String name) throws JMSException {
try {
return map.getShortProperty(new SimpleString(name));
} catch (ActiveMQPropertyConversionException e) {
throw new MessageFormatException(e.getMessage());
}
}
@Override
public char getChar(final String name) throws JMSException {
try {
return map.getCharProperty(new SimpleString(name));
} catch (ActiveMQPropertyConversionException e) {
throw new MessageFormatException(e.getMessage());
}
}
@Override
public int getInt(final String name) throws JMSException {
try {
return map.getIntProperty(new SimpleString(name));
} catch (ActiveMQPropertyConversionException e) {
throw new MessageFormatException(e.getMessage());
}
}
@Override
public long getLong(final String name) throws JMSException {
try {
return map.getLongProperty(new SimpleString(name));
} catch (ActiveMQPropertyConversionException e) {
throw new MessageFormatException(e.getMessage());
}
}
@Override
public float getFloat(final String name) throws JMSException {
try {
return map.getFloatProperty(new SimpleString(name));
} catch (ActiveMQPropertyConversionException e) {
throw new MessageFormatException(e.getMessage());
}
}
@Override
public double getDouble(final String name) throws JMSException {
try {
return map.getDoubleProperty(new SimpleString(name));
} catch (ActiveMQPropertyConversionException e) {
throw new MessageFormatException(e.getMessage());
}
}
@Override
public String getString(final String name) throws JMSException {
try {
SimpleString str = map.getSimpleStringProperty(new SimpleString(name));
if (str == null) {
return null;
} else {
return str.toString();
}
} catch (ActiveMQPropertyConversionException e) {
throw new MessageFormatException(e.getMessage());
}
}
@Override
public byte[] getBytes(final String name) throws JMSException {
try {
return map.getBytesProperty(new SimpleString(name));
} catch (ActiveMQPropertyConversionException e) {
throw new MessageFormatException(e.getMessage());
}
}
@Override
public Object getObject(final String name) throws JMSException {
Object val = map.getProperty(new SimpleString(name));
if (val instanceof SimpleString) {
val = ((SimpleString) val).toString();
}
return val;
}
@Override
public Enumeration getMapNames() throws JMSException {
return Collections.enumeration(map.getMapNames());
}
@Override
public boolean itemExists(final String name) throws JMSException {
return map.containsProperty(new SimpleString(name));
}
@Override
public void clearBody() throws JMSException {
super.clearBody();
map.clear();
}
@Override
public void encode() throws Exception {
super.encode();
writeBodyMap(getWriteBodyBuffer(), map);
}
@Override
public void decode() throws Exception {
super.decode();
readBodyMap(getReadBodyBuffer(), map);
}
// Package protected ---------------------------------------------
// Protected -----------------------------------------------------
// Private -------------------------------------------------------
}

384
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/jms/ServerJMSMessage.java
View File

@ -1,384 +0,0 @@
/*
* 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.activemq.artemis.protocol.amqp.converter.jms;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import java.util.Collections;
import java.util.Enumeration;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ICoreMessage;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.jms.client.ActiveMQDestination;
import org.apache.activemq.artemis.reader.MessageUtil;
import static org.apache.activemq.artemis.api.core.FilterConstants.NATIVE_MESSAGE_ID;
import static org.apache.activemq.artemis.api.core.Message.BYTES_TYPE;
import static org.apache.activemq.artemis.api.core.Message.MAP_TYPE;
import static org.apache.activemq.artemis.api.core.Message.OBJECT_TYPE;
import static org.apache.activemq.artemis.api.core.Message.STREAM_TYPE;
import static org.apache.activemq.artemis.api.core.Message.TEXT_TYPE;
public class ServerJMSMessage implements Message {
protected final ICoreMessage message;
private ActiveMQBuffer readBodyBuffer;
public ServerJMSMessage(ICoreMessage message) {
this.message = message;
}
public static ServerJMSMessage wrapCoreMessage(ICoreMessage wrapped) {
switch (wrapped.getType()) {
case STREAM_TYPE:
return new ServerJMSStreamMessage(wrapped);
case BYTES_TYPE:
return new ServerJMSBytesMessage(wrapped);
case MAP_TYPE:
return new ServerJMSMapMessage(wrapped);
case TEXT_TYPE:
return new ServerJMSTextMessage(wrapped);
case OBJECT_TYPE:
return new ServerJMSObjectMessage(wrapped);
default:
return new ServerJMSMessage(wrapped);
}
}
public ICoreMessage getInnerMessage() {
return message;
}
/**
* When reading we use a protected copy so multi-threads can work fine
*/
protected ActiveMQBuffer getReadBodyBuffer() {
if (readBodyBuffer == null) {
// to avoid clashes between multiple threads
readBodyBuffer = message.getDataBuffer();
}
return readBodyBuffer;
}
/**
* When writing on the conversion we use the buffer directly
*/
protected ActiveMQBuffer getWriteBodyBuffer() {
readBodyBuffer = null; // it invalidates this buffer if anything is written
return message.getBodyBuffer();
}
@Override
public final String getJMSMessageID() throws JMSException {
if (message.containsProperty(NATIVE_MESSAGE_ID)) {
return getStringProperty(NATIVE_MESSAGE_ID);
}
return null;
}
@Override
public final void setJMSMessageID(String id) throws JMSException {
if (id != null) {
message.putStringProperty(NATIVE_MESSAGE_ID, id);
}
}
@Override
public final long getJMSTimestamp() throws JMSException {
return message.getTimestamp();
}
@Override
public final void setJMSTimestamp(long timestamp) throws JMSException {
message.setTimestamp(timestamp);
}
@Override
public final byte[] getJMSCorrelationIDAsBytes() throws JMSException {
return MessageUtil.getJMSCorrelationIDAsBytes(message);
}
@Override
public final void setJMSCorrelationIDAsBytes(byte[] correlationID) throws JMSException {
if (correlationID == null || correlationID.length == 0) {
throw new JMSException("Please specify a non-zero length byte[]");
}
message.setCorrelationID(correlationID);
}
@Override
public final String getJMSCorrelationID() throws JMSException {
Object correlationID = message.getCorrelationID();
if (correlationID instanceof String) {
return ((String) correlationID);
} else if (correlationID != null) {
return String.valueOf(correlationID);
} else {
return null;
}
}
@Override
public final void setJMSCorrelationID(String correlationID) throws JMSException {
message.setCorrelationID(correlationID);
}
@Override
public final Destination getJMSReplyTo() throws JMSException {
SimpleString reply = MessageUtil.getJMSReplyTo(message);
if (reply != null) {
return ActiveMQDestination.fromPrefixedName(reply.toString());
} else {
return null;
}
}
@Override
public final void setJMSReplyTo(Destination replyTo) throws JMSException {
MessageUtil.setJMSReplyTo(message, replyTo == null ? null : ((ActiveMQDestination) replyTo).getSimpleAddress());
}
@Override
public Destination getJMSDestination() throws JMSException {
return ActiveMQDestination.createDestination(message.getRoutingType(), message.getAddressSimpleString());
}
@Override
public final void setJMSDestination(Destination destination) throws JMSException {
if (destination == null) {
message.setAddress((SimpleString) null);
} else {
message.setAddress(((ActiveMQDestination) destination).getSimpleAddress());
}
}
@Override
public final int getJMSDeliveryMode() throws JMSException {
return message.isDurable() ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT;
}
@Override
public final void setJMSDeliveryMode(int deliveryMode) throws JMSException {
if (deliveryMode == DeliveryMode.PERSISTENT) {
message.setDurable(true);
} else if (deliveryMode == DeliveryMode.NON_PERSISTENT) {
message.setDurable(false);
} else {
throw new JMSException("Invalid mode " + deliveryMode);
}
}
@Override
public final boolean getJMSRedelivered() throws JMSException {
return false;
}
@Override
public final void setJMSRedelivered(boolean redelivered) throws JMSException {
// no op
}
@Override
public final String getJMSType() throws JMSException {
return MessageUtil.getJMSType(message);
}
@Override
public final void setJMSType(String type) throws JMSException {
MessageUtil.setJMSType(message, type);
}
@Override
public final long getJMSExpiration() throws JMSException {
return message.getExpiration();
}
@Override
public final void setJMSExpiration(long expiration) throws JMSException {
message.setExpiration(expiration);
}
@Override
public final long getJMSDeliveryTime() throws JMSException {
// no op
return 0;
}
@Override
public final void setJMSDeliveryTime(long deliveryTime) throws JMSException {
// no op
}
@Override
public final int getJMSPriority() throws JMSException {
return message.getPriority();
}
@Override
public final void setJMSPriority(int priority) throws JMSException {
message.setPriority((byte) priority);
}
@Override
public final void clearProperties() throws JMSException {
MessageUtil.clearProperties(message);
}
@Override
public final boolean propertyExists(String name) throws JMSException {
return MessageUtil.propertyExists(message, name);
}
@Override
public final boolean getBooleanProperty(String name) throws JMSException {
return message.getBooleanProperty(name);
}
@Override
public final byte getByteProperty(String name) throws JMSException {
return message.getByteProperty(name);
}
@Override
public final short getShortProperty(String name) throws JMSException {
return message.getShortProperty(name);
}
@Override
public final int getIntProperty(String name) throws JMSException {
return MessageUtil.getIntProperty(message, name);
}
@Override
public final long getLongProperty(String name) throws JMSException {
return MessageUtil.getLongProperty(message, name);
}
@Override
public final float getFloatProperty(String name) throws JMSException {
return message.getFloatProperty(name);
}
@Override
public final double getDoubleProperty(String name) throws JMSException {
return message.getDoubleProperty(name);
}
@Override
public final String getStringProperty(String name) throws JMSException {
return MessageUtil.getStringProperty(message, name);
}
@Override
public final Object getObjectProperty(String name) throws JMSException {
return MessageUtil.getObjectProperty(message, name);
}
@Override
public final Enumeration getPropertyNames() throws JMSException {
return Collections.enumeration(MessageUtil.getPropertyNames(message));
}
@Override
public final void setBooleanProperty(String name, boolean value) throws JMSException {
message.putBooleanProperty(name, value);
}
@Override
public final void setByteProperty(String name, byte value) throws JMSException {
message.putByteProperty(name, value);
}
@Override
public final void setShortProperty(String name, short value) throws JMSException {
message.putShortProperty(name, value);
}
@Override
public final void setIntProperty(String name, int value) throws JMSException {
MessageUtil.setIntProperty(message, name, value);
}
@Override
public final void setLongProperty(String name, long value) throws JMSException {
MessageUtil.setLongProperty(message, name, value);
}
@Override
public final void setFloatProperty(String name, float value) throws JMSException {
message.putFloatProperty(name, value);
}
@Override
public final void setDoubleProperty(String name, double value) throws JMSException {
message.putDoubleProperty(name, value);
}
@Override
public final void setStringProperty(String name, String value) throws JMSException {
MessageUtil.setStringProperty(message, name, value);
}
@Override
public final void setObjectProperty(String name, Object value) throws JMSException {
MessageUtil.setObjectProperty(message, name, value);
}
@Override
public final void acknowledge() throws JMSException {
// no op
}
@Override
public void clearBody() throws JMSException {
message.getBodyBuffer().clear();
}
@Override
public final <T> T getBody(Class<T> c) throws JMSException {
// no op.. jms2 not used on the conversion
return null;
}
/**
* Encode the body into the internal message
*/
public void encode() throws Exception {
if (!message.isLargeMessage()) {
message.getBodyBuffer().resetReaderIndex();
}
}
public void decode() throws Exception {
if (!message.isLargeMessage()) {
message.getBodyBuffer().resetReaderIndex();
}
}
@Override
public final boolean isBodyAssignableTo(Class c) throws JMSException {
// no op.. jms2 not used on the conversion
return false;
}
}

73
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/jms/ServerJMSObjectMessage.java
View File

@ -1,73 +0,0 @@
/**
* 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
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* 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.activemq.artemis.protocol.amqp.converter.jms;
import java.io.Serializable;
import javax.jms.JMSException;
import javax.jms.ObjectMessage;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ICoreMessage;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.qpid.proton.amqp.Binary;
public class ServerJMSObjectMessage extends ServerJMSMessage implements ObjectMessage {
public static final byte TYPE = Message.OBJECT_TYPE;
private Binary payload;
public ServerJMSObjectMessage(ICoreMessage message) {
super(message);
}
@Override
public void setObject(Serializable object) throws JMSException {
throw new UnsupportedOperationException("Cannot set Object on this internal message");
}
@Override
public Serializable getObject() throws JMSException {
throw new UnsupportedOperationException("Cannot set Object on this internal message");
}
public void setSerializedForm(Binary payload) {
this.payload = payload;
}
public Binary getSerializedForm() {
return payload;
}
@Override
public void encode() throws Exception {
super.encode();
getInnerMessage().getBodyBuffer().writeInt(payload.getLength());
getInnerMessage().getBodyBuffer().writeBytes(payload.getArray(), payload.getArrayOffset(), payload.getLength());
}
@Override
public void decode() throws Exception {
super.decode();
ActiveMQBuffer buffer = getInnerMessage().getDataBuffer();
int size = buffer.readInt();
byte[] bytes = new byte[size];
buffer.readBytes(bytes);
payload = new Binary(bytes);
}
}

4
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerSenderContext.java
View File

@ -42,7 +42,6 @@ import org.apache.activemq.artemis.core.server.MessageReference;
import org.apache.activemq.artemis.core.server.QueueQueryResult;
import org.apache.activemq.artemis.core.server.ServerConsumer;
import org.apache.activemq.artemis.core.server.impl.ServerConsumerImpl;
import org.apache.activemq.artemis.jms.client.ActiveMQDestination;
import org.apache.activemq.artemis.protocol.amqp.broker.AMQPLargeMessage;
import org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessage;
import org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessageBrokerAccessor;
@ -64,6 +63,7 @@ import org.apache.activemq.artemis.selector.filter.FilterException;
import org.apache.activemq.artemis.selector.impl.SelectorParser;
import org.apache.activemq.artemis.spi.core.remoting.ReadyListener;
import org.apache.activemq.artemis.utils.CompositeAddress;
import org.apache.activemq.artemis.utils.DestinationUtil;
import org.apache.qpid.proton.amqp.DescribedType;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.messaging.Accepted;
@ -842,7 +842,7 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
final boolean durable = !isVolatile;
final String subscriptionName = pubId.contains("|") ? pubId.split("\\|")[0] : pubId;
final String clientID = clientId == null || clientId.isEmpty() || global ? null : clientId;
return ActiveMQDestination.createQueueNameForSubscription(durable, clientID, subscriptionName);
return DestinationUtil.createQueueNameForSubscription(durable, clientID, subscriptionName);
} else {
String queue = clientId == null || clientId.isEmpty() || global ? pubId : clientId + "." + pubId;
if (shared) {

34
artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/converter/TestConversions.java
View File

@ -31,11 +31,11 @@ import org.apache.activemq.artemis.core.buffers.impl.ResetLimitWrappedActiveMQBu
import org.apache.activemq.artemis.core.message.impl.CoreMessage;
import org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessage;
import org.apache.activemq.artemis.protocol.amqp.broker.AMQPStandardMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSBytesMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSMapMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSStreamMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSTextMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreBytesMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreMapMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreStreamMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreTextMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.util.NettyReadable;
import org.apache.activemq.artemis.protocol.amqp.util.NettyWritable;
import org.apache.activemq.artemis.protocol.amqp.util.TLSEncode;
@ -85,7 +85,7 @@ public class TestConversions extends Assert {
ICoreMessage serverMessage = encodedMessage.toCore();
verifyProperties(ServerJMSMessage.wrapCoreMessage(serverMessage));
verifyProperties(CoreMessageWrapper.wrap(serverMessage));
}
@Test
@ -107,7 +107,7 @@ public class TestConversions extends Assert {
ICoreMessage serverMessage = encodedMessage.toCore();
ServerJMSBytesMessage bytesMessage = (ServerJMSBytesMessage) ServerJMSMessage.wrapCoreMessage(serverMessage);
CoreBytesMessageWrapper bytesMessage = (CoreBytesMessageWrapper) CoreMessageWrapper.wrap(serverMessage);
verifyProperties(bytesMessage);
@ -120,7 +120,7 @@ public class TestConversions extends Assert {
Assert.assertArrayEquals(bodyBytes, newBodyBytes);
}
private void verifyProperties(javax.jms.Message message) throws Exception {
private void verifyProperties(CoreMessageWrapper message) throws Exception {
assertEquals(true, message.getBooleanProperty("true"));
assertEquals(false, message.getBooleanProperty("false"));
assertEquals("bar", message.getStringProperty("foo"));
@ -161,7 +161,7 @@ public class TestConversions extends Assert {
ICoreMessage serverMessage = encodedMessage.toCore();
serverMessage.getReadOnlyBodyBuffer();
ServerJMSMapMessage mapMessage = (ServerJMSMapMessage) ServerJMSMessage.wrapCoreMessage(serverMessage);
CoreMapMessageWrapper mapMessage = (CoreMapMessageWrapper) CoreMessageWrapper.wrap(serverMessage);
mapMessage.decode();
verifyProperties(mapMessage);
@ -190,7 +190,7 @@ public class TestConversions extends Assert {
ICoreMessage serverMessage = encodedMessage.toCore();
ServerJMSStreamMessage streamMessage = (ServerJMSStreamMessage) ServerJMSMessage.wrapCoreMessage(serverMessage);
CoreStreamMessageWrapper streamMessage = (CoreStreamMessageWrapper) CoreMessageWrapper.wrap(serverMessage);
verifyProperties(streamMessage);
@ -214,7 +214,7 @@ public class TestConversions extends Assert {
ICoreMessage serverMessage = encodedMessage.toCore();
ServerJMSTextMessage textMessage = (ServerJMSTextMessage) ServerJMSMessage.wrapCoreMessage(serverMessage);
CoreTextMessageWrapper textMessage = (CoreTextMessageWrapper) CoreMessageWrapper.wrap(serverMessage);
textMessage.decode();
verifyProperties(textMessage);
@ -236,7 +236,7 @@ public class TestConversions extends Assert {
ICoreMessage serverMessage = encodedMessage.toCore();
ServerJMSTextMessage textMessage = (ServerJMSTextMessage) ServerJMSMessage.wrapCoreMessage(serverMessage);
CoreTextMessageWrapper textMessage = (CoreTextMessageWrapper) CoreMessageWrapper.wrap(serverMessage);
textMessage.decode();
verifyProperties(textMessage);
@ -277,7 +277,7 @@ public class TestConversions extends Assert {
ICoreMessage serverMessage = encodedMessage.toCore();
serverMessage.getReadOnlyBodyBuffer();
ServerJMSMapMessage mapMessage = (ServerJMSMapMessage) ServerJMSMessage.wrapCoreMessage(serverMessage);
CoreMapMessageWrapper mapMessage = (CoreMapMessageWrapper) CoreMessageWrapper.wrap(serverMessage);
mapMessage.decode();
verifyProperties(mapMessage);
@ -329,7 +329,7 @@ public class TestConversions extends Assert {
ICoreMessage serverMessage = encodedMessage.toCore();
serverMessage.getReadOnlyBodyBuffer();
ServerJMSMapMessage mapMessage = (ServerJMSMapMessage) ServerJMSMessage.wrapCoreMessage(serverMessage);
CoreMapMessageWrapper mapMessage = (CoreMapMessageWrapper) CoreMessageWrapper.wrap(serverMessage);
mapMessage.decode();
verifyProperties(mapMessage);
@ -363,7 +363,7 @@ public class TestConversions extends Assert {
byte[] encodedEmbeddedMap = encodeObject(embeddedMap);
ServerJMSMessage serverMessage = createMessage();
CoreMessageWrapper serverMessage = createMessage();
serverMessage.setShortProperty(JMS_AMQP_ORIGINAL_ENCODING, AMQP_NULL);
serverMessage.setObjectProperty(JMS_AMQP_ENCODED_DELIVERY_ANNOTATION_PREFIX + annotationName, encodedEmbeddedMap);
@ -524,8 +524,8 @@ public class TestConversions extends Assert {
return new AMQPStandardMessage(AMQPMessage.DEFAULT_MESSAGE_FORMAT, readable, null, null);
}
private ServerJMSMessage createMessage() {
return new ServerJMSMessage(newMessage(org.apache.activemq.artemis.api.core.Message.DEFAULT_TYPE));
private CoreMessageWrapper createMessage() {
return new CoreMessageWrapper(newMessage(org.apache.activemq.artemis.api.core.Message.DEFAULT_TYPE));
}
private CoreMessage newMessage(byte messageType) {

137
artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/converter/message/JMSMappingInboundTransformerTest.java
View File

@ -29,37 +29,31 @@ import java.util.List;
import java.util.Map;
import java.util.UUID;
import javax.jms.Destination;
import javax.jms.Queue;
import javax.jms.TemporaryQueue;
import javax.jms.TemporaryTopic;
import javax.jms.TextMessage;
import javax.jms.Topic;
import org.apache.activemq.artemis.api.core.ICoreMessage;
import org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessage;
import org.apache.activemq.artemis.protocol.amqp.broker.AMQPStandardMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSBytesMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSMapMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSObjectMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSStreamMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSTextMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreBytesMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreMapMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreObjectMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreStreamMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreTextMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.util.NettyReadable;
import org.apache.activemq.artemis.protocol.amqp.util.NettyWritable;
import org.apache.qpid.proton.amqp.Binary;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.messaging.AmqpSequence;
import org.apache.qpid.proton.amqp.messaging.AmqpValue;
import org.apache.qpid.proton.amqp.messaging.Data;
import org.apache.qpid.proton.amqp.messaging.MessageAnnotations;
import org.apache.qpid.proton.message.Message;
import org.apache.qpid.proton.message.impl.MessageImpl;
import org.junit.Before;
import org.junit.Test;
import io.netty.buffer.Unpooled;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.messaging.MessageAnnotations;
public class JMSMappingInboundTransformerTest {
@ -85,10 +79,10 @@ public class JMSMappingInboundTransformerTest {
ICoreMessage coreMessage = messageEncode.toCore();
javax.jms.Message jmsMessage = ServerJMSMessage.wrapCoreMessage(coreMessage);
CoreMessageWrapper jmsMessage = CoreMessageWrapper.wrap(coreMessage);
assertNotNull("Message should not be null", jmsMessage);
assertEquals("Unexpected message class type", ServerJMSBytesMessage.class, jmsMessage.getClass());
assertEquals("Unexpected message class type", CoreBytesMessageWrapper.class, jmsMessage.getClass());
}
/**
@ -102,10 +96,10 @@ public class JMSMappingInboundTransformerTest {
public void testCreateBytesMessageFromNoBodySectionAndNoContentType() throws Exception {
MessageImpl message = (MessageImpl) Message.Factory.create();
javax.jms.Message jmsMessage = ServerJMSMessage.wrapCoreMessage(encodeAndCreateAMQPMessage(message).toCore());
CoreMessageWrapper jmsMessage = CoreMessageWrapper.wrap(encodeAndCreateAMQPMessage(message).toCore());
assertNotNull("Message should not be null", jmsMessage);
assertEquals("Unexpected message class type", ServerJMSBytesMessage.class, jmsMessage.getClass());
assertEquals("Unexpected message class type", CoreBytesMessageWrapper.class, jmsMessage.getClass());
}
@Test
@ -113,10 +107,10 @@ public class JMSMappingInboundTransformerTest {
MessageImpl message = (MessageImpl) Message.Factory.create();
message.setContentType("text/plain");
javax.jms.Message jmsMessage = ServerJMSMessage.wrapCoreMessage(encodeAndCreateAMQPMessage(message).toCore());
CoreMessageWrapper jmsMessage = CoreMessageWrapper.wrap(encodeAndCreateAMQPMessage(message).toCore());
assertNotNull("Message should not be null", jmsMessage);
assertEquals("Unexpected message class type", ServerJMSTextMessage.class, jmsMessage.getClass());
assertEquals("Unexpected message class type", CoreTextMessageWrapper.class, jmsMessage.getClass());
}
// ----- Data Body Section ------------------------------------------------//
@ -137,10 +131,10 @@ public class JMSMappingInboundTransformerTest {
message.setContentType(AMQPMessageSupport.OCTET_STREAM_CONTENT_TYPE);
AMQPStandardMessage amqp = encodeAndCreateAMQPMessage(message);
javax.jms.Message jmsMessage = ServerJMSMessage.wrapCoreMessage(amqp.toCore());
CoreMessageWrapper jmsMessage = CoreMessageWrapper.wrap(amqp.toCore());
assertNotNull("Message should not be null", jmsMessage);
assertEquals("Unexpected message class type", ServerJMSBytesMessage.class, jmsMessage.getClass());
assertEquals("Unexpected message class type", CoreBytesMessageWrapper.class, jmsMessage.getClass());
}
/**
@ -158,10 +152,10 @@ public class JMSMappingInboundTransformerTest {
message.setBody(new Data(binary));
message.setContentType("unknown-content-type");
javax.jms.Message jmsMessage = ServerJMSMessage.wrapCoreMessage(encodeAndCreateAMQPMessage(message).toCore());
CoreMessageWrapper jmsMessage = CoreMessageWrapper.wrap(encodeAndCreateAMQPMessage(message).toCore());
assertNotNull("Message should not be null", jmsMessage);
assertEquals("Unexpected message class type", ServerJMSBytesMessage.class, jmsMessage.getClass());
assertEquals("Unexpected message class type", CoreBytesMessageWrapper.class, jmsMessage.getClass());
}
/**
@ -179,20 +173,12 @@ public class JMSMappingInboundTransformerTest {
assertNull(message.getContentType());
javax.jms.Message jmsMessage = ServerJMSMessage.wrapCoreMessage(encodeAndCreateAMQPMessage(message).toCore());
CoreMessageWrapper jmsMessage = CoreMessageWrapper.wrap(encodeAndCreateAMQPMessage(message).toCore());
assertNotNull("Message should not be null", jmsMessage);
assertEquals("Unexpected message class type", ServerJMSBytesMessage.class, jmsMessage.getClass());
assertEquals("Unexpected message class type", CoreBytesMessageWrapper.class, jmsMessage.getClass());
}
/**
* Test that receiving a data body containing nothing, but with the content type set to
* {@value AMQPMessageSupport#SERIALIZED_JAVA_OBJECT_CONTENT_TYPE} results in an
* ObjectMessage when not otherwise annotated to indicate the type of JMS message it is.
*
* @throws Exception
* if an error occurs during the test.
*/
@Test
public void testCreateObjectMessageFromDataWithContentTypeAndEmptyBinary() throws Exception {
MessageImpl message = (MessageImpl) Message.Factory.create();
@ -200,10 +186,10 @@ public class JMSMappingInboundTransformerTest {
message.setBody(new Data(binary));
message.setContentType(AMQPMessageSupport.SERIALIZED_JAVA_OBJECT_CONTENT_TYPE.toString());
javax.jms.Message jmsMessage = ServerJMSMessage.wrapCoreMessage(encodeAndCreateAMQPMessage(message).toCore());
CoreMessageWrapper jmsMessage = CoreMessageWrapper.wrap(encodeAndCreateAMQPMessage(message).toCore());
assertNotNull("Message should not be null", jmsMessage);
assertEquals("Unexpected message class type", ServerJMSObjectMessage.class, jmsMessage.getClass());
assertEquals("Unexpected message class type", CoreObjectMessageWrapper.class, jmsMessage.getClass());
}
@Test
@ -300,13 +286,13 @@ public class JMSMappingInboundTransformerTest {
message.setBody(new Data(binary));
message.setContentType(contentType);
javax.jms.Message jmsMessage = ServerJMSMessage.wrapCoreMessage(encodeAndCreateAMQPMessage(message).toCore());
CoreMessageWrapper jmsMessage = CoreMessageWrapper.wrap(encodeAndCreateAMQPMessage(message).toCore());
assertNotNull("Message should not be null", jmsMessage);
if (StandardCharsets.UTF_8.equals(expectedCharset)) {
assertEquals("Unexpected message class type", ServerJMSTextMessage.class, jmsMessage.getClass());
assertEquals("Unexpected message class type", CoreTextMessageWrapper.class, jmsMessage.getClass());
} else {
assertEquals("Unexpected message class type", ServerJMSBytesMessage.class, jmsMessage.getClass());
assertEquals("Unexpected message class type", CoreBytesMessageWrapper.class, jmsMessage.getClass());
}
}
@ -324,10 +310,10 @@ public class JMSMappingInboundTransformerTest {
MessageImpl message = (MessageImpl) Message.Factory.create();
message.setBody(new AmqpValue("content"));
javax.jms.Message jmsMessage = ServerJMSMessage.wrapCoreMessage(encodeAndCreateAMQPMessage(message).toCore());
CoreMessageWrapper jmsMessage = CoreMessageWrapper.wrap(encodeAndCreateAMQPMessage(message).toCore());
assertNotNull("Message should not be null", jmsMessage);
assertEquals("Unexpected message class type", ServerJMSTextMessage.class, jmsMessage.getClass());
assertEquals("Unexpected message class type", CoreTextMessageWrapper.class, jmsMessage.getClass());
}
/**
@ -342,10 +328,10 @@ public class JMSMappingInboundTransformerTest {
MessageImpl message = (MessageImpl) Message.Factory.create();
message.setBody(new AmqpValue(null));
javax.jms.Message jmsMessage = ServerJMSMessage.wrapCoreMessage(encodeAndCreateAMQPMessage(message).toCore());
CoreMessageWrapper jmsMessage = CoreMessageWrapper.wrap(encodeAndCreateAMQPMessage(message).toCore());
assertNotNull("Message should not be null", jmsMessage);
assertEquals("Unexpected message class type", ServerJMSTextMessage.class, jmsMessage.getClass());
assertEquals("Unexpected message class type", CoreTextMessageWrapper.class, jmsMessage.getClass());
}
/**
@ -362,10 +348,10 @@ public class JMSMappingInboundTransformerTest {
message.setBody(new AmqpValue(new Binary(new byte[0])));
message.setContentType(AMQPMessageSupport.SERIALIZED_JAVA_OBJECT_CONTENT_TYPE.toString());
javax.jms.Message jmsMessage = ServerJMSMessage.wrapCoreMessage(encodeAndCreateAMQPMessage(message).toCore());
CoreMessageWrapper jmsMessage = CoreMessageWrapper.wrap(encodeAndCreateAMQPMessage(message).toCore());
assertNotNull("Message should not be null", jmsMessage);
assertEquals("Unexpected message class type", ServerJMSObjectMessage.class, jmsMessage.getClass());
assertEquals("Unexpected message class type", CoreObjectMessageWrapper.class, jmsMessage.getClass());
}
/**
@ -381,10 +367,10 @@ public class JMSMappingInboundTransformerTest {
Map<String, String> map = new HashMap<>();
message.setBody(new AmqpValue(map));
javax.jms.Message jmsMessage = ServerJMSMessage.wrapCoreMessage(encodeAndCreateAMQPMessage(message).toCore());
CoreMessageWrapper jmsMessage = CoreMessageWrapper.wrap(encodeAndCreateAMQPMessage(message).toCore());
assertNotNull("Message should not be null", jmsMessage);
assertEquals("Unexpected message class type", ServerJMSMapMessage.class, jmsMessage.getClass());
assertEquals("Unexpected message class type", CoreMapMessageWrapper.class, jmsMessage.getClass());
}
/**
@ -400,10 +386,10 @@ public class JMSMappingInboundTransformerTest {
List<String> list = new ArrayList<>();
message.setBody(new AmqpValue(list));
javax.jms.Message jmsMessage = ServerJMSMessage.wrapCoreMessage(encodeAndCreateAMQPMessage(message).toCore());
CoreMessageWrapper jmsMessage = CoreMessageWrapper.wrap(encodeAndCreateAMQPMessage(message).toCore());
assertNotNull("Message should not be null", jmsMessage);
assertEquals("Unexpected message class type", ServerJMSStreamMessage.class, jmsMessage.getClass());
assertEquals("Unexpected message class type", CoreStreamMessageWrapper.class, jmsMessage.getClass());
}
/**
@ -419,10 +405,10 @@ public class JMSMappingInboundTransformerTest {
List<String> list = new ArrayList<>();
message.setBody(new AmqpSequence(list));
javax.jms.Message jmsMessage = ServerJMSMessage.wrapCoreMessage(encodeAndCreateAMQPMessage(message).toCore());
CoreMessageWrapper jmsMessage = CoreMessageWrapper.wrap(encodeAndCreateAMQPMessage(message).toCore());
assertNotNull("Message should not be null", jmsMessage);
assertEquals("Unexpected message class type", ServerJMSStreamMessage.class, jmsMessage.getClass());
assertEquals("Unexpected message class type", CoreStreamMessageWrapper.class, jmsMessage.getClass());
}
/**
@ -438,10 +424,10 @@ public class JMSMappingInboundTransformerTest {
Binary binary = new Binary(new byte[0]);
message.setBody(new AmqpValue(binary));
javax.jms.Message jmsMessage = ServerJMSMessage.wrapCoreMessage(encodeAndCreateAMQPMessage(message).toCore());
CoreMessageWrapper jmsMessage = CoreMessageWrapper.wrap(encodeAndCreateAMQPMessage(message).toCore());
assertNotNull("Message should not be null", jmsMessage);
assertEquals("Unexpected message class type", ServerJMSBytesMessage.class, jmsMessage.getClass());
assertEquals("Unexpected message class type", CoreBytesMessageWrapper.class, jmsMessage.getClass());
}
/**
@ -457,10 +443,10 @@ public class JMSMappingInboundTransformerTest {
MessageImpl message = (MessageImpl) Message.Factory.create();
message.setBody(new AmqpValue(UUID.randomUUID()));
javax.jms.Message jmsMessage = ServerJMSMessage.wrapCoreMessage(encodeAndCreateAMQPMessage(message).toCore());
CoreMessageWrapper jmsMessage = CoreMessageWrapper.wrap(encodeAndCreateAMQPMessage(message).toCore());
assertNotNull("Message should not be null", jmsMessage);
assertEquals("Unexpected message class type", ServerJMSBytesMessage.class, jmsMessage.getClass());
assertEquals("Unexpected message class type", CoreBytesMessageWrapper.class, jmsMessage.getClass());
}
@Test
@ -469,13 +455,13 @@ public class JMSMappingInboundTransformerTest {
MessageImpl message = (MessageImpl) Message.Factory.create();
message.setBody(new AmqpValue(contentString));
ServerJMSTextMessage jmsMessage = (ServerJMSTextMessage)ServerJMSMessage.wrapCoreMessage(encodeAndCreateAMQPMessage(message).toCore());
CoreTextMessageWrapper jmsMessage = (CoreTextMessageWrapper) CoreMessageWrapper.wrap(encodeAndCreateAMQPMessage(message).toCore());
jmsMessage.decode();
assertTrue("Expected TextMessage", jmsMessage instanceof TextMessage);
assertEquals("Unexpected message class type", ServerJMSTextMessage.class, jmsMessage.getClass());
assertTrue("Expected TextMessage", jmsMessage instanceof CoreTextMessageWrapper);
assertEquals("Unexpected message class type", CoreTextMessageWrapper.class, jmsMessage.getClass());
TextMessage textMessage = (TextMessage) jmsMessage;
CoreTextMessageWrapper textMessage = jmsMessage;
assertNotNull(textMessage.getText());
assertEquals(contentString, textMessage.getText());
@ -485,30 +471,30 @@ public class JMSMappingInboundTransformerTest {
@Test
public void testTransformWithNoToTypeDestinationTypeAnnotation() throws Exception {
doTransformWithToTypeDestinationTypeAnnotationTestImpl(null, Destination.class);
doTransformWithToTypeDestinationTypeAnnotationTestImpl(null);
}
@Test
public void testTransformWithQueueStringToTypeDestinationTypeAnnotation() throws Exception {
doTransformWithToTypeDestinationTypeAnnotationTestImpl("queue", Queue.class);
doTransformWithToTypeDestinationTypeAnnotationTestImpl("queue");
}
@Test
public void testTransformWithTemporaryQueueStringToTypeDestinationTypeAnnotation() throws Exception {
doTransformWithToTypeDestinationTypeAnnotationTestImpl("queue,temporary", TemporaryQueue.class);
doTransformWithToTypeDestinationTypeAnnotationTestImpl("queue,temporary");
}
@Test
public void testTransformWithTopicStringToTypeDestinationTypeAnnotation() throws Exception {
doTransformWithToTypeDestinationTypeAnnotationTestImpl("topic", Topic.class);
doTransformWithToTypeDestinationTypeAnnotationTestImpl("topic");
}
@Test
public void testTransformWithTemporaryTopicStringToTypeDestinationTypeAnnotation() throws Exception {
doTransformWithToTypeDestinationTypeAnnotationTestImpl("topic,temporary", TemporaryTopic.class);
doTransformWithToTypeDestinationTypeAnnotationTestImpl("topic,temporary");
}
private void doTransformWithToTypeDestinationTypeAnnotationTestImpl(Object toTypeAnnotationValue, Class<? extends Destination> expectedClass)
private void doTransformWithToTypeDestinationTypeAnnotationTestImpl(Object toTypeAnnotationValue)
throws Exception {
String toAddress = "toAddress";
@ -522,38 +508,39 @@ public class JMSMappingInboundTransformerTest {
message.setMessageAnnotations(ma);
}
javax.jms.Message jmsMessage = ServerJMSMessage.wrapCoreMessage(encodeAndCreateAMQPMessage(message).toCore());
assertTrue("Expected TextMessage", jmsMessage instanceof TextMessage);
CoreMessageWrapper jmsMessage = CoreMessageWrapper.wrap(encodeAndCreateAMQPMessage(message).toCore());
assertTrue("Expected ServerJMSTextMessage", jmsMessage instanceof CoreTextMessageWrapper);
}
// ----- ReplyTo Conversions ----------------------------------------------//
@Test
public void testTransformWithNoReplyToTypeDestinationTypeAnnotation() throws Exception {
doTransformWithReplyToTypeDestinationTypeAnnotationTestImpl(null, Destination.class);
doTransformWithReplyToTypeDestinationTypeAnnotationTestImpl(null);
}
@Test
public void testTransformWithQueueStringReplyToTypeDestinationTypeAnnotation() throws Exception {
doTransformWithReplyToTypeDestinationTypeAnnotationTestImpl("queue", Queue.class);
doTransformWithReplyToTypeDestinationTypeAnnotationTestImpl("queue");
}
@Test
public void testTransformWithTemporaryQueueStringReplyToTypeDestinationTypeAnnotation() throws Exception {
doTransformWithReplyToTypeDestinationTypeAnnotationTestImpl("queue,temporary", TemporaryQueue.class);
doTransformWithReplyToTypeDestinationTypeAnnotationTestImpl("queue,temporary");
}
@Test
public void testTransformWithTopicStringReplyToTypeDestinationTypeAnnotation() throws Exception {
doTransformWithReplyToTypeDestinationTypeAnnotationTestImpl("topic", Topic.class);
doTransformWithReplyToTypeDestinationTypeAnnotationTestImpl("topic");
}
@Test
public void testTransformWithTemporaryTopicStringReplyToTypeDestinationTypeAnnotation() throws Exception {
doTransformWithReplyToTypeDestinationTypeAnnotationTestImpl("topic,temporary", TemporaryTopic.class);
doTransformWithReplyToTypeDestinationTypeAnnotationTestImpl("topic,temporary");
}
private void doTransformWithReplyToTypeDestinationTypeAnnotationTestImpl(Object replyToTypeAnnotationValue, Class<? extends Destination> expectedClass)
private void doTransformWithReplyToTypeDestinationTypeAnnotationTestImpl(Object replyToTypeAnnotationValue)
throws Exception {
String replyToAddress = "replyToAddress";
@ -567,8 +554,8 @@ public class JMSMappingInboundTransformerTest {
message.setMessageAnnotations(ma);
}
javax.jms.Message jmsMessage = ServerJMSMessage.wrapCoreMessage(encodeAndCreateAMQPMessage(message).toCore());
assertTrue("Expected TextMessage", jmsMessage instanceof TextMessage);
CoreMessageWrapper jmsMessage = CoreMessageWrapper.wrap(encodeAndCreateAMQPMessage(message).toCore());
assertTrue("Expected TextMessage", jmsMessage instanceof CoreTextMessageWrapper);
}
private AMQPStandardMessage encodeAndCreateAMQPMessage(MessageImpl message) {

125
artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/converter/message/JMSMappingOutboundTransformerTest.java
View File

@ -40,24 +40,18 @@ import java.util.List;
import java.util.Map;
import java.util.UUID;
import javax.jms.Destination;
import javax.jms.JMSException;
import org.apache.activemq.artemis.core.buffers.impl.ResetLimitWrappedActiveMQBuffer;
import org.apache.activemq.artemis.core.message.impl.CoreMessage;
import org.apache.activemq.artemis.jms.client.ActiveMQQueue;
import org.apache.activemq.artemis.jms.client.ActiveMQTemporaryQueue;
import org.apache.activemq.artemis.jms.client.ActiveMQTemporaryTopic;
import org.apache.activemq.artemis.jms.client.ActiveMQTopic;
import org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.AMQPConverter;
import org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSBytesMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSMapMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSObjectMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSStreamMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSTextMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreBytesMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreMapMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreObjectMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreStreamMessageWrapper;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreTextMessageWrapper;
import org.apache.activemq.artemis.utils.PrefixUtil;
import org.apache.qpid.proton.amqp.Binary;
import org.apache.qpid.proton.amqp.Symbol;
@ -81,7 +75,7 @@ public class JMSMappingOutboundTransformerTest {
@Test
public void testConvertMessageToAmqpMessageWithNoBody() throws Exception {
ServerJMSMessage outbound = createMessage();
CoreMessageWrapper outbound = createMessage();
outbound.encode();
AMQPMessage amqp = AMQPConverter.getInstance().fromCore(outbound.getInnerMessage(), null);
@ -91,7 +85,7 @@ public class JMSMappingOutboundTransformerTest {
@Test
public void testConvertTextMessageToAmqpMessageWithNoBodyOriginalEncodingWasNull() throws Exception {
ServerJMSMessage outbound = createMessage();
CoreMessageWrapper outbound = createMessage();
outbound.setShortProperty(JMS_AMQP_ORIGINAL_ENCODING, AMQP_NULL);
outbound.encode();
@ -105,7 +99,7 @@ public class JMSMappingOutboundTransformerTest {
@Test
public void testConvertBytesMessageToAmqpMessageWithDataBody() throws Exception {
byte[] expectedPayload = new byte[]{8, 16, 24, 32};
ServerJMSBytesMessage outbound = createBytesMessage();
CoreBytesMessageWrapper outbound = createBytesMessage();
outbound.writeBytes(expectedPayload);
outbound.encode();
@ -124,7 +118,7 @@ public class JMSMappingOutboundTransformerTest {
@Test
public void testConvertEmptyBytesMessageToAmqpMessageWithAmqpValueBody() throws Exception {
ServerJMSBytesMessage outbound = createBytesMessage();
CoreBytesMessageWrapper outbound = createBytesMessage();
outbound.setShortProperty(JMS_AMQP_ORIGINAL_ENCODING, AMQP_VALUE_BINARY);
outbound.encode();
@ -139,7 +133,7 @@ public class JMSMappingOutboundTransformerTest {
@Test
public void testConvertBytesMessageToAmqpMessageWithAmqpValueBody() throws Exception {
byte[] expectedPayload = new byte[]{8, 16, 24, 32};
ServerJMSBytesMessage outbound = createBytesMessage();
CoreBytesMessageWrapper outbound = createBytesMessage();
outbound.setShortProperty(JMS_AMQP_ORIGINAL_ENCODING, AMQP_VALUE_BINARY);
outbound.writeBytes(expectedPayload);
outbound.encode();
@ -161,7 +155,7 @@ public class JMSMappingOutboundTransformerTest {
@Test
public void testConvertMapMessageToAmqpMessageWithNoBody() throws Exception {
ServerJMSMapMessage outbound = createMapMessage();
CoreMapMessageWrapper outbound = createMapMessage();
outbound.encode();
AMQPMessage amqp = AMQPConverter.getInstance().fromCore(outbound.getInnerMessage(), null);
@ -175,7 +169,7 @@ public class JMSMappingOutboundTransformerTest {
public void testConvertMapMessageToAmqpMessageWithByteArrayValueInBody() throws Exception {
final byte[] byteArray = new byte[]{1, 2, 3, 4, 5};
ServerJMSMapMessage outbound = createMapMessage();
CoreMapMessageWrapper outbound = createMapMessage();
outbound.setBytes("bytes", byteArray);
outbound.encode();
@ -195,7 +189,7 @@ public class JMSMappingOutboundTransformerTest {
@Test
public void testConvertMapMessageToAmqpMessage() throws Exception {
ServerJMSMapMessage outbound = createMapMessage();
CoreMapMessageWrapper outbound = createMapMessage();
outbound.setString("property-1", "string");
outbound.setInt("property-2", 1);
outbound.setBoolean("property-3", true);
@ -218,7 +212,7 @@ public class JMSMappingOutboundTransformerTest {
@Test
public void testConvertStreamMessageToAmqpMessageWithAmqpValueBodyNoPropertySet() throws Exception {
ServerJMSStreamMessage outbound = createStreamMessage();
CoreStreamMessageWrapper outbound = createStreamMessage();
outbound.writeBoolean(false);
outbound.writeString("test");
outbound.encode();
@ -238,7 +232,7 @@ public class JMSMappingOutboundTransformerTest {
@Test
public void testConvertStreamMessageToAmqpMessageWithAmqpValueBody() throws Exception {
ServerJMSStreamMessage outbound = createStreamMessage();
CoreStreamMessageWrapper outbound = createStreamMessage();
outbound.setShortProperty(JMS_AMQP_ORIGINAL_ENCODING, AMQP_VALUE_LIST);
outbound.writeBoolean(false);
outbound.writeString("test");
@ -259,7 +253,7 @@ public class JMSMappingOutboundTransformerTest {
@Test
public void testConvertStreamMessageToAmqpMessageWithAmqpSequencey() throws Exception {
ServerJMSStreamMessage outbound = createStreamMessage();
CoreStreamMessageWrapper outbound = createStreamMessage();
outbound.setShortProperty(JMS_AMQP_ORIGINAL_ENCODING, AMQP_SEQUENCE);
outbound.writeBoolean(false);
outbound.writeString("test");
@ -281,7 +275,7 @@ public class JMSMappingOutboundTransformerTest {
@Test
public void testConvertEmptyObjectMessageToAmqpMessageWithDataBody() throws Exception {
ServerJMSObjectMessage outbound = createObjectMessage();
CoreObjectMessageWrapper outbound = createObjectMessage();
outbound.encode();
AMQPMessage amqp = AMQPConverter.getInstance().fromCore(outbound.getInnerMessage(), null);
@ -293,7 +287,7 @@ public class JMSMappingOutboundTransformerTest {
@Test
public void testConvertEmptyObjectMessageToAmqpMessageUnknownEncodingGetsDataSection() throws Exception {
ServerJMSObjectMessage outbound = createObjectMessage();
CoreObjectMessageWrapper outbound = createObjectMessage();
outbound.setShortProperty(JMS_AMQP_ORIGINAL_ENCODING, AMQP_UNKNOWN);
outbound.encode();
@ -306,7 +300,7 @@ public class JMSMappingOutboundTransformerTest {
@Test
public void testConvertObjectMessageToAmqpMessageWithDataBody() throws Exception {
ServerJMSObjectMessage outbound = createObjectMessage(TEST_OBJECT_VALUE);
CoreObjectMessageWrapper outbound = createObjectMessage(TEST_OBJECT_VALUE);
outbound.encode();
AMQPMessage amqp = AMQPConverter.getInstance().fromCore(outbound.getInnerMessage(), null);
@ -322,7 +316,7 @@ public class JMSMappingOutboundTransformerTest {
@Test
public void testConvertObjectMessageToAmqpMessageUnknownEncodingGetsDataSection() throws Exception {
ServerJMSObjectMessage outbound = createObjectMessage(TEST_OBJECT_VALUE);
CoreObjectMessageWrapper outbound = createObjectMessage(TEST_OBJECT_VALUE);
outbound.setShortProperty(JMS_AMQP_ORIGINAL_ENCODING, AMQP_UNKNOWN);
outbound.encode();
@ -339,7 +333,7 @@ public class JMSMappingOutboundTransformerTest {
@Test
public void testConvertObjectMessageToAmqpMessageWithAmqpValueBody() throws Exception {
ServerJMSObjectMessage outbound = createObjectMessage(TEST_OBJECT_VALUE);
CoreObjectMessageWrapper outbound = createObjectMessage(TEST_OBJECT_VALUE);
outbound.setShortProperty(JMS_AMQP_ORIGINAL_ENCODING, AMQP_VALUE_BINARY);
outbound.encode();
@ -357,7 +351,7 @@ public class JMSMappingOutboundTransformerTest {
@Test
public void testConvertEmptyObjectMessageToAmqpMessageWithAmqpValueBody() throws Exception {
ServerJMSObjectMessage outbound = createObjectMessage();
CoreObjectMessageWrapper outbound = createObjectMessage();
outbound.setShortProperty(JMS_AMQP_ORIGINAL_ENCODING, AMQP_VALUE_BINARY);
outbound.encode();
@ -373,7 +367,7 @@ public class JMSMappingOutboundTransformerTest {
@Test
public void testConvertTextMessageToAmqpMessageWithNoBody() throws Exception {
ServerJMSTextMessage outbound = createTextMessage();
CoreTextMessageWrapper outbound = createTextMessage();
outbound.encode();
AMQPMessage amqp = AMQPConverter.getInstance().fromCore(outbound.getInnerMessage(), null);
@ -386,7 +380,7 @@ public class JMSMappingOutboundTransformerTest {
@Test
public void testConvertTextMessageCreatesAmqpValueStringBody() throws Exception {
String contentString = "myTextMessageContent";
ServerJMSTextMessage outbound = createTextMessage(contentString);
CoreTextMessageWrapper outbound = createTextMessage(contentString);
outbound.encode();
AMQPMessage amqp = AMQPConverter.getInstance().fromCore(outbound.getInnerMessage(), null);
@ -399,7 +393,7 @@ public class JMSMappingOutboundTransformerTest {
@Test
public void testConvertTextMessageContentNotStoredCreatesAmqpValueStringBody() throws Exception {
String contentString = "myTextMessageContent";
ServerJMSTextMessage outbound = createTextMessage(contentString);
CoreTextMessageWrapper outbound = createTextMessage(contentString);
outbound.encode();
AMQPMessage amqp = AMQPConverter.getInstance().fromCore(outbound.getInnerMessage(), null);
@ -412,7 +406,7 @@ public class JMSMappingOutboundTransformerTest {
@Test
public void testConvertTextMessageCreatesDataSectionBody() throws Exception {
String contentString = "myTextMessageContent";
ServerJMSTextMessage outbound = createTextMessage(contentString);
CoreTextMessageWrapper outbound = createTextMessage(contentString);
outbound.encode();
AMQPMessage amqp = AMQPConverter.getInstance().fromCore(outbound.getInnerMessage(), null);
@ -428,7 +422,7 @@ public class JMSMappingOutboundTransformerTest {
@Test
public void testConvertTextMessageCreatesBodyUsingOriginalEncodingWithDataSection() throws Exception {
String contentString = "myTextMessageContent";
ServerJMSTextMessage outbound = createTextMessage(contentString);
CoreTextMessageWrapper outbound = createTextMessage(contentString);
outbound.setShortProperty(JMS_AMQP_ORIGINAL_ENCODING, AMQP_DATA);
outbound.encode();
@ -446,7 +440,7 @@ public class JMSMappingOutboundTransformerTest {
@Test
public void testConvertTextMessageContentNotStoredCreatesBodyUsingOriginalEncodingWithDataSection() throws Exception {
String contentString = "myTextMessageContent";
ServerJMSTextMessage outbound = createTextMessage(contentString);
CoreTextMessageWrapper outbound = createTextMessage(contentString);
outbound.setShortProperty(JMS_AMQP_ORIGINAL_ENCODING, AMQP_DATA);
outbound.encode();
@ -473,10 +467,11 @@ public class JMSMappingOutboundTransformerTest {
doTestConvertMessageWithJMSDestination(createDestination(QUEUE_TYPE), QUEUE_TYPE);
}
private void doTestConvertMessageWithJMSDestination(Destination jmsDestination, Object expectedAnnotationValue) throws Exception {
ServerJMSTextMessage textMessage = createTextMessage();
private void doTestConvertMessageWithJMSDestination(String jmsDestination, Object expectedAnnotationValue) throws Exception {
CoreTextMessageWrapper textMessage = createTextMessage();
textMessage.setText("myTextMessageContent");
textMessage.setJMSDestination(jmsDestination);
textMessage.setDestination(jmsDestination);
AMQPMessage amqp = AMQPConverter.getInstance().fromCore(textMessage.getInnerMessage(), null);
@ -490,7 +485,7 @@ public class JMSMappingOutboundTransformerTest {
}
if (jmsDestination != null) {
assertEquals("Unexpected 'to' address", AMQPMessageSupport.toAddress(jmsDestination), amqp.getAddress());
assertEquals("Unexpected 'to' address", jmsDestination, amqp.getAddress());
}
}
@ -506,8 +501,8 @@ public class JMSMappingOutboundTransformerTest {
doTestConvertMessageWithJMSReplyTo(createDestination(QUEUE_TYPE), QUEUE_TYPE);
}
private void doTestConvertMessageWithJMSReplyTo(Destination jmsReplyTo, Object expectedAnnotationValue) throws Exception {
ServerJMSTextMessage textMessage = createTextMessage();
private void doTestConvertMessageWithJMSReplyTo(String jmsReplyTo, Object expectedAnnotationValue) throws Exception {
CoreTextMessageWrapper textMessage = createTextMessage();
textMessage.setText("myTextMessageContent");
textMessage.setJMSReplyTo(jmsReplyTo);
@ -523,58 +518,48 @@ public class JMSMappingOutboundTransformerTest {
}
if (jmsReplyTo != null) {
assertEquals("Unexpected 'reply-to' address", AMQPMessageSupport.toAddress(jmsReplyTo).toString(), amqp.getReplyTo().toString());
assertEquals("Unexpected 'reply-to' address", jmsReplyTo, amqp.getReplyTo().toString());
}
}
// ----- Utility Methods used for this Test -------------------------------//
private Destination createDestination(byte destType) {
Destination destination = null;
private String createDestination(byte destType) {
String prefix = PrefixUtil.getURIPrefix(TEST_ADDRESS);
String address = PrefixUtil.removePrefix(TEST_ADDRESS, prefix);
switch (destType) {
case QUEUE_TYPE:
destination = new ActiveMQQueue(address);
break;
case TOPIC_TYPE:
destination = new ActiveMQTopic(address);
break;
case TEMP_QUEUE_TYPE:
destination = new ActiveMQTemporaryQueue(address, null);
break;
case TEMP_TOPIC_TYPE:
destination = new ActiveMQTemporaryTopic(address, null);
break;
return address;
default:
throw new IllegalArgumentException("Invliad Destination Type given/");
}
return destination;
}
private ServerJMSMessage createMessage() {
return new ServerJMSMessage(newMessage(org.apache.activemq.artemis.api.core.Message.DEFAULT_TYPE));
private CoreMessageWrapper createMessage() {
return new CoreMessageWrapper(newMessage(org.apache.activemq.artemis.api.core.Message.DEFAULT_TYPE));
}
private ServerJMSBytesMessage createBytesMessage() {
return new ServerJMSBytesMessage(newMessage(org.apache.activemq.artemis.api.core.Message.BYTES_TYPE));
private CoreBytesMessageWrapper createBytesMessage() {
return new CoreBytesMessageWrapper(newMessage(org.apache.activemq.artemis.api.core.Message.BYTES_TYPE));
}
private ServerJMSMapMessage createMapMessage() {
return new ServerJMSMapMessage(newMessage(org.apache.activemq.artemis.api.core.Message.MAP_TYPE));
private CoreMapMessageWrapper createMapMessage() {
return new CoreMapMessageWrapper(newMessage(org.apache.activemq.artemis.api.core.Message.MAP_TYPE));
}
private ServerJMSStreamMessage createStreamMessage() {
return new ServerJMSStreamMessage(newMessage(org.apache.activemq.artemis.api.core.Message.STREAM_TYPE));
private CoreStreamMessageWrapper createStreamMessage() {
return new CoreStreamMessageWrapper(newMessage(org.apache.activemq.artemis.api.core.Message.STREAM_TYPE));
}
private ServerJMSObjectMessage createObjectMessage() {
private CoreObjectMessageWrapper createObjectMessage() {
return createObjectMessage(null);
}
private ServerJMSObjectMessage createObjectMessage(Serializable payload) {
ServerJMSObjectMessage result = AMQPMessageSupport.createObjectMessage(0, null);
private CoreObjectMessageWrapper createObjectMessage(Serializable payload) {
CoreObjectMessageWrapper result = AMQPMessageSupport.createObjectMessage(0, null);
try (ByteArrayOutputStream baos = new ByteArrayOutputStream(); ObjectOutputStream oos = new ObjectOutputStream(baos);) {
@ -588,16 +573,16 @@ public class JMSMappingOutboundTransformerTest {
return result;
}
private ServerJMSTextMessage createTextMessage() {
private CoreTextMessageWrapper createTextMessage() {
return createTextMessage(null);
}
private ServerJMSTextMessage createTextMessage(String text) {
ServerJMSTextMessage result = AMQPMessageSupport.createTextMessage(0, null);
private CoreTextMessageWrapper createTextMessage(String text) {
CoreTextMessageWrapper result = AMQPMessageSupport.createTextMessage(0, null);
try {
result.setText(text);
} catch (JMSException e) {
} catch (Exception e) {
}
return result;

4
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpManagementTest.java
View File

@ -22,7 +22,7 @@ import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.apache.activemq.artemis.api.core.management.ResourceNames;
import org.apache.activemq.artemis.protocol.amqp.converter.jms.ServerJMSMapMessage;
import org.apache.activemq.artemis.protocol.amqp.converter.coreWrapper.CoreMapMessageWrapper;
import org.apache.activemq.transport.amqp.client.AmqpClient;
import org.apache.activemq.transport.amqp.client.AmqpConnection;
import org.apache.activemq.transport.amqp.client.AmqpMessage;
@ -89,7 +89,7 @@ public class AmqpManagementTest extends AmqpClientTestSupport {
int sequence = 42;
LinkedHashMap<String, Object> map = new LinkedHashMap<>();
map.put("sequence", new UnsignedInteger(sequence));
ServerJMSMapMessage msg = createMapMessage(1, map, null);
CoreMapMessageWrapper msg = createMapMessage(1, map, null);
assertEquals(msg.getInt("sequence"), sequence);
map.clear();

2
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/crossprotocol/RequestReplyMultiProtocolTest.java
View File

@ -62,7 +62,7 @@ public class RequestReplyMultiProtocolTest extends OpenWireTestBase {
this.protocolConsumer = protocolConsumer;
}
@Parameterized.Parameters(name = "openWireOnSender={0},openWireOnConsumer={1}")
@Parameterized.Parameters(name = "senderProtocol={0},receiverProtocol={1}")
public static Iterable<Object[]> data() {
return Arrays.asList(new Object[][] {
{"OPENWIRE", "OPENWIRE"},