[hornetq-commits] JBoss hornetq SVN: r10793 - in branches/Branch_2_2_EAP-cluster-cleanup: src/main/org/hornetq/core/client/impl and 15 other directories.
do-not-reply at jboss.org
do-not-reply at jboss.org
Sun Jun 12 22:59:48 EDT 2011
Author: clebert.suconic at jboss.com
Date: 2011-06-12 22:59:47 -0400 (Sun, 12 Jun 2011)
New Revision: 10793
Modified:
branches/Branch_2_2_EAP-cluster-cleanup/src/config/common/schema/hornetq-configuration.xsd
branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java
branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/cluster/impl/DiscoveryGroupImpl.java
branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/config/BridgeConfiguration.java
branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/config/ClusterConnectionConfiguration.java
branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java
branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java
branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java
branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/postoffice/impl/BindingsImpl.java
branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/postoffice/impl/DivertBinding.java
branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java
branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/protocol/core/impl/ChannelImpl.java
branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/cluster/Bridge.java
branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/cluster/MessageFlowRecord.java
branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java
branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionBridge.java
branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java
branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java
branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/impl/QueueImpl.java
branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java
branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/management/impl/ManagementServiceImpl.java
branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeReconnectTest.java
branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeStartTest.java
branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeTest.java
branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeWithDiscoveryGroupStartTest.java
branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeWithPagingTest.java
branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/cluster/distribution/OnewayTwoNodeClusterTest.java
branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/management/BridgeControlTest.java
branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/management/BridgeControlUsingCoreTest.java
branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java
Log:
Cleanup branch
Modified: branches/Branch_2_2_EAP-cluster-cleanup/src/config/common/schema/hornetq-configuration.xsd
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/src/config/common/schema/hornetq-configuration.xsd 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/src/config/common/schema/hornetq-configuration.xsd 2011-06-13 02:59:47 UTC (rev 10793)
@@ -313,12 +313,18 @@
</xsd:element>
<xsd:element maxOccurs="1" minOccurs="0" name="transformer-class-name" type="xsd:string">
</xsd:element>
+ <xsd:element maxOccurs="1" minOccurs="0" name="check-period" type="xsd:long">
+ </xsd:element>
+ <xsd:element maxOccurs="1" minOccurs="0" name="connection-ttl" type="xsd:long">
+ </xsd:element>
<xsd:element maxOccurs="1" minOccurs="0" name="retry-interval" type="xsd:long">
- </xsd:element>
+ </xsd:element>
<xsd:element maxOccurs="1" minOccurs="0" name="retry-interval-multiplier" type="xsd:double">
- </xsd:element>
- <xsd:element maxOccurs="1" minOccurs="0" name="reconnect-attempts" type="xsd:int">
- </xsd:element>
+ </xsd:element>
+ <xsd:element maxOccurs="1" minOccurs="0" name="max-retry-interval" type="xsd:long">
+ </xsd:element>
+ <xsd:element maxOccurs="1" minOccurs="0" name="reconnect-attempts" type="xsd:long">
+ </xsd:element>
<xsd:element maxOccurs="1" minOccurs="0" name="failover-on-server-shutdown" type="xsd:boolean">
</xsd:element>
<xsd:element maxOccurs="1" minOccurs="0" name="use-duplicate-detection" type="xsd:boolean">
@@ -354,8 +360,18 @@
</xsd:element>
<xsd:element maxOccurs="1" minOccurs="1" name="connector-ref" type="xsd:string">
</xsd:element>
+ <xsd:element maxOccurs="1" minOccurs="0" name="check-period" type="xsd:long">
+ </xsd:element>
+ <xsd:element maxOccurs="1" minOccurs="0" name="connection-ttl" type="xsd:long">
+ </xsd:element>
<xsd:element maxOccurs="1" minOccurs="0" name="retry-interval" type="xsd:long">
</xsd:element>
+ <xsd:element maxOccurs="1" minOccurs="0" name="retry-interval-multiplier" type="xsd:double">
+ </xsd:element>
+ <xsd:element maxOccurs="1" minOccurs="0" name="max-retry-interval" type="xsd:long">
+ </xsd:element>
+ <xsd:element maxOccurs="1" minOccurs="0" name="reconnect-attempts" type="xsd:long">
+ </xsd:element>
<xsd:element maxOccurs="1" minOccurs="0" name="use-duplicate-detection" type="xsd:boolean">
</xsd:element>
<xsd:element maxOccurs="1" minOccurs="0" name="forward-when-no-consumers" type="xsd:boolean">
Modified: branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -83,6 +83,8 @@
private static final long serialVersionUID = 2512460695662741413L;
private static final Logger log = Logger.getLogger(ClientSessionFactoryImpl.class);
+
+ private static final boolean isTrace = log.isTraceEnabled();
// Attributes
// -----------------------------------------------------------------------------------
@@ -497,6 +499,12 @@
return;
}
+
+ if (isTrace)
+ {
+ log.trace("Client Connection failed, calling failure listeners and trying to reconnect, reconnectAttempts=" + reconnectAttempts);
+ }
+
// We call before reconnection occurs to give the user a chance to do cleanup, like cancel messages
callFailureListeners(me, false, false);
@@ -910,6 +918,13 @@
return;
}
+ if (isTrace)
+ {
+ log.trace("Waiting " + interval +
+ " milliseconds before next retry. RetryInterval=" + retryInterval +
+ " and multiplier = " + retryIntervalMultiplier);
+ }
+
try
{
waitLock.wait(interval);
@@ -1086,6 +1101,13 @@
}
}
}
+ else
+ {
+ if (isTrace)
+ {
+ log.trace("No Backup configured!");
+ }
+ }
}
catch (Exception e)
{
Modified: branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/cluster/impl/DiscoveryGroupImpl.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/cluster/impl/DiscoveryGroupImpl.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/cluster/impl/DiscoveryGroupImpl.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -49,6 +49,8 @@
public class DiscoveryGroupImpl implements Runnable, DiscoveryGroup
{
private static final Logger log = Logger.getLogger(DiscoveryGroupImpl.class);
+
+ private static final boolean isTrace = log.isTraceEnabled();
private static final int SOCKET_TIMEOUT = 500;
@@ -375,6 +377,14 @@
if (changed)
{
+ if (isTrace)
+ {
+ log.trace("Connectors changed on Discovery:");
+ for (DiscoveryEntry connector : connectors.values())
+ {
+ log.trace(connector);
+ }
+ }
callListeners();
}
@@ -438,6 +448,10 @@
if (entry.getValue().getLastUpdate() + timeout <= now)
{
+ if (isTrace)
+ {
+ log.trace("Timed out node on discovery:" + entry.getValue());
+ }
iter.remove();
changed = true;
Modified: branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/config/BridgeConfiguration.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/config/BridgeConfiguration.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/config/BridgeConfiguration.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -61,17 +61,24 @@
private String password;
+ private final long connectionTTL;
+
+ private final long maxRetryInterval;
+
+
public BridgeConfiguration(final String name,
final String queueName,
final String forwardingAddress,
final String filterString,
final String transformerClassName,
+ final long clientFailureCheckPeriod,
+ final long connectionTTL,
final long retryInterval,
+ final long maxRetryInterval,
final double retryIntervalMultiplier,
final int reconnectAttempts,
final boolean useDuplicateDetection,
final int confirmationWindowSize,
- final long clientFailureCheckPeriod,
final List<String> staticConnectors,
final boolean ha,
final String user,
@@ -89,8 +96,10 @@
this.confirmationWindowSize = confirmationWindowSize;
this.clientFailureCheckPeriod = clientFailureCheckPeriod;
this.staticConnectors = staticConnectors;
- this.user = user;
+ this. user = user;
this.password = password;
+ this.connectionTTL = connectionTTL;
+ this.maxRetryInterval = maxRetryInterval;
discoveryGroupName = null;
}
@@ -99,12 +108,14 @@
final String forwardingAddress,
final String filterString,
final String transformerClassName,
+ final long clientFailureCheckPeriod,
+ final long connectionTTL,
final long retryInterval,
+ final long maxRetryInterval,
final double retryIntervalMultiplier,
final int reconnectAttempts,
final boolean useDuplicateDetection,
final int confirmationWindowSize,
- final long clientFailureCheckPeriod,
final String discoveryGroupName,
final boolean ha,
final String user,
@@ -126,6 +137,8 @@
this.ha = ha;
this.user = user;
this.password = password;
+ this.connectionTTL = connectionTTL;
+ this.maxRetryInterval = maxRetryInterval;
}
public String getName()
@@ -138,6 +151,22 @@
return queueName;
}
+ /**
+ * @return the connectionTTL
+ */
+ public long getConnectionTTL()
+ {
+ return connectionTTL;
+ }
+
+ /**
+ * @return the maxRetryInterval
+ */
+ public long getMaxRetryInterval()
+ {
+ return maxRetryInterval;
+ }
+
public String getForwardingAddress()
{
return forwardingAddress;
Modified: branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/config/ClusterConnectionConfiguration.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/config/ClusterConnectionConfiguration.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/config/ClusterConnectionConfiguration.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -16,6 +16,8 @@
import java.io.Serializable;
import java.util.List;
+import org.hornetq.core.config.impl.ConfigurationImpl;
+
/**
* A ClusterConnectionConfiguration
*
@@ -35,7 +37,17 @@
private final String connectorName;
+ private final long clientFailureCheckPeriod;
+
+ private final long connectionTTL;
+
private final long retryInterval;
+
+ private final double retryIntervalMultiplier;
+
+ private final long maxRetryInterval;
+
+ private final int reconnectAttempts;
private final boolean duplicateDetection;
@@ -50,11 +62,45 @@
private final int confirmationWindowSize;
private final boolean allowDirectConnectionsOnly;
+
+ public ClusterConnectionConfiguration(final String name,
+ final String address,
+ final String connectorName,
+ final long retryInterval,
+ final boolean duplicateDetection,
+ final boolean forwardWhenNoConsumers,
+ final int maxHops,
+ final int confirmationWindowSize,
+ final List<String> staticConnectors,
+ final boolean allowDirectConnectionsOnly)
+ {
+ this(name,
+ address,
+ connectorName,
+ ConfigurationImpl.DEFAULT_CLUSTER_FAILURE_CHECK_PERIOD,
+ ConfigurationImpl.DEFAULT_CLUSTER_CONNECTION_TTL,
+ retryInterval,
+ ConfigurationImpl.DEFAULT_CLUSTER_RETRY_INTERVAL_MULTIPLIER,
+ ConfigurationImpl.DEFAULT_CLUSTER_MAX_RETRY_INTERVAL,
+ ConfigurationImpl.DEFAULT_CLUSTER_RECONNECT_ATTEMPTS,
+ duplicateDetection,
+ forwardWhenNoConsumers,
+ maxHops,
+ confirmationWindowSize,
+ staticConnectors,
+ allowDirectConnectionsOnly);
+ }
+
public ClusterConnectionConfiguration(final String name,
final String address,
final String connectorName,
+ final long clientFailureCheckPeriod,
+ final long connectionTTL,
final long retryInterval,
+ final double retryIntervalMultiplier,
+ final long maxRetryInterval,
+ final int reconnectAttempts,
final boolean duplicateDetection,
final boolean forwardWhenNoConsumers,
final int maxHops,
@@ -65,7 +111,12 @@
this.name = name;
this.address = address;
this.connectorName = connectorName;
+ this.clientFailureCheckPeriod = clientFailureCheckPeriod;
+ this.connectionTTL = connectionTTL;
this.retryInterval = retryInterval;
+ this.retryIntervalMultiplier = retryIntervalMultiplier;
+ this.maxRetryInterval = maxRetryInterval;
+ this.reconnectAttempts = reconnectAttempts;
this.staticConnectors = staticConnectors;
this.duplicateDetection = duplicateDetection;
this.forwardWhenNoConsumers = forwardWhenNoConsumers;
@@ -75,6 +126,7 @@
this.allowDirectConnectionsOnly = allowDirectConnectionsOnly;
}
+
public ClusterConnectionConfiguration(final String name,
final String address,
final String connectorName,
@@ -85,10 +137,47 @@
final int confirmationWindowSize,
final String discoveryGroupName)
{
+ this(name,
+ address,
+ connectorName,
+ ConfigurationImpl.DEFAULT_CLUSTER_FAILURE_CHECK_PERIOD,
+ ConfigurationImpl.DEFAULT_CLUSTER_CONNECTION_TTL,
+ retryInterval,
+ ConfigurationImpl.DEFAULT_CLUSTER_RETRY_INTERVAL_MULTIPLIER,
+ ConfigurationImpl.DEFAULT_CLUSTER_MAX_RETRY_INTERVAL,
+ ConfigurationImpl.DEFAULT_CLUSTER_RECONNECT_ATTEMPTS,
+ duplicateDetection,
+ forwardWhenNoConsumers,
+ maxHops,
+ confirmationWindowSize,
+ discoveryGroupName);
+ }
+
+
+ public ClusterConnectionConfiguration(final String name,
+ final String address,
+ final String connectorName,
+ final long clientFailureCheckPeriod,
+ final long connectionTTL,
+ final long retryInterval,
+ final double retryIntervalMultiplier,
+ final long maxRetryInterval,
+ final int reconnectAttempts,
+ final boolean duplicateDetection,
+ final boolean forwardWhenNoConsumers,
+ final int maxHops,
+ final int confirmationWindowSize,
+ final String discoveryGroupName)
+ {
this.name = name;
this.address = address;
this.connectorName = connectorName;
+ this.clientFailureCheckPeriod = clientFailureCheckPeriod;
+ this.connectionTTL = connectionTTL;
this.retryInterval = retryInterval;
+ this.retryIntervalMultiplier = retryIntervalMultiplier;
+ this.maxRetryInterval = maxRetryInterval;
+ this.reconnectAttempts = reconnectAttempts;
this.duplicateDetection = duplicateDetection;
this.forwardWhenNoConsumers = forwardWhenNoConsumers;
this.discoveryGroupName = discoveryGroupName;
@@ -108,6 +197,46 @@
return address;
}
+ /**
+ * @return the clientFailureCheckPeriod
+ */
+ public long getClientFailureCheckPeriod()
+ {
+ return clientFailureCheckPeriod;
+ }
+
+ /**
+ * @return the connectionTTL
+ */
+ public long getConnectionTTL()
+ {
+ return connectionTTL;
+ }
+
+ /**
+ * @return the retryIntervalMultiplier
+ */
+ public double getRetryIntervalMultiplier()
+ {
+ return retryIntervalMultiplier;
+ }
+
+ /**
+ * @return the maxRetryInterval
+ */
+ public long getMaxRetryInterval()
+ {
+ return maxRetryInterval;
+ }
+
+ /**
+ * @return the reconnectAttempts
+ */
+ public int getReconnectAttempts()
+ {
+ return reconnectAttempts;
+ }
+
public String getConnectorName()
{
return connectorName;
Modified: branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -24,6 +24,7 @@
import org.hornetq.api.core.DiscoveryGroupConfiguration;
import org.hornetq.api.core.SimpleString;
import org.hornetq.api.core.TransportConfiguration;
+import org.hornetq.api.core.client.HornetQClient;
import org.hornetq.core.config.BridgeConfiguration;
import org.hornetq.core.config.BroadcastGroupConfiguration;
import org.hornetq.core.config.ClusterConnectionConfiguration;
@@ -166,6 +167,16 @@
public static final long DEFAULT_CLUSTER_RETRY_INTERVAL = 500;
+ public static final int DEFAULT_CLUSTER_RECONNECT_ATTEMPTS = -1;
+
+ public static final long DEFAULT_CLUSTER_FAILURE_CHECK_PERIOD = HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD;
+
+ public static final long DEFAULT_CLUSTER_CONNECTION_TTL = HornetQClient.DEFAULT_CONNECTION_TTL;
+
+ public static final double DEFAULT_CLUSTER_RETRY_INTERVAL_MULTIPLIER = HornetQClient.DEFAULT_RETRY_INTERVAL_MULTIPLIER;
+
+ public static final long DEFAULT_CLUSTER_MAX_RETRY_INTERVAL = HornetQClient.DEFAULT_MAX_RETRY_INTERVAL;
+
public static final boolean DEFAULT_DIVERT_EXCLUSIVE = false;
public static final boolean DEFAULT_BRIDGE_DUPLICATE_DETECTION = true;
Modified: branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -1003,12 +1003,27 @@
"max-hops",
ConfigurationImpl.DEFAULT_CLUSTER_MAX_HOPS,
Validators.GE_ZERO);
+
+ long clientFailureCheckPeriod = XMLConfigurationUtil.getLong(e, "check-period",
+ ConfigurationImpl.DEFAULT_CLUSTER_FAILURE_CHECK_PERIOD, Validators.GT_ZERO) ;
+ long connectionTTL = XMLConfigurationUtil.getLong(e, "connection-ttl",
+ ConfigurationImpl.DEFAULT_CLUSTER_CONNECTION_TTL, Validators.GT_ZERO) ;
+
+
long retryInterval = XMLConfigurationUtil.getLong(e,
"retry-interval",
ConfigurationImpl.DEFAULT_CLUSTER_RETRY_INTERVAL,
Validators.GT_ZERO);
+
+ double retryIntervalMultiplier = XMLConfigurationUtil.getDouble(e, "retry-interval-multiplier",
+ ConfigurationImpl.DEFAULT_CLUSTER_RETRY_INTERVAL_MULTIPLIER, Validators.GT_ZERO);
+
+ long maxRetryInterval = XMLConfigurationUtil.getLong(e, "max-retry-interval", ConfigurationImpl.DEFAULT_CLUSTER_MAX_RETRY_INTERVAL, Validators.GT_ZERO);
+
+ int reconnectAttempts = XMLConfigurationUtil.getInteger(e, "reconnect-attempts", ConfigurationImpl.DEFAULT_CLUSTER_RECONNECT_ATTEMPTS, Validators.GT_ZERO);
+
int confirmationWindowSize = XMLConfigurationUtil.getInteger(e,
"confirmation-window-size",
FileConfiguration.DEFAULT_CONFIRMATION_WINDOW_SIZE,
@@ -1048,7 +1063,12 @@
config = new ClusterConnectionConfiguration(name,
address,
connectorName,
+ clientFailureCheckPeriod,
+ connectionTTL,
retryInterval,
+ retryIntervalMultiplier,
+ maxRetryInterval,
+ reconnectAttempts,
duplicateDetection,
forwardWhenNoConsumers,
maxHops,
@@ -1061,7 +1081,12 @@
config = new ClusterConnectionConfiguration(name,
address,
connectorName,
+ clientFailureCheckPeriod,
+ connectionTTL,
retryInterval,
+ retryIntervalMultiplier,
+ maxRetryInterval,
+ reconnectAttempts,
duplicateDetection,
forwardWhenNoConsumers,
maxHops,
@@ -1101,17 +1126,26 @@
null,
Validators.NO_CHECK);
+ // Default bridge conf
+ int confirmationWindowSize = XMLConfigurationUtil.getInteger(brNode,
+ "confirmation-window-size",
+ FileConfiguration.DEFAULT_CONFIRMATION_WINDOW_SIZE,
+ Validators.GT_ZERO);
+
long retryInterval = XMLConfigurationUtil.getLong(brNode,
"retry-interval",
HornetQClient.DEFAULT_RETRY_INTERVAL,
Validators.GT_ZERO);
- // Default bridge conf
- int confirmationWindowSize = XMLConfigurationUtil.getInteger(brNode,
- "confirmation-window-size",
- FileConfiguration.DEFAULT_CONFIRMATION_WINDOW_SIZE,
- Validators.GT_ZERO);
+ long clientFailureCheckPeriod = XMLConfigurationUtil.getLong(brNode, "check-period",
+ HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, Validators.GT_ZERO) ;
+ long connectionTTL = XMLConfigurationUtil.getLong(brNode, "connection-ttl",
+ HornetQClient.DEFAULT_CONNECTION_TTL, Validators.GT_ZERO) ;
+
+ long maxRetryInterval = XMLConfigurationUtil.getLong(brNode, "max-retry-interval", HornetQClient.DEFAULT_MAX_RETRY_INTERVAL, Validators.GT_ZERO);
+
+
double retryIntervalMultiplier = XMLConfigurationUtil.getDouble(brNode,
"retry-interval-multiplier",
HornetQClient.DEFAULT_RETRY_INTERVAL_MULTIPLIER,
@@ -1173,12 +1207,14 @@
forwardingAddress,
filterString,
transformerClassName,
+ clientFailureCheckPeriod,
+ connectionTTL,
retryInterval,
+ maxRetryInterval,
retryIntervalMultiplier,
reconnectAttempts,
useDuplicateDetection,
confirmationWindowSize,
- HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
staticConnectorNames,
ha,
user,
@@ -1191,12 +1227,14 @@
forwardingAddress,
filterString,
transformerClassName,
+ clientFailureCheckPeriod,
+ connectionTTL,
retryInterval,
+ maxRetryInterval,
retryIntervalMultiplier,
reconnectAttempts,
useDuplicateDetection,
confirmationWindowSize,
- HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
discoveryGroupName,
ha,
user,
Modified: branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -38,6 +38,7 @@
import org.hornetq.api.core.HornetQException;
import org.hornetq.api.core.SimpleString;
import org.hornetq.api.core.TransportConfiguration;
+import org.hornetq.api.core.client.HornetQClient;
import org.hornetq.api.core.management.AddressControl;
import org.hornetq.api.core.management.BridgeControl;
import org.hornetq.api.core.management.DivertControl;
@@ -1702,12 +1703,14 @@
forwardingAddress,
filterString,
transformerClassName,
+ clientFailureCheckPeriod,
+ HornetQClient.DEFAULT_CONNECTION_TTL,
retryInterval,
+ HornetQClient.DEFAULT_MAX_RETRY_INTERVAL,
retryIntervalMultiplier,
reconnectAttempts,
useDuplicateDetection,
confirmationWindowSize,
- clientFailureCheckPeriod,
connectorNames,
ha,
user,
@@ -1721,12 +1724,14 @@
forwardingAddress,
filterString,
transformerClassName,
+ clientFailureCheckPeriod,
+ HornetQClient.DEFAULT_CONNECTION_TTL,
retryInterval,
+ HornetQClient.DEFAULT_MAX_RETRY_INTERVAL,
retryIntervalMultiplier,
reconnectAttempts,
useDuplicateDetection,
confirmationWindowSize,
- clientFailureCheckPeriod,
connectors,
ha,
user,
Modified: branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/postoffice/impl/BindingsImpl.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/postoffice/impl/BindingsImpl.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/postoffice/impl/BindingsImpl.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -13,6 +13,8 @@
package org.hornetq.core.postoffice.impl;
+import java.io.PrintWriter;
+import java.io.StringWriter;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.List;
@@ -50,6 +52,8 @@
{
private static final Logger log = Logger.getLogger(BindingsImpl.class);
+ private static boolean isTrace = log.isTraceEnabled();
+
private final ConcurrentMap<SimpleString, List<Binding>> routingNameBindingMap = new ConcurrentHashMap<SimpleString, List<Binding>>();
private final Map<SimpleString, Integer> routingNamePositions = new ConcurrentHashMap<SimpleString, Integer>();
@@ -61,13 +65,16 @@
private volatile boolean routeWhenNoConsumers;
private final GroupingHandler groupingHandler;
-
+
private final PagingStore pageStore;
- public BindingsImpl(final GroupingHandler groupingHandler, final PagingStore pageStore)
+ private final SimpleString name;
+
+ public BindingsImpl(final SimpleString name, final GroupingHandler groupingHandler, final PagingStore pageStore)
{
this.groupingHandler = groupingHandler;
this.pageStore = pageStore;
+ this.name = name;
}
public void setRouteWhenNoConsumers(final boolean routeWhenNoConsumers)
@@ -82,6 +89,10 @@
public void addBinding(final Binding binding)
{
+ if (isTrace)
+ {
+ log.trace("addBinding(" + binding + ") being called");
+ }
if (binding.isExclusive())
{
exclusiveBindings.add(binding);
@@ -108,6 +119,12 @@
}
bindingsMap.put(binding.getID(), binding);
+
+ if (isTrace)
+ {
+ log.trace("Adding binding " + binding + " into " + this + " bindingTable: " + debugBindings());
+ }
+
}
public void removeBinding(final Binding binding)
@@ -134,6 +151,11 @@
}
bindingsMap.remove(binding.getID());
+
+ if (isTrace)
+ {
+ log.trace("Removing binding " + binding + " into " + this + " bindingTable: " + debugBindings());
+ }
}
public boolean redistribute(final ServerMessage message, final Queue originatingQueue, final RoutingContext context) throws Exception
@@ -144,6 +166,11 @@
return false;
}
+ if (isTrace)
+ {
+ log.trace("Redistributing message " + message);
+ }
+
SimpleString routingName = originatingQueue.getName();
List<Binding> bindings = routingNameBindingMap.get(routingName);
@@ -222,12 +249,12 @@
return false;
}
}
-
+
public PagingStore getPagingStore()
{
return pageStore;
}
-
+
public void route(final ServerMessage message, final RoutingContext context) throws Exception
{
boolean routed = false;
@@ -247,8 +274,8 @@
if (!routed)
{
- //TODO this is a little inefficient since we do the lookup once to see if the property
- //is there, then do it again to remove the actual property
+ // TODO this is a little inefficient since we do the lookup once to see if the property
+ // is there, then do it again to remove the actual property
if (message.containsProperty(MessageImpl.HDR_ROUTE_TO_IDS))
{
routeFromCluster(message, context);
@@ -259,6 +286,10 @@
}
else
{
+ if (isTrace)
+ {
+ log.trace("Routing message " + message + " on binding=" + this);
+ }
for (Map.Entry<SimpleString, List<Binding>> entry : routingNameBindingMap.entrySet())
{
SimpleString routingName = entry.getKey();
@@ -283,6 +314,15 @@
}
}
+ /* (non-Javadoc)
+ * @see java.lang.Object#toString()
+ */
+ @Override
+ public String toString()
+ {
+ return "BindingsImpl [name=" + name + "]";
+ }
+
private Binding getNextBinding(final ServerMessage message,
final SimpleString routingName,
final List<Binding> bindings)
@@ -290,9 +330,9 @@
Integer ipos = routingNamePositions.get(routingName);
int pos = ipos != null ? ipos : 0;
-
+
int length = bindings.size();
-
+
int startPos = pos;
Binding theBinding = null;
@@ -470,7 +510,75 @@
}
}
}
+
+ private String debugBindings()
+ {
+ StringWriter writer = new StringWriter();
+ PrintWriter out = new PrintWriter(writer);
+
+ out.println("\n***************************************");
+
+ out.println("routingNameBindingMap:");
+ if (routingNameBindingMap.isEmpty())
+ {
+ out.println("EMPTY!");
+ }
+ for (Map.Entry<SimpleString, List<Binding>> entry : routingNameBindingMap.entrySet())
+ {
+ out.print("key=" + entry.getKey() + ", value=" + entry.getValue());
+// for (Binding bind : entry.getValue())
+// {
+// out.print(bind + ",");
+// }
+ out.println();
+ }
+
+ out.println();
+
+ out.println("RoutingNamePositions:");
+ if (routingNamePositions.isEmpty())
+ {
+ out.println("EMPTY!");
+ }
+ for (Map.Entry<SimpleString, Integer> entry : routingNamePositions.entrySet())
+ {
+ out.println("key=" + entry.getKey() + ", value=" + entry.getValue());
+ }
+
+ out.println();
+
+ out.println("BindingsMap:");
+
+ if (bindingsMap.isEmpty())
+ {
+ out.println("EMPTY!");
+ }
+ for (Map.Entry<Long, Binding> entry : bindingsMap.entrySet())
+ {
+ out.println("Key=" + entry.getKey() + ", value=" + entry.getValue());
+ }
+
+ out.println();
+
+ out.println("ExclusiveBindings:");
+ if (exclusiveBindings.isEmpty())
+ {
+ out.println("EMPTY!");
+ }
+
+ for (Binding binding: exclusiveBindings)
+ {
+ out.println(binding);
+ }
+
+ out.println("#####################################################");
+
+
+ return writer.toString();
+ }
+
+
private void routeFromCluster(final ServerMessage message, final RoutingContext context) throws Exception
{
byte[] ids = (byte[])message.removeProperty(MessageImpl.HDR_ROUTE_TO_IDS);
Modified: branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/postoffice/impl/DivertBinding.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/postoffice/impl/DivertBinding.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/postoffice/impl/DivertBinding.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -124,15 +124,30 @@
return BindingType.DIVERT;
}
+
+
/* (non-Javadoc)
* @see java.lang.Object#toString()
*/
@Override
public String toString()
{
- return "DivertBinding [divert=" + divert + "]";
+ return "DivertBinding [id=" + id +
+ ", address=" +
+ address +
+ ", divert=" +
+ divert +
+ ", filter=" +
+ filter +
+ ", uniqueName=" +
+ uniqueName +
+ ", routingName=" +
+ routingName +
+ ", exclusive=" +
+ exclusive +
+ "]";
}
-
+
public void close() throws Exception
{
}
Modified: branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -77,6 +77,8 @@
public class PostOfficeImpl implements PostOffice, NotificationListener, BindingsFactory
{
private static final Logger log = Logger.getLogger(PostOfficeImpl.class);
+
+ private static final boolean isTrace = log.isTraceEnabled();
public static final SimpleString HDR_RESET_QUEUE_DATA = new SimpleString("_HQ_RESET_QUEUE_DATA");
@@ -209,6 +211,10 @@
public void onNotification(final Notification notification)
{
+ if (isTrace)
+ {
+ log.trace("Receiving notification : " + notification);
+ }
synchronized (notificationLock)
{
NotificationType type = notification.getType();
@@ -1306,6 +1312,6 @@
public Bindings createBindings(final SimpleString address) throws Exception
{
- return new BindingsImpl(server.getGroupingHandler(), pagingManager.getPageStore(address));
+ return new BindingsImpl(address, server.getGroupingHandler(), pagingManager.getPageStore(address));
}
}
Modified: branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/protocol/core/impl/ChannelImpl.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/protocol/core/impl/ChannelImpl.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/protocol/core/impl/ChannelImpl.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -38,6 +38,8 @@
public class ChannelImpl implements Channel
{
private static final Logger log = Logger.getLogger(ChannelImpl.class);
+
+ private static final boolean isTrace = log.isTraceEnabled();
private volatile long id;
@@ -159,6 +161,11 @@
synchronized (sendLock)
{
packet.setChannelID(id);
+
+ if (isTrace)
+ {
+ log.trace("Sending packet nonblocking " + packet + " on channeID=" + id);
+ }
HornetQBuffer buffer = packet.encode(connection);
@@ -168,6 +175,10 @@
{
while (failingOver)
{
+ if (isTrace)
+ {
+ log.trace("Waiting fail over condition to clear on channelID=" + id);
+ }
// TODO - don't hardcode this timeout
try
{
@@ -176,6 +187,10 @@
catch (InterruptedException e)
{
}
+ if (isTrace)
+ {
+ log.trace("FailOver condition cleared on channelID=" + id);
+ }
}
// Sanity check
@@ -193,7 +208,13 @@
{
lock.unlock();
}
+
+ if (isTrace)
+ {
+ log.trace("Writing buffer for channelID=" + id);
+ }
+
// The actual send must be outside the lock, or with OIO transport, the write can block if the tcp
// buffer is full, preventing any incoming buffers being handled and blocking failover
connection.getTransportConnection().write(buffer, flush, batch);
@@ -350,6 +371,10 @@
{
if (resendCache != null)
{
+ if (isTrace)
+ {
+ log.trace("Replaying commands on channelID=" + id);
+ }
clearUpTo(otherLastConfirmedCommandID);
for (final Packet packet : resendCache)
Modified: branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/cluster/Bridge.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/cluster/Bridge.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/cluster/Bridge.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -13,6 +13,8 @@
package org.hornetq.core.server.cluster;
+import java.util.concurrent.Executor;
+
import org.hornetq.api.core.SimpleString;
import org.hornetq.core.server.Consumer;
import org.hornetq.core.server.HornetQComponent;
@@ -33,6 +35,12 @@
public interface Bridge extends Consumer, HornetQComponent
{
SimpleString getName();
+
+ /**
+ * Get the executor that is on the context of this bridge
+ * @return
+ */
+ Executor getExecutor();
Queue getQueue();
@@ -43,7 +51,9 @@
boolean isUseDuplicateDetection();
void activate();
-
+
+ void flushExecutor();
+
void setNotificationService(NotificationService notificationService);
RemotingConnection getForwardingConnection();
Modified: branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/cluster/MessageFlowRecord.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/cluster/MessageFlowRecord.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/cluster/MessageFlowRecord.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -33,6 +33,10 @@
Bridge getBridge();
void close() throws Exception;
+
+ public void resume() throws Exception;
+
+ boolean isClosed();
void reset() throws Exception;
Modified: branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -195,8 +195,17 @@
while ((ref = refs.poll()) != null)
{
+ if (isTrace)
+ {
+ log.trace("Cancelling reference " + ref + " on bridge " + this);
+ }
list.addFirst(ref);
}
+
+ if (isTrace && list.isEmpty())
+ {
+ log.trace("didn't have any references to cancel on bridge " + this);
+ }
Queue queue = null;
@@ -210,18 +219,45 @@
}
}
+
+ public Executor getExecutor()
+ {
+ return executor;
+ }
+ public void flushExecutor()
+ {
+ // Wait for any create objects runnable to complete
+ Future future = new Future();
+ executor.execute(future);
+
+ boolean ok = future.await(10000);
+
+ if (!ok)
+ {
+ BridgeImpl.log.warn("Timed out waiting to stop");
+ }
+ }
+
+
public void stop() throws Exception
{
if (started)
{
- // We need to stop the csf here otherwise the stop runnable never runs since the createobjectsrunnable is
- // trying to connect to the target
- // server which isn't up in an infinite loop
- if (csf != null)
+ executor.execute(new Runnable()
{
- csf.close();
- }
+ public void run()
+ {
+ // We need to stop the csf here otherwise the stop runnable never runs since the createobjectsrunnable is
+ // trying to connect to the target
+ // server which isn't up in an infinite loop
+ if (csf != null)
+ {
+ //csf.close();
+ csf = null;
+ }
+ }
+ });
}
log.info("Bridge " + this.name + " being stopped");
@@ -229,9 +265,7 @@
stopping = true;
executor.execute(new StopRunnable());
-
- waitForRunnablesToComplete();
-
+
if (notificationService != null)
{
TypedProperties props = new TypedProperties();
@@ -424,6 +458,10 @@
public void connectionFailed(final HornetQException me, boolean failedOver)
{
log.warn(name + "::Connection failed with failedOver=" + failedOver, me);
+ if (isTrace)
+ {
+ log.trace("Calling BridgeImpl::connectionFailed(HOrnetQException me=" + me + ", boolean failedOver=" + failedOver);
+ }
fail(false);
}
@@ -432,6 +470,8 @@
log.warn(name + "::Connection failed before reconnect ", exception);
fail(true);
}
+
+
// Package protected ---------------------------------------------
@@ -439,19 +479,29 @@
// Private -------------------------------------------------------
- private void waitForRunnablesToComplete()
+ /* (non-Javadoc)
+ * @see java.lang.Object#toString()
+ */
+ @Override
+ public String toString()
{
- // Wait for any create objects runnable to complete
- Future future = new Future();
-
- executor.execute(future);
-
- boolean ok = future.await(10000);
-
- if (!ok)
- {
- BridgeImpl.log.warn("Timed out waiting to stop");
- }
+ return this.getClass().getName() +
+ " [name=" + name +
+ ", nodeUUID=" +
+ nodeUUID +
+ ", queue=" +
+ queue +
+ ", filter=" +
+ filter +
+ ", forwardingAddress=" +
+ forwardingAddress +
+ ", useDuplicateDetection=" +
+ useDuplicateDetection +
+ ", active=" +
+ active +
+ ", stopping=" +
+ stopping +
+ "]";
}
private void fail(final boolean beforeReconnect)
@@ -654,20 +704,17 @@
{
try
{
+ // We need to close the session outside of the lock,
+ // so any pending operation will be canceled right away
+ if (session != null)
+ {
+ session.close();
+ }
+
synchronized (BridgeImpl.this)
{
- if (!started)
- {
- return;
- }
-
log.debug("Closing Session for bridge " + BridgeImpl.this.name);
- if (session != null)
- {
- session.close();
- }
-
started = false;
active = false;
Modified: branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionBridge.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionBridge.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionBridge.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -50,6 +50,8 @@
public class ClusterConnectionBridge extends BridgeImpl
{
private static final Logger log = Logger.getLogger(ClusterConnectionBridge.class);
+
+ private static final boolean isTrace = log.isTraceEnabled();
private final MessageFlowRecord flowRecord;
@@ -147,7 +149,8 @@
{
if (flowRecord != null)
{
- flowRecord.reset();
+ // TODO: can I really remove this?
+ // flowRecord.reset();
if (notifConsumer != null)
{
@@ -195,7 +198,7 @@
flowRecord.getAddress() +
"%')");
- session.createQueue(managementNotificationAddress, notifQueueName, filter, false);
+ session.createTemporaryQueue(managementNotificationAddress, notifQueueName, filter);
notifConsumer = session.createConsumer(notifQueueName);
@@ -224,6 +227,12 @@
}
@Override
+ public void stop() throws Exception
+ {
+ super.stop();
+ }
+
+ @Override
protected ClientSessionFactory createSessionFactory() throws Exception
{
//We create the session factory using the specified connector
@@ -234,6 +243,11 @@
@Override
public void connectionFailed(HornetQException me, boolean failedOver)
{
+ if (isTrace)
+ {
+ log.trace("Connection Failed on ClusterConnectionBridge, failedOver = " + failedOver + ", sessionClosed = " + session.isClosed(), new Exception ("trace"));
+ }
+
if (!failedOver && !session.isClosed())
{
try
Modified: branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -65,6 +65,8 @@
public class ClusterConnectionImpl implements ClusterConnection
{
private static final Logger log = Logger.getLogger(ClusterConnectionImpl.class);
+
+ private static final boolean isTrace = log.isTraceEnabled();
private final org.hornetq.utils.ExecutorFactory executorFactory;
@@ -78,7 +80,17 @@
private final SimpleString address;
+ private final long clientFailureCheckPeriod;
+
+ private final long connectionTTL;
+
private final long retryInterval;
+
+ private final double retryIntervalMultiplier;
+
+ private final long maxRetryInterval;
+
+ private final int reconnectAttempts;
private final boolean useDuplicateDetection;
@@ -114,7 +126,12 @@
final TransportConfiguration connector,
final SimpleString name,
final SimpleString address,
+ final long clientFailureCheckPeriod,
+ final long connectionTTL,
final long retryInterval,
+ final double retryIntervalMultiplier,
+ final long maxRetryInterval,
+ final int reconnectAttempts,
final boolean useDuplicateDetection,
final boolean routeWhenNoConsumers,
final int confirmationWindowSize,
@@ -144,7 +161,17 @@
this.address = address;
+ this.clientFailureCheckPeriod = clientFailureCheckPeriod;
+
+ this.connectionTTL = connectionTTL;
+
this.retryInterval = retryInterval;
+
+ this.retryIntervalMultiplier = retryIntervalMultiplier;
+
+ this.maxRetryInterval = maxRetryInterval;
+
+ this.reconnectAttempts = reconnectAttempts;
this.useDuplicateDetection = useDuplicateDetection;
@@ -188,7 +215,12 @@
final TransportConfiguration connector,
final SimpleString name,
final SimpleString address,
- final long retryInterval,
+ final long clientFailureCheckPeriod,
+ final long connectionTTL,
+ final long retryInterval,
+ final double retryIntervalMultiplier,
+ final long maxRetryInterval,
+ final int reconnectAttempts,
final boolean useDuplicateDetection,
final boolean routeWhenNoConsumers,
final int confirmationWindowSize,
@@ -218,7 +250,17 @@
this.address = address;
+ this.clientFailureCheckPeriod = clientFailureCheckPeriod;
+
+ this.connectionTTL = connectionTTL;
+
this.retryInterval = retryInterval;
+
+ this.retryIntervalMultiplier = retryIntervalMultiplier;
+
+ this.maxRetryInterval = maxRetryInterval;
+
+ this.reconnectAttempts = reconnectAttempts;
this.useDuplicateDetection = useDuplicateDetection;
@@ -355,13 +397,26 @@
{
serverLocator.setNodeID(nodeUUID.toString());
- serverLocator.setReconnectAttempts(-1);
+ serverLocator.setReconnectAttempts(reconnectAttempts);
serverLocator.setClusterConnection(true);
serverLocator.setClusterTransportConfiguration(connector);
serverLocator.setBackup(server.getConfiguration().isBackup());
serverLocator.setInitialConnectAttempts(-1);
- serverLocator.setConfirmationWindowSize(0);
+ if (serverLocator.getConfirmationWindowSize() < 0)
+ {
+ // We can't have confirmationSize = -1 on the cluster Bridge
+ // Otherwise we won't have confirmation working
+ serverLocator.setConfirmationWindowSize(0);
+ }
+
+ if (!useDuplicateDetection)
+ {
+ log.debug("DuplicateDetection is disabled, sending clustered messages blocked");
+ }
+ // if not using duplicate detection, we will send blocked
+ serverLocator.setBlockOnDurableSend(!useDuplicateDetection);
+ serverLocator.setBlockOnNonDurableSend(!useDuplicateDetection);
if(retryInterval > 0)
{
@@ -393,7 +448,7 @@
public synchronized void nodeDown(final String nodeID)
{
- log.debug("node " + nodeID + " being considered down on cluster connection for nodeID=" + nodeUUID);
+ log.debug("node " + nodeID + " being considered down on cluster connection for nodeID=" + nodeUUID, new Exception ("trace"));
if (nodeID.equals(nodeUUID.toString()))
{
return;
@@ -407,7 +462,11 @@
{
try
{
- record.reset();
+ if (isTrace)
+ {
+ log.trace("Closing clustering record " + record);
+ }
+ record.close();
}
catch (Exception e)
{
@@ -485,11 +544,11 @@
}
else
{
- // FIXME apple and orange comparison. I don't understand it...
- //if (!connectorPair.a.equals(record.getBridge().getForwardingConnection().getTransportConnection()))
- // {
- // // New live node - close it and recreate it - TODO - CAN THIS EVER HAPPEN?
- //}
+ log.info("Reattaching nodeID=" + nodeID);
+ if (record.isClosed())
+ {
+ record.resume();
+ }
}
}
catch (Exception e)
@@ -502,6 +561,11 @@
public void nodeAnnounced(final String nodeID,
final Pair<TransportConfiguration, TransportConfiguration> connectorPair)
{
+ if (isTrace)
+ {
+ log.trace("nodeAnnouncedUp:" + nodeID);
+ }
+
if (nodeID.equals(nodeUUID.toString()))
{
return;
@@ -510,6 +574,10 @@
// if the node is more than 1 hop away, we do not create a bridge for direct cluster connection
if (allowDirectConnectionsOnly && !allowableConnections.contains(connectorPair.a))
{
+ if (isTrace)
+ {
+ log.trace("Ignoring nodeUp message as it only allows direct connections");
+ }
return;
}
@@ -517,20 +585,32 @@
// and empty static connectors to create bridges... ulgy!
if (serverLocator == null)
{
+ if (isTrace)
+ {
+ log.trace("Ignoring nodeUp as serverLocator==null");
+ }
return;
}
/*we dont create bridges to backups*/
if(connectorPair.a == null)
{
+ if (isTrace)
+ {
+ log.trace("Igoring nodeup as connectorPair.a==null (backup)");
+ }
return;
}
synchronized (records)
{
+ if (isTrace)
+ {
+ log.trace("Adding record for nodeID=" + nodeID);
+ }
try
{
MessageFlowRecord record = records.get(nodeID);
-
+
if (record == null)
{
// New node - create a new flow record
@@ -556,11 +636,10 @@
}
else
{
- // FIXME apple and orange comparison. I don't understand it...
- //if (!connectorPair.a.equals(record.getBridge().getForwardingConnection().getTransportConnection()))
- // {
- // // New live node - close it and recreate it - TODO - CAN THIS EVER HAPPEN?
- //}
+ if (isTrace)
+ {
+ log.trace("It already had a node created before, ignoring the nodeUp message");
+ }
}
}
catch (Exception e)
@@ -576,13 +655,32 @@
final Queue queue,
final boolean start) throws Exception
{
- MessageFlowRecordImpl record = new MessageFlowRecordImpl(queue);
+ MessageFlowRecordImpl record = new MessageFlowRecordImpl(nodeID, connector, queueName, queue);
- Bridge bridge = new ClusterConnectionBridge(serverLocator,
+ records.put(nodeID, record);
+
+ Bridge bridge = createBridge(record);
+
+ record.setBridge(bridge);
+
+ if (start)
+ {
+ bridge.start();
+ }
+ }
+
+ /**
+ * @param record
+ * @return
+ * @throws Exception
+ */
+ protected Bridge createBridge(MessageFlowRecordImpl record) throws Exception
+ {
+ ClusterConnectionBridge bridge = new ClusterConnectionBridge(serverLocator,
nodeUUID,
- nodeID,
- queueName,
- queue,
+ record.getNodeID(),
+ record.getQueueName(),
+ record.getQueue(),
executorFactory.getExecutor(),
null,
null,
@@ -596,16 +694,9 @@
managementService.getManagementAddress(),
managementService.getManagementNotificationAddress(),
record,
- connector);
+ record.getConnector());
- record.setBridge(bridge);
-
- records.put(nodeID, record);
-
- if (start)
- {
- bridge.start();
- }
+ return bridge;
}
// Inner classes -----------------------------------------------------------------------------------
@@ -614,15 +705,26 @@
{
private Bridge bridge;
+ private final String nodeID;
+ private final TransportConfiguration connector;
+ private final SimpleString queueName;
private final Queue queue;
private final Map<SimpleString, RemoteQueueBinding> bindings = new HashMap<SimpleString, RemoteQueueBinding>();
+
+ private volatile boolean isClosed = false;
private volatile boolean firstReset = false;
- public MessageFlowRecordImpl(final Queue queue)
+ public MessageFlowRecordImpl(final String nodeID,
+ final TransportConfiguration connector,
+ final SimpleString queueName,
+ final Queue queue)
{
this.queue = queue;
+ this.nodeID = nodeID;
+ this.connector = connector;
+ this.queueName = queueName;
}
public String getAddress()
@@ -630,6 +732,38 @@
return address.toString();
}
+ /**
+ * @return the nodeID
+ */
+ public String getNodeID()
+ {
+ return nodeID;
+ }
+
+ /**
+ * @return the connector
+ */
+ public TransportConfiguration getConnector()
+ {
+ return connector;
+ }
+
+ /**
+ * @return the queueName
+ */
+ public SimpleString getQueueName()
+ {
+ return queueName;
+ }
+
+ /**
+ * @return the queue
+ */
+ public Queue getQueue()
+ {
+ return queue;
+ }
+
public int getMaxHops()
{
return maxHops;
@@ -637,10 +771,29 @@
public void close() throws Exception
{
+ if (isTrace)
+ {
+ log.trace("Stopping bridge " + bridge);
+ }
+
+ isClosed = true;
+ clearBindings();
+
bridge.stop();
-
- clearBindings();
}
+
+ public void resume() throws Exception
+ {
+ isClosed = false;
+ this.bridge = createBridge(this);
+ bridge.start();
+ bridge.activate();
+ }
+
+ public boolean isClosed()
+ {
+ return isClosed;
+ }
public void reset() throws Exception
{
@@ -659,6 +812,10 @@
public synchronized void onMessage(final ClientMessage message)
{
+ if (isTrace)
+ {
+ log.trace("Receiving message " + message);
+ }
try
{
// Reset the bindings
Modified: branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -17,7 +17,12 @@
import java.lang.reflect.Array;
import java.net.InetAddress;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
@@ -32,7 +37,10 @@
import org.hornetq.core.client.impl.ServerLocatorInternal;
import org.hornetq.core.client.impl.Topology;
import org.hornetq.core.client.impl.TopologyMember;
-import org.hornetq.core.config.*;
+import org.hornetq.core.config.BridgeConfiguration;
+import org.hornetq.core.config.BroadcastGroupConfiguration;
+import org.hornetq.core.config.ClusterConnectionConfiguration;
+import org.hornetq.core.config.Configuration;
import org.hornetq.core.logging.Logger;
import org.hornetq.core.postoffice.Binding;
import org.hornetq.core.postoffice.PostOffice;
@@ -682,8 +690,16 @@
serverLocator.setReconnectAttempts(config.getReconnectAttempts());
serverLocator.setRetryInterval(config.getRetryInterval());
serverLocator.setRetryIntervalMultiplier(config.getRetryIntervalMultiplier());
+ serverLocator.setMaxRetryInterval(config.getMaxRetryInterval());
serverLocator.setClientFailureCheckPeriod(config.getClientFailureCheckPeriod());
serverLocator.setInitialConnectAttempts(config.getReconnectAttempts());
+ serverLocator.setBlockOnDurableSend(!config.isUseDuplicateDetection());
+ serverLocator.setBlockOnNonDurableSend(!config.isUseDuplicateDetection());
+ if (!config.isUseDuplicateDetection())
+ {
+ log.debug("Bridge " + config.getName() +
+ " is configured to not use duplicate detecion, it will send messages synchronously");
+ }
clusterLocators.add(serverLocator);
Bridge bridge = new BridgeImpl(serverLocator,
nodeUUID,
@@ -769,7 +785,12 @@
connector,
new SimpleString(config.getName()),
new SimpleString(config.getAddress()),
+ config.getClientFailureCheckPeriod(),
+ config.getConnectionTTL(),
config.getRetryInterval(),
+ config.getRetryIntervalMultiplier(),
+ config.getMaxRetryInterval(),
+ config.getReconnectAttempts(),
config.isDuplicateDetection(),
config.isForwardWhenNoConsumers(),
config.getConfirmationWindowSize(),
@@ -794,7 +815,12 @@
connector,
new SimpleString(config.getName()),
new SimpleString(config.getAddress()),
+ config.getClientFailureCheckPeriod(),
+ config.getConnectionTTL(),
config.getRetryInterval(),
+ config.getRetryIntervalMultiplier(),
+ config.getMaxRetryInterval(),
+ config.getReconnectAttempts(),
config.isDuplicateDetection(),
config.isForwardWhenNoConsumers(),
config.getConfirmationWindowSize(),
Modified: branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/impl/QueueImpl.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/impl/QueueImpl.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/impl/QueueImpl.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -853,6 +853,7 @@
public synchronized void cancel(final MessageReference reference, final long timeBase) throws Exception
{
+ deliveringCount.decrementAndGet();
if (checkRedelivery(reference, timeBase))
{
if (!scheduledDeliveryHandler.checkAndSchedule(reference, false))
@@ -870,7 +871,7 @@
{
if (isTrace)
{
- log.trace("moving expired reference " + ref + " to address = " + expiryAddress + " from queue=" + this.getName(), new Exception ("trace"));
+ log.trace("moving expired reference " + ref + " to address = " + expiryAddress + " from queue=" + this.getName());
}
move(expiryAddress, ref, true, false);
}
Modified: branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -88,6 +88,8 @@
// Constants -----------------------------------------------------------------------------
private static final Logger log = Logger.getLogger(ServerSessionImpl.class);
+
+ private static final boolean isTrace = log.isTraceEnabled();
// Static -------------------------------------------------------------------------------
@@ -598,6 +600,10 @@
public void commit() throws Exception
{
+ if (isTrace)
+ {
+ log.trace("Calling commit");
+ }
try
{
tx.commit();
@@ -1075,7 +1081,7 @@
public void send(final ServerMessage message, final boolean direct) throws Exception
{
long id = storageManager.generateUniqueID();
-
+
SimpleString address = message.getAddress();
message.setMessageID(id);
@@ -1096,6 +1102,12 @@
}
}
+ if (isTrace)
+ {
+ log.trace("send(message=" + message + ", direct=" + direct + ") being called");
+ }
+
+
if (message.getAddress().equals(managementAddress))
{
// It's a management message
Modified: branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/management/impl/ManagementServiceImpl.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/management/impl/ManagementServiceImpl.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/src/main/org/hornetq/core/server/management/impl/ManagementServiceImpl.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -84,6 +84,8 @@
// Constants -----------------------------------------------------
private static final Logger log = Logger.getLogger(ManagementServiceImpl.class);
+
+ private static final boolean isTrace = log.isTraceEnabled();
private final MBeanServer mbeanServer;
@@ -635,6 +637,12 @@
public void sendNotification(final Notification notification) throws Exception
{
+ if (isTrace)
+ {
+ log.trace("Sending Notification = " + notification +
+ ", notificationEnabled=" + notificationsEnabled +
+ " messagingServerControl=" + messagingServerControl, new Exception ("trace"));
+ }
if (messagingServerControl != null && notificationsEnabled)
{
// This needs to be synchronized since we need to ensure notifications are processed in strict sequence
Modified: branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeReconnectTest.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeReconnectTest.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeReconnectTest.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -118,12 +118,14 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
+ HornetQClient.DEFAULT_CONNECTION_TTL,
retryInterval,
+ HornetQClient.DEFAULT_MAX_RETRY_INTERVAL,
retryIntervalMultiplier,
reconnectAttempts,
false,
confirmationWindowSize,
- HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
staticConnectors,
false,
ConfigurationImpl.DEFAULT_CLUSTER_USER,
@@ -260,12 +262,14 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
+ HornetQClient.DEFAULT_CONNECTION_TTL,
retryInterval,
+ HornetQClient.DEFAULT_MAX_RETRY_INTERVAL,
retryIntervalMultiplier,
reconnectAttempts,
false,
confirmationWindowSize,
- HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
staticConnectors,
false,
ConfigurationImpl.DEFAULT_CLUSTER_USER,
@@ -386,12 +390,14 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
+ HornetQClient.DEFAULT_CONNECTION_TTL,
retryInterval,
+ HornetQClient.DEFAULT_MAX_RETRY_INTERVAL,
retryIntervalMultiplier,
reconnectAttempts,
false,
confirmationWindowSize,
- HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
staticConnectors,
false,
ConfigurationImpl.DEFAULT_CLUSTER_USER,
@@ -527,12 +533,14 @@
forwardAddress,
null,
null,
+ clientFailureCheckPeriod,
+ HornetQClient.DEFAULT_CONNECTION_TTL,
retryInterval,
+ HornetQClient.DEFAULT_MAX_RETRY_INTERVAL,
retryIntervalMultiplier,
reconnectAttempts,
false,
confirmationWindowSize,
- clientFailureCheckPeriod,
staticConnectors,
false,
ConfigurationImpl.DEFAULT_CLUSTER_USER,
@@ -661,12 +669,14 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
+ HornetQClient.DEFAULT_CONNECTION_TTL,
retryInterval,
+ HornetQClient.DEFAULT_MAX_RETRY_INTERVAL,
retryIntervalMultiplier,
reconnectAttempts,
false,
confirmationWindowSize,
- HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
staticConnectors,
false,
ConfigurationImpl.DEFAULT_CLUSTER_USER,
Modified: branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeStartTest.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeStartTest.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeStartTest.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -104,12 +104,14 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
+ HornetQClient.DEFAULT_CONNECTION_TTL,
1000,
+ HornetQClient.DEFAULT_MAX_RETRY_INTERVAL,
1d,
0,
true,
1024,
- HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
staticConnectors,
false,
ConfigurationImpl.DEFAULT_CLUSTER_USER,
@@ -176,6 +178,8 @@
Bridge bridge = server0.getClusterManager().getBridges().get(bridgeName);
bridge.stop();
+
+ bridge.flushExecutor();
for (int i = 0; i < numMessages; i++)
{
@@ -267,12 +271,14 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
+ HornetQClient.DEFAULT_CONNECTION_TTL,
500,
+ HornetQClient.DEFAULT_MAX_RETRY_INTERVAL,
1d,
-1,
true,
1024,
- HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
staticConnectors,
false,
ConfigurationImpl.DEFAULT_CLUSTER_USER,
@@ -478,12 +484,14 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
+ HornetQClient.DEFAULT_CONNECTION_TTL,
1000,
+ HornetQClient.DEFAULT_MAX_RETRY_INTERVAL,
1d,
0,
false,
1024,
- HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
staticConnectors,
false,
ConfigurationImpl.DEFAULT_CLUSTER_USER,
@@ -629,12 +637,14 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
+ HornetQClient.DEFAULT_CONNECTION_TTL,
1000,
+ HornetQClient.DEFAULT_MAX_RETRY_INTERVAL,
1d,
1,
true,
1024,
- HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
staticConnectors,
false,
ConfigurationImpl.DEFAULT_CLUSTER_USER,
@@ -706,6 +716,8 @@
BridgeStartTest.log.info("stopping bridge manually");
bridge.stop();
+
+ bridge.flushExecutor();
for (int i = numMessages; i < numMessages * 2; i++)
{
@@ -739,6 +751,8 @@
Assert.assertNull(consumer1.receiveImmediate());
bridge.stop();
+
+ bridge.flushExecutor();
for (int i = 0; i < numMessages; i++)
{
Modified: branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeTest.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeTest.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeTest.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -139,14 +139,16 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
+ HornetQClient.DEFAULT_CONNECTION_TTL,
1000,
+ HornetQClient.DEFAULT_MAX_RETRY_INTERVAL,
1d,
-1,
false,
// Choose confirmation size to make sure acks
// are sent
numMessages * messageSize / 2,
- HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
connectorConfig,
false,
ConfigurationImpl.DEFAULT_CLUSTER_USER,
@@ -339,12 +341,14 @@
forwardAddress,
filterString,
null,
+ HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
+ HornetQClient.DEFAULT_CONNECTION_TTL,
1000,
+ HornetQClient.DEFAULT_MAX_RETRY_INTERVAL,
1d,
-1,
false,
1024,
- HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
staticConnectors,
false,
ConfigurationImpl.DEFAULT_CLUSTER_USER,
@@ -510,12 +514,14 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
+ HornetQClient.DEFAULT_CONNECTION_TTL,
100,
+ HornetQClient.DEFAULT_MAX_RETRY_INTERVAL,
1d,
-1,
false,
1024,
- HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
staticConnectors,
false,
ConfigurationImpl.DEFAULT_CLUSTER_USER,
@@ -665,12 +671,14 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
+ HornetQClient.DEFAULT_CONNECTION_TTL,
100,
+ HornetQClient.DEFAULT_MAX_RETRY_INTERVAL,
1d,
-1,
true,
0,
- HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
staticConnectors,
false,
ConfigurationImpl.DEFAULT_CLUSTER_USER,
@@ -853,12 +861,14 @@
forwardAddress,
null,
SimpleTransformer.class.getName(),
+ HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
+ HornetQClient.DEFAULT_CONNECTION_TTL,
1000,
+ HornetQClient.DEFAULT_MAX_RETRY_INTERVAL,
1d,
-1,
false,
1024,
- HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
staticConnectors,
false,
ConfigurationImpl.DEFAULT_CLUSTER_USER,
@@ -1001,12 +1011,14 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
+ HornetQClient.DEFAULT_CONNECTION_TTL,
1000,
+ HornetQClient.DEFAULT_MAX_RETRY_INTERVAL,
1d,
-1,
false,
0,
- HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
staticConnectors,
false,
ConfigurationImpl.DEFAULT_CLUSTER_USER,
@@ -1242,12 +1254,14 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
+ HornetQClient.DEFAULT_CONNECTION_TTL,
1000,
+ HornetQClient.DEFAULT_MAX_RETRY_INTERVAL,
1d,
-1,
false,
1024,
- HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
staticConnectors,
false,
ConfigurationImpl.DEFAULT_CLUSTER_USER,
@@ -1385,14 +1399,16 @@
// address
null,
null,
+ HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
+ HornetQClient.DEFAULT_CONNECTION_TTL,
1000,
+ HornetQClient.DEFAULT_MAX_RETRY_INTERVAL,
1d,
-1,
false,
// Choose confirmation size to make sure acks
// are sent
numMessages * messageSize / 2,
- HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
staticConnectors,
false,
ConfigurationImpl.DEFAULT_CLUSTER_USER,
Modified: branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeWithDiscoveryGroupStartTest.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeWithDiscoveryGroupStartTest.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeWithDiscoveryGroupStartTest.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -116,12 +116,14 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
+ HornetQClient.DEFAULT_CONNECTION_TTL,
1000,
+ HornetQClient.DEFAULT_MAX_RETRY_INTERVAL,
1d,
0,
true,
1024,
- HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
staticConnectors,
false,
ConfigurationImpl.DEFAULT_CLUSTER_USER,
@@ -188,6 +190,7 @@
Bridge bridge = server0.getClusterManager().getBridges().get(bridgeName);
bridge.stop();
+ bridge.flushExecutor();
for (int i = 0; i < numMessages; i++)
{
Modified: branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeWithPagingTest.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeWithPagingTest.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeWithPagingTest.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -113,13 +113,15 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
+ HornetQClient.DEFAULT_CONNECTION_TTL,
retryInterval,
+ HornetQClient.DEFAULT_MAX_RETRY_INTERVAL,
retryIntervalMultiplier,
reconnectAttempts,
false,
confirmationWindowSize,
- HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
- staticConnectors,
+ staticConnectors,
false,
ConfigurationImpl.DEFAULT_CLUSTER_USER,
ConfigurationImpl.DEFAULT_CLUSTER_PASSWORD);
Modified: branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/cluster/distribution/OnewayTwoNodeClusterTest.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/cluster/distribution/OnewayTwoNodeClusterTest.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/cluster/distribution/OnewayTwoNodeClusterTest.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -119,7 +119,7 @@
send(0, "queues.testaddress", 10, false, null);
verifyNotReceive(0);
}
-
+
public void testStopAndStartTarget() throws Exception
{
startServers(0, 1);
@@ -150,9 +150,14 @@
stopServers(1);
- OnewayTwoNodeClusterTest.log.info("restarting server 1");
+ OnewayTwoNodeClusterTest.log.info("restarting server 1(" + servers[1].getIdentity() + ")");
startServers(1);
+
+ //Thread.sleep(1000);
+
+ log.info("Server 1 id=" + servers[1].getNodeID());
+
long end = System.currentTimeMillis();
Modified: branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/management/BridgeControlTest.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/management/BridgeControlTest.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/management/BridgeControlTest.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -166,12 +166,14 @@
targetQueueConfig.getAddress(),
null,
null,
+ HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
+ HornetQClient.DEFAULT_CONNECTION_TTL,
RandomUtil.randomPositiveLong(),
+ HornetQClient.DEFAULT_MAX_RETRY_INTERVAL,
RandomUtil.randomDouble(),
RandomUtil.randomPositiveInt(),
RandomUtil.randomBoolean(),
RandomUtil.randomPositiveInt(),
- HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
connectors,
false,
ConfigurationImpl.DEFAULT_CLUSTER_USER,
Modified: branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/management/BridgeControlUsingCoreTest.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/management/BridgeControlUsingCoreTest.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/integration/management/BridgeControlUsingCoreTest.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -144,12 +144,14 @@
targetQueueConfig.getAddress(),
null,
null,
+ HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
+ HornetQClient.DEFAULT_CONNECTION_TTL,
RandomUtil.randomPositiveLong(),
+ HornetQClient.DEFAULT_MAX_RETRY_INTERVAL,
RandomUtil.randomDouble(),
RandomUtil.randomPositiveInt(),
RandomUtil.randomBoolean(),
RandomUtil.randomPositiveInt(),
- HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
connectors,
false,
ConfigurationImpl.DEFAULT_CLUSTER_USER,
Modified: branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java
===================================================================
--- branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java 2011-06-13 02:55:40 UTC (rev 10792)
+++ branches/Branch_2_2_EAP-cluster-cleanup/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java 2011-06-13 02:59:47 UTC (rev 10793)
@@ -87,7 +87,7 @@
{
final FakeBinding fake = new FakeBinding(new SimpleString("a"));
- final BindingsImpl bind = new BindingsImpl(null, null);
+ final BindingsImpl bind = new BindingsImpl(null, null, null);
bind.addBinding(fake);
bind.addBinding(new FakeBinding(new SimpleString("a")));
bind.addBinding(new FakeBinding(new SimpleString("a")));
More information about the hornetq-commits
mailing list