ServerMessageDatagram


1   package com.ubermq.jms.common.datagram.impl;
2   
3   import com.ubermq.jms.common.datagram.*;
4   import java.util.*;
5   
6   import com.ubermq.kernel.AbstractDatagram;
7   import java.nio.ByteBuffer  ;
8   
9   /**
10   * A slimmed down version of a message datagram intended for use
11   * on servers that simply route messages around, perform limited
12   * (if any) modifications, and will not retain messages for long
13   * periods.
14   * <P>
15   * This datagram creates no garbage, and is much faster than
16   * the alternative fully featured implementation.
17   * <P>
18   * It is very dependent on the accompanying MessageDatagram's wire format,
19   * but the implementation pattern can be used for other formats as well.
20   */
21  public class ServerMessageDatagram
22      extends AbstractDatagram
23      implements IMessageDatagram
24  {
25      private ByteBuffer   buffer;
26      private String   topicName;
27  
28      private int propsOffset;
29      private Map props;
30  
31      private static final int SENDER_ID_OFFSET = 0;
32      private static final int SEQUENCE_OFFSET = 8;
33      private static final int VARIOUS_OFFSET = 12;
34      private static final int TIMESTAMP_OFFSET = 20;
35      private static final int PROPS_LEN_OFFSET = 28;
36      private static final int TOPIC_NAME_OFFSET = 32;
37  
38  
39      /**
40       * Creates a server message datagram. The datagram is intended
41       * to be reusable, but doing so requires delicate state management
42       * so that datagram instances are not shared amongst various concurrent processes.
43       */
44      ServerMessageDatagram()
45      {
46          super(DatagramFactory.DGRAM_MSG, 0);
47      }
48  
49      /**
50       * Loads the datagram from a ByteBuffer.
51       * The position need not be zero, but should be set to the beginning
52       * of the datagram's data. The limit should represent the end of the
53       * datagram's data.
54       */
55      public void incoming(ByteBuffer   bb) throws java.io.IOException  
56      {
57          super.incoming(bb);
58          this.buffer = bb.slice();
59  
60          // the only thing we need to read a priori is the
61          // topic name - it's always used.
62          topicName =
63              AbstractDatagram.readPascalString((ByteBuffer  )buffer.position(TOPIC_NAME_OFFSET));
64  
65          // save where the properties begin, and blow away old properties from
66          // the last message.
67          propsOffset = buffer.position();
68          props = null;
69      }
70  
71      /**
72       * Outputs the datagram to a ByteBuffer, usually for output into
73       * a channel.
74       */
75      public void outgoing(java.nio.ByteBuffer   bb)
76      {
77          super.outgoing(bb);
78          bb.put((ByteBuffer  )buffer.rewind());
79      }
80  
81      /**
82       * Sets the sequence number associated with this message.
83       */
84      public void setSequence(int sequence)
85      {
86          throw new UnsupportedOperationException  ("mutation is not supported for server datagrams.");
87      }
88  
89      /**
90       * Gets the name of the topic that the message was sent to,
91       * or should be sent to.
92       * @return a topic name
93       */
94      public String   getTopicName()
95      {
96          return topicName;
97      }
98  
99      /**
100      * Gets the sequence number associated with this message. The sequence number
101      * is defined by the sender. Clients will expect the sequence number to
102      * monotonically increase and not skip any values. The starting point is not
103      * used.
104      */
105     public int getSequence()
106     {
107         return buffer.getInt(SEQUENCE_OFFSET);
108     }
109 
110     /**
111      * Gets the unique identifier for the sender of this message. The sender ID
112      * should be unique throughout the messaging space. A message ID is defined
113      * as a <code>{senderID, sequence}</code> tuple.
114      * @return a sender ID
115      */
116     public long getSenderId()
117     {
118         return buffer.getLong(SENDER_ID_OFFSET);
119     }
120 
121     /**
122      * Gets a unique identifier for this message. This ID can change
123      * over time if the Message datagram is reused for multiple publish()
124      * calls, or if clients call the setSenderID and setSequence methods.
125      * @return a message ID
126      */
127     public MessageId getMessageId()
128     {
129         return getIncomingMessageId();
130     }
131 
132     /**
133      * Gets the original unique identifier for this message
134      * when it arrived at the subscriber. This is used for acknowledgement
135      * and identification.
136      * @return the original message ID, constant for the life of the datagram.
137      */
138     public MessageId getIncomingMessageId()
139     {
140         return new MessageId(getSenderId(), getSequence());
141     }
142 
143     /**
144      * Gets the value of a standard property, one of the STDPROP_xxx constants.
145      * @param property the property index
146      * @return the value
147      */
148     public Object   getStandardProperty(int property)
149     {
150         switch(property)
151         {
152             case STDPROP_PRIORITY:
153                 return new Integer  (MessageDatagram.getFlagBasedProperty(buffer.getLong(VARIOUS_OFFSET), MessageDatagram.PRI_SHIFT, MessageDatagram.PRI_MASK));
154             case STDPROP_MSGTYPE:
155                 return new Integer  (MessageDatagram.getFlagBasedProperty(buffer.getLong(VARIOUS_OFFSET), MessageDatagram.MSGTYPE_SHIFT, MessageDatagram.MSGTYPE_MASK));
156             case STDPROP_TTL:
157                 return new Integer  (MessageDatagram.getFlagBasedProperty(buffer.getLong(VARIOUS_OFFSET), MessageDatagram.TTL_SHIFT, MessageDatagram.TTL_MASK) * 100);
158             case STDPROP_REDELIVERY:
159                 return new Boolean  (MessageDatagram.getFlagBasedProperty(buffer.getLong(VARIOUS_OFFSET), MessageDatagram.REDELIVERED_SHIFT, MessageDatagram.REDELIVERED_MASK) != 0 ? true : false);
160             case STDPROP_TIMESTAMP:
161                 return new Long  (buffer.getLong(TIMESTAMP_OFFSET));
162             case STDPROP_BODY:
163                 {
164                     return readTypedData((ByteBuffer  )this.buffer.duplicate().position(propsOffset + buffer.getInt(PROPS_LEN_OFFSET)));
165                 }
166             case STDPROP_DELIVERYMODE:
167                 return new Integer  (MessageDatagram.getFlagBasedProperty(buffer.getLong(VARIOUS_OFFSET), MessageDatagram.DELIVERY_SHIFT, MessageDatagram.DELIVERY_MASK));
168             default:
169                 return (hasProcessedProperties() ? props.get(new Integer  (property)) : null);
170         }
171     }
172 
173     public void setStandardProperty(int property, Object   value)
174     {
175         switch(property)
176         {
177             case STDPROP_REDELIVERY:
178                 buffer.putLong(VARIOUS_OFFSET,
179                                MessageDatagram.setFlagBasedProperty(buffer.getLong(VARIOUS_OFFSET),
180                                                                     MessageDatagram.REDELIVERED_SHIFT,
181                                                                     MessageDatagram.REDELIVERED_MASK,
182                                                                         ((Boolean  )value).booleanValue() ? 1 : 0));
183                 break;
184             case STDPROP_TIMESTAMP:
185                 buffer.putLong(TIMESTAMP_OFFSET, ((Number  )value).longValue());
186                 break;
187             default:
188                 throw new UnsupportedOperationException  ("mutation is not supported for server datagrams.");
189         }
190     }
191 
192     /**
193      * Gets the value of a custom (named) property. This is a very
194      * costly call for a server mesage datagram, because we need to parse
195      * the properties that were sent with the message to get the data out.
196      * <P>
197      * @param property the name of the property
198      * @return the value.
199      */
200     public Object   getCustomProperty(String   property)
201     {
202         int props_len = buffer.getInt(PROPS_LEN_OFFSET);
203 
204         if (!hasProcessedProperties()) {
205             props = new HashMap();
206             MessageDatagram.parsePropsBuffer(((ByteBuffer  )buffer.asReadOnlyBuffer().position(propsOffset).limit(propsOffset + props_len)).slice(),
207                                              props);
208         }
209 
210         return props.get(property);
211     }
212 
213     /**
214      * Indicates whether the property block has been processed by this
215      * datagram.  This can be used as a hint to determine if
216      * property access will have a significant performance cost, or if we've
217      * already incurred that cost.
218      * <P>
219      * @return true if the properties have been processed.
220      */
221     protected boolean hasProcessedProperties() {return (props != null);}
222 
223     /**
224      * Gets a Collection containing the names of all custom properties.
225      * @return the names of all custom properties, as String objects.
226      */
227     public java.util.Collection   getCustomPropertyNames()
228     {
229         // we don't support this. either you know the property name, or hey man, get out of here.
230         throw new UnsupportedOperationException  ("mutation is not supported for server datagrams.");
231     }
232 
233     /**
234      * Sets a custom (named) property.
235      * @param property the name of the property
236      * @param value the value of the property
237      */
238     public void setCustomProperty(String   property, Object   value)
239     {
240         throw new UnsupportedOperationException  ("mutation is not supported for server datagrams.");
241     }
242 
243     /**
244      * Removes all custom properties.
245      */
246     public void clearProperties()
247     {
248         throw new UnsupportedOperationException  ("mutation is not supported for server datagrams.");
249     }
250 
251     /**
252      * Sets the name of the topic to send the message to.
253      * @param a topic name
254      */
255     public void setTopicName(String   sz)
256     {
257         throw new UnsupportedOperationException  ("mutation is not supported for server datagrams.");
258     }
259 
260     /**
261      * Sets the sender identifier that should be attached to this message.
262      * @param senderId a sender ID
263      */
264     public void setSenderId(long senderId)
265     {
266         throw new UnsupportedOperationException  ("mutation is not supported for server datagrams.");
267     }
268 
269     /**
270      * Prepares a message datagram for sending, and specifies the
271      * options available to a JMS client at send-time.
272      * @param deliveryMode the JMS delivery mode to use
273      * @param pri the JMS priority to send the message at.
274      * @param ttl the TTL to use
275      */
276     public void prepareToSend(int deliveryMode, int pri, long ttl)
277     {
278         throw new UnsupportedOperationException  ("mutation is not supported for server datagrams.");
279     }
280 
281 }
282
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags