You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2007/01/08 18:02:37 UTC
svn commit: r494121 [3/5] - in /incubator/qpid/trunk/qpid:
gentools/src/org/apache/qpid/gentools/
java/broker/src/main/java/org/apache/qpid/server/
java/broker/src/main/java/org/apache/qpid/server/ack/
java/broker/src/main/java/org/apache/qpid/server/c...
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ChannelCloseMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ChannelCloseMethodHandler.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ChannelCloseMethodHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ChannelCloseMethodHandler.java Mon Jan 8 09:02:26 2007
@@ -33,6 +33,7 @@
import org.apache.qpid.framing.AMQFrame;
import org.apache.qpid.framing.ChannelCloseBody;
import org.apache.qpid.framing.ChannelCloseOkBody;
+import org.apache.qpid.framing.AMQShortString;
public class ChannelCloseMethodHandler implements StateAwareMethodListener
{
@@ -51,7 +52,7 @@
ChannelCloseBody method = (ChannelCloseBody) evt.getMethod();
int errorCode = method.replyCode;
- String reason = method.replyText;
+ AMQShortString reason = method.replyText;
if (_logger.isDebugEnabled())
{
_logger.debug("Channel close reply code: " + errorCode + ", reason: " + reason);
@@ -77,7 +78,7 @@
{
_logger.info("Broker responded with Invalid Selector.");
- throw new AMQInvalidSelectorException(reason);
+ throw new AMQInvalidSelectorException(String.valueOf(reason));
}
else
{
@@ -85,6 +86,6 @@
}
}
- evt.getProtocolSession().channelClosed(evt.getChannelId(), errorCode, reason);
+ evt.getProtocolSession().channelClosed(evt.getChannelId(), errorCode, String.valueOf(reason));
}
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionCloseMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionCloseMethodHandler.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionCloseMethodHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionCloseMethodHandler.java Mon Jan 8 09:02:26 2007
@@ -31,6 +31,7 @@
import org.apache.qpid.client.AMQAuthenticationException;
import org.apache.qpid.framing.ConnectionCloseBody;
import org.apache.qpid.framing.ConnectionCloseOkBody;
+import org.apache.qpid.framing.AMQShortString;
public class ConnectionCloseMethodHandler implements StateAwareMethodListener
{
@@ -56,7 +57,7 @@
//stateManager.changeState(AMQState.CONNECTION_CLOSING);
int errorCode = method.replyCode;
- String reason = method.replyText;
+ AMQShortString reason = method.replyText;
// TODO: check whether channel id of zero is appropriate
// AMQP version change: Hardwire the version to 0-8 (major=8, minor=0)
@@ -75,7 +76,7 @@
//todo this is a bit of a fudge (could be conssidered such as each new connection needs a new state manager or at least a fresh state.
stateManager.changeState(AMQState.CONNECTION_NOT_STARTED);
- throw new AMQAuthenticationException(errorCode, reason);
+ throw new AMQAuthenticationException(errorCode, reason == null ? null : reason.toString());
}
else
{
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionRedirectMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionRedirectMethodHandler.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionRedirectMethodHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionRedirectMethodHandler.java Mon Jan 8 09:02:26 2007
@@ -49,19 +49,20 @@
_logger.info("ConnectionRedirect frame received");
ConnectionRedirectBody method = (ConnectionRedirectBody) evt.getMethod();
+ String host = method.host.toString();
// the host is in the form hostname:port with the port being optional
- int portIndex = method.host.indexOf(':');
- String host;
+ int portIndex = host.indexOf(':');
+
int port;
if (portIndex == -1)
{
- host = method.host;
port = DEFAULT_REDIRECT_PORT;
}
else
{
- host = method.host.substring(0, portIndex);
- port = Integer.parseInt(method.host.substring(portIndex + 1));
+ port = Integer.parseInt(host.substring(portIndex + 1));
+ host = host.substring(0, portIndex);
+
}
evt.getProtocolSession().failover(host, port);
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionStartMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionStartMethodHandler.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionStartMethodHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionStartMethodHandler.java Mon Jan 8 09:02:26 2007
@@ -31,10 +31,7 @@
import org.apache.qpid.client.state.AMQState;
import org.apache.qpid.client.state.AMQStateManager;
import org.apache.qpid.client.state.StateAwareMethodListener;
-import org.apache.qpid.framing.ConnectionStartBody;
-import org.apache.qpid.framing.ConnectionStartOkBody;
-import org.apache.qpid.framing.FieldTable;
-import org.apache.qpid.framing.FieldTableFactory;
+import org.apache.qpid.framing.*;
import javax.security.sasl.Sasl;
import javax.security.sasl.SaslClient;
@@ -122,18 +119,18 @@
stateManager.changeState(AMQState.CONNECTION_NOT_TUNED);
FieldTable clientProperties = FieldTableFactory.newFieldTable();
- clientProperties.put(ClientProperties.instance.toString(), ps.getClientID());
- clientProperties.put(ClientProperties.product.toString(), QpidProperties.getProductName());
- clientProperties.put(ClientProperties.version.toString(), QpidProperties.getReleaseVersion());
- clientProperties.put(ClientProperties.platform.toString(), getFullSystemInfo());
+ clientProperties.setString(ClientProperties.instance.toString(), ps.getClientID());
+ clientProperties.setString(ClientProperties.product.toString(), QpidProperties.getProductName());
+ clientProperties.setString(ClientProperties.version.toString(), QpidProperties.getReleaseVersion());
+ clientProperties.setString(ClientProperties.platform.toString(), getFullSystemInfo());
// AMQP version change: Hardwire the version to 0-8 (major=8, minor=0)
// TODO: Connect this to the session version obtained from ProtocolInitiation for this session.
// Be aware of possible changes to parameter order as versions change.
ps.writeFrame(ConnectionStartOkBody.createAMQFrame(evt.getChannelId(),
(byte)8, (byte)0, // AMQP version (major, minor)
clientProperties, // clientProperties
- selectedLocale, // locale
- mechanism, // mechanism
+ new AMQShortString(selectedLocale), // locale
+ new AMQShortString(mechanism), // mechanism
saslResponse)); // response
}
catch (UnsupportedEncodingException e)
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionTuneMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionTuneMethodHandler.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionTuneMethodHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionTuneMethodHandler.java Mon Jan 8 09:02:26 2007
@@ -28,10 +28,7 @@
import org.apache.qpid.client.state.AMQState;
import org.apache.qpid.client.state.AMQStateManager;
import org.apache.qpid.client.state.StateAwareMethodListener;
-import org.apache.qpid.framing.ConnectionOpenBody;
-import org.apache.qpid.framing.ConnectionTuneBody;
-import org.apache.qpid.framing.ConnectionTuneOkBody;
-import org.apache.qpid.framing.AMQFrame;
+import org.apache.qpid.framing.*;
public class ConnectionTuneMethodHandler implements StateAwareMethodListener
{
@@ -67,10 +64,10 @@
stateManager.changeState(AMQState.CONNECTION_NOT_OPENED);
session.writeFrame(createTuneOkFrame(evt.getChannelId(), params));
- session.writeFrame(createConnectionOpenFrame(evt.getChannelId(), session.getAMQConnection().getVirtualHost(), null, true));
+ session.writeFrame(createConnectionOpenFrame(evt.getChannelId(), new AMQShortString(session.getAMQConnection().getVirtualHost()), null, true));
}
- protected AMQFrame createConnectionOpenFrame(int channel, String path, String capabilities, boolean insist)
+ protected AMQFrame createConnectionOpenFrame(int channel, AMQShortString path, AMQShortString capabilities, boolean insist)
{
// AMQP version change: Hardwire the version to 0-8 (major=8, minor=0)
// TODO: Connect this to the session version obtained from ProtocolInitiation for this session.
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?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- 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 Mon Jan 8 09:02:26 2007
@@ -55,7 +55,7 @@
AbstractBytesMessage(ByteBuffer data)
{
super(data); // this instanties a content header
- getJmsContentHeaderProperties().setContentType(getMimeType());
+ getJmsContentHeaderProperties().setContentType(getMimeTypeAsShortString());
if (_data == null)
{
@@ -74,7 +74,7 @@
{
// TODO: this casting is ugly. Need to review whole ContentHeaderBody idea
super(messageNbr, (BasicContentHeaderProperties) contentHeader.properties, data);
- getJmsContentHeaderProperties().setContentType(getMimeType());
+ getJmsContentHeaderProperties().setContentType(getMimeTypeAsShortString());
}
public void clearBodyImpl() throws JMSException
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?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- 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 Mon Jan 8 09:02:26 2007
@@ -30,6 +30,7 @@
import org.apache.qpid.client.BasicMessageConsumer;
import org.apache.qpid.framing.BasicContentHeaderProperties;
import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.framing.AMQShortString;
import javax.jms.Destination;
import javax.jms.JMSException;
@@ -168,7 +169,7 @@
}
final AMQDestination amqd = (AMQDestination) destination;
- final String encodedDestination = amqd.getEncodedName();
+ final AMQShortString encodedDestination = amqd.getEncodedName();
_destinationCache.put(encodedDestination, destination);
getJmsContentHeaderProperties().setReplyTo(encodedDestination);
}
@@ -235,7 +236,7 @@
public void clearProperties() throws JMSException
{
- getJmsContentHeaderProperties().getJMSHeaders().clear();
+ getJmsContentHeaderProperties().clear();
_readableProperties = false;
}
@@ -247,139 +248,168 @@
}
+ public boolean propertyExists(AMQShortString propertyName) throws JMSException
+ {
+ checkPropertyName(propertyName);
+ return getJmsContentHeaderProperties().propertyExists(propertyName);
+ }
+
+
public boolean propertyExists(String propertyName) throws JMSException
{
checkPropertyName(propertyName);
- return getJmsContentHeaderProperties().getJMSHeaders().propertyExists(propertyName);
+ return getJmsContentHeaderProperties().propertyExists(propertyName);
}
+ public boolean getBooleanProperty(AMQShortString propertyName) throws JMSException
+ {
+ checkPropertyName(propertyName);
+
+ return getJmsContentHeaderProperties().getBoolean(propertyName);
+ }
+
+
public boolean getBooleanProperty(String propertyName) throws JMSException
{
checkPropertyName(propertyName);
- return getJmsContentHeaderProperties().getJMSHeaders().getBoolean(propertyName);
+ return getJmsContentHeaderProperties().getBoolean(propertyName);
}
public byte getByteProperty(String propertyName) throws JMSException
{
checkPropertyName(propertyName);
- return getJmsContentHeaderProperties().getJMSHeaders().getByte(propertyName);
+ return getJmsContentHeaderProperties().getByte(propertyName);
}
public short getShortProperty(String propertyName) throws JMSException
{
checkPropertyName(propertyName);
- return getJmsContentHeaderProperties().getJMSHeaders().getShort(propertyName);
+ return getJmsContentHeaderProperties().getShort(propertyName);
}
public int getIntProperty(String propertyName) throws JMSException
{
checkPropertyName(propertyName);
- return getJmsContentHeaderProperties().getJMSHeaders().getInteger(propertyName);
+ return getJmsContentHeaderProperties().getInteger(propertyName);
}
public long getLongProperty(String propertyName) throws JMSException
{
checkPropertyName(propertyName);
- return getJmsContentHeaderProperties().getJMSHeaders().getLong(propertyName);
+ return getJmsContentHeaderProperties().getLong(propertyName);
}
public float getFloatProperty(String propertyName) throws JMSException
{
checkPropertyName(propertyName);
- return getJmsContentHeaderProperties().getJMSHeaders().getFloat(propertyName);
+ return getJmsContentHeaderProperties().getFloat(propertyName);
}
public double getDoubleProperty(String propertyName) throws JMSException
{
checkPropertyName(propertyName);
- return getJmsContentHeaderProperties().getJMSHeaders().getDouble(propertyName);
+ return getJmsContentHeaderProperties().getDouble(propertyName);
}
public String getStringProperty(String propertyName) throws JMSException
{
checkPropertyName(propertyName);
- return getJmsContentHeaderProperties().getJMSHeaders().getString(propertyName);
+ return getJmsContentHeaderProperties().getString(propertyName);
}
public Object getObjectProperty(String propertyName) throws JMSException
{
checkPropertyName(propertyName);
- return getJmsContentHeaderProperties().getJMSHeaders().getObject(propertyName);
+ return getJmsContentHeaderProperties().getObject(propertyName);
}
public Enumeration getPropertyNames() throws JMSException
{
- return getJmsContentHeaderProperties().getJMSHeaders().getPropertyNames();
+ return getJmsContentHeaderProperties().getPropertyNames();
+ }
+
+ public void setBooleanProperty(AMQShortString propertyName, boolean b) throws JMSException
+ {
+ checkWritableProperties();
+ checkPropertyName(propertyName);
+ getJmsContentHeaderProperties().setBoolean(propertyName, b);
}
public void setBooleanProperty(String propertyName, boolean b) throws JMSException
{
checkWritableProperties();
checkPropertyName(propertyName);
- getJmsContentHeaderProperties().getJMSHeaders().setBoolean(propertyName, b);
+ getJmsContentHeaderProperties().setBoolean(propertyName, b);
}
public void setByteProperty(String propertyName, byte b) throws JMSException
{
checkWritableProperties();
checkPropertyName(propertyName);
- getJmsContentHeaderProperties().getJMSHeaders().setByte(propertyName, new Byte(b));
+ getJmsContentHeaderProperties().setByte(propertyName, new Byte(b));
}
public void setShortProperty(String propertyName, short i) throws JMSException
{
checkWritableProperties();
checkPropertyName(propertyName);
- getJmsContentHeaderProperties().getJMSHeaders().setShort(propertyName, new Short(i));
+ getJmsContentHeaderProperties().setShort(propertyName, new Short(i));
}
public void setIntProperty(String propertyName, int i) throws JMSException
{
checkWritableProperties();
checkPropertyName(propertyName);
- getJmsContentHeaderProperties().getJMSHeaders().setInteger(propertyName, new Integer(i));
+ getJmsContentHeaderProperties().setInteger(propertyName, new Integer(i));
}
public void setLongProperty(String propertyName, long l) throws JMSException
{
checkWritableProperties();
checkPropertyName(propertyName);
- getJmsContentHeaderProperties().getJMSHeaders().setLong(propertyName, new Long(l));
+ getJmsContentHeaderProperties().setLong(propertyName, new Long(l));
}
public void setFloatProperty(String propertyName, float f) throws JMSException
{
checkWritableProperties();
checkPropertyName(propertyName);
- getJmsContentHeaderProperties().getJMSHeaders().setFloat(propertyName, new Float(f));
+ getJmsContentHeaderProperties().setFloat(propertyName, new Float(f));
}
public void setDoubleProperty(String propertyName, double v) throws JMSException
{
checkWritableProperties();
checkPropertyName(propertyName);
- getJmsContentHeaderProperties().getJMSHeaders().setDouble(propertyName, new Double(v));
+ getJmsContentHeaderProperties().setDouble(propertyName, new Double(v));
}
public void setStringProperty(String propertyName, String value) throws JMSException
{
checkWritableProperties();
checkPropertyName(propertyName);
- getJmsContentHeaderProperties().getJMSHeaders().setString(propertyName, value);
+ getJmsContentHeaderProperties().setString(propertyName, value);
}
public void setObjectProperty(String propertyName, Object object) throws JMSException
{
checkWritableProperties();
checkPropertyName(propertyName);
- getJmsContentHeaderProperties().getJMSHeaders().setObject(propertyName, object);
+ getJmsContentHeaderProperties().setObject(propertyName, object);
}
+ protected void removeProperty(AMQShortString propertyName) throws JMSException
+ {
+ checkPropertyName(propertyName);
+ getJmsContentHeaderProperties().remove(propertyName);
+ }
+
+
protected void removeProperty(String propertyName) throws JMSException
{
checkPropertyName(propertyName);
- getJmsContentHeaderProperties().getJMSHeaders().remove(propertyName);
+ getJmsContentHeaderProperties().remove(propertyName);
}
public void acknowledgeThis() throws JMSException
@@ -421,7 +451,12 @@
*/
public abstract String toBodyString() throws JMSException;
- public abstract String getMimeType();
+ public String getMimeType()
+ {
+ return getMimeTypeAsShortString().toString();
+ }
+
+ public abstract AMQShortString getMimeTypeAsShortString();
public String toString()
{
@@ -436,13 +471,13 @@
buf.append("\nJMS reply to: ").append(String.valueOf(getJMSReplyTo()));
buf.append("\nAMQ message number: ").append(_deliveryTag);
buf.append("\nProperties:");
- if (getJmsContentHeaderProperties().getJMSHeaders().isEmpty())
+ if (getJmsContentHeaderProperties().isEmpty())
{
buf.append("<NONE>");
}
else
{
- buf.append('\n').append(getJmsContentHeaderProperties().getJMSHeaders());
+ buf.append('\n').append(getJmsContentHeaderProperties().getHeaders());
}
return buf.toString();
}
@@ -458,13 +493,13 @@
getJmsContentHeaderProperties().setHeaders(messageProperties);
}
- private void checkPropertyName(String propertyName)
+ private void checkPropertyName(CharSequence propertyName)
{
if (propertyName == null)
{
throw new IllegalArgumentException("Property name must not be null");
}
- else if ("".equals(propertyName))
+ else if (propertyName.length()==0)
{
throw new IllegalArgumentException("Property name must not be the empty string");
}
@@ -536,5 +571,12 @@
public void setConsumer(BasicMessageConsumer basicMessageConsumer)
{
_consumer = basicMessageConsumer;
+ }
+
+ public byte[] getBytesProperty(AMQShortString propertyName) throws JMSException
+ {
+ checkPropertyName(propertyName);
+ return getJmsContentHeaderProperties().getBytes(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?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- 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 Mon Jan 8 09:02:26 2007
@@ -43,16 +43,23 @@
List bodies) throws AMQException
{
ByteBuffer data;
+ final boolean debug = _logger.isDebugEnabled();
// we optimise the non-fragmented case to avoid copying
if (bodies != null && bodies.size() == 1)
{
- _logger.debug("Non-fragmented message body (bodySize=" + contentHeader.bodySize +")");
+ if(debug)
+ {
+ _logger.debug("Non-fragmented message body (bodySize=" + contentHeader.bodySize +")");
+ }
data = ((ContentBody)bodies.get(0)).payload;
}
else
{
- _logger.debug("Fragmented message body (" + bodies.size() + " frames, bodySize=" + contentHeader.bodySize + ")");
+ if(debug)
+ {
+ _logger.debug("Fragmented message body (" + bodies.size() + " frames, bodySize=" + contentHeader.bodySize + ")");
+ }
data = ByteBuffer.allocate((int)contentHeader.bodySize); // XXX: Is cast a problem?
final Iterator it = bodies.iterator();
while (it.hasNext())
@@ -63,7 +70,10 @@
}
data.flip();
}
- _logger.debug("Creating message from buffer with position=" + data.position() + " and remaining=" + data.remaining());
+ if(debug)
+ {
+ _logger.debug("Creating message from buffer with position=" + data.position() + " and remaining=" + data.remaining());
+ }
return createMessage(messageNbr, data, contentHeader);
}
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?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- 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 Mon Jan 8 09:02:26 2007
@@ -23,6 +23,7 @@
import org.apache.mina.common.ByteBuffer;
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.ContentHeaderBody;
+import org.apache.qpid.framing.AMQShortString;
import javax.jms.BytesMessage;
import javax.jms.JMSException;
@@ -35,9 +36,11 @@
public class JMSBytesMessage extends AbstractBytesMessage implements BytesMessage
{
- private static final String MIME_TYPE = "application/octet-stream";
+ public static final String MIME_TYPE = "application/octet-stream";
+ private static final AMQShortString MIME_TYPE_SHORT_STRING = new AMQShortString(MIME_TYPE);
- JMSBytesMessage()
+
+ public JMSBytesMessage()
{
this(null);
}
@@ -65,9 +68,9 @@
_readableMessage = true;
}
- public String getMimeType()
+ public AMQShortString getMimeTypeAsShortString()
{
- return MIME_TYPE;
+ return MIME_TYPE_SHORT_STRING;
}
public long getBodyLength() throws JMSException
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?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- 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 Mon Jan 8 09:02:26 2007
@@ -22,6 +22,7 @@
import org.apache.mina.common.ByteBuffer;
import org.apache.qpid.framing.ContentHeaderBody;
+import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.AMQException;
import org.apache.log4j.Logger;
@@ -37,10 +38,11 @@
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>();
- JMSMapMessage() throws JMSException
+ public JMSMapMessage() throws JMSException
{
this(null);
}
@@ -74,9 +76,9 @@
return _map.toString();
}
- public String getMimeType()
+ public AMQShortString getMimeTypeAsShortString()
{
- return MIME_TYPE;
+ return MIME_TYPE_SHORT_STRING;
}
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?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- 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 Mon Jan 8 09:02:26 2007
@@ -24,6 +24,7 @@
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.BasicContentHeaderProperties;
import org.apache.qpid.framing.ContentHeaderBody;
+import org.apache.qpid.framing.AMQShortString;
import javax.jms.JMSException;
import javax.jms.MessageFormatException;
@@ -34,14 +35,15 @@
public class JMSObjectMessage extends AbstractJMSMessage implements ObjectMessage
{
- static final String MIME_TYPE = "application/java-object-stream";
+ 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
*/
- JMSObjectMessage()
+ public JMSObjectMessage()
{
this(null);
}
@@ -54,7 +56,7 @@
_data = ByteBuffer.allocate(DEFAULT_BUFFER_SIZE);
_data.setAutoExpand(true);
}
- getJmsContentHeaderProperties().setContentType(MIME_TYPE);
+ getJmsContentHeaderProperties().setContentType(MIME_TYPE_SHORT_STRING);
}
/**
@@ -80,9 +82,9 @@
return toString(_data);
}
- public String getMimeType()
+ public AMQShortString getMimeTypeAsShortString()
{
- return MIME_TYPE;
+ return MIME_TYPE_SHORT_STRING;
}
public void setObject(Serializable serializable) throws JMSException
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?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- 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 Mon Jan 8 09:02:26 2007
@@ -23,6 +23,7 @@
import org.apache.mina.common.ByteBuffer;
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.ContentHeaderBody;
+import org.apache.qpid.framing.AMQShortString;
import javax.jms.*;
import java.nio.charset.CharacterCodingException;
@@ -34,6 +35,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);
/**
@@ -42,7 +44,7 @@
*/
private int _byteArrayRemaining = -1;
- JMSStreamMessage()
+ public JMSStreamMessage()
{
this(null);
}
@@ -71,9 +73,9 @@
_readableMessage = true;
}
- public String getMimeType()
+ public AMQShortString getMimeTypeAsShortString()
{
- return MIME_TYPE;
+ return MIME_TYPE_SHORT_STRING;
}
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?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- 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 Mon Jan 8 09:02:26 2007
@@ -21,6 +21,7 @@
package org.apache.qpid.client.message;
import org.apache.qpid.framing.BasicContentHeaderProperties;
+import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.AMQException;
import org.apache.mina.common.ByteBuffer;
@@ -32,15 +33,18 @@
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 String PAYLOAD_NULL_PROPERTY = "JMS_QPID_NULL";
+ private static final AMQShortString PAYLOAD_NULL_PROPERTY = new AMQShortString("JMS_QPID_NULL");
+ private static final Charset DEFAULT_CHARSET = Charset.defaultCharset();
- JMSTextMessage() throws JMSException
+ public JMSTextMessage() throws JMSException
{
this(null, null);
}
@@ -48,7 +52,7 @@
JMSTextMessage(ByteBuffer data, String encoding) throws JMSException
{
super(data); // this instantiates a content header
- getJmsContentHeaderProperties().setContentType(MIME_TYPE);
+ getJmsContentHeaderProperties().setContentType(MIME_TYPE_SHORT_STRING);
getJmsContentHeaderProperties().setEncoding(encoding);
}
@@ -56,7 +60,7 @@
throws AMQException
{
super(deliveryTag, contentHeader, data);
- contentHeader.setContentType(MIME_TYPE);
+ contentHeader.setContentType(MIME_TYPE_SHORT_STRING);
_data = data;
}
@@ -91,9 +95,9 @@
_data = data;
}
- public String getMimeType()
+ public AMQShortString getMimeTypeAsShortString()
{
- return MIME_TYPE;
+ return MIME_TYPE_SHORT_STRING;
}
public void setText(String text) throws JMSException
@@ -109,13 +113,14 @@
_data.limit(text.length()) ;
//_data.sweep();
_data.setAutoExpand(true);
- if (getJmsContentHeaderProperties().getEncoding() == null)
+ final String encoding = getJmsContentHeaderProperties().getEncoding();
+ if (encoding == null)
{
_data.put(text.getBytes());
}
else
{
- _data.put(text.getBytes(getJmsContentHeaderProperties().getEncoding()));
+ _data.put(text.getBytes(encoding));
}
_changedData=true;
}
@@ -164,7 +169,7 @@
{
try
{
- _decodedValue = _data.getString(Charset.defaultCharset().newDecoder());
+ _decodedValue = _data.getString(DEFAULT_CHARSET.newDecoder());
}
catch (CharacterCodingException e)
{
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageFactoryRegistry.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageFactoryRegistry.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageFactoryRegistry.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/MessageFactoryRegistry.java Mon Jan 8 09:02:26 2007
@@ -23,6 +23,7 @@
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.BasicContentHeaderProperties;
import org.apache.qpid.framing.ContentHeaderBody;
+import org.apache.qpid.framing.AMQShortString;
import javax.jms.JMSException;
import java.util.HashMap;
@@ -31,7 +32,8 @@
public class MessageFactoryRegistry
{
- private final Map _mimeToFactoryMap = new HashMap();
+ private final Map<String, MessageFactory> _mimeStringToFactoryMap = new HashMap<String, MessageFactory>();
+ private final Map<AMQShortString, MessageFactory> _mimeShortStringToFactoryMap = new HashMap<AMQShortString, MessageFactory>();
public void registerFactory(String mimeType, MessageFactory mf)
{
@@ -39,12 +41,14 @@
{
throw new IllegalArgumentException("Message factory must not be null");
}
- _mimeToFactoryMap.put(mimeType, mf);
+ _mimeStringToFactoryMap.put(mimeType, mf);
+ _mimeShortStringToFactoryMap.put(new AMQShortString(mimeType), mf);
}
public MessageFactory deregisterFactory(String mimeType)
{
- return (MessageFactory) _mimeToFactoryMap.remove(mimeType);
+ _mimeShortStringToFactoryMap.remove(new AMQShortString(mimeType));
+ return _mimeStringToFactoryMap.remove(mimeType);
}
/**
@@ -63,7 +67,7 @@
List bodies) throws AMQException, JMSException
{
BasicContentHeaderProperties properties = (BasicContentHeaderProperties) contentHeader.properties;
- MessageFactory mf = (MessageFactory) _mimeToFactoryMap.get(properties.getContentType());
+ MessageFactory mf = _mimeShortStringToFactoryMap.get(properties.getContentTypeShortString());
if (mf == null)
{
throw new AMQException("Unsupport MIME type of " + properties.getContentType());
@@ -80,7 +84,7 @@
{
throw new IllegalArgumentException("Mime type must not be null");
}
- MessageFactory mf = (MessageFactory) _mimeToFactoryMap.get(mimeType);
+ MessageFactory mf = _mimeStringToFactoryMap.get(mimeType);
if (mf == null)
{
throw new AMQException("Unsupport MIME type of " + mimeType);
@@ -101,7 +105,7 @@
mf.registerFactory(JMSMapMessage.MIME_TYPE, new JMSMapMessageFactory());
mf.registerFactory("text/plain", new JMSTextMessageFactory());
mf.registerFactory("text/xml", new JMSTextMessageFactory());
- mf.registerFactory("application/octet-stream", new JMSBytesMessageFactory());
+ mf.registerFactory(JMSBytesMessage.MIME_TYPE, new JMSBytesMessageFactory());
mf.registerFactory(JMSObjectMessage.MIME_TYPE, new JMSObjectMessageFactory());
mf.registerFactory(JMSStreamMessage.MIME_TYPE, new JMSStreamMessageFactory());
mf.registerFactory(null, new JMSBytesMessageFactory());
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java Mon Jan 8 09:02:26 2007
@@ -37,14 +37,7 @@
import org.apache.qpid.client.state.AMQStateManager;
import org.apache.qpid.client.state.listener.SpecificMethodFrameListener;
import org.apache.qpid.codec.AMQCodecFactory;
-import org.apache.qpid.framing.AMQDataBlock;
-import org.apache.qpid.framing.AMQFrame;
-import org.apache.qpid.framing.AMQMethodBody;
-import org.apache.qpid.framing.ConnectionCloseBody;
-import org.apache.qpid.framing.ConnectionCloseOkBody;
-import org.apache.qpid.framing.ContentBody;
-import org.apache.qpid.framing.ContentHeaderBody;
-import org.apache.qpid.framing.HeartbeatBody;
+import org.apache.qpid.framing.*;
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.ssl.BogusSSLContextFactory;
@@ -99,7 +92,7 @@
// We add a proxy for the state manager so that we can substitute the state manager easily in this class.
// We substitute the state manager when performing failover
- _frameListeners.add(new AMQMethodListener()
+/* _frameListeners.add(new AMQMethodListener()
{
public boolean methodReceived(AMQMethodEvent evt) throws AMQException
{
@@ -110,7 +103,7 @@
{
_stateManager.error(e);
}
- });
+ });*/
}
public boolean isUseSSL()
@@ -284,11 +277,14 @@
public void propagateExceptionToWaiters(Exception e)
{
_stateManager.error(e);
- final Iterator it = _frameListeners.iterator();
- while (it.hasNext())
+ if(!_frameListeners.isEmpty())
{
- final AMQMethodListener ml = (AMQMethodListener) it.next();
- ml.error(e);
+ final Iterator it = _frameListeners.iterator();
+ while (it.hasNext())
+ {
+ final AMQMethodListener ml = (AMQMethodListener) it.next();
+ ml.error(e);
+ }
}
}
@@ -296,12 +292,13 @@
public void messageReceived(IoSession session, Object message) throws Exception
{
+ final long msgNumber = ++_messageReceivedCount;
- if (_messageReceivedCount++ % 1000 == 0)
+ if (_logger.isDebugEnabled() && (msgNumber % 1000 == 0))
{
_logger.debug("Received " + _messageReceivedCount + " protocol messages");
}
- Iterator it = _frameListeners.iterator();
+
AMQFrame frame = (AMQFrame) message;
HeartbeatDiagnostics.received(frame.bodyFrame instanceof HeartbeatBody);
@@ -314,13 +311,19 @@
}
final AMQMethodEvent evt = new AMQMethodEvent(frame.channel, (AMQMethodBody) frame.bodyFrame, _protocolSession);
+
try
{
- boolean wasAnyoneInterested = false;
- while (it.hasNext())
+
+ boolean wasAnyoneInterested = _stateManager.methodReceived(evt);
+ if(!_frameListeners.isEmpty())
{
- final AMQMethodListener listener = (AMQMethodListener) it.next();
- wasAnyoneInterested = listener.methodReceived(evt) || wasAnyoneInterested;
+ Iterator it = _frameListeners.iterator();
+ while (it.hasNext())
+ {
+ final AMQMethodListener listener = (AMQMethodListener) it.next();
+ wasAnyoneInterested = listener.methodReceived(evt) || wasAnyoneInterested;
+ }
}
if (!wasAnyoneInterested)
{
@@ -329,11 +332,15 @@
}
catch (AMQException e)
{
- it = _frameListeners.iterator();
- while (it.hasNext())
+ _stateManager.error(e);
+ if(!_frameListeners.isEmpty())
{
- final AMQMethodListener listener = (AMQMethodListener) it.next();
- listener.error(e);
+ Iterator it = _frameListeners.iterator();
+ while (it.hasNext())
+ {
+ final AMQMethodListener listener = (AMQMethodListener) it.next();
+ listener.error(e);
+ }
}
exceptionCaught(session, e);
}
@@ -359,17 +366,21 @@
public void messageSent(IoSession session, Object message) throws Exception
{
- if (_messagesOut++ % 1000 == 0)
+ final long sentMessages = _messagesOut++;
+
+ final boolean debug = _logger.isDebugEnabled();
+
+ if (debug && (sentMessages % 1000 == 0))
{
_logger.debug("Sent " + _messagesOut + " protocol messages");
}
_connection.bytesSent(session.getWrittenBytes());
- if (_logger.isDebugEnabled())
+ if (debug)
{
_logger.debug("Sent frame " + message);
}
}
-
+/*
public void addFrameListener(AMQMethodListener listener)
{
_frameListeners.add(listener);
@@ -379,7 +390,7 @@
{
_frameListeners.remove(listener);
}
-
+ */
public void attainState(AMQState s) throws AMQException
{
_stateManager.attainState(s);
@@ -423,9 +434,13 @@
// When control resumes before this line, a reply will have been received
// that matches the criteria defined in the blocking listener
}
+ catch (AMQException e)
+ {
+ throw e;
+ }
finally
{
- // If we don't remove the listener then no-one will
+ // If we don't removeKey the listener then no-one will
_frameListeners.remove(listener);
}
@@ -480,7 +495,7 @@
0, // classId
0, // methodId
AMQConstant.REPLY_SUCCESS.getCode(), // replyCode
- "JMS client is closing the connection."); // replyText
+ new AMQShortString("JMS client is closing the connection.")); // replyText
syncWrite(frame, ConnectionCloseOkBody.class);
_protocolSession.closeProtocolSession();
@@ -518,7 +533,7 @@
}
}
- public String generateQueueName()
+ public AMQShortString generateQueueName()
{
return _protocolSession.generateQueueName();
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java Mon Jan 8 09:02:26 2007
@@ -31,11 +31,7 @@
import org.apache.qpid.client.ConnectionTuneParameters;
import org.apache.qpid.client.message.UnexpectedBodyReceivedException;
import org.apache.qpid.client.message.UnprocessedMessage;
-import org.apache.qpid.framing.AMQDataBlock;
-import org.apache.qpid.framing.ContentBody;
-import org.apache.qpid.framing.ContentHeaderBody;
-import org.apache.qpid.framing.ProtocolInitiation;
-import org.apache.qpid.framing.ProtocolVersionList;
+import org.apache.qpid.framing.*;
import org.apache.commons.lang.StringUtils;
import javax.jms.JMSException;
@@ -381,7 +377,7 @@
_protocolHandler.failover(host, port);
}
- protected String generateQueueName()
+ protected AMQShortString generateQueueName()
{
int id;
synchronized(_queueIdLock)
@@ -390,7 +386,7 @@
}
//get rid of / and : and ; from address for spec conformance
String localAddress = StringUtils.replaceChars(_minaProtocolSession.getLocalAddress().toString(),"/;:","");
- return "tmp_" + localAddress + "_" + id;
+ return new AMQShortString("tmp_" + localAddress + "_" + id);
}
/**
@@ -407,7 +403,7 @@
}
}
- public void confirmConsumerCancelled(int channelId, String consumerTag)
+ public void confirmConsumerCancelled(int channelId, AMQShortString consumerTag)
{
final Integer chId = channelId;
final AMQSession session = (AMQSession) _channelId2SessionMap.get(chId);
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/security/amqplain/AmqPlainSaslClient.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/security/amqplain/AmqPlainSaslClient.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/security/amqplain/AmqPlainSaslClient.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/security/amqplain/AmqPlainSaslClient.java Mon Jan 8 09:02:26 2007
@@ -73,8 +73,8 @@
throw new SaslException("Error handling SASL callbacks: " + e, e);
}
FieldTable table = FieldTableFactory.newFieldTable();
- table.put("LOGIN", nameCallback.getName());
- table.put("PASSWORD", pwdCallback.getPassword());
+ table.setString("LOGIN", nameCallback.getName());
+ table.setString("PASSWORD", new String(pwdCallback.getPassword()));
return table.getDataAsBytes();
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jndi/PropertiesFileInitialContextFactory.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jndi/PropertiesFileInitialContextFactory.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jndi/PropertiesFileInitialContextFactory.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jndi/PropertiesFileInitialContextFactory.java Mon Jan 8 09:02:26 2007
@@ -29,6 +29,7 @@
import org.apache.qpid.url.AMQBindingURL;
import org.apache.qpid.url.BindingURL;
import org.apache.qpid.url.URLSyntaxException;
+import org.apache.qpid.framing.AMQShortString;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
@@ -232,10 +233,14 @@
*/
protected Queue createQueue(Object value)
{
- if (value instanceof String)
+ if(value instanceof AMQShortString)
+ {
+ return new AMQQueue((AMQShortString) value);
+ }
+ else if (value instanceof String)
{
- return new AMQQueue((String) value);
+ return new AMQQueue(new AMQShortString((String) value));
}
else if (value instanceof BindingURL)
@@ -251,9 +256,13 @@
*/
protected Topic createTopic(Object value)
{
- if (value instanceof String)
+ if(value instanceof AMQShortString)
+ {
+ return new AMQTopic((AMQShortString)value);
+ }
+ else if (value instanceof String)
{
- return new AMQTopic((String) value);
+ return new AMQTopic(new AMQShortString((String) value));
}
else if (value instanceof BindingURL)
Modified: incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/ack/RecoverTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/ack/RecoverTest.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/ack/RecoverTest.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/ack/RecoverTest.java Mon Jan 8 09:02:26 2007
@@ -25,6 +25,8 @@
import org.apache.qpid.client.AMQQueue;
import org.apache.qpid.client.AMQSession;
import org.apache.qpid.client.transport.TransportConnection;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.exchange.ExchangeDefaults;
import javax.jms.*;
@@ -50,10 +52,10 @@
Connection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "/test");
Session consumerSession = con.createSession(false, Session.CLIENT_ACKNOWLEDGE);
- Queue queue = new AMQQueue("someQ", "someQ", false, true);
+ Queue queue = new AMQQueue(new AMQShortString("someQ"), new AMQShortString("someQ"), false, true);
MessageConsumer consumer = consumerSession.createConsumer(queue);
//force synch to ensure the consumer has resulted in a bound queue
- ((AMQSession) consumerSession).declareExchangeSynch("amq.direct", "direct");
+ ((AMQSession) consumerSession).declareExchangeSynch(ExchangeDefaults.DIRECT_EXCHANGE_NAME, ExchangeDefaults.DIRECT_EXCHANGE_NAME);
Connection con2 = new AMQConnection("vm://:1", "guest", "guest", "producer1", "/test");
Session producerSession = con2.createSession(false, Session.CLIENT_ACKNOWLEDGE);
@@ -107,10 +109,10 @@
Connection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "/test");
Session consumerSession = con.createSession(false, Session.CLIENT_ACKNOWLEDGE);
- Queue queue = new AMQQueue("someQ", "someQ", false, true);
+ Queue queue = new AMQQueue(new AMQShortString("someQ"), new AMQShortString("someQ"), false, true);
MessageConsumer consumer = consumerSession.createConsumer(queue);
//force synch to ensure the consumer has resulted in a bound queue
- ((AMQSession) consumerSession).declareExchangeSynch("amq.direct", "direct");
+ ((AMQSession) consumerSession).declareExchangeSynch(ExchangeDefaults.DIRECT_EXCHANGE_NAME, ExchangeDefaults.DIRECT_EXCHANGE_NAME);
Connection con2 = new AMQConnection("vm://:1", "guest", "guest", "producer1", "/test");
Session producerSession = con2.createSession(false, Session.CLIENT_ACKNOWLEDGE);
@@ -171,8 +173,8 @@
Connection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "/test");
Session consumerSession = con.createSession(false, Session.CLIENT_ACKNOWLEDGE);
- Queue queue = new AMQQueue("Q1", "Q1", false, true);
- Queue queue2 = new AMQQueue("Q2", "Q2", false, true);
+ Queue queue = new AMQQueue(new AMQShortString("Q1"), new AMQShortString("Q1"), false, true);
+ Queue queue2 = new AMQQueue(new AMQShortString("Q2"), new AMQShortString("Q2"), false, true);
MessageConsumer consumer = consumerSession.createConsumer(queue);
MessageConsumer consumer2 = consumerSession.createConsumer(queue2);
@@ -210,7 +212,7 @@
Connection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "/test");
final Session consumerSession = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
- Queue queue = new AMQQueue("Q1", "Q1", false, true);
+ Queue queue = new AMQQueue(new AMQShortString("Q1"), new AMQShortString("Q1"), false, true);
MessageProducer producer = consumerSession.createProducer(queue);
producer.send(consumerSession.createTextMessage("hello"));
MessageConsumer consumer = consumerSession.createConsumer(queue);
Modified: incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/FieldTableKeyEnumeratorTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/FieldTableKeyEnumeratorTest.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/FieldTableKeyEnumeratorTest.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/FieldTableKeyEnumeratorTest.java Mon Jan 8 09:02:26 2007
@@ -35,16 +35,20 @@
public class FieldTableKeyEnumeratorTest extends TestCase
{
+ public void testTrue()
+ {
+
+ }
public void testKeyEnumeration()
{
FieldTable result = FieldTableFactory.newFieldTable();
- result.put("one", 1L);
- result.put("two", 2L);
- result.put("three", 3L);
- result.put("four", 4L);
- result.put("five", 5L);
+ result.setObject("one", 1L);
+ result.setObject("two", 2L);
+ result.setObject("three", 3L);
+ result.setObject("four", 4L);
+ result.setObject("five", 5L);
- Iterator iterator = result.keySet().iterator();
+ Iterator iterator = result.keys().iterator();
try
{
Modified: incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/FieldTableMessageTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/FieldTableMessageTest.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/FieldTableMessageTest.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/FieldTableMessageTest.java Mon Jan 8 09:02:26 2007
@@ -85,11 +85,11 @@
private FieldTable load() throws IOException
{
FieldTable result = FieldTableFactory.newFieldTable();
- result.put("one", 1L);
- result.put("two", 2L);
- result.put("three", 3L);
- result.put("four", 4L);
- result.put("five", 5L);
+ result.setLong("one", 1L);
+ result.setLong("two", 2L);
+ result.setLong("three", 3L);
+ result.setLong("four", 4L);
+ result.setLong("five", 5L);
return result;
}
@@ -133,10 +133,9 @@
{
ByteBuffer buffer = ((JMSBytesMessage) m).getData();
FieldTable actual = FieldTableFactory.newFieldTable(buffer, buffer.remaining());
- for (Object o : _expected.keySet())
- {
- String key = (String) o;
- assertEquals("Values for " + key + " did not match", _expected.get(key), actual.get(key));
+ for (String key : _expected.keys())
+ {
+ assertEquals("Values for " + key + " did not match", _expected.getObject(key), actual.getObject(key));
}
}
}
Modified: incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/PropertyValueTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/PropertyValueTest.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/PropertyValueTest.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/PropertyValueTest.java Mon Jan 8 09:02:26 2007
@@ -29,12 +29,7 @@
import org.apache.qpid.client.message.JMSTextMessage;
import org.apache.qpid.testutil.VMBrokerSetup;
-import javax.jms.Destination;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageListener;
-import javax.jms.MessageProducer;
-import javax.jms.Queue;
+import javax.jms.*;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
@@ -81,7 +76,7 @@
{
_connection = connection;
_destination = destination;
- _session = (AMQSession) connection.createSession(false, AMQSession.AUTO_ACKNOWLEDGE);
+ _session = (AMQSession) connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//set up a slow consumer
_session.createConsumer(destination).setMessageListener(this);
Modified: incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/TextMessageTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/TextMessageTest.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/TextMessageTest.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/TextMessageTest.java Mon Jan 8 09:02:26 2007
@@ -75,7 +75,7 @@
{
_connection = connection;
_destination = destination;
- _session = (AMQSession) connection.createSession(false, AMQSession.AUTO_ACKNOWLEDGE);
+ _session = (AMQSession) connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//set up a slow consumer
_session.createConsumer(destination).setMessageListener(this);
Modified: incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/SpecialQueue.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/SpecialQueue.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/SpecialQueue.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/SpecialQueue.java Mon Jan 8 09:02:26 2007
@@ -21,6 +21,7 @@
package org.apache.qpid.test.unit.client.forwardall;
import org.apache.qpid.client.AMQQueue;
+import org.apache.qpid.framing.AMQShortString;
/**
* Queue that allows several private queues to be registered and bound
@@ -29,15 +30,19 @@
*/
class SpecialQueue extends AMQQueue
{
- private final String name;
+ private final AMQShortString name;
SpecialQueue(String name)
{
+ this(new AMQShortString(name));
+ }
+ SpecialQueue(AMQShortString name)
+ {
super(name, true);
this.name = name;
}
- public String getRoutingKey()
+ public AMQShortString getRoutingKey()
{
return name;
}
Modified: incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/protocol/AMQProtocolSessionTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/protocol/AMQProtocolSessionTest.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/protocol/AMQProtocolSessionTest.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/protocol/AMQProtocolSessionTest.java Mon Jan 8 09:02:26 2007
@@ -23,6 +23,7 @@
import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.client.protocol.AMQProtocolHandler;
import org.apache.qpid.client.protocol.AMQProtocolSession;
+import org.apache.qpid.framing.AMQShortString;
import org.apache.mina.common.IoSession;
import junit.framework.TestCase;
@@ -45,7 +46,7 @@
return (TestIoSession) _minaProtocolSession;
}
- public String genQueueName()
+ public AMQShortString genQueueName()
{
return generateQueueName();
}
@@ -80,26 +81,26 @@
public void testGenerateQueueName()
{
- String testAddress;
+ AMQShortString testAddress;
- //test address with / and ; chars which generateQueueName should remove
+ //test address with / and ; chars which generateQueueName should removeKey
_testSession.getMinaProtocolSession().setStringLocalAddress(_brokenAddress);
_testSession.getMinaProtocolSession().setLocalPort(_port);
testAddress = _testSession.genQueueName();
- assertEquals("Failure when generating a queue exchange from an address with special chars",_generatedAddress,testAddress);
+ assertEquals("Failure when generating a queue exchange from an address with special chars",_generatedAddress,testAddress.toString());
//test empty address
_testSession.getMinaProtocolSession().setStringLocalAddress(_emptyAddress);
testAddress = _testSession.genQueueName();
- assertEquals("Failure when generating a queue exchange from an empty address",_generatedAddress_2,testAddress);
+ assertEquals("Failure when generating a queue exchange from an empty address",_generatedAddress_2,testAddress.toString());
//test address with no special chars
_testSession.getMinaProtocolSession().setStringLocalAddress(_validAddress);
testAddress = _testSession.genQueueName();
- assertEquals("Failure when generating a queue exchange from an address with no special chars",_generatedAddress_3,testAddress);
+ assertEquals("Failure when generating a queue exchange from an address with no special chars",_generatedAddress_3,testAddress.toString());
}
Modified: incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/JMSDestinationTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/JMSDestinationTest.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/JMSDestinationTest.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/JMSDestinationTest.java Mon Jan 8 09:02:26 2007
@@ -11,6 +11,7 @@
import org.apache.qpid.url.BindingURL;
import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.framing.AMQShortString;
import javax.jms.*;
@@ -41,7 +42,7 @@
{
Connection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "/test");
AMQSession consumerSession = (AMQSession) con.createSession(false, Session.CLIENT_ACKNOWLEDGE);
- Queue queue = new AMQQueue("someQ", "someQ", false, true);
+ Queue queue = new AMQQueue(new AMQShortString("someQ"), new AMQShortString("someQ"), false, true);
MessageConsumer consumer = consumerSession.createConsumer(queue);
Connection con2 = new AMQConnection("vm://:1", "guest", "guest", "producer1", "/test");
Modified: incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java Mon Jan 8 09:02:26 2007
@@ -170,7 +170,7 @@
con.start();
TextMessage tm = session1.createTextMessage("Hello");
publisher.publish(tm);
- tm = (TextMessage) consumer1.receive(2000);
+ tm = (TextMessage) consumer1.receive(200000L);
assertNotNull(tm);
String msgText = tm.getText();
assertEquals("Hello", msgText);
@@ -178,7 +178,7 @@
msgText = tm.getText();
assertNull(msgText);
publisher.publish(tm);
- tm = (TextMessage) consumer1.receive(2000);
+ tm = (TextMessage) consumer1.receive(20000000L);
assertNotNull(tm);
msgText = tm.getText();
assertNull(msgText);
Modified: incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/ClientHandlerRegistry.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/ClientHandlerRegistry.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/ClientHandlerRegistry.java (original)
+++ incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/ClientHandlerRegistry.java Mon Jan 8 09:02:26 2007
@@ -29,13 +29,7 @@
import org.apache.qpid.client.state.AMQStateManager;
import org.apache.qpid.client.state.IllegalStateTransitionException;
import org.apache.qpid.client.state.StateAwareMethodListener;
-import org.apache.qpid.framing.AMQFrame;
-import org.apache.qpid.framing.AMQMethodBody;
-import org.apache.qpid.framing.ConnectionCloseBody;
-import org.apache.qpid.framing.ConnectionOpenOkBody;
-import org.apache.qpid.framing.ConnectionSecureBody;
-import org.apache.qpid.framing.ConnectionStartBody;
-import org.apache.qpid.framing.ConnectionTuneBody;
+import org.apache.qpid.framing.*;
import java.util.HashMap;
import java.util.Map;
@@ -127,9 +121,9 @@
class ConnectionTuneHandler extends ConnectionTuneMethodHandler
{
- protected AMQFrame createConnectionOpenFrame(int channel, String path, String capabilities, boolean insist)
+ protected AMQFrame createConnectionOpenFrame(int channel, AMQShortString path, AMQShortString capabilities, boolean insist)
{
- return super.createConnectionOpenFrame(channel, path, ClusterCapability.add(capabilities, _identity), insist);
+ return super.createConnectionOpenFrame(channel, path, new AMQShortString(ClusterCapability.add(capabilities, _identity)), insist);
}
}
}
Modified: incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/ClusterCapability.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/ClusterCapability.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/ClusterCapability.java (original)
+++ incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/ClusterCapability.java Mon Jan 8 09:02:26 2007
@@ -20,6 +20,8 @@
*/
package org.apache.qpid.server.cluster;
+import org.apache.qpid.framing.AMQShortString;
+
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -28,22 +30,22 @@
public static final String PATTERN = ".*\\bcluster_peer=(\\S*:\\d*)\b*.*";
public static final String PEER = "cluster_peer";
- public static String add(String original, MemberHandle identity)
+ public static AMQShortString add(AMQShortString original, MemberHandle identity)
{
- return original == null ? peer(identity) : original + " " + peer(identity);
+ return original == null ? peer(identity) : new AMQShortString(original + " " + peer(identity));
}
- private static String peer(MemberHandle identity)
+ private static AMQShortString peer(MemberHandle identity)
{
- return PEER + "=" + identity.getDetails();
+ return new AMQShortString(PEER + "=" + identity.getDetails());
}
- public static boolean contains(String in)
+ public static boolean contains(AMQShortString in)
{
- return in != null && in.contains(in);
+ return in != null; // && in.contains(in);
}
- public static MemberHandle getPeer(String in)
+ public static MemberHandle getPeer(AMQShortString in)
{
Matcher matcher = Pattern.compile(PATTERN).matcher(in);
if (matcher.matches())
Modified: incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/DefaultGroupManager.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/DefaultGroupManager.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/DefaultGroupManager.java (original)
+++ incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/DefaultGroupManager.java Mon Jan 8 09:02:26 2007
@@ -109,7 +109,7 @@
// AMQP version change: Hardwire the version to 0-8 (major=8, minor=0)
// TODO: Connect this to the session version obtained from ProtocolInitiation for this session.
ClusterPingBody ping = new ClusterPingBody((byte)8, (byte)0);
- ping.broker = _group.getLocal().getDetails();
+ ping.broker = new AMQShortString(_group.getLocal().getDetails());
ping.responseRequired = true;
ping.load = _loadTable.getLocalLoad();
BlockingHandler handler = new BlockingHandler();
@@ -157,7 +157,7 @@
// AMQP version change: Hardwire the version to 0-8 (major=8, minor=0)
// TODO: Connect this to the session version obtained from ProtocolInitiation for this session.
ClusterJoinBody join = new ClusterJoinBody((byte)8, (byte)0);
- join.broker = _group.getLocal().getDetails();
+ join.broker = new AMQShortString(_group.getLocal().getDetails());
send(leader, new SimpleBodySendable(join));
}
@@ -178,7 +178,7 @@
// AMQP version change: Hardwire the version to 0-8 (major=8, minor=0)
// TODO: Connect this to the session version obtained from ProtocolInitiation for this session.
ClusterLeaveBody leave = new ClusterLeaveBody((byte)8, (byte)0);
- leave.broker = _group.getLocal().getDetails();
+ leave.broker = new AMQShortString(_group.getLocal().getDetails());
send(getLeader(), new SimpleBodySendable(leave));
}
@@ -201,7 +201,7 @@
// AMQP version change: Hardwire the version to 0-8 (major=8, minor=0)
// TODO: Connect this to the session version obtained from ProtocolInitiation for this session.
ClusterSuspectBody suspect = new ClusterSuspectBody((byte)8, (byte)0);
- suspect.broker = broker.getDetails();
+ suspect.broker = new AMQShortString(broker.getDetails());
send(getLeader(), new SimpleBodySendable(suspect));
}
}
@@ -225,7 +225,7 @@
// AMQP version change: Hardwire the version to 0-8 (major=8, minor=0)
// TODO: Connect this to the session version obtained from ProtocolInitiation for this session.
ClusterJoinBody request = new ClusterJoinBody((byte)8, (byte)0);
- request.broker = member.getDetails();
+ request.broker = new AMQShortString(member.getDetails());
Broker leader = getLeader();
send(leader, new SimpleBodySendable(request));
_logger.info(new LogMessage("Passed join request for {0} to {1}", member, leader));
Modified: incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/MemberHandle.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/MemberHandle.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/MemberHandle.java (original)
+++ incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/MemberHandle.java Mon Jan 8 09:02:26 2007
@@ -20,6 +20,8 @@
*/
package org.apache.qpid.server.cluster;
+import org.apache.qpid.framing.AMQShortString;
+
public interface MemberHandle
{
public String getHost();
@@ -30,5 +32,5 @@
public boolean matches(String host, int port);
- public String getDetails();
+ public AMQShortString getDetails();
}
Modified: incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/MinaBrokerProxy.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/MinaBrokerProxy.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/MinaBrokerProxy.java (original)
+++ incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/MinaBrokerProxy.java Mon Jan 8 09:02:26 2007
@@ -180,7 +180,7 @@
{
//signal redirection to waiting thread
ConnectionRedirectBody redirect = (ConnectionRedirectBody) method;
- String[] parts = redirect.host.split(":");
+ String[] parts = redirect.host.toString().split(":");
_connectionMonitor.redirect(parts[0], Integer.parseInt(parts[1]));
}
else
Modified: incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/SimpleMemberHandle.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/SimpleMemberHandle.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/SimpleMemberHandle.java (original)
+++ incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/SimpleMemberHandle.java Mon Jan 8 09:02:26 2007
@@ -20,6 +20,8 @@
*/
package org.apache.qpid.server.cluster;
+import org.apache.qpid.framing.AMQShortString;
+
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
@@ -37,6 +39,11 @@
_port = port;
}
+ public SimpleMemberHandle(AMQShortString details)
+ {
+ this(details.toString());
+ }
+
public SimpleMemberHandle(String details)
{
String[] parts = details.split(":");
@@ -84,14 +91,14 @@
return _host.equals(host) && _port == port;
}
- public String getDetails()
+ public AMQShortString getDetails()
{
- return _host + ":" + _port;
+ return new AMQShortString(_host + ":" + _port);
}
public String toString()
{
- return getDetails();
+ return getDetails().toString();
}
static List<MemberHandle> stringToMembers(String membership)
Modified: incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/ChannelQueueManager.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/ChannelQueueManager.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/ChannelQueueManager.java (original)
+++ incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/ChannelQueueManager.java Mon Jan 8 09:02:26 2007
@@ -28,11 +28,7 @@
import org.apache.qpid.server.protocol.AMQMethodEvent;
import org.apache.qpid.server.cluster.util.LogMessage;
import org.apache.qpid.AMQException;
-import org.apache.qpid.framing.AMQMethodBody;
-import org.apache.qpid.framing.QueueDeclareBody;
-import org.apache.qpid.framing.QueueBindBody;
-import org.apache.qpid.framing.QueueDeleteBody;
-import org.apache.qpid.framing.BasicConsumeBody;
+import org.apache.qpid.framing.*;
import org.apache.log4j.Logger;
import java.util.Map;
@@ -46,7 +42,7 @@
class ChannelQueueManager
{
private static final Logger _logger = Logger.getLogger(ChannelQueueManager.class);
- private final Map<Integer, String> _channelQueues = new HashMap<Integer, String>();
+ private final Map<Integer, AMQShortString> _channelQueues = new HashMap<Integer, AMQShortString>();
ClusterMethodHandler<QueueDeclareBody> createQueueDeclareHandler()
{
@@ -68,15 +64,15 @@
return new BasicConsumeHandler();
}
- private void set(int channel, String queue)
+ private void set(int channel, AMQShortString queue)
{
_channelQueues.put(channel, queue);
_logger.info(new LogMessage("Set default queue for {0} to {1}", channel, queue));
}
- private String get(int channel)
+ private AMQShortString get(int channel)
{
- String queue = _channelQueues.get(channel);
+ AMQShortString queue = _channelQueues.get(channel);
_logger.info(new LogMessage("Default queue for {0} is {1}", channel, queue));
return queue;
}
Modified: incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/ClusterMethodHandlerFactory.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/ClusterMethodHandlerFactory.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/ClusterMethodHandlerFactory.java (original)
+++ incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/ClusterMethodHandlerFactory.java Mon Jan 8 09:02:26 2007
@@ -21,34 +21,7 @@
package org.apache.qpid.server.cluster.handler;
import org.apache.qpid.AMQException;
-import org.apache.qpid.framing.AMQFrame;
-import org.apache.qpid.framing.AMQMethodBody;
-import org.apache.qpid.framing.BasicCancelBody;
-import org.apache.qpid.framing.BasicConsumeBody;
-import org.apache.qpid.framing.BasicPublishBody;
-import org.apache.qpid.framing.ChannelCloseBody;
-import org.apache.qpid.framing.ChannelFlowBody;
-import org.apache.qpid.framing.ChannelOpenBody;
-import org.apache.qpid.framing.ClusterJoinBody;
-import org.apache.qpid.framing.ClusterLeaveBody;
-import org.apache.qpid.framing.ClusterMembershipBody;
-import org.apache.qpid.framing.ClusterPingBody;
-import org.apache.qpid.framing.ClusterSuspectBody;
-import org.apache.qpid.framing.ConnectionCloseBody;
-import org.apache.qpid.framing.ConnectionOpenBody;
-import org.apache.qpid.framing.ConnectionSecureOkBody;
-import org.apache.qpid.framing.ConnectionStartOkBody;
-import org.apache.qpid.framing.ConnectionTuneOkBody;
-import org.apache.qpid.framing.ExchangeDeclareBody;
-import org.apache.qpid.framing.ExchangeDeleteBody;
-import org.apache.qpid.framing.QueueBindBody;
-import org.apache.qpid.framing.QueueDeclareBody;
-import org.apache.qpid.framing.QueueDeleteBody;
-import org.apache.qpid.framing.ClusterSynchBody;
-import org.apache.qpid.framing.BasicQosBody;
-import org.apache.qpid.framing.TxSelectBody;
-import org.apache.qpid.framing.TxCommitBody;
-import org.apache.qpid.framing.TxRollbackBody;
+import org.apache.qpid.framing.*;
import org.apache.qpid.server.cluster.ClusterCapability;
import org.apache.qpid.server.cluster.ClusteredProtocolSession;
import org.apache.qpid.server.cluster.GroupManager;
@@ -236,7 +209,7 @@
void postHandle(AMQStateManager stateMgr, AMQProtocolSession session, AMQMethodEvent<ConnectionOpenBody> evt)
{
- String capabilities = evt.getMethod().capabilities;
+ AMQShortString capabilities = evt.getMethod().capabilities;
if (ClusterCapability.contains(capabilities))
{
ClusteredProtocolSession.setSessionPeer(session, ClusterCapability.getPeer(capabilities));
Modified: incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/LocalQueueDeclareHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/LocalQueueDeclareHandler.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/LocalQueueDeclareHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/LocalQueueDeclareHandler.java Mon Jan 8 09:02:26 2007
@@ -23,6 +23,7 @@
import org.apache.log4j.Logger;
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.QueueDeclareBody;
+import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.server.cluster.ClusteredProtocolSession;
import org.apache.qpid.server.cluster.GroupManager;
import org.apache.qpid.server.cluster.util.LogMessage;
@@ -45,9 +46,9 @@
_groupMgr = groupMgr;
}
- protected String createName()
+ protected AMQShortString createName()
{
- return super.createName() + "@" + _groupMgr.getLocal().getDetails();
+ return new AMQShortString(super.createName().toString() + "@" + _groupMgr.getLocal().getDetails());
}
protected AMQQueue createQueue(QueueDeclareBody body, QueueRegistry registry, AMQProtocolSession session) throws AMQException
@@ -60,7 +61,7 @@
//need to get peer from the session...
MemberHandle peer = ClusteredProtocolSession.getSessionPeer(session);
_logger.debug(new LogMessage("Creating proxied queue {0} on behalf of {1}", body.queue, peer));
- return new RemoteQueueProxy(peer, _groupMgr, body.queue, body.durable, peer.getDetails(), body.autoDelete, registry);
+ return new RemoteQueueProxy(peer, _groupMgr, body.queue, body.durable, new AMQShortString(peer.getDetails()), body.autoDelete, registry);
}
else
{
Modified: incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/replay/ConsumerCounts.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/replay/ConsumerCounts.java?view=diff&rev=494121&r1=494120&r2=494121
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/replay/ConsumerCounts.java (original)
+++ incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/replay/ConsumerCounts.java Mon Jan 8 09:02:26 2007
@@ -22,6 +22,7 @@
import org.apache.qpid.framing.AMQMethodBody;
import org.apache.qpid.framing.BasicConsumeBody;
+import org.apache.qpid.framing.AMQShortString;
import java.util.Map;
import java.util.HashMap;
@@ -29,19 +30,19 @@
class ConsumerCounts
{
- private final Map<String, Integer> _counts = new HashMap<String, Integer>();
+ private final Map<AMQShortString, Integer> _counts = new HashMap<AMQShortString, Integer>();
- synchronized void increment(String queue)
+ synchronized void increment(AMQShortString queue)
{
_counts.put(queue, get(queue) + 1);
}
- synchronized void decrement(String queue)
+ synchronized void decrement(AMQShortString queue)
{
_counts.put(queue, get(queue) - 1);
}
- private int get(String queue)
+ private int get(AMQShortString queue)
{
Integer count = _counts.get(queue);
return count == null ? 0 : count;
@@ -49,7 +50,7 @@
synchronized void replay(List<AMQMethodBody> messages)
{
- for(String queue : _counts.keySet())
+ for(AMQShortString queue : _counts.keySet())
{
// AMQP version change: Hardwire the version to 0-8 (major=8, minor=0)
// TODO: Connect this to the session version obtained from ProtocolInitiation for this session.