[hornetq-commits] JBoss hornetq SVN: r7954 - in branches/Branch_Replication_Changes: src/main/org/hornetq/core/client/impl and 25 other directories.

Author: timfox Date: 2009-09-11 06:56:25 -0400 (Fri, 11 Sep 2009) New Revision: 7954 Removed: branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/impl/ReplicationOperationInvokerImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/jmx/ branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicateCreateSessionMessage.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/SessionFailoverCompleteMessage.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/replication/ branches/Branch_Replication_Changes/src/main/org/hornetq/jms/server/management/jmx/ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/ActivationTimeoutTest.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/management/ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/jms/cluster/management/ Modified: branches/Branch_Replication_Changes/src/config/common/schema/hornetq-configuration.xsd branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/config/Configuration.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/config/impl/FileConfiguration.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/HornetQServerControl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/ManagementService.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/QueueControl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/impl/ManagementServiceImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/impl/QueueControlImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/paging/PagingManager.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/postoffice/PostOffice.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/Channel.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/Packet.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/RemotingConnection.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/ChannelImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/PacketDecoder.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/RemotingConnectionImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/CreateSessionResponseMessage.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/PacketImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/Ping.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/ReattachSessionResponseMessage.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/HornetQServer.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/Queue.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/ScheduledDeliveryHandler.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/ServerConsumer.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/ServerSession.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/cluster/impl/Redistributor.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/HornetQPacketHandler.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/QueueImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/ScheduledDeliveryHandlerImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/ServerSessionPacketHandler.java branches/Branch_Replication_Changes/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java branches/Branch_Replication_Changes/src/main/org/hornetq/jms/server/management/impl/JMSManagementServiceImpl.java branches/Branch_Replication_Changes/tests/config/ConfigurationTest-full-config.xml branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/management/HornetQServerControlTest.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/management/QueueControlTest.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/unit/core/config/impl/ConfigurationImplTest.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/unit/core/config/impl/DefaultsFileConfigurationTest.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/unit/core/config/impl/FileConfigurationTest.java branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingImplTest.java Log: initial commit Modified: branches/Branch_Replication_Changes/src/config/common/schema/hornetq-configuration.xsd =================================================================== --- branches/Branch_Replication_Changes/src/config/common/schema/hornetq-configuration.xsd 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/config/common/schema/hornetq-configuration.xsd 2009-09-11 10:56:25 UTC (rev 7954) @@ -63,10 +63,10 @@ </xsd:element> <xsd:element maxOccurs="1" minOccurs="0" ref="remoting-interceptors"> </xsd:element> - <xsd:element maxOccurs="1" minOccurs="0" name="queue-activation-timeout" type="xsd:long"> - </xsd:element> <xsd:element maxOccurs="1" minOccurs="0" name="backup" type="xsd:boolean"> - </xsd:element> + </xsd:element> + <xsd:element maxOccurs="1" minOccurs="0" name="shared-store" type="xsd:boolean"> + </xsd:element> <xsd:element maxOccurs="1" minOccurs="0" name="persist-delivery-count-before-delivery" type="xsd:boolean"> </xsd:element> <xsd:element maxOccurs="1" minOccurs="0" name="backup-connector-ref" type="backup-connectorType"> Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -50,7 +50,6 @@ import org.hornetq.core.remoting.impl.wireformat.SessionCreateConsumerMessage; import org.hornetq.core.remoting.impl.wireformat.SessionDeleteQueueMessage; import org.hornetq.core.remoting.impl.wireformat.SessionExpiredMessage; -import org.hornetq.core.remoting.impl.wireformat.SessionFailoverCompleteMessage; import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryMessage; import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryResponseMessage; import org.hornetq.core.remoting.impl.wireformat.SessionReceiveContinuationMessage; @@ -743,7 +742,7 @@ try { - channel.transferConnection(backupConnection, channel.getID(), null); + channel.transferConnection(backupConnection); backupConnection.syncIDGeneratorSequence(remotingConnection.getIDGeneratorSequence()); @@ -793,8 +792,6 @@ channel.unlock(); } - channel.send(new SessionFailoverCompleteMessage(name)); - return ok; } Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/config/Configuration.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/config/Configuration.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/config/Configuration.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -57,6 +57,10 @@ void setBackup(boolean backup); + boolean isSharedStore(); + + void setSharedStore(boolean sharedStore); + boolean isFileDeploymentEnabled(); void setFileDeploymentEnabled(boolean enable); @@ -69,10 +73,6 @@ void setFileDeployerScanPeriod(long period); - long getQueueActivationTimeout(); - - void setQueueActivationTimeout(long timeout); - int getThreadPoolMaxSize(); void setThreadPoolMaxSize(int maxSize); Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -47,6 +47,8 @@ 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; @@ -54,8 +56,6 @@ public static final long DEFAULT_FILE_DEPLOYER_SCAN_PERIOD = 5000; - public static final long DEFAULT_QUEUE_ACTIVATION_TIMEOUT = 30000; - public static final int DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE = 5; public static final int DEFAULT_THREAD_POOL_MAX_SIZE = -1; @@ -163,7 +163,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; @@ -172,8 +174,6 @@ protected boolean persistDeliveryCountBeforeDelivery = DEFAULT_PERSIST_DELIVERY_COUNT_BEFORE_DELIVERY; - protected long queueActivationTimeout = DEFAULT_QUEUE_ACTIVATION_TIMEOUT; - protected int scheduledThreadPoolMaxSize = DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE; protected int threadPoolMaxSize = DEFAULT_THREAD_POOL_MAX_SIZE; @@ -364,15 +364,15 @@ { this.backup = backup; } - - public long getQueueActivationTimeout() + + public boolean isSharedStore() { - return queueActivationTimeout; + return sharedStore; } - public void setQueueActivationTimeout(final long timeout) + public void setSharedStore(final boolean sharedStore) { - queueActivationTimeout = timeout; + this.sharedStore = sharedStore; } public int getScheduledThreadPoolMaxSize() Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/config/impl/FileConfiguration.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/config/impl/FileConfiguration.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/config/impl/FileConfiguration.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -104,6 +104,8 @@ backup = getBoolean(e, "backup", backup); + sharedStore = getBoolean(e, "shared-store", sharedStore); + //Defaults to true when using FileConfiguration fileDeploymentEnabled = getBoolean(e, "file-deployment-enabled", true); @@ -111,8 +113,6 @@ persistDeliveryCountBeforeDelivery = getBoolean(e, "persist-delivery-count-before-delivery", persistDeliveryCountBeforeDelivery); - queueActivationTimeout = getLong(e, "queue-activation-timeout", queueActivationTimeout, GE_ZERO); - // NOTE! All the defaults come from the super class scheduledThreadPoolMaxSize = getInteger(e, "scheduled-thread-pool-max-size", scheduledThreadPoolMaxSize, GT_ZERO); Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/HornetQServerControl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/HornetQServerControl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/HornetQServerControl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -25,7 +25,7 @@ { // Attributes ---------------------------------------------------- - public String getBackupConnectorName(); + String getBackupConnectorName(); String getVersion(); @@ -77,18 +77,18 @@ void setMessageCounterSamplePeriod(long newPeriod) throws Exception; - public boolean isBackup(); + boolean isBackup(); + + boolean isSharedStore(); int getAIOBufferSize(); int getAIOBufferTimeout(); - public String getPagingDirectory(); + String getPagingDirectory(); boolean isPersistDeliveryCountBeforeDelivery(); - long getQueueActivationTimeout(); - long getConnectionTTLOverride(); String getManagementAddress(); @@ -156,7 +156,7 @@ void resetAllMessageCounterHistories() throws Exception; @Operation(desc = "List all the prepared transaction, sorted by date, oldest first") - public String[] listPreparedTransactions() throws Exception; + String[] listPreparedTransactions() throws Exception; @Operation(desc = "Commit a prepared transaction") boolean commitPreparedTransaction(@Parameter(desc = "the Base64 representation of a transaction", name = "transactionAsBase64") String transactionAsBase64) throws Exception; Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/ManagementService.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/ManagementService.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/ManagementService.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -67,8 +67,6 @@ SimpleString getManagementNotificationAddress(); - ReplicationOperationInvoker getReplicationOperationInvoker(); - // Resource Registration HornetQServerControlImpl registerServer(PostOffice postOffice, Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/QueueControl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/QueueControl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/QueueControl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -58,8 +58,6 @@ void setDeadLetterAddress(@Parameter(name = "deadLetterAddress", desc = "Dead-letter address of the queue") String deadLetterAddress) throws Exception; - boolean isBackup(); - // Operations ---------------------------------------------------- @Operation(desc = "List the messages scheduled for delivery", impact = INFO) Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -127,6 +127,11 @@ { return configuration.isBackup(); } + + public boolean isSharedStore() + { + return configuration.isSharedStore(); + } public String getBackupConnectorName() { @@ -602,11 +607,6 @@ return configuration.getMessageExpiryThreadPriority(); } - public long getQueueActivationTimeout() - { - return configuration.getQueueActivationTimeout(); - } - public long getTransactionTimeout() { return configuration.getTransactionTimeout(); Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/impl/ManagementServiceImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/impl/ManagementServiceImpl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/impl/ManagementServiceImpl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -50,11 +50,7 @@ import org.hornetq.core.management.Notification; import org.hornetq.core.management.NotificationListener; import org.hornetq.core.management.ObjectNames; -import org.hornetq.core.management.ReplicationOperationInvoker; import org.hornetq.core.management.ResourceNames; -import org.hornetq.core.management.jmx.impl.ReplicationAwareAddressControlWrapper; -import org.hornetq.core.management.jmx.impl.ReplicationAwareHornetQServerControlWrapper; -import org.hornetq.core.management.jmx.impl.ReplicationAwareQueueControlWrapper; import org.hornetq.core.messagecounter.MessageCounter; import org.hornetq.core.messagecounter.MessageCounterManager; import org.hornetq.core.messagecounter.impl.MessageCounterManagerImpl; @@ -130,8 +126,6 @@ private final Set<NotificationListener> listeners = new org.hornetq.utils.ConcurrentHashSet<NotificationListener>(); - private ReplicationOperationInvoker replicationInvoker; - // Static -------------------------------------------------------- private static void checkDefaultManagementClusterCredentials(String user, String password) @@ -163,12 +157,6 @@ registry = new HashMap<String, Object>(); broadcaster = new NotificationBroadcasterSupport(); notificationsEnabled = true; - - replicationInvoker = new ReplicationOperationInvokerImpl(managementClusterUser, - managementClusterPassword, - managementAddress, - managementRequestTimeout, - managementConnectorID); } // Public -------------------------------------------------------- @@ -210,8 +198,7 @@ messageCounterManager, broadcaster); ObjectName objectName = ObjectNames.getHornetQServerObjectName(); - registerInJMX(objectName, new ReplicationAwareHornetQServerControlWrapper(messagingServerControl, - replicationInvoker)); + registerInJMX(objectName, messagingServerControl); registerInRegistry(ResourceNames.CORE_SERVER, messagingServerControl); return messagingServerControl; @@ -229,7 +216,7 @@ ObjectName objectName = ObjectNames.getAddressObjectName(address); AddressControlImpl addressControl = new AddressControlImpl(address, postOffice, securityRepository); - registerInJMX(objectName, new ReplicationAwareAddressControlWrapper(addressControl, replicationInvoker)); + registerInJMX(objectName, addressControl); registerInRegistry(ResourceNames.CORE_ADDRESS + address, addressControl); @@ -264,7 +251,7 @@ queueControl.setMessageCounter(counter); messageCounterManager.registerMessageCounter(queue.getName().toString(), counter); ObjectName objectName = ObjectNames.getQueueObjectName(address, queue.getName()); - registerInJMX(objectName, new ReplicationAwareQueueControlWrapper(queueControl, replicationInvoker)); + registerInJMX(objectName, queueControl); registerInRegistry(ResourceNames.CORE_QUEUE + queue.getName(), queueControl); if (log.isDebugEnabled()) @@ -572,11 +559,6 @@ return managementClusterPassword; } - public ReplicationOperationInvoker getReplicationOperationInvoker() - { - return replicationInvoker; - } - // HornetQComponent implementation ----------------------------- public void start() throws Exception @@ -639,8 +621,6 @@ registeredNames.clear(); - replicationInvoker.stop(); - started = false; } Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/impl/QueueControlImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/impl/QueueControlImpl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/impl/QueueControlImpl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -133,11 +133,6 @@ return queue.isTemporary(); } - public boolean isBackup() - { - return queue.isBackup(); - } - public int getMessageCount() { return queue.getMessageCount(); Deleted: branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/impl/ReplicationOperationInvokerImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/impl/ReplicationOperationInvokerImpl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/management/impl/ReplicationOperationInvokerImpl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -1,134 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.core.management.impl; - -import java.util.HashMap; - -import org.hornetq.core.client.ClientMessage; -import org.hornetq.core.client.ClientRequestor; -import org.hornetq.core.client.ClientSession; -import org.hornetq.core.client.ClientSessionFactory; -import org.hornetq.core.client.impl.ClientSessionFactoryImpl; -import org.hornetq.core.client.management.impl.ManagementHelper; -import org.hornetq.core.config.TransportConfiguration; -import org.hornetq.core.logging.Logger; -import org.hornetq.core.management.ReplicationOperationInvoker; -import org.hornetq.core.remoting.impl.invm.InVMConnectorFactory; -import org.hornetq.core.remoting.impl.invm.TransportConstants; -import org.hornetq.utils.SimpleString; - -/** - * A ReplicationOperationInvoker - * - * @author <a href=&quot;jmesnil(a)redhat.com&quot;&gt;Jeff Mesnil</a> - * - */ -public class ReplicationOperationInvokerImpl implements ReplicationOperationInvoker -{ - - // Constants ----------------------------------------------------- - - private static final Logger log = Logger.getLogger(ReplicationOperationInvokerImpl.class); - - // Attributes ---------------------------------------------------- - - private final long timeout; - - private final String clusterUser; - - private final String clusterPassword; - - private final SimpleString managementAddress; - - private ClientSession clientSession; - - private ClientRequestor requestor; - - private final int managementConnectorID; - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - public ReplicationOperationInvokerImpl(final String clusterUser, - final String clusterPassword, - final SimpleString managementAddress, - final long managementRequestTimeout, - final int managementConnectorID) - { - this.timeout = managementRequestTimeout; - this.clusterUser = clusterUser; - this.clusterPassword = clusterPassword; - this.managementAddress = managementAddress; - this.managementConnectorID = managementConnectorID; - } - - // Public -------------------------------------------------------- - - public synchronized Object invoke(final String resourceName, final String operationName, final Object... parameters) throws Exception - { - if (clientSession == null) - { - ClientSessionFactory sf = new ClientSessionFactoryImpl(new TransportConfiguration(InVMConnectorFactory.class.getName(), - new HashMap<String, Object>() - { - { - put(TransportConstants.SERVER_ID_PROP_NAME, - managementConnectorID); - } - })); - - clientSession = sf.createSession(clusterUser, clusterPassword, false, true, true, false, 1); - requestor = new ClientRequestor(clientSession, managementAddress); - clientSession.start(); - } - ClientMessage mngmntMessage = clientSession.createClientMessage(false); - ManagementHelper.putOperationInvocation(mngmntMessage, resourceName, operationName, parameters); - ClientMessage reply = requestor.request(mngmntMessage, timeout); - - if (reply == null) - { - throw new Exception("did not receive reply for message " + mngmntMessage); - } - reply.acknowledge(); - if (ManagementHelper.hasOperationSucceeded(reply)) - { - return ManagementHelper.getResult(reply); - } - else - { - throw new Exception((String)ManagementHelper.getResult(reply)); - } - } - - public void stop() throws Exception - { - if (clientSession != null) - { - clientSession.close(); - } - } - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - // Private ------------------------------------------------------- - - // Inner classes ------------------------------------------------- - -} Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/paging/PagingManager.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/paging/PagingManager.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/paging/PagingManager.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -47,10 +47,6 @@ */ public interface PagingManager extends HornetQComponent { - void activate(); - - boolean isBackup(); - /** To return the PageStore associated with the address */ PagingStore getPageStore(SimpleString address) throws Exception; Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -47,7 +47,7 @@ private volatile boolean started = false; - private volatile boolean backup; + // private volatile boolean backup; private final AtomicLong totalMemoryBytes = new AtomicLong(0); @@ -74,14 +74,12 @@ public PagingManagerImpl(final PagingStoreFactory pagingSPI, final StorageManager storageManager, final HierarchicalRepository<AddressSettings> addressSettingsRepository, - final boolean syncNonTransactional, - final boolean backup) + final boolean syncNonTransactional) { pagingStoreFactory = pagingSPI; this.addressSettingsRepository = addressSettingsRepository; this.storageManager = storageManager; this.syncNonTransactional = syncNonTransactional; - this.backup = backup; } // Public @@ -95,25 +93,7 @@ Set<SimpleString> names = stores.keySet(); return (SimpleString[])names.toArray(new SimpleString[names.size()]); } - - public void activate() - { - backup = false; - for (PagingStore store : stores.values()) - { - if (store.isPaging()) - { - store.startDepaging(); - } - } - } - - public boolean isBackup() - { - return backup; - } - /* (non-Javadoc) * @see org.hornetq.core.paging.PagingManager#reloadStores() */ @@ -270,14 +250,11 @@ } synchronized (this) { - if (!isBackup()) + for (PagingStore store : stores.values()) { - for (PagingStore store : stores.values()) + if (store.isPaging()) { - if (store.isPaging()) - { - store.startDepaging(); - } + store.startDepaging(); } } } Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -401,11 +401,6 @@ return false; } - if (pagingManager.isBackup()) - { - return false; - } - currentPageLock.readLock().lock(); try { Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/postoffice/PostOffice.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/postoffice/PostOffice.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/postoffice/PostOffice.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -13,8 +13,6 @@ package org.hornetq.core.postoffice; -import java.util.List; - import org.hornetq.core.paging.PagingManager; import org.hornetq.core.server.HornetQComponent; import org.hornetq.core.server.Queue; @@ -58,8 +56,6 @@ boolean redistribute(ServerMessage message, final Queue originatingQueue, Transaction tx) throws Exception; - List<Queue> activate(); - PagingManager getPagingManager(); DuplicateIDCache getDuplicateIDCache(SimpleString address); Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -75,8 +75,6 @@ public static final SimpleString HDR_RESET_QUEUE_DATA = new SimpleString("_HQ_RESET_QUEUE_DATA"); - private HornetQServer server; - private final AddressManager addressManager; private final QueueFactory queueFactory; @@ -87,7 +85,7 @@ private volatile boolean started; - private volatile boolean backup; + // private volatile boolean backup; private final ManagementService managementService; @@ -123,23 +121,19 @@ private final HierarchicalRepository<AddressSettings> addressSettingsRepository; - public PostOfficeImpl(final HornetQServer server, - final StorageManager storageManager, + public PostOfficeImpl(final StorageManager storageManager, final PagingManager pagingManager, final QueueFactory bindableFactory, final ManagementService managementService, final long reaperPeriod, final int reaperPriority, - final boolean enableWildCardRouting, - final boolean backup, + final boolean enableWildCardRouting, final int idCacheSize, final boolean persistIDCache, final ExecutorFactory orderedExecutorFactory, HierarchicalRepository<AddressSettings> addressSettingsRepository) { - this.server = server; - this.storageManager = storageManager; this.queueFactory = bindableFactory; @@ -161,8 +155,6 @@ addressManager = new SimpleAddressManager(); } - this.backup = backup; - this.idCacheSize = idCacheSize; this.persistIDCache = persistIDCache; @@ -190,10 +182,7 @@ // This is to avoid thread leakages where the Reaper would run beyong the life cycle of the PostOffice started = true; - if (!backup) - { - startExpiryScanner(); - } + startExpiryScanner(); } public synchronized void stop() throws Exception @@ -356,8 +345,7 @@ if (redistributionDelay != -1) { queue.addRedistributor(redistributionDelay, - redistributorExecutorFactory.getExecutor(), - server.getReplicatingChannel()); + redistributorExecutorFactory.getExecutor()); } } } @@ -428,8 +416,7 @@ if (redistributionDelay != -1) { queue.addRedistributor(redistributionDelay, - redistributorExecutorFactory.getExecutor(), - server.getReplicatingChannel()); + redistributorExecutorFactory.getExecutor()); } } } @@ -463,11 +450,6 @@ { Queue queue = (Queue)binding.getBindable(); - if (backup) - { - queue.setBackup(); - } - managementService.registerQueue(queue, binding.getAddress(), storageManager); if (!existed) @@ -683,37 +665,6 @@ return pagingManager; } - public List<Queue> activate() - { - - backup = false; - - pagingManager.activate(); - - Map<SimpleString, Binding> nameMap = addressManager.getBindings(); - - List<Queue> queues = new ArrayList<Queue>(); - - for (Binding binding : nameMap.values()) - { - if (binding.getType() == BindingType.LOCAL_QUEUE) - { - Queue queue = (Queue)binding.getBindable(); - - boolean activated = queue.activate(); - - if (!activated) - { - queues.add(queue); - } - } - } - - startExpiryScanner(); - - return queues; - } - public DuplicateIDCache getDuplicateIDCache(final SimpleString address) { DuplicateIDCache cache = duplicateIDCaches.get(address); Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/Channel.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/Channel.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/Channel.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -31,15 +31,13 @@ Packet sendBlocking(Packet packet) throws HornetQException; - void replicatePacket(Packet packet, long replicatedChannelID, Runnable action); - void setHandler(ChannelHandler handler); ChannelHandler getHandler(); void close(); - void transferConnection(RemotingConnection newConnection, final long newID, final Channel replicatingChannel); + void transferConnection(RemotingConnection newConnection); void replayCommands(int lastReceivedCommandID, final long newID); @@ -55,8 +53,6 @@ RemotingConnection getConnection(); - void executeOutstandingDelayedResults(); - void confirm(Packet packet); void setCommandConfirmationHandler(CommandConfirmationHandler handler); @@ -64,6 +60,4 @@ void flushConfirmations(); void handlePacket(Packet packet); - - void waitForAllReplicationResponse(); } Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/Packet.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/Packet.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/Packet.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -41,8 +41,6 @@ int getRequiredBufferSize(); boolean isRequiresConfirmations(); - - boolean isWriteAlways(); } Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/RemotingConnection.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/RemotingConnection.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/RemotingConnection.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -62,14 +62,8 @@ long getIDGeneratorSequence(); - void activate(); - - void freeze(); - Connection getTransportConnection(); - boolean isActive(); - boolean isClient(); boolean isDestroyed(); Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/ChannelImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/ChannelImpl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/ChannelImpl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -14,11 +14,7 @@ package org.hornetq.core.remoting.impl; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.PACKETS_CONFIRMED; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_RESPONSE; -import java.util.ArrayList; -import java.util.List; -import java.util.Queue; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; @@ -73,12 +69,8 @@ private final Object sendBlockingLock = new Object(); - private final Object replicationLock = new Object(); - private boolean failingOver; - private final Queue<Runnable> responseActions = new ConcurrentLinkedQueue<Runnable>(); - private final int windowSize; private final int confWindowSize; @@ -89,10 +81,6 @@ private CommandConfirmationHandler commandConfirmationHandler; - private int responseActionCount; - - private boolean playedResponsesOnFailure; - public ChannelImpl(final RemotingConnection connection, final long id, final int windowSize, final boolean block) { this.connection = connection; @@ -200,10 +188,7 @@ resendCache.add(packet); } - if (connection.isActive() || packet.isWriteAlways()) - { - connection.getTransportConnection().write(buffer, flush); - } + connection.getTransportConnection().write(buffer, flush); } finally { @@ -333,99 +318,11 @@ } } - // Must be synchronized since can be called by incoming session commands but also by deliveries - // Also needs to be synchronized with respect to replicatingChannelDead - public void replicatePacket(final Packet packet, final long replicatedChannelID, final Runnable action) - { - packet.setChannelID(replicatedChannelID); - - boolean runItNow = false; - - synchronized (replicationLock) - { - if (playedResponsesOnFailure && action != null) - { - // Already replicating channel failed, so just play the action now - - runItNow = true; - } - else - { - if (action != null) - { - responseActions.add(action); - - responseActionCount++; - } - - final HornetQBuffer buffer = connection.getTransportConnection() - .createBuffer(packet.getRequiredBufferSize()); - - packet.encode(buffer); - - connection.getTransportConnection().write(buffer); - } - } - - // Execute outside lock - - if (runItNow) - { - action.run(); - } - } - public void setCommandConfirmationHandler(final CommandConfirmationHandler handler) { this.commandConfirmationHandler = handler; } - public void executeOutstandingDelayedResults() - { - // Execute on different thread to avoid deadlock - - new Thread() - { - public void run() - { - doExecuteOutstandingDelayedResults(); - } - }.start(); - } - - private void doExecuteOutstandingDelayedResults() - { - List<Runnable> toRun = new ArrayList<Runnable>(); - - synchronized (replicationLock) - { - playedResponsesOnFailure = true; - - responseActionCount = 0; - } - - while (true) - { - // Execute all the response actions now - - Runnable action = responseActions.poll(); - - if (action != null) - { - toRun.add(action); - } - else - { - break; - } - } - - for (Runnable action : toRun) - { - action.run(); - } - } - public void setHandler(final ChannelHandler handler) { this.handler = handler; @@ -451,9 +348,7 @@ closed = true; } - public void transferConnection(final RemotingConnection newConnection, - final long newChannelID, - final Channel replicatingChannel) + public void transferConnection(final RemotingConnection newConnection) { // Needs to synchronize on the connection to make sure no packets from // the old connection get processed after transfer has occurred @@ -461,25 +356,13 @@ { connection.removeChannel(id); - if (replicatingChannel != null) - { - // If we're reconnecting to a live node which is replicated then there will be a replicating channel - // too. We need to then make sure that all replication responses come back since packets aren't - // considered confirmed until response comes back and is processed. Otherwise responses to previous - // message sends could come back after reconnection resulting in clients resending same message - // since it wasn't confirmed yet. - replicatingChannel.waitForAllReplicationResponse(); - } - // And switch it final RemotingConnectionImpl rnewConnection = (RemotingConnectionImpl)newConnection; - rnewConnection.putChannel(newChannelID, this); + rnewConnection.putChannel(id, this); connection = rnewConnection; - - this.id = newChannelID; } } @@ -522,7 +405,7 @@ public void flushConfirmations() { - if (receivedBytes != 0 && connection.isActive()) + if (receivedBytes != 0) { receivedBytes = 0; @@ -546,14 +429,11 @@ { receivedBytes = 0; - if (connection.isActive()) - { - final Packet confirmed = new PacketsConfirmedMessage(lastReceivedCommandID); + final Packet confirmed = new PacketsConfirmedMessage(lastReceivedCommandID); - confirmed.setChannelID(id); + confirmed.setChannelID(id); - doWrite(confirmed); - } + doWrite(confirmed); } } } @@ -576,12 +456,6 @@ return; } - else if (packet.getType() == REPLICATION_RESPONSE) - { - replicateResponseReceived(); - - return; - } else { if (packet.isResponse()) @@ -606,98 +480,8 @@ handler.handlePacket(packet); } } - - replicateComplete(); } - public void waitForAllReplicationResponse() - { - synchronized (replicationLock) - { - long toWait = 10000; // TODO don't hardcode timeout - - long start = System.currentTimeMillis(); - - while (responseActionCount > 0 && toWait > 0) - { - try - { - replicationLock.wait(); - } - catch (InterruptedException e) - { - } - - long now = System.currentTimeMillis(); - - toWait -= now - start; - - start = now; - } - - if (toWait <= 0) - { - log.warn("Timed out waiting for replication responses to return"); - } - } - } - - private void replicateComplete() - { - if (!connection.isActive() && id != 0) - { - // We're on backup and not ping channel so send back a replication response - - Packet packet = new PacketImpl(REPLICATION_RESPONSE); - - packet.setChannelID(2); - - doWrite(packet); - } - } - - // This will never get called concurrently by more than one thread - - // TODO it's not ideal synchronizing this since it forms a contention point with replication - // but we need to do this to protect it w.r.t. the check on replicatingChannel - - private void replicateResponseReceived() - { - Runnable result = null; - - synchronized (replicationLock) - { - if (playedResponsesOnFailure) - { - return; - } - - result = responseActions.poll(); - - if (result == null) - { - throw new IllegalStateException("Cannot find response action"); - } - } - - // Must execute outside of lock - if (result != null) - { - result.run(); - - // TODO - we can optimise this not to lock every time - only if waiting for all replications to return - synchronized (replicationLock) - { - responseActionCount--; - - if (responseActionCount == 0) - { - replicationLock.notify(); - } - } - } - } - private void doWrite(final Packet packet) { final HornetQBuffer buffer = connection.getTransportConnection().createBuffer(packet.getRequiredBufferSize()); Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/PacketDecoder.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/PacketDecoder.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/PacketDecoder.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -24,15 +24,6 @@ import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.PING; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REATTACH_SESSION; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REATTACH_SESSION_RESP; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATE_ACKNOWLEDGE; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATE_ADD_REMOTE_CONSUMER; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATE_ADD_REMOTE_QUEUE_BINDING; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATE_CREATESESSION; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATE_REDISTRIBUTION; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATE_REMOVE_REMOTE_CONSUMER; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATE_REMOVE_REMOTE_QUEUE_BINDING; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATE_STARTUP_INFO; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_RESPONSE; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_ACKNOWLEDGE; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_BINDINGQUERY; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_BINDINGQUERY_RESP; @@ -41,13 +32,11 @@ import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_CONSUMER_CLOSE; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_CREATECONSUMER; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_EXPIRED; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_FAILOVER_COMPLETE; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_FLOWTOKEN; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_QUEUEQUERY; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_QUEUEQUERY_RESP; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_RECEIVE_CONTINUATION; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_RECEIVE_MSG; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_REPLICATE_DELIVERY; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_ROLLBACK; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_SEND; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_SEND_CONTINUATION; @@ -82,7 +71,6 @@ import org.hornetq.core.remoting.impl.wireformat.Ping; import org.hornetq.core.remoting.impl.wireformat.ReattachSessionMessage; import org.hornetq.core.remoting.impl.wireformat.ReattachSessionResponseMessage; -import org.hornetq.core.remoting.impl.wireformat.ReplicateCreateSessionMessage; import org.hornetq.core.remoting.impl.wireformat.RollbackMessage; import org.hornetq.core.remoting.impl.wireformat.SessionAcknowledgeMessage; import org.hornetq.core.remoting.impl.wireformat.SessionBindingQueryMessage; @@ -93,7 +81,6 @@ import org.hornetq.core.remoting.impl.wireformat.SessionCreateConsumerMessage; import org.hornetq.core.remoting.impl.wireformat.SessionDeleteQueueMessage; import org.hornetq.core.remoting.impl.wireformat.SessionExpiredMessage; -import org.hornetq.core.remoting.impl.wireformat.SessionFailoverCompleteMessage; import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryMessage; import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryResponseMessage; import org.hornetq.core.remoting.impl.wireformat.SessionReceiveContinuationMessage; @@ -114,14 +101,6 @@ import org.hornetq.core.remoting.impl.wireformat.SessionXASetTimeoutMessage; import org.hornetq.core.remoting.impl.wireformat.SessionXASetTimeoutResponseMessage; import org.hornetq.core.remoting.impl.wireformat.SessionXAStartMessage; -import org.hornetq.core.remoting.impl.wireformat.replication.ReplicateAcknowledgeMessage; -import org.hornetq.core.remoting.impl.wireformat.replication.ReplicateRedistributionMessage; -import org.hornetq.core.remoting.impl.wireformat.replication.ReplicateRemoteBindingAddedMessage; -import org.hornetq.core.remoting.impl.wireformat.replication.ReplicateRemoteBindingRemovedMessage; -import org.hornetq.core.remoting.impl.wireformat.replication.ReplicateRemoteConsumerAddedMessage; -import org.hornetq.core.remoting.impl.wireformat.replication.ReplicateRemoteConsumerRemovedMessage; -import org.hornetq.core.remoting.impl.wireformat.replication.ReplicateStartupInfoMessage; -import org.hornetq.core.remoting.impl.wireformat.replication.SessionReplicateDeliveryMessage; import org.hornetq.core.remoting.spi.HornetQBuffer; /** @@ -161,21 +140,11 @@ packet = new PacketsConfirmedMessage(); break; } - case REPLICATION_RESPONSE: - { - packet = new PacketImpl(REPLICATION_RESPONSE); - break; - } case CREATESESSION: { packet = new CreateSessionMessage(); break; } - case REPLICATE_CREATESESSION: - { - packet = new ReplicateCreateSessionMessage(); - break; - } case CREATESESSION_RESP: { packet = new CreateSessionResponseMessage(); @@ -191,11 +160,6 @@ packet = new ReattachSessionResponseMessage(); break; } - case SESS_FAILOVER_COMPLETE: - { - packet = new SessionFailoverCompleteMessage(); - break; - } case SESS_CLOSE: { packet = new SessionCloseMessage(); @@ -385,47 +349,7 @@ { packet = new SessionSendContinuationMessage(); break; - } - case REPLICATE_ADD_REMOTE_QUEUE_BINDING: - { - packet = new ReplicateRemoteBindingAddedMessage(); - break; - } - case REPLICATE_REMOVE_REMOTE_QUEUE_BINDING: - { - packet = new ReplicateRemoteBindingRemovedMessage(); - break; - } - case REPLICATE_ADD_REMOTE_CONSUMER: - { - packet = new ReplicateRemoteConsumerAddedMessage(); - break; - } - case REPLICATE_REMOVE_REMOTE_CONSUMER: - { - packet = new ReplicateRemoteConsumerRemovedMessage(); - break; - } - case SESS_REPLICATE_DELIVERY: - { - packet = new SessionReplicateDeliveryMessage(); - break; - } - case REPLICATE_STARTUP_INFO: - { - packet = new ReplicateStartupInfoMessage(); - break; - } - case REPLICATE_ACKNOWLEDGE: - { - packet = new ReplicateAcknowledgeMessage(); - break; - } - case REPLICATE_REDISTRIBUTION: - { - packet = new ReplicateRedistributionMessage(); - break; - } + } default: { throw new IllegalArgumentException("Invalid type: " + packetType); Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/RemotingConnectionImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/RemotingConnectionImpl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/RemotingConnectionImpl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -65,7 +65,7 @@ private volatile boolean destroyed; - private volatile boolean active; + // private volatile boolean active; private final boolean client; @@ -79,7 +79,7 @@ private final Object transferLock = new Object(); - private boolean frozen; + // private boolean frozen; private final Object failLock = new Object(); @@ -99,7 +99,7 @@ final long blockingCallTimeout, final List<Interceptor> interceptors) { - this(transportConnection, blockingCallTimeout, interceptors, true, true, null); + this(transportConnection, blockingCallTimeout, interceptors, true, null); } /* @@ -107,17 +107,15 @@ */ public RemotingConnectionImpl(final Connection transportConnection, final List<Interceptor> interceptors, - final boolean active, final Executor executor) { - this(transportConnection, -1, interceptors, active, false, executor); + this(transportConnection, -1, interceptors, false, executor); } private RemotingConnectionImpl(final Connection transportConnection, final long blockingCallTimeout, final List<Interceptor> interceptors, - final boolean active, final boolean client, final Executor executor) @@ -128,8 +126,6 @@ this.interceptors = interceptors; - this.active = active; - this.client = client; this.executor = executor; @@ -306,11 +302,6 @@ return transferLock; } - public boolean isActive() - { - return active; - } - public boolean isClient() { return client; @@ -366,55 +357,37 @@ { synchronized (transferLock) { - if (!frozen) + if (interceptors != null) { - if (interceptors != null) + for (final Interceptor interceptor : interceptors) { - for (final Interceptor interceptor : interceptors) + try { - try + boolean callNext = interceptor.intercept(packet, this); + + if (!callNext) { - boolean callNext = interceptor.intercept(packet, this); + // abort - if (!callNext) - { - // abort - - return; - } + return; } - catch (final Throwable e) - { - log.warn("Failure in calling interceptor: " + interceptor, e); - } } + catch (final Throwable e) + { + log.warn("Failure in calling interceptor: " + interceptor, e); + } } + } - final Channel channel = channels.get(packet.getChannelID()); + final Channel channel = channels.get(packet.getChannelID()); - if (channel != null) - { - channel.handlePacket(packet); - } + if (channel != null) + { + channel.handlePacket(packet); } } } - public void activate() - { - active = true; - } - - public void freeze() - { - // Prevent any more packets being handled on this connection - - synchronized (transferLock) - { - frozen = true; - } - } - // Package protected // ---------------------------------------------------------------------------- Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/CreateSessionResponseMessage.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/CreateSessionResponseMessage.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/CreateSessionResponseMessage.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -29,14 +29,18 @@ // Attributes ---------------------------------------------------- private int serverVersion; + + private boolean created; // Static -------------------------------------------------------- // Constructors -------------------------------------------------- - public CreateSessionResponseMessage(final int serverVersion) + public CreateSessionResponseMessage(final boolean created, final int serverVersion) { super(CREATESESSION_RESP); + + this.created = created; this.serverVersion = serverVersion; } @@ -48,6 +52,11 @@ // Public -------------------------------------------------------- + public boolean isCreated() + { + return created; + } + @Override public boolean isResponse() { @@ -62,18 +71,20 @@ @Override public void encodeBody(final HornetQBuffer buffer) { + buffer.writeBoolean(created); buffer.writeInt(serverVersion); } @Override public void decodeBody(final HornetQBuffer buffer) { + created = buffer.readBoolean(); serverVersion = buffer.readInt(); } public int getRequiredBufferSize() { - return BASIC_PACKET_SIZE + DataConstants.SIZE_INT; + return BASIC_PACKET_SIZE + DataConstants.SIZE_BOOLEAN + DataConstants.SIZE_INT; } @Override @@ -86,7 +97,7 @@ CreateSessionResponseMessage r = (CreateSessionResponseMessage)other; - boolean matches = super.equals(other) && serverVersion == r.serverVersion; + boolean matches = super.equals(other) && serverVersion == r.serverVersion && created == r.created; return matches; } Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/PacketImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/PacketImpl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/PacketImpl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -52,8 +52,6 @@ public static final byte PACKETS_CONFIRMED = 22; - public static final byte REPLICATION_RESPONSE = 23; - // Server public static final byte CREATESESSION = 30; @@ -63,11 +61,9 @@ public static final byte REATTACH_SESSION_RESP = 33; - public static final byte REPLICATE_CREATESESSION = 34; - - public static final byte CREATE_QUEUE = 35; + public static final byte CREATE_QUEUE = 34; - public static final byte DELETE_QUEUE = 36; + public static final byte DELETE_QUEUE = 35; // Session @@ -141,26 +137,6 @@ public static final byte SESS_RECEIVE_CONTINUATION = 76; - public static final byte SESS_FAILOVER_COMPLETE = 77; - - //Replication - - public static final byte SESS_REPLICATE_DELIVERY = 90; - - public static final byte REPLICATE_ADD_REMOTE_QUEUE_BINDING = 91; - - public static final byte REPLICATE_REMOVE_REMOTE_QUEUE_BINDING = 92; - - public static final byte REPLICATE_ADD_REMOTE_CONSUMER = 93; - - public static final byte REPLICATE_REMOVE_REMOTE_CONSUMER = 94; - - public static final byte REPLICATE_ACKNOWLEDGE = 95; - - public static final byte REPLICATE_STARTUP_INFO = 96; - - public static final byte REPLICATE_REDISTRIBUTION = 97; - // Static -------------------------------------------------------- public PacketImpl(final byte type) @@ -242,11 +218,6 @@ return true; } - public boolean isWriteAlways() - { - return false; - } - @Override public String toString() { Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/Ping.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/Ping.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/Ping.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -49,11 +49,6 @@ // Public -------------------------------------------------------- - public boolean isWriteAlways() - { - return true; - } - public long getConnectionTTL() { return connectionTTL; Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/ReattachSessionResponseMessage.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/ReattachSessionResponseMessage.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/ReattachSessionResponseMessage.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -30,21 +30,20 @@ // Attributes ---------------------------------------------------- private int lastReceivedCommandID; + + private boolean sessionFound; - //Is this flag really necessary - try removing it - private boolean removed; - // Static -------------------------------------------------------- // Constructors -------------------------------------------------- - public ReattachSessionResponseMessage(final int lastReceivedCommandID, final boolean removed) + public ReattachSessionResponseMessage(final int lastReceivedCommandID, final boolean sessionFound) { super(REATTACH_SESSION_RESP); this.lastReceivedCommandID = lastReceivedCommandID; - this.removed = removed; + this.sessionFound = sessionFound; } public ReattachSessionResponseMessage() @@ -59,9 +58,9 @@ return lastReceivedCommandID; } - public boolean isRemoved() + public boolean isSessionFound() { - return removed; + return sessionFound; } public int getRequiredBufferSize() @@ -73,13 +72,13 @@ public void encodeBody(final HornetQBuffer buffer) { buffer.writeInt(lastReceivedCommandID); - buffer.writeBoolean(removed); + buffer.writeBoolean(sessionFound); } public void decodeBody(final HornetQBuffer buffer) { lastReceivedCommandID = buffer.readInt(); - removed = buffer.readBoolean(); + sessionFound = buffer.readBoolean(); } public boolean isResponse() Deleted: branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicateCreateSessionMessage.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicateCreateSessionMessage.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicateCreateSessionMessage.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -1,255 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.core.remoting.impl.wireformat; - -import org.hornetq.core.remoting.spi.HornetQBuffer; -import org.hornetq.utils.DataConstants; - -/** - * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> - * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a>. - * - * @version <tt>$Revision$</tt> - */ -public class ReplicateCreateSessionMessage extends PacketImpl -{ - // Constants ----------------------------------------------------- - - // Attributes ---------------------------------------------------- - - private String name; - - private long replicatedSessionChannelID; - - private long originalSessionChannelID; - - private int version; - - private String username; - - private String password; - - private int minLargeMessageSize; - - private boolean xa; - - private boolean autoCommitSends; - - private boolean autoCommitAcks; - - private boolean preAcknowledge; - - private int windowSize; - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - public ReplicateCreateSessionMessage(final String name, - final long replicatedSessionChannelID, - final long originalSessionChannelID, - final int version, - final String username, - final String password, - final int minLargeMessageSize, - final boolean xa, - final boolean autoCommitSends, - final boolean autoCommitAcks, - final boolean preAcknowledge, - final int windowSize) - { - super(REPLICATE_CREATESESSION); - - this.name = name; - - this.replicatedSessionChannelID = replicatedSessionChannelID; - - this.originalSessionChannelID = originalSessionChannelID; - - this.version = version; - - this.username = username; - - this.password = password; - - this.minLargeMessageSize = minLargeMessageSize; - - this.xa = xa; - - this.autoCommitSends = autoCommitSends; - - this.autoCommitAcks = autoCommitAcks; - - this.preAcknowledge = preAcknowledge; - - this.windowSize = windowSize; - } - - public ReplicateCreateSessionMessage() - { - super(REPLICATE_CREATESESSION); - } - - // Public -------------------------------------------------------- - - public String getName() - { - return name; - } - - public long getReplicatedSessionChannelID() - { - return replicatedSessionChannelID; - } - - public long getOriginalSessionChannelID() - { - return originalSessionChannelID; - } - - public int getVersion() - { - return version; - } - - public String getUsername() - { - return username; - } - - public String getPassword() - { - return password; - } - - public int getMinLargeMessageSize() - { - return minLargeMessageSize; - } - - public boolean isXA() - { - return xa; - } - - public boolean isAutoCommitSends() - { - return autoCommitSends; - } - - public boolean isAutoCommitAcks() - { - return autoCommitAcks; - } - - public boolean isPreAcknowledge() - { - return preAcknowledge; - } - - public int getWindowSize() - { - return windowSize; - } - - public int getRequiredBufferSize() - { - return BASIC_PACKET_SIZE + - stringEncodeSize(name) + // buffer.writeString(name); - DataConstants.SIZE_LONG + // buffer.writeLong(originalSessionChannelID); - DataConstants.SIZE_LONG + // buffer.writeLong(replicatedSessionChannelID); - DataConstants.SIZE_INT + // buffer.writeInt(version); - nullableStringEncodeSize(username) + // buffer.writeNullableString(username); - nullableStringEncodeSize(password) + // buffer.writeNullableString(password); - DataConstants.SIZE_INT + // buffer.writeInt(minLargeMessageSize); - DataConstants.SIZE_BOOLEAN + // buffer.writeBoolean(xa); - DataConstants.SIZE_BOOLEAN + // buffer.writeBoolean(autoCommitSends); - DataConstants.SIZE_BOOLEAN + // buffer.writeBoolean(autoCommitAcks); - DataConstants.SIZE_INT + // buffer.writeInt(windowSize); - DataConstants.SIZE_BOOLEAN; // buffer.writeBoolean(preAcknowledge); - } - - @Override - public void encodeBody(final HornetQBuffer buffer) - { - buffer.writeString(name); - buffer.writeLong(originalSessionChannelID); - buffer.writeLong(replicatedSessionChannelID); - buffer.writeInt(version); - buffer.writeNullableString(username); - buffer.writeNullableString(password); - buffer.writeInt(minLargeMessageSize); - buffer.writeBoolean(xa); - buffer.writeBoolean(autoCommitSends); - buffer.writeBoolean(autoCommitAcks); - buffer.writeInt(windowSize); - buffer.writeBoolean(preAcknowledge); - } - - @Override - public void decodeBody(final HornetQBuffer buffer) - { - name = buffer.readString(); - originalSessionChannelID = buffer.readLong(); - replicatedSessionChannelID = buffer.readLong(); - version = buffer.readInt(); - username = buffer.readNullableString(); - password = buffer.readNullableString(); - minLargeMessageSize = buffer.readInt(); - xa = buffer.readBoolean(); - autoCommitSends = buffer.readBoolean(); - autoCommitAcks = buffer.readBoolean(); - windowSize = buffer.readInt(); - preAcknowledge = buffer.readBoolean(); - } - - @Override - public boolean equals(final Object other) - { - if (other instanceof ReplicateCreateSessionMessage == false) - { - return false; - } - - ReplicateCreateSessionMessage r = (ReplicateCreateSessionMessage)other; - - boolean matches = super.equals(other) && name.equals(r.name) && - originalSessionChannelID == r.originalSessionChannelID && - replicatedSessionChannelID == r.replicatedSessionChannelID && - version == r.version && - xa == r.xa && - autoCommitSends == r.autoCommitSends && - autoCommitAcks == r.autoCommitAcks && - (username == null ? r.username == null : username.equals(r.username)) && - (password == null ? r.password == null : password.equals(r.password)) && - minLargeMessageSize == r.minLargeMessageSize && - windowSize == r.windowSize; - - return matches; - } - - @Override - public final boolean isRequiresConfirmations() - { - return false; - } - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - // Private ------------------------------------------------------- - - // Inner classes ------------------------------------------------- -} Deleted: branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/SessionFailoverCompleteMessage.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/SessionFailoverCompleteMessage.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/impl/wireformat/SessionFailoverCompleteMessage.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -1,99 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.core.remoting.impl.wireformat; - -import org.hornetq.core.remoting.spi.HornetQBuffer; - -/** - * - * A SessionFailoverCompleteMessage - * - * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> - * - */ -public class SessionFailoverCompleteMessage extends PacketImpl -{ - // Constants ----------------------------------------------------- - - // Attributes ---------------------------------------------------- - - private String name; - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - public SessionFailoverCompleteMessage(final String name) - { - super(SESS_FAILOVER_COMPLETE); - - this.name = name; - } - - public SessionFailoverCompleteMessage() - { - super(SESS_FAILOVER_COMPLETE); - } - - // Public -------------------------------------------------------- - - public String getName() - { - return name; - } - - public int getRequiredBufferSize() - { - return BASIC_PACKET_SIZE + stringEncodeSize(name); - } - - @Override - public void encodeBody(final HornetQBuffer buffer) - { - buffer.writeString(name); - } - - @Override - public void decodeBody(final HornetQBuffer buffer) - { - name = buffer.readString(); - } - - @Override - public boolean isRequiresConfirmations() - { - return false; - } - - @Override - public boolean equals(final Object other) - { - if (other instanceof SessionFailoverCompleteMessage == false) - { - return false; - } - - SessionFailoverCompleteMessage r = (SessionFailoverCompleteMessage)other; - - return super.equals(other) && name.equals(r.name); - } - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - // Private ------------------------------------------------------- - - // Inner classes ------------------------------------------------- -} Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -303,8 +303,7 @@ } RemotingConnection rc = new RemotingConnectionImpl(connection, - interceptors, - !config.isBackup(), + interceptors, server.getConfiguration().isAsyncConnectionExecutionEnabled() ? server.getExecutorFactory() .getExecutor() : null); Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/HornetQServer.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/HornetQServer.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/HornetQServer.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -23,7 +23,6 @@ import org.hornetq.core.management.impl.HornetQServerControlImpl; import org.hornetq.core.persistence.StorageManager; import org.hornetq.core.postoffice.PostOffice; -import org.hornetq.core.remoting.Channel; import org.hornetq.core.remoting.RemotingConnection; import org.hornetq.core.remoting.impl.wireformat.CreateSessionResponseMessage; import org.hornetq.core.remoting.impl.wireformat.ReattachSessionResponseMessage; @@ -65,15 +64,10 @@ HornetQServerControlImpl getHornetQServerControl(); - void registerActivateCallback(ActivateCallback callback); - - void unregisterActivateCallback(ActivateCallback callback); - ReattachSessionResponseMessage reattachSession(RemotingConnection connection, String name, int lastReceivedCommandID) throws Exception; CreateSessionResponseMessage createSession(String name, - long channelID, - long replicatedSessionID, + long channelID, String username, String password, int minLargeMessageSize, @@ -85,20 +79,6 @@ boolean xa, int producerWindowSize) throws Exception; - void replicateCreateSession(String name, - long channelID, - long originalSessionID, - String username, - String password, - int minLargeMessageSize, - int incrementingVersion, - RemotingConnection remotingConnection, - boolean autoCommitSends, - boolean autoCommitAcks, - boolean preAcknowledge, - boolean xa, - int sendWindowSize) throws Exception; - void removeSession(String name) throws Exception; ServerSession getSession(String name); @@ -125,10 +105,8 @@ SimpleString getNodeID(); - Channel getReplicatingChannel(); + // void initialiseBackup(UUID nodeID, long currentMessageID) throws Exception; - void initialiseBackup(UUID nodeID, long currentMessageID) throws Exception; - boolean isInitialised(); Queue createQueue(SimpleString address, @@ -145,7 +123,5 @@ void destroyQueue(SimpleString queueName, ServerSession session) throws Exception; - void handleReplicateRedistribution(final SimpleString queueName, final long messageID) throws Exception; - ExecutorFactory getExecutorFactory(); } Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/Queue.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/Queue.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/Queue.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -121,18 +121,8 @@ int moveReferences(Filter filter, SimpleString toAddress) throws Exception; - void setBackup(); + void addRedistributor(long delay, Executor executor); - boolean activate(); - - void activateNow(Executor executor); - - boolean isBackup(); - - boolean consumerFailedOver(); - - void addRedistributor(long delay, Executor executor, final Channel replicatingChannel); - void cancelRedistributor() throws Exception; // Only used in testing Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/ScheduledDeliveryHandler.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/ScheduledDeliveryHandler.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/ScheduledDeliveryHandler.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -19,7 +19,7 @@ */ public interface ScheduledDeliveryHandler { - boolean checkAndSchedule(MessageReference ref, boolean backup); + boolean checkAndSchedule(MessageReference ref); void reSchedule(); Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/ServerConsumer.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/ServerConsumer.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/ServerConsumer.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -43,12 +43,4 @@ MessageReference getExpired(long messageID) throws Exception; void acknowledge(boolean autoCommitAcks, Transaction tx, long messageID) throws Exception; - - void failedOver(); - - void deliverReplicated(long messageID) throws Exception; - - void lock(); - - void unlock(); } Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/ServerSession.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/ServerSession.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/ServerSession.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -17,7 +17,6 @@ import org.hornetq.core.remoting.Packet; import org.hornetq.core.remoting.RemotingConnection; import org.hornetq.core.remoting.impl.wireformat.CreateQueueMessage; -import org.hornetq.core.remoting.impl.wireformat.PacketsConfirmedMessage; import org.hornetq.core.remoting.impl.wireformat.RollbackMessage; import org.hornetq.core.remoting.impl.wireformat.SessionAcknowledgeMessage; import org.hornetq.core.remoting.impl.wireformat.SessionBindingQueryMessage; @@ -39,7 +38,6 @@ import org.hornetq.core.remoting.impl.wireformat.SessionXARollbackMessage; import org.hornetq.core.remoting.impl.wireformat.SessionXASetTimeoutMessage; import org.hornetq.core.remoting.impl.wireformat.SessionXAStartMessage; -import org.hornetq.core.remoting.impl.wireformat.replication.SessionReplicateDeliveryMessage; import org.hornetq.core.server.impl.ServerSessionPacketHandler; /** @@ -126,14 +124,8 @@ void handleSendLargeMessage(SessionSendLargeMessage packet); - void handleFailedOver(Packet packet); - void handleClose(Packet packet); - void handleReplicatedDelivery(SessionReplicateDeliveryMessage packet); - - void handlePacketsConfirmed(PacketsConfirmedMessage packet); - int transferConnection(RemotingConnection newConnection, int lastReceivedCommandID); Channel getChannel(); Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -41,11 +41,8 @@ import org.hornetq.core.message.impl.MessageImpl; import org.hornetq.core.persistence.StorageManager; import org.hornetq.core.postoffice.BindingType; -import org.hornetq.core.remoting.Channel; import org.hornetq.core.remoting.FailureListener; -import org.hornetq.core.remoting.Packet; import org.hornetq.core.remoting.RemotingConnection; -import org.hornetq.core.remoting.impl.wireformat.replication.ReplicateAcknowledgeMessage; import org.hornetq.core.server.HandleStatus; import org.hornetq.core.server.MessageReference; import org.hornetq.core.server.Queue; @@ -131,8 +128,6 @@ private final String clusterPassword; - private Channel replicatingChannel; - private boolean activated; private NotificationService notificationService; @@ -169,7 +164,6 @@ final String clusterUser, final String clusterPassword, final MessageFlowRecord flowRecord, - final Channel replicatingChannel, final boolean activated, final StorageManager storageManager) throws Exception { @@ -215,8 +209,6 @@ this.flowRecord = flowRecord; - this.replicatingChannel = replicatingChannel; - this.activated = activated; } @@ -309,8 +301,6 @@ public synchronized void activate() { - replicatingChannel = null; - activated = true; executor.execute(new CreateObjectsRunnable()); @@ -374,30 +364,7 @@ if (ref != null) { - if (replicatingChannel == null) - { - // Acknowledge when we know send has been processed on the server - ref.getQueue().acknowledge(ref); - } - else - { - Packet packet = new ReplicateAcknowledgeMessage(name, ref.getMessage().getMessageID()); - - replicatingChannel.replicatePacket(packet, 1, new Runnable() - { - public void run() - { - try - { - ref.getQueue().acknowledge(ref); - } - catch (Exception e) - { - log.error("Failed to ack", e); - } - } - }); - } + ref.getQueue().acknowledge(ref); } } catch (Exception e) Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -37,13 +37,6 @@ import org.hornetq.core.postoffice.Binding; import org.hornetq.core.postoffice.Bindings; import org.hornetq.core.postoffice.PostOffice; -import org.hornetq.core.remoting.Channel; -import org.hornetq.core.remoting.Packet; -import org.hornetq.core.remoting.impl.wireformat.CreateQueueMessage; -import org.hornetq.core.remoting.impl.wireformat.replication.ReplicateRemoteBindingAddedMessage; -import org.hornetq.core.remoting.impl.wireformat.replication.ReplicateRemoteBindingRemovedMessage; -import org.hornetq.core.remoting.impl.wireformat.replication.ReplicateRemoteConsumerAddedMessage; -import org.hornetq.core.remoting.impl.wireformat.replication.ReplicateRemoteConsumerRemovedMessage; import org.hornetq.core.server.HornetQServer; import org.hornetq.core.server.Queue; import org.hornetq.core.server.cluster.Bridge; @@ -97,16 +90,12 @@ private final UUID nodeUUID; - private final Channel replicatingChannel; - private final List<Pair<TransportConfiguration, TransportConfiguration>> staticConnectors; private boolean backup; private volatile boolean started; - - private int replicationCount; - + /* * Constructor using static list of connectors */ @@ -123,7 +112,6 @@ final List<Pair<TransportConfiguration, TransportConfiguration>> connectors, final int maxHops, final UUID nodeUUID, - final Channel replicatingChannel, final boolean backup) throws Exception { this.name = name; @@ -157,8 +145,6 @@ this.nodeUUID = nodeUUID; - this.replicatingChannel = replicatingChannel; - this.backup = backup; this.staticConnectors = connectors; @@ -185,7 +171,6 @@ final DiscoveryGroup discoveryGroup, final int maxHops, final UUID nodeUUID, - final Channel replicatingChannel, final boolean backup) throws Exception { this.name = name; @@ -214,8 +199,6 @@ this.nodeUUID = nodeUUID; - this.replicatingChannel = replicatingChannel; - this.backup = backup; this.staticConnectors = null; @@ -237,7 +220,8 @@ } public synchronized void stop() throws Exception - { if (!started) + { + if (!started) { return; } @@ -374,47 +358,9 @@ // Add binding in storage so the queue will get reloaded on startup and we can find it - it's never // actually routed to at that address though - if (replicatingChannel == null) - { - queue = server.createQueue(queueName, queueName, null, true, false); + queue = server.createQueue(queueName, queueName, null, true, false); - createNewRecord(entry.getKey(), connectorPair, queueName, queue, true); - } - else - { - // We need to create the record before we replicate, since otherwise, two updates can come in for - // the same entry before the first replication comes back, and it won't find the record, so it - // will try and create the queue twice - createNewRecord(entry.getKey(), connectorPair, queueName, null, false); - - // Replicate the createQueue first - Packet packet = new CreateQueueMessage(queueName, queueName, null, true, false); - - replicatingChannel.replicatePacket(packet, 1, new Runnable() - { - public void run() - { - try - { - Queue queue = server.createQueue(queueName, queueName, null, true, false); - - synchronized (ClusterConnectionImpl.this) - { - MessageFlowRecord record = records.get(entry.getKey()); - - if (record != null) - { - record.activate(queue); - } - } - } - catch (Exception e) - { - log.error("Failed create record", e); - } - } - }); - } + createNewRecord(entry.getKey(), connectorPair, queueName, queue, true); } } } @@ -449,7 +395,6 @@ managementService.getClusterUser(), managementService.getClusterPassword(), record, - replicatingChannel, !backup, server.getStorageManager()); @@ -495,52 +440,8 @@ bridge.stop(); clearBindings(); - - waitForReplicationsToComplete(3000); } - - private synchronized void waitForReplicationsToComplete(long timeout) - { - long toWait = timeout; - long start = System.currentTimeMillis(); - - while (replicationCount > 0 && toWait > 0) - { - try - { - wait(toWait); - } - catch (InterruptedException e) - { - } - - long now = System.currentTimeMillis(); - - toWait -= now - start; - - start = now; - } - - if (toWait <= 0) - { - log.warn("Timed out waiting for replication responses to return"); - } - - } - - private synchronized void replicationComplete() - { - replicationCount--; - - notify(); - } - - private synchronized void beforeReplicate() - { - replicationCount++; - } - public void activate(final Queue queue) throws Exception { this.queue = queue; @@ -584,7 +485,7 @@ { case BINDING_ADDED: { - doBindingAdded(message, replicatingChannel); + doBindingAdded(message); break; } @@ -596,13 +497,13 @@ } case CONSUMER_CREATED: { - doConsumerCreated(message, replicatingChannel); + doConsumerCreated(message); break; } case CONSUMER_CLOSED: { - doConsumerClosed(message, replicatingChannel); + doConsumerClosed(message); break; } @@ -622,14 +523,14 @@ } private synchronized void clearBindings() throws Exception - { + { for (RemoteQueueBinding binding : new HashSet<RemoteQueueBinding>(bindings.values())) { - removeBinding(binding.getClusterName(), replicatingChannel); - } + removeBinding(binding.getClusterName()); + } } - private synchronized void doBindingAdded(final ClientMessage message, final Channel replChannel) throws Exception + private synchronized void doBindingAdded(final ClientMessage message) throws Exception { Integer distance = (Integer)message.getProperty(ManagementHelper.HDR_DISTANCE); @@ -668,73 +569,42 @@ throw new IllegalStateException("queueID is null"); } - if (replChannel != null) - { - Packet packet = new ReplicateRemoteBindingAddedMessage(name, - queueAddress, - clusterName, - routingName, - queueID, - filterString, - queue.getName(), - distance + 1); + RemoteQueueBinding binding = new RemoteQueueBindingImpl(queueAddress, + clusterName, + routingName, + queueID, + filterString, + queue, + bridge.getName(), + distance + 1); - beforeReplicate(); - replChannel.replicatePacket(packet, 1, new Runnable() - { - public void run() - { - try - { - doBindingAdded(message, null); - } - catch (Exception e) - { - log.error("Failed to add remote queue binding", e); - } - - replicationComplete(); - } - }); - } - else + bindings.put(clusterName, binding); + + if (postOffice.getBinding(clusterName) != null) { - RemoteQueueBinding binding = new RemoteQueueBindingImpl(queueAddress, - clusterName, - routingName, - queueID, - filterString, - queue, - bridge.getName(), - distance + 1); + // Sanity check - this means the binding has already been added via another bridge, probably max + // hops is too high + // or there are multiple cluster connections for the same address - bindings.put(clusterName, binding); + log.warn("Remote queue binding " + clusterName + + " has already been bound in the post office. Most likely cause for this is you have a loop " + + "in your cluster due to cluster max-hops being too large or you have multiple cluster connections to the same nodes using overlapping addresses"); - if (postOffice.getBinding(clusterName) != null) - { - // Sanity check - this means the binding has already been added via another bridge, probably max - // hops is too high - // or there are multiple cluster connections for the same address + return; + } - log.warn("Remote queue binding " + clusterName + - " has already been bound in the post office. Most likely cause for this is you have a loop " + - "in your cluster due to cluster max-hops being too large or you have multiple cluster connections to the same nodes using overlapping addresses"); + try + { + postOffice.addBinding(binding); + } + catch (Exception ignore) + { + } - return; - } + Bindings theBindings = postOffice.getBindingsForAddress(queueAddress); - try - { - postOffice.addBinding(binding); - } - catch (Exception ignore) - { - } + theBindings.setRouteWhenNoConsumers(routeWhenNoConsumers); - Bindings theBindings = postOffice.getBindingsForAddress(queueAddress); - - theBindings.setRouteWhenNoConsumers(routeWhenNoConsumers); - } } private void doBindingRemoved(final ClientMessage message) throws Exception @@ -746,47 +616,22 @@ throw new IllegalStateException("clusterName is null"); } - removeBinding(clusterName, replicatingChannel); + removeBinding(clusterName); } - - private synchronized void removeBinding(final SimpleString clusterName, final Channel replChannel) throws Exception + + private synchronized void removeBinding(final SimpleString clusterName) throws Exception { - if (replChannel != null) - { - Packet packet = new ReplicateRemoteBindingRemovedMessage(clusterName); + RemoteQueueBinding binding = bindings.remove(clusterName); - beforeReplicate(); - replChannel.replicatePacket(packet, 1, new Runnable() - { - public void run() - { - try - { - removeBinding(clusterName, null); - } - catch (Exception e) - { - log.error("Failed to remove remote queue binding", e); - } - - replicationComplete(); - } - }); - } - else + if (binding == null) { - RemoteQueueBinding binding = bindings.remove(clusterName); - - if (binding == null) - { - throw new IllegalStateException("Cannot find binding for queue " + clusterName); - } - - postOffice.removeBinding(binding.getUniqueName()); + throw new IllegalStateException("Cannot find binding for queue " + clusterName); } + + postOffice.removeBinding(binding.getUniqueName()); } - private synchronized void doConsumerCreated(final ClientMessage message, final Channel replChannel) throws Exception + private synchronized void doConsumerCreated(final ClientMessage message) throws Exception { Integer distance = (Integer)message.getProperty(ManagementHelper.HDR_DISTANCE); @@ -806,47 +651,22 @@ SimpleString filterString = (SimpleString)message.getProperty(ManagementHelper.HDR_FILTERSTRING); - if (replChannel != null) - { - Packet packet = new ReplicateRemoteConsumerAddedMessage(clusterName, filterString, message.getProperties()); + RemoteQueueBinding binding = bindings.get(clusterName); - beforeReplicate(); - replChannel.replicatePacket(packet, 1, new Runnable() - { - public void run() - { - try - { - doConsumerCreated(message, null); - } - catch (Exception e) - { - log.error("Failed to add remote consumer", e); - } - - replicationComplete(); - } - }); - } - else + if (binding == null) { - RemoteQueueBinding binding = bindings.get(clusterName); + throw new IllegalStateException("Cannot find binding for " + clusterName); + } - if (binding == null) - { - throw new IllegalStateException("Cannot find binding for " + clusterName); - } + binding.addConsumer(filterString); - binding.addConsumer(filterString); + // Need to propagate the consumer add + Notification notification = new Notification(null, CONSUMER_CREATED, message.getProperties()); - // Need to propagate the consumer add - Notification notification = new Notification(null, CONSUMER_CREATED, message.getProperties()); - - managementService.sendNotification(notification); - } + managementService.sendNotification(notification); } - private synchronized void doConsumerClosed(final ClientMessage message, final Channel replChannel) throws Exception + private synchronized void doConsumerClosed(final ClientMessage message) throws Exception { Integer distance = (Integer)message.getProperty(ManagementHelper.HDR_DISTANCE); @@ -866,46 +686,19 @@ SimpleString filterString = (SimpleString)message.getProperty(ManagementHelper.HDR_FILTERSTRING); - if (replChannel != null) - { - Packet packet = new ReplicateRemoteConsumerRemovedMessage(clusterName, - filterString, - message.getProperties()); + RemoteQueueBinding binding = bindings.get(clusterName); - beforeReplicate(); - replChannel.replicatePacket(packet, 1, new Runnable() - { - public void run() - { - try - { - doConsumerClosed(message, null); - } - catch (Exception e) - { - log.error("Failed to remove remote consumer", e); - } - - replicationComplete(); - } - }); - } - else + if (binding == null) { - RemoteQueueBinding binding = bindings.get(clusterName); + throw new IllegalStateException("Cannot find binding for " + clusterName); + } - if (binding == null) - { - throw new IllegalStateException("Cannot find binding for " + clusterName); - } + binding.removeConsumer(filterString); - binding.removeConsumer(filterString); + // Need to propagate the consumer close + Notification notification = new Notification(null, CONSUMER_CLOSED, message.getProperties()); - // Need to propagate the consumer close - Notification notification = new Notification(null, CONSUMER_CLOSED, message.getProperties()); - - managementService.sendNotification(notification); - } + managementService.sendNotification(notification); } } Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -84,8 +84,6 @@ private final UUID nodeUUID; - private Channel replicatingChannel; - private volatile boolean started; private boolean backup; @@ -96,8 +94,7 @@ final ScheduledExecutorService scheduledExecutor, final ManagementService managementService, final Configuration configuration, - final UUID nodeUUID, - final Channel replicatingChannel, + final UUID nodeUUID, final boolean backup) { if (nodeUUID == null) @@ -119,8 +116,6 @@ this.nodeUUID = nodeUUID; - this.replicatingChannel = replicatingChannel; - this.backup = backup; } @@ -236,8 +231,6 @@ cc.activate(); } - replicatingChannel = null; - backup = false; } @@ -420,8 +413,7 @@ managementService.getManagementNotificationAddress(), managementService.getClusterUser(), managementService.getClusterPassword(), - null, - replicatingChannel, + null, !backup, server.getStorageManager()); } @@ -473,8 +465,7 @@ managementService.getManagementNotificationAddress(), managementService.getClusterUser(), managementService.getClusterPassword(), - null, - replicatingChannel, + null, !backup, server.getStorageManager()); } @@ -553,8 +544,7 @@ scheduledExecutor, connectors, config.getMaxHops(), - nodeUUID, - replicatingChannel, + nodeUUID, backup); } else @@ -579,8 +569,7 @@ scheduledExecutor, dg, config.getMaxHops(), - nodeUUID, - replicatingChannel, + nodeUUID, backup); } Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/cluster/impl/Redistributor.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/cluster/impl/Redistributor.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/cluster/impl/Redistributor.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -19,9 +19,6 @@ import org.hornetq.core.logging.Logger; import org.hornetq.core.persistence.StorageManager; import org.hornetq.core.postoffice.PostOffice; -import org.hornetq.core.remoting.Channel; -import org.hornetq.core.remoting.Packet; -import org.hornetq.core.remoting.impl.wireformat.replication.ReplicateRedistributionMessage; import org.hornetq.core.server.Consumer; import org.hornetq.core.server.HandleStatus; import org.hornetq.core.server.MessageReference; @@ -57,14 +54,11 @@ private int count; - private final Channel replicatingChannel; - public Redistributor(final Queue queue, final StorageManager storageManager, final PostOffice postOffice, final Executor executor, - final int batchSize, - final Channel replicatingChannel) + final int batchSize) { this.queue = queue; @@ -75,8 +69,6 @@ this.executor = executor; this.batchSize = batchSize; - - this.replicatingChannel = replicatingChannel; } public Filter getFilter() @@ -134,30 +126,8 @@ if (routed) { - if (replicatingChannel == null) - { - doRedistribute(reference, tx); - } - else - { - Packet packet = new ReplicateRedistributionMessage(queue.getName(), reference.getMessage().getMessageID()); - - replicatingChannel.replicatePacket(packet, 1, new Runnable() - { - public void run() - { - try - { - doRedistribute(reference, tx); - } - catch (Exception e) - { - log.error("Failed to handle redistribution", e); - } - } - }); - } - + doRedistribute(reference, tx); + return HandleStatus.HANDLED; } else Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/HornetQPacketHandler.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/HornetQPacketHandler.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/HornetQPacketHandler.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -13,18 +13,12 @@ package org.hornetq.core.server.impl; -import static org.hornetq.core.management.NotificationType.CONSUMER_CLOSED; -import static org.hornetq.core.management.NotificationType.CONSUMER_CREATED; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.CREATESESSION; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.CREATE_QUEUE; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REATTACH_SESSION; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATE_CREATESESSION; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATE_STARTUP_INFO; import org.hornetq.core.exception.HornetQException; import org.hornetq.core.logging.Logger; -import org.hornetq.core.management.Notification; -import org.hornetq.core.postoffice.Binding; import org.hornetq.core.remoting.Channel; import org.hornetq.core.remoting.ChannelHandler; import org.hornetq.core.remoting.Packet; @@ -32,21 +26,8 @@ import org.hornetq.core.remoting.impl.wireformat.CreateQueueMessage; import org.hornetq.core.remoting.impl.wireformat.CreateSessionMessage; import org.hornetq.core.remoting.impl.wireformat.HornetQExceptionMessage; -import org.hornetq.core.remoting.impl.wireformat.PacketImpl; import org.hornetq.core.remoting.impl.wireformat.ReattachSessionMessage; -import org.hornetq.core.remoting.impl.wireformat.ReplicateCreateSessionMessage; -import org.hornetq.core.remoting.impl.wireformat.replication.ReplicateAcknowledgeMessage; -import org.hornetq.core.remoting.impl.wireformat.replication.ReplicateRedistributionMessage; -import org.hornetq.core.remoting.impl.wireformat.replication.ReplicateRemoteBindingAddedMessage; -import org.hornetq.core.remoting.impl.wireformat.replication.ReplicateRemoteBindingRemovedMessage; -import org.hornetq.core.remoting.impl.wireformat.replication.ReplicateRemoteConsumerAddedMessage; -import org.hornetq.core.remoting.impl.wireformat.replication.ReplicateRemoteConsumerRemovedMessage; -import org.hornetq.core.remoting.impl.wireformat.replication.ReplicateStartupInfoMessage; import org.hornetq.core.server.HornetQServer; -import org.hornetq.core.server.MessageReference; -import org.hornetq.core.server.Queue; -import org.hornetq.core.server.cluster.ClusterConnection; -import org.hornetq.core.server.cluster.RemoteQueueBinding; /** * A packet handler for all packets that need to be handled at the server level @@ -80,30 +61,10 @@ { byte type = packet.getType(); - if (!server.isInitialised() && type != PacketImpl.REPLICATE_STARTUP_INFO) - { - throw new IllegalStateException("First packet must be startup info for backup " + type); - } - // All these operations need to be idempotent since they are outside of the session // reliability replay functionality switch (type) - { - case REPLICATE_STARTUP_INFO: - { - ReplicateStartupInfoMessage msg = (ReplicateStartupInfoMessage)packet; - - try - { - server.initialiseBackup(msg.getNodeID(), msg.getCurrentMessageID()); - } - catch (Exception e) - { - log.error("Failed to initialise", e); - } - - break; - } + { case CREATESESSION: { CreateSessionMessage request = (CreateSessionMessage)packet; @@ -111,15 +72,7 @@ handleCreateSession(request); break; - } - case REPLICATE_CREATESESSION: - { - ReplicateCreateSessionMessage request = (ReplicateCreateSessionMessage)packet; - - handleReplicateCreateSession(request); - - break; - } + } case REATTACH_SESSION: { ReattachSessionMessage request = (ReattachSessionMessage)packet; @@ -137,55 +90,7 @@ handleCreateQueue(request); break; - } - case PacketImpl.REPLICATE_ADD_REMOTE_QUEUE_BINDING: - { - ReplicateRemoteBindingAddedMessage request = (ReplicateRemoteBindingAddedMessage)packet; - - handleAddRemoteQueueBinding(request); - - break; - } - case PacketImpl.REPLICATE_REMOVE_REMOTE_QUEUE_BINDING: - { - ReplicateRemoteBindingRemovedMessage request = (ReplicateRemoteBindingRemovedMessage)packet; - - handleRemoveRemoteQueueBinding(request); - - break; - } - case PacketImpl.REPLICATE_ADD_REMOTE_CONSUMER: - { - ReplicateRemoteConsumerAddedMessage request = (ReplicateRemoteConsumerAddedMessage)packet; - - handleAddRemoteConsumer(request); - - break; - } - case PacketImpl.REPLICATE_REMOVE_REMOTE_CONSUMER: - { - ReplicateRemoteConsumerRemovedMessage request = (ReplicateRemoteConsumerRemovedMessage)packet; - - handleRemoveRemoteConsumer(request); - - break; - } - case PacketImpl.REPLICATE_ACKNOWLEDGE: - { - ReplicateAcknowledgeMessage request = (ReplicateAcknowledgeMessage)packet; - - handleReplicateAcknowledge(request); - - break; - } - case PacketImpl.REPLICATE_REDISTRIBUTION: - { - ReplicateRedistributionMessage message = (ReplicateRedistributionMessage)packet; - - handleReplicateRedistribution(message); - - break; - } + } default: { log.error("Invalid packet " + packet); @@ -193,14 +98,13 @@ } } - private void doHandleCreateSession(final CreateSessionMessage request, final long oppositeChannelID) + private void handleCreateSession(final CreateSessionMessage request) { Packet response; try { response = server.createSession(request.getName(), - request.getSessionChannelID(), - oppositeChannelID, + request.getSessionChannelID(), request.getUsername(), request.getPassword(), request.getMinLargeMessageSize(), @@ -231,68 +135,9 @@ } } - channel1.send(response); + channel1.send(response); } - - private void handleCreateSession(final CreateSessionMessage request) - { - Channel replicatingChannel = server.getReplicatingChannel(); - - if (replicatingChannel == null) - { - doHandleCreateSession(request, -1); - } - else - { - final long replicatedChannelID = replicatingChannel.getConnection().generateChannelID(); - - Packet replPacket = new ReplicateCreateSessionMessage(request.getName(), - replicatedChannelID, - request.getSessionChannelID(), - request.getVersion(), - request.getUsername(), - request.getPassword(), - request.getMinLargeMessageSize(), - request.isXA(), - request.isAutoCommitSends(), - request.isAutoCommitAcks(), - request.isPreAcknowledge(), - request.getWindowSize()); - - replicatingChannel.replicatePacket(replPacket, 1, new Runnable() - { - public void run() - { - doHandleCreateSession(request, replicatedChannelID); - } - }); - } - } - - private void handleReplicateCreateSession(final ReplicateCreateSessionMessage request) - { - try - { - server.replicateCreateSession(request.getName(), - request.getReplicatedSessionChannelID(), - request.getOriginalSessionChannelID(), - request.getUsername(), - request.getPassword(), - request.getMinLargeMessageSize(), - request.getVersion(), - connection, - request.isAutoCommitSends(), - request.isAutoCommitAcks(), - request.isPreAcknowledge(), - request.isXA(), - request.getWindowSize()); - } - catch (Exception e) - { - log.error("Failed to handle replicate create session", e); - } - } - + private void handleReattachSession(final ReattachSessionMessage request) { Packet response; @@ -330,151 +175,5 @@ } } - private void handleAddRemoteQueueBinding(final ReplicateRemoteBindingAddedMessage request) - { - ClusterConnection cc = server.getClusterManager().getClusterConnection(request.getClusterConnectionName()); - - if (cc == null) - { - throw new IllegalStateException("No cluster connection found with name " + request.getClusterConnectionName()); - } - - try - { - cc.handleReplicatedAddBinding(request.getAddress(), - request.getUniqueName(), - request.getRoutingName(), - request.getRemoteQueueID(), - request.getFilterString(), - request.getSfQueueName(), - request.getDistance()); - } - catch (Exception e) - { - log.error("Failed to handle add remote queue binding", e); - } - } - - private void handleRemoveRemoteQueueBinding(final ReplicateRemoteBindingRemovedMessage request) - { - try - { - Binding binding = server.getPostOffice().removeBinding(request.getUniqueName()); - - if (binding == null) - { - throw new IllegalStateException("Cannot find binding to remove " + request.getUniqueName()); - } - } - catch (Exception e) - { - log.error("Failed to handle remove remote queue binding", e); - } - } - - private void handleAddRemoteConsumer(final ReplicateRemoteConsumerAddedMessage request) - { - RemoteQueueBinding binding = (RemoteQueueBinding)server.getPostOffice() - .getBinding(request.getUniqueBindingName()); - - if (binding == null) - { - throw new IllegalStateException("Cannot find binding to remove " + request.getUniqueBindingName()); - } - - try - { - binding.addConsumer(request.getFilterString()); - } - catch (Exception e) - { - log.error("Failed to handle add remote consumer", e); - } - - // Need to propagate the consumer add - Notification notification = new Notification(null, CONSUMER_CREATED, request.getProperties()); - - try - { - server.getManagementService().sendNotification(notification); - } - catch (Exception e) - { - log.error("Failed to handle add remote consumer", e); - } - } - - private void handleRemoveRemoteConsumer(final ReplicateRemoteConsumerRemovedMessage request) - { - RemoteQueueBinding binding = (RemoteQueueBinding)server.getPostOffice() - .getBinding(request.getUniqueBindingName()); - - if (binding == null) - { - throw new IllegalStateException("Cannot find binding to remove " + request.getUniqueBindingName()); - } - - try - { - binding.removeConsumer(request.getFilterString()); - } - catch (Exception e) - { - log.error("Failed to handle remove remote consumer", e); - } - - // Need to propagate the consumer close - Notification notification = new Notification(null, CONSUMER_CLOSED, request.getProperties()); - - try - { - server.getManagementService().sendNotification(notification); - } - catch (Exception e) - { - log.error("Failed to handle remove remote consumer", e); - } - } - - private void handleReplicateAcknowledge(final ReplicateAcknowledgeMessage request) - { - Binding binding = server.getPostOffice().getBinding(request.getUniqueName()); - - if (binding == null) - { - throw new IllegalStateException("Cannot find binding " + request.getUniqueName()); - } - - try - { - Queue queue = (Queue)binding.getBindable(); - - MessageReference ref = queue.removeFirstReference(request.getMessageID()); - - queue.acknowledge(ref); - } - catch (Exception e) - { - log.error("Failed to handle remove remote consumer", e); - } - } - private void handleReplicateRedistribution(final ReplicateRedistributionMessage request) - { - Binding binding = server.getPostOffice().getBinding(request.getQueueName()); - - if (binding == null) - { - throw new IllegalStateException("Cannot find binding " + request.getQueueName()); - } - - try - { - server.handleReplicateRedistribution(request.getQueueName(), request.getMessageID()); - } - catch (Exception e) - { - log.error("Failed to handle remove remote consumer", e); - } - } } \ No newline at end of file Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -32,11 +32,7 @@ import javax.management.MBeanServer; -import org.hornetq.core.client.impl.ClientSessionFactoryImpl; -import org.hornetq.core.client.impl.ConnectionManager; -import org.hornetq.core.client.impl.ConnectionManagerImpl; import org.hornetq.core.config.Configuration; -import org.hornetq.core.config.TransportConfiguration; import org.hornetq.core.config.cluster.DivertConfiguration; import org.hornetq.core.config.cluster.QueueConfiguration; import org.hornetq.core.config.impl.ConfigurationImpl; @@ -68,12 +64,9 @@ import org.hornetq.core.postoffice.impl.LocalQueueBinding; import org.hornetq.core.postoffice.impl.PostOfficeImpl; import org.hornetq.core.remoting.Channel; -import org.hornetq.core.remoting.FailureListener; -import org.hornetq.core.remoting.Packet; import org.hornetq.core.remoting.RemotingConnection; import org.hornetq.core.remoting.impl.wireformat.CreateSessionResponseMessage; import org.hornetq.core.remoting.impl.wireformat.ReattachSessionResponseMessage; -import org.hornetq.core.remoting.impl.wireformat.replication.ReplicateStartupInfoMessage; import org.hornetq.core.remoting.server.RemotingService; import org.hornetq.core.remoting.server.impl.RemotingServiceImpl; import org.hornetq.core.security.CheckType; @@ -81,7 +74,6 @@ import org.hornetq.core.security.Role; import org.hornetq.core.security.SecurityStore; import org.hornetq.core.security.impl.SecurityStoreImpl; -import org.hornetq.core.server.ActivateCallback; import org.hornetq.core.server.Divert; import org.hornetq.core.server.HornetQServer; import org.hornetq.core.server.MemoryManager; @@ -101,7 +93,6 @@ import org.hornetq.core.transaction.impl.TransactionImpl; import org.hornetq.core.version.Version; import org.hornetq.utils.ExecutorFactory; -import org.hornetq.utils.Future; import org.hornetq.utils.HornetQThreadFactory; import org.hornetq.utils.OrderedExecutorFactory; import org.hornetq.utils.Pair; @@ -142,8 +133,6 @@ private final MBeanServer mbeanServer; - private final Set<ActivateCallback> activateCallbacks = new HashSet<ActivateCallback>(); - private volatile boolean started; private SecurityStore securityStore; @@ -189,17 +178,11 @@ private Deployer securityDeployer; private final Map<String, ServerSession> sessions = new ConcurrentHashMap<String, ServerSession>(); - - private RemotingConnection replicatingConnection; - - private Channel replicatingChannel; - + private final Object initialiseLock = new Object(); private boolean initialised; - - private ConnectionManager replicatingConnectionManager; - + private int managementConnectorID; private static AtomicInteger managementConnectorSequence = new AtomicInteger(0); @@ -276,7 +259,7 @@ if (configuration.isBackup()) { // We defer actually initialisation until the live node has contacted the backup - log.info("Backup server will await live server before becoming operational"); + log.info("Backup server initialised"); } else { @@ -354,22 +337,7 @@ { securityManager.stop(); } - - if (replicatingConnection != null) - { - try - { - replicatingConnection.destroy(); - } - catch (Exception ignore) - { - } - - replicatingConnection = null; - replicatingChannel = null; - replicatingConnectionManager = null; - } - + resourceManager.stop(); postOffice.stop(); @@ -496,57 +464,28 @@ { ServerSession session = sessions.get(name); - // Need to activate the connection even if session can't be found - since otherwise response - // will never get back + if (!checkActivate()) + { + return new ReattachSessionResponseMessage(-1, false, false); + } - checkActivate(connection); - if (session == null) { - return new ReattachSessionResponseMessage(-1, true); + create the session + + return new ReattachSessionResponseMessage(-1, false, true); } else { // Reconnect the channel to the new connection int serverLastReceivedCommandID = session.transferConnection(connection, lastReceivedCommandID); - return new ReattachSessionResponseMessage(serverLastReceivedCommandID, false); + return new ReattachSessionResponseMessage(serverLastReceivedCommandID, true, true); } } - public void replicateCreateSession(final String name, - final long replicatedChannelID, - final long originalChannelID, - final String username, - final String password, - final int minLargeMessageSize, - final int incrementingVersion, - final RemotingConnection connection, - final boolean autoCommitSends, - final boolean autoCommitAcks, - final boolean preAcknowledge, - final boolean xa, - final int producerWindowSize) throws Exception - { - doCreateSession(name, - replicatedChannelID, - originalChannelID, - username, - password, - minLargeMessageSize, - incrementingVersion, - connection, - autoCommitSends, - autoCommitAcks, - preAcknowledge, - xa, - producerWindowSize, - true); - } - public CreateSessionResponseMessage createSession(final String name, - final long channelID, - final long replicatedChannelID, + final long channelID, final String username, final String password, final int minLargeMessageSize, @@ -557,23 +496,70 @@ final boolean preAcknowledge, final boolean xa, final int sendWindowSize) throws Exception - { - checkActivate(connection); + { + if (version.getIncrementingVersion() != incrementingVersion) + { + log.warn("Client with version " + incrementingVersion + + " and address " + + connection.getRemoteAddress() + + " is not compatible with server version " + + version.getFullVersion() + + ". " + + "Please ensure all clients and servers are upgraded to the same version for them to " + + "interoperate properly"); + return null; + } + + if (!checkActivate()) + { + //Backup server is not ready to accept connections + + return new CreateSessionResponseMessage(false, version.getIncrementingVersion()); + } - return doCreateSession(name, - channelID, - replicatedChannelID, - username, - password, - minLargeMessageSize, - incrementingVersion, - connection, - autoCommitSends, - autoCommitAcks, - preAcknowledge, - xa, - sendWindowSize, - false); + securityStore.authenticate(username, password); + + ServerSession currentSession = sessions.remove(name); + + if (currentSession != null) + { + // This session may well be on a different connection and different channel id, so we must get rid + // of it and create another + currentSession.getChannel().close(); + } + + Channel channel = connection.getChannel(channelID, sendWindowSize, false); + + final ServerSessionImpl session = new ServerSessionImpl(name, + username, + password, + minLargeMessageSize, + autoCommitSends, + autoCommitAcks, + preAcknowledge, + configuration.isPersistDeliveryCountBeforeDelivery(), + xa, + connection, + storageManager, + postOffice, + resourceManager, + securityStore, + executorFactory.getExecutor(), + channel, + managementService, + queueFactory, + this, + configuration.getManagementAddress()); + + sessions.put(name, session); + + ServerSessionPacketHandler handler = new ServerSessionPacketHandler(session); + + session.setHandler(handler); + + channel.setHandler(handler); + + return new CreateSessionResponseMessage(true, version.getIncrementingVersion()); } public void removeSession(final String name) throws Exception @@ -614,42 +600,42 @@ } } - public void initialiseBackup(final UUID theUUID, final long liveUniqueID) throws Exception - { - if (theUUID == null) - { - throw new IllegalArgumentException("node id is null"); - } +// public void initialiseBackup(final UUID theUUID, final long liveUniqueID) throws Exception +// { +// if (theUUID == null) +// { +// throw new IllegalArgumentException("node id is null"); +// } +// +// synchronized (initialiseLock) +// { +// if (initialised) +// { +// throw new IllegalStateException("Server is already initialised"); +// } +// +// this.uuid = theUUID; +// +// this.nodeID = new SimpleString(uuid.toString()); +// +// initialisePart2(); +// +// long backupID = storageManager.getCurrentUniqueID(); +// +// if (liveUniqueID != backupID) +// { +// initialised = false; +// +// throw new IllegalStateException("Live and backup unique ids different (" + liveUniqueID + +// ":" + +// backupID + +// "). You're probably trying to restart a live backup pair after a crash"); +// } +// +// log.info("Backup server is now operational"); +// } +// } - synchronized (initialiseLock) - { - if (initialised) - { - throw new IllegalStateException("Server is already initialised"); - } - - this.uuid = theUUID; - - this.nodeID = new SimpleString(uuid.toString()); - - initialisePart2(); - - long backupID = storageManager.getCurrentUniqueID(); - - if (liveUniqueID != backupID) - { - initialised = false; - - throw new IllegalStateException("Live and backup unique ids different (" + liveUniqueID + - ":" + - backupID + - "). You're probably trying to restart a live backup pair after a crash"); - } - - log.info("Backup server is now operational"); - } - } - public HornetQServerControlImpl getHornetQServerControl() { return messagingServerControl; @@ -761,16 +747,6 @@ postOffice.removeBinding(queueName); } - public synchronized void registerActivateCallback(final ActivateCallback callback) - { - activateCallbacks.add(callback); - } - - public synchronized void unregisterActivateCallback(final ActivateCallback callback) - { - activateCallbacks.remove(callback); - } - public ExecutorFactory getExecutorFactory() { return executorFactory; @@ -790,92 +766,31 @@ return new PagingManagerImpl(new PagingStoreFactoryNIO(configuration.getPagingDirectory(), executorFactory), storageManager, addressSettingsRepository, - configuration.isJournalSyncNonTransactional(), - configuration.isBackup()); + configuration.isJournalSyncNonTransactional()); } // Private // -------------------------------------------------------------------------------------- - private synchronized void callActivateCallbacks() + private boolean checkActivate() throws Exception { - for (ActivateCallback callback : activateCallbacks) - { - callback.activated(); - } - } - - private void checkActivate(final RemotingConnection connection) throws Exception - { if (configuration.isBackup()) { - log.info("A connection has been made to the backup server so it will be activated! This will result in the live server being considered failed."); - - synchronized (this) + //Handle backup server activation + + if (configuration.isSharedStore()) { - freezeBackupConnection(); - - List<Queue> toActivate = postOffice.activate(); - - for (Queue queue : toActivate) - { - scheduledPool.schedule(new ActivateRunner(queue), - configuration.getQueueActivationTimeout(), - TimeUnit.MILLISECONDS); - } - + //load shared store + configuration.setBackup(false); - - if (clusterManager != null) - { - clusterManager.activate(); - } - - if (configuration.isFileDeploymentEnabled()) - { - queueDeployer = new QueueDeployer(deploymentManager, messagingServerControl); - - queueDeployer.start(); - } } - } - - connection.activate(); - } - - // We need to prevent any more packets being handled on replicating connection as soon as first live connection - // is created or re-attaches, to prevent a situation like the following: - // connection 1 create queue A - // connection 2 fails over - // A gets activated since no consumers - // connection 1 create consumer on A - // connection 1 delivery - // connection 1 delivery gets replicated - // can't find message in queue since active was delivered immediately - private void freezeBackupConnection() - { - // Sanity check - // All replicated sessions should be on the same connection - RemotingConnection replConnection = null; - - for (ServerSession session : sessions.values()) - { - RemotingConnection rc = session.getChannel().getConnection(); - - if (replConnection == null) + else { - replConnection = rc; + //just load journal } - else if (replConnection != rc) - { - throw new IllegalStateException("More than one replicating connection!"); - } } - - if (replConnection != null) - { - replConnection.freeze(); - } + + return true; } private void initialisePart1() throws Exception @@ -948,15 +863,13 @@ resourceManager = new ResourceManagerImpl((int)(configuration.getTransactionTimeout() / 1000), configuration.getTransactionTimeoutScanPeriod(), scheduledPool); - postOffice = new PostOfficeImpl(this, - storageManager, + postOffice = new PostOfficeImpl(storageManager, pagingManager, queueFactory, managementService, configuration.getMessageExpiryScanPeriod(), configuration.getMessageExpiryThreadPriority(), - configuration.isWildcardRoutingEnabled(), - configuration.isBackup(), + configuration.isWildcardRoutingEnabled(), configuration.getIDCacheSize(), configuration.isPersistIDCache(), executorFactory, @@ -1022,28 +935,16 @@ // Deploy any predefined queues - // We don't activate queue deployer on the backup - all queues deployed on live are deployed on backup by - // replicating them - if (configuration.isFileDeploymentEnabled() && !configuration.isBackup()) + if (configuration.isFileDeploymentEnabled()) { queueDeployer = new QueueDeployer(deploymentManager, messagingServerControl); queueDeployer.start(); } - // We need to call this here, this gives any dependent server a chance to deploy its own destinations - // this needs to be done before clustering is initialised, and in the same order on live and backup - callActivateCallbacks(); - // Deply any pre-defined diverts deployDiverts(); - // Set-up the replicating connection - if (!setupReplicatingConnection()) - { - return; - } - if (configuration.isClustered()) { // This can't be created until node id is set @@ -1053,8 +954,7 @@ scheduledPool, managementService, configuration, - uuid, - replicatingChannel, + uuid, configuration.isBackup()); clusterManager.start(); @@ -1068,7 +968,9 @@ pagingManager.resumeDepages(); final ServerInfo dumper = new ServerInfo(this, pagingManager); + long dumpInfoInterval = configuration.getServerDumpInterval(); + if (dumpInfoInterval > 0) { scheduledPool.scheduleWithFixedDelay(new Runnable() @@ -1079,6 +981,7 @@ } }, 0, dumpInfoInterval, TimeUnit.MILLISECONDS); } + initialised = true; started = true; @@ -1094,89 +997,7 @@ config.isDurable()); } } - - public Channel getReplicatingChannel() - { - return replicatingChannel; - } - - private boolean setupReplicatingConnection() throws Exception - { - String backupConnectorName = configuration.getBackupConnectorName(); - - if (backupConnectorName != null) - { - TransportConfiguration backupConnector = configuration.getConnectorConfigurations().get(backupConnectorName); - - if (backupConnector == null) - { - log.warn("connector with name '" + backupConnectorName + "' is not defined in the configuration."); - } - else - { - replicatingConnectionManager = new ConnectionManagerImpl(null, - backupConnector, - null, - false, - 1, - ClientSessionFactoryImpl.DEFAULT_CALL_TIMEOUT, - ClientSessionFactoryImpl.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, - ClientSessionFactoryImpl.DEFAULT_CONNECTION_TTL, - 0, - 1.0d, - 0, - threadPool, - scheduledPool); - - replicatingConnection = replicatingConnectionManager.getConnection(1); - - if (replicatingConnection != null) - { - replicatingChannel = replicatingConnection.getChannel(2, -1, false); - - replicatingConnection.addFailureListener(new FailureListener() - { - public void connectionFailed(HornetQException me) - { - replicatingChannel.executeOutstandingDelayedResults(); - } - }); - - // First time we get channel we send a message down it informing the backup of our node id - - // backup and live must have the same node id - - Packet packet = new ReplicateStartupInfoMessage(uuid, storageManager.getCurrentUniqueID()); - - final Future future = new Future(); - - replicatingChannel.replicatePacket(packet, 1, new Runnable() - { - public void run() - { - future.run(); - } - }); - - // This may take a while especially if the journal is large - boolean ok = future.await(60000); - - if (!ok) - { - throw new IllegalStateException("Timed out waiting for response from backup for initialisation"); - } - } - else - { - log.warn("Backup server MUST be started before live server. Initialisation will not proceed."); - - return false; - } - } - } - - return true; - } - + private void loadJournal() throws Exception { List<QueueBindingInfo> queueBindingInfos = new ArrayList<QueueBindingInfo>(); @@ -1369,95 +1190,6 @@ } } - private CreateSessionResponseMessage doCreateSession(final String name, - final long channelID, - final long oppositeChannelID, - final String username, - final String password, - final int minLargeMessageSize, - final int incrementingVersion, - final RemotingConnection connection, - final boolean autoCommitSends, - final boolean autoCommitAcks, - final boolean preAcknowledge, - final boolean xa, - final int sendWindowSize, - final boolean backup) throws Exception - { - if (version.getIncrementingVersion() != incrementingVersion) - { - log.warn("Client with version " + incrementingVersion + - " and address " + - connection.getRemoteAddress() + - " is not compatible with server version " + - version.getFullVersion() + - ". " + - "Please ensure all clients and servers are upgraded to the same version for them to " + - "interoperate properly"); - return null; - } - - // Is this comment relevant any more ? - - // Authenticate. Successful autentication will place a new SubjectContext - // on thread local, - // which will be used in the authorization process. However, we need to - // make sure we clean - // up thread local immediately after we used the information, otherwise - // some other people - // security my be screwed up, on account of thread local security stack - // being corrupted. - - securityStore.authenticate(username, password); - - ServerSession currentSession = sessions.remove(name); - - if (currentSession != null) - { - // This session may well be on a different connection and different channel id, so we must get rid - // of it and create another - currentSession.getChannel().close(); - } - - Channel channel = connection.getChannel(channelID, sendWindowSize, false); - - Channel replicatingChannel = getReplicatingChannel(); - - final ServerSessionImpl session = new ServerSessionImpl(name, - oppositeChannelID, - username, - password, - minLargeMessageSize, - autoCommitSends, - autoCommitAcks, - preAcknowledge, - configuration.isPersistDeliveryCountBeforeDelivery(), - xa, - connection, - storageManager, - postOffice, - resourceManager, - securityStore, - executorFactory.getExecutor(), - channel, - managementService, - queueFactory, - this, - configuration.getManagementAddress(), - replicatingChannel, - backup); - - sessions.put(name, session); - - ServerSessionPacketHandler handler = new ServerSessionPacketHandler(session); - - session.setHandler(handler); - - channel.setHandler(handler); - - return new CreateSessionResponseMessage(version.getIncrementingVersion()); - } - private Transformer instantiateTransformer(final String transformerClassName) { Transformer transformer = null; @@ -1480,21 +1212,5 @@ } // Inner classes - // -------------------------------------------------------------------------------- - - private class ActivateRunner implements Runnable - { - private Queue queue; - - ActivateRunner(final Queue queue) - { - this.queue = queue; - } - - public void run() - { - queue.activateNow(threadPool); - } - } - + // -------------------------------------------------------------------------------- } Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/QueueImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/QueueImpl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/QueueImpl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -46,7 +46,6 @@ import org.hornetq.core.persistence.StorageManager; import org.hornetq.core.postoffice.Bindings; import org.hornetq.core.postoffice.PostOffice; -import org.hornetq.core.remoting.Channel; import org.hornetq.core.server.Consumer; import org.hornetq.core.server.Distributor; import org.hornetq.core.server.HandleStatus; @@ -127,10 +126,6 @@ private final ScheduledExecutorService scheduledExecutor; - private volatile boolean backup; - - private int consumersToFailover = -1; - private SimpleString address; private Redistributor redistributor; @@ -146,7 +141,7 @@ private final Map<Consumer, Iterator<MessageReference>> iterators = new HashMap<Consumer, Iterator<MessageReference>>(); private ConcurrentMap<SimpleString, Consumer> groups = new ConcurrentHashMap<SimpleString, Consumer>(); - + private volatile SimpleString expiryAddress; public QueueImpl(final long persistenceID, @@ -192,7 +187,7 @@ direct = true; scheduledDeliveryHandler = new ScheduledDeliveryHandlerImpl(scheduledExecutor); - + if (addressSettingsRepository != null) { expiryAddress = addressSettingsRepository.getMatch(address.toString()).getExpiryAddress(); @@ -202,7 +197,7 @@ expiryAddress = null; } } - + // Bindable implementation ------------------------------------------------------------------------------------- public SimpleString getRoutingName() @@ -352,11 +347,6 @@ public void lockDelivery() { - if (backup) - { - return; - } - try { lock.acquire(); @@ -369,11 +359,6 @@ public void unlockDelivery() { - if (backup) - { - return; - } - lock.release(); } @@ -471,7 +456,7 @@ return removed; } - public synchronized void addRedistributor(final long delay, final Executor executor, final Channel replicatingChannel) + public synchronized void addRedistributor(final long delay, final Executor executor) { if (future != null) { @@ -490,7 +475,7 @@ { if (consumers.size() == 0) { - DelayedAddRedistributor dar = new DelayedAddRedistributor(executor, replicatingChannel); + DelayedAddRedistributor dar = new DelayedAddRedistributor(executor); future = scheduledExecutor.schedule(dar, delay, TimeUnit.MILLISECONDS); @@ -499,7 +484,7 @@ } else { - internalAddRedistributor(executor, replicatingChannel); + internalAddRedistributor(executor); } } @@ -743,26 +728,26 @@ { if (checkDLQ(reference)) { - if (!scheduledDeliveryHandler.checkAndSchedule(reference, backup)) + if (!scheduledDeliveryHandler.checkAndSchedule(reference)) { messageReferences.addFirst(reference, reference.getMessage().getPriority()); } } - } + } public void expire(final MessageReference ref) throws Exception - { + { log.info("expiring ref " + this.expiryAddress); if (expiryAddress != null) { - move(expiryAddress, ref, true); + move(expiryAddress, ref, true); } else - { + { acknowledge(ref); } } - + public void setExpiryAddress(final SimpleString expiryAddress) { this.expiryAddress = expiryAddress; @@ -1003,69 +988,6 @@ return false; } - public boolean isBackup() - { - return backup; - } - - public synchronized void setBackup() - { - backup = true; - - direct = false; - } - - public synchronized boolean activate() - { - consumersToFailover = consumers.size(); - - if (consumersToFailover == 0) - { - backup = false; - - return true; - } - else - { - return false; - } - } - - public synchronized void activateNow(final Executor executor) - { - if (backup) - { - log.info("Timed out waiting for all consumers to reconnect to queue " + name + - " so queue will be activated now"); - - backup = false; - - scheduledDeliveryHandler.reSchedule(); - - deliverAsync(executor); - } - } - - public synchronized boolean consumerFailedOver() - { - consumersToFailover--; - - if (consumersToFailover == 0) - { - // All consumers for the queue have failed over, can re-activate it now - - backup = false; - - scheduledDeliveryHandler.reSchedule(); - - return true; - } - else - { - return false; - } - } - // Public // ----------------------------------------------------------------------------- @@ -1097,17 +1019,12 @@ // Private // ------------------------------------------------------------------------------ - private void internalAddRedistributor(final Executor executor, final Channel replicatingChannel) + private void internalAddRedistributor(final Executor executor) { // create the redistributor only once if there are no local consumers if (consumers.size() == 0 && redistributor == null) { - redistributor = new Redistributor(this, - storageManager, - postOffice, - executor, - REDISTRIBUTOR_BATCH_SIZE, - replicatingChannel); + redistributor = new Redistributor(this, storageManager, postOffice, executor, REDISTRIBUTOR_BATCH_SIZE); distributionPolicy.addConsumer(redistributor); @@ -1287,16 +1204,6 @@ */ private synchronized void deliver() { - // We don't do actual delivery if the queue is on a backup node - this is - // because it's async and could get out of step - // with the live node. Instead, when we replicate the delivery we remove - // the ref from the queue - - if (backup) - { - return; - } - direct = false; if (distributionPolicy.getConsumerCount() == 0) @@ -1310,11 +1217,11 @@ Iterator<MessageReference> iterator = null; - //TODO - this needs to be optimised!! Creating too much stuff on an inner loop + // TODO - this needs to be optimised!! Creating too much stuff on an inner loop int totalConsumers = distributionPolicy.getConsumerCount(); Set<Consumer> busyConsumers = new HashSet<Consumer>(); Set<Consumer> nullReferences = new HashSet<Consumer>(); - + while (true) { consumer = distributionPolicy.getNextConsumer(); @@ -1334,7 +1241,7 @@ else { reference = null; - + if (consumer.getFilter() != null) { // we have iterated on the whole queue for @@ -1347,7 +1254,7 @@ if (reference == null) { - nullReferences.add(consumer); + nullReferences.add(consumer); if (nullReferences.size() + busyConsumers.size() == totalConsumers) { startDepaging(); @@ -1361,10 +1268,10 @@ else { nullReferences.remove(consumer); - + if (reference.getMessage().isExpired()) { - //We expire messages on the server too + // We expire messages on the server too if (iterator == null) { messageReferences.removeFirst(); @@ -1373,9 +1280,9 @@ { iterator.remove(); } - + referenceHandled(); - + try { expire(reference); @@ -1384,7 +1291,7 @@ { log.error("Failed to expire ref", e); } - + continue; } } @@ -1443,14 +1350,14 @@ messagesAdded.incrementAndGet(); } - if (scheduledDeliveryHandler.checkAndSchedule(ref, backup)) + if (scheduledDeliveryHandler.checkAndSchedule(ref)) { return; } boolean add = false; - if (direct && !backup) + if (direct) { // Deliver directly @@ -1670,7 +1577,7 @@ if (message.decrementRefCount() == 0 && store != null) { - store.addSize(-ref.getMessage().getMemoryEstimate()); + store.addSize(-ref.getMessage().getMemoryEstimate()); } } @@ -1682,7 +1589,7 @@ { ServerMessage msg = ref.getMessage(); - if (!scheduledDeliveryHandler.checkAndSchedule(ref, backup)) + if (!scheduledDeliveryHandler.checkAndSchedule(ref)) { messageReferences.addFirst(ref, msg.getPriority()); } @@ -1871,20 +1778,16 @@ { private final Executor executor; - private final Channel replicatingChannel; - - DelayedAddRedistributor(final Executor executor, final Channel replicatingChannel) + DelayedAddRedistributor(final Executor executor) { this.executor = executor; - - this.replicatingChannel = replicatingChannel; } public void run() { synchronized (QueueImpl.this) { - internalAddRedistributor(executor, replicatingChannel); + internalAddRedistributor(executor); futures.remove(this); } Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/ScheduledDeliveryHandlerImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/ScheduledDeliveryHandlerImpl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/ScheduledDeliveryHandlerImpl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -49,7 +49,7 @@ this.scheduledExecutor = scheduledExecutor; } - public boolean checkAndSchedule(final MessageReference ref, final boolean backup) + public boolean checkAndSchedule(final MessageReference ref) { long deliveryTime = ref.getScheduledDeliveryTime(); @@ -67,10 +67,7 @@ scheduledRunnables.put(ref.getMessage().getMessageID(), runnable); } - if (!backup) - { - scheduleDelivery(runnable, deliveryTime); - } + scheduleDelivery(runnable, deliveryTime); return true; } Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -34,10 +34,8 @@ import org.hornetq.core.postoffice.Binding; import org.hornetq.core.postoffice.QueueBinding; import org.hornetq.core.remoting.Channel; -import org.hornetq.core.remoting.Packet; import org.hornetq.core.remoting.impl.wireformat.SessionReceiveContinuationMessage; import org.hornetq.core.remoting.impl.wireformat.SessionReceiveMessage; -import org.hornetq.core.remoting.impl.wireformat.replication.SessionReplicateDeliveryMessage; import org.hornetq.core.remoting.spi.HornetQBuffer; import org.hornetq.core.server.HandleStatus; import org.hornetq.core.server.LargeServerMessage; @@ -78,8 +76,6 @@ private final long id; - private final long replicatedSessionID; - private final Queue messageQueue; private final Filter filter; @@ -119,8 +115,6 @@ private final Channel channel; - private final Channel replicatingChannel; - private volatile boolean closed; private final boolean preAcknowledge; @@ -132,7 +126,6 @@ // Constructors --------------------------------------------------------------------------------- public ServerConsumerImpl(final long id, - final long replicatedSessionID, final ServerSession session, final QueueBinding binding, final Filter filter, @@ -141,7 +134,6 @@ final StorageManager storageManager, final PagingManager pagingManager, final Channel channel, - final Channel replicatingChannel, final boolean preAcknowledge, final boolean updateDeliveries, final Executor executor, @@ -149,8 +141,6 @@ { this.id = id; - this.replicatedSessionID = replicatedSessionID; - this.filter = filter; this.session = session; @@ -169,8 +159,6 @@ this.channel = channel; - this.replicatingChannel = replicatingChannel; - this.preAcknowledge = preAcknowledge; this.pagingManager = pagingManager; @@ -180,7 +168,7 @@ this.minLargeMessageSize = session.getMinLargeMessageSize(); this.updateDeliveries = updateDeliveries; - + if (browseOnly) { browserDeliverer = new BrowserDeliverer(messageQueue.iterator()); @@ -222,7 +210,7 @@ { messageQueue.removeConsumer(this); } - + session.removeConsumer(this); LinkedList<MessageReference> refs = cancelRefs(false, null); @@ -371,8 +359,6 @@ id + ", messageId = " + messageID + - " backup = " + - messageQueue.isBackup() + " queue = " + messageQueue.getName() + " closed = " + @@ -397,7 +383,7 @@ { return null; } - + // Expiries can come in out of sequence with respect to delivery order Iterator<MessageReference> iter = deliveringRefs.iterator(); @@ -421,70 +407,6 @@ return ref; } - public void deliverReplicated(final long messageID) throws Exception - { - MessageReference ref = messageQueue.removeFirstReference(messageID); - - if (ref == null) - { - // The order is correct, but it hasn't been depaged yet, so we need to force a depage - PagingStore store = pagingManager.getPageStore(binding.getAddress()); - - // force a depage - if (!store.readPage()) // This returns false if there are no pages - { - throw new IllegalStateException("Cannot find Reference[" + messageID + - "] in queue " + - messageQueue.getName()); - } - else - { - ref = messageQueue.removeFirstReference(messageID); - - if (ref == null) - { - throw new IllegalStateException("Cannot find Reference[" + messageID + - "] after depaging on Queue " + - messageQueue.getName()); - } - } - } - - // We call doHandle rather than handle, since we don't want to check available credits - // This is because delivery and receive credits can be processed in different order on live - // and backup, and otherwise we could have a situation where the delivery is replicated - // but the credits haven't arrived yet, so the delivery gets rejected on backup - HandleStatus handled = doHandle(ref); - - if (handled != HandleStatus.HANDLED) - { - throw new IllegalStateException("Reference " + ref + - " was not handled on backup node, handleStatus = " + - handled); - } - } - - public void failedOver() - { - if (messageQueue.consumerFailedOver()) - { - if (started) - { - promptDelivery(); - } - } - } - - public void lock() - { - lock.lock(); - } - - public void unlock() - { - lock.unlock(); - } - // Public --------------------------------------------------------------------------------------- /** To be used on tests only */ @@ -533,23 +455,14 @@ */ private void resumeLargeMessage() { - if (messageQueue.isBackup()) - { - // We need to deliver the largeMessage on backup also, exactly as done on the live node. - // In case of failure the same packets will be available for resume sending. - largeMessageDeliverer.deliver(); - } - else - { - executor.execute(resumeLargeMessageRunnable); - } + executor.execute(resumeLargeMessageRunnable); } private HandleStatus doHandle(final MessageReference ref) throws Exception { if (availableCredits != null && availableCredits.get() <= 0) { - // log.info("busy - available credits is " + availableCredits.get()); + // log.info("busy - available credits is " + availableCredits.get()); return HandleStatus.BUSY; } @@ -571,13 +484,6 @@ // This has to be checked inside the lock as the set to null is done inside the lock if (pendingLargeMessagesCounter.get() > 0) { - if (messageQueue.isBackup()) - { - log.warn("doHandle: rejecting message while send is pending, ignoring reference = " + ref + - " backup = " + - messageQueue.isBackup()); - } - return HandleStatus.BUSY; } @@ -646,40 +552,9 @@ final LargeMessageDeliverer localDeliverer = new LargeMessageDeliverer((LargeServerMessage)message, ref); - if (replicatingChannel == null) - { - // it doesn't need lock because deliverLargeMesasge is already inside the lock() - largeMessageDeliverer = localDeliverer; - largeMessageDeliverer.deliver(); - } - else - { - Packet replPacket = new SessionReplicateDeliveryMessage(id, message.getMessageID()); - replPacket.setChannelID(channel.getID()); - - replicatingChannel.replicatePacket(replPacket, replicatedSessionID, new Runnable() - { - public void run() - { - // setting & unsetting largeMessageDeliver is done inside the lock, - // so this needs to be locked - lock.lock(); - try - { - largeMessageDeliverer = localDeliverer; - if (largeMessageDeliverer.deliver()) - { - promptDelivery(); - } - } - finally - { - lock.unlock(); - } - } - }); - } - + // it doesn't need lock because deliverLargeMesasge is already inside the lock() + largeMessageDeliverer = localDeliverer; + largeMessageDeliverer.deliver(); } /** @@ -699,33 +574,7 @@ } } - if (replicatingChannel == null) - { - // Not replicated - just send now - - if (trace) - { - log.trace("delivering Message " + ref + " on backup"); - } - channel.send(packet); - } - else - { - Packet replPacket = new SessionReplicateDeliveryMessage(id, message.getMessageID()); - replPacket.setChannelID(channel.getID()); - - replicatingChannel.replicatePacket(replPacket, replicatedSessionID, new Runnable() - { - public void run() - { - if (trace) - { - log.trace("delivering Message " + ref + " on live"); - } - channel.send(packet); - } - }); - } + channel.send(packet); } // Inner classes @@ -849,7 +698,7 @@ { if (trace) { - trace("deliverLargeMessage: Leaving loop of send LargeMessage because of credits, backup = " + messageQueue.isBackup()); + trace("deliverLargeMessage: Leaving loop of send LargeMessage because of credits"); } return false; } @@ -870,9 +719,7 @@ { trace("deliverLargeMessage: Sending " + chunk.getRequiredBufferSize() + " availableCredits now is " + - availableCredits + - " isBackup = " + - messageQueue.isBackup()); + availableCredits); } channel.send(chunk); @@ -887,7 +734,7 @@ if (trace) { - trace("Finished deliverLargeMessage isBackup = " + messageQueue.isBackup()); + trace("Finished deliverLargeMessage"); } close(); @@ -997,18 +844,18 @@ return chunk; } } - + private class BrowserDeliverer implements Runnable { private MessageReference current = null; - + public BrowserDeliverer(final Iterator<MessageReference> iterator) { this.iterator = iterator; } private final Iterator<MessageReference> iterator; - + public void run() { // if the reference was busy during the previous iteration, handle it now @@ -1020,7 +867,7 @@ if (status == HandleStatus.BUSY) { return; - } + } } catch (Exception e) { @@ -1050,6 +897,6 @@ } } } - + } } Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -52,7 +52,6 @@ import org.hornetq.core.remoting.impl.wireformat.HornetQExceptionMessage; import org.hornetq.core.remoting.impl.wireformat.NullResponseMessage; import org.hornetq.core.remoting.impl.wireformat.PacketImpl; -import org.hornetq.core.remoting.impl.wireformat.PacketsConfirmedMessage; import org.hornetq.core.remoting.impl.wireformat.RollbackMessage; import org.hornetq.core.remoting.impl.wireformat.SessionAcknowledgeMessage; import org.hornetq.core.remoting.impl.wireformat.SessionBindingQueryMessage; @@ -80,7 +79,6 @@ import org.hornetq.core.remoting.impl.wireformat.SessionXASetTimeoutMessage; import org.hornetq.core.remoting.impl.wireformat.SessionXASetTimeoutResponseMessage; import org.hornetq.core.remoting.impl.wireformat.SessionXAStartMessage; -import org.hornetq.core.remoting.impl.wireformat.replication.SessionReplicateDeliveryMessage; import org.hornetq.core.remoting.spi.HornetQBuffer; import org.hornetq.core.security.CheckType; import org.hornetq.core.security.SecurityStore; @@ -114,23 +112,12 @@ private static final Logger log = Logger.getLogger(ServerSessionImpl.class); - private static final boolean trace = log.isTraceEnabled(); - - private static void trace(String message) - { - log.trace(message); - } - // Static ------------------------------------------------------------------------------- // Attributes ---------------------------------------------------------------------------- - private volatile long id; + private final long id; - // The id of the opposite channel - i.e. if this is the live then its the id on the backup, if this is the backup - // then it's the id on the live, or -1 if non clustered - private volatile long oppositeChannelID; - private final String username; private final String password; @@ -147,8 +134,6 @@ private RemotingConnection remotingConnection; - private Channel replicatingChannel; - private final Map<Long, ServerConsumer> consumers = new ConcurrentHashMap<Long, ServerConsumer>(); private final Executor executor; @@ -183,8 +168,6 @@ private final SimpleString nodeID; - private boolean backup; - // The current currentLargeMessage being processed // In case of replication, currentLargeMessage should only be accessed within the replication callbacks private volatile LargeServerMessage currentLargeMessage; @@ -193,8 +176,7 @@ // Constructors --------------------------------------------------------------------------------- - public ServerSessionImpl(final String name, - final long oppositeChannelID, + public ServerSessionImpl(final String name, final String username, final String password, final int minLargeMessageSize, @@ -213,14 +195,10 @@ final ManagementService managementService, final QueueFactory queueFactory, final HornetQServer server, - final SimpleString managementAddress, - final Channel replicatingChannel, - final boolean backup) throws Exception + final SimpleString managementAddress) throws Exception { this.id = channel.getID(); - this.oppositeChannelID = oppositeChannelID; - this.username = username; this.password = password; @@ -266,10 +244,6 @@ this.nodeID = server.getNodeID(); - this.replicatingChannel = replicatingChannel; - - this.backup = backup; - remotingConnection.addFailureListener(this); remotingConnection.addCloseListener(this); @@ -367,890 +341,6 @@ public void handleCreateConsumer(final SessionCreateConsumerMessage packet) { - if (replicatingChannel == null) - { - doHandleCreateConsumer(packet); - } - else - { - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - doHandleCreateConsumer(packet); - } - }); - } - } - - public void handleCreateQueue(final CreateQueueMessage packet) - { - if (replicatingChannel == null) - { - doHandleCreateQueue(packet); - } - else - { - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - doHandleCreateQueue(packet); - } - }); - } - } - - public void handleDeleteQueue(final SessionDeleteQueueMessage packet) - { - if (replicatingChannel == null) - { - doHandleDeleteQueue(packet); - } - else - { - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - doHandleDeleteQueue(packet); - } - }); - } - } - - public void handleExecuteQueueQuery(final SessionQueueQueryMessage packet) - { - if (replicatingChannel == null) - { - doHandleExecuteQueueQuery(packet); - } - else - { - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - doHandleExecuteQueueQuery(packet); - } - }); - } - } - - public void handleExecuteBindingQuery(final SessionBindingQueryMessage packet) - { - if (replicatingChannel == null) - { - doHandleExecuteBindingQuery(packet); - } - else - { - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - doHandleExecuteBindingQuery(packet); - } - }); - } - } - - public void handleAcknowledge(final SessionAcknowledgeMessage packet) - { - if (replicatingChannel == null) - { - doHandleAcknowledge(packet); - } - else - { - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - doHandleAcknowledge(packet); - } - }); - } - } - - public void handleExpired(final SessionExpiredMessage packet) - { - if (replicatingChannel == null) - { - doHandleExpired(packet); - } - else - { - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - doHandleExpired(packet); - } - }); - } - } - - public void handleCommit(final Packet packet) - { - if (replicatingChannel == null) - { - doHandleCommit(packet); - } - else - { - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - doHandleCommit(packet); - } - }); - } - } - - public void handleRollback(final RollbackMessage packet) - { - - if (replicatingChannel == null) - { - doHandleRollback(packet); - } - else - { - final HashSet<Queue> queues = lockUsedQueues(null); - - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - try - { - doHandleRollback(packet); - } - finally - { - for (Queue queue : queues) - { - queue.unlockDelivery(); - } - } - } - }); - } - } - - public void handleXACommit(final SessionXACommitMessage packet) - { - if (replicatingChannel == null) - { - doHandleXACommit(packet); - } - else - { - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - doHandleXACommit(packet); - } - }); - } - } - - public void handleXAEnd(final SessionXAEndMessage packet) - { - if (replicatingChannel == null) - { - doHandleXAEnd(packet); - } - else - { - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - doHandleXAEnd(packet); - } - }); - } - } - - public void handleXAForget(final SessionXAForgetMessage packet) - { - if (replicatingChannel == null) - { - doHandleXAForget(packet); - } - else - { - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - doHandleXAForget(packet); - } - }); - } - } - - public void handleXAJoin(final SessionXAJoinMessage packet) - { - if (replicatingChannel == null) - { - doHandleXAJoin(packet); - } - else - { - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - doHandleXAJoin(packet); - } - }); - } - } - - public void handleXAResume(final SessionXAResumeMessage packet) - { - if (replicatingChannel == null) - { - doHandleXAResume(packet); - } - else - { - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - doHandleXAResume(packet); - } - }); - } - } - - public void handleXARollback(final SessionXARollbackMessage packet) - { - - if (replicatingChannel == null) - { - doHandleXARollback(packet); - } - else - { - final Set<Queue> queues = lockUsedQueues(packet.getXid()); - - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - try - { - doHandleXARollback(packet); - } - finally - { - for (Queue queue : queues) - { - queue.unlockDelivery(); - } - } - } - }); - } - } - - public void handleXAStart(final SessionXAStartMessage packet) - { - if (replicatingChannel == null) - { - doHandleXAStart(packet); - } - else - { - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - doHandleXAStart(packet); - } - }); - } - } - - public void handleXASuspend(final Packet packet) - { - if (replicatingChannel == null) - { - doHandleXASuspend(packet); - } - else - { - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - doHandleXASuspend(packet); - } - }); - } - } - - public void handleXAPrepare(final SessionXAPrepareMessage packet) - { - if (replicatingChannel == null) - { - doHandleXAPrepare(packet); - } - else - { - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - doHandleXAPrepare(packet); - } - }); - } - } - - public void handleGetInDoubtXids(final Packet packet) - { - if (replicatingChannel == null) - { - doHandleGetInDoubtXids(packet); - } - else - { - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - doHandleGetInDoubtXids(packet); - } - }); - } - } - - public void handleGetXATimeout(final Packet packet) - { - if (replicatingChannel == null) - { - doHandleGetXATimeout(packet); - } - else - { - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - doHandleGetXATimeout(packet); - } - }); - } - } - - public void handleSetXATimeout(final SessionXASetTimeoutMessage packet) - { - if (replicatingChannel == null) - { - doHandleSetXATimeout(packet); - } - else - { - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - doHandleSetXATimeout(packet); - } - }); - } - } - - private void lockConsumers() - { - for (ServerConsumer consumer : consumers.values()) - { - consumer.lock(); - } - } - - private void unlockConsumers() - { - for (ServerConsumer consumer : consumers.values()) - { - consumer.unlock(); - } - } - - public void handleStart(final Packet packet) - { - if (replicatingChannel == null) - { - setStarted(true); - - channel.confirm(packet); - } - else - { - lockConsumers(); - - try - { - setStarted(true); - - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - // setStarted(true); - - channel.confirm(packet); - } - }); - } - finally - { - unlockConsumers(); - } - } - } - - public void handleStop(final Packet packet) - { - final Packet response = new NullResponseMessage(); - - // We need to stop the session *before* replicating, to avoid this situation: - // session stop, - // stop is replicated - // delivery occurs, local is not stopped yet - // stop is processed on backup - // backup is stopped - // delivery is processed on backup - // it's stopped so barfs and cannot process delivery - - if (replicatingChannel == null) - { - setStarted(false); - - channel.confirm(packet); - - channel.send(response); - } - else - { - lockConsumers(); - - try - { - - setStarted(false); - - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - channel.confirm(packet); - - channel.send(response); - - } - }); - } - finally - { - unlockConsumers(); - } - } - } - - public void handleFailedOver(final Packet packet) - { - Set<ServerConsumer> consumersClone = new HashSet<ServerConsumer>(consumers.values()); - - for (ServerConsumer consumer : consumersClone) - { - consumer.failedOver(); - } - } - - public void handleClose(final Packet packet) - { - - if (replicatingChannel == null) - { - doHandleClose(packet); - } - else - { - final HashSet<Queue> queues = lockUsedQueues(null); - - // We need to stop the consumers first before replicating, to ensure no deliveries occur after this, - // but we need to process the actual close() when the replication response returns, otherwise things - // can happen like acks can come in after close - - for (ServerConsumer consumer : consumers.values()) - { - consumer.setStarted(false); - } - - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - try - { - doHandleClose(packet); - } - finally - { - for (Queue queue : queues) - { - queue.unlockDelivery(); - } - } - } - }); - } - } - - public void handleCloseConsumer(final SessionConsumerCloseMessage packet) - { - final ServerConsumer consumer = consumers.get(packet.getConsumerID()); - - consumer.setStarted(false); - - if (replicatingChannel == null) - { - doHandleCloseConsumer(packet, consumer); - } - else - { - final Queue queue; - - if (consumer.getCountOfPendingDeliveries() > 0) - { - queue = consumer.getQueue(); - queue.lockDelivery(); - } - else - { - queue = null; - } - - // We need to stop the consumer first before replicating, to ensure no deliveries occur after this, - // but we need to process the actual close() when the replication response returns, otherwise things - // can happen like acks can come in after close - - consumer.setStarted(false); - - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - try - { - doHandleCloseConsumer(packet, consumer); - } - finally - { - if (queue != null) - { - queue.unlockDelivery(); - } - } - } - }); - } - } - - public void handleReceiveConsumerCredits(final SessionConsumerFlowCreditMessage packet) - { - if (replicatingChannel == null) - { - doReceiveCredits(packet); - } - else - { - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - doReceiveCredits(packet); - } - }); - } - } - - public void handleSendLargeMessage(final SessionSendLargeMessage packet) - { - // need to create the LargeMessage before continue - final LargeServerMessage msg = doCreateLargeMessage(packet); - - if (msg == null) - { - // packet logged an error, and played with channel.returns... and nothing needs to be done now - return; - } - - // With a send we must make sure it is replicated to backup before being processed on live - // or can end up with delivery being processed on backup before original send - - if (replicatingChannel == null) - { - if (currentLargeMessage != null) - { - log.warn("Replacing incomplete LargeMessage with ID=" + currentLargeMessage.getMessageID()); - } - - currentLargeMessage = msg; - - doSendLargeMessage(packet); - } - else - { - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - if (trace) - { - trace("(Replication) SendLargeMessage, id=" + msg.getMessageID()); - } - - if (currentLargeMessage != null) - { - log.warn("Replacing incomplete LargeMessage with ID=" + currentLargeMessage.getMessageID()); - } - - currentLargeMessage = msg; - - doSendLargeMessage(packet); - } - }); - } - } - - public void handleSend(final SessionSendMessage packet) - { - if (replicatingChannel == null) - { - doSend(packet); - } - else - { - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - doSend(packet); - } - }); - } - } - - public void handleSendContinuations(final SessionSendContinuationMessage packet) - { - if (replicatingChannel == null) - { - doSendContinuations(packet); - } - else - { - replicatingChannel.replicatePacket(packet, oppositeChannelID, new Runnable() - { - public void run() - { - doSendContinuations(packet); - } - }); - } - } - - public void handleReplicatedDelivery(final SessionReplicateDeliveryMessage packet) - { - ServerConsumer consumer = consumers.get(packet.getConsumerID()); - - if (consumer == null) - { - throw new IllegalStateException("Cannot handle replicated delivery, consumer is closed " + packet.getConsumerID() + - " session " + - id + - " " + - System.identityHashCode(this)); - } - - try - { - consumer.deliverReplicated(packet.getMessageID()); - } - catch (Exception e) - { - log.error("Failed to handle replicated delivery", e); - } - } - - public void handlePacketsConfirmed(final PacketsConfirmedMessage packet) - { - if (replicatingChannel != null) - { - replicatingChannel.replicatePacket(packet, oppositeChannelID, null); - } - } - - public int transferConnection(final RemotingConnection newConnection, final int lastReceivedCommandID) - { - boolean wasStarted = this.started; - - if (wasStarted) - { - this.setStarted(false); - } - - if (backup) - { - // Put the id back to the original client session id - this.id = this.oppositeChannelID; - - this.oppositeChannelID = -1; - - backup = false; - } - - remotingConnection.removeFailureListener(this); - remotingConnection.removeCloseListener(this); - - // Note. We do not destroy the replicating connection here. In the case the live server has really crashed - // then the connection will get cleaned up anyway when the server ping timeout kicks in. - // In the case the live server is really still up, i.e. a split brain situation (or in tests), then closing - // the replicating connection will cause the outstanding responses to be be replayed on the live server, - // if these reach the client who then subsequently fails over, on reconnection to backup, it will have - // received responses that the backup did not know about. - - channel.transferConnection(newConnection, this.id, replicatingChannel); - - newConnection.syncIDGeneratorSequence(remotingConnection.getIDGeneratorSequence()); - - remotingConnection = newConnection; - - remotingConnection.addFailureListener(this); - remotingConnection.addCloseListener(this); - - int serverLastReceivedCommandID = channel.getLastReceivedCommandID(); - - channel.replayCommands(lastReceivedCommandID, this.id); - - if (wasStarted) - { - this.setStarted(true); - } - - return serverLastReceivedCommandID; - } - - public Channel getChannel() - { - return channel; - } - - // FailureListener implementation - // -------------------------------------------------------------------- - - public void connectionFailed(final HornetQException me) - { - try - { - log.warn("Client connection failed, clearing up resources for session " + name); - - for (Runnable runner : failureRunners) - { - try - { - runner.run(); - } - catch (Throwable t) - { - log.error("Failed to execute failure runner", t); - } - } - - handleClose(new PacketImpl(PacketImpl.SESS_CLOSE)); - - log.warn("Cleared up resources for session " + name); - } - catch (Throwable t) - { - log.error("Failed to close connection " + this); - } - } - - public void connectionClosed() - { - try - { - for (Runnable runner : failureRunners) - { - try - { - runner.run(); - } - catch (Throwable t) - { - log.error("Failed to execute failure runner", t); - } - } - } - catch (Throwable t) - { - log.error("Failed fire listeners " + this); - } - - } - - // Public - // ---------------------------------------------------------------------------- - - public Transaction getTransaction() - { - return tx; - } - - // Private - // ---------------------------------------------------------------------------- - - private void doHandleCloseConsumer(final SessionConsumerCloseMessage packet, final ServerConsumer consumer) - { - Packet response; - - try - { - consumer.close(); - - response = new NullResponseMessage(); - } - catch (Exception e) - { - log.error("Failed to close consumer", e); - - if (e instanceof HornetQException) - { - response = new HornetQExceptionMessage((HornetQException)e); - } - else - { - response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR)); - } - } - - channel.confirm(packet); - - channel.send(response); - } - - private void doHandleCreateConsumer(final SessionCreateConsumerMessage packet) - { SimpleString name = packet.getQueueName(); SimpleString filterString = packet.getFilterString(); @@ -1301,8 +391,7 @@ theQueue = (Queue)binding.getBindable(); } - ServerConsumer consumer = new ServerConsumerImpl(idGenerator.generateID(), - oppositeChannelID, + ServerConsumer consumer = new ServerConsumerImpl(idGenerator.generateID(), this, (QueueBinding)binding, filter, @@ -1310,8 +399,7 @@ browseOnly, storageManager, postOffice.getPagingManager(), - channel, - replicatingChannel, + channel, preAcknowledge, updateDeliveries, executor, @@ -1362,10 +450,9 @@ channel.confirm(packet); channel.send(response); - } - private void doHandleCreateQueue(final CreateQueueMessage packet) + public void handleCreateQueue(final CreateQueueMessage packet) { SimpleString address = packet.getAddress(); @@ -1406,7 +493,7 @@ public void run() { try - { + { if (postOffice.getBinding(name) != null) { postOffice.removeBinding(name); @@ -1439,7 +526,7 @@ channel.send(response); } - private void doHandleDeleteQueue(final SessionDeleteQueueMessage packet) + public void handleDeleteQueue(final SessionDeleteQueueMessage packet) { SimpleString name = packet.getQueueName(); @@ -1477,7 +564,7 @@ channel.send(response); } - private void doHandleExecuteQueueQuery(final SessionQueueQueryMessage packet) + public void handleExecuteQueueQuery(final SessionQueueQueryMessage packet) { SimpleString name = packet.getQueueName(); @@ -1530,7 +617,7 @@ channel.send(response); } - private void doHandleExecuteBindingQuery(final SessionBindingQueryMessage packet) + public void handleExecuteBindingQuery(final SessionBindingQueryMessage packet) { SimpleString address = packet.getAddress(); @@ -1576,7 +663,7 @@ channel.send(response); } - private void doHandleAcknowledge(final SessionAcknowledgeMessage packet) + public void handleAcknowledge(final SessionAcknowledgeMessage packet) { Packet response = null; @@ -1616,7 +703,7 @@ } } - private void doHandleExpired(final SessionExpiredMessage packet) + public void handleExpired(final SessionExpiredMessage packet) { try { @@ -1635,7 +722,7 @@ channel.confirm(packet); } - private void doHandleCommit(final Packet packet) + public void handleCommit(final Packet packet) { Packet response = null; @@ -1668,7 +755,7 @@ channel.send(response); } - private void doHandleRollback(final RollbackMessage packet) + public void handleRollback(final RollbackMessage packet) { Packet response = null; @@ -1697,7 +784,7 @@ channel.send(response); } - private void doHandleXACommit(final SessionXACommitMessage packet) + public void handleXACommit(final SessionXACommitMessage packet) { Packet response = null; @@ -1760,7 +847,7 @@ channel.send(response); } - private void doHandleXAEnd(final SessionXAEndMessage packet) + public void handleXAEnd(final SessionXAEndMessage packet) { Packet response = null; @@ -1834,7 +921,7 @@ channel.send(response); } - private void doHandleXAForget(final SessionXAForgetMessage packet) + public void handleXAForget(final SessionXAForgetMessage packet) { // Do nothing since we don't support heuristic commits / rollback from the // resource manager @@ -1846,7 +933,7 @@ channel.send(response); } - private void doHandleXAJoin(final SessionXAJoinMessage packet) + public void handleXAJoin(final SessionXAJoinMessage packet) { Packet response = null; @@ -1897,7 +984,7 @@ channel.send(response); } - private void doHandleXAResume(final SessionXAResumeMessage packet) + public void handleXAResume(final SessionXAResumeMessage packet) { Packet response = null; @@ -1959,7 +1046,7 @@ channel.send(response); } - private void doHandleXARollback(final SessionXARollbackMessage packet) + public void handleXARollback(final SessionXARollbackMessage packet) { Packet response = null; @@ -2022,7 +1109,7 @@ channel.send(response); } - private void doHandleXAStart(final SessionXAStartMessage packet) + public void handleXAStart(final SessionXAStartMessage packet) { Packet response = null; @@ -2073,7 +1160,7 @@ channel.send(response); } - private void doHandleXASuspend(final Packet packet) + public void handleXASuspend(final Packet packet) { Packet response = null; @@ -2122,7 +1209,7 @@ channel.send(response); } - private void doHandleXAPrepare(final SessionXAPrepareMessage packet) + public void handleXAPrepare(final SessionXAPrepareMessage packet) { Packet response = null; @@ -2182,7 +1269,7 @@ channel.send(response); } - private void doHandleGetInDoubtXids(final Packet packet) + public void handleGetInDoubtXids(final Packet packet) { Packet response = new SessionXAGetInDoubtXidsResponseMessage(resourceManager.getPreparedTransactions()); @@ -2191,7 +1278,7 @@ channel.send(response); } - private void doHandleGetXATimeout(final Packet packet) + public void handleGetXATimeout(final Packet packet) { Packet response = new SessionXAGetTimeoutResponseMessage(resourceManager.getTimeoutSeconds()); @@ -2200,7 +1287,7 @@ channel.send(response); } - private void doHandleSetXATimeout(final SessionXASetTimeoutMessage packet) + public void handleSetXATimeout(final SessionXASetTimeoutMessage packet) { Packet response = new SessionXASetTimeoutResponseMessage(resourceManager.setTimeoutSeconds(packet.getTimeoutSeconds())); @@ -2209,8 +1296,26 @@ channel.send(response); } - private void doHandleClose(final Packet packet) + public void handleStart(final Packet packet) { + setStarted(true); + + channel.confirm(packet); + } + + public void handleStop(final Packet packet) + { + final Packet response = new NullResponseMessage(); + + setStarted(false); + + channel.confirm(packet); + + channel.send(response); + } + + public void handleClose(final Packet packet) + { Packet response = null; try @@ -2243,45 +1348,38 @@ channel.close(); } - private void setStarted(final boolean s) + public void handleCloseConsumer(final SessionConsumerCloseMessage packet) { - Set<ServerConsumer> consumersClone = new HashSet<ServerConsumer>(consumers.values()); + final ServerConsumer consumer = consumers.get(packet.getConsumerID()); - for (ServerConsumer consumer : consumersClone) - { - consumer.setStarted(s); - } + Packet response; - started = s; - } - - /** - * We need to create the LargeMessage before replicating the packet, or else we won't know how to extract the destination, - * which is stored on the header - * @param packet - * @throws Exception - */ - private LargeServerMessage doCreateLargeMessage(final SessionSendLargeMessage packet) - { try { - return createLargeMessageStorage(packet.getLargeMessageHeader()); + consumer.close(); + + response = new NullResponseMessage(); } catch (Exception e) { - log.error("Failed to create large message", e); - Packet response = null; + log.error("Failed to close consumer", e); - channel.confirm(packet); - if (response != null) + if (e instanceof HornetQException) { - channel.send(response); + response = new HornetQExceptionMessage((HornetQException)e); } - return null; + else + { + response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR)); + } } + + channel.confirm(packet); + + channel.send(response); } - private void doReceiveCredits(final SessionConsumerFlowCreditMessage packet) + public void handleReceiveConsumerCredits(final SessionConsumerFlowCreditMessage packet) { try { @@ -2294,8 +1392,27 @@ channel.confirm(packet); } - private void doSendLargeMessage(final SessionSendLargeMessage packet) + public void handleSendLargeMessage(final SessionSendLargeMessage packet) { + // need to create the LargeMessage before continue + final LargeServerMessage msg = doCreateLargeMessage(packet); + + if (msg == null) + { + // packet logged an error, and played with channel.returns... and nothing needs to be done now + return; + } + + // With a send we must make sure it is replicated to backup before being processed on live + // or can end up with delivery being processed on backup before original send + + if (currentLargeMessage != null) + { + log.warn("Replacing incomplete LargeMessage with ID=" + currentLargeMessage.getMessageID()); + } + + currentLargeMessage = msg; + try { long id = storageManager.generateUniqueID(); @@ -2310,7 +1427,7 @@ channel.confirm(packet); } - private void doSend(final SessionSendMessage packet) + public void handleSend(final SessionSendMessage packet) { Packet response = null; @@ -2360,13 +1477,10 @@ if (response != null) { channel.send(response); - } + } } - /** - * @param packet - */ - private void doSendContinuations(final SessionSendContinuationMessage packet) + public void handleSendContinuations(final SessionSendContinuationMessage packet) { Packet response = null; @@ -2417,9 +1531,157 @@ if (response != null) { channel.send(response); + } + } + + public int transferConnection(final RemotingConnection newConnection, final int lastReceivedCommandID) + { + boolean wasStarted = this.started; + + if (wasStarted) + { + this.setStarted(false); } + + remotingConnection.removeFailureListener(this); + remotingConnection.removeCloseListener(this); + + // Note. We do not destroy the replicating connection here. In the case the live server has really crashed + // then the connection will get cleaned up anyway when the server ping timeout kicks in. + // In the case the live server is really still up, i.e. a split brain situation (or in tests), then closing + // the replicating connection will cause the outstanding responses to be be replayed on the live server, + // if these reach the client who then subsequently fails over, on reconnection to backup, it will have + // received responses that the backup did not know about. + + channel.transferConnection(newConnection); + + newConnection.syncIDGeneratorSequence(remotingConnection.getIDGeneratorSequence()); + + remotingConnection = newConnection; + + remotingConnection.addFailureListener(this); + remotingConnection.addCloseListener(this); + + int serverLastReceivedCommandID = channel.getLastReceivedCommandID(); + + channel.replayCommands(lastReceivedCommandID, this.id); + + if (wasStarted) + { + this.setStarted(true); + } + + return serverLastReceivedCommandID; } + public Channel getChannel() + { + return channel; + } + + // FailureListener implementation + // -------------------------------------------------------------------- + + public void connectionFailed(final HornetQException me) + { + try + { + log.warn("Client connection failed, clearing up resources for session " + name); + + for (Runnable runner : failureRunners) + { + try + { + runner.run(); + } + catch (Throwable t) + { + log.error("Failed to execute failure runner", t); + } + } + + handleClose(new PacketImpl(PacketImpl.SESS_CLOSE)); + + log.warn("Cleared up resources for session " + name); + } + catch (Throwable t) + { + log.error("Failed to close connection " + this); + } + } + + public void connectionClosed() + { + try + { + for (Runnable runner : failureRunners) + { + try + { + runner.run(); + } + catch (Throwable t) + { + log.error("Failed to execute failure runner", t); + } + } + } + catch (Throwable t) + { + log.error("Failed fire listeners " + this); + } + + } + + // Public + // ---------------------------------------------------------------------------- + + public Transaction getTransaction() + { + return tx; + } + + // Private + // ---------------------------------------------------------------------------- + + private void setStarted(final boolean s) + { + Set<ServerConsumer> consumersClone = new HashSet<ServerConsumer>(consumers.values()); + + for (ServerConsumer consumer : consumersClone) + { + consumer.setStarted(s); + } + + started = s; + } + + /** + * We need to create the LargeMessage before replicating the packet, or else we won't know how to extract the destination, + * which is stored on the header + * @param packet + * @throws Exception + */ + private LargeServerMessage doCreateLargeMessage(final SessionSendLargeMessage packet) + { + try + { + return createLargeMessageStorage(packet.getLargeMessageHeader()); + } + catch (Exception e) + { + log.error("Failed to create large message", e); + Packet response = null; + + channel.confirm(packet); + if (response != null) + { + channel.send(response); + } + return null; + } + } + private void handleManagementMessage(final ServerMessage message) throws Exception { try @@ -2528,40 +1790,5 @@ { postOffice.route(msg, tx); } - } - - /** - * We need to avoid delivery when rolling back while doing replication, or the backup node could be on a different order - * @return - */ - private HashSet<Queue> lockUsedQueues(Xid xid) - { - final HashSet<Queue> queues = new HashSet<Queue>(); - - for (ServerConsumer consumer : consumers.values()) - { - queues.add(consumer.getQueue()); - } - - Transaction localTX; - if (xid == null) - { - localTX = tx; - } - else - { - localTX = resourceManager.getTransaction(xid); - } - - if (localTX != null) - { - queues.addAll(localTX.getDistinctQueues()); - } - - for (Queue queue : queues) - { - queue.lockDelivery(); - } - return queues; - } + } } Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/ServerSessionPacketHandler.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/ServerSessionPacketHandler.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/core/server/impl/ServerSessionPacketHandler.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -22,10 +22,8 @@ import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_CONSUMER_CLOSE; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_CREATECONSUMER; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_EXPIRED; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_FAILOVER_COMPLETE; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_FLOWTOKEN; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_QUEUEQUERY; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_REPLICATE_DELIVERY; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_ROLLBACK; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_SEND; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_SEND_CONTINUATION; @@ -49,8 +47,6 @@ import org.hornetq.core.remoting.ChannelHandler; import org.hornetq.core.remoting.Packet; import org.hornetq.core.remoting.impl.wireformat.CreateQueueMessage; -import org.hornetq.core.remoting.impl.wireformat.PacketImpl; -import org.hornetq.core.remoting.impl.wireformat.PacketsConfirmedMessage; import org.hornetq.core.remoting.impl.wireformat.RollbackMessage; import org.hornetq.core.remoting.impl.wireformat.SessionAcknowledgeMessage; import org.hornetq.core.remoting.impl.wireformat.SessionBindingQueryMessage; @@ -72,7 +68,6 @@ import org.hornetq.core.remoting.impl.wireformat.SessionXARollbackMessage; import org.hornetq.core.remoting.impl.wireformat.SessionXASetTimeoutMessage; import org.hornetq.core.remoting.impl.wireformat.SessionXAStartMessage; -import org.hornetq.core.remoting.impl.wireformat.replication.SessionReplicateDeliveryMessage; import org.hornetq.core.server.ServerSession; /** @@ -106,11 +101,6 @@ { switch (type) { - case PacketImpl.PACKETS_CONFIRMED: - { - session.handlePacketsConfirmed((PacketsConfirmedMessage)packet); - break; - } case SESS_CREATECONSUMER: { SessionCreateConsumerMessage request = (SessionCreateConsumerMessage)packet; @@ -118,15 +108,15 @@ break; } case CREATE_QUEUE: - { - CreateQueueMessage request = (CreateQueueMessage)packet; - session.handleCreateQueue(request); + { + CreateQueueMessage request = (CreateQueueMessage)packet; + session.handleCreateQueue(request); break; } case DELETE_QUEUE: { SessionDeleteQueueMessage request = (SessionDeleteQueueMessage)packet; - session.handleDeleteQueue(request); + session.handleDeleteQueue(request); break; } case SESS_QUEUEQUERY: @@ -237,11 +227,6 @@ session.handleStart(packet); break; } - case SESS_FAILOVER_COMPLETE: - { - session.handleFailedOver(packet); - break; - } case SESS_STOP: { session.handleStop(packet); @@ -267,27 +252,21 @@ case SESS_SEND: { SessionSendMessage message = (SessionSendMessage)packet; - session.handleSend(message); - break; + session.handleSend(message); + break; } case SESS_SEND_LARGE: { SessionSendLargeMessage message = (SessionSendLargeMessage)packet; - session.handleSendLargeMessage(message); - break; + session.handleSendLargeMessage(message); + break; } case SESS_SEND_CONTINUATION: { SessionSendContinuationMessage message = (SessionSendContinuationMessage)packet; session.handleSendContinuations(message); - break; + break; } - case SESS_REPLICATE_DELIVERY: - { - SessionReplicateDeliveryMessage message = (SessionReplicateDeliveryMessage)packet; - session.handleReplicatedDelivery(message); - break; - } } } catch (Throwable t) Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -141,8 +141,6 @@ deploymentManager = new FileDeploymentManager(server.getConfiguration().getFileDeployerScanPeriod()); - server.registerActivateCallback(this); - server.start(); started = true; Modified: branches/Branch_Replication_Changes/src/main/org/hornetq/jms/server/management/impl/JMSManagementServiceImpl.java =================================================================== --- branches/Branch_Replication_Changes/src/main/org/hornetq/jms/server/management/impl/JMSManagementServiceImpl.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/src/main/org/hornetq/jms/server/management/impl/JMSManagementServiceImpl.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -13,7 +13,6 @@ package org.hornetq.jms.server.management.impl; - import java.util.List; import javax.management.ObjectName; @@ -31,10 +30,6 @@ import org.hornetq.jms.server.JMSServerManager; import org.hornetq.jms.server.management.JMSManagementService; import org.hornetq.jms.server.management.JMSServerControl; -import org.hornetq.jms.server.management.jmx.impl.ReplicationAwareConnectionFactoryControlWrapper; -import org.hornetq.jms.server.management.jmx.impl.ReplicationAwareJMSQueueControlWrapper; -import org.hornetq.jms.server.management.jmx.impl.ReplicationAwareJMSServerControlWrapper; -import org.hornetq.jms.server.management.jmx.impl.ReplicationAwareTopicControlWrapper; /* * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a> @@ -65,12 +60,10 @@ { ObjectName objectName = ObjectNames.getJMSServerObjectName(); JMSServerControlImpl control = new JMSServerControlImpl(server); - JMSServerControl replicatingProxy = new ReplicationAwareJMSServerControlWrapper(control, - managementService.getReplicationOperationInvoker()); managementService.registerInJMX(objectName, - replicatingProxy); + control); managementService.registerInRegistry(ResourceNames.JMS_SERVER, control); - return replicatingProxy; + return control; } public synchronized void unregisterJMSServer() throws Exception @@ -97,9 +90,7 @@ coreQueueControl, jndiBinding, counter); - managementService.registerInJMX(objectName, - new ReplicationAwareJMSQueueControlWrapper(control, - managementService.getReplicationOperationInvoker())); + managementService.registerInJMX(objectName, control); managementService.registerInRegistry(ResourceNames.JMS_QUEUE + queue.getQueueName(), control); } @@ -116,8 +107,7 @@ ObjectName objectName = ObjectNames.getJMSTopicObjectName(topic.getTopicName()); AddressControl addressControl = (AddressControl)managementService.getResource(ResourceNames.CORE_ADDRESS + topic.getAddress()); TopicControlImpl control = new TopicControlImpl(topic, addressControl, jndiBinding, managementService); - managementService.registerInJMX(objectName, new ReplicationAwareTopicControlWrapper(control, - managementService.getReplicationOperationInvoker())); + managementService.registerInJMX(objectName, control); managementService.registerInRegistry(ResourceNames.JMS_TOPIC + topic.getTopicName(), control); } @@ -134,9 +124,7 @@ { ObjectName objectName = ObjectNames.getConnectionFactoryObjectName(name); ConnectionFactoryControlImpl control = new ConnectionFactoryControlImpl(connectionFactory, name, bindings); - managementService.registerInJMX(objectName, - new ReplicationAwareConnectionFactoryControlWrapper(control, - managementService.getReplicationOperationInvoker())); + managementService.registerInJMX(objectName, control); managementService.registerInRegistry(ResourceNames.JMS_CONNECTION_FACTORY + name, control); } Modified: branches/Branch_Replication_Changes/tests/config/ConfigurationTest-full-config.xml =================================================================== --- branches/Branch_Replication_Changes/tests/config/ConfigurationTest-full-config.xml 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/tests/config/ConfigurationTest-full-config.xml 2009-09-11 10:56:25 UTC (rev 7954) @@ -28,6 +28,7 @@ <persist-id-cache>true</persist-id-cache> <queue-activation-timeout>12456</queue-activation-timeout> <backup>true</backup> + <shared-store>true</shared-store> <persist-delivery-count-before-delivery>true</persist-delivery-count-before-delivery> <paging-directory>pagingdir</paging-directory> <bindings-directory>somedir</bindings-directory> Deleted: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/ActivationTimeoutTest.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/ActivationTimeoutTest.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/cluster/failover/ActivationTimeoutTest.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -1,324 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.tests.integration.cluster.failover; - -import java.util.HashMap; -import java.util.Map; - -import org.hornetq.core.client.ClientConsumer; -import org.hornetq.core.client.ClientMessage; -import org.hornetq.core.client.ClientProducer; -import org.hornetq.core.client.ClientSession; -import org.hornetq.core.client.impl.ClientSessionFactoryImpl; -import org.hornetq.core.client.impl.ClientSessionFactoryInternal; -import org.hornetq.core.client.impl.ClientSessionInternal; -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.logging.Logger; -import org.hornetq.core.remoting.RemotingConnection; -import org.hornetq.core.remoting.impl.invm.InVMRegistry; -import org.hornetq.core.remoting.impl.invm.TransportConstants; -import org.hornetq.core.server.HornetQ; -import org.hornetq.core.server.HornetQServer; -import org.hornetq.jms.client.HornetQTextMessage; -import org.hornetq.tests.util.UnitTestCase; -import org.hornetq.utils.SimpleString; - -/** - * - * A ActivationTimeoutTest - * - * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> - * - * Created 4 Nov 2008 16:54:50 - * - * - */ -public class ActivationTimeoutTest extends UnitTestCase -{ - private static final Logger log = Logger.getLogger(SimpleAutomaticFailoverTest.class); - - // Constants ----------------------------------------------------- - - // Attributes ---------------------------------------------------- - - private static final long ACTIVATION_TIMEOUT = 5000; - - private static final SimpleString ADDRESS = new SimpleString("FailoverTestAddress"); - - private HornetQServer liveService; - - private HornetQServer backupService; - - private Map<String, Object> backupParams = new HashMap<String, Object>(); - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - // Public -------------------------------------------------------- - - public void testTimeoutAfterConsumerFailsToReattach() throws Exception - { - ClientSessionFactoryInternal sf1 = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - backupParams)); - - ClientSessionFactoryInternal sf2 = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - backupParams)); - - sf1.setProducerWindowSize(32 * 1024); - sf2.setProducerWindowSize(32 * 1024); - - ClientSession session1 = sf1.createSession(false, true, true); - - session1.createQueue(ADDRESS, ADDRESS, null, false); - - ClientProducer producer = session1.createProducer(ADDRESS); - - final int numMessages = 1000; - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = session1.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message.putIntProperty(new SimpleString("count"), i); - message.getBody().writeString("aardvarks"); - producer.send(message); - } - - ClientConsumer consumer1 = session1.createConsumer(ADDRESS); - - ClientSession session2 = sf2.createSession(false, true, true); - - // Create another consumer so we have two consumers on the queue - ClientConsumer consumer2 = session2.createConsumer(ADDRESS); - - long start = System.currentTimeMillis(); - - RemotingConnection conn1 = ((ClientSessionInternal)session1).getConnection(); - - // Now we fail ONLY the connections on sf1, not on sf2 - conn1.fail(new HornetQException(HornetQException.NOT_CONNECTED)); - - session1.start(); - - // The messages should not be delivered until after activationTimeout ms, since - // session 2 didn't reattach - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = consumer1.receive(2 * ACTIVATION_TIMEOUT); - - assertNotNull(message); - - if (i == 0) - { - long now = System.currentTimeMillis(); - - assertTrue(now - start >= ACTIVATION_TIMEOUT); - } - - assertEquals("aardvarks", message.getBody().readString()); - - assertEquals(i, message.getProperty(new SimpleString("count"))); - - message.acknowledge(); - } - - ClientMessage message = consumer1.receive(1000); - - assertNull(message); - - session1.close(); - - RemotingConnection conn2 = ((ClientSessionInternal)session2).getConnection(); - - conn2.fail(new HornetQException(HornetQException.NOT_CONNECTED)); - - session2.close(); - } - - public void testTimeoutAfterAllConsumerFailsToReattach() throws Exception - { - for (int j = 0; j < 5; j++) - { - log.info("*************** ITERATION " + j); - - ClientSessionFactoryInternal sf1 = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - backupParams)); - - ClientSessionFactoryInternal sf2 = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory"), - new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - backupParams)); - - sf1.setProducerWindowSize(32 * 1024); - sf2.setProducerWindowSize(32 * 1024); - - ClientSession session1 = sf1.createSession(false, true, true); - - session1.createQueue(ADDRESS, ADDRESS, null, false); - - ClientProducer producer = session1.createProducer(ADDRESS); - - final int numMessages = 1000; - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = session1.createClientMessage(HornetQTextMessage.TYPE, - false, - 0, - System.currentTimeMillis(), - (byte)1); - message.putIntProperty(new SimpleString("count"), i); - message.getBody().writeString("aardvarks"); - producer.send(message); - } - - ClientSession session2 = sf2.createSession(false, true, true); - - ClientConsumer consumer1 = session2.createConsumer(ADDRESS); - - ClientConsumer consumer2 = session2.createConsumer(ADDRESS); - - long start = System.currentTimeMillis(); - - RemotingConnection conn1 = ((ClientSessionInternal)session1).getConnection(); - - // Now we fail ONLY the connections on sf1, not on sf2 - conn1.fail(new HornetQException(HornetQException.NOT_CONNECTED)); - - session1.start(); - - // The messages should not be delivered until after activationTimeout ms, since - // session 2 didn't reattach - - // We now create a new consumer but it shouldn't receive the messages until after the timeout - - ClientConsumer consumer3 = session1.createConsumer(ADDRESS); - - for (int i = 0; i < numMessages; i++) - { - ClientMessage message = consumer3.receive(2 * ACTIVATION_TIMEOUT); - - assertNotNull(message); - - if (i == 0) - { - long now = System.currentTimeMillis(); - - assertTrue(now - start >= ACTIVATION_TIMEOUT); - } - - assertEquals("aardvarks", message.getBody().readString()); - - assertEquals(i, message.getProperty(new SimpleString("count"))); - - message.acknowledge(); - } - - ClientMessage message = consumer3.receive(1000); - - assertNull(message); - - session1.close(); - - RemotingConnection conn2 = ((ClientSessionInternal)session2).getConnection(); - - conn2.fail(new HornetQException(HornetQException.NOT_CONNECTED)); - - session2.close(); - - stop(); - - start(); - - } - } - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - private void start() throws Exception - { - Configuration backupConf = new ConfigurationImpl(); - backupConf.setSecurityEnabled(false); - backupConf.setQueueActivationTimeout(ACTIVATION_TIMEOUT); - backupParams.put(TransportConstants.SERVER_ID_PROP_NAME, 1); - backupConf.getAcceptorConfigurations() - .add(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory", - backupParams)); - backupConf.setBackup(true); - backupService = HornetQ.newHornetQServer(backupConf, false); - backupService.start(); - - Configuration liveConf = new ConfigurationImpl(); - liveConf.setSecurityEnabled(false); - liveConf.getAcceptorConfigurations() - .add(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory")); - Map<String, TransportConfiguration> connectors = new HashMap<String, TransportConfiguration>(); - TransportConfiguration backupTC = new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory", - backupParams, - "backup-connector"); - connectors.put(backupTC.getName(), backupTC); - liveConf.setConnectorConfigurations(connectors); - liveConf.setBackupConnectorName(backupTC.getName()); - liveService = HornetQ.newHornetQServer(liveConf, false); - liveService.start(); - } - - private void stop() throws Exception - { - backupService.stop(); - - liveService.stop(); - - assertEquals(0, InVMRegistry.instance.size()); - - } - - @Override - protected void setUp() throws Exception - { - super.setUp(); - - start(); - } - - @Override - protected void tearDown() throws Exception - { - stop(); - - backupService = null; - - liveService = null; - - backupParams = null; - - super.tearDown(); - } - - // Private ------------------------------------------------------- - - // Inner classes ------------------------------------------------- -} Modified: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/management/HornetQServerControlTest.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/management/HornetQServerControlTest.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/management/HornetQServerControlTest.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -72,7 +72,7 @@ assertEquals(conf.isClustered(), serverControl.isClustered()); assertEquals(conf.isPersistDeliveryCountBeforeDelivery(), serverControl.isPersistDeliveryCountBeforeDelivery()); assertEquals(conf.isBackup(), serverControl.isBackup()); - assertEquals(conf.getQueueActivationTimeout(), serverControl.getQueueActivationTimeout()); + assertEquals(conf.isSharedStore(), serverControl.isSharedStore()); assertEquals(conf.getScheduledThreadPoolMaxSize(), serverControl.getScheduledThreadPoolMaxSize()); assertEquals(conf.getThreadPoolMaxSize(), serverControl.getThreadPoolMaxSize()); assertEquals(conf.getSecurityInvalidationInterval(), serverControl.getSecurityInvalidationInterval()); Modified: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/management/QueueControlTest.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/management/QueueControlTest.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/integration/management/QueueControlTest.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -84,7 +84,6 @@ assertEquals(filter.toString(), queueControl.getFilter()); assertEquals(durable, queueControl.isDurable()); assertEquals(false, queueControl.isTemporary()); - assertEquals(false, queueControl.isBackup()); session.deleteQueue(queue); } Modified: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/unit/core/config/impl/ConfigurationImplTest.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/unit/core/config/impl/ConfigurationImplTest.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/unit/core/config/impl/ConfigurationImplTest.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -44,7 +44,7 @@ { assertEquals(ConfigurationImpl.DEFAULT_CLUSTERED, conf.isClustered()); assertEquals(ConfigurationImpl.DEFAULT_BACKUP, conf.isBackup()); - assertEquals(ConfigurationImpl.DEFAULT_QUEUE_ACTIVATION_TIMEOUT, conf.getQueueActivationTimeout()); + assertEquals(ConfigurationImpl.DEFAULT_SHARED_STORE, conf.isSharedStore()); assertEquals(ConfigurationImpl.DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE, conf.getScheduledThreadPoolMaxSize()); assertEquals(ConfigurationImpl.DEFAULT_SECURITY_INVALIDATION_INTERVAL, conf.getSecurityInvalidationInterval()); assertEquals(ConfigurationImpl.DEFAULT_SECURITY_ENABLED, conf.isSecurityEnabled()); @@ -83,15 +83,15 @@ conf.setBackup(b); assertEquals(b, conf.isBackup()); - long l = randomLong(); - conf.setQueueActivationTimeout(l); - assertEquals(l, conf.getQueueActivationTimeout()); + b = randomBoolean(); + conf.setSharedStore(b); + assertEquals(b, conf.isSharedStore()); int i = randomInt(); conf.setScheduledThreadPoolMaxSize(i); assertEquals(i, conf.getScheduledThreadPoolMaxSize()); - l = randomLong(); + long l = randomLong(); conf.setSecurityInvalidationInterval(l); assertEquals(l, conf.getSecurityInvalidationInterval()); @@ -176,14 +176,13 @@ b = randomBoolean(); conf.setBackup(b); - long l = randomLong(); - conf.setQueueActivationTimeout(l); - - + b = randomBoolean(); + conf.setSharedStore(b); + int i = randomInt(); conf.setScheduledThreadPoolMaxSize(i); - l = randomLong(); + long l = randomLong(); conf.setSecurityInvalidationInterval(l); b = randomBoolean(); Modified: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/unit/core/config/impl/DefaultsFileConfigurationTest.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/unit/core/config/impl/DefaultsFileConfigurationTest.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/unit/core/config/impl/DefaultsFileConfigurationTest.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -37,7 +37,7 @@ assertEquals(ConfigurationImpl.DEFAULT_BACKUP, conf.isBackup()); - assertEquals(ConfigurationImpl.DEFAULT_QUEUE_ACTIVATION_TIMEOUT, conf.getQueueActivationTimeout()); + assertEquals(ConfigurationImpl.DEFAULT_SHARED_STORE, conf.isSharedStore()); assertEquals(ConfigurationImpl.DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE, conf.getScheduledThreadPoolMaxSize()); Modified: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/unit/core/config/impl/FileConfigurationTest.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/unit/core/config/impl/FileConfigurationTest.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/unit/core/config/impl/FileConfigurationTest.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -59,8 +59,8 @@ assertEquals(8, conf.getMessageExpiryThreadPriority()); assertEquals(127, conf.getIDCacheSize()); assertEquals(true, conf.isPersistIDCache()); - assertEquals(12456, conf.getQueueActivationTimeout()); assertEquals(true, conf.isBackup()); + assertEquals(true, conf.isSharedStore()); assertEquals(true, conf.isPersistDeliveryCountBeforeDelivery()); assertEquals("pagingdir", conf.getPagingDirectory()); assertEquals("somedir", conf.getBindingsDirectory()); Modified: branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingImplTest.java =================================================================== --- branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingImplTest.java 2009-09-11 02:25:29 UTC (rev 7953) +++ branches/Branch_Replication_Changes/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingImplTest.java 2009-09-11 10:56:25 UTC (rev 7954) @@ -1089,7 +1089,7 @@ /* (non-Javadoc) * @see org.hornetq.core.server.Queue#addRedistributor(long, java.util.concurrent.Executor, org.hornetq.core.remoting.Channel) */ - public void addRedistributor(long delay, Executor executor, Channel replicatingChannel) + public void addRedistributor(long delay, Executor executor) { }