Author: clebert.suconic(a)jboss.com Date: 2009-11-03 13:12:24 -0500 (Tue, 03 Nov 2009) New Revision: 8197 Modified: trunk/docs/user-manual/en/configuration-index.xml trunk/src/config/common/schema/hornetq-configuration.xsd trunk/src/main/org/hornetq/core/config/Configuration.java trunk/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java trunk/src/main/org/hornetq/core/config/impl/FileConfiguration.java trunk/src/main/org/hornetq/core/replication/impl/ReplicationManagerImpl.java trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java trunk/tests/config/ConfigurationTest-full-config.xml trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java trunk/tests/src/org/hornetq/tests/unit/core/config/impl/ConfigurationImplTest.java trunk/tests/src/org/hornetq/tests/unit/core/config/impl/FileConfigurationTest.java Log: https://jira.jboss.org/jira/browse/HORNETQ-125 - Few Tweaks, adding parameter Modified: trunk/docs/user-manual/en/configuration-index.xml =================================================================== --- trunk/docs/user-manual/en/configuration-index.xml 2009-11-03 18:03:04 UTC (rev 8196) +++ trunk/docs/user-manual/en/configuration-index.xml 2009-11-03 18:12:24 UTC (rev 8197) @@ -57,6 +57,14 @@ <entry/> </row> <row> + <entry><link linkend="configuring.live.backup" + >backup-window-size</link></entry> + <entry>int</entry> + <entry>The Window Size used to flow control between live and backup</entry> + <entry/> + <entry>1 MiB</entry> + </row> + <row> <entry><link linkend="configuring.bindings.journal" <entry>String</entry> Modified: trunk/src/config/common/schema/hornetq-configuration.xsd =================================================================== --- trunk/src/config/common/schema/hornetq-configuration.xsd 2009-11-03 18:03:04 UTC (rev 8196) +++ trunk/src/config/common/schema/hornetq-configuration.xsd 2009-11-03 18:12:24 UTC (rev 8197) @@ -75,6 +75,8 @@ </xsd:element> <xsd:element maxOccurs="1" minOccurs="0" name="backup-connector-ref" type="backup-connectorType"> </xsd:element> + <xsd:element maxOccurs="1" minOccurs="0" name="backup-window-size" type="xsd:int"> + </xsd:element> <xsd:element maxOccurs="1" minOccurs="0" name="connectors"> <xsd:complexType> <xsd:sequence> Modified: trunk/src/main/org/hornetq/core/config/Configuration.java =================================================================== --- trunk/src/main/org/hornetq/core/config/Configuration.java 2009-11-03 18:03:04 UTC (rev 8196) +++ trunk/src/main/org/hornetq/core/config/Configuration.java 2009-11-03 18:12:24 UTC (rev 8197) @@ -119,6 +119,10 @@ void setConnectorConfigurations(Map<String, TransportConfiguration> infos); String getBackupConnectorName(); + + int getBackupWindowSize(); + + void setBackupWindowSize(int windowSize); void setBackupConnectorName(String name); Modified: trunk/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java 2009-11-03 18:03:04 UTC (rev 8196) +++ trunk/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java 2009-11-03 18:12:24 UTC (rev 8197) @@ -49,7 +49,7 @@ public static final boolean DEFAULT_PERSIST_DELIVERY_COUNT_BEFORE_DELIVERY = false; public static final boolean DEFAULT_BACKUP = false; - + public static final boolean DEFAULT_SHARED_STORE = false; public static final boolean DEFAULT_FILE_DEPLOYMENT_ENABLED = false; @@ -165,9 +165,11 @@ public static final long DEFAULT_SERVER_DUMP_INTERVAL = -1; public static final int DEFAULT_MEMORY_WARNING_THRESHOLD = 25; - + public static final long DEFAULT_MEMORY_MEASURE_INTERVAL = 3000; // in milliseconds - + + public static final int DEFAULT_BACKUP_WINDOW_SIZE = 1024 * 1024; + public static final String DEFAULT_LOG_DELEGATE_FACTORY_CLASS_NAME = JULLogDelegateFactory.class.getCanonicalName(); // Attributes ----------------------------------------------------------------------------- @@ -175,9 +177,9 @@ protected boolean clustered = DEFAULT_CLUSTERED; protected boolean backup = DEFAULT_BACKUP; - + protected boolean sharedStore = DEFAULT_SHARED_STORE; - + protected boolean fileDeploymentEnabled = DEFAULT_FILE_DEPLOYMENT_ENABLED; protected boolean persistenceEnabled = DEFAULT_PERSISTENCE_ENABLED; @@ -209,7 +211,7 @@ protected int idCacheSize = DEFAULT_ID_CACHE_SIZE; protected boolean persistIDCache = DEFAULT_PERSIST_ID_CACHE; - + protected String logDelegateFactoryClassName = DEFAULT_LOG_DELEGATE_FACTORY_CLASS_NAME; protected List<String> interceptorClassNames = new ArrayList<String>(); @@ -220,6 +222,8 @@ protected String backupConnectorName; + protected int backupWindowSize = DEFAULT_BACKUP_WINDOW_SIZE; + protected List<BridgeConfiguration> bridgeConfigurations = new ArrayList<BridgeConfiguration>(); protected List<DivertConfiguration> divertConfigurations = new ArrayList<DivertConfiguration>(); @@ -302,7 +306,7 @@ // percentage of free memory which triggers warning from the memory manager protected int memoryWarningThreshold = DEFAULT_MEMORY_WARNING_THRESHOLD; - + protected long memoryMeasureInterval = DEFAULT_MEMORY_MEASURE_INTERVAL; protected GroupingHandlerConfiguration groupingHandlerConfiguration; @@ -387,7 +391,7 @@ { this.backup = backup; } - + public boolean isSharedStore() { return sharedStore; @@ -488,6 +492,19 @@ this.backupConnectorName = backupConnectorName; } + /* (non-Javadoc) + * @see org.hornetq.core.config.Configuration#getBackupWindowSize() + */ + public int getBackupWindowSize() + { + return backupWindowSize; + } + + public void setBackupWindowSize(int windowSize) + { + this.backupWindowSize = windowSize; + } + public GroupingHandlerConfiguration getGroupingHandlerConfiguration() { return groupingHandlerConfiguration; @@ -498,7 +515,6 @@ this.groupingHandlerConfiguration = groupingHandlerConfiguration; } - public List<BridgeConfiguration> getBridgeConfigurations() { return bridgeConfigurations; @@ -788,12 +804,12 @@ { jmxManagementEnabled = enabled; } - + public String getJMXDomain() { return jmxDomain; } - + public void setJMXDomain(String domain) { jmxDomain = domain; @@ -944,15 +960,21 @@ } else if (!bindingsDirectory.equals(other.bindingsDirectory)) return false; - if (clustered != other.clustered) + + if (backupWindowSize != other.backupWindowSize) + { return false; + } + + if (clustered != other.clustered) + return false; if (connectionTTLOverride != other.connectionTTLOverride) return false; if (createBindingsDir != other.createBindingsDir) return false; if (createJournalDir != other.createJournalDir) return false; - + if (fileDeploymentEnabled != other.fileDeploymentEnabled) return false; if (fileDeploymentScanPeriod != other.fileDeploymentScanPeriod) @@ -1107,12 +1129,12 @@ { this.serverDumpInterval = intervalInMilliseconds; } - + public int getMemoryWarningThreshold() { - return memoryWarningThreshold ; + return memoryWarningThreshold; } - + public void setMemoryWarningThreshold(int memoryWarningThreshold) { this.memoryWarningThreshold = memoryWarningThreshold; @@ -1120,9 +1142,9 @@ public long getMemoryMeasureInterval() { - return memoryMeasureInterval ; + return memoryMeasureInterval; } - + public void setMemoryMeasureInterval(long memoryMeasureInterval) { this.memoryMeasureInterval = memoryMeasureInterval; Modified: trunk/src/main/org/hornetq/core/config/impl/FileConfiguration.java =================================================================== --- trunk/src/main/org/hornetq/core/config/impl/FileConfiguration.java 2009-11-03 18:03:04 UTC (rev 8196) +++ trunk/src/main/org/hornetq/core/config/impl/FileConfiguration.java 2009-11-03 18:12:24 UTC (rev 8197) @@ -347,6 +347,8 @@ memoryMeasureInterval = getLong(e, "memory-measure-interval", memoryMeasureInterval, MINUS_ONE_OR_GT_ZERO); // in milliseconds + backupWindowSize = getInteger(e, "backup-window-size", DEFAULT_BACKUP_WINDOW_SIZE, MINUS_ONE_OR_GT_ZERO); + started = true; } Modified: trunk/src/main/org/hornetq/core/replication/impl/ReplicationManagerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/replication/impl/ReplicationManagerImpl.java 2009-11-03 18:03:04 UTC (rev 8196) +++ trunk/src/main/org/hornetq/core/replication/impl/ReplicationManagerImpl.java 2009-11-03 18:12:24 UTC (rev 8197) @@ -16,7 +16,6 @@ import java.util.Queue; import java.util.Set; import java.util.concurrent.ConcurrentLinkedQueue; -import java.util.concurrent.Executor; import org.hornetq.core.client.SessionFailureListener; import org.hornetq.core.client.impl.FailoverManager; @@ -64,8 +63,7 @@ // Attributes ---------------------------------------------------- - // TODO: where should this be configured? - private static final int CONF_WINDOW_SIZE = 1024 * 1024; + private final int backupWindowSize; private final ResponseHandler responseHandler = new ResponseHandler(); @@ -81,8 +79,6 @@ private final Object replicationLock = new Object(); - private final Executor executor; - private final ThreadLocal<ReplicationContext> tlReplicationContext = new ThreadLocal<ReplicationContext>(); private final Queue<ReplicationContext> pendingTokens = new ConcurrentLinkedQueue<ReplicationContext>(); @@ -96,11 +92,11 @@ /** * @param replicationConnectionManager */ - public ReplicationManagerImpl(final FailoverManager failoverManager, final Executor executor) + public ReplicationManagerImpl(final FailoverManager failoverManager, final int backupWindowSize) { super(); this.failoverManager = failoverManager; - this.executor = executor; + this.backupWindowSize = backupWindowSize; } // Public -------------------------------------------------------- @@ -323,12 +319,12 @@ Channel mainChannel = connection.getChannel(1, -1); - replicatingChannel = connection.getChannel(channelID, CONF_WINDOW_SIZE); + replicatingChannel = connection.getChannel(channelID, backupWindowSize); replicatingChannel.setHandler(responseHandler); CreateReplicationSessionMessage replicationStartPackage = new CreateReplicationSessionMessage(channelID, - CONF_WINDOW_SIZE); + backupWindowSize); mainChannel.sendBlocking(replicationStartPackage); Modified: trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-11-03 18:03:04 UTC (rev 8196) +++ trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-11-03 18:12:24 UTC (rev 8197) @@ -912,8 +912,7 @@ replicationFailoverManager = createBackupConnection(backupConnector, threadPool, scheduledPool); - this.replicationManager = new ReplicationManagerImpl(replicationFailoverManager, - this.executorFactory.getExecutor()); + this.replicationManager = new ReplicationManagerImpl(replicationFailoverManager, configuration.getBackupWindowSize()); replicationManager.start(); } } Modified: trunk/tests/config/ConfigurationTest-full-config.xml =================================================================== --- trunk/tests/config/ConfigurationTest-full-config.xml 2009-11-03 18:03:04 UTC (rev 8196) +++ trunk/tests/config/ConfigurationTest-full-config.xml 2009-11-03 18:12:24 UTC (rev 8197) @@ -57,6 +57,7 @@ </remoting-interceptors> <backup-connector-ref connector-name="backup-connector" /> + <backup-window-size>1024</backup-window-size> <connectors> <connector name="connector1"> <factory-class>org.hornetq.tests.unit.core.config.impl.TestConnectorFactory1</factory-class> Modified: trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java 2009-11-03 18:03:04 UTC (rev 8196) +++ trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java 2009-11-03 18:12:24 UTC (rev 8197) @@ -34,6 +34,7 @@ import org.hornetq.core.client.impl.FailoverManagerImpl; import org.hornetq.core.config.Configuration; import org.hornetq.core.config.TransportConfiguration; +import org.hornetq.core.config.impl.ConfigurationImpl; import org.hornetq.core.exception.HornetQException; import org.hornetq.core.journal.EncodingSupport; import org.hornetq.core.journal.Journal; @@ -108,8 +109,7 @@ try { - ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, - executor); + ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); manager.start(); manager.stop(); } @@ -134,8 +134,7 @@ try { - ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, - executor); + ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); manager.start(); try { @@ -174,15 +173,13 @@ try { - ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, - executor); + ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); manager.start(); try { - ReplicationManagerImpl manager2 = new ReplicationManagerImpl(failoverManager, - executor); + ReplicationManagerImpl manager2 = new ReplicationManagerImpl(failoverManager, ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); manager2.start(); fail("Exception was expected"); @@ -215,8 +212,7 @@ try { - ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, - executor); + ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); try { @@ -250,8 +246,7 @@ try { - ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, - executor); + ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); manager.start(); Journal replicatedJournal = new ReplicatedJournal((byte)1, new FakeJournal(), manager); @@ -368,8 +363,7 @@ try { - ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, - executor); + ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); manager.start(); Journal replicatedJournal = new ReplicatedJournal((byte)1, new FakeJournal(), manager); @@ -431,8 +425,7 @@ try { - ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, - executor); + ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); manager.start(); fail("Exception expected"); } @@ -457,8 +450,7 @@ try { - ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, - executor); + ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); manager.start(); Journal replicatedJournal = new ReplicatedJournal((byte)1, new FakeJournal(), manager); Modified: trunk/tests/src/org/hornetq/tests/unit/core/config/impl/ConfigurationImplTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/config/impl/ConfigurationImplTest.java 2009-11-03 18:03:04 UTC (rev 8196) +++ trunk/tests/src/org/hornetq/tests/unit/core/config/impl/ConfigurationImplTest.java 2009-11-03 18:12:24 UTC (rev 8197) @@ -95,6 +95,7 @@ assertEquals(ConfigurationImpl.DEFAULT_SERVER_DUMP_INTERVAL, conf.getServerDumpInterval()); assertEquals(ConfigurationImpl.DEFAULT_MEMORY_WARNING_THRESHOLD, conf.getMemoryWarningThreshold()); assertEquals(ConfigurationImpl.DEFAULT_MEMORY_MEASURE_INTERVAL, conf.getMemoryMeasureInterval()); + assertEquals(ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE, conf.getBackupWindowSize()); } public void testSetGetAttributes() @@ -294,6 +295,10 @@ conf.setTransactionTimeoutScanPeriod(l); assertEquals(l, conf.getTransactionTimeoutScanPeriod()); + i = randomInt(); + conf.setBackupWindowSize(i); + assertEquals(i, conf.getBackupWindowSize()); + s = randomString(); conf.setManagementClusterPassword(s); assertEquals(s, conf.getManagementClusterPassword()); @@ -510,6 +515,10 @@ s = randomString(); conf.setManagementClusterPassword(s); assertEquals(s, conf.getManagementClusterPassword()); + + i = randomInt(); + conf.setBackupWindowSize(i); + assertEquals(i, conf.getBackupWindowSize()); ByteArrayOutputStream baos = new ByteArrayOutputStream(); ObjectOutputStream oos = new ObjectOutputStream(baos); Modified: trunk/tests/src/org/hornetq/tests/unit/core/config/impl/FileConfigurationTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/config/impl/FileConfigurationTest.java 2009-11-03 18:03:04 UTC (rev 8196) +++ trunk/tests/src/org/hornetq/tests/unit/core/config/impl/FileConfigurationTest.java 2009-11-03 18:12:24 UTC (rev 8197) @@ -82,6 +82,7 @@ assertEquals(56546, conf.getJournalMaxAIO()); assertEquals("largemessagesdir", conf.getLargeMessagesDirectory()); assertEquals(95, conf.getMemoryWarningThreshold()); + assertEquals(1024, conf.getBackupWindowSize()); assertEquals(2, conf.getInterceptorClassNames().size()); assertTrue(conf.getInterceptorClassNames().contains("org.hornetq.tests.unit.core.config.impl.TestInterceptor1"));