You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rh...@apache.org on 2008/08/07 21:25:14 UTC
svn commit: r683683 [2/3] - in /incubator/qpid/trunk/qpid/java:
broker/src/test/java/org/apache/qpid/server/util/
client/src/main/java/org/apache/qpid/client/
client/src/main/java/org/apache/qpid/client/message/
client/src/main/java/org/apache/qpid/fil...
Added: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_8.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_8.java?rev=683683&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_8.java (added)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_8.java Thu Aug 7 12:25:12 2008
@@ -0,0 +1,567 @@
+/*
+ *
+ * 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.qpid.client.message;
+
+import org.apache.commons.collections.map.ReferenceMap;
+import org.apache.mina.common.ByteBuffer;
+import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.client.CustomJMSXProperty;
+import org.apache.qpid.client.AMQDestination;
+import org.apache.qpid.client.AMQQueue;
+import org.apache.qpid.client.AMQTopic;
+import org.apache.qpid.client.AMQUndefinedDestination;
+import org.apache.qpid.client.JMSAMQException;
+import org.apache.qpid.framing.ContentHeaderProperties;
+import org.apache.qpid.framing.BasicContentHeaderProperties;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.url.BindingURL;
+import org.apache.qpid.url.AMQBindingURL;
+
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.MessageNotWriteableException;
+import java.util.Map;
+import java.util.Collections;
+import java.util.Enumeration;
+import java.util.UUID;
+import java.net.URISyntaxException;
+
+public class AMQMessageDelegate_0_8 implements AMQMessageDelegate
+{
+ private static final Map _destinationCache = Collections.synchronizedMap(new ReferenceMap());
+
+ public static final String JMS_TYPE = "x-jms-type";
+
+
+ private boolean _readableProperties = false;
+
+ private Destination _destination;
+ private JMSHeaderAdapter _headerAdapter;
+ private static final boolean STRICT_AMQP_COMPLIANCE =
+ Boolean.parseBoolean(System.getProperties().getProperty(AMQSession.STRICT_AMQP, AMQSession.STRICT_AMQP_DEFAULT));
+
+ private ContentHeaderProperties _contentHeaderProperties;
+ /** If the acknowledge mode is CLIENT_ACKNOWLEDGE the session is required */
+ private AMQSession _session;
+ private final long _deliveryTag;
+
+ protected AMQMessageDelegate_0_8()
+ {
+ this(new BasicContentHeaderProperties(), -1);
+ _readableProperties = false;
+ _headerAdapter = new JMSHeaderAdapter(((BasicContentHeaderProperties) _contentHeaderProperties).getHeaders());
+
+ }
+
+ protected AMQMessageDelegate_0_8(long deliveryTag, BasicContentHeaderProperties contentHeader, AMQShortString exchange,
+ AMQShortString routingKey)
+ {
+ this(contentHeader, deliveryTag);
+
+ Integer type = contentHeader.getHeaders().getInteger(CustomJMSXProperty.JMS_QPID_DESTTYPE.getShortStringName());
+
+ if(type == null)
+ {
+ type = AMQDestination.UNKNOWN_TYPE;
+ }
+
+ AMQDestination dest;
+
+ switch(type.intValue())
+ {
+ case AMQDestination.QUEUE_TYPE:
+ dest = new AMQQueue(exchange, routingKey, routingKey);
+ break;
+ case AMQDestination.TOPIC_TYPE:
+ dest = new AMQTopic(exchange, routingKey, null);
+ break;
+ default:
+ dest = new AMQUndefinedDestination(exchange, routingKey, null);
+ }
+
+
+
+ // Destination dest = AMQDestination.createDestination(url);
+ setJMSDestination(dest);
+
+
+
+ }
+
+ protected AMQMessageDelegate_0_8(BasicContentHeaderProperties properties, long deliveryTag)
+ {
+ _contentHeaderProperties = properties;
+ _deliveryTag = deliveryTag;
+ _readableProperties = (_contentHeaderProperties != null);
+ _headerAdapter = new JMSHeaderAdapter(((BasicContentHeaderProperties) _contentHeaderProperties).getHeaders());
+ }
+
+
+ public String getJMSMessageID() throws JMSException
+ {
+ return getContentHeaderProperties().getMessageIdAsString();
+ }
+
+ public void setJMSMessageID(String messageId) throws JMSException
+ {
+ getContentHeaderProperties().setMessageId(messageId);
+ }
+
+ public void setJMSMessageID(UUID messageId) throws JMSException
+ {
+ getContentHeaderProperties().setMessageId("ID:" + messageId);
+ }
+
+
+ public long getJMSTimestamp() throws JMSException
+ {
+ return getContentHeaderProperties().getTimestamp();
+ }
+
+ public void setJMSTimestamp(long timestamp) throws JMSException
+ {
+ getContentHeaderProperties().setTimestamp(timestamp);
+ }
+
+ public byte[] getJMSCorrelationIDAsBytes() throws JMSException
+ {
+ return getContentHeaderProperties().getCorrelationIdAsString().getBytes();
+ }
+
+ public void setJMSCorrelationIDAsBytes(byte[] bytes) throws JMSException
+ {
+ getContentHeaderProperties().setCorrelationId(new String(bytes));
+ }
+
+ public void setJMSCorrelationID(String correlationId) throws JMSException
+ {
+ getContentHeaderProperties().setCorrelationId(correlationId);
+ }
+
+ public String getJMSCorrelationID() throws JMSException
+ {
+ return getContentHeaderProperties().getCorrelationIdAsString();
+ }
+
+ public Destination getJMSReplyTo() throws JMSException
+ {
+ String replyToEncoding = getContentHeaderProperties().getReplyToAsString();
+ if (replyToEncoding == null)
+ {
+ return null;
+ }
+ else
+ {
+ Destination dest = (Destination) _destinationCache.get(replyToEncoding);
+ if (dest == null)
+ {
+ try
+ {
+ BindingURL binding = new AMQBindingURL(replyToEncoding);
+ dest = AMQDestination.createDestination(binding);
+ }
+ catch (URISyntaxException e)
+ {
+ throw new JMSAMQException("Illegal value in JMS_ReplyTo property: " + replyToEncoding, e);
+ }
+
+ _destinationCache.put(replyToEncoding, dest);
+ }
+
+ return dest;
+ }
+ }
+
+ public void setJMSReplyTo(Destination destination) throws JMSException
+ {
+ if (destination == null)
+ {
+ throw new IllegalArgumentException("Null destination not allowed");
+ }
+
+ if (!(destination instanceof AMQDestination))
+ {
+ throw new IllegalArgumentException(
+ "ReplyTo destination may only be an AMQDestination - passed argument was type " + destination.getClass());
+ }
+
+ final AMQDestination amqd = (AMQDestination) destination;
+
+ final AMQShortString encodedDestination = amqd.getEncodedName();
+ _destinationCache.put(encodedDestination, destination);
+ getContentHeaderProperties().setReplyTo(encodedDestination);
+ }
+
+ public Destination getJMSDestination() throws JMSException
+ {
+ return _destination;
+ }
+
+ public void setJMSDestination(Destination destination)
+ {
+ _destination = destination;
+ }
+
+ public void setContentType(String contentType)
+ {
+ getContentHeaderProperties().setContentType(contentType);
+ }
+
+ public String getContentType()
+ {
+ return getContentHeaderProperties().getContentTypeAsString();
+ }
+
+ public void setEncoding(String encoding)
+ {
+ getContentHeaderProperties().setEncoding(encoding);
+ }
+
+ public String getEncoding()
+ {
+ return getContentHeaderProperties().getEncodingAsString();
+ }
+
+ public String getReplyToString()
+ {
+ return getContentHeaderProperties().getReplyToAsString();
+ }
+
+ public int getJMSDeliveryMode() throws JMSException
+ {
+ return getContentHeaderProperties().getDeliveryMode();
+ }
+
+ public void setJMSDeliveryMode(int i) throws JMSException
+ {
+ getContentHeaderProperties().setDeliveryMode((byte) i);
+ }
+
+ public BasicContentHeaderProperties getContentHeaderProperties()
+ {
+ return (BasicContentHeaderProperties) _contentHeaderProperties;
+ }
+
+
+ public String getJMSType() throws JMSException
+ {
+ return getContentHeaderProperties().getTypeAsString();
+ }
+
+ public void setJMSType(String string) throws JMSException
+ {
+ getContentHeaderProperties().setType(string);
+ }
+
+ public long getJMSExpiration() throws JMSException
+ {
+ return getContentHeaderProperties().getExpiration();
+ }
+
+ public void setJMSExpiration(long l) throws JMSException
+ {
+ getContentHeaderProperties().setExpiration(l);
+ }
+
+
+
+ public boolean propertyExists(String propertyName) throws JMSException
+ {
+ return getJmsHeaders().propertyExists(propertyName);
+ }
+
+ public boolean getBooleanProperty(String propertyName) throws JMSException
+ {
+ if (STRICT_AMQP_COMPLIANCE)
+ {
+ throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
+ }
+
+ return getJmsHeaders().getBoolean(propertyName);
+ }
+
+ public byte getByteProperty(String propertyName) throws JMSException
+ {
+ if (STRICT_AMQP_COMPLIANCE)
+ {
+ throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
+ }
+
+ return getJmsHeaders().getByte(propertyName);
+ }
+
+ public short getShortProperty(String propertyName) throws JMSException
+ {
+ if (STRICT_AMQP_COMPLIANCE)
+ {
+ throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
+ }
+
+ return getJmsHeaders().getShort(propertyName);
+ }
+
+ public int getIntProperty(String propertyName) throws JMSException
+ {
+ if (STRICT_AMQP_COMPLIANCE)
+ {
+ throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
+ }
+
+ return getJmsHeaders().getInteger(propertyName);
+ }
+
+ public long getLongProperty(String propertyName) throws JMSException
+ {
+ if (STRICT_AMQP_COMPLIANCE)
+ {
+ throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
+ }
+
+ return getJmsHeaders().getLong(propertyName);
+ }
+
+ public float getFloatProperty(String propertyName) throws JMSException
+ {
+ if (STRICT_AMQP_COMPLIANCE)
+ {
+ throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
+ }
+
+ return getJmsHeaders().getFloat(propertyName);
+ }
+
+ public double getDoubleProperty(String propertyName) throws JMSException
+ {
+ if (STRICT_AMQP_COMPLIANCE)
+ {
+ throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
+ }
+
+ return getJmsHeaders().getDouble(propertyName);
+ }
+
+ public String getStringProperty(String propertyName) throws JMSException
+ {
+ //NOTE: if the JMSX Property is a non AMQP property then we must check _strictAMQP and throw as below.
+ if (propertyName.equals(CustomJMSXProperty.JMSXUserID.toString()))
+ {
+ return ((BasicContentHeaderProperties) _contentHeaderProperties).getUserIdAsString();
+ }
+ else
+ {
+ if (STRICT_AMQP_COMPLIANCE)
+ {
+ throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
+ }
+
+ return getJmsHeaders().getString(propertyName);
+ }
+ }
+
+ public Object getObjectProperty(String propertyName) throws JMSException
+ {
+ return getJmsHeaders().getObject(propertyName);
+ }
+
+ public Enumeration getPropertyNames() throws JMSException
+ {
+ return getJmsHeaders().getPropertyNames();
+ }
+
+ public void setBooleanProperty(String propertyName, boolean b) throws JMSException
+ {
+ if (STRICT_AMQP_COMPLIANCE)
+ {
+ throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
+ }
+
+ checkWritableProperties();
+ getJmsHeaders().setBoolean(propertyName, b);
+ }
+
+ public void setByteProperty(String propertyName, byte b) throws JMSException
+ {
+ if (STRICT_AMQP_COMPLIANCE)
+ {
+ throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
+ }
+
+ checkWritableProperties();
+ getJmsHeaders().setByte(propertyName, new Byte(b));
+ }
+
+ public void setShortProperty(String propertyName, short i) throws JMSException
+ {
+ if (STRICT_AMQP_COMPLIANCE)
+ {
+ throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
+ }
+
+ checkWritableProperties();
+ getJmsHeaders().setShort(propertyName, new Short(i));
+ }
+
+ public void setIntProperty(String propertyName, int i) throws JMSException
+ {
+ checkWritableProperties();
+ getJmsHeaders().setInteger(propertyName, new Integer(i));
+ }
+
+ public void setLongProperty(String propertyName, long l) throws JMSException
+ {
+ if (STRICT_AMQP_COMPLIANCE)
+ {
+ throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
+ }
+
+ checkWritableProperties();
+ getJmsHeaders().setLong(propertyName, new Long(l));
+ }
+
+ public void setFloatProperty(String propertyName, float f) throws JMSException
+ {
+ if (STRICT_AMQP_COMPLIANCE)
+ {
+ throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
+ }
+
+ checkWritableProperties();
+ getJmsHeaders().setFloat(propertyName, new Float(f));
+ }
+
+ public void setDoubleProperty(String propertyName, double v) throws JMSException
+ {
+ if (STRICT_AMQP_COMPLIANCE)
+ {
+ throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
+ }
+
+ checkWritableProperties();
+ getJmsHeaders().setDouble(propertyName, new Double(v));
+ }
+
+ public void setStringProperty(String propertyName, String value) throws JMSException
+ {
+ checkWritableProperties();
+ getJmsHeaders().setString(propertyName, value);
+ }
+
+ public void setObjectProperty(String propertyName, Object object) throws JMSException
+ {
+ checkWritableProperties();
+ getJmsHeaders().setObject(propertyName, object);
+ }
+
+ public void removeProperty(String propertyName) throws JMSException
+ {
+ getJmsHeaders().remove(propertyName);
+ }
+
+
+ private JMSHeaderAdapter getJmsHeaders()
+ {
+ return _headerAdapter;
+ }
+
+ protected void checkWritableProperties() throws MessageNotWriteableException
+ {
+ if (_readableProperties)
+ {
+ throw new MessageNotWriteableException("You need to call clearProperties() to make the message writable");
+ }
+ _contentHeaderProperties.updated();
+ }
+
+
+ public int getJMSPriority() throws JMSException
+ {
+ return getContentHeaderProperties().getPriority();
+ }
+
+ public void setJMSPriority(int i) throws JMSException
+ {
+ getContentHeaderProperties().setPriority((byte) i);
+ }
+
+ public void clearProperties() throws JMSException
+ {
+ getJmsHeaders().clear();
+
+ _readableProperties = false;
+ }
+
+
+ public void acknowledgeThis() throws JMSException
+ {
+ // the JMS 1.1 spec says in section 3.6 that calls to acknowledge are ignored when client acknowledge
+ // is not specified. In our case, we only set the session field where client acknowledge mode is specified.
+ if (_session != null)
+ {
+ if (_session.getAMQConnection().isClosed())
+ {
+ throw new javax.jms.IllegalStateException("Connection is already closed");
+ }
+
+ // we set multiple to true here since acknowledgement implies acknowledge of all previous messages
+ // received on the session
+ _session.acknowledgeMessage(_deliveryTag, true);
+ }
+ }
+
+ public void acknowledge() throws JMSException
+ {
+ if (_session != null)
+ {
+ _session.acknowledge();
+ }
+ }
+
+
+ /**
+ * The session is set when CLIENT_ACKNOWLEDGE mode is used so that the CHANNEL ACK can be sent when the user calls
+ * acknowledge()
+ *
+ * @param s the AMQ session that delivered this message
+ */
+ public void setAMQSession(AMQSession s)
+ {
+ _session = s;
+ }
+
+ public AMQSession getAMQSession()
+ {
+ return _session;
+ }
+
+ /**
+ * Get the AMQ message number assigned to this message
+ *
+ * @return the message number
+ */
+ public long getDeliveryTag()
+ {
+ return _deliveryTag;
+ }
+
+
+}
Propchange: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_8.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractBytesMessage.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractBytesMessage.java?rev=683683&r1=683682&r2=683683&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractBytesMessage.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractBytesMessage.java Thu Aug 7 12:25:12 2008
@@ -31,7 +31,6 @@
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.BasicContentHeaderProperties;
-import org.apache.qpid.framing.ContentHeaderBody;
/**
* @author Apache Software Foundation
@@ -44,21 +43,21 @@
*/
private static final int DEFAULT_BUFFER_INITIAL_SIZE = 1024;
- AbstractBytesMessage()
+ AbstractBytesMessage(AMQMessageDelegateFactory delegateFactory)
{
- this(null);
+ this(delegateFactory, null);
}
/**
* Construct a bytes message with existing data.
*
+ * @param delegateFactory
* @param data the data that comprises this message. If data is null, you get a 1024 byte buffer that is
- * set to auto expand
*/
- AbstractBytesMessage(ByteBuffer data)
+ AbstractBytesMessage(AMQMessageDelegateFactory delegateFactory, ByteBuffer data)
{
- super(data); // this instanties a content header
- getContentHeaderProperties().setContentType(getMimeTypeAsShortString());
+ super(delegateFactory, data); // this instanties a content header
+ setContentType(getMimeType());
if (_data == null)
{
@@ -72,13 +71,12 @@
_data.setAutoExpand(true);
}
- AbstractBytesMessage(long messageNbr, BasicContentHeaderProperties contentHeader, AMQShortString exchange,
- AMQShortString routingKey, ByteBuffer data) throws AMQException
- {
- // TODO: this casting is ugly. Need to review whole ContentHeaderBody idea
- super(messageNbr, contentHeader, exchange, routingKey, data);
- getContentHeaderProperties().setContentType(getMimeTypeAsShortString());
- }
+ AbstractBytesMessage(AMQMessageDelegate delegate, ByteBuffer data) throws AMQException
+ {
+ super(delegate, data);
+ setContentType(getMimeType());
+ }
+
public void clearBodyImpl() throws JMSException
{
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractBytesTypedMessage.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractBytesTypedMessage.java?rev=683683&r1=683682&r2=683683&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractBytesTypedMessage.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractBytesTypedMessage.java Thu Aug 7 12:25:12 2008
@@ -33,7 +33,6 @@
import org.apache.mina.common.ByteBuffer;
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.ContentHeaderBody;
import org.apache.qpid.framing.BasicContentHeaderProperties;
/**
@@ -70,27 +69,28 @@
*/
private int _byteArrayRemaining = -1;
- AbstractBytesTypedMessage()
+ AbstractBytesTypedMessage(AMQMessageDelegateFactory delegateFactory)
{
- this(null);
+
+ this(delegateFactory, null);
}
/**
* Construct a stream message with existing data.
*
+ * @param delegateFactory
* @param data the data that comprises this message. If data is null, you get a 1024 byte buffer that is
- * set to auto expand
*/
- AbstractBytesTypedMessage(ByteBuffer data)
+ AbstractBytesTypedMessage(AMQMessageDelegateFactory delegateFactory, ByteBuffer data)
{
- super(data); // this instanties a content header
- }
+ super(delegateFactory, data); // this instanties a content header
+ }
- AbstractBytesTypedMessage(long messageNbr, BasicContentHeaderProperties contentHeader, AMQShortString exchange,
- AMQShortString routingKey, ByteBuffer data) throws AMQException
+ AbstractBytesTypedMessage(AMQMessageDelegate delegate, ByteBuffer data) throws AMQException
{
- super(messageNbr, contentHeader, exchange, routingKey, data);
+
+ super(delegate, data);
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java?rev=683683&r1=683682&r2=683683&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java Thu Aug 7 12:25:12 2008
@@ -21,10 +21,7 @@
package org.apache.qpid.client.message;
import java.io.IOException;
-import java.net.URISyntaxException;
-import java.util.Collections;
import java.util.Enumeration;
-import java.util.Map;
import java.util.UUID;
import javax.jms.Destination;
@@ -32,122 +29,54 @@
import javax.jms.MessageNotReadableException;
import javax.jms.MessageNotWriteableException;
-import org.apache.commons.collections.map.ReferenceMap;
import org.apache.mina.common.ByteBuffer;
import org.apache.qpid.AMQException;
-import org.apache.qpid.client.AMQDestination;
-import org.apache.qpid.client.AMQQueue;
import org.apache.qpid.client.AMQSession;
-import org.apache.qpid.client.AMQTopic;
-import org.apache.qpid.client.AMQUndefinedDestination;
-import org.apache.qpid.client.BasicMessageConsumer;
-import org.apache.qpid.client.CustomJMSXProperty;
-import org.apache.qpid.client.JMSAMQException;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.BasicContentHeaderProperties;
-import org.apache.qpid.framing.FieldTable;
-import org.apache.qpid.url.AMQBindingURL;
-import org.apache.qpid.url.BindingURL;
-public abstract class AbstractJMSMessage extends AMQMessage implements org.apache.qpid.jms.Message
+public abstract class AbstractJMSMessage implements org.apache.qpid.jms.Message
{
- private static final Map _destinationCache = Collections.synchronizedMap(new ReferenceMap());
- public static final String JMS_TYPE = "x-jms-type";
- protected boolean _redelivered;
protected ByteBuffer _data;
- private boolean _readableProperties = false;
protected boolean _readableMessage = false;
protected boolean _changedData = true;
- private Destination _destination;
- private JMSHeaderAdapter _headerAdapter;
- private static final boolean STRICT_AMQP_COMPLIANCE =
- Boolean.parseBoolean(System.getProperties().getProperty(AMQSession.STRICT_AMQP, AMQSession.STRICT_AMQP_DEFAULT));
+
/**
* This is 0_10 specific
*/
private org.apache.qpid.api.Message _010message = null;
+ /** If the acknowledge mode is CLIENT_ACKNOWLEDGE the session is required */
- public void set010Message(org.apache.qpid.api.Message m )
- {
- _010message = m;
- }
- public void dataChanged()
- {
- if (_010message != null)
- {
- _010message.clearData();
- try
- {
- if (_data != null)
- {
- _010message.appendData(_data.buf().slice());
- }
- else
- {
- _010message.appendData(java.nio.ByteBuffer.allocate(0));
- }
- }
- catch (IOException e)
- {
- throw new RuntimeException(e);
- }
- }
- }
- /**
- * End 010 specific
- */
- public org.apache.qpid.api.Message get010Message()
- {
- return _010message;
- }
+ private AMQMessageDelegate _delegate;
+ private boolean _redelivered;
- protected AbstractJMSMessage(ByteBuffer data)
+ protected AbstractJMSMessage(AMQMessageDelegateFactory delegateFactory, ByteBuffer data)
{
- super(new BasicContentHeaderProperties());
+ _delegate = delegateFactory.createDelegate();
_data = data;
if (_data != null)
{
_data.acquire();
}
- _readableProperties = false;
+
_readableMessage = (data != null);
_changedData = (data == null);
- _headerAdapter = new JMSHeaderAdapter(((BasicContentHeaderProperties) _contentHeaderProperties).getHeaders());
}
- protected AbstractJMSMessage(long deliveryTag, BasicContentHeaderProperties contentHeader, AMQShortString exchange,
- AMQShortString routingKey, ByteBuffer data) throws AMQException
+ protected AbstractJMSMessage(AMQMessageDelegate delegate, ByteBuffer data) throws AMQException
{
- this(contentHeader, deliveryTag);
-
- Integer type = contentHeader.getHeaders().getInteger(CustomJMSXProperty.JMS_QPID_DESTTYPE.getShortStringName());
-
- AMQDestination dest;
- if (AMQDestination.QUEUE_TYPE.equals(type))
- {
- dest = new AMQQueue(exchange, routingKey, routingKey);
- }
- else if (AMQDestination.TOPIC_TYPE.equals(type))
- {
- dest = new AMQTopic(exchange, routingKey, null);
- }
- else
- {
- dest = new AMQUndefinedDestination(exchange, routingKey, null);
- }
- // Destination dest = AMQDestination.createDestination(url);
- setJMSDestination(dest);
+ _delegate = delegate;
_data = data;
if (_data != null)
@@ -159,126 +88,82 @@
}
- protected AbstractJMSMessage(BasicContentHeaderProperties contentHeader, long deliveryTag)
+ public String getJMSMessageID() throws JMSException
{
- super(contentHeader, deliveryTag);
- _readableProperties = (_contentHeaderProperties != null);
- _headerAdapter = new JMSHeaderAdapter(((BasicContentHeaderProperties) _contentHeaderProperties).getHeaders());
+ return _delegate.getJMSMessageID();
}
- public String getJMSMessageID() throws JMSException
+ public void setJMSMessageID(String messageId) throws JMSException
{
- return getContentHeaderProperties().getMessageIdAsString();
+ _delegate.setJMSMessageID(messageId);
}
- public void setJMSMessageID(String messageId) throws JMSException
+ public void setJMSMessageID(UUID messageId) throws JMSException
{
- getContentHeaderProperties().setMessageId(messageId);
+ _delegate.setJMSMessageID(messageId);
}
+
public long getJMSTimestamp() throws JMSException
{
- return getContentHeaderProperties().getTimestamp();
+ return _delegate.getJMSTimestamp();
}
public void setJMSTimestamp(long timestamp) throws JMSException
{
- getContentHeaderProperties().setTimestamp(timestamp);
+ _delegate.setJMSTimestamp(timestamp);
}
public byte[] getJMSCorrelationIDAsBytes() throws JMSException
{
- return getContentHeaderProperties().getCorrelationIdAsString().getBytes();
+ return _delegate.getJMSCorrelationIDAsBytes();
}
public void setJMSCorrelationIDAsBytes(byte[] bytes) throws JMSException
{
- getContentHeaderProperties().setCorrelationId(new String(bytes));
+ _delegate.setJMSCorrelationIDAsBytes(bytes);
}
public void setJMSCorrelationID(String correlationId) throws JMSException
{
- getContentHeaderProperties().setCorrelationId(correlationId);
+ _delegate.setJMSCorrelationID(correlationId);
}
public String getJMSCorrelationID() throws JMSException
{
- return getContentHeaderProperties().getCorrelationIdAsString();
+ return _delegate.getJMSCorrelationID();
}
public Destination getJMSReplyTo() throws JMSException
{
- String replyToEncoding = getContentHeaderProperties().getReplyToAsString();
- if (replyToEncoding == null)
- {
- return null;
- }
- else
- {
- Destination dest = (Destination) _destinationCache.get(replyToEncoding);
- if (dest == null)
- {
- try
- {
- BindingURL binding = new AMQBindingURL(replyToEncoding);
- dest = AMQDestination.createDestination(binding);
- }
- catch (URISyntaxException e)
- {
- throw new JMSAMQException("Illegal value in JMS_ReplyTo property: " + replyToEncoding, e);
- }
-
- _destinationCache.put(replyToEncoding, dest);
- }
-
- return dest;
- }
+ return _delegate.getJMSReplyTo();
}
public void setJMSReplyTo(Destination destination) throws JMSException
{
- if (destination == null)
- {
- throw new IllegalArgumentException("Null destination not allowed");
- }
-
- if (!(destination instanceof AMQDestination))
- {
- throw new IllegalArgumentException(
- "ReplyTo destination may only be an AMQDestination - passed argument was type " + destination.getClass());
- }
-
- final AMQDestination amqd = (AMQDestination) destination;
-
- final AMQShortString encodedDestination = amqd.getEncodedName();
- _destinationCache.put(encodedDestination, destination);
- getContentHeaderProperties().setReplyTo(encodedDestination);
+ _delegate.setJMSReplyTo(destination);
}
public Destination getJMSDestination() throws JMSException
{
- return _destination;
+ return _delegate.getJMSDestination();
}
public void setJMSDestination(Destination destination)
{
- _destination = destination;
+ _delegate.setJMSDestination(destination);
}
public int getJMSDeliveryMode() throws JMSException
{
- return getContentHeaderProperties().getDeliveryMode();
+ return _delegate.getJMSDeliveryMode();
}
public void setJMSDeliveryMode(int i) throws JMSException
{
- getContentHeaderProperties().setDeliveryMode((byte) i);
+ _delegate.setJMSDeliveryMode(i);
}
- public BasicContentHeaderProperties getContentHeaderProperties()
- {
- return (BasicContentHeaderProperties) _contentHeaderProperties;
- }
public boolean getJMSRedelivered() throws JMSException
{
@@ -290,318 +175,180 @@
_redelivered = b;
}
+
public String getJMSType() throws JMSException
{
- return getContentHeaderProperties().getTypeAsString();
+ return _delegate.getJMSType();
}
public void setJMSType(String string) throws JMSException
{
- getContentHeaderProperties().setType(string);
+ _delegate.setJMSType(string);
}
public long getJMSExpiration() throws JMSException
{
- return getContentHeaderProperties().getExpiration();
+ return _delegate.getJMSExpiration();
}
public void setJMSExpiration(long l) throws JMSException
{
- getContentHeaderProperties().setExpiration(l);
+ _delegate.setJMSExpiration(l);
}
public int getJMSPriority() throws JMSException
{
- return getContentHeaderProperties().getPriority();
+ return _delegate.getJMSPriority();
}
public void setJMSPriority(int i) throws JMSException
{
- getContentHeaderProperties().setPriority((byte) i);
+ _delegate.setJMSPriority(i);
}
- public void clearProperties() throws JMSException
- {
- getJmsHeaders().clear();
-
- _readableProperties = false;
- }
-
- public void clearBody() throws JMSException
- {
- clearBodyImpl();
- _readableMessage = false;
- }
-
- public boolean propertyExists(AMQShortString propertyName) throws JMSException
- {
- return getJmsHeaders().propertyExists(propertyName);
- }
public boolean propertyExists(String propertyName) throws JMSException
{
- return getJmsHeaders().propertyExists(propertyName);
+ return _delegate.propertyExists(propertyName);
}
- public boolean getBooleanProperty(AMQShortString propertyName) throws JMSException
+ public boolean getBooleanProperty(final String s)
+ throws JMSException
{
- if (STRICT_AMQP_COMPLIANCE)
- {
- throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
- }
-
- return getJmsHeaders().getBoolean(propertyName);
+ return _delegate.getBooleanProperty(s);
}
- public boolean getBooleanProperty(String propertyName) throws JMSException
+ public byte getByteProperty(final String s)
+ throws JMSException
{
- if (STRICT_AMQP_COMPLIANCE)
- {
- throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
- }
-
- return getJmsHeaders().getBoolean(propertyName);
+ return _delegate.getByteProperty(s);
}
- public byte getByteProperty(String propertyName) throws JMSException
+ public short getShortProperty(final String s)
+ throws JMSException
{
- if (STRICT_AMQP_COMPLIANCE)
- {
- throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
- }
-
- return getJmsHeaders().getByte(propertyName);
+ return _delegate.getShortProperty(s);
}
- public byte[] getBytesProperty(AMQShortString propertyName) throws JMSException
+ public int getIntProperty(final String s)
+ throws JMSException
{
- if (STRICT_AMQP_COMPLIANCE)
- {
- throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
- }
-
- return getJmsHeaders().getBytes(propertyName);
+ return _delegate.getIntProperty(s);
}
- public short getShortProperty(String propertyName) throws JMSException
+ public long getLongProperty(final String s)
+ throws JMSException
{
- if (STRICT_AMQP_COMPLIANCE)
- {
- throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
- }
-
- return getJmsHeaders().getShort(propertyName);
- }
-
- public int getIntProperty(String propertyName) throws JMSException
- {
- if (STRICT_AMQP_COMPLIANCE)
- {
- throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
- }
-
- return getJmsHeaders().getInteger(propertyName);
+ return _delegate.getLongProperty(s);
}
- public long getLongProperty(String propertyName) throws JMSException
+ public float getFloatProperty(final String s)
+ throws JMSException
{
- if (STRICT_AMQP_COMPLIANCE)
- {
- throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
- }
-
- return getJmsHeaders().getLong(propertyName);
+ return _delegate.getFloatProperty(s);
}
- public float getFloatProperty(String propertyName) throws JMSException
+ public double getDoubleProperty(final String s)
+ throws JMSException
{
- if (STRICT_AMQP_COMPLIANCE)
- {
- throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
- }
-
- return getJmsHeaders().getFloat(propertyName);
+ return _delegate.getDoubleProperty(s);
}
- public double getDoubleProperty(String propertyName) throws JMSException
+ public String getStringProperty(final String s)
+ throws JMSException
{
- if (STRICT_AMQP_COMPLIANCE)
- {
- throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
- }
-
- return getJmsHeaders().getDouble(propertyName);
+ return _delegate.getStringProperty(s);
}
- public String getStringProperty(String propertyName) throws JMSException
+ public Object getObjectProperty(final String s)
+ throws JMSException
{
- //NOTE: if the JMSX Property is a non AMQP property then we must check _strictAMQP and throw as below.
- if (propertyName.equals(CustomJMSXProperty.JMSXUserID.toString()))
- {
- return ((BasicContentHeaderProperties) _contentHeaderProperties).getUserIdAsString();
- }
- else
- {
- if (STRICT_AMQP_COMPLIANCE)
- {
- throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
- }
-
- return getJmsHeaders().getString(propertyName);
- }
+ return _delegate.getObjectProperty(s);
}
- public Object getObjectProperty(String propertyName) throws JMSException
+ public Enumeration getPropertyNames()
+ throws JMSException
{
- return getJmsHeaders().getObject(propertyName);
+ return _delegate.getPropertyNames();
}
- public Enumeration getPropertyNames() throws JMSException
+ public void setBooleanProperty(final String s, final boolean b)
+ throws JMSException
{
- return getJmsHeaders().getPropertyNames();
+ _delegate.setBooleanProperty(s, b);
}
- public void setBooleanProperty(AMQShortString propertyName, boolean b) throws JMSException
+ public void setByteProperty(final String s, final byte b)
+ throws JMSException
{
- if (STRICT_AMQP_COMPLIANCE)
- {
- throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
- }
-
- checkWritableProperties();
- getJmsHeaders().setBoolean(propertyName, b);
+ _delegate.setByteProperty(s, b);
}
- public void setBooleanProperty(String propertyName, boolean b) throws JMSException
+ public void setShortProperty(final String s, final short i)
+ throws JMSException
{
- if (STRICT_AMQP_COMPLIANCE)
- {
- throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
- }
-
- checkWritableProperties();
- getJmsHeaders().setBoolean(propertyName, b);
+ _delegate.setShortProperty(s, i);
}
- public void setByteProperty(String propertyName, byte b) throws JMSException
+ public void setIntProperty(final String s, final int i)
+ throws JMSException
{
- if (STRICT_AMQP_COMPLIANCE)
- {
- throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
- }
-
- checkWritableProperties();
- getJmsHeaders().setByte(propertyName, new Byte(b));
+ _delegate.setIntProperty(s, i);
}
- public void setBytesProperty(AMQShortString propertyName, byte[] bytes) throws JMSException
+ public void setLongProperty(final String s, final long l)
+ throws JMSException
{
- if (STRICT_AMQP_COMPLIANCE)
- {
- throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
- }
-
- checkWritableProperties();
- getJmsHeaders().setBytes(propertyName, bytes);
+ _delegate.setLongProperty(s, l);
}
- public void setShortProperty(String propertyName, short i) throws JMSException
+ public void setFloatProperty(final String s, final float v)
+ throws JMSException
{
- if (STRICT_AMQP_COMPLIANCE)
- {
- throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
- }
-
- checkWritableProperties();
- getJmsHeaders().setShort(propertyName, new Short(i));
+ _delegate.setFloatProperty(s, v);
}
- public void setIntProperty(String propertyName, int i) throws JMSException
+ public void setDoubleProperty(final String s, final double v)
+ throws JMSException
{
- checkWritableProperties();
- JMSHeaderAdapter.checkPropertyName(propertyName);
- super.setIntProperty(new AMQShortString(propertyName), new Integer(i));
+ _delegate.setDoubleProperty(s, v);
}
- public void setLongProperty(String propertyName, long l) throws JMSException
+ public void setStringProperty(final String s, final String s1)
+ throws JMSException
{
- if (STRICT_AMQP_COMPLIANCE)
- {
- throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
- }
-
- checkWritableProperties();
- getJmsHeaders().setLong(propertyName, new Long(l));
+ _delegate.setStringProperty(s, s1);
}
- public void setFloatProperty(String propertyName, float f) throws JMSException
+ public void setObjectProperty(final String s, final Object o)
+ throws JMSException
{
- if (STRICT_AMQP_COMPLIANCE)
- {
- throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
- }
-
- checkWritableProperties();
- getJmsHeaders().setFloat(propertyName, new Float(f));
+ _delegate.setObjectProperty(s, o);
}
- public void setDoubleProperty(String propertyName, double v) throws JMSException
- {
- if (STRICT_AMQP_COMPLIANCE)
- {
- throw new UnsupportedOperationException("JMS Proprerties not supported in AMQP");
- }
- checkWritableProperties();
- getJmsHeaders().setDouble(propertyName, new Double(v));
- }
- public void setStringProperty(String propertyName, String value) throws JMSException
+ public void clearProperties() throws JMSException
{
- checkWritableProperties();
- JMSHeaderAdapter.checkPropertyName(propertyName);
- super.setLongStringProperty(new AMQShortString(propertyName), value);
+ _delegate.clearProperties();
}
- public void setObjectProperty(String propertyName, Object object) throws JMSException
+ public void clearBody() throws JMSException
{
- checkWritableProperties();
- getJmsHeaders().setObject(propertyName, object);
- }
+ clearBodyImpl();
+ _readableMessage = false;
- protected void removeProperty(AMQShortString propertyName) throws JMSException
- {
- getJmsHeaders().remove(propertyName);
}
- protected void removeProperty(String propertyName) throws JMSException
- {
- getJmsHeaders().remove(propertyName);
- }
public void acknowledgeThis() throws JMSException
{
- // the JMS 1.1 spec says in section 3.6 that calls to acknowledge are ignored when client acknowledge
- // is not specified. In our case, we only set the session field where client acknowledge mode is specified.
- if (_session != null)
- {
- if (_session.getAMQConnection().isClosed())
- {
- throw new javax.jms.IllegalStateException("Connection is already closed");
- }
-
- // we set multiple to true here since acknowledgement implies acknowledge of all previous messages
- // received on the session
- _session.acknowledgeMessage(_deliveryTag, true);
- }
+ _delegate.acknowledgeThis();
}
public void acknowledge() throws JMSException
{
- if (_session != null)
- {
- _session.acknowledge();
- }
+ _delegate.acknowledge();
}
/**
@@ -617,12 +364,9 @@
*/
public abstract String toBodyString() throws JMSException;
- public String getMimeType()
- {
- return getMimeTypeAsShortString().toString();
- }
+ protected abstract String getMimeType();
+
- public abstract AMQShortString getMimeTypeAsShortString();
public String toString()
{
@@ -640,16 +384,23 @@
buf.append("\nJMS Destination: ").append(getJMSDestination());
buf.append("\nJMS Type: ").append(getJMSType());
buf.append("\nJMS MessageID: ").append(getJMSMessageID());
- buf.append("\nAMQ message number: ").append(_deliveryTag);
+ buf.append("\nAMQ message number: ").append(getDeliveryTag());
buf.append("\nProperties:");
- if (getJmsHeaders().isEmpty())
+ final Enumeration propertyNames = getPropertyNames();
+ if (!propertyNames.hasMoreElements())
{
buf.append("<NONE>");
}
else
{
- buf.append('\n').append(getJmsHeaders().getHeaders());
+ buf.append('\n');
+ while(propertyNames.hasMoreElements())
+ {
+ String propertyName = (String) propertyNames.nextElement();
+ buf.append(propertyName).append(":\t").append(getObjectProperty(propertyName));
+ }
+
}
return buf.toString();
@@ -660,14 +411,10 @@
}
}
- public void setUnderlyingMessagePropertiesMap(FieldTable messageProperties)
- {
- getContentHeaderProperties().setHeaders(messageProperties);
- }
- public JMSHeaderAdapter getJmsHeaders()
+ public AMQMessageDelegate getDelegate()
{
- return _headerAdapter;
+ return _delegate;
}
public ByteBuffer getData()
@@ -698,25 +445,6 @@
}
}
- protected void checkWritableProperties() throws MessageNotWriteableException
- {
- if (_readableProperties)
- {
- throw new MessageNotWriteableException("You need to call clearProperties() to make the message writable");
- }
- _contentHeaderProperties.updated();
- }
-
- public boolean isReadable()
- {
- return _readableMessage;
- }
-
- public boolean isWritable()
- {
- return !_readableMessage;
- }
-
public void reset()
{
if (!_changedData)
@@ -731,6 +459,47 @@
}
}
+ public void set010Message(org.apache.qpid.api.Message m )
+ {
+ _010message = m;
+ }
+
+ public void dataChanged()
+ {
+ if (_010message != null)
+ {
+ _010message.clearData();
+ try
+ {
+ if (_data != null)
+ {
+ _010message.appendData(_data.buf().slice());
+ }
+ else
+ {
+ _010message.appendData(java.nio.ByteBuffer.allocate(0));
+ }
+ }
+ catch (IOException e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ /**
+ * End 010 specific
+ */
+
+ public org.apache.qpid.api.Message get010Message()
+ {
+ return _010message;
+ }
+
+
+
+
+
public int getContentLength()
{
if(_data != null)
@@ -748,4 +517,66 @@
_changedData = false;
}
+ /**
+ * The session is set when CLIENT_ACKNOWLEDGE mode is used so that the CHANNEL ACK can be sent when the user calls
+ * acknowledge()
+ *
+ * @param s the AMQ session that delivered this message
+ */
+ public void setAMQSession(AMQSession s)
+ {
+ _delegate.setAMQSession(s);
+ }
+
+ public AMQSession getAMQSession()
+ {
+ return _delegate.getAMQSession();
+ }
+
+ /**
+ * Get the AMQ message number assigned to this message
+ *
+ * @return the message number
+ */
+ public long getDeliveryTag()
+ {
+ return _delegate.getDeliveryTag();
+ }
+
+ /** Invoked prior to sending the message. Allows the message to be modified if necessary before sending. */
+ public void prepareForSending() throws JMSException
+ {
+ }
+
+
+ public void setContentType(String contentType)
+ {
+ _delegate.setContentType(contentType);
+ }
+
+ public String getContentType()
+ {
+ return _delegate.getContentType();
+ }
+
+ public void setEncoding(String encoding)
+ {
+ _delegate.setEncoding(encoding);
+ }
+
+ public String getEncoding()
+ {
+ return _delegate.getEncoding();
+ }
+
+ public String getReplyToString()
+ {
+ return _delegate.getReplyToString();
+ }
+
+ protected void removeProperty(final String propertyName) throws JMSException
+ {
+ _delegate.removeProperty(propertyName);
+ }
+
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java?rev=683683&r1=683682&r2=683683&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java Thu Aug 7 12:25:12 2008
@@ -38,17 +38,11 @@
import java.util.Iterator;
import java.util.List;
-import java.util.Map;
-import java.util.UUID;
public abstract class AbstractJMSMessageFactory implements MessageFactory
{
private static final Logger _logger = LoggerFactory.getLogger(AbstractJMSMessageFactory.class);
- protected abstract AbstractJMSMessage createMessage(long messageNbr, ByteBuffer data, AMQShortString exchange,
- AMQShortString routingKey,
- BasicContentHeaderProperties contentHeader) throws AMQException;
-
protected AbstractJMSMessage create08MessageWithBody(long messageNbr, ContentHeaderBody contentHeader,
AMQShortString exchange, AMQShortString routingKey,
List bodies) throws AMQException
@@ -105,13 +99,18 @@
.remaining());
}
- return createMessage(messageNbr, data, exchange, routingKey,
- (BasicContentHeaderProperties) contentHeader.properties);
+ AMQMessageDelegate delegate = new AMQMessageDelegate_0_8(messageNbr,
+ (BasicContentHeaderProperties) contentHeader.properties,
+ exchange, routingKey);
+
+ return createMessage(delegate, data);
}
+ protected abstract AbstractJMSMessage createMessage(AMQMessageDelegate delegate, ByteBuffer data) throws AMQException;
+
+
protected AbstractJMSMessage create010MessageWithBody(long messageNbr, Struct[] contentHeader,
- AMQShortString exchange, AMQShortString routingKey,
- List bodies, String replyToURL) throws AMQException
+ List bodies) throws AMQException
{
ByteBuffer data;
final boolean debug = _logger.isDebugEnabled();
@@ -131,40 +130,13 @@
_logger.debug("Creating message from buffer with position=" + data.position() + " and remaining=" + data
.remaining());
}
- BasicContentHeaderProperties props = new BasicContentHeaderProperties();
// set the properties of this message
MessageProperties mprop = (MessageProperties) contentHeader[0];
DeliveryProperties devprop = (DeliveryProperties) contentHeader[1];
- props.setContentType(mprop.getContentType());
- props.setCorrelationId(asString(mprop.getCorrelationId()));
- String encoding = mprop.getContentEncoding();
- if (encoding != null && !encoding.equals(""))
- {
- props.setEncoding(encoding);
- }
- if (devprop.hasDeliveryMode())
- {
- props.setDeliveryMode((byte) devprop.getDeliveryMode().getValue());
- }
- props.setExpiration(devprop.getExpiration());
- UUID mid = mprop.getMessageId();
- props.setMessageId(mid == null ? null : "ID:" + mid.toString());
- if (devprop.hasPriority())
- {
- props.setPriority((byte) devprop.getPriority().getValue());
- }
- props.setReplyTo(replyToURL);
- props.setTimestamp(devprop.getTimestamp());
- String type = null;
- Map<String,Object> map = mprop.getApplicationHeaders();
- if (map != null)
- {
- type = (String) map.get(AbstractJMSMessage.JMS_TYPE);
- }
- props.setType(type);
- props.setUserId(asString(mprop.getUserId()));
- props.setHeaders(FiledTableSupport.convertToFieldTable(mprop.getApplicationHeaders()));
- AbstractJMSMessage message = createMessage(messageNbr, data, exchange, routingKey, props);
+
+ AMQMessageDelegate delegate = new AMQMessageDelegate_0_10(mprop, devprop, messageNbr);
+
+ AbstractJMSMessage message = createMessage(delegate, data);
return message;
}
@@ -192,12 +164,11 @@
}
public AbstractJMSMessage createMessage(long messageNbr, boolean redelivered, Struct[] contentHeader,
- AMQShortString exchange, AMQShortString routingKey, List bodies,
- String replyToURL)
+ List bodies)
throws JMSException, AMQException
{
final AbstractJMSMessage msg =
- create010MessageWithBody(messageNbr, contentHeader, exchange, routingKey, bodies, replyToURL);
+ create010MessageWithBody(messageNbr, contentHeader, bodies);
msg.setJMSRedelivered(redelivered);
msg.receivedFromServer();
return msg;
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSBytesMessage.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSBytesMessage.java?rev=683683&r1=683682&r2=683683&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSBytesMessage.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSBytesMessage.java Thu Aug 7 12:25:12 2008
@@ -33,46 +33,47 @@
import org.apache.mina.common.ByteBuffer;
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.ContentHeaderBody;
import org.apache.qpid.framing.BasicContentHeaderProperties;
public class JMSBytesMessage extends AbstractBytesMessage implements BytesMessage
{
public static final String MIME_TYPE = "application/octet-stream";
- private static final AMQShortString MIME_TYPE_SHORT_STRING = new AMQShortString(MIME_TYPE);
- public JMSBytesMessage()
+
+ public JMSBytesMessage(AMQMessageDelegateFactory delegateFactory)
{
- this(null);
+ this(delegateFactory,null);
+
}
/**
* Construct a bytes message with existing data.
*
+ * @param delegateFactory
* @param data the data that comprises this message. If data is null, you get a 1024 byte buffer that is
- * set to auto expand
*/
- JMSBytesMessage(ByteBuffer data)
+ JMSBytesMessage(AMQMessageDelegateFactory delegateFactory, ByteBuffer data)
{
- super(data); // this instanties a content header
+
+ super(delegateFactory, data); // this instanties a content header
}
- JMSBytesMessage(long messageNbr, BasicContentHeaderProperties contentHeader, AMQShortString exchange,
- AMQShortString routingKey, ByteBuffer data) throws AMQException
+ JMSBytesMessage(AMQMessageDelegate delegate, ByteBuffer data) throws AMQException
{
- super(messageNbr, contentHeader, exchange, routingKey, data);
+ super(delegate, data);
}
+
public void reset()
{
super.reset();
_readableMessage = true;
}
- public AMQShortString getMimeTypeAsShortString()
+ protected String getMimeType()
{
- return MIME_TYPE_SHORT_STRING;
+ return MIME_TYPE;
}
public long getBodyLength() throws JMSException
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSBytesMessageFactory.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSBytesMessageFactory.java?rev=683683&r1=683682&r2=683683&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSBytesMessageFactory.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSBytesMessageFactory.java Thu Aug 7 12:25:12 2008
@@ -25,21 +25,18 @@
import org.apache.mina.common.ByteBuffer;
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.ContentHeaderBody;
import org.apache.qpid.framing.BasicContentHeaderProperties;
public class JMSBytesMessageFactory extends AbstractJMSMessageFactory
{
- protected AbstractJMSMessage createMessage(long deliveryTag, ByteBuffer data,
- AMQShortString exchange, AMQShortString routingKey,
- BasicContentHeaderProperties contentHeader) throws AMQException
+ protected AbstractJMSMessage createMessage(AMQMessageDelegate delegate, ByteBuffer data) throws AMQException
{
- return new JMSBytesMessage(deliveryTag, contentHeader, exchange, routingKey, data);
+ return new JMSBytesMessage(delegate, data);
}
- public AbstractJMSMessage createMessage() throws JMSException
+ public AbstractJMSMessage createMessage(AMQMessageDelegateFactory delegateFactory) throws JMSException
{
- return new JMSBytesMessage();
+ return new JMSBytesMessage(delegateFactory);
}
// 0_10 specific
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSHeaderAdapter.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSHeaderAdapter.java?rev=683683&r1=683682&r2=683683&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSHeaderAdapter.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSHeaderAdapter.java Thu Aug 7 12:25:12 2008
@@ -467,7 +467,7 @@
return getPropertyNames();
}
- protected static void checkPropertyName(CharSequence propertyName)
+ protected void checkPropertyName(CharSequence propertyName)
{
if (propertyName == null)
{
@@ -481,7 +481,7 @@
checkIdentiferFormat(propertyName);
}
- protected static void checkIdentiferFormat(CharSequence propertyName)
+ protected void checkIdentiferFormat(CharSequence propertyName)
{
// JMS requirements 3.5.1 Property Names
// Identifiers:
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSMapMessage.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSMapMessage.java?rev=683683&r1=683682&r2=683683&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSMapMessage.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSMapMessage.java Thu Aug 7 12:25:12 2008
@@ -24,7 +24,6 @@
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.ContentHeaderBody;
import org.apache.qpid.framing.BasicContentHeaderProperties;
import org.slf4j.Logger;
@@ -44,18 +43,19 @@
private static final Logger _logger = LoggerFactory.getLogger(JMSMapMessage.class);
public static final String MIME_TYPE = "jms/map-message";
- private static final AMQShortString MIME_TYPE_SHORT_STRING = new AMQShortString(MIME_TYPE);
+
private Map<String, Object> _map = new HashMap<String, Object>();
- public JMSMapMessage() throws JMSException
+ public JMSMapMessage(AMQMessageDelegateFactory delegateFactory) throws JMSException
{
- this(null);
+ this(delegateFactory, null);
}
- JMSMapMessage(ByteBuffer data) throws JMSException
+ JMSMapMessage(AMQMessageDelegateFactory delegateFactory, ByteBuffer data) throws JMSException
{
- super(data); // this instantiates a content header
+
+ super(delegateFactory, data); // this instantiates a content header
if(data != null)
{
populateMapFromData();
@@ -63,10 +63,10 @@
}
- JMSMapMessage(long messageNbr, BasicContentHeaderProperties contentHeader, AMQShortString exchange, AMQShortString routingKey,
- ByteBuffer data) throws AMQException
+ JMSMapMessage(AMQMessageDelegate delegate, ByteBuffer data) throws AMQException
{
- super(messageNbr, contentHeader, exchange, routingKey, data);
+
+ super(delegate, data);
try
{
populateMapFromData();
@@ -79,14 +79,15 @@
}
+
public String toBodyString() throws JMSException
{
return _map == null ? "" : _map.toString();
}
- public AMQShortString getMimeTypeAsShortString()
+ protected String getMimeType()
{
- return MIME_TYPE_SHORT_STRING;
+ return MIME_TYPE;
}
public ByteBuffer getData()
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSMapMessageFactory.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSMapMessageFactory.java?rev=683683&r1=683682&r2=683683&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSMapMessageFactory.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSMapMessageFactory.java Thu Aug 7 12:25:12 2008
@@ -25,21 +25,18 @@
import org.apache.mina.common.ByteBuffer;
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.ContentHeaderBody;
import org.apache.qpid.framing.BasicContentHeaderProperties;
public class JMSMapMessageFactory extends AbstractJMSMessageFactory
{
- public AbstractJMSMessage createMessage() throws JMSException
+ public AbstractJMSMessage createMessage(AMQMessageDelegateFactory delegateFactory) throws JMSException
{
- return new JMSMapMessage();
+ return new JMSMapMessage(delegateFactory);
}
- protected AbstractJMSMessage createMessage(long deliveryTag, ByteBuffer data,
- AMQShortString exchange, AMQShortString routingKey,
- BasicContentHeaderProperties contentHeader) throws AMQException
+ protected AbstractJMSMessage createMessage(AMQMessageDelegate delegate, ByteBuffer data) throws AMQException
{
- return new JMSMapMessage(deliveryTag, contentHeader, exchange, routingKey, data);
+ return new JMSMapMessage(delegate, data);
}
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSObjectMessage.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSObjectMessage.java?rev=683683&r1=683682&r2=683683&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSObjectMessage.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSObjectMessage.java Thu Aug 7 12:25:12 2008
@@ -37,52 +37,54 @@
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.BasicContentHeaderProperties;
-import org.apache.qpid.framing.ContentHeaderBody;
public class JMSObjectMessage extends AbstractJMSMessage implements ObjectMessage
{
public static final String MIME_TYPE = "application/java-object-stream";
- private static final AMQShortString MIME_TYPE_SHORT_STRING = new AMQShortString(MIME_TYPE);
+
private static final int DEFAULT_BUFFER_SIZE = 1024;
/**
* Creates empty, writable message for use by producers
+ * @param delegateFactory
*/
- public JMSObjectMessage()
+ public JMSObjectMessage(AMQMessageDelegateFactory delegateFactory)
{
- this(null);
+ this(delegateFactory, null);
}
- private JMSObjectMessage(ByteBuffer data)
+ private JMSObjectMessage(AMQMessageDelegateFactory delegateFactory, ByteBuffer data)
{
- super(data);
+ super(delegateFactory, data);
if (data == null)
{
_data = ByteBuffer.allocate(DEFAULT_BUFFER_SIZE);
_data.setAutoExpand(true);
}
- getContentHeaderProperties().setContentType(MIME_TYPE_SHORT_STRING);
+ setContentType(getMimeType());
}
/**
* Creates read only message for delivery to consumers
*/
- JMSObjectMessage(long messageNbr, BasicContentHeaderProperties contentHeader, AMQShortString exchange, AMQShortString routingKey,
- ByteBuffer data) throws AMQException
- {
- super(messageNbr, contentHeader, exchange, routingKey, data);
- }
+
+ JMSObjectMessage(AMQMessageDelegate delegate, ByteBuffer data) throws AMQException
+ {
+ super(delegate, data);
+ }
+
public void clearBodyImpl() throws JMSException
{
if (_data != null)
{
_data.release();
+ _data = null;
}
- _data = null;
+
}
@@ -91,9 +93,9 @@
return toString(_data);
}
- public AMQShortString getMimeTypeAsShortString()
+ public String getMimeType()
{
- return MIME_TYPE_SHORT_STRING;
+ return MIME_TYPE;
}
public void setObject(Serializable serializable) throws JMSException
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSObjectMessageFactory.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSObjectMessageFactory.java?rev=683683&r1=683682&r2=683683&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSObjectMessageFactory.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSObjectMessageFactory.java Thu Aug 7 12:25:12 2008
@@ -25,20 +25,17 @@
import org.apache.mina.common.ByteBuffer;
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.ContentHeaderBody;
import org.apache.qpid.framing.BasicContentHeaderProperties;
public class JMSObjectMessageFactory extends AbstractJMSMessageFactory
{
- protected AbstractJMSMessage createMessage(long deliveryTag, ByteBuffer data,
- AMQShortString exchange, AMQShortString routingKey,
- BasicContentHeaderProperties contentHeader) throws AMQException
+ protected AbstractJMSMessage createMessage(AMQMessageDelegate delegate, ByteBuffer data) throws AMQException
{
- return new JMSObjectMessage(deliveryTag, contentHeader, exchange, routingKey, data);
+ return new JMSObjectMessage(delegate, data);
}
- public AbstractJMSMessage createMessage() throws JMSException
+ public AbstractJMSMessage createMessage(AMQMessageDelegateFactory delegateFactory) throws JMSException
{
- return new JMSObjectMessage();
+ return new JMSObjectMessage(delegateFactory);
}
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSStreamMessage.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSStreamMessage.java?rev=683683&r1=683682&r2=683683&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSStreamMessage.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSStreamMessage.java Thu Aug 7 12:25:12 2008
@@ -26,7 +26,6 @@
import org.apache.mina.common.ByteBuffer;
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.ContentHeaderBody;
import org.apache.qpid.framing.BasicContentHeaderProperties;
/**
@@ -35,7 +34,7 @@
public class JMSStreamMessage extends AbstractBytesTypedMessage implements StreamMessage
{
public static final String MIME_TYPE="jms/stream-message";
- private static final AMQShortString MIME_TYPE_SHORT_STRING = new AMQShortString(MIME_TYPE);
+
/**
@@ -44,38 +43,40 @@
*/
private int _byteArrayRemaining = -1;
- public JMSStreamMessage()
+ public JMSStreamMessage(AMQMessageDelegateFactory delegateFactory)
{
- this(null);
+ this(delegateFactory,null);
+
}
/**
* Construct a stream message with existing data.
*
+ * @param delegateFactory
* @param data the data that comprises this message. If data is null, you get a 1024 byte buffer that is
- * set to auto expand
*/
- JMSStreamMessage(ByteBuffer data)
+ JMSStreamMessage(AMQMessageDelegateFactory delegateFactory, ByteBuffer data)
{
- super(data); // this instanties a content header
- }
+ super(delegateFactory, data); // this instanties a content header
+ }
- JMSStreamMessage(long messageNbr, BasicContentHeaderProperties contentHeader, AMQShortString exchange,
- AMQShortString routingKey, ByteBuffer data) throws AMQException
+ JMSStreamMessage(AMQMessageDelegate delegate, ByteBuffer data) throws AMQException
{
- super(messageNbr, contentHeader, exchange, routingKey, data);
+
+ super(delegate, data);
}
+
public void reset()
{
super.reset();
_readableMessage = true;
}
- public AMQShortString getMimeTypeAsShortString()
+ protected String getMimeType()
{
- return MIME_TYPE_SHORT_STRING;
+ return MIME_TYPE;
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSStreamMessageFactory.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSStreamMessageFactory.java?rev=683683&r1=683682&r2=683683&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSStreamMessageFactory.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSStreamMessageFactory.java Thu Aug 7 12:25:12 2008
@@ -25,19 +25,16 @@
import org.apache.mina.common.ByteBuffer;
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.ContentHeaderBody;
import org.apache.qpid.framing.BasicContentHeaderProperties;
public class JMSStreamMessageFactory extends AbstractJMSMessageFactory
{
- protected AbstractJMSMessage createMessage(long deliveryTag, ByteBuffer data,
- AMQShortString exchange, AMQShortString routingKey,
- BasicContentHeaderProperties contentHeader) throws AMQException
+ protected AbstractJMSMessage createMessage(AMQMessageDelegate delegate, ByteBuffer data) throws AMQException
{
- return new JMSStreamMessage(deliveryTag, contentHeader, exchange, routingKey, data);
+ return new JMSStreamMessage(delegate, data);
}
- public AbstractJMSMessage createMessage() throws JMSException
+ public AbstractJMSMessage createMessage(AMQMessageDelegateFactory delegateFactory) throws JMSException
{
- return new JMSStreamMessage();
+ return new JMSStreamMessage(delegateFactory);
}
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSTextMessage.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSTextMessage.java?rev=683683&r1=683682&r2=683683&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSTextMessage.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSTextMessage.java Thu Aug 7 12:25:12 2008
@@ -36,46 +36,44 @@
public class JMSTextMessage extends AbstractJMSMessage implements javax.jms.TextMessage
{
private static final String MIME_TYPE = "text/plain";
- private static final AMQShortString MIME_TYPE_SHORT_STRING = new AMQShortString(MIME_TYPE);
-
private String _decodedValue;
/**
* This constant represents the name of a property that is set when the message payload is null.
*/
- private static final AMQShortString PAYLOAD_NULL_PROPERTY = CustomJMSXProperty.JMS_AMQP_NULL.getShortStringName();
+ private static final String PAYLOAD_NULL_PROPERTY = CustomJMSXProperty.JMS_AMQP_NULL.toString();
private static final Charset DEFAULT_CHARSET = Charset.forName("UTF-8");
- public JMSTextMessage() throws JMSException
+ public JMSTextMessage(AMQMessageDelegateFactory delegateFactory) throws JMSException
{
- this(null, null);
+ this(delegateFactory, null, null);
}
- JMSTextMessage(ByteBuffer data, String encoding) throws JMSException
+ JMSTextMessage(AMQMessageDelegateFactory delegateFactory, ByteBuffer data, String encoding) throws JMSException
{
- super(data); // this instantiates a content header
- getContentHeaderProperties().setContentType(MIME_TYPE_SHORT_STRING);
- getContentHeaderProperties().setEncoding(encoding);
+ super(delegateFactory, data); // this instantiates a content header
+ setContentType(getMimeType());
+ setEncoding(encoding);
}
- JMSTextMessage(long deliveryTag, BasicContentHeaderProperties contentHeader, AMQShortString exchange,
- AMQShortString routingKey, ByteBuffer data)
+ JMSTextMessage(AMQMessageDelegate delegate, ByteBuffer data)
throws AMQException
{
- super(deliveryTag, contentHeader, exchange, routingKey, data);
- contentHeader.setContentType(MIME_TYPE_SHORT_STRING);
+ super(delegate, data);
+ setContentType(getMimeType());
_data = data;
}
- JMSTextMessage(ByteBuffer data) throws JMSException
+
+ JMSTextMessage(AMQMessageDelegateFactory delegateFactory, ByteBuffer data) throws JMSException
{
- this(data, null);
+ this(delegateFactory, data, null);
}
- JMSTextMessage(String text) throws JMSException
+ JMSTextMessage(AMQMessageDelegateFactory delegateFactory, String text) throws JMSException
{
- super((ByteBuffer) null);
+ super(delegateFactory, (ByteBuffer) null);
setText(text);
}
@@ -84,8 +82,9 @@
if (_data != null)
{
_data.release();
+ _data = null;
}
- _data = null;
+
_decodedValue = null;
}
@@ -94,14 +93,9 @@
return getText();
}
- public void setData(ByteBuffer data)
+ protected String getMimeType()
{
- _data = data;
- }
-
- public AMQShortString getMimeTypeAsShortString()
- {
- return MIME_TYPE_SHORT_STRING;
+ return MIME_TYPE;
}
public void setText(String text) throws JMSException
@@ -113,7 +107,7 @@
{
if (text != null)
{
- final String encoding = getContentHeaderProperties().getEncodingAsString();
+ final String encoding = getEncoding();
if (encoding == null || encoding.equalsIgnoreCase("UTF-8"))
{
_data = ByteBuffer.wrap(Strings.toUTF8(text));
@@ -154,11 +148,11 @@
{
return null;
}
- if (getContentHeaderProperties().getEncodingAsString() != null)
+ if (getEncoding() != null)
{
try
{
- _decodedValue = _data.getString(Charset.forName(getContentHeaderProperties().getEncodingAsString()).newDecoder());
+ _decodedValue = _data.getString(Charset.forName(getEncoding()).newDecoder());
}
catch (CharacterCodingException e)
{
@@ -197,4 +191,6 @@
removeProperty(PAYLOAD_NULL_PROPERTY);
}
}
+
+
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSTextMessageFactory.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSTextMessageFactory.java?rev=683683&r1=683682&r2=683683&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSTextMessageFactory.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSTextMessageFactory.java Thu Aug 7 12:25:12 2008
@@ -26,21 +26,17 @@
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.BasicContentHeaderProperties;
-import org.apache.qpid.framing.ContentHeaderBody;
public class JMSTextMessageFactory extends AbstractJMSMessageFactory
{
- public AbstractJMSMessage createMessage() throws JMSException
+ public AbstractJMSMessage createMessage(AMQMessageDelegateFactory delegateFactory) throws JMSException
{
- return new JMSTextMessage();
+ return new JMSTextMessage(delegateFactory);
}
- protected AbstractJMSMessage createMessage(long deliveryTag, ByteBuffer data,
- AMQShortString exchange, AMQShortString routingKey,
- BasicContentHeaderProperties contentHeader) throws AMQException
+ protected AbstractJMSMessage createMessage(AMQMessageDelegate delegate, ByteBuffer data) throws AMQException
{
- return new JMSTextMessage(deliveryTag, contentHeader,
- exchange, routingKey, data);
+ return new JMSTextMessage(delegate, data);
}
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageConverter.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageConverter.java?rev=683683&r1=683682&r2=683683&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageConverter.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageConverter.java Thu Aug 7 12:25:12 2008
@@ -22,15 +22,9 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.qpid.client.AMQSession;
-import javax.jms.BytesMessage;
-import javax.jms.JMSException;
-import javax.jms.MapMessage;
-import javax.jms.Message;
-import javax.jms.MessageEOFException;
-import javax.jms.ObjectMessage;
-import javax.jms.StreamMessage;
-import javax.jms.TextMessage;
+import javax.jms.*;
import java.util.Enumeration;
@@ -52,12 +46,11 @@
_newMessage = message;
}
- public MessageConverter(BytesMessage message) throws JMSException
+ public MessageConverter(AMQSession session, BytesMessage bytesMessage) throws JMSException
{
- BytesMessage bytesMessage = (BytesMessage) message;
bytesMessage.reset();
- JMSBytesMessage nativeMsg = new JMSBytesMessage();
+ JMSBytesMessage nativeMsg = (JMSBytesMessage) session.createBytesMessage();
byte[] buf = new byte[1024];
@@ -69,12 +62,12 @@
}
_newMessage = nativeMsg;
- setMessageProperties(message);
+ setMessageProperties(bytesMessage);
}
- public MessageConverter(MapMessage message) throws JMSException
+ public MessageConverter(AMQSession session, MapMessage message) throws JMSException
{
- MapMessage nativeMessage = new JMSMapMessage();
+ MapMessage nativeMessage = session.createMapMessage();
Enumeration mapNames = message.getMapNames();
while (mapNames.hasMoreElements())
@@ -87,21 +80,21 @@
setMessageProperties(message);
}
- public MessageConverter(ObjectMessage message) throws JMSException
+ public MessageConverter(AMQSession session, ObjectMessage origMessage) throws JMSException
{
- ObjectMessage origMessage = (ObjectMessage) message;
- ObjectMessage nativeMessage = new JMSObjectMessage();
+
+ ObjectMessage nativeMessage = session.createObjectMessage();
nativeMessage.setObject(origMessage.getObject());
_newMessage = (AbstractJMSMessage) nativeMessage;
- setMessageProperties(message);
+ setMessageProperties(origMessage);
}
- public MessageConverter(TextMessage message) throws JMSException
+ public MessageConverter(AMQSession session, TextMessage message) throws JMSException
{
- TextMessage nativeMessage = new JMSTextMessage();
+ TextMessage nativeMessage = session.createTextMessage();
nativeMessage.setText(message.getText());
@@ -109,9 +102,9 @@
setMessageProperties(message);
}
- public MessageConverter(StreamMessage message) throws JMSException
+ public MessageConverter(AMQSession session, StreamMessage message) throws JMSException
{
- StreamMessage nativeMessage = new JMSStreamMessage();
+ StreamMessage nativeMessage = session.createStreamMessage();
try
{
@@ -130,11 +123,11 @@
setMessageProperties(message);
}
- public MessageConverter(Message message) throws JMSException
+ public MessageConverter(AMQSession session, Message message) throws JMSException
{
// Send a message with just properties.
// Throwing away content
- BytesMessage nativeMessage = new JMSBytesMessage();
+ Message nativeMessage = session.createMessage();
_newMessage = (AbstractJMSMessage) nativeMessage;
setMessageProperties(message);
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageFactory.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageFactory.java?rev=683683&r1=683682&r2=683683&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageFactory.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageFactory.java Thu Aug 7 12:25:12 2008
@@ -39,10 +39,9 @@
throws JMSException, AMQException;
AbstractJMSMessage createMessage(long deliveryTag, boolean redelivered,
- Struct[] contentHeader,
- AMQShortString exchange, AMQShortString routingKey,
- List bodies, String replyToURL)
+ Struct[] contentHeader,
+ List bodies)
throws JMSException, AMQException;
- AbstractJMSMessage createMessage() throws JMSException;
+ AbstractJMSMessage createMessage(AMQMessageDelegateFactory delegateFactory) throws JMSException;
}