Author: borges Date: 2011-10-20 07:00:16 -0400 (Thu, 20 Oct 2011) New Revision: 11566 Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java trunk/hornetq-core/src/main/java/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java Log: HORNETQ-720 Revert changes merged from 2_2_EAP (they break replication) Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java =================================================================== --- trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2011-10-20 07:09:29 UTC (rev 11565) +++ trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2011-10-20 11:00:16 UTC (rev 11566) @@ -312,14 +312,7 @@ throw new IllegalArgumentException("Unsupported journal type " + config.getJournalType()); } - if (config.isBackup() && !config.isSharedStore()) - { - idGenerator = null; - } - else - { - idGenerator = new BatchingIDGenerator(0, JournalStorageManager.CHECKPOINT_BATCH_SIZE, bindingsJournal); - } + idGenerator = new BatchingIDGenerator(0, JournalStorageManager.CHECKPOINT_BATCH_SIZE, bindingsJournal); Journal localMessage = new JournalImpl(config.getJournalFileSize(), config.getJournalMinFiles(), config.getJournalCompactMinFiles(), Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java =================================================================== --- trunk/hornetq-core/src/main/java/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java 2011-10-20 07:09:29 UTC (rev 11565) +++ trunk/hornetq-core/src/main/java/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java 2011-10-20 11:00:16 UTC (rev 11566) @@ -65,12 +65,12 @@ import org.hornetq.utils.UUID; /** - * + * * A ClusterConnectionImpl * * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> * @author Clebert Suconic - * + * * Created 21 Jan 2009 14:43:05 * * @@ -142,11 +142,11 @@ private final Set<TransportConfiguration> allowableConnections = new HashSet<TransportConfiguration>(); private final ClusterManagerInternal manager; - - + + // Stuff that used to be on the ClusterManager - + private final Topology topology = new Topology(this); private volatile ServerLocatorInternal backupServerLocator; @@ -214,7 +214,7 @@ this.executorFactory = executorFactory; this.executor = executorFactory.getExecutor(); - + this.topology.setExecutor(executor); this.server = server; @@ -325,7 +325,7 @@ this.executorFactory = executorFactory; this.executor = executorFactory.getExecutor(); - + this.topology.setExecutor(executor); this.server = server; @@ -367,10 +367,10 @@ { return; } - - + + started = true; - + if (!backup) { activate(); @@ -378,7 +378,7 @@ } } - + public void flushExecutor() { Future future = new Future(); @@ -435,35 +435,23 @@ props); managementService.sendNotification(notification); } - - executor.execute(new Runnable() + if (backupServerLocator != null) { - public void run() - { - synchronized (ClusterConnectionImpl.this) - { - if (backupServerLocator != null) - { - backupServerLocator.close(); - backupServerLocator = null; - } + backupServerLocator.close(); + backupServerLocator = null; + } - if (serverLocator != null) - { - serverLocator.close(); - serverLocator = null; - } - } + if (serverLocator != null) + { + serverLocator.close(); + serverLocator = null; + } - } - }); - started = false; } - public void announceBackup() { executor.execute(new Runnable() @@ -501,7 +489,7 @@ { return topology.getMember(manager.getNodeId()); } - + public void addClusterTopologyListener(final ClusterTopologyListener listener, final boolean clusterConnection) { topology.addClusterTopologyListener(listener); @@ -519,7 +507,7 @@ { return topology; } - + public void nodeAnnounced(final long uniqueEventID, final String nodeID, final Pair<TransportConfiguration, TransportConfiguration> connectorPair, @@ -610,7 +598,7 @@ } backup = false; - + topology.updateAsLive(manager.getNodeId(), new TopologyMember(connector, null)); if (backupServerLocator != null) @@ -831,13 +819,13 @@ } } } - + public synchronized void informTopology() { String nodeID = server.getNodeID().toString(); - + TopologyMember localMember; - + if (backup) { localMember = new TopologyMember(null, connector); @@ -859,21 +847,21 @@ final boolean start) throws Exception { final ServerLocatorInternal targetLocator = new ServerLocatorImpl(topology, false, connector); - + String nodeId; - + synchronized (this) { if (!started) { return; } - + if (serverLocator == null) { return; } - + nodeId = serverLocator.getNodeID(); }