You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ai...@apache.org on 2008/04/24 03:54:42 UTC
svn commit: r651133 [6/17] - in /incubator/qpid/trunk/qpid: gentools/
gentools/src/org/apache/qpid/gentools/ java/ java/broker/
java/broker/src/main/grammar/
java/broker/src/main/java/org/apache/qpid/server/exchange/
java/broker/src/main/java/org/apach...
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?rev=651133&r1=651132&r2=651133&view=diff
==============================================================================
--- 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 Wed Apr 23 18:54:20 2008
@@ -21,10 +21,15 @@
package org.apache.qpid.client.protocol;
import org.apache.mina.common.IdleStatus;
+import org.apache.mina.common.IoFilterChain;
import org.apache.mina.common.IoHandlerAdapter;
import org.apache.mina.common.IoSession;
+import org.apache.mina.filter.ReadThrottleFilterBuilder;
import org.apache.mina.filter.SSLFilter;
+import org.apache.mina.filter.WriteBufferLimitFilterBuilder;
+import org.apache.mina.filter.codec.ProtocolCodecException;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
+import org.apache.mina.filter.executor.ExecutorFilter;
import org.apache.qpid.AMQConnectionClosedException;
import org.apache.qpid.AMQDisconnectedException;
import org.apache.qpid.AMQException;
@@ -39,16 +44,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.AMQBody;
-import org.apache.qpid.framing.AMQDataBlock;
-import org.apache.qpid.framing.AMQFrame;
-import org.apache.qpid.framing.AMQMethodBody;
-import org.apache.qpid.framing.AMQShortString;
-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.pool.ReadWriteThreadModel;
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.protocol.AMQMethodEvent;
@@ -57,7 +53,9 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.IOException;
import java.util.Iterator;
+import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.CountDownLatch;
@@ -155,9 +153,19 @@
/** Used to provide a condition to wait upon for operations that are required to wait for failover to complete. */
private CountDownLatch _failoverLatch;
+
+ /** The last failover exception that occured */
+ private FailoverException _lastFailoverException;
+
/** Defines the default timeout to use for synchronous protocol commands. */
private final long DEFAULT_SYNC_TIMEOUT = 1000 * 30;
+ /** Default buffer size for pending messages reads */
+ private static final String DEFAULT_READ_BUFFER_LIMIT = "262144";
+
+ /** Default buffer size for pending messages writes */
+ private static final String DEFAULT_WRITE_BUFFER_LIMIT = "262144";
+
/**
* Creates a new protocol handler, associated with the specified client connection instance.
*
@@ -211,9 +219,34 @@
}
catch (RuntimeException e)
{
- e.printStackTrace();
+ _logger.error(e.getMessage(), e);
}
+ if (Boolean.getBoolean("protectio"))
+ {
+ try
+ {
+ //Add IO Protection Filters
+ IoFilterChain chain = session.getFilterChain();
+
+ session.getFilterChain().addLast("tempExecutorFilterForFilterBuilder", new ExecutorFilter());
+
+ ReadThrottleFilterBuilder readfilter = new ReadThrottleFilterBuilder();
+ readfilter.setMaximumConnectionBufferSize(Integer.parseInt(System.getProperty("qpid.read.buffer.limit", DEFAULT_READ_BUFFER_LIMIT)));
+ readfilter.attach(chain);
+
+ WriteBufferLimitFilterBuilder writefilter = new WriteBufferLimitFilterBuilder();
+ writefilter.setMaximumConnectionBufferSize(Integer.parseInt(System.getProperty("qpid.write.buffer.limit", DEFAULT_WRITE_BUFFER_LIMIT)));
+ writefilter.attach(chain);
+ session.getFilterChain().remove("tempExecutorFilterForFilterBuilder");
+
+ _logger.info("Using IO Read/Write Filter Protection");
+ }
+ catch (Exception e)
+ {
+ _logger.error("Unable to attach IO Read/Write Filter Protection :" + e.getMessage());
+ }
+ }
_protocolSession = new AMQProtocolSession(this, session, _connection, getStateManager());
_protocolSession.init();
}
@@ -327,13 +360,27 @@
if (_failoverState == FailoverState.NOT_STARTED)
{
// if (!(cause instanceof AMQUndeliveredException) && (!(cause instanceof AMQAuthenticationException)))
- if (cause instanceof AMQConnectionClosedException)
+ if ((cause instanceof AMQConnectionClosedException) || cause instanceof IOException)
{
_logger.info("Exception caught therefore going to attempt failover: " + cause, cause);
// this will attemp failover
sessionClosed(session);
}
+ else
+ {
+
+ if (cause instanceof ProtocolCodecException)
+ {
+ _logger.info("Protocol Exception caught NOT going to attempt failover as " +
+ "cause isn't AMQConnectionClosedException: " + cause, cause);
+
+ AMQException amqe = new AMQException("Protocol handler error: " + cause, cause);
+ propagateExceptionToWaiters(amqe);
+ }
+ _connection.exceptionReceived(cause);
+
+ }
// FIXME Need to correctly handle other exceptions. Things like ...
// if (cause instanceof AMQChannelClosedException)
@@ -349,7 +396,7 @@
// we notify the state manager of the error in case we have any clients waiting on a state
// change. Those "waiters" will be interrupted and can handle the exception
- AMQException amqe = new AMQException(null, "Protocol handler error: " + cause, cause);
+ AMQException amqe = new AMQException("Protocol handler error: " + cause, cause);
propagateExceptionToWaiters(amqe);
_connection.exceptionReceived(cause);
}
@@ -364,7 +411,7 @@
*/
public void propagateExceptionToWaiters(Exception e)
{
- getStateManager().error(e);
+
if (!_frameListeners.isEmpty())
{
final Iterator it = _frameListeners.iterator();
@@ -376,98 +423,111 @@
}
}
- private static int _messageReceivedCount;
-
- public void messageReceived(IoSession session, Object message) throws Exception
+ public void notifyFailoverStarting()
{
- final boolean debug = _logger.isDebugEnabled();
- final long msgNumber = ++_messageReceivedCount;
-
- if (debug && ((msgNumber % 1000) == 0))
+ // Set the last exception in the sync block to ensure the ordering with add.
+ // either this gets done and the add does the ml.error
+ // or the add completes first and the iterator below will do ml.error
+ synchronized (_frameListeners)
{
- _logger.debug("Received " + _messageReceivedCount + " protocol messages");
+ _lastFailoverException = new FailoverException("Failing over about to start");
}
- AMQFrame frame = (AMQFrame) message;
+ propagateExceptionToWaiters(_lastFailoverException);
+ }
- final AMQBody bodyFrame = frame.getBodyFrame();
+ public void failoverInProgress()
+ {
+ _lastFailoverException = null;
+ }
- HeartbeatDiagnostics.received(bodyFrame instanceof HeartbeatBody);
+ private static int _messageReceivedCount;
- switch (bodyFrame.getFrameType())
+ public void messageReceived(IoSession session, Object message) throws Exception
+ {
+ if(message instanceof AMQFrame)
{
- case AMQMethodBody.TYPE:
+ final boolean debug = _logger.isDebugEnabled();
+ final long msgNumber = ++_messageReceivedCount;
- if (debug)
- {
- _logger.debug("(" + System.identityHashCode(this) + ")Method frame received: " + frame);
- }
+ if (debug && ((msgNumber % 1000) == 0))
+ {
+ _logger.debug("Received " + _messageReceivedCount + " protocol messages");
+ }
- final AMQMethodEvent<AMQMethodBody> evt =
- new AMQMethodEvent<AMQMethodBody>(frame.getChannel(), (AMQMethodBody) bodyFrame);
+ AMQFrame frame = (AMQFrame) message;
- try
- {
+ final AMQBody bodyFrame = frame.getBodyFrame();
- boolean wasAnyoneInterested = getStateManager().methodReceived(evt);
- if (!_frameListeners.isEmpty())
- {
- Iterator it = _frameListeners.iterator();
- while (it.hasNext())
- {
- final AMQMethodListener listener = (AMQMethodListener) it.next();
- wasAnyoneInterested = listener.methodReceived(evt) || wasAnyoneInterested;
- }
- }
-
- if (!wasAnyoneInterested)
- {
- throw new AMQException(null, "AMQMethodEvent " + evt + " was not processed by any listener. Listeners:"
- + _frameListeners, null);
- }
- }
- catch (AMQException e)
- {
- getStateManager().error(e);
- if (!_frameListeners.isEmpty())
- {
- Iterator it = _frameListeners.iterator();
- while (it.hasNext())
- {
- final AMQMethodListener listener = (AMQMethodListener) it.next();
- listener.error(e);
- }
- }
+ HeartbeatDiagnostics.received(bodyFrame instanceof HeartbeatBody);
- exceptionCaught(session, e);
- }
+ bodyFrame.handle(frame.getChannel(),_protocolSession);
- break;
+ _connection.bytesReceived(_protocolSession.getIoSession().getReadBytes());
+ }
+ else if (message instanceof ProtocolInitiation)
+ {
+ // We get here if the server sends a response to our initial protocol header
+ // suggesting an alternate ProtocolVersion; the server will then close the
+ // connection.
+ ProtocolInitiation protocolInit = (ProtocolInitiation) message;
+ ProtocolVersion pv = protocolInit.checkVersion();
+ getConnection().setProtocolVersion(pv);
- case ContentHeaderBody.TYPE:
+ // get round a bug in old versions of qpid whereby the connection is not closed
+ _stateManager.changeState(AMQState.CONNECTION_CLOSED);
+ }
+ }
- _protocolSession.messageContentHeaderReceived(frame.getChannel(), (ContentHeaderBody) bodyFrame);
- break;
+ public void methodBodyReceived(final int channelId, final AMQBody bodyFrame, IoSession session)//, final IoSession session)
+ throws AMQException
+ {
- case ContentBody.TYPE:
+ if (_logger.isDebugEnabled())
+ {
+ _logger.debug("(" + System.identityHashCode(this) + ")Method frame received: " + bodyFrame);
+ }
- _protocolSession.messageContentBodyReceived(frame.getChannel(), (ContentBody) bodyFrame);
- break;
+ final AMQMethodEvent<AMQMethodBody> evt =
+ new AMQMethodEvent<AMQMethodBody>(channelId, (AMQMethodBody) bodyFrame);
- case HeartbeatBody.TYPE:
+ try
+ {
- if (debug)
+ boolean wasAnyoneInterested = getStateManager().methodReceived(evt);
+ if (!_frameListeners.isEmpty())
+ {
+ //This iterator is safe from the error state as the frame listeners always add before they send so their
+ // will be ready and waiting for this response.
+ Iterator it = _frameListeners.iterator();
+ while (it.hasNext())
{
- _logger.debug("Received heartbeat");
+ final AMQMethodListener listener = (AMQMethodListener) it.next();
+ wasAnyoneInterested = listener.methodReceived(evt) || wasAnyoneInterested;
}
+ }
- break;
-
- default:
+ if (!wasAnyoneInterested)
+ {
+ throw new AMQException(null, "AMQMethodEvent " + evt + " was not processed by any listener. Listeners:"
+ + _frameListeners, null);
+ }
+ }
+ catch (AMQException e)
+ {
+ if (!_frameListeners.isEmpty())
+ {
+ Iterator it = _frameListeners.iterator();
+ while (it.hasNext())
+ {
+ final AMQMethodListener listener = (AMQMethodListener) it.next();
+ listener.error(e);
+ }
+ }
+ exceptionCaught(session, e);
}
- _connection.bytesReceived(_protocolSession.getIoSession().getReadBytes());
}
private static int _messagesOut;
@@ -506,6 +566,12 @@
getStateManager().attainState(s);
}
+ public AMQState attainState(Set<AMQState> states) throws AMQException
+ {
+ return getStateManager().attainState(states);
+ }
+
+
/**
* Convenience method that writes a frame to the protocol session. Equivalent to calling
* getProtocolSession().write().
@@ -547,7 +613,15 @@
{
try
{
- _frameListeners.add(listener);
+ synchronized (_frameListeners)
+ {
+ if (_lastFailoverException != null)
+ {
+ throw _lastFailoverException;
+ }
+
+ _frameListeners.add(listener);
+ }
_protocolSession.writeFrame(frame);
AMQMethodEvent e = listener.blockForFrame(timeout);
@@ -556,10 +630,6 @@
// 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 removeKey the listener then no-one will
@@ -600,16 +670,11 @@
{
getStateManager().changeState(AMQState.CONNECTION_CLOSING);
- // 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.
- final AMQFrame frame =
- ConnectionCloseBody.createAMQFrame(0, _protocolSession.getProtocolMajorVersion(),
- _protocolSession.getProtocolMinorVersion(), // AMQP version (major, minor)
- 0, // classId
- 0, // methodId
- AMQConstant.REPLY_SUCCESS.getCode(), // replyCode
- new AMQShortString("JMS client is closing the connection.")); // replyText
+ ConnectionCloseBody body = _protocolSession.getMethodRegistry().createConnectionCloseBody(AMQConstant.REPLY_SUCCESS.getCode(), // replyCode
+ new AMQShortString("JMS client is closing the connection."),0,0);
+
+
+ final AMQFrame frame = body.generateFrame(0);
try
{
@@ -682,7 +747,10 @@
public void setStateManager(AMQStateManager stateManager)
{
_stateManager = stateManager;
- _protocolSession.setStateManager(stateManager);
+ if (_protocolSession != null)
+ {
+ _protocolSession.setStateManager(stateManager);
+ }
}
public AMQProtocolSession getProtocolSession()
@@ -708,5 +776,15 @@
public byte getProtocolMinorVersion()
{
return _protocolSession.getProtocolMinorVersion();
+ }
+
+ public MethodRegistry getMethodRegistry()
+ {
+ return getStateManager().getMethodRegistry();
+ }
+
+ public ProtocolVersion getProtocolVersion()
+ {
+ return _protocolSession.getProtocolVersion();
}
}
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?rev=651133&r1=651132&r2=651133&view=diff
==============================================================================
--- 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 Wed Apr 23 18:54:20 2008
@@ -21,40 +21,31 @@
package org.apache.qpid.client.protocol;
import org.apache.commons.lang.StringUtils;
-
import org.apache.mina.common.CloseFuture;
import org.apache.mina.common.IdleStatus;
import org.apache.mina.common.IoSession;
import org.apache.mina.common.WriteFuture;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.jms.JMSException;
+import javax.security.sasl.SaslClient;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import org.apache.qpid.AMQException;
import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.client.AMQSession;
import org.apache.qpid.client.ConnectionTuneParameters;
-// import org.apache.qpid.client.message.UnexpectedBodyReceivedException;
import org.apache.qpid.client.message.ReturnMessage;
import org.apache.qpid.client.message.UnprocessedMessage;
import org.apache.qpid.client.message.UnprocessedMessage_0_8;
import org.apache.qpid.client.state.AMQStateManager;
-import org.apache.qpid.framing.AMQDataBlock;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.ContentBody;
-import org.apache.qpid.framing.ContentHeaderBody;
-import org.apache.qpid.framing.MainRegistry;
-import org.apache.qpid.framing.ProtocolInitiation;
-import org.apache.qpid.framing.ProtocolVersion;
-import org.apache.qpid.framing.VersionSpecificRegistry;
+import org.apache.qpid.framing.*;
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.protocol.AMQVersionAwareProtocolSession;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.jms.JMSException;
-import javax.security.sasl.SaslClient;
-
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
+import org.apache.qpid.client.handler.ClientMethodDispatcherImpl;
/**
* Wrapper for protocol session that provides type-safe access to session attributes. <p/> The underlying protocol
@@ -95,18 +86,27 @@
* Maps from a channel id to an unprocessed message. This is used to tie together the JmsDeliverBody (which arrives
* first) with the subsequent content header and content bodies.
*/
- protected ConcurrentMap<Integer,UnprocessedMessage_0_8> _channelId2UnprocessedMsgMap = new ConcurrentHashMap<Integer,UnprocessedMessage_0_8>();
+ private final ConcurrentMap<Integer,UnprocessedMessage> _channelId2UnprocessedMsgMap = new ConcurrentHashMap<Integer,UnprocessedMessage>();
+ private final UnprocessedMessage[] _channelId2UnprocessedMsgArray = new UnprocessedMessage[16];
/** Counter to ensure unique queue names */
protected int _queueId = 1;
protected final Object _queueIdLock = new Object();
- private byte _protocolMinorVersion;
- private byte _protocolMajorVersion;
- private VersionSpecificRegistry _registry =
- MainRegistry.getVersionSpecificRegistry(ProtocolVersion.getLatestSupportedVersion());
+ private ProtocolVersion _protocolVersion;
+// private VersionSpecificRegistry _registry =
+// MainRegistry.getVersionSpecificRegistry(ProtocolVersion.getLatestSupportedVersion());
+
+
+ private MethodRegistry _methodRegistry =
+ MethodRegistry.getMethodRegistry(ProtocolVersion.getLatestSupportedVersion());
+
+
+ private MethodDispatcher _methodDispatcher;
+
private final AMQConnection _connection;
+ private static final int FAST_CHANNEL_ACCESS_MASK = 0xFFFFFFF0;
public AMQProtocolSession(AMQProtocolHandler protocolHandler, IoSession protocolSession, AMQConnection connection)
{
@@ -126,6 +126,9 @@
_minaProtocolSession.setWriteTimeout(LAST_WRITE_FUTURE_JOIN_TIMEOUT);
_stateManager = stateManager;
_stateManager.setProtocolSession(this);
+ _protocolVersion = connection.getProtocolVersion();
+ _methodDispatcher = ClientMethodDispatcherImpl.newMethodDispatcher(ProtocolVersion.getLatestSupportedVersion(),
+ stateManager);
_connection = connection;
}
@@ -135,7 +138,7 @@
// start the process of setting up the connection. This is the first place that
// data is written to the server.
- _minaProtocolSession.write(new ProtocolInitiation(ProtocolVersion.getLatestSupportedVersion()));
+ _minaProtocolSession.write(new ProtocolInitiation(_connection.getProtocolVersion()));
}
public String getClientID()
@@ -164,6 +167,8 @@
public void setStateManager(AMQStateManager stateManager)
{
_stateManager = stateManager;
+ _methodDispatcher = ClientMethodDispatcherImpl.newMethodDispatcher(_protocolVersion,
+ stateManager);
}
public String getVirtualHost()
@@ -230,14 +235,25 @@
*
* @throws AMQException if this was not expected
*/
- public void unprocessedMessageReceived(UnprocessedMessage_0_8 message) throws AMQException
+ public void unprocessedMessageReceived(UnprocessedMessage message) throws AMQException
{
- _channelId2UnprocessedMsgMap.put(message.getChannelId(), message);
+ final int channelId = message.getChannelId();
+ if((channelId & FAST_CHANNEL_ACCESS_MASK) == 0)
+ {
+ _channelId2UnprocessedMsgArray[channelId] = message;
+ }
+ else
+ {
+ _channelId2UnprocessedMsgMap.put(channelId, message);
+ }
}
- public void messageContentHeaderReceived(int channelId, ContentHeaderBody contentHeader) throws AMQException
+ public void contentHeaderReceived(int channelId, ContentHeaderBody contentHeader) throws AMQException
{
- UnprocessedMessage_0_8 msg = (UnprocessedMessage_0_8) _channelId2UnprocessedMsgMap.get(channelId);
+ final UnprocessedMessage msg = (channelId & FAST_CHANNEL_ACCESS_MASK) == 0 ? _channelId2UnprocessedMsgArray[channelId]
+ : _channelId2UnprocessedMsgMap.get(channelId);
+
+
if (msg == null)
{
throw new AMQException(null, "Error: received content header without having received a BasicDeliver frame first", null);
@@ -255,9 +271,19 @@
}
}
- public void messageContentBodyReceived(int channelId, ContentBody contentBody) throws AMQException
+ public void contentBodyReceived(final int channelId, ContentBody contentBody) throws AMQException
{
- UnprocessedMessage_0_8 msg = _channelId2UnprocessedMsgMap.get(channelId);
+ UnprocessedMessage_0_8 msg;
+ final boolean fastAccess = (channelId & FAST_CHANNEL_ACCESS_MASK) == 0;
+ if(fastAccess)
+ {
+ msg = (UnprocessedMessage_0_8) _channelId2UnprocessedMsgArray[channelId];
+ }
+ else
+ {
+ msg = (UnprocessedMessage_0_8) _channelId2UnprocessedMsgMap.get(channelId);
+ }
+
if (msg == null)
{
throw new AMQException(null, "Error: received content body without having received a JMSDeliver frame first", null);
@@ -265,7 +291,14 @@
if (msg.getContentHeader() == null)
{
- _channelId2UnprocessedMsgMap.remove(channelId);
+ if(fastAccess)
+ {
+ _channelId2UnprocessedMsgArray[channelId] = null;
+ }
+ else
+ {
+ _channelId2UnprocessedMsgMap.remove(channelId);
+ }
throw new AMQException(null, "Error: received content body without having received a ContentHeader frame first", null);
}
@@ -285,6 +318,11 @@
}
}
+ public void heartbeatBodyReceived(int channelId, HeartbeatBody body) throws AMQException
+ {
+
+ }
+
/**
* Deliver a message to the appropriate session, removing the unprocessed message from our map
*
@@ -295,7 +333,14 @@
{
AMQSession session = getSession(channelId);
session.messageReceived(msg);
- _channelId2UnprocessedMsgMap.remove(channelId);
+ if((channelId & FAST_CHANNEL_ACCESS_MASK) == 0)
+ {
+ _channelId2UnprocessedMsgArray[channelId] = null;
+ }
+ else
+ {
+ _channelId2UnprocessedMsgMap.remove(channelId);
+ }
}
protected AMQSession getSession(int channelId)
@@ -440,26 +485,64 @@
session.confirmConsumerCancelled(consumerTag);
}
- public void setProtocolVersion(final byte versionMajor, final byte versionMinor)
+ public void setProtocolVersion(final ProtocolVersion pv)
{
- _protocolMajorVersion = versionMajor;
- _protocolMinorVersion = versionMinor;
- _registry = MainRegistry.getVersionSpecificRegistry(versionMajor, versionMinor);
+ _protocolVersion = pv;
+ _methodRegistry = MethodRegistry.getMethodRegistry(pv);
+ _methodDispatcher = ClientMethodDispatcherImpl.newMethodDispatcher(pv, _stateManager);
+
+ // _registry = MainRegistry.getVersionSpecificRegistry(versionMajor, versionMinor);
}
public byte getProtocolMinorVersion()
{
- return _protocolMinorVersion;
+ return _protocolVersion.getMinorVersion();
}
public byte getProtocolMajorVersion()
{
- return _protocolMajorVersion;
+ return _protocolVersion.getMajorVersion();
}
- public VersionSpecificRegistry getRegistry()
+ public ProtocolVersion getProtocolVersion()
{
- return _registry;
+ return _protocolVersion;
}
+// public VersionSpecificRegistry getRegistry()
+// {
+// return _registry;
+// }
+
+ public MethodRegistry getMethodRegistry()
+ {
+ return _methodRegistry;
+ }
+
+ public MethodDispatcher getMethodDispatcher()
+ {
+ return _methodDispatcher;
+ }
+
+
+ public void setTicket(int ticket, int channelId)
+ {
+ final AMQSession session = getSession(channelId);
+ session.setTicket(ticket);
+ }
+ public void setMethodDispatcher(MethodDispatcher methodDispatcher)
+ {
+ _methodDispatcher = methodDispatcher;
+ }
+
+ public void setFlowControl(final int channelId, final boolean active)
+ {
+ final AMQSession session = getSession(channelId);
+ session.setFlowControl(active);
+ }
+
+ public void methodFrameReceived(final int channel, final AMQMethodBody amqMethodBody) throws AMQException
+ {
+ _protocolHandler.methodBodyReceived(channel, amqMethodBody, _minaProtocolSession);
+ }
}
Added: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQMethodNotImplementedException.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQMethodNotImplementedException.java?rev=651133&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQMethodNotImplementedException.java (added)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQMethodNotImplementedException.java Wed Apr 23 18:54:20 2008
@@ -0,0 +1,32 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.client.state;
+
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.AMQException;
+
+public class AMQMethodNotImplementedException extends AMQException
+{
+ public AMQMethodNotImplementedException(AMQMethodBody body)
+ {
+ super(null, "Unexpected Method Received: " + body.getClass().getName(), null);
+ }
+}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQState.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQState.java?rev=651133&r1=651132&r2=651133&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQState.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQState.java Wed Apr 23 18:54:20 2008
@@ -24,8 +24,22 @@
* States used in the AMQ protocol. Used by the finite state machine to determine
* valid responses.
*/
-public class AMQState
+public enum AMQState
{
+
+ CONNECTION_NOT_STARTED(1, "CONNECTION_NOT_STARTED"),
+
+ CONNECTION_NOT_TUNED(2, "CONNECTION_NOT_TUNED"),
+
+ CONNECTION_NOT_OPENED(3, "CONNECTION_NOT_OPENED"),
+
+ CONNECTION_OPEN(4, "CONNECTION_OPEN"),
+
+ CONNECTION_CLOSING(5, "CONNECTION_CLOSING"),
+
+ CONNECTION_CLOSED(6, "CONNECTION_CLOSED");
+
+
private final int _id;
private final String _name;
@@ -41,16 +55,6 @@
return "AMQState: id = " + _id + " name: " + _name;
}
- public static final AMQState CONNECTION_NOT_STARTED = new AMQState(1, "CONNECTION_NOT_STARTED");
-
- public static final AMQState CONNECTION_NOT_TUNED = new AMQState(2, "CONNECTION_NOT_TUNED");
-
- public static final AMQState CONNECTION_NOT_OPENED = new AMQState(3, "CONNECTION_NOT_OPENED");
-
- public static final AMQState CONNECTION_OPEN = new AMQState(4, "CONNECTION_OPEN");
-
- public static final AMQState CONNECTION_CLOSING = new AMQState(5, "CONNECTION_CLOSING");
-
- public static final AMQState CONNECTION_CLOSED = new AMQState(6, "CONNECTION_CLOSED");
-
+
+
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java?rev=651133&r1=651132&r2=651133&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java Wed Apr 23 18:54:20 2008
@@ -21,49 +21,22 @@
package org.apache.qpid.client.state;
import org.apache.qpid.AMQException;
-import org.apache.qpid.client.handler.BasicCancelOkMethodHandler;
-import org.apache.qpid.client.handler.BasicDeliverMethodHandler;
-import org.apache.qpid.client.handler.BasicReturnMethodHandler;
-import org.apache.qpid.client.handler.ChannelCloseMethodHandler;
-import org.apache.qpid.client.handler.ChannelCloseOkMethodHandler;
-import org.apache.qpid.client.handler.ChannelFlowOkMethodHandler;
-import org.apache.qpid.client.handler.ConnectionCloseMethodHandler;
-import org.apache.qpid.client.handler.ConnectionOpenOkMethodHandler;
-import org.apache.qpid.client.handler.ConnectionSecureMethodHandler;
-import org.apache.qpid.client.handler.ConnectionStartMethodHandler;
-import org.apache.qpid.client.handler.ConnectionTuneMethodHandler;
-import org.apache.qpid.client.handler.ExchangeBoundOkMethodHandler;
-import org.apache.qpid.client.handler.QueueDeleteOkMethodHandler;
import org.apache.qpid.client.protocol.AMQProtocolSession;
-import org.apache.qpid.framing.AMQMethodBody;
-import org.apache.qpid.framing.BasicCancelOkBody;
-import org.apache.qpid.framing.BasicDeliverBody;
-import org.apache.qpid.framing.BasicReturnBody;
-import org.apache.qpid.framing.ChannelCloseBody;
-import org.apache.qpid.framing.ChannelCloseOkBody;
-import org.apache.qpid.framing.ChannelFlowOkBody;
-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.ExchangeBoundOkBody;
-import org.apache.qpid.framing.QueueDeleteOkBody;
+import org.apache.qpid.framing.*;
import org.apache.qpid.protocol.AMQMethodEvent;
import org.apache.qpid.protocol.AMQMethodListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.HashMap;
import java.util.Iterator;
-import java.util.Map;
+import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
/**
* The state manager is responsible for managing the state of the protocol session. <p/> For each AMQProtocolHandler
* there is a separate state manager.
*/
-public class AMQStateManager implements AMQMethodListener
+public class AMQStateManager
{
private static final Logger _logger = LoggerFactory.getLogger(AMQStateManager.class);
@@ -72,15 +45,15 @@
/** The current state */
private AMQState _currentState;
+
/**
* Maps from an AMQState instance to a Map from Class to StateTransitionHandler. The class must be a subclass of
* AMQFrame.
*/
- protected final Map _state2HandlersMap = new HashMap();
- private final CopyOnWriteArraySet _stateListeners = new CopyOnWriteArraySet();
+
private final Object _stateLock = new Object();
- private static final long MAXIMUM_STATE_WAIT_TIME = 30000L;
+ private static final long MAXIMUM_STATE_WAIT_TIME = Long.parseLong(System.getProperty("amqj.MaximumStateWait", "30000"));
public AMQStateManager()
{
@@ -96,53 +69,10 @@
{
_protocolSession = protocolSession;
_currentState = state;
- if (register)
- {
- registerListeners();
- }
+
}
- protected void registerListeners()
- {
- Map frame2handlerMap = new HashMap();
- // we need to register a map for the null (i.e. all state) handlers otherwise you get
- // a stack overflow in the handler searching code when you present it with a frame for which
- // no handlers are registered
- //
- _state2HandlersMap.put(null, frame2handlerMap);
-
- frame2handlerMap = new HashMap();
- frame2handlerMap.put(ConnectionStartBody.class, ConnectionStartMethodHandler.getInstance());
- frame2handlerMap.put(ConnectionCloseBody.class, ConnectionCloseMethodHandler.getInstance());
- _state2HandlersMap.put(AMQState.CONNECTION_NOT_STARTED, frame2handlerMap);
-
- frame2handlerMap = new HashMap();
- frame2handlerMap.put(ConnectionTuneBody.class, ConnectionTuneMethodHandler.getInstance());
- frame2handlerMap.put(ConnectionSecureBody.class, ConnectionSecureMethodHandler.getInstance());
- frame2handlerMap.put(ConnectionCloseBody.class, ConnectionCloseMethodHandler.getInstance());
- _state2HandlersMap.put(AMQState.CONNECTION_NOT_TUNED, frame2handlerMap);
-
- frame2handlerMap = new HashMap();
- frame2handlerMap.put(ConnectionOpenOkBody.class, ConnectionOpenOkMethodHandler.getInstance());
- frame2handlerMap.put(ConnectionCloseBody.class, ConnectionCloseMethodHandler.getInstance());
- _state2HandlersMap.put(AMQState.CONNECTION_NOT_OPENED, frame2handlerMap);
-
- //
- // ConnectionOpen handlers
- //
- frame2handlerMap = new HashMap();
- frame2handlerMap.put(ChannelCloseBody.class, ChannelCloseMethodHandler.getInstance());
- frame2handlerMap.put(ChannelCloseOkBody.class, ChannelCloseOkMethodHandler.getInstance());
- frame2handlerMap.put(ConnectionCloseBody.class, ConnectionCloseMethodHandler.getInstance());
- frame2handlerMap.put(BasicDeliverBody.class, BasicDeliverMethodHandler.getInstance());
- frame2handlerMap.put(BasicReturnBody.class, BasicReturnMethodHandler.getInstance());
- frame2handlerMap.put(BasicCancelOkBody.class, BasicCancelOkMethodHandler.getInstance());
- frame2handlerMap.put(ChannelFlowOkBody.class, ChannelFlowOkMethodHandler.getInstance());
- frame2handlerMap.put(QueueDeleteOkBody.class, QueueDeleteOkMethodHandler.getInstance());
- frame2handlerMap.put(ExchangeBoundOkBody.class, ExchangeBoundOkMethodHandler.getInstance());
- _state2HandlersMap.put(AMQState.CONNECTION_OPEN, frame2handlerMap);
- }
public AMQState getCurrentState()
{
@@ -160,72 +90,17 @@
}
}
- public void error(Exception e)
- {
- _logger.debug("State manager receive error notification: " + e);
- synchronized (_stateListeners)
- {
- final Iterator it = _stateListeners.iterator();
- while (it.hasNext())
- {
- final StateListener l = (StateListener) it.next();
- l.error(e);
- }
- }
- }
public <B extends AMQMethodBody> boolean methodReceived(AMQMethodEvent<B> evt) throws AMQException
{
- StateAwareMethodListener handler = findStateTransitionHandler(_currentState, evt.getMethod());
- if (handler != null)
- {
- handler.methodReceived(this, _protocolSession, evt);
-
- return true;
- }
- return false;
+ B method = evt.getMethod();
+
+ // StateAwareMethodListener handler = findStateTransitionHandler(_currentState, evt.getMethod());
+ method.execute(_protocolSession.getMethodDispatcher(), evt.getChannelId());
+ return true;
}
- protected StateAwareMethodListener findStateTransitionHandler(AMQState currentState, AMQMethodBody frame)
- // throws IllegalStateTransitionException
- {
- final Class clazz = frame.getClass();
- if (_logger.isDebugEnabled())
- {
- _logger.debug("Looking for state[" + currentState + "] transition handler for frame " + clazz);
- }
-
- final Map classToHandlerMap = (Map) _state2HandlersMap.get(currentState);
-
- if (classToHandlerMap == null)
- {
- // if no specialised per state handler is registered look for a
- // handler registered for "all" states
- return findStateTransitionHandler(null, frame);
- }
-
- final StateAwareMethodListener handler = (StateAwareMethodListener) classToHandlerMap.get(clazz);
- if (handler == null)
- {
- if (currentState == null)
- {
- _logger.debug("No state transition handler defined for receiving frame " + frame);
-
- return null;
- }
- else
- {
- // if no specialised per state handler is registered look for a
- // handler registered for "all" states
- return findStateTransitionHandler(null, frame);
- }
- }
- else
- {
- return handler;
- }
- }
public void attainState(final AMQState s) throws AMQException
{
@@ -271,5 +146,47 @@
public void setProtocolSession(AMQProtocolSession session)
{
_protocolSession = session;
+ }
+
+ public MethodRegistry getMethodRegistry()
+ {
+ return getProtocolSession().getMethodRegistry();
+ }
+
+ public AMQState attainState(Set<AMQState> stateSet) throws AMQException
+ {
+ synchronized (_stateLock)
+ {
+ final long waitUntilTime = System.currentTimeMillis() + MAXIMUM_STATE_WAIT_TIME;
+ long waitTime = MAXIMUM_STATE_WAIT_TIME;
+
+ while (!stateSet.contains(_currentState) && (waitTime > 0))
+ {
+ try
+ {
+ _stateLock.wait(MAXIMUM_STATE_WAIT_TIME);
+ }
+ catch (InterruptedException e)
+ {
+ _logger.warn("Thread interrupted");
+ }
+
+ if (!stateSet.contains(_currentState))
+ {
+ waitTime = waitUntilTime - System.currentTimeMillis();
+ }
+ }
+
+ if (!stateSet.contains(_currentState))
+ {
+ _logger.warn("State not achieved within permitted time. Current state " + _currentState
+ + ", desired state: " + stateSet);
+ throw new AMQException(null, "State not achieved within permitted time. Current state " + _currentState
+ + ", desired state: " + stateSet, null);
+ }
+ return _currentState;
+ }
+
+
}
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/StateAwareMethodListener.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/StateAwareMethodListener.java?rev=651133&r1=651132&r2=651133&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/StateAwareMethodListener.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/StateAwareMethodListener.java Wed Apr 23 18:54:20 2008
@@ -21,6 +21,7 @@
package org.apache.qpid.client.state;
import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQMethodBody;
import org.apache.qpid.client.protocol.AMQProtocolSession;
import org.apache.qpid.protocol.AMQMethodEvent;
@@ -29,8 +30,9 @@
* the opportunity to update state.
*
*/
-public interface StateAwareMethodListener
+public interface StateAwareMethodListener<B extends AMQMethodBody>
{
- void methodReceived(AMQStateManager stateManager, AMQProtocolSession protocolSession,
- AMQMethodEvent evt) throws AMQException;
+
+ void methodReceived(AMQStateManager stateManager, B body, int channelId) throws AMQException;
+
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/transport/SocketTransportConnection.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/transport/SocketTransportConnection.java?rev=651133&r1=651132&r2=651133&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/transport/SocketTransportConnection.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/transport/SocketTransportConnection.java Wed Apr 23 18:54:20 2008
@@ -24,6 +24,7 @@
import org.apache.mina.common.ConnectFuture;
import org.apache.mina.common.IoConnector;
import org.apache.mina.common.SimpleByteBufferAllocator;
+import org.apache.mina.transport.socket.nio.ExistingSocketConnector;
import org.apache.mina.transport.socket.nio.SocketConnectorConfig;
import org.apache.mina.transport.socket.nio.SocketSessionConfig;
@@ -36,6 +37,9 @@
import java.io.IOException;
import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
public class SocketTransportConnection implements ITransportConnection
{
@@ -83,8 +87,34 @@
_logger.info("send-buffer-size = " + scfg.getSendBufferSize());
scfg.setReceiveBufferSize(Integer.getInteger("amqj.receiveBufferSize", DEFAULT_BUFFER_SIZE));
_logger.info("recv-buffer-size = " + scfg.getReceiveBufferSize());
- final InetSocketAddress address = new InetSocketAddress(brokerDetail.getHost(), brokerDetail.getPort());
- _logger.info("Attempting connection to " + address);
+
+ final InetSocketAddress address;
+
+ if (brokerDetail.getTransport().equals(BrokerDetails.SOCKET))
+ {
+ address = null;
+
+ Socket socket = TransportConnection.removeOpenSocket(brokerDetail.getHost());
+
+ if (socket != null)
+ {
+ _logger.info("Using existing Socket:" + socket);
+
+ ((ExistingSocketConnector) ioConnector).setOpenSocket(socket);
+ }
+ else
+ {
+ throw new IllegalArgumentException("Active Socket must be provided for broker " +
+ "with 'socket://<SocketID>' transport:" + brokerDetail);
+ }
+ }
+ else
+ {
+ address = new InetSocketAddress(brokerDetail.getHost(), brokerDetail.getPort());
+ _logger.info("Attempting connection to " + address);
+ }
+
+
ConnectFuture future = ioConnector.connect(address, protocolHandler);
// wait for connection to complete
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java?rev=651133&r1=651132&r2=651133&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java Wed Apr 23 18:54:20 2008
@@ -23,6 +23,8 @@
import org.apache.mina.common.IoConnector;
import org.apache.mina.common.IoHandlerAdapter;
import org.apache.mina.common.IoServiceConfig;
+import org.apache.mina.transport.socket.nio.ExistingSocketConnector;
+import org.apache.mina.transport.socket.nio.MultiThreadSocketConnector;
import org.apache.mina.transport.socket.nio.SocketConnector;
import org.apache.mina.transport.vmpipe.VmPipeAcceptor;
import org.apache.mina.transport.vmpipe.VmPipeAddress;
@@ -34,8 +36,10 @@
import java.io.IOException;
import java.util.HashMap;
-import java.util.Iterator;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.net.Socket;
+
/**
* The TransportConnection is a helper class responsible for connecting to an AMQ server. It sets up the underlying
@@ -54,12 +58,25 @@
private static final int TCP = 0;
private static final int VM = 1;
+ private static final int SOCKET = 2;
private static Logger _logger = LoggerFactory.getLogger(TransportConnection.class);
private static final String DEFAULT_QPID_SERVER = "org.apache.qpid.server.protocol.AMQPFastProtocolHandler";
- public static ITransportConnection getInstance(BrokerDetails details) throws AMQTransportConnectionException
+ private static Map<String, Socket> _openSocketRegister = new ConcurrentHashMap<String, Socket>();
+
+ public static void registerOpenSocket(String socketID, Socket openSocket)
+ {
+ _openSocketRegister.put(socketID, openSocket);
+ }
+
+ public static Socket removeOpenSocket(String socketID)
+ {
+ return _openSocketRegister.remove(socketID);
+ }
+
+ public static synchronized ITransportConnection getInstance(BrokerDetails details) throws AMQTransportConnectionException
{
int transport = getTransport(details.getTransport());
@@ -68,7 +85,7 @@
throw new AMQNoTransportForProtocolException(details, null, null);
}
- if (transport == _currentInstance)
+ /* if (transport == _currentInstance)
{
if (transport == VM)
{
@@ -83,13 +100,23 @@
}
}
- _currentInstance = transport;
+ _currentInstance = transport;*/
+ ITransportConnection instance;
switch (transport)
{
-
+ case SOCKET:
+ instance =
+ new SocketTransportConnection(new SocketTransportConnection.SocketConnectorFactory()
+ {
+ public IoConnector newSocketConnector()
+ {
+ return new ExistingSocketConnector();
+ }
+ });
+ break;
case TCP:
- _instance = new SocketTransportConnection(new SocketTransportConnection.SocketConnectorFactory()
+ instance = new SocketTransportConnection(new SocketTransportConnection.SocketConnectorFactory()
{
public IoConnector newSocketConnector()
{
@@ -97,38 +124,44 @@
// FIXME - this needs to be sorted to use the new Mina MultiThread SA.
if (Boolean.getBoolean("qpidnio"))
{
- _logger.error("Using Qpid NIO - sysproperty 'qpidnio' is set.");
- // result = new org.apache.qpid.nio.SocketConnector(); // non-blocking connector
+ _logger.warn("Using Qpid MultiThreaded NIO - " + (System.getProperties().containsKey("qpidnio")
+ ? "Qpid NIO is new default"
+ : "Sysproperty 'qpidnio' is set"));
+ result = new MultiThreadSocketConnector();
}
- // else
-
+ else
{
_logger.info("Using Mina NIO");
result = new SocketConnector(); // non-blocking connector
}
-
// Don't have the connector's worker thread wait around for other connections (we only use
// one SocketConnector per connection at the moment anyway). This allows short-running
// clients (like unit tests) to complete quickly.
result.setWorkerTimeout(0);
-
return result;
}
});
break;
-
case VM:
{
- _instance = getVMTransport(details, Boolean.getBoolean("amqj.AutoCreateVMBroker"));
+ instance = getVMTransport(details, Boolean.getBoolean("amqj.AutoCreateVMBroker"));
break;
}
+ default:
+ // FIXME: TGM
+ throw new AMQNoTransportForProtocolException(details, null, null);
}
- return _instance;
+ return instance;
}
private static int getTransport(String transport)
{
+ if (transport.equals(BrokerDetails.SOCKET))
+ {
+ return SOCKET;
+ }
+
if (transport.equals(BrokerDetails.TCP))
{
return TCP;
@@ -151,7 +184,15 @@
{
if (AutoCreate)
{
- createVMBroker(port);
+ if (AutoCreate)
+ {
+ createVMBroker(port);
+ }
+ else
+ {
+ throw new AMQVMBrokerCreationException(null, port, "VM Broker on port " + port
+ + " does not exist. Auto create disabled.", null);
+ }
}
else
{
@@ -281,16 +322,17 @@
public static void killAllVMBrokers()
{
_logger.info("Killing all VM Brokers");
- _acceptor.unbindAll();
-
- Iterator keys = _inVmPipeAddress.keySet().iterator();
-
- while (keys.hasNext())
+ if (_acceptor != null)
{
- int id = (Integer) keys.next();
- _inVmPipeAddress.remove(id);
+ _acceptor.unbindAll();
}
-
+ synchronized (_inVmPipeAddress)
+ {
+ _inVmPipeAddress.clear();
+ }
+ _acceptor = null;
+ _currentInstance = -1;
+ _currentVMPort = -1;
}
public static void killVMBroker(int port)
@@ -300,6 +342,8 @@
{
_logger.info("Killing VM Broker:" + port);
_inVmPipeAddress.remove(port);
+ // This does need to be sychronized as otherwise mina can hang
+ // if a new connection is made
_acceptor.unbind(pipe);
}
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/transport/VmPipeTransportConnection.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/transport/VmPipeTransportConnection.java?rev=651133&r1=651132&r2=651133&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/transport/VmPipeTransportConnection.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/transport/VmPipeTransportConnection.java Wed Apr 23 18:54:20 2008
@@ -22,15 +22,12 @@
import org.apache.mina.common.ConnectFuture;
import org.apache.mina.common.IoServiceConfig;
+import org.apache.mina.transport.vmpipe.QpidVmPipeConnector;
import org.apache.mina.transport.vmpipe.VmPipeAddress;
import org.apache.mina.transport.vmpipe.VmPipeConnector;
-
import org.apache.qpid.client.protocol.AMQProtocolHandler;
import org.apache.qpid.jms.BrokerDetails;
-import org.apache.qpid.pool.PoolingFilter;
-import org.apache.qpid.pool.ReferenceCountingExecutorService;
import org.apache.qpid.pool.ReadWriteThreadModel;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -49,10 +46,10 @@
public void connect(AMQProtocolHandler protocolHandler, BrokerDetails brokerDetail) throws IOException
{
- final VmPipeConnector ioConnector = new VmPipeConnector();
+ final VmPipeConnector ioConnector = new QpidVmPipeConnector();
final IoServiceConfig cfg = ioConnector.getDefaultConfig();
- cfg.setThreadModel(ReadWriteThreadModel.getInstance());
+ cfg.setThreadModel(ReadWriteThreadModel.getInstance());
final VmPipeAddress address = new VmPipeAddress(_port);
_logger.info("Attempting connection to " + address);
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java?rev=651133&r1=651132&r2=651133&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java Wed Apr 23 18:54:20 2008
@@ -33,8 +33,10 @@
*/
public static final String OPTIONS_RETRY = "retries";
public static final String OPTIONS_CONNECT_TIMEOUT = "connecttimeout";
+ public static final String OPTIONS_CONNECT_DELAY = "connectdelay";
public static final int DEFAULT_PORT = 5672;
+ public static final String SOCKET = "socket";
public static final String TCP = "tcp";
public static final String VM = "vm";
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/ConnectionURL.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/ConnectionURL.java?rev=651133&r1=651132&r2=651133&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/ConnectionURL.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/ConnectionURL.java Wed Apr 23 18:54:20 2008
@@ -21,6 +21,7 @@
package org.apache.qpid.jms;
import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.ProtocolVersion;
import java.util.List;
@@ -45,7 +46,7 @@
public static final String OPTIONS_TEMPORARY_QUEUE_EXCHANGE = "temporaryQueueExchange";
public static final byte URL_0_8 = 1;
public static final byte URL_0_10 = 2;
-
+
String getURL();
String getFailoverMethod();
@@ -89,4 +90,5 @@
AMQShortString getTemporaryQueueExchangeName();
AMQShortString getTemporaryTopicExchangeName();
+
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/FailoverPolicy.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/FailoverPolicy.java?rev=651133&r1=651132&r2=651133&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/FailoverPolicy.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/FailoverPolicy.java Wed Apr 23 18:54:20 2008
@@ -34,7 +34,6 @@
private static final long MINUTE = 60000L;
private static final long DEFAULT_METHOD_TIMEOUT = 1 * MINUTE;
- private static final long DEFAULT_FAILOVER_TIMEOUT = 4 * MINUTE;
private FailoverMethod[] _methods = new FailoverMethod[1];
@@ -161,16 +160,7 @@
}
else
{
- if ((now - _lastFailTime) >= DEFAULT_FAILOVER_TIMEOUT)
- {
- _logger.info("Failover timeout");
-
- return false;
- }
- else
- {
- _lastMethodTime = now;
- }
+ _lastMethodTime = now;
}
}
else
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/MessageProducer.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/MessageProducer.java?rev=651133&r1=651132&r2=651133&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/MessageProducer.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/MessageProducer.java Wed Apr 23 18:54:20 2008
@@ -50,4 +50,8 @@
void send(Destination destination, Message message, int deliveryMode,
int priority, long timeToLive, boolean mandatory, boolean immediate)
throws JMSException;
+
+ void send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive,
+ boolean mandatory, boolean immediate, boolean waitUntilSent) throws JMSException;
+
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverRoundRobinServers.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverRoundRobinServers.java?rev=651133&r1=651132&r2=651133&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverRoundRobinServers.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverRoundRobinServers.java Wed Apr 23 18:54:20 2008
@@ -22,7 +22,6 @@
import org.apache.qpid.jms.BrokerDetails;
import org.apache.qpid.jms.ConnectionURL;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -35,34 +34,22 @@
/** The default number of times to retry each server */
public static final int DEFAULT_SERVER_RETRIES = 0;
- /**
- * The index into the hostDetails array of the broker to which we are connected
- */
+ /** The index into the hostDetails array of the broker to which we are connected */
private int _currentBrokerIndex = -1;
- /**
- * The number of times to retry connecting for each server
- */
+ /** The number of times to retry connecting for each server */
private int _serverRetries;
- /**
- * The current number of retry attempts made
- */
+ /** The current number of retry attempts made */
private int _currentServerRetry;
- /**
- * The number of times to cycle through the servers
- */
+ /** The number of times to cycle through the servers */
private int _cycleRetries;
- /**
- * The current number of cycles performed.
- */
+ /** The current number of cycles performed. */
private int _currentCycleRetries;
- /**
- * Array of BrokerDetail used to make connections.
- */
+ /** Array of BrokerDetail used to make connections. */
private ConnectionURL _connectionDetails;
public FailoverRoundRobinServers(ConnectionURL connectionDetails)
@@ -128,6 +115,8 @@
public BrokerDetails getNextBrokerDetails()
{
+ boolean doDelay = false;
+
if (_currentBrokerIndex == (_connectionDetails.getBrokerCount() - 1))
{
if (_currentServerRetry < _serverRetries)
@@ -143,6 +132,7 @@
else
{
_logger.info("Retrying " + _connectionDetails.getBrokerDetails(_currentBrokerIndex));
+ doDelay=true;
}
_currentServerRetry++;
@@ -175,6 +165,7 @@
else
{
_logger.info("Retrying " + _connectionDetails.getBrokerDetails(_currentBrokerIndex));
+ doDelay=true;
}
_currentServerRetry++;
@@ -189,7 +180,28 @@
}
}
- return _connectionDetails.getBrokerDetails(_currentBrokerIndex);
+ BrokerDetails broker = _connectionDetails.getBrokerDetails(_currentBrokerIndex);
+
+ String delayStr = broker.getProperty(BrokerDetails.OPTIONS_CONNECT_DELAY);
+ if (delayStr != null && doDelay)
+ {
+ Long delay = Long.parseLong(delayStr);
+ _logger.info("Delay between connect retries:" + delay);
+ try
+ {
+ Thread.sleep(delay);
+ }
+ catch (InterruptedException ie)
+ {
+ return null;
+ }
+ }
+ else
+ {
+ _logger.info("No delay between connect retries, use tcp://host:port?connectdelay='value' to enable.");
+ }
+
+ return broker;
}
public void setBroker(BrokerDetails broker)
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverSingleServer.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverSingleServer.java?rev=651133&r1=651132&r2=651133&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverSingleServer.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverSingleServer.java Wed Apr 23 18:54:20 2008
@@ -22,25 +22,23 @@
import org.apache.qpid.jms.BrokerDetails;
import org.apache.qpid.jms.ConnectionURL;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class FailoverSingleServer implements FailoverMethod
{
+ private static final Logger _logger = LoggerFactory.getLogger(FailoverSingleServer.class);
+
/** The default number of times to rety a conection to this server */
public static final int DEFAULT_SERVER_RETRIES = 1;
- /**
- * The details of the Single Server
- */
+ /** The details of the Single Server */
private BrokerDetails _brokerDetail;
- /**
- * The number of times to retry connecting to the sever
- */
+ /** The number of times to retry connecting to the sever */
private int _retries;
- /**
- * The current number of attempts made to the server
- */
+ /** The current number of attempts made to the server */
private int _currentRetries;
@@ -78,7 +76,7 @@
public BrokerDetails getCurrentBrokerDetails()
{
- return _brokerDetail;
+ return _brokerDetail;
}
public BrokerDetails getNextBrokerDetails()
@@ -91,11 +89,29 @@
{
if (_currentRetries < _retries)
{
- _currentRetries ++;
+ _currentRetries++;
}
+ }
+
- return _brokerDetail;
+ String delayStr = _brokerDetail.getProperty(BrokerDetails.OPTIONS_CONNECT_DELAY);
+ if (delayStr != null && _currentRetries != 1)
+ {
+ Long delay = Long.parseLong(delayStr);
+ _logger.info("Delay between connect retries:" + delay);
+ try
+ {
+
+ Thread.sleep(delay);
+ }
+ catch (InterruptedException ie)
+ {
+ _logger.info("No delay between connect retries, use tcp://host:port?connectdelay='value' to enable.");
+ return null;
+ }
}
+
+ return _brokerDetail;
}
public void setBroker(BrokerDetails broker)
@@ -138,10 +154,10 @@
public String toString()
{
- return "SingleServer:\n"+
- "Max Retries:"+_retries+
- "\nCurrent Retry:"+_currentRetries+
- "\n"+_brokerDetail+"\n";
+ return "SingleServer:\n" +
+ "Max Retries:" + _retries +
+ "\nCurrent Retry:" + _currentRetries +
+ "\n" + _brokerDetail + "\n";
}
}
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?rev=651133&r1=651132&r2=651133&view=diff
==============================================================================
--- 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 Wed Apr 23 18:54:20 2008
@@ -99,8 +99,6 @@
_logger.warn("Unable to load property file specified in Provider_URL:" + environment.get(Context.PROVIDER_URL));
}
-
-
createConnectionFactories(data, environment);
createDestinations(data, environment);
Modified: incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation/TestLargePublisher.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation/TestLargePublisher.java?rev=651133&r1=651132&r2=651133&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation/TestLargePublisher.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation/TestLargePublisher.java Wed Apr 23 18:54:20 2008
@@ -183,7 +183,7 @@
}
catch (UnknownHostException e)
{
- e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
+ e.printStackTrace();
}
catch (AMQException e)
{
Modified: incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/pubsub1/TestPublisher.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/pubsub1/TestPublisher.java?rev=651133&r1=651132&r2=651133&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/pubsub1/TestPublisher.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/pubsub1/TestPublisher.java Wed Apr 23 18:54:20 2008
@@ -133,7 +133,7 @@
}
catch (JMSException e)
{
- e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
+ e.printStackTrace();
}
}
@@ -163,7 +163,7 @@
}
catch (UnknownHostException e)
{
- e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
+ e.printStackTrace();
}
catch (AMQException e)
{
Modified: incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/AMQQueueDeferredOrderingTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/AMQQueueDeferredOrderingTest.java?rev=651133&r1=651132&r2=651133&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/AMQQueueDeferredOrderingTest.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/AMQQueueDeferredOrderingTest.java Wed Apr 23 18:54:20 2008
@@ -126,7 +126,7 @@
_logger.info("Consuming messages");
for (int i = 0; i < NUM_MESSAGES; i++)
{
- Message msg = consumer.receive(1500);
+ Message msg = consumer.receive(3000);
assertNotNull("Message should not be null", msg);
assertTrue("Message should be a text message", msg instanceof TextMessage);
assertEquals("Message content does not match expected", Integer.toString(i), ((TextMessage) msg).getText());
Modified: incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/MessageListenerMultiConsumerTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/MessageListenerMultiConsumerTest.java?rev=651133&r1=651132&r2=651133&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/MessageListenerMultiConsumerTest.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/MessageListenerMultiConsumerTest.java Wed Apr 23 18:54:20 2008
@@ -20,13 +20,17 @@
*/
package org.apache.qpid.client;
+import junit.framework.TestCase;
+import org.apache.qpid.client.transport.TransportConnection;
+import org.apache.qpid.jndi.PropertiesFileInitialContextFactory;
import org.apache.qpid.testutil.QpidTestCase;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
@@ -34,7 +38,9 @@
import javax.jms.Queue;
import javax.jms.Session;
import javax.naming.Context;
+import javax.naming.spi.InitialContextFactory;
+import java.util.Hashtable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
Modified: incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/MessageListenerTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/MessageListenerTest.java?rev=651133&r1=651132&r2=651133&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/MessageListenerTest.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/MessageListenerTest.java Wed Apr 23 18:54:20 2008
@@ -20,13 +20,17 @@
*/
package org.apache.qpid.client;
+import junit.framework.TestCase;
+import org.apache.qpid.client.transport.TransportConnection;
+import org.apache.qpid.jndi.PropertiesFileInitialContextFactory;
import org.apache.qpid.testutil.QpidTestCase;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
@@ -34,6 +38,9 @@
import javax.jms.Queue;
import javax.jms.Session;
import javax.naming.Context;
+import javax.naming.spi.InitialContextFactory;
+
+import java.util.Hashtable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
Modified: incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/ResetMessageListenerTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/ResetMessageListenerTest.java?rev=651133&r1=651132&r2=651133&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/ResetMessageListenerTest.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/ResetMessageListenerTest.java Wed Apr 23 18:54:20 2008
@@ -29,7 +29,16 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.jms.*;
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageListener;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.spi.InitialContextFactory;
@@ -65,12 +74,15 @@
private final CountDownLatch _allFirstMessagesSent = new CountDownLatch(2); // all messages Sent Lock
private final CountDownLatch _allSecondMessagesSent = new CountDownLatch(2); // all messages Sent Lock
private final CountDownLatch _allFirstMessagesSent010 = new CountDownLatch(MSG_COUNT); // all messages Sent Lock
- private final CountDownLatch _allSecondMessagesSent010 = new CountDownLatch(MSG_COUNT); // all messages Sent Lock
+ private final CountDownLatch _allSecondMessagesSent010 = new CountDownLatch(MSG_COUNT); // all messages Sent Lock
+
+ private String oldImmediatePrefetch;
protected void setUp() throws Exception
{
super.setUp();
+ oldImmediatePrefetch = System.getProperty(AMQSession.IMMEDIATE_PREFETCH);
System.setProperty(AMQSession.IMMEDIATE_PREFETCH, "true");
_clientConnection = getConnection("guest", "guest");
@@ -109,8 +121,12 @@
{
_clientConnection.close();
- _producerConnection.close();
super.tearDown();
+ if (oldImmediatePrefetch == null)
+ {
+ oldImmediatePrefetch = AMQSession.IMMEDIATE_PREFETCH_DEFAULT;
+ }
+ System.setProperty(AMQSession.IMMEDIATE_PREFETCH, oldImmediatePrefetch);
}
public void testAsynchronousRecieve()
@@ -238,7 +254,7 @@
try
{
- _allSecondMessagesSent.await(1000, TimeUnit.MILLISECONDS);
+ _allSecondMessagesSent.await(5000, TimeUnit.MILLISECONDS);
}
catch (InterruptedException e)
{
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?rev=651133&r1=651132&r2=651133&view=diff
==============================================================================
--- 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 Wed Apr 23 18:54:20 2008
@@ -60,6 +60,7 @@
private final List<String> messages = new ArrayList<String>();
private int _count = 1;
public String _connectionString = "vm://:1";
+ private static final String USERNAME = "guest";
protected void setUp() throws Exception
{
@@ -171,7 +172,7 @@
m.setJMSReplyTo(q);
m.setStringProperty("TempQueue", q.toString());
- _logger.trace("Message:" + m);
+ _logger.debug("Message:" + m);
Assert.assertEquals("Check temp queue has been set correctly", m.getJMSReplyTo().toString(),
m.getStringProperty("TempQueue"));
@@ -287,7 +288,14 @@
((AMQMessage) m).setVoidProperty(new AMQShortString("void"));
Assert.assertTrue("Check void properties are correctly transported",
- ((AMQMessage) m).getPropertyHeaders().containsKey("void"));
+ ((AMQMessage) m).getPropertyHeaders().containsKey("void"));
+
+ //JMSXUserID
+ if (m.getStringProperty("JMSXUserID") != null)
+ {
+ Assert.assertEquals("Check 'JMSXUserID' is supported ", USERNAME,
+ m.getStringProperty("JMSXUserID"));
+ }
}
received.clear();
Modified: incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/SelectorTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/SelectorTest.java?rev=651133&r1=651132&r2=651133&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/SelectorTest.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/SelectorTest.java Wed Apr 23 18:54:20 2008
@@ -21,7 +21,7 @@
package org.apache.qpid.test.unit.basic;
import junit.framework.TestCase;
-
+import org.apache.qpid.AMQException;
import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.client.AMQDestination;
import org.apache.qpid.client.AMQQueue;
@@ -29,11 +29,14 @@
import org.apache.qpid.client.BasicMessageProducer;
import org.apache.qpid.client.transport.TransportConnection;
import org.apache.qpid.testutil.QpidTestCase;
+import org.apache.qpid.url.URLSyntaxException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.jms.Connection;
import javax.jms.DeliveryMode;
+import javax.jms.InvalidSelectorException;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
@@ -47,11 +50,12 @@
private AMQSession _session;
private int count;
public String _connectionString = "vm://:1";
+ private static final String INVALID_SELECTOR = "Cost LIKE 5";
protected void setUp() throws Exception
{
super.setUp();
- init((AMQConnection) getConnection("guest", "guest"));
+ init((AMQConnection) getConnection("guest", "guest"));
}
protected void tearDown() throws Exception
@@ -59,19 +63,19 @@
super.tearDown();
}
- private void init(AMQConnection connection) throws Exception
+ private void init(AMQConnection connection) throws JMSException
{
init(connection, new AMQQueue(connection, randomize("SessionStartTest"), true));
}
- private void init(AMQConnection connection, AMQDestination destination) throws Exception
+ private void init(AMQConnection connection, AMQDestination destination) throws JMSException
{
_connection = connection;
_destination = destination;
connection.start();
String selector = null;
- // selector = "Cost = 2 AND JMSDeliveryMode=" + DeliveryMode.NON_PERSISTENT;
+ selector = "Cost = 2 AND \"property-with-hyphen\" = 'wibble'";
// selector = "JMSType = Special AND Cost = 2 AND AMQMessageID > 0 AND JMSDeliveryMode=" + DeliveryMode.NON_PERSISTENT;
_session = (AMQSession) connection.createSession(false, AMQSession.NO_ACKNOWLEDGE);
@@ -79,13 +83,17 @@
_session.createConsumer(destination, selector).setMessageListener(this);
}
- public synchronized void test() throws JMSException, InterruptedException
+ public synchronized void test() throws Exception
{
try
{
+
+ init((AMQConnection) getConnection("guest", "guest", randomize("Client")));
+
Message msg = _session.createTextMessage("Message");
msg.setJMSPriority(1);
msg.setIntProperty("Cost", 2);
+ msg.setStringProperty("property-with-hyphen", "wibble");
msg.setJMSType("Special");
_logger.info("Sending Message:" + msg);
@@ -105,10 +113,147 @@
// throw new RuntimeException("Did not get message!");
}
}
+ catch (JMSException e)
+ {
+ _logger.debug("JMS:" + e.getClass().getSimpleName() + ":" + e.getMessage());
+ if (!(e instanceof InvalidSelectorException))
+ {
+ fail("Wrong exception:" + e.getMessage());
+ }
+ else
+ {
+ System.out.println("SUCCESS!!");
+ }
+ }
+ catch (InterruptedException e)
+ {
+ _logger.debug("IE :" + e.getClass().getSimpleName() + ":" + e.getMessage());
+ }
+ catch (URLSyntaxException e)
+ {
+ _logger.debug("URL:" + e.getClass().getSimpleName() + ":" + e.getMessage());
+ fail("Wrong exception");
+ }
+ catch (AMQException e)
+ {
+ _logger.debug("AMQ:" + e.getClass().getSimpleName() + ":" + e.getMessage());
+ fail("Wrong exception");
+ }
+
+ finally
+ {
+ if (_session != null)
+ {
+ _session.close();
+ }
+ if (_connection != null)
+ {
+ _connection.close();
+ }
+ }
+ }
+
+
+ public void testInvalidSelectors() throws Exception
+ {
+ Connection connection = null;
+
+ try
+ {
+ connection = getConnection("guest", "guest", randomize("Client"));
+ _session = (AMQSession) connection.createSession(false, AMQSession.NO_ACKNOWLEDGE);
+ }
+ catch (JMSException e)
+ {
+ fail(e.getMessage());
+ }
+ catch (AMQException e)
+ {
+ fail(e.getMessage());
+ }
+ catch (URLSyntaxException e)
+ {
+ fail("Error:" + e.getMessage());
+ }
+
+ //Try Creating a Browser
+ try
+ {
+ _session.createBrowser(_session.createQueue("Ping"), INVALID_SELECTOR);
+ }
+ catch (JMSException e)
+ {
+ _logger.debug("JMS:" + e.getClass().getSimpleName() + ":" + e.getMessage());
+ if (!(e instanceof InvalidSelectorException))
+ {
+ fail("Wrong exception:" + e.getMessage());
+ }
+ else
+ {
+ _logger.debug("SUCCESS!!");
+ }
+ }
+
+ //Try Creating a Consumer
+ try
+ {
+ _session.createConsumer(_session.createQueue("Ping"), INVALID_SELECTOR);
+ }
+ catch (JMSException e)
+ {
+ _logger.debug("JMS:" + e.getClass().getSimpleName() + ":" + e.getMessage());
+ if (!(e instanceof InvalidSelectorException))
+ {
+ fail("Wrong exception:" + e.getMessage());
+ }
+ else
+ {
+ _logger.debug("SUCCESS!!");
+ }
+ }
+
+ //Try Creating a Receiever
+ try
+ {
+ _session.createReceiver(_session.createQueue("Ping"), INVALID_SELECTOR);
+ }
+ catch (JMSException e)
+ {
+ _logger.debug("JMS:" + e.getClass().getSimpleName() + ":" + e.getMessage());
+ if (!(e instanceof InvalidSelectorException))
+ {
+ fail("Wrong exception:" + e.getMessage());
+ }
+ else
+ {
+ _logger.debug("SUCCESS!!");
+ }
+ }
+
finally
{
- _session.close();
- _connection.close();
+ if (_session != null)
+ {
+ try
+ {
+ _session.close();
+ }
+ catch (JMSException e)
+ {
+ fail("Error cleaning up:" + e.getMessage());
+ }
+ }
+ if (_connection != null)
+ {
+ try
+ {
+ _connection.close();
+ }
+ catch (JMSException e)
+ {
+ fail("Error cleaning up:" + e.getMessage());
+ }
+ }
}
}
@@ -127,9 +272,29 @@
public static void main(String[] argv) throws Exception
{
SelectorTest test = new SelectorTest();
- test._connectionString = (argv.length == 0) ? "localhost:5672" : argv[0];
- test.setUp();
- test.test();
+ test._connectionString = (argv.length == 0) ? "localhost:3000" : argv[0];
+
+ try
+ {
+ while (true)
+ {
+ if (test._connectionString.contains("vm://:1"))
+ {
+ test.setUp();
+ }
+ test.test();
+
+ if (test._connectionString.contains("vm://:1"))
+ {
+ test.tearDown();
+ }
+ }
+ }
+ catch (Exception e)
+ {
+ System.err.println(e.getMessage());
+ e.printStackTrace();
+ }
}
public static junit.framework.Test suite()