Author: rajith
Date: Fri Jun 15 17:22:05 2012
New Revision: 1350707
URL: http://svn.apache.org/viewvc?rev=1350707&view=rev
Log:
QPID-4027 Added an AMQP 0-10 message factory implementation. The
StringMessage is not really 0-10 specific. When we add AMQP 1.0 support,
perhaps we could have a AbstractMessageFactory which contains the common
stuff.
Added GenericMessageAdapter to be used as a base for all kinds of
Message adapters.
Added:
qpid/branches/address-refactor2/qpid/java/client-api/src/main/java/org/apache/qpid/messaging/util/GenericMessageAdapter.java
- copied, changed from r1350706,
qpid/branches/address-refactor2/qpid/java/client-api/src/main/java/org/apache/qpid/messaging/util/ReadOnlyMessageAdapter.java
qpid/branches/address-refactor2/qpid/java/client-api/src/main/java/org/apache/qpid/messaging/util/MessageFactory_AMQP_0_10.java
Modified:
qpid/branches/address-refactor2/qpid/java/client-api/src/main/java/org/apache/qpid/messaging/util/ReadOnlyMessageAdapter.java
Copied:
qpid/branches/address-refactor2/qpid/java/client-api/src/main/java/org/apache/qpid/messaging/util/GenericMessageAdapter.java
(from r1350706,
qpid/branches/address-refactor2/qpid/java/client-api/src/main/java/org/apache/qpid/messaging/util/ReadOnlyMessageAdapter.java)
URL:
http://svn.apache.org/viewvc/qpid/branches/address-refactor2/qpid/java/client-api/src/main/java/org/apache/qpid/messaging/util/GenericMessageAdapter.java?p2=qpid/branches/address-refactor2/qpid/java/client-api/src/main/java/org/apache/qpid/messaging/util/GenericMessageAdapter.java&p1=qpid/branches/address-refactor2/qpid/java/client-api/src/main/java/org/apache/qpid/messaging/util/ReadOnlyMessageAdapter.java&r1=1350706&r2=1350707&rev=1350707&view=diff
==============================================================================
---
qpid/branches/address-refactor2/qpid/java/client-api/src/main/java/org/apache/qpid/messaging/util/ReadOnlyMessageAdapter.java
(original)
+++
qpid/branches/address-refactor2/qpid/java/client-api/src/main/java/org/apache/qpid/messaging/util/GenericMessageAdapter.java
Fri Jun 15 17:22:05 2012
@@ -17,6 +17,7 @@
*/
package org.apache.qpid.messaging.util;
+import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.Map;
@@ -25,25 +26,24 @@ import org.apache.qpid.messaging.Message
import org.apache.qpid.messaging.MessagingException;
/**
- * Ensures the message is read only by blocking the delegates
- * setter methods.
+ * A generic message adapter that simply delegates
+ * all method calls to the underlying message delegate.
+ * This is not intended to be used by itself,
+ * rather as a base class for other adapters. For example,
+ * @see ReadOnlyMessageAdapter
+ * @see StringMessage_AMQP_0_10
+ * @see MapMessage_AMQP_0_10
*/
-public class ReadOnlyMessageAdapter implements Message
+public class GenericMessageAdapter implements Message
{
private Message _delegate;
- ReadOnlyMessageAdapter(Message delegate)
+ GenericMessageAdapter(Message delegate)
{
_delegate = delegate;
}
@Override
- public Object getContent() throws MessagingException
- {
- return _delegate.getContent();
- }
-
- @Override
public String getMessageId() throws MessagingException
{
return _delegate.getMessageId();
@@ -52,7 +52,7 @@ public class ReadOnlyMessageAdapter impl
@Override
public void setMessageId(String messageId) throws MessagingException
{
- throwMessageNotWritableException();
+ _delegate.setMessageId(messageId);
}
@Override
@@ -64,7 +64,7 @@ public class ReadOnlyMessageAdapter impl
@Override
public void setSubject(String subject) throws MessagingException
{
- throwMessageNotWritableException();
+ _delegate.setSubject(subject);
}
@Override
@@ -76,7 +76,7 @@ public class ReadOnlyMessageAdapter impl
@Override
public void setContentType(String contentType) throws MessagingException
{
- throwMessageNotWritableException();
+ _delegate.setContentType(contentType);
}
@Override
@@ -88,7 +88,7 @@ public class ReadOnlyMessageAdapter impl
@Override
public void setCorrelationId(String correlationId) throws
MessagingException
{
- throwMessageNotWritableException();
+ _delegate.setCorrelationId(correlationId);
}
@Override
@@ -100,7 +100,7 @@ public class ReadOnlyMessageAdapter impl
@Override
public void setReplyTo(String replyTo) throws MessagingException
{
- throwMessageNotWritableException();
+ _delegate.setReplyTo(replyTo);
}
@Override
@@ -112,7 +112,7 @@ public class ReadOnlyMessageAdapter impl
@Override
public void setUserId(String userId) throws MessagingException
{
- throwMessageNotWritableException();
+ _delegate.setUserId(userId);
}
@Override
@@ -124,7 +124,7 @@ public class ReadOnlyMessageAdapter impl
@Override
public void setDurable(boolean durable) throws MessagingException
{
- throwMessageNotWritableException();
+ _delegate.setDurable(durable);
}
@Override
@@ -136,7 +136,7 @@ public class ReadOnlyMessageAdapter impl
@Override
public void setRedelivered(boolean redelivered) throws MessagingException
{
- throwMessageNotWritableException();
+ _delegate.setRedelivered(redelivered);
}
@Override
@@ -148,7 +148,7 @@ public class ReadOnlyMessageAdapter impl
@Override
public void setPriority(int priority) throws MessagingException
{
- throwMessageNotWritableException();
+ _delegate.setPriority(priority);
}
@Override
@@ -160,7 +160,7 @@ public class ReadOnlyMessageAdapter impl
@Override
public void setTtl(long ttl) throws MessagingException
{
- throwMessageNotWritableException();
+ _delegate.setTtl(ttl);
}
@Override
@@ -172,24 +172,24 @@ public class ReadOnlyMessageAdapter impl
@Override
public void setTimestamp(long timestamp) throws MessagingException
{
- throwMessageNotWritableException();
+ _delegate.setTimestamp(timestamp);
}
@Override
public Map<String, Object> getProperties() throws MessagingException
{
- return Collections.unmodifiableMap(_delegate.getProperties());
+ return _delegate.getProperties();
}
@Override
public void setProperty(String key, Object value) throws MessagingException
{
- throwMessageNotWritableException();
+ _delegate.setProperty(key, value);
}
- private void throwMessageNotWritableException() throws
MessageNotWritableException
+ @Override
+ public ByteBuffer getContent()
{
- throw new MessageNotWritableException("Message is read-only");
+ return _delegate.getContent();
}
-
}
Added:
qpid/branches/address-refactor2/qpid/java/client-api/src/main/java/org/apache/qpid/messaging/util/MessageFactory_AMQP_0_10.java
URL:
http://svn.apache.org/viewvc/qpid/branches/address-refactor2/qpid/java/client-api/src/main/java/org/apache/qpid/messaging/util/MessageFactory_AMQP_0_10.java?rev=1350707&view=auto
==============================================================================
---
qpid/branches/address-refactor2/qpid/java/client-api/src/main/java/org/apache/qpid/messaging/util/MessageFactory_AMQP_0_10.java
(added)
+++
qpid/branches/address-refactor2/qpid/java/client-api/src/main/java/org/apache/qpid/messaging/util/MessageFactory_AMQP_0_10.java
Fri Jun 15 17:22:05 2012
@@ -0,0 +1,626 @@
+package org.apache.qpid.messaging.util;
+
+import java.nio.ByteBuffer;
+import java.nio.CharBuffer;
+import java.nio.charset.CharacterCodingException;
+import java.nio.charset.Charset;
+import java.nio.charset.CharsetDecoder;
+import java.nio.charset.CharsetEncoder;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.qpid.messaging.ListMessage;
+import org.apache.qpid.messaging.MapMessage;
+import org.apache.qpid.messaging.Message;
+import org.apache.qpid.messaging.MessageEncodingException;
+import org.apache.qpid.messaging.MessageFactory;
+import org.apache.qpid.messaging.MessagingException;
+import org.apache.qpid.messaging.StringMessage;
+import org.apache.qpid.transport.DeliveryProperties;
+import org.apache.qpid.transport.MessageDeliveryMode;
+import org.apache.qpid.transport.MessageDeliveryPriority;
+import org.apache.qpid.transport.MessageProperties;
+import org.apache.qpid.transport.codec.BBDecoder;
+import org.apache.qpid.transport.codec.BBEncoder;
+import org.apache.qpid.util.UUIDGen;
+import org.apache.qpid.util.UUIDs;
+
+/**
+ * A generic message factory that is based on the AMQO 0-10 encoding.
+ *
+ */
+public class MessageFactory_AMQP_0_10 implements MessageFactory
+{
+ private static final Charset DEFAULT_CHARSET = Charset.forName("UTF-8");
+ private static boolean ALLOCATE_DIRECT =
Boolean.getBoolean("qpid.allocate-direct");
+ private static final ByteBuffer EMPTY_BYTE_BUFFER = ALLOCATE_DIRECT ?
ByteBuffer.allocateDirect(0) : ByteBuffer.allocate(0);
+
+ @Override
+ public Message createMessage(String text) throws MessageEncodingException
+ {
+ return new StringMessage_AMQP_0_10(new Mesage_AMQP_0_10(), text);
+ }
+
+ @Override
+ public Message createMessage(byte[] bytes) throws MessageEncodingException
+ {
+ ByteBuffer b;
+ if (ALLOCATE_DIRECT)
+ {
+ b = ByteBuffer.allocateDirect(bytes.length);
+ b.put(bytes);
+ }
+ else
+ {
+ b = ByteBuffer.wrap(bytes);
+ }
+ return new Mesage_AMQP_0_10(b);
+ }
+
+ @Override
+ public Message createMessage(ByteBuffer buf) throws
MessageEncodingException
+ {
+ if (ALLOCATE_DIRECT)
+ {
+ if (buf.isDirect())
+ {
+ return new Mesage_AMQP_0_10(buf);
+ }
+ else
+ {
+ // Silently copying the data to a direct buffer is not a good
thing as it can
+ // add a perf overhead. So an exception is a more reasonable
option.
+ throw new MessageEncodingException("The ByteBuffer needs to be
direct allocated");
+ }
+ }
+ else
+ {
+ return new Mesage_AMQP_0_10(buf);
+ }
+ }
+
+ @Override
+ public Message createMessage(Map<String, Object> map) throws
MessageEncodingException
+ {
+ return new MapMessage_AMQP_0_10(new Mesage_AMQP_0_10(), map);
+ }
+
+ @Override
+ public Message createMessage(List<Object> list) throws
MessageEncodingException
+ {
+ return new ListMessage_AMQP_0_10(new Mesage_AMQP_0_10(), list);
+ }
+
+ @Override
+ public String getContentAsString(Message m) throws MessageEncodingException
+ {
+ if (m instanceof StringMessage)
+ {
+ return ((StringMessage)m).getString();
+ }
+ else
+ {
+ return decodeAsString(m.getContent());
+ }
+ }
+
+ @Override
+ public Map<String, Object> getContentAsMap(Message m) throws
MessageEncodingException
+ {
+ if (m instanceof MapMessage)
+ {
+ return ((MapMessage)m).getMap();
+ }
+ else
+ {
+ return decodeAsMap(m.getContent());
+ }
+ }
+
+ @Override
+ public List<Object> getContentAsList(Message m) throws
MessageEncodingException
+ {
+ if (m instanceof ListMessage)
+ {
+ return ((ListMessage)m).getList();
+ }
+ else
+ {
+ return decodeAsList(m.getContent());
+ }
+ }
+
+ class Mesage_AMQP_0_10 implements Message
+ {
+ private MessageProperties _messageProps;
+ private DeliveryProperties _deliveryProps;
+ private ByteBuffer _data;
+
+ private UUIDGen _ssnNameGenerator = UUIDs.newGenerator();
+
+ protected Mesage_AMQP_0_10(MessageProperties messageProps,
DeliveryProperties deliveryProps)
+ {
+ this(messageProps, deliveryProps,EMPTY_BYTE_BUFFER);
+ }
+ protected Mesage_AMQP_0_10(MessageProperties messageProps,
+ DeliveryProperties deliveryProps,
+ ByteBuffer buf)
+ {
+ _messageProps = messageProps;
+ _deliveryProps = deliveryProps;
+ _data = buf;
+ }
+
+ protected Mesage_AMQP_0_10()
+ {
+ _messageProps = new MessageProperties();
+ _deliveryProps = new DeliveryProperties();
+ }
+
+ protected Mesage_AMQP_0_10(ByteBuffer buf)
+ {
+ _messageProps = new MessageProperties();
+ _deliveryProps = new DeliveryProperties();
+ _data = buf;
+ }
+
+ @Override
+ public String getMessageId() throws MessagingException
+ {
+ return _messageProps.getMessageId().toString();
+ }
+
+ @Override
+ public void setMessageId(String messageId) throws MessagingException
+ {
+ // Temp hack for the time being
+ _messageProps.setMessageId(_ssnNameGenerator.generate());
+ }
+
+ @Override
+ public String getSubject() throws MessagingException
+ {
+ Map<String,Object> props = _messageProps.getApplicationHeaders();
+ return props == null ? null : (String)props.get(QPID_SUBJECT);
+ }
+
+ @Override
+ public void setSubject(String subject) throws MessagingException
+ {
+ Map<String,Object> props = _messageProps.getApplicationHeaders();
+ if (props == null)
+ {
+ props = new HashMap<String,Object>();
+ _messageProps.setApplicationHeaders(props);
+ }
+ props.put(QPID_SUBJECT, subject);
+ }
+
+ @Override
+ public String getContentType() throws MessagingException
+ {
+ return _messageProps.getContentType();
+ }
+
+ @Override
+ public void setContentType(String contentType)
+ throws MessagingException
+ {
+ _messageProps.setContentType(contentType);
+ }
+
+ @Override
+ public String getCorrelationId() throws MessagingException
+ {
+ return new String(_messageProps.getCorrelationId());
+ }
+
+ @Override
+ public void setCorrelationId(String correlationId)
+ throws MessagingException
+ {
+ _messageProps.setCorrelationId(correlationId.getBytes());
+ }
+
+ @Override
+ public String getReplyTo() throws MessagingException
+ {
+ return addressFrom0_10_ReplyTo(_deliveryProps.getExchange(),
+ _deliveryProps.getRoutingKey());
+ }
+
+ @Override
+ public void setReplyTo(String replyTo) throws MessagingException
+ {
+ // TODO
+ }
+
+ @Override
+ public String getUserId() throws MessagingException
+ {
+ return new String(_messageProps.getUserId());
+ }
+
+ @Override
+ public void setUserId(String userId) throws MessagingException
+ {
+ _messageProps.setUserId(userId.getBytes());
+ }
+
+ @Override
+ public boolean isDurable() throws MessagingException
+ {
+ return _deliveryProps.getDeliveryMode() ==
MessageDeliveryMode.PERSISTENT;
+ }
+
+ @Override
+ public void setDurable(boolean durable) throws MessagingException
+ {
+ _deliveryProps.setDeliveryMode(durable ?
+ MessageDeliveryMode.PERSISTENT:
MessageDeliveryMode.NON_PERSISTENT);
+ }
+
+ @Override
+ public boolean isRedelivered() throws MessagingException
+ {
+ return _deliveryProps.getRedelivered();
+ }
+
+ @Override
+ public void setRedelivered(boolean redelivered)
+ throws MessagingException
+ {
+
+ _deliveryProps.setRedelivered(redelivered);
+ }
+
+ @Override
+ public int getPriority() throws MessagingException
+ {
+ return _deliveryProps.getPriority().getValue();
+ }
+
+ @Override
+ public void setPriority(int priority) throws MessagingException
+ {
+
_deliveryProps.setPriority(MessageDeliveryPriority.get((short)priority));
+
+ }
+
+ @Override
+ public long getTtl() throws MessagingException
+ {
+ return _deliveryProps.getTtl();
+ }
+
+ @Override
+ public void setTtl(long ttl) throws MessagingException
+ {
+ _deliveryProps.setTtl(ttl);
+ }
+
+ @Override
+ public long getTimestamp() throws MessagingException
+ {
+ return _deliveryProps.getTimestamp();
+ }
+
+ @Override
+ public void setTimestamp(long timestamp) throws MessagingException
+ {
+ _deliveryProps.setTimestamp(timestamp);
+ }
+
+ @Override
+ public Map<String, Object> getProperties() throws MessagingException
+ {
+ return _messageProps.getApplicationHeaders();
+ }
+
+ @Override
+ public void setProperty(String key, Object value)
+ throws MessagingException
+ {
+ Map<String,Object> props = _messageProps.getApplicationHeaders();
+ if (props == null)
+ {
+ props = new HashMap<String,Object>();
+ _messageProps.setApplicationHeaders(props);
+ }
+ props.put(key, value);
+ }
+
+ private String addressFrom0_10_ReplyTo(String exchange, String
routingKey)
+ {
+ if ("".equals(exchange)) // type Queue
+ {
+ return routingKey;
+ }
+ else
+ {
+ return exchange + "/" + routingKey;
+ }
+ }
+
+ @Override
+ public ByteBuffer getContent()
+ {
+ return _data;
+ }
+ }
+
+ class StringMessage_AMQP_0_10 extends GenericMessageAdapter implements
StringMessage
+ {
+ private String _str;
+ private ByteBuffer _rawData;
+ private MessageEncodingException _exception;
+
+ /**
+ * @param data The ByteBuffer passed will be read from position zero.
+ */
+ public StringMessage_AMQP_0_10(MessageProperties messageProps,
+ DeliveryProperties deliveryProps,
+ ByteBuffer data)
+ {
+ super(new Mesage_AMQP_0_10(messageProps, deliveryProps));
+ _rawData = (ByteBuffer) data.rewind();
+ try
+ {
+ _str = decodeAsString(_rawData.duplicate());
+ }
+ catch (MessageEncodingException e)
+ {
+ _exception = e;
+ }
+ }
+
+ public StringMessage_AMQP_0_10(Message delegate, String str) throws
MessageEncodingException
+ {
+ super(delegate);
+ if(_str == null || _str.isEmpty())
+ {
+ _rawData = EMPTY_BYTE_BUFFER;
+ }
+ else
+ {
+ _rawData = encodeString(str);
+ }
+ }
+
+ @Override
+ public String getString() throws MessageEncodingException
+ {
+ if (_exception != null)
+ {
+ throw _exception;
+ }
+ else
+ {
+ return _str;
+ }
+ }
+
+ @Override
+ public ByteBuffer getContent()
+ {
+ return _rawData;
+ }
+ }
+
+ /**
+ * @param data The ByteBuffer passed will be read from position zero.
+ */
+ class MapMessage_AMQP_0_10 extends GenericMessageAdapter implements
MapMessage
+ {
+ private Map<String,Object> _map;
+ private ByteBuffer _rawData;
+ private MessageEncodingException _exception;
+
+ public MapMessage_AMQP_0_10(MessageProperties messageProps,
+ DeliveryProperties deliveryProps,
+ ByteBuffer data)
+ {
+ super(new Mesage_AMQP_0_10(messageProps, deliveryProps));
+ _rawData = (ByteBuffer) data.rewind();
+ try
+ {
+ _map = decodeAsMap(_rawData.duplicate());
+ }
+ catch (MessageEncodingException e)
+ {
+ _exception = e;
+ }
+ }
+
+ public MapMessage_AMQP_0_10(Message delegate, Map<String,Object> map)
throws MessageEncodingException
+ {
+ super(delegate);
+ if(map == null || map.isEmpty())
+ {
+ _rawData = EMPTY_BYTE_BUFFER;
+ }
+ else
+ {
+ _rawData = encodeMap(map);
+ }
+ }
+
+ @Override
+ public Map<String,Object> getMap() throws MessageEncodingException
+ {
+ if (_exception != null)
+ {
+ throw _exception;
+ }
+ else
+ {
+ return _map;
+ }
+ }
+
+ @Override
+ public ByteBuffer getContent()
+ {
+ return _rawData;
+ }
+ }
+
+ /**
+ * @param data The ByteBuffer passed will be read from position zero.
+ */
+ class ListMessage_AMQP_0_10 extends GenericMessageAdapter implements
ListMessage
+ {
+ private List<Object> _list;
+ private ByteBuffer _rawData;
+ private MessageEncodingException _exception;
+
+ public ListMessage_AMQP_0_10(MessageProperties messageProps,
+ DeliveryProperties deliveryProps,
+ ByteBuffer data)
+ {
+ super(new Mesage_AMQP_0_10(messageProps, deliveryProps));
+ _rawData = (ByteBuffer) data.rewind();
+ try
+ {
+ _list = decodeAsList(_rawData.duplicate());
+ }
+ catch (MessageEncodingException e)
+ {
+ _exception = e;
+ }
+ }
+
+ public ListMessage_AMQP_0_10(Message delegate, List<Object> list)
throws MessageEncodingException
+ {
+ super(delegate);
+ if(list == null || list.isEmpty())
+ {
+ _rawData = EMPTY_BYTE_BUFFER;
+ }
+ else
+ {
+ _rawData = encodeList(list);
+ }
+ }
+
+ @Override
+ public List<Object> getList() throws MessageEncodingException
+ {
+ if (_exception != null)
+ {
+ throw _exception;
+ }
+ else
+ {
+ return _list;
+ }
+ }
+
+ @Override
+ public ByteBuffer getContent()
+ {
+ return _rawData;
+ }
+ }
+
+ protected static String decodeAsString(ByteBuffer buf) throws
MessageEncodingException
+ {
+ final CharsetDecoder decoder = DEFAULT_CHARSET.newDecoder();
+ try
+ {
+ return decoder.decode(buf).toString();
+ }
+ catch (CharacterCodingException e)
+ {
+ throw new MessageEncodingException("Error decoding content as
String using UTF-8",e);
+ }
+
+ }
+
+ protected static ByteBuffer encodeString(String str) throws
MessageEncodingException
+ {
+ final CharsetEncoder encoder = DEFAULT_CHARSET.newEncoder();
+ ByteBuffer b;
+ try
+ {
+ b = encoder.encode(CharBuffer.wrap(str));
+ b.flip();
+ }
+ catch (CharacterCodingException e)
+ {
+ throw new MessageEncodingException("Cannot encode string in UFT-8:
" + str,e);
+ }
+ if (ALLOCATE_DIRECT)
+ {
+ // unfortunately this extra copy is required as it does not seem
possible
+ // to create a CharSetEncoder that returns a buffer allocated
directly.
+ ByteBuffer direct = ByteBuffer.allocateDirect(b.remaining());
+ direct.put(b);
+ direct.flip();
+ return direct;
+ }
+ else
+ {
+ return b;
+ }
+ }
+
+ protected static Map<String,Object> decodeAsMap(ByteBuffer buf) throws
MessageEncodingException
+ {
+ try
+ {
+ BBDecoder decorder = new BBDecoder();
+ decorder.init(buf);
+ return decorder.readMap();
+ }
+ catch (Exception e)
+ {
+ throw new MessageEncodingException("Error decoding content as
Map",e);
+ }
+ }
+
+ protected static ByteBuffer encodeMap(Map<String,Object> map) throws
MessageEncodingException
+ {
+ try
+ {
+ //need to investigate the capacity here.
+ BBEncoder encoder = new BBEncoder(1024);
+ encoder.writeMap(map);
+ return (ByteBuffer)encoder.buffer().flip();
+ }
+ catch (Exception e)
+ {
+ throw new MessageEncodingException("Cannot encode Map" ,e);
+ }
+ }
+
+ protected static List<Object> decodeAsList(ByteBuffer buf) throws
MessageEncodingException
+ {
+ try
+ {
+ BBDecoder decorder = new BBDecoder();
+ decorder.init(buf);
+ return decorder.readList();
+ }
+ catch (Exception e)
+ {
+ throw new MessageEncodingException("Error decoding content as
List",e);
+ }
+ }
+
+ protected static ByteBuffer encodeList(List<Object> list) throws
MessageEncodingException
+ {
+ try
+ {
+ //need to investigate the capacity here.
+ BBEncoder encoder = new BBEncoder(1024);
+ encoder.writeList(list);
+ return (ByteBuffer)encoder.buffer().flip();
+ }
+ catch (Exception e)
+ {
+ throw new MessageEncodingException("Cannot encode List" ,e);
+ }
+ }
+}
Modified:
qpid/branches/address-refactor2/qpid/java/client-api/src/main/java/org/apache/qpid/messaging/util/ReadOnlyMessageAdapter.java
URL:
http://svn.apache.org/viewvc/qpid/branches/address-refactor2/qpid/java/client-api/src/main/java/org/apache/qpid/messaging/util/ReadOnlyMessageAdapter.java?rev=1350707&r1=1350706&r2=1350707&view=diff
==============================================================================
---
qpid/branches/address-refactor2/qpid/java/client-api/src/main/java/org/apache/qpid/messaging/util/ReadOnlyMessageAdapter.java
(original)
+++
qpid/branches/address-refactor2/qpid/java/client-api/src/main/java/org/apache/qpid/messaging/util/ReadOnlyMessageAdapter.java
Fri Jun 15 17:22:05 2012
@@ -17,6 +17,7 @@
*/
package org.apache.qpid.messaging.util;
+import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.Map;
@@ -25,28 +26,13 @@ import org.apache.qpid.messaging.Message
import org.apache.qpid.messaging.MessagingException;
/**
- * Ensures the message is read only by blocking the delegates
- * setter methods.
+ * Ensures the message properties and content are read only.
*/
-public class ReadOnlyMessageAdapter implements Message
+public class ReadOnlyMessageAdapter extends GenericMessageAdapter
{
- private Message _delegate;
-
ReadOnlyMessageAdapter(Message delegate)
{
- _delegate = delegate;
- }
-
- @Override
- public Object getContent() throws MessagingException
- {
- return _delegate.getContent();
- }
-
- @Override
- public String getMessageId() throws MessagingException
- {
- return _delegate.getMessageId();
+ super(delegate);
}
@Override
@@ -56,120 +42,60 @@ public class ReadOnlyMessageAdapter impl
}
@Override
- public String getSubject() throws MessagingException
- {
- return _delegate.getSubject();
- }
-
- @Override
public void setSubject(String subject) throws MessagingException
{
throwMessageNotWritableException();
}
@Override
- public String getContentType() throws MessagingException
- {
- return _delegate.getContentType();
- }
-
- @Override
public void setContentType(String contentType) throws MessagingException
{
throwMessageNotWritableException();
}
@Override
- public String getCorrelationId() throws MessagingException
- {
- return _delegate.getCorrelationId();
- }
-
- @Override
public void setCorrelationId(String correlationId) throws
MessagingException
{
throwMessageNotWritableException();
}
@Override
- public String getReplyTo() throws MessagingException
- {
- return _delegate.getReplyTo();
- }
-
- @Override
public void setReplyTo(String replyTo) throws MessagingException
{
throwMessageNotWritableException();
}
@Override
- public String getUserId() throws MessagingException
- {
- return _delegate.getUserId();
- }
-
- @Override
public void setUserId(String userId) throws MessagingException
{
throwMessageNotWritableException();
}
@Override
- public boolean isDurable() throws MessagingException
- {
- return _delegate.isDurable();
- }
-
- @Override
public void setDurable(boolean durable) throws MessagingException
{
throwMessageNotWritableException();
}
@Override
- public boolean isRedelivered() throws MessagingException
- {
- return _delegate.isRedelivered();
- }
-
- @Override
public void setRedelivered(boolean redelivered) throws MessagingException
{
throwMessageNotWritableException();
}
@Override
- public int getPriority() throws MessagingException
- {
- return _delegate.getPriority();
- }
-
- @Override
public void setPriority(int priority) throws MessagingException
{
throwMessageNotWritableException();
}
@Override
- public long getTtl() throws MessagingException
- {
- return _delegate.getTtl();
- }
-
- @Override
public void setTtl(long ttl) throws MessagingException
{
throwMessageNotWritableException();
}
@Override
- public long getTimestamp() throws MessagingException
- {
- return _delegate.getTimestamp();
- }
-
- @Override
public void setTimestamp(long timestamp) throws MessagingException
{
throwMessageNotWritableException();
@@ -178,7 +104,7 @@ public class ReadOnlyMessageAdapter impl
@Override
public Map<String, Object> getProperties() throws MessagingException
{
- return Collections.unmodifiableMap(_delegate.getProperties());
+ return Collections.unmodifiableMap(super.getProperties());
}
@Override
@@ -187,9 +113,14 @@ public class ReadOnlyMessageAdapter impl
throwMessageNotWritableException();
}
+ @Override
+ public ByteBuffer getContent()
+ {
+ return super.getContent().asReadOnlyBuffer();
+ }
+
private void throwMessageNotWritableException() throws
MessageNotWritableException
{
throw new MessageNotWritableException("Message is read-only");
}
-
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
