You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ri...@apache.org on 2007/06/15 18:28:54 UTC
svn commit: r547730 [5/9] - in /incubator/qpid/trunk/qpid: ./ java/
java/broker/ java/broker/src/main/java/org/apache/qpid/server/
java/broker/src/main/java/org/apache/qpid/server/output/amqp0_8/
java/broker/src/main/java/org/apache/qpid/server/queue/ ...
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/security/CallbackHandlerRegistry.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/security/CallbackHandlerRegistry.java?view=diff&rev=547730&r1=547729&r2=547730
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/security/CallbackHandlerRegistry.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/security/CallbackHandlerRegistry.java Fri Jun 15 09:28:46 2007
@@ -20,10 +20,11 @@
*/
package org.apache.qpid.client.security;
-import java.io.BufferedInputStream;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileNotFoundException;
+import org.apache.qpid.util.FileUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.io.IOException;
import java.io.InputStream;
import java.util.Enumeration;
@@ -31,10 +32,6 @@
import java.util.Map;
import java.util.Properties;
-import org.apache.log4j.Logger;
-
-import org.apache.qpid.util.FileUtils;
-
/**
* CallbackHandlerRegistry is a registry for call back handlers for user authentication and interaction during user
* authentication. It is capable of reading its configuration from a properties file containing call back handler
@@ -60,7 +57,7 @@
*/
public class CallbackHandlerRegistry
{
- private static final Logger _logger = Logger.getLogger(CallbackHandlerRegistry.class);
+ private static final Logger _logger = LoggerFactory.getLogger(CallbackHandlerRegistry.class);
/** The name of the system property that holds the name of the callback handler properties file. */
private static final String FILE_PROPERTY = "amq.callbackhandler.properties";
@@ -121,7 +118,7 @@
String filename = System.getProperty(FILE_PROPERTY);
InputStream is =
FileUtils.openFileOrDefaultResource(filename, DEFAULT_RESOURCE_NAME,
- CallbackHandlerRegistry.class.getClassLoader());
+ CallbackHandlerRegistry.class.getClassLoader());
try
{
@@ -167,12 +164,12 @@
_logger.error("Unable to read from file " + filename + ": " + e, e);
}
}
-
+
if (useDefault)
{
is = CallbackHandlerRegistry.class.getResourceAsStream(DEFAULT_RESOURCE_NAME);
}
-
+
return is;
}*/
@@ -207,7 +204,7 @@
if (!AMQCallbackHandler.class.isAssignableFrom(clazz))
{
_logger.warn("SASL provider " + clazz + " does not implement " + AMQCallbackHandler.class
- + ". Skipping");
+ + ". Skipping");
continue;
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/security/DynamicSaslRegistrar.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/security/DynamicSaslRegistrar.java?view=diff&rev=547730&r1=547729&r2=547730
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/security/DynamicSaslRegistrar.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/security/DynamicSaslRegistrar.java Fri Jun 15 09:28:46 2007
@@ -20,6 +20,13 @@
*/
package org.apache.qpid.client.security;
+import org.apache.qpid.util.FileUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.security.sasl.SaslClientFactory;
+
import java.io.IOException;
import java.io.InputStream;
import java.security.Security;
@@ -28,13 +35,6 @@
import java.util.Properties;
import java.util.TreeMap;
-import javax.security.sasl.SaslClientFactory;
-
-
-import org.apache.log4j.Logger;
-
-import org.apache.qpid.util.FileUtils;
-
/**
* DynamicSaslRegistrar provides a collection of helper methods for reading a configuration file that contains a mapping
* from SASL mechanism names to implementing client factory class names and registering a security provider with the
@@ -55,7 +55,7 @@
*/
public class DynamicSaslRegistrar
{
- private static final Logger _logger = Logger.getLogger(DynamicSaslRegistrar.class);
+ private static final Logger _logger = LoggerFactory.getLogger(DynamicSaslRegistrar.class);
/** The name of the system property that holds the name of the SASL configuration properties. */
private static final String FILE_PROPERTY = "amq.dynamicsaslregistrar.properties";
@@ -71,8 +71,8 @@
// Open the SASL properties file, using the default name is one is not specified.
String filename = System.getProperty(FILE_PROPERTY);
InputStream is =
- FileUtils.openFileOrDefaultResource(filename, DEFAULT_RESOURCE_NAME,
- DynamicSaslRegistrar.class.getClassLoader());
+ FileUtils.openFileOrDefaultResource(filename, DEFAULT_RESOURCE_NAME,
+ DynamicSaslRegistrar.class.getClassLoader());
try
{
@@ -169,7 +169,7 @@
Enumeration e = props.propertyNames();
TreeMap<String, Class<? extends SaslClientFactory>> factoriesToRegister =
- new TreeMap<String, Class<? extends SaslClientFactory>>();
+ new TreeMap<String, Class<? extends SaslClientFactory>>();
while (e.hasMoreElements())
{
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/security/JCAProvider.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/security/JCAProvider.java?view=diff&rev=547730&r1=547729&r2=547730
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/security/JCAProvider.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/security/JCAProvider.java Fri Jun 15 09:28:46 2007
@@ -20,13 +20,13 @@
*/
package org.apache.qpid.client.security;
-import java.security.Provider;
-import java.security.Security;
-import java.util.Map;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import javax.security.sasl.SaslClientFactory;
-import org.apache.log4j.Logger;
+import java.security.Provider;
+import java.util.Map;
/**
* JCAProvider is a security provider for SASL client factories that is configured from a map of SASL mechanism names
@@ -40,7 +40,7 @@
*/
public class JCAProvider extends Provider
{
- private static final Logger log = Logger.getLogger(JCAProvider.class);
+ private static final Logger log = LoggerFactory.getLogger(JCAProvider.class);
/**
* Creates the security provider with a map from SASL mechanisms to implementing factories.
@@ -50,9 +50,9 @@
public JCAProvider(Map<String, Class<? extends SaslClientFactory>> providerMap)
{
super("AMQSASLProvider", 1.0, "A JCA provider that registers all "
- + "AMQ SASL providers that want to be registered");
+ + "AMQ SASL providers that want to be registered");
register(providerMap);
-// Security.addProvider(this);
+ // Security.addProvider(this);
}
/**
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/security/UsernameHashedPasswordCallbackHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/security/UsernameHashedPasswordCallbackHandler.java?view=diff&rev=547730&r1=547729&r2=547730
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/security/UsernameHashedPasswordCallbackHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/security/UsernameHashedPasswordCallbackHandler.java Fri Jun 15 09:28:46 2007
@@ -20,26 +20,24 @@
*/
package org.apache.qpid.client.security;
-import java.io.IOException;
-import java.io.UnsupportedEncodingException;
-import java.security.MessageDigest;
-import java.security.NoSuchAlgorithmException;
+import org.apache.qpid.client.protocol.AMQProtocolSession;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import javax.security.auth.callback.Callback;
import javax.security.auth.callback.NameCallback;
import javax.security.auth.callback.PasswordCallback;
import javax.security.auth.callback.UnsupportedCallbackException;
-import javax.security.sasl.RealmCallback;
-
-import com.sun.crypto.provider.HmacMD5;
-import org.apache.log4j.Logger;
-
-import org.apache.qpid.client.protocol.AMQProtocolSession;
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
public class UsernameHashedPasswordCallbackHandler implements AMQCallbackHandler
{
- private static final Logger _logger = Logger.getLogger(UsernameHashedPasswordCallbackHandler.class);
+ private static final Logger _logger = LoggerFactory.getLogger(UsernameHashedPasswordCallbackHandler.class);
private AMQProtocolSession _protocolSession;
@@ -91,11 +89,11 @@
byte[] digest = md.digest();
- char[] hash = new char[digest.length ];
+ char[] hash = new char[digest.length];
int index = 0;
for (byte b : digest)
- {
+ {
hash[index++] = (char) b;
}
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?view=diff&rev=547730&r1=547729&r2=547730
==============================================================================
--- 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 Fri Jun 15 09:28:46 2007
@@ -20,12 +20,6 @@
*/
package org.apache.qpid.client.state;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.concurrent.CopyOnWriteArraySet;
-
-import org.apache.log4j.Logger;
import org.apache.qpid.AMQException;
import org.apache.qpid.client.handler.BasicCancelOkMethodHandler;
import org.apache.qpid.client.handler.BasicDeliverMethodHandler;
@@ -58,13 +52,22 @@
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.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
{
- private static final Logger _logger = Logger.getLogger(AMQStateManager.class);
+ private static final Logger _logger = LoggerFactory.getLogger(AMQStateManager.class);
+
private AMQProtocolSession _protocolSession;
/** The current state */
@@ -78,14 +81,13 @@
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 = 30000L;
public AMQStateManager()
{
this(null);
}
-
public AMQStateManager(AMQProtocolSession protocolSession)
{
this(AMQState.CONNECTION_NOT_STARTED, true, protocolSession);
@@ -179,20 +181,22 @@
if (handler != null)
{
handler.methodReceived(this, _protocolSession, evt);
+
return true;
}
+
return false;
}
- protected StateAwareMethodListener findStateTransitionHandler(AMQState currentState,
- AMQMethodBody frame)
- // throws IllegalStateTransitionException
+ 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)
@@ -201,12 +205,14 @@
// 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
@@ -222,7 +228,6 @@
}
}
-
public void attainState(final AMQState s) throws AMQException
{
synchronized (_stateLock)
@@ -230,7 +235,7 @@
final long waitUntilTime = System.currentTimeMillis() + MAXIMUM_STATE_WAIT_TIME;
long waitTime = MAXIMUM_STATE_WAIT_TIME;
- while (_currentState != s && waitTime > 0)
+ while ((_currentState != s) && (waitTime > 0))
{
try
{
@@ -240,11 +245,13 @@
{
_logger.warn("Thread interrupted");
}
+
if (_currentState != s)
{
waitTime = waitUntilTime - System.currentTimeMillis();
}
}
+
if (_currentState != s)
{
_logger.warn("State not achieved within permitted time. Current state " + _currentState + ", desired state: " + s);
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/StateWaiter.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/StateWaiter.java?view=diff&rev=547730&r1=547729&r2=547730
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/StateWaiter.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/StateWaiter.java Fri Jun 15 09:28:46 2007
@@ -7,9 +7,9 @@
* 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
@@ -20,15 +20,17 @@
*/
package org.apache.qpid.client.state;
-import org.apache.log4j.Logger;
import org.apache.qpid.AMQException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
/**
* Waits for a particular state to be reached.
*/
public class StateWaiter implements StateListener
{
- private static final Logger _logger = Logger.getLogger(StateWaiter.class);
+ private static final Logger _logger = LoggerFactory.getLogger(StateWaiter.class);
private final AMQState _state;
@@ -52,13 +54,13 @@
// The guard is required in case we are woken up by a spurious
// notify().
//
- while (!_newStateAchieved && _throwable == null)
+ while (!_newStateAchieved && (_throwable == null))
{
try
{
_logger.debug("State " + _state + " not achieved so waiting...");
_monitor.wait(TIME_OUT);
- //fixme this won't cause the timeout to exit the loop. need to set _throwable
+ // fixme this won't cause the timeout to exit the loop. need to set _throwable
}
catch (InterruptedException e)
{
@@ -72,7 +74,7 @@
_logger.debug("Throwable reached state waiter: " + _throwable);
if (_throwable instanceof AMQException)
{
- throw(AMQException) _throwable;
+ throw (AMQException) _throwable;
}
else
{
@@ -89,6 +91,7 @@
{
_logger.debug("stateChanged called changing from :" + oldState + " to :" + newState);
}
+
if (_state == newState)
{
_newStateAchieved = true;
@@ -97,6 +100,7 @@
{
_logger.debug("New state reached so notifying monitor");
}
+
_monitor.notifyAll();
}
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/listener/SpecificMethodFrameListener.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/listener/SpecificMethodFrameListener.java?view=diff&rev=547730&r1=547729&r2=547730
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/listener/SpecificMethodFrameListener.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/listener/SpecificMethodFrameListener.java Fri Jun 15 09:28:46 2007
@@ -34,7 +34,7 @@
_expectedClass = expectedClass;
}
- public boolean processMethod(int channelId, AMQMethodBody frame) throws AMQException
+ public boolean processMethod(int channelId, AMQMethodBody frame) //throws AMQException
{
return _expectedClass.isInstance(frame);
}
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?view=diff&rev=547730&r1=547729&r2=547730
==============================================================================
--- 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 Fri Jun 15 09:28:46 2007
@@ -7,9 +7,9 @@
* 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
@@ -20,28 +20,32 @@
*/
package org.apache.qpid.client.transport;
-import java.io.IOException;
-import java.net.InetSocketAddress;
-
-import org.apache.log4j.Logger;
import org.apache.mina.common.ByteBuffer;
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.SocketConnectorConfig;
import org.apache.mina.transport.socket.nio.SocketSessionConfig;
+
import org.apache.qpid.client.protocol.AMQProtocolHandler;
import org.apache.qpid.jms.BrokerDetails;
import org.apache.qpid.pool.ReadWriteThreadModel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+
public class SocketTransportConnection implements ITransportConnection
{
- private static final Logger _logger = Logger.getLogger(SocketTransportConnection.class);
+ private static final Logger _logger = LoggerFactory.getLogger(SocketTransportConnection.class);
private static final int DEFAULT_BUFFER_SIZE = 32 * 1024;
private SocketConnectorFactory _socketConnectorFactory;
- static interface SocketConnectorFactory {
+ static interface SocketConnectorFactory
+ {
IoConnector newSocketConnector();
}
@@ -50,8 +54,7 @@
_socketConnectorFactory = socketConnectorFactory;
}
- public void connect(AMQProtocolHandler protocolHandler, BrokerDetails brokerDetail)
- throws IOException
+ public void connect(AMQProtocolHandler protocolHandler, BrokerDetails brokerDetail) throws IOException
{
ByteBuffer.setUseDirectBuffers(Boolean.getBoolean("amqj.enableDirectBuffers"));
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?view=diff&rev=547730&r1=547729&r2=547730
==============================================================================
--- 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 Fri Jun 15 09:28:46 2007
@@ -20,13 +20,6 @@
*/
package org.apache.qpid.client.transport;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.Map;
-
-import org.apache.log4j.Logger;
-
import org.apache.mina.common.IoConnector;
import org.apache.mina.common.IoHandlerAdapter;
import org.apache.mina.common.IoServiceConfig;
@@ -34,11 +27,18 @@
import org.apache.mina.transport.vmpipe.VmPipeAcceptor;
import org.apache.mina.transport.vmpipe.VmPipeAddress;
-import org.apache.qpid.client.AMQBrokerDetails;
import org.apache.qpid.client.vmbroker.AMQVMBrokerCreationException;
import org.apache.qpid.jms.BrokerDetails;
import org.apache.qpid.pool.ReadWriteThreadModel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+
/**
* The TransportConnection is a helper class responsible for connecting to an AMQ server. It sets up the underlying
* connector, which currently always uses TCP/IP sockets. It creates the "protocol handler" which deals with MINA
@@ -57,7 +57,7 @@
private static final int TCP = 0;
private static final int VM = 1;
- private static Logger _logger = Logger.getLogger(TransportConnection.class);
+ private static Logger _logger = LoggerFactory.getLogger(TransportConnection.class);
private static final String DEFAULT_QPID_SERVER = "org.apache.qpid.server.protocol.AMQPFastProtocolHandler";
@@ -99,7 +99,7 @@
// FIXME - this needs to be sorted to use the new Mina MultiThread SA.
if (Boolean.getBoolean("qpidnio"))
{
- _logger.fatal("Using Qpid NIO - sysproperty 'qpidnio' is set.");
+ _logger.error("Using Qpid NIO - sysproperty 'qpidnio' is set.");
// result = new org.apache.qpid.nio.SocketConnector(); // non-blocking connector
}
// else
@@ -193,7 +193,7 @@
}
catch (IOException e)
{
- _logger.error(e);
+ _logger.error("Got IOException.", e);
// Try and unbind provider
try
@@ -262,7 +262,6 @@
catch (Exception e)
{
_logger.info("Unable to create InVM Qpid.AMQP on port " + port + ". Because: " + e.getCause());
- _logger.error(e);
String because;
if (e.getCause() == null)
{
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?view=diff&rev=547730&r1=547729&r2=547730
==============================================================================
--- 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 Fri Jun 15 09:28:46 2007
@@ -7,9 +7,9 @@
* 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
@@ -20,21 +20,25 @@
*/
package org.apache.qpid.client.transport;
-import java.io.IOException;
-
-import org.apache.log4j.Logger;
import org.apache.mina.common.ConnectFuture;
import org.apache.mina.common.IoServiceConfig;
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;
+
+import java.io.IOException;
public class VmPipeTransportConnection implements ITransportConnection
{
- private static final Logger _logger = Logger.getLogger(VmPipeTransportConnection.class);
+ private static final Logger _logger = LoggerFactory.getLogger(VmPipeTransportConnection.class);
private static int _port;
@@ -47,14 +51,9 @@
{
final VmPipeConnector ioConnector = new VmPipeConnector();
final IoServiceConfig cfg = ioConnector.getDefaultConfig();
- ReferenceCountingExecutorService executorService = ReferenceCountingExecutorService.getInstance();
- PoolingFilter asyncRead = PoolingFilter.createAynschReadPoolingFilter(executorService,
- "AsynchronousReadFilter");
- cfg.getFilterChain().addFirst("AsynchronousReadFilter", asyncRead);
- PoolingFilter asyncWrite = PoolingFilter.createAynschWritePoolingFilter(executorService,
- "AsynchronousWriteFilter");
- cfg.getFilterChain().addLast("AsynchronousWriteFilter", asyncWrite);
-
+
+ cfg.setThreadModel(ReadWriteThreadModel.getInstance());
+
final VmPipeAddress address = new VmPipeAddress(_port);
_logger.info("Attempting connection to " + address);
ConnectFuture future = ioConnector.connect(address, protocolHandler);
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/util/FlowControllingBlockingQueue.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/util/FlowControllingBlockingQueue.java?view=diff&rev=547730&r1=547729&r2=547730
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/util/FlowControllingBlockingQueue.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/util/FlowControllingBlockingQueue.java Fri Jun 15 09:28:46 2007
@@ -7,9 +7,9 @@
* 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
@@ -20,16 +20,17 @@
*/
package org.apache.qpid.client.util;
-
+import java.util.Iterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
-import java.util.Iterator;
/**
* A blocking queue that emits events above a user specified threshold allowing the caller to take action (e.g. flow
* control) to try to prevent the queue growing (much) further. The underlying queue itself is not bounded therefore the
* caller is not obliged to react to the events. <p/> This implementation is <b>only</b> safe where we have a single
* thread adding items and a single (different) thread removing items.
+ *
+ * @todo Make this implement java.util.Queue and hide the implementation. Then different queue types can be substituted.
*/
public class FlowControllingBlockingQueue
{
@@ -81,6 +82,7 @@
}
}
}
+
return o;
}
@@ -104,4 +106,3 @@
return _queue.iterator();
}
}
-
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?view=diff&rev=547730&r1=547729&r2=547730
==============================================================================
--- 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 Fri Jun 15 09:28:46 2007
@@ -7,9 +7,9 @@
* 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
@@ -20,22 +20,23 @@
*/
package org.apache.qpid.jms;
-import org.apache.log4j.Logger;
import org.apache.qpid.jms.failover.FailoverMethod;
import org.apache.qpid.jms.failover.FailoverRoundRobinServers;
import org.apache.qpid.jms.failover.FailoverSingleServer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class FailoverPolicy
{
- private static final Logger _logger = Logger.getLogger(FailoverPolicy.class);
+ private static final Logger _logger = LoggerFactory.getLogger(FailoverPolicy.class);
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];
+ private FailoverMethod[] _methods = new FailoverMethod[1];
private int _currentMethod;
@@ -52,7 +53,7 @@
{
FailoverMethod method;
- //todo This should be integrated in to the connection url when it supports
+ // todo This should be integrated in to the connection url when it supports
// multiple strategies.
_methodsRetries = 0;
@@ -72,12 +73,12 @@
{
String failoverMethod = connectionDetails.getFailoverMethod();
-/*
- if (failoverMethod.equals(FailoverMethod.RANDOM))
- {
- //todo write a random connection Failover
- }
-*/
+ /*
+ if (failoverMethod.equals(FailoverMethod.RANDOM))
+ {
+ //todo write a random connection Failover
+ }
+ */
if (failoverMethod.equals(FailoverMethod.SINGLE_BROKER))
{
method = new FailoverRoundRobinServers(connectionDetails);
@@ -92,12 +93,12 @@
{
try
{
- Class[] constructorSpec = {ConnectionURL.class};
- Object [] params = {connectionDetails};
+ Class[] constructorSpec = { ConnectionURL.class };
+ Object[] params = { connectionDetails };
- method = (FailoverMethod) ClassLoader.getSystemClassLoader().
- loadClass(failoverMethod).
- getConstructor(constructorSpec).newInstance(params);
+ method =
+ (FailoverMethod) ClassLoader.getSystemClassLoader().loadClass(failoverMethod)
+ .getConstructor(constructorSpec).newInstance(params);
}
catch (Exception cnfe)
{
@@ -157,13 +158,13 @@
return false;
}
-
}
else
{
if ((now - _lastFailTime) >= DEFAULT_FAILOVER_TIMEOUT)
{
_logger.info("Failover timeout");
+
return false;
}
else
@@ -179,7 +180,6 @@
_lastFailTime = _lastMethodTime;
}
-
if (_methods[_currentMethod].failoverAllowed())
{
failoverAllowed = true;
@@ -190,6 +190,7 @@
{
nextMethod();
_logger.info("Changing method to " + _methods[_currentMethod].methodName());
+
return failoverAllowed();
}
else
@@ -207,6 +208,7 @@
{
_currentMethod++;
_methods[_currentMethod].reset();
+
return true;
}
else
@@ -225,11 +227,13 @@
_logger.info("Retrying methods starting with " + _methods[_currentMethod].methodName());
_methods[_currentMethod].reset();
+
return failoverAllowed();
}
else
{
_logger.debug("All failover methods exhausted");
+
return false;
}
}
@@ -278,7 +282,7 @@
public FailoverMethod getCurrentMethod()
{
- if (_currentMethod >= 0 && _currentMethod < (_methods.length - 1))
+ if ((_currentMethod >= 0) && (_currentMethod < (_methods.length - 1)))
{
return _methods[_currentMethod];
}
@@ -311,6 +315,7 @@
{
sb.append(">");
}
+
sb.append(_methods[i].toString());
}
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?view=diff&rev=547730&r1=547729&r2=547730
==============================================================================
--- 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 Fri Jun 15 09:28:46 2007
@@ -7,9 +7,9 @@
* 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
@@ -20,13 +20,15 @@
*/
package org.apache.qpid.jms.failover;
-import org.apache.log4j.Logger;
import org.apache.qpid.jms.BrokerDetails;
import org.apache.qpid.jms.ConnectionURL;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
public class FailoverRoundRobinServers implements FailoverMethod
{
- private static final Logger _logger = Logger.getLogger(FailoverRoundRobinServers.class);
+ private static final Logger _logger = LoggerFactory.getLogger(FailoverRoundRobinServers.class);
/** The default number of times to cycle through all servers */
public static final int DEFAULT_CYCLE_RETRIES = 0;
@@ -72,7 +74,7 @@
_connectionDetails = connectionDetails;
- //There is no current broker at startup so set it to -1.
+ // There is no current broker at startup so set it to -1.
_currentBrokerIndex = -1;
String cycleRetries = _connectionDetails.getFailoverOption(ConnectionURL.OPTIONS_FAILOVER_CYCLE);
@@ -104,9 +106,8 @@
public boolean failoverAllowed()
{
- return ((_currentCycleRetries < _cycleRetries)
- || (_currentServerRetry < _serverRetries)
- || (_currentBrokerIndex < (_connectionDetails.getBrokerCount() - 1)));
+ return ((_currentCycleRetries < _cycleRetries) || (_currentServerRetry < _serverRetries)
+ || (_currentBrokerIndex < (_connectionDetails.getBrokerCount() - 1)));
}
public void attainedConnection()
@@ -125,8 +126,6 @@
return _connectionDetails.getBrokerDetails(_currentBrokerIndex);
}
-
-
public BrokerDetails getNextBrokerDetails()
{
if (_currentBrokerIndex == (_connectionDetails.getBrokerCount() - 1))
@@ -137,7 +136,7 @@
{
_currentBrokerIndex = 0;
- setBroker(_connectionDetails.getBrokerDetails(_currentBrokerIndex ));
+ setBroker(_connectionDetails.getBrokerDetails(_currentBrokerIndex));
_logger.info("First run using " + _connectionDetails.getBrokerDetails(_currentBrokerIndex));
}
@@ -151,15 +150,15 @@
else
{
_currentCycleRetries++;
- //failed to connect to first broker
+ // failed to connect to first broker
_currentBrokerIndex = 0;
- setBroker(_connectionDetails.getBrokerDetails(_currentBrokerIndex ));
+ setBroker(_connectionDetails.getBrokerDetails(_currentBrokerIndex));
// This is zero rather than -1 as we are already retrieving the details.
_currentServerRetry = 0;
}
- //else - should force client to stop as max retries has been reached.
+ // else - should force client to stop as max retries has been reached.
}
else
{
@@ -169,7 +168,7 @@
{
_currentBrokerIndex = 0;
- setBroker(_connectionDetails.getBrokerDetails(_currentBrokerIndex ));
+ setBroker(_connectionDetails.getBrokerDetails(_currentBrokerIndex));
_logger.info("First run using " + _connectionDetails.getBrokerDetails(_currentBrokerIndex));
}
@@ -177,13 +176,14 @@
{
_logger.info("Retrying " + _connectionDetails.getBrokerDetails(_currentBrokerIndex));
}
+
_currentServerRetry++;
}
else
{
_currentBrokerIndex++;
- setBroker(_connectionDetails.getBrokerDetails(_currentBrokerIndex ));
+ setBroker(_connectionDetails.getBrokerDetails(_currentBrokerIndex));
// This is zero rather than -1 as we are already retrieving the details.
_currentServerRetry = 0;
}
@@ -192,7 +192,6 @@
return _connectionDetails.getBrokerDetails(_currentBrokerIndex);
}
-
public void setBroker(BrokerDetails broker)
{
@@ -246,12 +245,13 @@
sb.append(_currentBrokerIndex);
sb.append("\n");
- for(int i=0; i < _connectionDetails.getBrokerCount() ; i++)
+ for (int i = 0; i < _connectionDetails.getBrokerCount(); i++)
{
if (i == _currentBrokerIndex)
{
sb.append(">");
}
+
sb.append(_connectionDetails.getBrokerDetails(i));
sb.append("\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?view=diff&rev=547730&r1=547729&r2=547730
==============================================================================
--- 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 Fri Jun 15 09:28:46 2007
@@ -7,9 +7,9 @@
* 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
@@ -20,38 +20,40 @@
*/
package org.apache.qpid.jndi;
-import java.io.BufferedInputStream;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.util.Hashtable;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.ConcurrentHashMap;
-
-import javax.jms.ConnectionFactory;
-import javax.jms.Destination;
-import javax.jms.Queue;
-import javax.jms.Topic;
-import javax.naming.Context;
-import javax.naming.NamingException;
-import javax.naming.spi.InitialContextFactory;
-
-import org.apache.log4j.Logger;
import org.apache.qpid.client.AMQConnectionFactory;
import org.apache.qpid.client.AMQDestination;
import org.apache.qpid.client.AMQHeadersExchange;
import org.apache.qpid.client.AMQQueue;
import org.apache.qpid.client.AMQTopic;
+import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.url.AMQBindingURL;
import org.apache.qpid.url.BindingURL;
import org.apache.qpid.url.URLSyntaxException;
-import org.apache.qpid.exchange.ExchangeDefaults;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.jms.ConnectionFactory;
+import javax.jms.Destination;
+import javax.jms.Queue;
+import javax.jms.Topic;
+import javax.naming.Context;
+import javax.naming.NamingException;
+import javax.naming.spi.InitialContextFactory;
+
+import java.io.BufferedInputStream;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.util.Hashtable;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
public class PropertiesFileInitialContextFactory implements InitialContextFactory
{
- protected final Logger _logger = Logger.getLogger(PropertiesFileInitialContextFactory.class);
+ protected final Logger _logger = LoggerFactory.getLogger(PropertiesFileInitialContextFactory.class);
private String CONNECTION_FACTORY_PREFIX = "connectionfactory.";
private String DESTINATION_PREFIX = "destination.";
@@ -78,7 +80,7 @@
if (file != null)
{
_logger.info("Loading Properties from:" + file);
- //Load the properties specified
+ // Load the properties specified
Properties p = new Properties();
p.load(new BufferedInputStream(new FileInputStream(file)));
@@ -93,8 +95,7 @@
}
catch (IOException ioe)
{
- _logger.warn("Unable to load property file specified in Provider_URL:" +
- environment.get(Context.PROVIDER_URL));
+ _logger.warn("Unable to load property file specified in Provider_URL:" + environment.get(Context.PROVIDER_URL));
}
createConnectionFactories(data, environment);
@@ -109,7 +110,7 @@
}
// Implementation methods
- //-------------------------------------------------------------------------
+ // -------------------------------------------------------------------------
protected ReadOnlyContext createContext(Map data, Hashtable environment)
{
return new ReadOnlyContext(environment, data);
@@ -200,6 +201,7 @@
{
_logger.warn("Unable to createFactories:" + urlse);
}
+
return null;
}
@@ -216,6 +218,7 @@
catch (URLSyntaxException urlse)
{
_logger.warn("Unable to destination:" + urlse);
+
return null;
}
@@ -226,6 +229,7 @@
catch (IllegalArgumentException iaw)
{
_logger.warn("Binding: '" + binding + "' not supported");
+
return null;
}
}
@@ -235,17 +239,15 @@
*/
protected Queue createQueue(Object value)
{
- if(value instanceof AMQShortString)
+ if (value instanceof AMQShortString)
{
return new AMQQueue(ExchangeDefaults.DIRECT_EXCHANGE_NAME, (AMQShortString) value);
}
else if (value instanceof String)
-
{
return new AMQQueue(ExchangeDefaults.DIRECT_EXCHANGE_NAME, new AMQShortString((String) value));
}
else if (value instanceof BindingURL)
-
{
return new AMQQueue((BindingURL) value);
}
@@ -258,16 +260,15 @@
*/
protected Topic createTopic(Object value)
{
- if(value instanceof AMQShortString)
+ if (value instanceof AMQShortString)
{
- return new AMQTopic(ExchangeDefaults.TOPIC_EXCHANGE_NAME, (AMQShortString)value);
+ return new AMQTopic(ExchangeDefaults.TOPIC_EXCHANGE_NAME, (AMQShortString) value);
}
else if (value instanceof String)
{
return new AMQTopic(ExchangeDefaults.TOPIC_EXCHANGE_NAME, new AMQShortString((String) value));
}
else if (value instanceof BindingURL)
-
{
return new AMQTopic((BindingURL) value);
}
@@ -293,7 +294,7 @@
}
// Properties
- //-------------------------------------------------------------------------
+ // -------------------------------------------------------------------------
public String getConnectionPrefix()
{
return CONNECTION_FACTORY_PREFIX;
Modified: incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/DispatcherTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/DispatcherTest.java?view=diff&rev=547730&r1=547729&r2=547730
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/DispatcherTest.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/DispatcherTest.java Fri Jun 15 09:28:46 2007
@@ -20,9 +20,13 @@
*/
package org.apache.qpid.client;
-import java.util.Hashtable;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
+import junit.framework.TestCase;
+
+import org.apache.qpid.client.transport.TransportConnection;
+import org.apache.qpid.jndi.PropertiesFileInitialContextFactory;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
@@ -36,11 +40,9 @@
import javax.naming.Context;
import javax.naming.spi.InitialContextFactory;
-import junit.framework.TestCase;
-
-import org.apache.log4j.Logger;
-import org.apache.qpid.client.transport.TransportConnection;
-import org.apache.qpid.jndi.PropertiesFileInitialContextFactory;
+import java.util.Hashtable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
/**
* QPID-293 Setting MessageListener after connection has started can cause messages to be "lost" on a internal delivery queue
@@ -56,7 +58,7 @@
*/
public class DispatcherTest extends TestCase
{
- private static final Logger _logger = Logger.getLogger(DispatcherTest.class);
+ private static final Logger _logger = LoggerFactory.getLogger(DispatcherTest.class);
Context _context;
@@ -68,8 +70,8 @@
MessageProducer _producer;
Session _clientSession, _producerSession;
- private final CountDownLatch _allFirstMessagesSent = new CountDownLatch(1); //all messages Sent Lock
- private final CountDownLatch _allSecondMessagesSent = new CountDownLatch(1); //all messages Sent Lock
+ private final CountDownLatch _allFirstMessagesSent = new CountDownLatch(1); // all messages Sent Lock
+ private final CountDownLatch _allSecondMessagesSent = new CountDownLatch(1); // all messages Sent Lock
private volatile boolean _connectionStopped = false;
@@ -89,14 +91,14 @@
Queue queue = (Queue) _context.lookup("queue");
- //Create Client 1
+ // Create Client 1
_clientConnection = ((ConnectionFactory) _context.lookup("connection")).createConnection();
_clientSession = _clientConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
_consumer = _clientSession.createConsumer(queue);
- //Create Producer
+ // Create Producer
_producerConnection = ((ConnectionFactory) _context.lookup("connection")).createConnection();
_producerConnection.start();
@@ -109,7 +111,6 @@
{
_producer.send(_producerSession.createTextMessage("Message " + msg));
}
-
}
protected void tearDown() throws Exception
@@ -122,67 +123,63 @@
TransportConnection.killAllVMBrokers();
}
-
public void testAsynchronousRecieve()
{
-
_logger.info("Test Start");
-
assertTrue(!((AMQConnection) _clientConnection).started());
- //Set default Message Listener
+ // Set default Message Listener
try
{
_consumer.setMessageListener(new MessageListener()
- {
- public void onMessage(Message message)
{
- _logger.info("Client 1 ML 1 Received Message(" + _receivedCount + "):" + message);
+ public void onMessage(Message message)
+ {
+ _logger.info("Client 1 ML 1 Received Message(" + _receivedCount + "):" + message);
- _receivedCount++;
+ _receivedCount++;
- if (_receivedCount == MSG_COUNT)
- {
- _allFirstMessagesSent.countDown();
- }
+ if (_receivedCount == MSG_COUNT)
+ {
+ _allFirstMessagesSent.countDown();
+ }
- if (_connectionStopped)
- {
- _logger.info("Running with Message:" + _receivedCount);
- }
+ if (_connectionStopped)
+ {
+ _logger.info("Running with Message:" + _receivedCount);
+ }
- if (_connectionStopped && _allFirstMessagesSent.getCount() == 0)
- {
- _receivedCountWhileStopped++;
- }
+ if (_connectionStopped && (_allFirstMessagesSent.getCount() == 0))
+ {
+ _receivedCountWhileStopped++;
+ }
- if (_allFirstMessagesSent.getCount() == 0)
- {
- if (_receivedCount == MSG_COUNT * 2)
+ if (_allFirstMessagesSent.getCount() == 0)
{
- _allSecondMessagesSent.countDown();
+ if (_receivedCount == (MSG_COUNT * 2))
+ {
+ _allSecondMessagesSent.countDown();
+ }
}
}
- }
- });
-
+ });
+
assertTrue("Connecion should not be started", !((AMQConnection) _clientConnection).started());
- _clientConnection.start();
+ _clientConnection.start();
}
catch (JMSException e)
{
_logger.error("Error Setting Default ML on consumer1");
}
-
try
{
_allFirstMessagesSent.await(1000, TimeUnit.MILLISECONDS);
}
catch (InterruptedException e)
{
- //do nothing
+ // do nothing
}
try
@@ -196,7 +193,6 @@
_logger.error("Error stopping connection");
}
-
try
{
_logger.error("Send additional messages");
@@ -211,14 +207,13 @@
_logger.error("Unable to send additional messages", e);
}
-
try
{
Thread.sleep(1000);
}
catch (InterruptedException e)
{
- //ignore
+ // ignore
}
try
@@ -233,7 +228,6 @@
_logger.error("Error Setting Better ML on consumer1", e);
}
-
_logger.info("Waiting upto 2 seconds for messages");
try
@@ -242,16 +236,14 @@
}
catch (InterruptedException e)
{
- //do nothing
+ // do nothing
}
assertEquals("Messages not received correctly", 0, _allFirstMessagesSent.getCount());
assertEquals("Messages not received correctly", 0, _allSecondMessagesSent.getCount());
assertEquals("Client didn't get all messages", MSG_COUNT * 2, _receivedCount);
assertEquals("Messages received while stopped is not 0", 0, _receivedCountWhileStopped);
-
}
-
public static junit.framework.Test suite()
{
Modified: incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/MessageListenerMultiConsumerImmediatePrefetch.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/MessageListenerMultiConsumerImmediatePrefetch.java?view=diff&rev=547730&r1=547729&r2=547730
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/MessageListenerMultiConsumerImmediatePrefetch.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/MessageListenerMultiConsumerImmediatePrefetch.java Fri Jun 15 09:28:46 2007
@@ -20,27 +20,6 @@
*/
package org.apache.qpid.client;
-import java.util.Hashtable;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
-import javax.jms.Connection;
-import javax.jms.ConnectionFactory;
-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.naming.Context;
-import javax.naming.spi.InitialContextFactory;
-
-import junit.framework.TestCase;
-
-import org.apache.log4j.Logger;
-import org.apache.qpid.client.transport.TransportConnection;
-import org.apache.qpid.jndi.PropertiesFileInitialContextFactory;
-
/**
* QPID-293 Setting MessageListener after connection has started can cause messages to be "lost" on a internal delivery
* queue <p/> The message delivery process: Mina puts a message on _queue in AMQSession and the dispatcher thread
@@ -52,16 +31,11 @@
*/
public class MessageListenerMultiConsumerImmediatePrefetch extends MessageListenerMultiConsumerTest
{
-
-
protected void setUp() throws Exception
{
-
System.setProperty(AMQSession.IMMEDIATE_PREFETCH, "true");
super.setUp();
-
}
-
public static junit.framework.Test suite()
{
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?view=diff&rev=547730&r1=547729&r2=547730
==============================================================================
--- 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 Fri Jun 15 09:28:46 2007
@@ -20,9 +20,13 @@
*/
package org.apache.qpid.client;
-import java.util.Hashtable;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
+import junit.framework.TestCase;
+
+import org.apache.qpid.client.transport.TransportConnection;
+import org.apache.qpid.jndi.PropertiesFileInitialContextFactory;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
@@ -35,11 +39,9 @@
import javax.naming.Context;
import javax.naming.spi.InitialContextFactory;
-import junit.framework.TestCase;
-
-import org.apache.log4j.Logger;
-import org.apache.qpid.client.transport.TransportConnection;
-import org.apache.qpid.jndi.PropertiesFileInitialContextFactory;
+import java.util.Hashtable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
/**
* QPID-293 Setting MessageListener after connection has started can cause messages to be "lost" on a internal delivery
@@ -52,7 +54,7 @@
*/
public class MessageListenerMultiConsumerTest extends TestCase
{
- private static final Logger _logger = Logger.getLogger(MessageListenerMultiConsumerTest.class);
+ private static final Logger _logger = LoggerFactory.getLogger(MessageListenerMultiConsumerTest.class);
Context _context;
@@ -64,8 +66,7 @@
private MessageConsumer _consumer2;
private Session _clientSession1;
private Queue _queue;
- private final CountDownLatch _allMessagesSent = new CountDownLatch(2); //all messages Sent Lock
-
+ private final CountDownLatch _allMessagesSent = new CountDownLatch(2); // all messages Sent Lock
protected void setUp() throws Exception
{
@@ -77,13 +78,13 @@
Hashtable<String, String> env = new Hashtable<String, String>();
env.put("connectionfactory.connection", "amqp://guest:guest@MLT_ID/test?brokerlist='vm://:1'");
- env.put("queue.queue", "direct://amq.direct//"+this.getClass().getName());
+ env.put("queue.queue", "direct://amq.direct//" + this.getClass().getName());
_context = factory.getInitialContext(env);
_queue = (Queue) _context.lookup("queue");
- //Create Client 1
+ // Create Client 1
_clientConnection = ((ConnectionFactory) _context.lookup("connection")).createConnection();
_clientConnection.start();
@@ -92,17 +93,16 @@
_consumer1 = _clientSession1.createConsumer(_queue);
- //Create Client 2
+ // Create Client 2
Session clientSession2 = _clientConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
_consumer2 = clientSession2.createConsumer(_queue);
- //Create Producer
+ // Create Producer
Connection producerConnection = ((ConnectionFactory) _context.lookup("connection")).createConnection();
producerConnection.start();
-
Session producerSession = producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
MessageProducer producer = producerSession.createProducer(_queue);
@@ -124,18 +124,18 @@
TransportConnection.killAllVMBrokers();
}
-
public void testRecieveInterleaved() throws Exception
{
int msg = 0;
int MAX_LOOPS = MSG_COUNT * 2;
- for (int loops = 0; msg < MSG_COUNT || loops < MAX_LOOPS; loops++)
+ for (int loops = 0; (msg < MSG_COUNT) || (loops < MAX_LOOPS); loops++)
{
if (_consumer1.receive(100) != null)
{
msg++;
}
+
if (_consumer2.receive(100) != null)
{
msg++;
@@ -145,39 +145,37 @@
assertEquals("Not all messages received.", MSG_COUNT, msg);
}
-
public void testAsynchronousRecieve() throws Exception
{
_consumer1.setMessageListener(new MessageListener()
- {
- public void onMessage(Message message)
{
- _logger.info("Client 1 Received Message(" + receivedCount1 + "):" + message);
+ public void onMessage(Message message)
+ {
+ _logger.info("Client 1 Received Message(" + receivedCount1 + "):" + message);
- receivedCount1++;
+ receivedCount1++;
- if (receivedCount1 == MSG_COUNT / 2)
- {
- _allMessagesSent.countDown();
- }
+ if (receivedCount1 == (MSG_COUNT / 2))
+ {
+ _allMessagesSent.countDown();
+ }
- }
- });
+ }
+ });
_consumer2.setMessageListener(new MessageListener()
- {
- public void onMessage(Message message)
{
- _logger.info("Client 2 Received Message(" + receivedCount2 + "):" + message);
-
- receivedCount2++;
- if (receivedCount2 == MSG_COUNT / 2)
+ public void onMessage(Message message)
{
- _allMessagesSent.countDown();
- }
- }
- });
+ _logger.info("Client 2 Received Message(" + receivedCount2 + "):" + message);
+ receivedCount2++;
+ if (receivedCount2 == (MSG_COUNT / 2))
+ {
+ _allMessagesSent.countDown();
+ }
+ }
+ });
_logger.info("Waiting upto 2 seconds for messages");
@@ -187,7 +185,7 @@
}
catch (InterruptedException e)
{
- //do nothing
+ // do nothing
}
assertEquals(MSG_COUNT, receivedCount1 + receivedCount2);
@@ -195,34 +193,37 @@
public void testRecieveC2Only() throws Exception
{
- if (!Boolean.parseBoolean(System.getProperties().
- getProperty(AMQSession.IMMEDIATE_PREFETCH, AMQSession.IMMEDIATE_PREFETCH_DEFAULT)))
+ if (
+ !Boolean.parseBoolean(
+ System.getProperties().getProperty(AMQSession.IMMEDIATE_PREFETCH,
+ AMQSession.IMMEDIATE_PREFETCH_DEFAULT)))
{
_logger.info("Performing Receive only on C2");
for (int msg = 0; msg < MSG_COUNT; msg++)
{
- assertTrue(MSG_COUNT + " msg should be received. Only received:" + msg,
- _consumer2.receive(1000) != null);
+ assertTrue(MSG_COUNT + " msg should be received. Only received:" + msg, _consumer2.receive(1000) != null);
}
}
}
public void testRecieveBoth() throws Exception
{
- if (!Boolean.parseBoolean(System.getProperties().
- getProperty(AMQSession.IMMEDIATE_PREFETCH, AMQSession.IMMEDIATE_PREFETCH_DEFAULT)))
+ if (
+ !Boolean.parseBoolean(
+ System.getProperties().getProperty(AMQSession.IMMEDIATE_PREFETCH,
+ AMQSession.IMMEDIATE_PREFETCH_DEFAULT)))
{
_logger.info("Performing Receive only with two consumers on one session ");
MessageConsumer consumer2 = _clientSession1.createConsumer(_queue);
- for (int msg = 0; msg < MSG_COUNT / 2; msg++)
+ for (int msg = 0; msg < (MSG_COUNT / 2); msg++)
{
assertTrue(_consumer1.receive() != null);
}
- for (int msg = 0; msg < MSG_COUNT / 2; msg++)
+ for (int msg = 0; msg < (MSG_COUNT / 2); msg++)
{
assertTrue(consumer2.receive() != null);
}
@@ -231,19 +232,18 @@
{
_logger.info("Performing Receive only on both C1 and C2");
- for (int msg = 0; msg < MSG_COUNT / 2; msg++)
+ for (int msg = 0; msg < (MSG_COUNT / 2); msg++)
{
assertTrue(_consumer1.receive() != null);
}
- for (int msg = 0; msg < MSG_COUNT / 2; msg++)
+ for (int msg = 0; msg < (MSG_COUNT / 2); msg++)
{
assertTrue(_consumer2.receive() != null);
}
}
}
-
public static junit.framework.Test suite()
{
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?view=diff&rev=547730&r1=547729&r2=547730
==============================================================================
--- 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 Fri Jun 15 09:28:46 2007
@@ -14,15 +14,19 @@
* "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.
+ * under the License.
+ *
*
- *
*/
package org.apache.qpid.client;
-import java.util.Hashtable;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
+import junit.framework.TestCase;
+
+import org.apache.qpid.client.transport.TransportConnection;
+import org.apache.qpid.jndi.PropertiesFileInitialContextFactory;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
@@ -35,13 +39,9 @@
import javax.naming.Context;
import javax.naming.spi.InitialContextFactory;
-import junit.framework.TestCase;
-
-import org.apache.log4j.Logger;
-import org.apache.qpid.client.transport.TransportConnection;
-import org.apache.qpid.jndi.PropertiesFileInitialContextFactory;
-import org.apache.qpid.url.BindingURL;
-import org.apache.qpid.url.AMQBindingURL;
+import java.util.Hashtable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
/**
* QPID-293 Setting MessageListener after connection has started can cause messages to be "lost" on a internal delivery
@@ -54,7 +54,7 @@
*/
public class MessageListenerTest extends TestCase implements MessageListener
{
- private static final Logger _logger = Logger.getLogger(MessageListenerTest.class);
+ private static final Logger _logger = LoggerFactory.getLogger(MessageListenerTest.class);
Context _context;
@@ -73,6 +73,7 @@
{
TransportConnection.createVMBroker(1);
}
+
InitialContextFactory factory = new PropertiesFileInitialContextFactory();
Hashtable<String, String> env = new Hashtable<String, String>();
@@ -84,7 +85,7 @@
Queue queue = (Queue) _context.lookup("queue");
- //Create Client
+ // Create Client
_clientConnection = ((ConnectionFactory) _context.lookup("connection")).createConnection();
_clientConnection.start();
@@ -93,7 +94,7 @@
_consumer = clientSession.createConsumer(queue);
- //Create Producer
+ // Create Producer
Connection producerConnection = ((ConnectionFactory) _context.lookup("connection")).createConnection();
@@ -123,7 +124,6 @@
}
}
-
public void testSynchronousRecieve() throws Exception
{
for (int msg = 0; msg < MSG_COUNT; msg++)
@@ -144,9 +144,9 @@
}
catch (InterruptedException e)
{
- //do nothing
+ // do nothing
}
- //Should have recieved all async messages
+ // Should have recieved all async messages
assertEquals(MSG_COUNT, receivedCount);
}
@@ -156,11 +156,11 @@
_logger.error("Test disabled as initial receive is not called first");
// Perform initial receive to start connection
-// assertTrue(_consumer.receive(2000) != null);
-// receivedCount++;
+ // assertTrue(_consumer.receive(2000) != null);
+ // receivedCount++;
// Sleep to ensure remaining 4 msgs end up on _synchronousQueue
-// Thread.sleep(1000);
+ // Thread.sleep(1000);
// Set the message listener and wait for the messages to come in.
_consumer.setMessageListener(this);
@@ -173,13 +173,12 @@
}
catch (InterruptedException e)
{
- //do nothing
+ // do nothing
}
- //Should have recieved all async messages
+ // Should have recieved all async messages
assertEquals(MSG_COUNT, receivedCount);
}
-
public void onMessage(Message message)
{
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?view=diff&rev=547730&r1=547729&r2=547730
==============================================================================
--- 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 Fri Jun 15 09:28:46 2007
@@ -20,9 +20,13 @@
*/
package org.apache.qpid.client;
-import java.util.Hashtable;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
+import junit.framework.TestCase;
+
+import org.apache.qpid.client.transport.TransportConnection;
+import org.apache.qpid.jndi.PropertiesFileInitialContextFactory;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
@@ -36,11 +40,9 @@
import javax.naming.Context;
import javax.naming.spi.InitialContextFactory;
-import junit.framework.TestCase;
-
-import org.apache.log4j.Logger;
-import org.apache.qpid.client.transport.TransportConnection;
-import org.apache.qpid.jndi.PropertiesFileInitialContextFactory;
+import java.util.Hashtable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
/**
* QPID-293 Setting MessageListener after connection has started can cause messages to be "lost" on a internal delivery
@@ -53,7 +55,7 @@
*/
public class ResetMessageListenerTest extends TestCase
{
- private static final Logger _logger = Logger.getLogger(ResetMessageListenerTest.class);
+ private static final Logger _logger = LoggerFactory.getLogger(ResetMessageListenerTest.class);
Context _context;
@@ -67,8 +69,8 @@
MessageProducer _producer;
Session _clientSession, _producerSession;
- 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 _allFirstMessagesSent = new CountDownLatch(2); // all messages Sent Lock
+ private final CountDownLatch _allSecondMessagesSent = new CountDownLatch(2); // all messages Sent Lock
protected void setUp() throws Exception
{
@@ -88,17 +90,17 @@
Queue queue = (Queue) _context.lookup("queue");
- //Create Client 1
+ // Create Client 1
_clientConnection = ((ConnectionFactory) _context.lookup("connection")).createConnection();
_clientSession = _clientConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
_consumer1 = _clientSession.createConsumer(queue);
- //Create Client 2 on same session
+ // Create Client 2 on same session
_consumer2 = _clientSession.createConsumer(queue);
- //Create Producer
+ // Create Producer
_producerConnection = ((ConnectionFactory) _context.lookup("connection")).createConnection();
_producerConnection.start();
@@ -129,57 +131,55 @@
TransportConnection.killAllVMBrokers();
}
-
public void testAsynchronousRecieve()
{
_logger.info("Test Start");
- //Set default Message Listener
+ // Set default Message Listener
try
{
_consumer1.setMessageListener(new MessageListener()
- {
- public void onMessage(Message message)
{
- _logger.info("Client 1 ML 1 Received Message(" + receivedCount1ML1 + "):" + message);
-
- receivedCount1ML1++;
- if (receivedCount1ML1 == MSG_COUNT / 2)
+ public void onMessage(Message message)
{
- _allFirstMessagesSent.countDown();
+ _logger.info("Client 1 ML 1 Received Message(" + receivedCount1ML1 + "):" + message);
+
+ receivedCount1ML1++;
+ if (receivedCount1ML1 == (MSG_COUNT / 2))
+ {
+ _allFirstMessagesSent.countDown();
+ }
}
- }
- });
+ });
}
catch (JMSException e)
{
_logger.error("Error Setting Default ML on consumer1");
}
-
try
{
_consumer2.setMessageListener(new MessageListener()
- {
- public void onMessage(Message message)
{
- _logger.info("Client 2 Received Message(" + receivedCount2 + "):" + message);
-
- receivedCount2++;
- if (receivedCount2 == MSG_COUNT / 2)
+ public void onMessage(Message message)
{
- _logger.info("Client 2 received all its messages1");
- _allFirstMessagesSent.countDown();
- }
+ _logger.info("Client 2 Received Message(" + receivedCount2 + "):" + message);
- if (receivedCount2 == MSG_COUNT)
- {
- _logger.info("Client 2 received all its messages2");
- _allSecondMessagesSent.countDown();
+ receivedCount2++;
+ if (receivedCount2 == (MSG_COUNT / 2))
+ {
+ _logger.info("Client 2 received all its messages1");
+ _allFirstMessagesSent.countDown();
+ }
+
+ if (receivedCount2 == MSG_COUNT)
+ {
+ _logger.info("Client 2 received all its messages2");
+ _allSecondMessagesSent.countDown();
+ }
}
- }
- });
+ });
_clientConnection.start();
}
@@ -189,7 +189,6 @@
}
-
try
{
_allFirstMessagesSent.await(1000, TimeUnit.MILLISECONDS);
@@ -197,7 +196,7 @@
}
catch (InterruptedException e)
{
- //do nothing
+ // do nothing
}
try
@@ -213,18 +212,18 @@
try
{
_consumer1.setMessageListener(new MessageListener()
- {
- public void onMessage(Message message)
{
- _logger.info("Client 1 ML2 Received Message(" + receivedCount1ML1 + "):" + message);
-
- receivedCount1ML2++;
- if (receivedCount1ML2 == MSG_COUNT / 2)
+ public void onMessage(Message message)
{
- _allSecondMessagesSent.countDown();
+ _logger.info("Client 1 ML2 Received Message(" + receivedCount1ML1 + "):" + message);
+
+ receivedCount1ML2++;
+ if (receivedCount1ML2 == (MSG_COUNT / 2))
+ {
+ _allSecondMessagesSent.countDown();
+ }
}
- }
- });
+ });
_clientConnection.start();
}
@@ -260,10 +259,9 @@
}
catch (InterruptedException e)
{
- //do nothing
+ // do nothing
}
}
-
public static junit.framework.Test suite()
{
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=547730&r1=547729&r2=547730
==============================================================================
--- 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 Fri Jun 15 09:28:46 2007
@@ -19,32 +19,30 @@
*/
package org.apache.qpid.test.unit.ack;
-import java.util.concurrent.atomic.AtomicInteger;
+import junit.framework.TestCase;
+
+import org.apache.qpid.client.AMQConnection;
+import org.apache.qpid.client.AMQQueue;
+import org.apache.qpid.client.transport.TransportConnection;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.jms.Session;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
-import javax.jms.Connection;
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.TextMessage;
-import junit.framework.TestCase;
-
-import org.apache.log4j.Logger;
-import org.apache.qpid.client.AMQConnection;
-import org.apache.qpid.client.AMQQueue;
-import org.apache.qpid.client.AMQSession;
-import org.apache.qpid.client.transport.TransportConnection;
-import org.apache.qpid.exchange.ExchangeDefaults;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.jms.Session;
+import java.util.concurrent.atomic.AtomicInteger;
public class RecoverTest extends TestCase
{
- private static final Logger _logger = Logger.getLogger(RecoverTest.class);
+ private static final Logger _logger = LoggerFactory.getLogger(RecoverTest.class);
private Exception _error;
private AtomicInteger count;
@@ -64,16 +62,17 @@
count = null;
}
-
public void testRecoverResendsMsgs() throws Exception
{
AMQConnection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "test");
Session consumerSession = con.createSession(false, Session.CLIENT_ACKNOWLEDGE);
- Queue queue = new AMQQueue(consumerSession.getDefaultQueueExchangeName(),new AMQShortString("someQ"), new AMQShortString("someQ"), false, true);
+ Queue queue =
+ new AMQQueue(consumerSession.getDefaultQueueExchangeName(), 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(ExchangeDefaults.DIRECT_EXCHANGE_NAME, ExchangeDefaults.DIRECT_EXCHANGE_CLASS);
+ // force synch to ensure the consumer has resulted in a bound queue
+ // ((AMQSession) consumerSession).declareExchangeSynch(ExchangeDefaults.DIRECT_EXCHANGE_NAME, ExchangeDefaults.DIRECT_EXCHANGE_CLASS);
// This is the default now
AMQConnection con2 = new AMQConnection("vm://:1", "guest", "guest", "producer1", "test");
@@ -122,16 +121,17 @@
con.close();
}
-
public void testRecoverResendsMsgsAckOnEarlier() throws Exception
{
AMQConnection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "test");
Session consumerSession = con.createSession(false, Session.CLIENT_ACKNOWLEDGE);
- Queue queue = new AMQQueue(consumerSession.getDefaultQueueExchangeName(), new AMQShortString("someQ"), new AMQShortString("someQ"), false, true);
+ Queue queue =
+ new AMQQueue(consumerSession.getDefaultQueueExchangeName(), 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(ExchangeDefaults.DIRECT_EXCHANGE_NAME, ExchangeDefaults.DIRECT_EXCHANGE_CLASS);
+ // force synch to ensure the consumer has resulted in a bound queue
+ // ((AMQSession) consumerSession).declareExchangeSynch(ExchangeDefaults.DIRECT_EXCHANGE_NAME, ExchangeDefaults.DIRECT_EXCHANGE_CLASS);
// This is the default now
AMQConnection con2 = new AMQConnection("vm://:1", "guest", "guest", "producer1", "test");
@@ -164,7 +164,6 @@
TextMessage tm4 = (TextMessage) consumer.receive(3000);
assertEquals("msg4", tm4.getText());
-
_logger.info("Received redelivery of two messages. calling acknolwedgeThis() first of those message");
((org.apache.qpid.jms.Message) tm3).acknowledgeThis();
@@ -180,7 +179,6 @@
// all acked so no messages to be delivered
consumerSession.recover();
-
tm = (TextMessage) consumer.receiveNoWait();
assertNull(tm);
_logger.info("No messages redelivered as is expected");
@@ -193,8 +191,12 @@
AMQConnection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "test");
Session consumerSession = con.createSession(false, Session.CLIENT_ACKNOWLEDGE);
- Queue queue = new AMQQueue(consumerSession.getDefaultQueueExchangeName(), new AMQShortString("Q1"), new AMQShortString("Q1"), false, true);
- Queue queue2 = new AMQQueue(consumerSession.getDefaultQueueExchangeName(), new AMQShortString("Q2"), new AMQShortString("Q2"), false, true);
+ Queue queue =
+ new AMQQueue(consumerSession.getDefaultQueueExchangeName(), new AMQShortString("Q1"), new AMQShortString("Q1"),
+ false, true);
+ Queue queue2 =
+ new AMQQueue(consumerSession.getDefaultQueueExchangeName(), new AMQShortString("Q2"), new AMQShortString("Q2"),
+ false, true);
MessageConsumer consumer = consumerSession.createConsumer(queue);
MessageConsumer consumer2 = consumerSession.createConsumer(queue2);
@@ -232,68 +234,72 @@
AMQConnection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "test");
final Session consumerSession = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
- Queue queue = new AMQQueue(consumerSession.getDefaultQueueExchangeName(), new AMQShortString("Q3"), new AMQShortString("Q3"), false, true);
+ Queue queue =
+ new AMQQueue(consumerSession.getDefaultQueueExchangeName(), new AMQShortString("Q3"), new AMQShortString("Q3"),
+ false, true);
MessageConsumer consumer = consumerSession.createConsumer(queue);
MessageProducer producer = consumerSession.createProducer(queue);
producer.send(consumerSession.createTextMessage("hello"));
-
final Object lock = new Object();
consumer.setMessageListener(new MessageListener()
- {
-
-
-
- public void onMessage(Message message)
{
- try
+
+ public void onMessage(Message message)
{
- count.incrementAndGet();
- if (count.get() == 1)
+ try
{
- if(message.getJMSRedelivered())
+ count.incrementAndGet();
+ if (count.get() == 1)
{
- setError(new Exception("Message marked as redilvered on what should be first delivery attempt"));
+ if (message.getJMSRedelivered())
+ {
+ setError(
+ new Exception("Message marked as redilvered on what should be first delivery attempt"));
+ }
+
+ consumerSession.recover();
}
- consumerSession.recover();
- }
- else if (count.get() == 2)
- {
- if(!message.getJMSRedelivered())
+ else if (count.get() == 2)
{
- setError(new Exception("Message not marked as redilvered on what should be second delivery attempt"));
+ if (!message.getJMSRedelivered())
+ {
+ setError(
+ new Exception(
+ "Message not marked as redilvered on what should be second delivery attempt"));
+ }
+ }
+ else
+ {
+ System.err.println(message);
+ fail("Message delivered too many times!: " + count);
}
}
- else
+ catch (JMSException e)
{
- System.err.println(message);
- fail("Message delivered too many times!: " + count);
+ _logger.error("Error recovering session: " + e, e);
+ setError(e);
+ }
+
+ synchronized (lock)
+ {
+ lock.notify();
}
}
- catch (JMSException e)
- {
- _logger.error("Error recovering session: " + e, e);
- setError(e);
- }
- synchronized(lock)
- {
- lock.notify();
- }
- }
- });
+ });
con.start();
long waitTime = 300000L;
long waitUntilTime = System.currentTimeMillis() + waitTime;
- synchronized(lock)
+ synchronized (lock)
{
- while((count.get() <= 1) && (waitTime > 0))
+ while ((count.get() <= 1) && (waitTime > 0))
{
lock.wait(waitTime);
- if(count.get() <= 1)
+ if (count.get() <= 1)
{
waitTime = waitUntilTime - System.currentTimeMillis();
}
@@ -302,15 +308,16 @@
Thread.sleep(1000);
- if(count.get() != 2)
+ if (count.get() != 2)
{
System.err.println("Count != 2 : " + count);
}
- assertTrue(count.get() == 2);
+
+ assertTrue(count.get() == 2);
con.close();
- if(_error != null)
+ if (_error != null)
{
throw _error;
}