You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by an...@apache.org on 2014/11/11 19:41:50 UTC
[20/51] [abbrv] [partial] activemq-6 git commit: ACTIVEMQ6-4 - Rename
packages to ActiveMQ
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/23e8edd9/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQDestination.java
----------------------------------------------------------------------
diff --git a/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQDestination.java b/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQDestination.java
new file mode 100644
index 0000000..7ecc457
--- /dev/null
+++ b/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQDestination.java
@@ -0,0 +1,376 @@
+/*
+ * Copyright 2005-2014 Red Hat, Inc.
+ * Red Hat 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.activemq6.jms.client;
+
+import java.io.Serializable;
+import java.util.UUID;
+
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.JMSRuntimeException;
+import javax.naming.NamingException;
+import javax.naming.Reference;
+import javax.naming.Referenceable;
+
+import org.apache.activemq6.api.core.Pair;
+import org.apache.activemq6.api.core.SimpleString;
+import org.apache.activemq6.jms.referenceable.DestinationObjectFactory;
+import org.apache.activemq6.jms.referenceable.SerializableObjectRefAddr;
+
+/**
+ * HornetQ implementation of a JMS Destination.
+ *
+ * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a>
+ *
+ */
+public class HornetQDestination implements Destination, Serializable, Referenceable
+{
+ // Constants -----------------------------------------------------
+
+ // Static --------------------------------------------------------
+
+ /**
+ *
+ */
+ private static final long serialVersionUID = 5027962425462382883L;
+
+ public static final String JMS_QUEUE_ADDRESS_PREFIX = "jms.queue.";
+
+ public static final String JMS_TEMP_QUEUE_ADDRESS_PREFIX = "jms.tempqueue.";
+
+ public static final String JMS_TOPIC_ADDRESS_PREFIX = "jms.topic.";
+
+ public static final String JMS_TEMP_TOPIC_ADDRESS_PREFIX = "jms.temptopic.";
+
+ private static final char SEPARATOR = '.';
+
+ private static String escape(final String input)
+ {
+ if (input == null)
+ {
+ return "";
+ }
+ return input.replace("\\", "\\\\").replace(".", "\\.");
+ }
+
+ public static Destination fromAddress(final String address)
+ {
+ if (address.startsWith(HornetQDestination.JMS_QUEUE_ADDRESS_PREFIX))
+ {
+ String name = address.substring(HornetQDestination.JMS_QUEUE_ADDRESS_PREFIX.length());
+
+ return createQueue(name);
+ }
+ else if (address.startsWith(HornetQDestination.JMS_TOPIC_ADDRESS_PREFIX))
+ {
+ String name = address.substring(HornetQDestination.JMS_TOPIC_ADDRESS_PREFIX.length());
+
+ return createTopic(name);
+ }
+ else if (address.startsWith(HornetQDestination.JMS_TEMP_QUEUE_ADDRESS_PREFIX))
+ {
+ String name = address.substring(HornetQDestination.JMS_TEMP_QUEUE_ADDRESS_PREFIX.length());
+
+ return new HornetQTemporaryQueue(address, name, null);
+ }
+ else if (address.startsWith(HornetQDestination.JMS_TEMP_TOPIC_ADDRESS_PREFIX))
+ {
+ String name = address.substring(HornetQDestination.JMS_TEMP_TOPIC_ADDRESS_PREFIX.length());
+
+ return new HornetQTemporaryTopic(address, name, null);
+ }
+ else
+ {
+ throw new JMSRuntimeException("Invalid address " + address);
+ }
+ }
+
+ public static String createQueueNameForDurableSubscription(final boolean isDurable, final String clientID, final String subscriptionName)
+ {
+ if (clientID != null)
+ {
+ if (isDurable)
+ {
+ return HornetQDestination.escape(clientID) + SEPARATOR +
+ HornetQDestination.escape(subscriptionName);
+ }
+ else
+ {
+ return "nonDurable" + SEPARATOR +
+ HornetQDestination.escape(clientID) + SEPARATOR +
+ HornetQDestination.escape(subscriptionName);
+ }
+ }
+ else
+ {
+ if (isDurable)
+ {
+ return HornetQDestination.escape(subscriptionName);
+ }
+ else
+ {
+ return "nonDurable" + SEPARATOR +
+ HornetQDestination.escape(subscriptionName);
+ }
+ }
+ }
+
+ public static String createQueueNameForSharedSubscription(final boolean isDurable, final String clientID, final String subscriptionName)
+ {
+ if (clientID != null)
+ {
+ return (isDurable ? "Durable" : "nonDurable") + SEPARATOR +
+ HornetQDestination.escape(clientID) + SEPARATOR +
+ HornetQDestination.escape(subscriptionName);
+ }
+ else
+ {
+ return (isDurable ? "Durable" : "nonDurable") + SEPARATOR +
+ HornetQDestination.escape(subscriptionName);
+ }
+ }
+
+ public static Pair<String, String> decomposeQueueNameForDurableSubscription(final String queueName)
+ {
+ StringBuffer[] parts = new StringBuffer[2];
+ int currentPart = 0;
+
+ parts[0] = new StringBuffer();
+ parts[1] = new StringBuffer();
+
+ int pos = 0;
+ while (pos < queueName.length())
+ {
+ char ch = queueName.charAt(pos);
+ pos++;
+
+ if (ch == SEPARATOR)
+ {
+ currentPart++;
+ if (currentPart >= parts.length)
+ {
+ throw new JMSRuntimeException("Invalid message queue name: " + queueName);
+ }
+
+ continue;
+ }
+
+ if (ch == '\\')
+ {
+ if (pos >= queueName.length())
+ {
+ throw new JMSRuntimeException("Invalid message queue name: " + queueName);
+ }
+ ch = queueName.charAt(pos);
+ pos++;
+ }
+
+ parts[currentPart].append(ch);
+ }
+
+ if (currentPart != 1)
+ {
+ throw new JMSRuntimeException("Invalid message queue name: " + queueName);
+ }
+
+ Pair<String, String> pair = new Pair<String, String>(parts[0].toString(), parts[1].toString());
+
+ return pair;
+ }
+
+ public static SimpleString createQueueAddressFromName(final String name)
+ {
+ return new SimpleString(JMS_QUEUE_ADDRESS_PREFIX + name);
+ }
+
+ public static SimpleString createTopicAddressFromName(final String name)
+ {
+ return new SimpleString(JMS_TOPIC_ADDRESS_PREFIX + name);
+ }
+
+ public static HornetQQueue createQueue(final String name)
+ {
+ return new HornetQQueue(name);
+ }
+
+ public static HornetQTopic createTopic(final String name)
+ {
+ return new HornetQTopic(name);
+ }
+
+ public static HornetQTemporaryQueue createTemporaryQueue(final String name, final HornetQSession session)
+ {
+ return new HornetQTemporaryQueue(JMS_TEMP_QUEUE_ADDRESS_PREFIX.concat(name), name, session);
+ }
+
+ public static HornetQTemporaryQueue createTemporaryQueue(final String name)
+ {
+ return createTemporaryQueue(name, null);
+ }
+
+ public static HornetQTemporaryQueue createTemporaryQueue(final HornetQSession session)
+ {
+ String name = UUID.randomUUID().toString();
+
+ return createTemporaryQueue(name, session);
+ }
+
+ public static HornetQTemporaryTopic createTemporaryTopic(final HornetQSession session)
+ {
+ String name = UUID.randomUUID().toString();
+
+ return createTemporaryTopic(name, session);
+ }
+
+ public static HornetQTemporaryTopic createTemporaryTopic(String name, final HornetQSession session)
+ {
+ return new HornetQTemporaryTopic(JMS_TEMP_TOPIC_ADDRESS_PREFIX.concat(name), name, session);
+ }
+
+ public static HornetQTemporaryTopic createTemporaryTopic(String name)
+ {
+ return createTemporaryTopic(name, null);
+ }
+
+ // Attributes ----------------------------------------------------
+
+ /**
+ * The JMS name
+ */
+ protected final String name;
+
+ /**
+ * The core address
+ */
+ private final String address;
+
+ /**
+ * SimpleString version of address
+ */
+ private final SimpleString simpleAddress;
+
+ private final boolean temporary;
+
+ private final boolean queue;
+
+ private final transient HornetQSession session;
+
+ // Constructors --------------------------------------------------
+
+ protected HornetQDestination(final String address, final String name,
+ final boolean temporary,
+ final boolean queue,
+ final HornetQSession session)
+ {
+ this.address = address;
+
+ this.name = name;
+
+ simpleAddress = new SimpleString(address);
+
+ this.temporary = temporary;
+
+ this.queue = queue;
+
+ this.session = session;
+ }
+
+ // Referenceable implementation ---------------------------------------
+
+ public Reference getReference() throws NamingException
+ {
+ return new Reference(this.getClass().getCanonicalName(),
+ new SerializableObjectRefAddr("HornetQ-DEST", this),
+ DestinationObjectFactory.class.getCanonicalName(),
+ null);
+ }
+
+ public void delete() throws JMSException
+ {
+ if (session != null)
+ {
+ if (session.getCoreSession().isClosed())
+ {
+ // Temporary queues will be deleted when the connection is closed.. nothing to be done then!
+ return;
+ }
+ if (queue)
+ {
+ session.deleteTemporaryQueue(this);
+ }
+ else
+ {
+ session.deleteTemporaryTopic(this);
+ }
+ }
+ }
+
+ public boolean isQueue()
+ {
+ return queue;
+ }
+
+ // Public --------------------------------------------------------
+
+ public String getAddress()
+ {
+ return address;
+ }
+
+ public SimpleString getSimpleAddress()
+ {
+ return simpleAddress;
+ }
+
+ public String getName()
+ {
+ return name;
+ }
+
+ public boolean isTemporary()
+ {
+ return temporary;
+ }
+
+ @Override
+ public boolean equals(final Object o)
+ {
+ if (this == o)
+ {
+ return true;
+ }
+
+ if (!(o instanceof HornetQDestination))
+ {
+ return false;
+ }
+
+ HornetQDestination that = (HornetQDestination)o;
+
+ return address.equals(that.address);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return address.hashCode();
+ }
+
+ // Package protected ---------------------------------------------
+
+ // Protected -----------------------------------------------------
+
+ // Private -------------------------------------------------------
+
+ // Inner classes -------------------------------------------------
+}
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/23e8edd9/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQJMSClientBundle.java
----------------------------------------------------------------------
diff --git a/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQJMSClientBundle.java b/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQJMSClientBundle.java
new file mode 100644
index 0000000..d5af6df
--- /dev/null
+++ b/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQJMSClientBundle.java
@@ -0,0 +1,97 @@
+/*
+ * Copyright 2005-2014 Red Hat, Inc.
+ * Red Hat 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.activemq6.jms.client;
+
+
+import javax.jms.IllegalStateException;
+import javax.jms.IllegalStateRuntimeException;
+import javax.jms.InvalidDestinationException;
+import javax.jms.JMSException;
+import javax.jms.JMSRuntimeException;
+import javax.jms.MessageNotReadableException;
+import javax.jms.MessageNotWriteableException;
+
+import org.apache.activemq6.api.core.HornetQIllegalStateException;
+import org.apache.activemq6.api.core.HornetQInvalidFilterExpressionException;
+import org.apache.activemq6.api.core.HornetQNonExistentQueueException;
+import org.apache.activemq6.api.core.SimpleString;
+import org.jboss.logging.annotations.Cause;
+import org.jboss.logging.annotations.Message;
+import org.jboss.logging.annotations.MessageBundle;
+import org.jboss.logging.Messages;
+
+/**
+ * @author <a href="mailto:andy.taylor@jboss.org">Andy Taylor</a>
+ * 3/12/12
+ *
+ * Logger Code 12
+ *
+ * each message id must be 6 digits long starting with 10, the 3rd digit should be 9
+ *
+ * so 129000 to 129999
+ */
+@MessageBundle(projectCode = "HQ")
+public interface HornetQJMSClientBundle
+{
+ HornetQJMSClientBundle BUNDLE = Messages.getBundle(HornetQJMSClientBundle.class);
+
+ @Message(id = 129000, value = "Invalid filter: {0}", format = Message.Format.MESSAGE_FORMAT)
+ HornetQInvalidFilterExpressionException invalidFilter(@Cause Throwable e, SimpleString filter);
+
+ @Message(id = 129001, value = "Invalid Subscription Name. It is required to set the subscription name", format = Message.Format.MESSAGE_FORMAT)
+ HornetQIllegalStateException invalidSubscriptionName();
+
+ @Message(id = 129002, value = "Destination {0} does not exist", format = Message.Format.MESSAGE_FORMAT)
+ HornetQNonExistentQueueException destinationDoesNotExist(SimpleString destination);
+
+ @Message(id = 129003, value = "name cannot be null", format = Message.Format.MESSAGE_FORMAT)
+ IllegalArgumentException nameCannotBeNull();
+
+ @Message(id = 129004, value = "name cannot be empty", format = Message.Format.MESSAGE_FORMAT)
+ IllegalArgumentException nameCannotBeEmpty();
+
+ @Message(id = 129005, value = "It is illegal to call this method from within a Message Listener", format = Message.Format.MESSAGE_FORMAT)
+ IllegalStateRuntimeException callingMethodFromListenerRuntime();
+
+ @Message(id = 129006, value = "It is illegal to call this method from within a Message Listener", format = Message.Format.MESSAGE_FORMAT)
+ IllegalStateException callingMethodFromListener();
+
+ @Message(id = 129007, value = "It is illegal to call this method from within a Completion Listener", format = Message.Format.MESSAGE_FORMAT)
+ IllegalStateRuntimeException callingMethodFromCompletionListenerRuntime();
+
+ @Message(id = 129008, value = "It is illegal to call this method from within a Completion Listener", format = Message.Format.MESSAGE_FORMAT)
+ IllegalStateException callingMethodFromCompletionListener();
+
+ @Message(id = 129009, value = "Null {0} is not allowed", format = Message.Format.MESSAGE_FORMAT)
+ IllegalArgumentException nullArgumentNotAllowed(String type);
+
+ @Message(id = 129010, value = "Topic (Destination) cannot be null", format = Message.Format.MESSAGE_FORMAT)
+ InvalidDestinationException nullTopic();
+
+ @Message(id = 129011, value = "LargeMessage streaming is only possible on ByteMessage or StreamMessage",
+ format = Message.Format.MESSAGE_FORMAT)
+ IllegalStateException onlyValidForByteOrStreamMessages();
+
+ @Message(id = 129012, value = "The property name ''{0}'' is not a valid java identifier.",
+ format = Message.Format.MESSAGE_FORMAT)
+ JMSRuntimeException invalidJavaIdentifier(String propertyName);
+
+ @Message(id = 129013, value = "Message is read-only", format = Message.Format.MESSAGE_FORMAT)
+ MessageNotWriteableException messageNotWritable();
+
+ @Message(id = 129014, value = "Message is write-only", format = Message.Format.MESSAGE_FORMAT)
+ MessageNotReadableException messageNotReadable();
+
+ @Message(id = 129015, value = "Illegal deliveryMode value: {0}", format = Message.Format.MESSAGE_FORMAT)
+ JMSException illegalDeliveryMode(int deliveryMode);
+}
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/23e8edd9/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQJMSClientLogger.java
----------------------------------------------------------------------
diff --git a/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQJMSClientLogger.java b/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQJMSClientLogger.java
new file mode 100644
index 0000000..1111b98
--- /dev/null
+++ b/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQJMSClientLogger.java
@@ -0,0 +1,79 @@
+/*
+ * Copyright 2005-2014 Red Hat, Inc.
+ * Red Hat 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.activemq6.jms.client;
+
+import org.jboss.logging.BasicLogger;
+import org.jboss.logging.Logger;
+import org.jboss.logging.annotations.Cause;
+import org.jboss.logging.annotations.LogMessage;
+import org.jboss.logging.annotations.Message;
+import org.jboss.logging.annotations.MessageLogger;
+
+/**
+ * @author <a href="mailto:andy.taylor@jboss.org">Andy Taylor</a>
+ * 3/15/12
+ *
+ * Logger Code 12
+ *
+ * each message id must be 6 digits long starting with 12, the 3rd digit donates the level so
+ *
+ * INF0 1
+ * WARN 2
+ * DEBUG 3
+ * ERROR 4
+ * TRACE 5
+ * FATAL 6
+ *
+ * so an INFO message would be 121000 to 121999
+ */
+@MessageLogger(projectCode = "HQ")
+public interface HornetQJMSClientLogger extends BasicLogger
+{
+ /**
+ * The default logger.
+ */
+ HornetQJMSClientLogger LOGGER = Logger.getMessageLogger(HornetQJMSClientLogger.class, HornetQJMSClientLogger.class.getPackage().getName());
+
+ @LogMessage(level = Logger.Level.WARN)
+ @Message(id = 122000, value = "I''m closing a JMS connection you left open. Please make sure you close all JMS connections explicitly before letting them go out of scope! see stacktrace to find out where it was created" , format = Message.Format.MESSAGE_FORMAT)
+ void connectionLeftOpen(@Cause Exception e);
+
+ @LogMessage(level = Logger.Level.WARN)
+ @Message(id = 122001, value = "Unhandled exception thrown from onMessage" , format = Message.Format.MESSAGE_FORMAT)
+ void onMessageError(@Cause Exception e);
+
+ @LogMessage(level = Logger.Level.ERROR)
+ @Message(id = 124000, value = "Failed to call JMS exception listener" , format = Message.Format.MESSAGE_FORMAT)
+ void errorCallingExcListener(@Cause Exception e);
+
+ @LogMessage(level = Logger.Level.ERROR)
+ @Message(id = 124002, value = "Queue Browser failed to create message" , format = Message.Format.MESSAGE_FORMAT)
+ void errorCreatingMessage(@Cause Throwable e);
+
+ @LogMessage(level = Logger.Level.ERROR)
+ @Message(id = 124003, value = "Message Listener failed to prepare message for receipt" , format = Message.Format.MESSAGE_FORMAT)
+ void errorPreparingMessageForReceipt(@Cause Throwable e);
+
+ @LogMessage(level = Logger.Level.ERROR)
+ @Message(id = 124004, value = "Message Listener failed to process message" , format = Message.Format.MESSAGE_FORMAT)
+ void errorProcessingMessage(@Cause Throwable e);
+
+ @LogMessage(level = Logger.Level.ERROR)
+ @Message(id = 124005, value = "Message Listener failed to recover session" , format = Message.Format.MESSAGE_FORMAT)
+ void errorRecoveringSession(@Cause Throwable e);
+
+ @LogMessage(level = Logger.Level.ERROR)
+ @Message(id = 124006, value = "Failed to call Failover listener" , format = Message.Format.MESSAGE_FORMAT)
+ void errorCallingFailoverListener(@Cause Exception e);
+
+}
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/23e8edd9/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQJMSConnectionFactory.java
----------------------------------------------------------------------
diff --git a/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQJMSConnectionFactory.java b/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQJMSConnectionFactory.java
new file mode 100644
index 0000000..d1d1deb
--- /dev/null
+++ b/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQJMSConnectionFactory.java
@@ -0,0 +1,67 @@
+/*
+ * Copyright 2005-2014 Red Hat, Inc.
+ * Red Hat 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.activemq6.jms.client;
+
+import javax.jms.QueueConnectionFactory;
+import javax.jms.TopicConnectionFactory;
+
+import org.apache.activemq6.api.core.DiscoveryGroupConfiguration;
+import org.apache.activemq6.api.core.TransportConfiguration;
+import org.apache.activemq6.api.core.client.ServerLocator;
+
+
+/**
+ * A class that represents a ConnectionFactory.
+ *
+ * @author <a href="mailto:hgao@redhat.com">Howard Gao</a>
+ */
+public class HornetQJMSConnectionFactory extends HornetQConnectionFactory implements TopicConnectionFactory, QueueConnectionFactory
+{
+
+ private static final long serialVersionUID = -2810634789345348326L;
+
+ /**
+ *
+ */
+ public HornetQJMSConnectionFactory()
+ {
+ super();
+ }
+
+ /**
+ * @param serverLocator
+ */
+ public HornetQJMSConnectionFactory(ServerLocator serverLocator)
+ {
+ super(serverLocator);
+ }
+
+ /**
+ * @param ha
+ * @param groupConfiguration
+ */
+ public HornetQJMSConnectionFactory(boolean ha, final DiscoveryGroupConfiguration groupConfiguration)
+ {
+ super(ha, groupConfiguration);
+ }
+
+ /**
+ * @param ha
+ * @param initialConnectors
+ */
+ public HornetQJMSConnectionFactory(boolean ha, TransportConfiguration... initialConnectors)
+ {
+ super(ha, initialConnectors);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/23e8edd9/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQJMSConsumer.java
----------------------------------------------------------------------
diff --git a/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQJMSConsumer.java b/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQJMSConsumer.java
new file mode 100644
index 0000000..bdbeb7d
--- /dev/null
+++ b/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQJMSConsumer.java
@@ -0,0 +1,198 @@
+/*
+ * Copyright 2005-2014 Red Hat, Inc.
+ * Red Hat 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.activemq6.jms.client;
+
+import javax.jms.JMSConsumer;
+import javax.jms.JMSException;
+import javax.jms.JMSRuntimeException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageListener;
+
+/**
+ * @author <a href="http://jmesnil.net/">Jeff Mesnil</a> (c) 2013 Red Hat inc.
+ */
+public class HornetQJMSConsumer implements JMSConsumer
+{
+
+ private final HornetQJMSContext context;
+ private final MessageConsumer consumer;
+
+ HornetQJMSConsumer(HornetQJMSContext context, MessageConsumer consumer)
+ {
+ this.context = context;
+ this.consumer = consumer;
+ }
+
+ @Override
+ public String getMessageSelector()
+ {
+ try
+ {
+ return consumer.getMessageSelector();
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public MessageListener getMessageListener() throws JMSRuntimeException
+ {
+ try
+ {
+ return consumer.getMessageListener();
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public void setMessageListener(MessageListener listener) throws JMSRuntimeException
+ {
+ try
+ {
+ consumer.setMessageListener(new MessageListenerWrapper(listener));
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public Message receive()
+ {
+ try
+ {
+ return context.setLastMessage(this, consumer.receive());
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public Message receive(long timeout)
+ {
+ try
+ {
+ return context.setLastMessage(this, consumer.receive(timeout));
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public Message receiveNoWait()
+ {
+ try
+ {
+ return context.setLastMessage(this, consumer.receiveNoWait());
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public void close()
+ {
+ try
+ {
+ consumer.close();
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public <T> T receiveBody(Class<T> c)
+ {
+ try
+ {
+ Message message = consumer.receive();
+ context.setLastMessage(HornetQJMSConsumer.this, message);
+ return message == null ? null : message.getBody(c);
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public <T> T receiveBody(Class<T> c, long timeout)
+ {
+ try
+ {
+ Message message = consumer.receive(timeout);
+ context.setLastMessage(HornetQJMSConsumer.this, message);
+ return message == null ? null : message.getBody(c);
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public <T> T receiveBodyNoWait(Class<T> c)
+ {
+ try
+ {
+ Message message = consumer.receiveNoWait();
+ context.setLastMessage(HornetQJMSConsumer.this, message);
+ return message == null ? null : message.getBody(c);
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ final class MessageListenerWrapper implements MessageListener
+ {
+ private final MessageListener wrapped;
+
+ public MessageListenerWrapper(MessageListener wrapped)
+ {
+ this.wrapped = wrapped;
+ }
+
+ @Override
+ public void onMessage(Message message)
+ {
+ context.setLastMessage(HornetQJMSConsumer.this, message);
+
+ context.getThreadAwareContext().setCurrentThread(false);
+ try
+ {
+ wrapped.onMessage(message);
+ }
+ finally
+ {
+ context.getThreadAwareContext().clearCurrentThread(false);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/23e8edd9/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQJMSContext.java
----------------------------------------------------------------------
diff --git a/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQJMSContext.java b/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQJMSContext.java
new file mode 100644
index 0000000..ba3364d
--- /dev/null
+++ b/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQJMSContext.java
@@ -0,0 +1,769 @@
+/*
+ * Copyright 2005-2014 Red Hat, Inc.
+ * Red Hat 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.activemq6.jms.client;
+
+import javax.jms.BytesMessage;
+import javax.jms.ConnectionMetaData;
+import javax.jms.Destination;
+import javax.jms.ExceptionListener;
+import javax.jms.IllegalStateRuntimeException;
+import javax.jms.JMSConsumer;
+import javax.jms.JMSContext;
+import javax.jms.JMSException;
+import javax.jms.JMSProducer;
+import javax.jms.MapMessage;
+import javax.jms.Message;
+import javax.jms.MessageProducer;
+import javax.jms.ObjectMessage;
+import javax.jms.Queue;
+import javax.jms.QueueBrowser;
+import javax.jms.Session;
+import javax.jms.StreamMessage;
+import javax.jms.TemporaryQueue;
+import javax.jms.TemporaryTopic;
+import javax.jms.TextMessage;
+import javax.jms.Topic;
+import javax.jms.XAConnection;
+import javax.jms.XASession;
+import javax.transaction.xa.XAResource;
+import java.io.Serializable;
+
+/**
+ * HornetQ implementation of a JMSContext.
+ *
+ * @author <a href="http://jmesnil.net/">Jeff Mesnil</a> (c) 2013 Red Hat inc
+ */
+public class HornetQJMSContext implements JMSContext
+{
+ private static final boolean DEFAULT_AUTO_START = true;
+ private final int sessionMode;
+
+ private final ThreadAwareContext threadAwareContext;
+
+ /**
+ * Client ACK needs to hold last acked messages, so context.ack calls will be respected.
+ */
+ private volatile Message lastMessagesWaitingAck;
+
+ private final HornetQConnectionForContext connection;
+ private Session session;
+ private boolean autoStart = HornetQJMSContext.DEFAULT_AUTO_START;
+ private MessageProducer innerProducer;
+ private boolean xa;
+ private boolean closed;
+
+ HornetQJMSContext(final HornetQConnectionForContext connection, final int ackMode, final boolean xa, ThreadAwareContext threadAwareContext)
+ {
+ this.connection = connection;
+ this.sessionMode = ackMode;
+ this.xa = xa;
+ this.threadAwareContext = threadAwareContext;
+ }
+
+ public HornetQJMSContext(HornetQConnectionForContext connection, int ackMode, ThreadAwareContext threadAwareContext)
+ {
+ this(connection, ackMode, false, threadAwareContext);
+ }
+
+ public HornetQJMSContext(HornetQConnectionForContext connection, ThreadAwareContext threadAwareContext)
+ {
+ this(connection, SESSION_TRANSACTED, true, threadAwareContext);
+ }
+
+ // XAJMSContext implementation -------------------------------------
+
+ public JMSContext getContext()
+ {
+ return this;
+ }
+
+ public Session getSession()
+ {
+ return session;
+ }
+
+ public XAResource getXAResource()
+ {
+ checkSession();
+ return ((XASession) session).getXAResource();
+ }
+
+ // JMSContext implementation -------------------------------------
+
+ @Override
+ public JMSContext createContext(int sessionMode)
+ {
+ return connection.createContext(sessionMode);
+ }
+
+ @Override
+ public JMSProducer createProducer()
+ {
+ checkSession();
+ try
+ {
+ return new HornetQJMSProducer(this, getInnerProducer());
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ private synchronized MessageProducer getInnerProducer() throws JMSException
+ {
+ if (innerProducer == null)
+ {
+ innerProducer = session.createProducer(null);
+ }
+
+ return innerProducer;
+ }
+
+ /**
+ *
+ */
+ private void checkSession()
+ {
+ if (session == null)
+ {
+ synchronized (this)
+ {
+ if (closed)
+ throw new IllegalStateRuntimeException("Context is closed");
+ if (session == null)
+ {
+ try
+ {
+ if (xa)
+ {
+ session = ((XAConnection) connection).createXASession();
+ }
+ else
+ {
+ session = connection.createSession(sessionMode);
+ }
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+ }
+ }
+ }
+
+ @Override
+ public String getClientID()
+ {
+ try
+ {
+ return connection.getClientID();
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public void setClientID(String clientID)
+ {
+ try
+ {
+ connection.setClientID(clientID);
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public ConnectionMetaData getMetaData()
+ {
+ try
+ {
+ return connection.getMetaData();
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public ExceptionListener getExceptionListener()
+ {
+ try
+ {
+ return connection.getExceptionListener();
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public void setExceptionListener(ExceptionListener listener)
+ {
+ try
+ {
+ connection.setExceptionListener(listener);
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public void start()
+ {
+ try
+ {
+ connection.start();
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public void stop()
+ {
+ threadAwareContext.assertNotMessageListenerThreadRuntime();
+ try
+ {
+ connection.stop();
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public void setAutoStart(boolean autoStart)
+ {
+ this.autoStart = autoStart;
+ }
+
+ @Override
+ public boolean getAutoStart()
+ {
+ return autoStart;
+ }
+
+ @Override
+ public void close()
+ {
+ threadAwareContext.assertNotCompletionListenerThreadRuntime();
+ threadAwareContext.assertNotMessageListenerThreadRuntime();
+ try
+ {
+ synchronized (this)
+ {
+ if (session != null)
+ session.close();
+ connection.closeFromContext();
+ closed = true;
+ }
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public BytesMessage createBytesMessage()
+ {
+ checkSession();
+ try
+ {
+ return session.createBytesMessage();
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public MapMessage createMapMessage()
+ {
+ checkSession();
+ try
+ {
+ return session.createMapMessage();
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public Message createMessage()
+ {
+ checkSession();
+ try
+ {
+ return session.createMessage();
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public ObjectMessage createObjectMessage()
+ {
+ checkSession();
+ try
+ {
+ return session.createObjectMessage();
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public ObjectMessage createObjectMessage(Serializable object)
+ {
+ checkSession();
+ try
+ {
+ return session.createObjectMessage(object);
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public StreamMessage createStreamMessage()
+ {
+ checkSession();
+ try
+ {
+ return session.createStreamMessage();
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public TextMessage createTextMessage()
+ {
+ checkSession();
+ try
+ {
+ return session.createTextMessage();
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public TextMessage createTextMessage(String text)
+ {
+ checkSession();
+ try
+ {
+ return session.createTextMessage(text);
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public boolean getTransacted()
+ {
+ checkSession();
+ try
+ {
+ return session.getTransacted();
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public int getSessionMode()
+ {
+ return sessionMode;
+ }
+
+ @Override
+ public void commit()
+ {
+ threadAwareContext.assertNotCompletionListenerThreadRuntime();
+ checkSession();
+ try
+ {
+ session.commit();
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public void rollback()
+ {
+ threadAwareContext.assertNotCompletionListenerThreadRuntime();
+ checkSession();
+ try
+ {
+ session.rollback();
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public void recover()
+ {
+ checkSession();
+ try
+ {
+ session.recover();
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public JMSConsumer createConsumer(Destination destination)
+ {
+ checkSession();
+ try
+ {
+ HornetQJMSConsumer consumer = new HornetQJMSConsumer(this, session.createConsumer(destination));
+ checkAutoStart();
+ return consumer;
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public JMSConsumer createConsumer(Destination destination, String messageSelector)
+ {
+ checkSession();
+ try
+ {
+ HornetQJMSConsumer consumer = new HornetQJMSConsumer(this, session.createConsumer(destination, messageSelector));
+ checkAutoStart();
+ return consumer;
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public JMSConsumer createConsumer(Destination destination, String messageSelector, boolean noLocal)
+ {
+ checkSession();
+ try
+ {
+ HornetQJMSConsumer consumer = new HornetQJMSConsumer(this, session.createConsumer(destination, messageSelector, noLocal));
+ checkAutoStart();
+ return consumer;
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public Queue createQueue(String queueName)
+ {
+ checkSession();
+ try
+ {
+ return session.createQueue(queueName);
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public Topic createTopic(String topicName)
+ {
+ checkSession();
+ try
+ {
+ return session.createTopic(topicName);
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public JMSConsumer createDurableConsumer(Topic topic, String name)
+ {
+ checkSession();
+ try
+ {
+ HornetQJMSConsumer consumer = new HornetQJMSConsumer(this, session.createDurableConsumer(topic, name));
+ checkAutoStart();
+ return consumer;
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public JMSConsumer createDurableConsumer(Topic topic, String name, String messageSelector, boolean noLocal)
+ {
+ checkSession();
+ try
+ {
+ HornetQJMSConsumer consumer = new HornetQJMSConsumer(this, session.createDurableConsumer(topic, name, messageSelector, noLocal));
+ checkAutoStart();
+ return consumer;
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public JMSConsumer createSharedDurableConsumer(Topic topic, String name)
+ {
+ checkSession();
+ try
+ {
+ HornetQJMSConsumer consumer = new HornetQJMSConsumer(this, session.createSharedDurableConsumer(topic, name));
+ checkAutoStart();
+ return consumer;
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public JMSConsumer createSharedDurableConsumer(Topic topic, String name, String messageSelector)
+ {
+ checkSession();
+ try
+ {
+ HornetQJMSConsumer consumer = new HornetQJMSConsumer(this, session.createSharedDurableConsumer(topic, name, messageSelector));
+ checkAutoStart();
+ return consumer;
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public JMSConsumer createSharedConsumer(Topic topic, String sharedSubscriptionName)
+ {
+ checkSession();
+ try
+ {
+ HornetQJMSConsumer consumer = new HornetQJMSConsumer(this, session.createSharedConsumer(topic, sharedSubscriptionName));
+ checkAutoStart();
+ return consumer;
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public JMSConsumer createSharedConsumer(Topic topic, String sharedSubscriptionName, String messageSelector)
+ {
+ checkSession();
+ try
+ {
+ HornetQJMSConsumer consumer = new HornetQJMSConsumer(this, session.createSharedConsumer(topic, sharedSubscriptionName, messageSelector));
+ checkAutoStart();
+ return consumer;
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public QueueBrowser createBrowser(Queue queue)
+ {
+ checkSession();
+ try
+ {
+ QueueBrowser browser = session.createBrowser(queue);
+ checkAutoStart();
+ return browser;
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public QueueBrowser createBrowser(Queue queue, String messageSelector)
+ {
+ checkSession();
+ try
+ {
+ QueueBrowser browser = session.createBrowser(queue, messageSelector);
+ checkAutoStart();
+ return browser;
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public TemporaryQueue createTemporaryQueue()
+ {
+ checkSession();
+ try
+ {
+ return session.createTemporaryQueue();
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public TemporaryTopic createTemporaryTopic()
+ {
+ checkSession();
+ try
+ {
+ return session.createTemporaryTopic();
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public void unsubscribe(String name)
+ {
+ checkSession();
+ try
+ {
+ session.unsubscribe(name);
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public void acknowledge()
+ {
+ checkSession();
+ if (closed)
+ throw new IllegalStateRuntimeException("Context is closed");
+ try
+ {
+ if (lastMessagesWaitingAck != null)
+ {
+ lastMessagesWaitingAck.acknowledge();
+ }
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ /**
+ * This is to be used on tests only. It's not part of the interface and it's not guaranteed to be kept
+ * on the API contract.
+ *
+ * @return
+ */
+ public Session getUsedSession()
+ {
+ return this.session;
+ }
+
+ private synchronized void checkAutoStart() throws JMSException
+ {
+ if (closed)
+ throw new IllegalStateRuntimeException("Context is closed");
+ if (autoStart)
+ {
+ connection.start();
+ }
+ }
+
+ /**
+ * this is to ensure Context.acknowledge would work on ClientACK
+ */
+ Message setLastMessage(final JMSConsumer consumer, final Message lastMessageReceived)
+ {
+ if (sessionMode == CLIENT_ACKNOWLEDGE)
+ {
+ lastMessagesWaitingAck = lastMessageReceived;
+ }
+ return lastMessageReceived;
+ }
+
+ public ThreadAwareContext getThreadAwareContext()
+ {
+ return threadAwareContext;
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/23e8edd9/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQJMSProducer.java
----------------------------------------------------------------------
diff --git a/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQJMSProducer.java b/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQJMSProducer.java
new file mode 100644
index 0000000..f39f9ca
--- /dev/null
+++ b/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQJMSProducer.java
@@ -0,0 +1,800 @@
+/*
+ * Copyright 2005-2014 Red Hat, Inc.
+ * Red Hat 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.activemq6.jms.client;
+
+import javax.jms.BytesMessage;
+import javax.jms.CompletionListener;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.JMSProducer;
+import javax.jms.JMSRuntimeException;
+import javax.jms.MapMessage;
+import javax.jms.Message;
+import javax.jms.MessageFormatRuntimeException;
+import javax.jms.MessageProducer;
+import javax.jms.ObjectMessage;
+import javax.jms.TextMessage;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+
+import org.apache.activemq6.api.core.HornetQPropertyConversionException;
+import org.apache.activemq6.api.core.SimpleString;
+import org.apache.activemq6.utils.TypedProperties;
+
+/**
+ * NOTE: this class forwards {@link #setDisableMessageID(boolean)} and
+ * {@link #setDisableMessageTimestamp(boolean)} calls their equivalent at the
+ * {@link MessageProducer}. IF the user is using the producer in async mode, this may lead to races.
+ * We allow/tolerate this because these are just optional optimizations.
+ *
+ * @author <a href="http://jmesnil.net/">Jeff Mesnil</a> (c) 2013 Red Hat inc.
+ */
+public final class HornetQJMSProducer implements JMSProducer
+{
+ private final HornetQJMSContext context;
+ private final MessageProducer producer;
+ private final TypedProperties properties = new TypedProperties();
+
+ //we convert Strings to SimpleStrings so if getProperty is called the wrong object is returned, this list let's us return the
+ //correct type
+ private final List<SimpleString> stringPropertyNames = new ArrayList<>();
+
+ private volatile CompletionListener completionListener;
+
+ private Destination jmsHeaderReplyTo;
+ private String jmsHeaderCorrelationID;
+ private byte[] jmsHeaderCorrelationIDAsBytes;
+ private String jmsHeaderType;
+
+ HornetQJMSProducer(HornetQJMSContext context, MessageProducer producer)
+ {
+ this.context = context;
+ this.producer = producer;
+ }
+
+ @Override
+ public JMSProducer send(Destination destination, Message message)
+ {
+ if (message == null)
+ {
+ throw new MessageFormatRuntimeException("null message");
+ }
+
+ try
+ {
+ if (jmsHeaderCorrelationID != null)
+ {
+ message.setJMSCorrelationID(jmsHeaderCorrelationID);
+ }
+ if (jmsHeaderCorrelationIDAsBytes != null && jmsHeaderCorrelationIDAsBytes.length > 0)
+ {
+ message.setJMSCorrelationIDAsBytes(jmsHeaderCorrelationIDAsBytes);
+ }
+ if (jmsHeaderReplyTo != null)
+ {
+ message.setJMSReplyTo(jmsHeaderReplyTo);
+ }
+ if (jmsHeaderType != null)
+ {
+ message.setJMSType(jmsHeaderType);
+ }
+ // XXX HORNETQ-1209 "JMS 2.0" can this be a foreign msg?
+ // if so, then "SimpleString" properties will trigger an error.
+ setProperties(message);
+ if (completionListener != null)
+ {
+ CompletionListener wrapped = new CompletionListenerWrapper(completionListener);
+ producer.send(destination, message, wrapped);
+ }
+ else
+ {
+ producer.send(destination, message);
+ }
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ return this;
+ }
+
+ /**
+ * Sets all properties we carry onto the message.
+ *
+ * @param message
+ * @throws JMSException
+ */
+ private void setProperties(Message message) throws JMSException
+ {
+ for (SimpleString name : properties.getPropertyNames())
+ {
+ message.setObjectProperty(name.toString(), properties.getProperty(name));
+ }
+ }
+
+ @Override
+ public JMSProducer send(Destination destination, String body)
+ {
+ TextMessage message = context.createTextMessage(body);
+ send(destination, message);
+ return this;
+ }
+
+ @Override
+ public JMSProducer send(Destination destination, Map<String, Object> body)
+ {
+ MapMessage message = context.createMapMessage();
+ if (body != null)
+ {
+ try
+ {
+ for (Entry<String, Object> entry : body.entrySet())
+ {
+ final String name = entry.getKey();
+ final Object v = entry.getValue();
+ if (v instanceof String)
+ {
+ message.setString(name, (String) v);
+ }
+ else if (v instanceof Long)
+ {
+ message.setLong(name, (Long) v);
+ }
+ else if (v instanceof Double)
+ {
+ message.setDouble(name, (Double) v);
+ }
+ else if (v instanceof Integer)
+ {
+ message.setInt(name, (Integer) v);
+ }
+ else if (v instanceof Character)
+ {
+ message.setChar(name, (Character) v);
+ }
+ else if (v instanceof Short)
+ {
+ message.setShort(name, (Short) v);
+ }
+ else if (v instanceof Boolean)
+ {
+ message.setBoolean(name, (Boolean) v);
+ }
+ else if (v instanceof Float)
+ {
+ message.setFloat(name, (Float) v);
+ }
+ else if (v instanceof Byte)
+ {
+ message.setByte(name, (Byte) v);
+ }
+ else if (v instanceof byte[])
+ {
+ byte[] array = (byte[]) v;
+ message.setBytes(name, array, 0, array.length);
+ }
+ else
+ {
+ message.setObject(name, v);
+ }
+ }
+ }
+ catch (JMSException e)
+ {
+ throw new MessageFormatRuntimeException(e.getMessage());
+ }
+ }
+ send(destination, message);
+ return this;
+ }
+
+ @Override
+ public JMSProducer send(Destination destination, byte[] body)
+ {
+ BytesMessage message = context.createBytesMessage();
+ if (body != null)
+ {
+ try
+ {
+ message.writeBytes(body);
+ }
+ catch (JMSException e)
+ {
+ throw new MessageFormatRuntimeException(e.getMessage());
+ }
+ }
+ send(destination, message);
+ return this;
+ }
+
+ @Override
+ public JMSProducer send(Destination destination, Serializable body)
+ {
+ ObjectMessage message = context.createObjectMessage(body);
+ send(destination, message);
+ return this;
+ }
+
+ @Override
+ public JMSProducer setDisableMessageID(boolean value)
+ {
+ try
+ {
+ producer.setDisableMessageID(value);
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public boolean getDisableMessageID()
+ {
+ try
+ {
+ return producer.getDisableMessageID();
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public JMSProducer setDisableMessageTimestamp(boolean value)
+ {
+ try
+ {
+ producer.setDisableMessageTimestamp(value);
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public boolean getDisableMessageTimestamp()
+ {
+ try
+ {
+ return producer.getDisableMessageTimestamp();
+ }
+ catch (JMSException e)
+ {
+ throw JmsExceptionUtils.convertToRuntimeException(e);
+ }
+ }
+
+ @Override
+ public JMSProducer setDeliveryMode(int deliveryMode)
+ {
+ try
+ {
+ producer.setDeliveryMode(deliveryMode);
+ }
+ catch (JMSException e)
+ {
+ JMSRuntimeException e2 = new JMSRuntimeException(e.getMessage());
+ e2.initCause(e);
+ throw e2;
+ }
+ return this;
+ }
+
+ @Override
+ public int getDeliveryMode()
+ {
+ try
+ {
+ return producer.getDeliveryMode();
+ }
+ catch (JMSException e)
+ {
+ JMSRuntimeException e2 = new JMSRuntimeException(e.getMessage());
+ e2.initCause(e);
+ throw e2;
+ }
+ }
+
+ @Override
+ public JMSProducer setPriority(int priority)
+ {
+ try
+ {
+ producer.setPriority(priority);
+ }
+ catch (JMSException e)
+ {
+ JMSRuntimeException e2 = new JMSRuntimeException(e.getMessage());
+ e2.initCause(e);
+ throw e2;
+ }
+ return this;
+ }
+
+ @Override
+ public int getPriority()
+ {
+ try
+ {
+ return producer.getPriority();
+ }
+ catch (JMSException e)
+ {
+ JMSRuntimeException e2 = new JMSRuntimeException(e.getMessage());
+ e2.initCause(e);
+ throw e2;
+ }
+ }
+
+ @Override
+ public JMSProducer setTimeToLive(long timeToLive)
+ {
+ try
+ {
+ producer.setTimeToLive(timeToLive);
+ return this;
+ }
+ catch (JMSException e)
+ {
+ JMSRuntimeException e2 = new JMSRuntimeException(e.getMessage());
+ e2.initCause(e);
+ throw e2;
+ }
+ }
+
+ @Override
+ public long getTimeToLive()
+ {
+ long timeToLive = 0;
+ try
+ {
+ timeToLive = producer.getTimeToLive();
+ return timeToLive;
+ }
+ catch (JMSException e)
+ {
+ JMSRuntimeException e2 = new JMSRuntimeException(e.getMessage());
+ e2.initCause(e);
+ throw e2;
+ }
+ }
+
+ @Override
+ public JMSProducer setDeliveryDelay(long deliveryDelay)
+ {
+ try
+ {
+ producer.setDeliveryDelay(deliveryDelay);
+ return this;
+ }
+ catch (JMSException e)
+ {
+ JMSRuntimeException e2 = new JMSRuntimeException(e.getMessage());
+ e2.initCause(e);
+ throw e2;
+ }
+ }
+
+ @Override
+ public long getDeliveryDelay()
+ {
+ long deliveryDelay = 0;
+ try
+ {
+ deliveryDelay = producer.getDeliveryDelay();
+ }
+ catch (Exception ignored)
+ {
+ }
+ return deliveryDelay;
+ }
+
+ @Override
+ public JMSProducer setAsync(CompletionListener completionListener)
+ {
+ this.completionListener = completionListener;
+ return this;
+ }
+
+ @Override
+ public CompletionListener getAsync()
+ {
+ return completionListener;
+ }
+
+ @Override
+ public JMSProducer setProperty(String name, boolean value)
+ {
+ checkName(name);
+ properties.putBooleanProperty(new SimpleString(name), value);
+ return this;
+ }
+
+ @Override
+ public JMSProducer setProperty(String name, byte value)
+ {
+ checkName(name);
+ properties.putByteProperty(new SimpleString(name), value);
+ return this;
+ }
+
+ @Override
+ public JMSProducer setProperty(String name, short value)
+ {
+ checkName(name);
+ properties.putShortProperty(new SimpleString(name), value);
+ return this;
+ }
+
+ @Override
+ public JMSProducer setProperty(String name, int value)
+ {
+ checkName(name);
+ properties.putIntProperty(new SimpleString(name), value);
+ return this;
+ }
+
+ @Override
+ public JMSProducer setProperty(String name, long value)
+ {
+ checkName(name);
+ properties.putLongProperty(new SimpleString(name), value);
+ return this;
+ }
+
+ @Override
+ public JMSProducer setProperty(String name, float value)
+ {
+ checkName(name);
+ properties.putFloatProperty(new SimpleString(name), value);
+ return this;
+ }
+
+ @Override
+ public JMSProducer setProperty(String name, double value)
+ {
+ checkName(name);
+ properties.putDoubleProperty(new SimpleString(name), value);
+ return this;
+ }
+
+ @Override
+ public JMSProducer setProperty(String name, String value)
+ {
+ checkName(name);
+ SimpleString key = new SimpleString(name);
+ properties.putSimpleStringProperty(key, new SimpleString(value));
+ stringPropertyNames.add(key);
+ return this;
+ }
+
+ @Override
+ public JMSProducer setProperty(String name, Object value)
+ {
+ checkName(name);
+ try
+ {
+ TypedProperties.setObjectProperty(new SimpleString(name), value, properties);
+ }
+ catch (HornetQPropertyConversionException hqe)
+ {
+ throw new MessageFormatRuntimeException(hqe.getMessage());
+ }
+ catch (RuntimeException e)
+ {
+ throw new JMSRuntimeException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public JMSProducer clearProperties()
+ {
+ try
+ {
+ stringPropertyNames.clear();
+ properties.clear();
+ }
+ catch (RuntimeException e)
+ {
+ throw new JMSRuntimeException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public boolean propertyExists(String name)
+ {
+ return properties.containsProperty(new SimpleString(name));
+ }
+
+ @Override
+ public boolean getBooleanProperty(String name)
+ {
+ try
+ {
+ return properties.getBooleanProperty(new SimpleString(name));
+ }
+ catch (HornetQPropertyConversionException ce)
+ {
+ throw new MessageFormatRuntimeException(ce.getMessage());
+ }
+ catch (RuntimeException e)
+ {
+ throw new JMSRuntimeException(e.getMessage());
+ }
+ }
+
+ @Override
+ public byte getByteProperty(String name)
+ {
+ try
+ {
+ return properties.getByteProperty(new SimpleString(name));
+ }
+ catch (HornetQPropertyConversionException ce)
+ {
+ throw new MessageFormatRuntimeException(ce.getMessage());
+ }
+ }
+
+ @Override
+ public short getShortProperty(String name)
+ {
+ try
+ {
+ return properties.getShortProperty(new SimpleString(name));
+ }
+ catch (HornetQPropertyConversionException ce)
+ {
+ throw new MessageFormatRuntimeException(ce.getMessage());
+ }
+ }
+
+ @Override
+ public int getIntProperty(String name)
+ {
+ try
+ {
+ return properties.getIntProperty(new SimpleString(name));
+ }
+ catch (HornetQPropertyConversionException ce)
+ {
+ throw new MessageFormatRuntimeException(ce.getMessage());
+ }
+ }
+
+ @Override
+ public long getLongProperty(String name)
+ {
+ try
+ {
+ return properties.getLongProperty(new SimpleString(name));
+ }
+ catch (HornetQPropertyConversionException ce)
+ {
+ throw new MessageFormatRuntimeException(ce.getMessage());
+ }
+ }
+
+ @Override
+ public float getFloatProperty(String name)
+ {
+ try
+ {
+ return properties.getFloatProperty(new SimpleString(name));
+ }
+ catch (HornetQPropertyConversionException ce)
+ {
+ throw new MessageFormatRuntimeException(ce.getMessage());
+ }
+ }
+
+ @Override
+ public double getDoubleProperty(String name)
+ {
+ try
+ {
+ return properties.getDoubleProperty(new SimpleString(name));
+ }
+ catch (HornetQPropertyConversionException ce)
+ {
+ throw new MessageFormatRuntimeException(ce.getMessage());
+ }
+ }
+
+ @Override
+ public String getStringProperty(String name)
+ {
+ try
+ {
+ SimpleString prop = properties.getSimpleStringProperty(new SimpleString(name));
+ if (prop == null)
+ return null;
+ return prop.toString();
+ }
+ catch (HornetQPropertyConversionException ce)
+ {
+ throw new MessageFormatRuntimeException(ce.getMessage());
+ }
+ catch (RuntimeException e)
+ {
+ throw new JMSRuntimeException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Object getObjectProperty(String name)
+ {
+ try
+ {
+ SimpleString key = new SimpleString(name);
+ Object property = properties.getProperty(key);
+ if (stringPropertyNames.contains(key))
+ {
+ property = property.toString();
+ }
+ return property;
+ }
+ catch (HornetQPropertyConversionException ce)
+ {
+ throw new MessageFormatRuntimeException(ce.getMessage());
+ }
+ catch (RuntimeException e)
+ {
+ throw new JMSRuntimeException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Set<String> getPropertyNames()
+ {
+ try
+ {
+ Set<SimpleString> simplePropNames = properties.getPropertyNames();
+ Set<String> propNames = new HashSet<String>(simplePropNames.size());
+
+ for (SimpleString str : simplePropNames)
+ {
+ propNames.add(str.toString());
+ }
+ return propNames;
+ }
+ catch (HornetQPropertyConversionException ce)
+ {
+ throw new MessageFormatRuntimeException(ce.getMessage());
+ }
+ catch (RuntimeException e)
+ {
+ throw new JMSRuntimeException(e.getMessage());
+ }
+ }
+
+ @Override
+ public JMSProducer setJMSCorrelationIDAsBytes(byte[] correlationID)
+ {
+ if (correlationID == null || correlationID.length == 0)
+ {
+ throw new JMSRuntimeException("Please specify a non-zero length byte[]");
+ }
+ jmsHeaderCorrelationIDAsBytes = Arrays.copyOf(correlationID, correlationID.length);
+ return this;
+ }
+
+ @Override
+ public byte[] getJMSCorrelationIDAsBytes()
+ {
+ return Arrays.copyOf(jmsHeaderCorrelationIDAsBytes, jmsHeaderCorrelationIDAsBytes.length);
+ }
+
+ @Override
+ public JMSProducer setJMSCorrelationID(String correlationID)
+ {
+ jmsHeaderCorrelationID = correlationID;
+ return this;
+ }
+
+ @Override
+ public String getJMSCorrelationID()
+ {
+ return jmsHeaderCorrelationID;
+ }
+
+ @Override
+ public JMSProducer setJMSType(String type)
+ {
+ jmsHeaderType = type;
+ return this;
+ }
+
+ @Override
+ public String getJMSType()
+ {
+ return jmsHeaderType;
+ }
+
+ @Override
+ public JMSProducer setJMSReplyTo(Destination replyTo)
+ {
+ jmsHeaderReplyTo = replyTo;
+ return this;
+ }
+
+ @Override
+ public Destination getJMSReplyTo()
+ {
+ return jmsHeaderReplyTo;
+ }
+
+ private void checkName(String name)
+ {
+ if (name == null)
+ {
+ throw HornetQJMSClientBundle.BUNDLE.nameCannotBeNull();
+ }
+ if (name.equals(""))
+ {
+ throw HornetQJMSClientBundle.BUNDLE.nameCannotBeEmpty();
+ }
+ }
+
+ final class CompletionListenerWrapper implements CompletionListener
+ {
+
+ private final CompletionListener wrapped;
+
+ public CompletionListenerWrapper(CompletionListener wrapped)
+ {
+ this.wrapped = wrapped;
+ }
+
+ @Override
+ public void onCompletion(Message message)
+ {
+ context.getThreadAwareContext().setCurrentThread(true);
+ try
+ {
+ wrapped.onCompletion(message);
+ }
+ finally
+ {
+ context.getThreadAwareContext().clearCurrentThread(true);
+ }
+ }
+
+ @Override
+ public void onException(Message message, Exception exception)
+ {
+ context.getThreadAwareContext().setCurrentThread(true);
+ try
+ {
+ wrapped.onException(message, exception);
+ }
+ finally
+ {
+ context.getThreadAwareContext().clearCurrentThread(true);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/23e8edd9/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQMapMessage.java
----------------------------------------------------------------------
diff --git a/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQMapMessage.java b/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQMapMessage.java
new file mode 100644
index 0000000..b6a8f04
--- /dev/null
+++ b/activemq6-jms-client/src/main/java/org/apache/activemq6/jms/client/HornetQMapMessage.java
@@ -0,0 +1,450 @@
+/*
+ * Copyright 2005-2014 Red Hat, Inc.
+ * Red Hat 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.activemq6.jms.client;
+
+import javax.jms.JMSException;
+import javax.jms.MapMessage;
+import javax.jms.MessageFormatException;
+import java.util.Collections;
+import java.util.Enumeration;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.activemq6.api.core.HornetQException;
+import org.apache.activemq6.api.core.HornetQPropertyConversionException;
+import org.apache.activemq6.api.core.Message;
+import org.apache.activemq6.api.core.SimpleString;
+import org.apache.activemq6.api.core.client.ClientMessage;
+import org.apache.activemq6.api.core.client.ClientSession;
+import org.apache.activemq6.utils.TypedProperties;
+
+
+import static org.apache.activemq6.reader.MapMessageUtil.writeBodyMap;
+import static org.apache.activemq6.reader.MapMessageUtil.readBodyMap;
+
+/**
+ * HornetQ implementation of a JMS MapMessage.
+ *
+ * @author Norbert Lataille (Norbert.Lataille@m4x.org)
+ * @author <a href="mailto:adrian@jboss.org">Adrian Brock</a>
+ * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a>
+ * @author <a href="mailto:ovidiu@feodorov.com">Ovidiu Feodorov</a>
+ * @author <a href="mailto:ataylor@redhat.com">Andy Taylor</a>
+ * @version $Revision: 3412 $
+ */
+public final class HornetQMapMessage extends HornetQMessage implements MapMessage
+{
+ // Constants -----------------------------------------------------
+
+ public static final byte TYPE = Message.MAP_TYPE;
+
+ // Attributes ----------------------------------------------------
+
+ private final TypedProperties map = new TypedProperties();
+
+ private boolean invalid;
+
+ // Static --------------------------------------------------------
+
+ // Constructors --------------------------------------------------
+
+ /*
+ * This constructor is used to construct messages prior to sending
+ */
+ protected HornetQMapMessage(final ClientSession session)
+ {
+ super(HornetQMapMessage.TYPE, session);
+
+ invalid = true;
+ }
+
+ /*
+ * This constructor is used during reading
+ */
+ protected HornetQMapMessage(final ClientMessage message, final ClientSession session)
+ {
+ super(message, session);
+
+ invalid = false;
+ }
+
+ public HornetQMapMessage()
+ {
+ invalid = false;
+ }
+
+ /**
+ * Constructor for a foreign MapMessage
+ *
+ * @param foreign
+ * @throws JMSException
+ */
+ public HornetQMapMessage(final MapMessage foreign, final ClientSession session) throws JMSException
+ {
+ super(foreign, HornetQMapMessage.TYPE, session);
+ Enumeration<?> names = foreign.getMapNames();
+ while (names.hasMoreElements())
+ {
+ String name = (String) names.nextElement();
+ Object obj = foreign.getObject(name);
+ setObject(name, obj);
+ }
+ }
+
+ // Public --------------------------------------------------------
+
+ @Override
+ public byte getType()
+ {
+ return HornetQMapMessage.TYPE;
+ }
+
+ // MapMessage implementation -------------------------------------
+
+ public void setBoolean(final String name, final boolean value) throws JMSException
+ {
+ checkName(name);
+ map.putBooleanProperty(new SimpleString(name), value);
+ invalid = true;
+ }
+
+ public void setByte(final String name, final byte value) throws JMSException
+ {
+ checkName(name);
+ map.putByteProperty(new SimpleString(name), value);
+ invalid = true;
+ }
+
+ public void setShort(final String name, final short value) throws JMSException
+ {
+ checkName(name);
+ map.putShortProperty(new SimpleString(name), value);
+ invalid = true;
+ }
+
+ public void setChar(final String name, final char value) throws JMSException
+ {
+ checkName(name);
+ map.putCharProperty(new SimpleString(name), value);
+ invalid = true;
+ }
+
+ public void setInt(final String name, final int value) throws JMSException
+ {
+ checkName(name);
+ map.putIntProperty(new SimpleString(name), value);
+ invalid = true;
+ }
+
+ public void setLong(final String name, final long value) throws JMSException
+ {
+ checkName(name);
+ map.putLongProperty(new SimpleString(name), value);
+ invalid = true;
+ }
+
+ public void setFloat(final String name, final float value) throws JMSException
+ {
+ checkName(name);
+ map.putFloatProperty(new SimpleString(name), value);
+ invalid = true;
+ }
+
+ public void setDouble(final String name, final double value) throws JMSException
+ {
+ checkName(name);
+ map.putDoubleProperty(new SimpleString(name), value);
+ invalid = true;
+ }
+
+ public void setString(final String name, final String value) throws JMSException
+ {
+ checkName(name);
+ map.putSimpleStringProperty(new SimpleString(name), value == null ? null : new SimpleString(value));
+ invalid = true;
+ }
+
+ public void setBytes(final String name, final byte[] value) throws JMSException
+ {
+ checkName(name);
+ map.putBytesProperty(new SimpleString(name), value);
+ invalid = true;
+ }
+
+ public void setBytes(final String name, final byte[] value, final int offset, final int length) throws JMSException
+ {
+ checkName(name);
+ 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);
+ invalid = true;
+ }
+
+ public void setObject(final String name, final Object value) throws JMSException
+ {
+ checkName(name);
+ try
+ {
+ TypedProperties.setObjectProperty(new SimpleString(name), value, map);
+ }
+ catch (HornetQPropertyConversionException e)
+ {
+ throw new MessageFormatException(e.getMessage());
+ }
+ invalid = true;
+ }
+
+ public boolean getBoolean(final String name) throws JMSException
+ {
+ try
+ {
+ return map.getBooleanProperty(new SimpleString(name));
+ }
+ catch (HornetQPropertyConversionException e)
+ {
+ throw new MessageFormatException(e.getMessage());
+ }
+ }
+
+ public byte getByte(final String name) throws JMSException
+ {
+ try
+ {
+ return map.getByteProperty(new SimpleString(name));
+ }
+ catch (HornetQPropertyConversionException e)
+ {
+ throw new MessageFormatException(e.getMessage());
+ }
+ }
+
+ public short getShort(final String name) throws JMSException
+ {
+ try
+ {
+ return map.getShortProperty(new SimpleString(name));
+ }
+ catch (HornetQPropertyConversionException e)
+ {
+ throw new MessageFormatException(e.getMessage());
+ }
+ }
+
+ public char getChar(final String name) throws JMSException
+ {
+ try
+ {
+ return map.getCharProperty(new SimpleString(name));
+ }
+ catch (HornetQPropertyConversionException e)
+ {
+ throw new MessageFormatException(e.getMessage());
+ }
+ }
+
+ public int getInt(final String name) throws JMSException
+ {
+ try
+ {
+ return map.getIntProperty(new SimpleString(name));
+ }
+ catch (HornetQPropertyConversionException e)
+ {
+ throw new MessageFormatException(e.getMessage());
+ }
+ }
+
+ public long getLong(final String name) throws JMSException
+ {
+ try
+ {
+ return map.getLongProperty(new SimpleString(name));
+ }
+ catch (HornetQPropertyConversionException e)
+ {
+ throw new MessageFormatException(e.getMessage());
+ }
+ }
+
+ public float getFloat(final String name) throws JMSException
+ {
+ try
+ {
+ return map.getFloatProperty(new SimpleString(name));
+ }
+ catch (HornetQPropertyConversionException e)
+ {
+ throw new MessageFormatException(e.getMessage());
+ }
+ }
+
+ public double getDouble(final String name) throws JMSException
+ {
+ try
+ {
+ return map.getDoubleProperty(new SimpleString(name));
+ }
+ catch (HornetQPropertyConversionException e)
+ {
+ throw new MessageFormatException(e.getMessage());
+ }
+ }
+
+ 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 (HornetQPropertyConversionException e)
+ {
+ throw new MessageFormatException(e.getMessage());
+ }
+ }
+
+ public byte[] getBytes(final String name) throws JMSException
+ {
+ try
+ {
+ return map.getBytesProperty(new SimpleString(name));
+ }
+ catch (HornetQPropertyConversionException e)
+ {
+ throw new MessageFormatException(e.getMessage());
+ }
+ }
+
+ 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;
+ }
+
+ public Enumeration getMapNames() throws JMSException
+ {
+ Set<SimpleString> simplePropNames = map.getPropertyNames();
+ Set<String> propNames = new HashSet<String>(simplePropNames.size());
+
+ for (SimpleString str : simplePropNames)
+ {
+ propNames.add(str.toString());
+ }
+
+ return Collections.enumeration(propNames);
+ }
+
+ public boolean itemExists(final String name) throws JMSException
+ {
+ return map.containsProperty(new SimpleString(name));
+ }
+
+
+ // HornetQRAMessage overrides ----------------------------------------
+
+ @Override
+ public void clearBody() throws JMSException
+ {
+ super.clearBody();
+
+ map.clear();
+
+ invalid = true;
+ }
+
+ @Override
+ public void doBeforeSend() throws Exception
+ {
+ if (invalid)
+ {
+ writeBodyMap(message, map);
+ invalid = false;
+ }
+
+ super.doBeforeSend();
+ }
+
+ @Override
+ public void doBeforeReceive() throws HornetQException
+ {
+ super.doBeforeReceive();
+
+ readBodyMap(message, map);
+ }
+
+ // Package protected ---------------------------------------------
+
+ // Protected -----------------------------------------------------
+
+ // Private -------------------------------------------------------
+
+ /**
+ * Check the name
+ *
+ * @param name the name
+ */
+ private void checkName(final String name) throws JMSException
+ {
+ checkWrite();
+
+ if (name == null)
+ {
+ throw HornetQJMSClientBundle.BUNDLE.nameCannotBeNull();
+ }
+ if (name.equals(""))
+ {
+ throw HornetQJMSClientBundle.BUNDLE.nameCannotBeEmpty();
+ }
+ }
+
+ @Override
+ protected boolean hasNoBody()
+ {
+ return map.isEmpty();
+ }
+
+ @Override
+ public boolean isBodyAssignableTo(@SuppressWarnings("rawtypes")
+ Class c)
+ {
+ if (hasNoBody())
+ {
+ return true;
+ }
+ return c.isAssignableFrom(java.util.Map.class);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ protected <T> T getBodyInternal(Class<T> c)
+ {
+ return (T) map.getMap();
+ }
+}