User: dmaplesden
Date: 01/08/28 14:38:25
Modified: src/main/org/jboss/mq SpyXAResourceManager.java
SpyMessage.java Connection.java
Log:
Made changes to allow a received message to be resent without having to copy it.
Previously this caused the acknowledgement of the original message to fail.
Revision Changes Path
1.3 +2 -6 jbossmq/src/main/org/jboss/mq/SpyXAResourceManager.java
Index: SpyXAResourceManager.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/SpyXAResourceManager.java,v
retrieving revision 1.2
retrieving revision 1.3
diff -u -r1.2 -r1.3
--- SpyXAResourceManager.java 2001/08/17 03:04:01 1.2
+++ SpyXAResourceManager.java 2001/08/28 21:38:24 1.3
@@ -20,7 +20,7 @@
*
* @author Hiram Chirino ([EMAIL PROTECTED])
* @created August 16, 2001
- * @version $Revision: 1.2 $
+ * @version $Revision: 1.3 $
*/
public class SpyXAResourceManager implements java.io.Serializable {
@@ -61,11 +61,7 @@
if ( state == null ) {
throw new JMSException( "Invalid transaction id." );
}
- AcknowledgementRequest item = new AcknowledgementRequest();
- item.destination = msg.getJMSDestination();
- item.messageID = msg.getJMSMessageID();
- item.isAck = true;
- item.subscriberId = msg.routeToSubscriber;
+ AcknowledgementRequest item = msg.getAcknowledgementRequest(true);
state.ackedMessages.addLast( item );
}
1.4 +105 -95 jbossmq/src/main/org/jboss/mq/SpyMessage.java
Index: SpyMessage.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/SpyMessage.java,v
retrieving revision 1.3
retrieving revision 1.4
diff -u -r1.3 -r1.4
--- SpyMessage.java 2001/08/17 05:38:13 1.3
+++ SpyMessage.java 2001/08/28 21:38:25 1.4
@@ -24,18 +24,18 @@
* @author Hiram Chirino ([EMAIL PROTECTED])
* @author David Maplesden ([EMAIL PROTECTED])
*
- * @version $Revision: 1.3 $
+ * @version $Revision: 1.4 $
*/
public class SpyMessage
implements Serializable, Message, Comparable, Externalizable
{
-
+
// Constants -----------------------------------------------------
-
+
public static final int DEFAULT_DELIVERY_MODE =
javax.jms.DeliveryMode.PERSISTENT;
public static final int DEFAULT_PRIORITY = 4;
public static final int DEFAULT_TIME_TO_LIVE = 0;
-
+
//Those attributes are not transient ---------------
//Header fields
//Set by send() method
@@ -62,200 +62,200 @@
public String producerClientId;
//For durable subscriptions
public DurableSubcriptionID durableSubscriberID = null;
-
+
// Transient Attributes ------------------------------------------
//For acknowledgment (set on the client side)
public transient SpySession session;
//For ordering in the JMSServerQueue (set on the server side)
public transient long messageId;
public transient Object persistData;
-
+
// Constructor ---------------------------------------------------
-
+
public SpyMessage()
{
-
+
prop=new Hashtable();
propReadWrite=true;
}
-
+
// Public --------------------------------------------------------
-
+
public String getJMSMessageID()
{
return jmsMessageID;
}
-
+
public void setJMSMessageID(String id) throws JMSException
{
jmsMessageID=id;
}
-
+
public long getJMSTimestamp()
{
return jmsTimeStamp;
}
-
+
public void setJMSTimestamp(long timestamp) throws JMSException
{
jmsTimeStamp=timestamp;
}
-
+
public byte [] getJMSCorrelationIDAsBytes() throws JMSException
{
if (jmsCorrelationID) throw new JMSException("JMSCorrelationID is a string");
return jmsCorrelationIDbyte;
}
-
+
public void setJMSCorrelationIDAsBytes(byte[] correlationID) throws JMSException
{
jmsCorrelationID=false;
jmsCorrelationIDbyte=(byte[])correlationID.clone();
jmsCorrelationIDString=null;
}
-
+
public void setJMSCorrelationID(String correlationID) throws JMSException
{
jmsCorrelationID=true;
jmsCorrelationIDString=correlationID;
jmsCorrelationIDbyte=null;
}
-
+
public String getJMSCorrelationID() throws JMSException
{
if (!jmsCorrelationID) throw new JMSException("JMSCorrelationID is an array");
return jmsCorrelationIDString;
}
-
+
public Destination getJMSReplyTo()
{
return jmsReplyTo;
}
-
+
public void setJMSReplyTo(Destination replyTo) throws JMSException
{
jmsReplyTo=replyTo;
}
-
+
public Destination getJMSDestination()
{
return jmsDestination;
}
-
+
public void setJMSDestination(Destination destination) throws JMSException
{
jmsDestination=destination;
}
-
+
public int getJMSDeliveryMode()
{
return jmsDeliveryMode;
}
-
+
public void setJMSDeliveryMode(int deliveryMode) throws JMSException
{
jmsDeliveryMode=deliveryMode;
}
-
+
public boolean getJMSRedelivered()
{
return jmsRedelivered;
}
-
+
public void setJMSRedelivered(boolean redelivered) throws JMSException
{
jmsRedelivered=redelivered;
}
-
+
public String getJMSType()
{
return jmsType;
}
-
+
public void setJMSType(String type) throws JMSException
{
jmsType=type;
}
-
+
public long getJMSExpiration()
{
return jmsExpiration;
}
-
+
public void setJMSExpiration(long expiration) throws JMSException
{
jmsExpiration=expiration;
}
-
+
public int getJMSPriority()
{
return jmsPriority;
}
-
+
public void setJMSPriority(int priority) throws JMSException
{
jmsPriority=priority;
}
-
+
public void clearProperties() throws JMSException
{
prop=new Hashtable();
propReadWrite=true;
}
-
+
public boolean propertyExists(String name) throws JMSException
{
return prop.containsKey(name);
}
-
+
public boolean getBooleanProperty(String name) throws JMSException
{
Object value=prop.get(name);
if (value==null) return Boolean.valueOf(null).booleanValue();
-
+
if (value instanceof Boolean) return ((Boolean)value).booleanValue();
else if (value instanceof String) return
Boolean.valueOf((String)value).booleanValue();
else throw new MessageFormatException("Invalid conversion");
}
-
+
public byte getByteProperty(String name) throws JMSException
{
Object value=prop.get(name);
if (value==null) return Byte.valueOf(null).byteValue();
-
+
if (value instanceof Byte) return ((Byte)value).byteValue();
else if (value instanceof String) return Byte.parseByte((String)value);
else throw new MessageFormatException("Invalid conversion");
}
-
+
public short getShortProperty(String name) throws JMSException
{
Object value=prop.get(name);
if (value==null) return Short.valueOf(null).shortValue();
-
+
if (value instanceof Byte) return ((Byte)value).shortValue();
else if (value instanceof Short) return ((Short)value).shortValue();
else if (value instanceof String) return Short.parseShort((String)value);
else throw new MessageFormatException("Invalid conversion");
}
-
+
public int getIntProperty(String name) throws JMSException
{
Object value=prop.get(name);
if (value==null) return Integer.valueOf(null).intValue();
-
+
if (value instanceof Byte) return ((Byte)value).intValue();
else if (value instanceof Short) return ((Short)value).intValue();
else if (value instanceof Integer) return ((Integer)value).intValue();
else if (value instanceof String) return Integer.parseInt((String)value);
else throw new MessageFormatException("Invalid conversion");
}
-
+
public long getLongProperty(String name) throws JMSException
{
Object value=prop.get(name);
if (value==null) return Long.valueOf(null).longValue();
-
+
if (value instanceof Byte) return ((Byte)value).longValue();
else if (value instanceof Short) return ((Short)value).longValue();
else if (value instanceof Integer) return ((Integer)value).longValue();
@@ -263,33 +263,33 @@
else if (value instanceof String) return Long.parseLong((String)value);
else throw new MessageFormatException("Invalid conversion");
}
-
+
public float getFloatProperty(String name) throws JMSException
{
Object value=prop.get(name);
if (value==null) return Float.valueOf(null).floatValue();
-
+
if (value instanceof Float) return ((Float)value).floatValue();
else if (value instanceof String) return Float.parseFloat((String)value);
else throw new MessageFormatException("Invalid conversion");
}
-
+
public double getDoubleProperty(String name) throws JMSException
{
Object value=prop.get(name);
if (value==null) return Double.valueOf(null).doubleValue();
-
+
if (value instanceof Float) return ((Float)value).doubleValue();
else if (value instanceof Double) return ((Double)value).doubleValue();
else if (value instanceof String) return Double.parseDouble((String)value);
else throw new MessageFormatException("Invalid conversion");
}
-
+
public String getStringProperty(String name) throws JMSException
{
Object value=prop.get(name);
if (value==null) return null;
-
+
if (value instanceof Boolean) return ((Boolean)value).toString();
else if (value instanceof Byte) return ((Byte)value).toString();
else if (value instanceof Short) return ((Short)value).toString();
@@ -300,95 +300,95 @@
else if (value instanceof String) return (String)value;
else throw new MessageFormatException("Invalid conversion");
}
-
+
public Object getObjectProperty(String name) throws JMSException
{
Object value=prop.get(name);
return value;
}
-
+
public Enumeration getPropertyNames() throws JMSException
{
return prop.keys();
}
-
+
void CheckPropertyName(String name) throws JMSException
{
if (name.regionMatches(false,0,"JMS_",0,4))
{
throw new JMSException("Bad property name");
}
-
+
if (name.regionMatches(false,0,"JMSX",0,4))
{
if (name.equals("JMSXGroupId")) return;
if (name.equals("JMSXGroupSeq")) return;
throw new JMSException("Bad property name");
}
-
+
}
-
+
public void setBooleanProperty(String name, boolean value) throws JMSException
{
CheckPropertyName(name);
if (!propReadWrite) throw new MessageNotWriteableException("Properties are
read-only");
prop.put(name,new Boolean(value));
}
-
+
public void setByteProperty(String name, byte value) throws JMSException
{
CheckPropertyName(name);
if (!propReadWrite) throw new MessageNotWriteableException("Properties are
read-only");
prop.put(name,new Byte(value));
}
-
+
public void setShortProperty(String name, short value) throws JMSException
{
CheckPropertyName(name);
if (!propReadWrite) throw new MessageNotWriteableException("Properties are
read-only");
prop.put(name,new Short(value));
}
-
+
public void setIntProperty(String name, int value) throws JMSException
{
CheckPropertyName(name);
if (!propReadWrite) throw new MessageNotWriteableException("Properties are
read-only");
prop.put(name,new Integer(value));
}
-
+
public void setLongProperty(String name, long value) throws JMSException
{
CheckPropertyName(name);
if (!propReadWrite) throw new MessageNotWriteableException("Properties are
read-only");
prop.put(name,new Long(value));
}
-
+
public void setFloatProperty(String name, float value) throws JMSException
{
CheckPropertyName(name);
if (!propReadWrite) throw new MessageNotWriteableException("Properties are
read-only");
prop.put(name,new Float(value));
}
-
+
public void setDoubleProperty(String name, double value) throws JMSException
{
CheckPropertyName(name);
if (!propReadWrite) throw new MessageNotWriteableException("Properties are
read-only");
prop.put(name,new Double(value));
}
-
+
public void setStringProperty(String name, String value) throws JMSException
{
CheckPropertyName(name);
if (!propReadWrite) throw new MessageNotWriteableException("Properties are
read-only");
prop.put(name,value);
}
-
+
public void setObjectProperty(String name, Object value) throws JMSException
{
CheckPropertyName(name);
if (!propReadWrite) throw new MessageNotWriteableException("Properties are
read-only");
-
+
if (value instanceof Boolean) prop.put(name,value);
else if (value instanceof Byte) prop.put(name,value);
else if (value instanceof Short) prop.put(name,value);
@@ -399,36 +399,36 @@
else if (value instanceof String) prop.put(name,value);
else throw new MessageFormatException("Invalid object type");
}
-
+
public void clearBody() throws JMSException
{
//Inherited classes clear their content here
msgReadOnly=false;
}
-
+
public void acknowledge() throws JMSException
{
if (session==null)
throw new JMSException("This message was not recieved from the provider");
-
+
if( session.acknowledgeMode == session.CLIENT_ACKNOWLEDGE )
doAcknowledge();
-
+
}
-
+
public void setReadOnlyMode()
{
propReadWrite=false;
msgReadOnly=true;
}
-
+
public SpyMessage myClone() throws JMSException
{
SpyMessage result = new SpyMessage();
result.copyProps(this);
return result;
}
-
+
public void copyProps(SpyMessage original) throws JMSException
{
try
@@ -453,7 +453,7 @@
String key = (String) en.nextElement();
this.setStringProperty(key,original.getStringProperty(key));
}
-
+
//Spy Message special props
this.propReadWrite = original.propReadWrite;
this.msgReadOnly = original.msgReadOnly;
@@ -461,15 +461,15 @@
if(original.durableSubscriberID != null)
this.durableSubscriberID = new
DurableSubcriptionID(original.durableSubscriberID.clientID,original.durableSubscriberID.subscriptionName);
}
-
+
public boolean isOutdated()
{
if (jmsExpiration==0) return false;
long ts=System.currentTimeMillis();
return jmsExpiration<ts;
}
-
-
+
+
/**
* Return a negative number if this message should be sent
* before the o message. Return a positive if should be sent
@@ -478,7 +478,7 @@
public int compareTo(Object o)
{
SpyMessage sm = (SpyMessage)o;
-
+
if( jmsPriority > sm.jmsPriority)
{
return -1;
@@ -489,38 +489,48 @@
}
return (int)(messageId - sm.messageId);
}
-
-
+
+
public void doAcknowledge() throws JMSException
{
if( session.closed )
throw new IllegalStateException("Session is closed.");
session.connection.send(getAcknowledgementRequest(true));
}
-
-
+
+
public void doNegAcknowledge() throws JMSException
{
if( session.closed )
throw new IllegalStateException("Session is closed.");
session.connection.send(getAcknowledgementRequest(false));
}
-
+
+ public void createAcknowledgementRequest(int subscriptionId)
+ {
+ ack = new AcknowledgementRequest();
+ ack.destination = jmsDestination;
+ ack.messageID = jmsMessageID;
+ ack.subscriberId = subscriptionId;
+ }
+
public AcknowledgementRequest getAcknowledgementRequest(boolean isAck) throws
JMSException
{
+ //don't know if we have to copy but to be on safe side...
AcknowledgementRequest item = new AcknowledgementRequest();
- item.destination = jmsDestination;
- item.messageID = jmsMessageID;
+ item.destination = ack.destination;
+ item.messageID = ack.messageID;
+ item.subscriberId = ack.subscriberId;
item.isAck = isAck;
- item.subscriberId = routeToSubscriber;
return item;
}
-
- public transient int routeToSubscriber;
+
+ public transient AcknowledgementRequest ack;
+
// So that every time we rebuild the sources we don't have
// Version serialization problems
private final static long serialVersionUID = 467206190892964404L;
-
+
//Custom serialization to attempt to get (big) speed improvement on message
passing and persistence
protected static final byte OBJECT_MESS = 1;
protected static final byte BYTES_MESS = 2;
@@ -529,7 +539,7 @@
protected static final byte STREAM_MESS = 5;
protected static final byte ENCAP_MESS = 6;
protected static final byte SPY_MESS = 7;
-
+
public static void writeMessage(SpyMessage message,ObjectOutput out)throws
IOException
{
if(message instanceof SpyEncapsulatedMessage)
@@ -556,7 +566,7 @@
}
message.writeExternal(out);
}
-
+
public static SpyMessage readMessage(ObjectInput in) throws IOException
{
SpyMessage message = null;
@@ -593,7 +603,7 @@
}
return message;
}
-
+
protected static final int BYTE = 0;
protected static final int SHORT = 1;
protected static final int INT = 2;
@@ -604,7 +614,7 @@
protected static final int STRING = 7;
protected static final int OBJECT = 8;
protected static final int NULL = 9;
-
+
public void writeExternal(java.io.ObjectOutput out) throws java.io.IOException
{
SpyDestination.writeDest(out,jmsDestination);
@@ -679,7 +689,7 @@
}
}
}
-
+
private static void writeString(java.io.ObjectOutput out,String s) throws
java.io.IOException
{
if(s == null)
@@ -691,7 +701,7 @@
out.writeUTF(s);
}
}
-
+
private static String readString(java.io.ObjectInput in) throws
java.io.IOException
{
byte b = in.readByte();
@@ -700,7 +710,7 @@
else
return in.readUTF();
}
-
+
public void readExternal(java.io.ObjectInput in) throws java.io.IOException,
ClassNotFoundException
{
jmsDestination = SpyDestination.readDest(in);
@@ -765,5 +775,5 @@
prop.put(key,value);
}
}
-
+
}
1.4 +4 -3 jbossmq/src/main/org/jboss/mq/Connection.java
Index: Connection.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/Connection.java,v
retrieving revision 1.3
retrieving revision 1.4
diff -u -r1.3 -r1.4
--- Connection.java 2001/08/17 22:57:42 1.3
+++ Connection.java 2001/08/28 21:38:25 1.4
@@ -36,7 +36,7 @@
* @author Norbert Lataille ([EMAIL PROTECTED])
* @author Hiram Chirino ([EMAIL PROTECTED])
* @created August 16, 2001
- * @version $Revision: 1.3 $
+ * @version $Revision: 1.4 $
*/
public class Connection implements java.io.Serializable, javax.jms.Connection {
//Maps a destination to a LinkedList of Subscriptions
@@ -335,7 +335,8 @@
continue;
}
- requests[i].message.routeToSubscriber =
requests[i].subscriptionId.intValue();
+
requests[i].message.createAcknowledgementRequest(requests[i].subscriptionId.intValue());
+
consumer.addMessage( requests[i].message );
}
} catch ( JMSException e ) {
@@ -587,7 +588,7 @@
try {
SpyMessage message = serverIL.receive( connectionToken,
sub.subscriptionId, wait );
if ( message != null ) {
- message.routeToSubscriber = sub.subscriptionId;
+ message.createAcknowledgementRequest(sub.subscriptionId);
}
return message;
} catch ( Exception e ) {
_______________________________________________
Jboss-development mailing list
[EMAIL PROTECTED]
http://lists.sourceforge.net/lists/listinfo/jboss-development
