[jboss-cvs] JBoss Messaging SVN: r5918 - in trunk: src/main/org/jboss/messaging/core/journal/impl and 12 other directories.
jboss-cvs-commits at lists.jboss.org
jboss-cvs-commits at lists.jboss.org
Sat Feb 21 09:22:13 EST 2009
Author: timfox
Date: 2009-02-21 09:22:13 -0500 (Sat, 21 Feb 2009)
New Revision: 5918
Modified:
trunk/src/main/org/jboss/messaging/core/journal/Journal.java
trunk/src/main/org/jboss/messaging/core/journal/impl/JournalImpl.java
trunk/src/main/org/jboss/messaging/core/list/impl/PriorityLinkedListImpl.java
trunk/src/main/org/jboss/messaging/core/message/impl/MessageImpl.java
trunk/src/main/org/jboss/messaging/core/persistence/StorageManager.java
trunk/src/main/org/jboss/messaging/core/persistence/impl/journal/JournalStorageManager.java
trunk/src/main/org/jboss/messaging/core/persistence/impl/nullpm/NullStorageManager.java
trunk/src/main/org/jboss/messaging/core/postoffice/impl/PostOfficeImpl.java
trunk/src/main/org/jboss/messaging/core/server/MessagingServer.java
trunk/src/main/org/jboss/messaging/core/server/cluster/impl/BridgeImpl.java
trunk/src/main/org/jboss/messaging/core/server/cluster/impl/ClusterConnectionImpl.java
trunk/src/main/org/jboss/messaging/core/server/cluster/impl/ClusterManagerImpl.java
trunk/src/main/org/jboss/messaging/core/server/impl/GroupingRoundRobinDistributor.java
trunk/src/main/org/jboss/messaging/core/server/impl/MessagingServerImpl.java
trunk/src/main/org/jboss/messaging/core/server/impl/RoundRobinDistributor.java
trunk/src/main/org/jboss/messaging/util/UUID.java
trunk/tests/src/org/jboss/messaging/tests/integration/cluster/bridge/BridgeStartTest.java
trunk/tests/src/org/jboss/messaging/tests/integration/management/DiscoveryGroupControlTest.java
Log:
mainly bridge duplicate detection
Modified: trunk/src/main/org/jboss/messaging/core/journal/Journal.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/journal/Journal.java 2009-02-21 01:16:23 UTC (rev 5917)
+++ trunk/src/main/org/jboss/messaging/core/journal/Journal.java 2009-02-21 14:22:13 UTC (rev 5918)
@@ -41,6 +41,8 @@
void appendAddRecord(long id, byte recordType, byte[] record) throws Exception;
void appendAddRecord(long id, byte recordType, EncodingSupport record) throws Exception;
+
+ void appendAddRecord(long id, byte recordType, EncodingSupport record, boolean sync) throws Exception;
void appendUpdateRecord(long id, byte recordType, byte[] record) throws Exception;
Modified: trunk/src/main/org/jboss/messaging/core/journal/impl/JournalImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/journal/impl/JournalImpl.java 2009-02-21 01:16:23 UTC (rev 5917)
+++ trunk/src/main/org/jboss/messaging/core/journal/impl/JournalImpl.java 2009-02-21 14:22:13 UTC (rev 5918)
@@ -286,11 +286,16 @@
public void appendAddRecord(final long id, final byte recordType, final byte[] record) throws Exception
{
- appendAddRecord(id, recordType, new ByteArrayEncoding(record));
+ appendAddRecord(id, recordType, new ByteArrayEncoding(record), false);
}
-
+
public void appendAddRecord(final long id, final byte recordType, final EncodingSupport record) throws Exception
{
+ appendAddRecord(id, recordType, record, syncNonTransactional);
+ }
+
+ public void appendAddRecord(final long id, final byte recordType, final EncodingSupport record, final boolean sync) throws Exception
+ {
if (state != STATE_LOADED)
{
throw new IllegalStateException("Journal must be loaded first");
@@ -312,7 +317,7 @@
try
{
- JournalFile usedFile = appendRecord(bb.getBuffer(), syncNonTransactional, null);
+ JournalFile usedFile = appendRecord(bb.getBuffer(), sync, null);
posFilesMap.put(id, new PosFiles(usedFile));
}
Modified: trunk/src/main/org/jboss/messaging/core/list/impl/PriorityLinkedListImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/list/impl/PriorityLinkedListImpl.java 2009-02-21 01:16:23 UTC (rev 5917)
+++ trunk/src/main/org/jboss/messaging/core/list/impl/PriorityLinkedListImpl.java 2009-02-21 14:22:13 UTC (rev 5918)
@@ -18,7 +18,7 @@
* License along with this software; if not, write to the Free
* Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
* 02110-1301 USA, or see the FSF site: http://www.fsf.org.
- */
+ */
package org.jboss.messaging.core.list.impl;
@@ -42,72 +42,72 @@
* $Id: BasicPrioritizedDeque.java 1174 2006-08-02 14:14:32Z timfox $
*/
public class PriorityLinkedListImpl<T> implements PriorityLinkedList<T>
-{
+{
private final List<LinkedList<T>> linkedLists;
-
+
private final int priorities;
-
+
private int size;
-
+
public PriorityLinkedListImpl(final int priorities)
{
this.priorities = priorities;
-
+
linkedLists = new ArrayList<LinkedList<T>>();
-
+
for (int i = 0; i < priorities; i++)
{
linkedLists.add(new LinkedList<T>());
}
}
-
+
public void addFirst(final T t, final int priority)
- {
+ {
linkedLists.get(priority).addFirst(t);
-
- size++;
+
+ size++;
}
-
+
public void addLast(final T t, final int priority)
- {
+ {
linkedLists.get(priority).addLast(t);
-
+
size++;
}
public T removeFirst()
{
T t = null;
-
- //Initially we are just using a simple prioritization algorithm:
- //Highest priority refs always get returned first.
- //This could cause starvation of lower priority refs.
-
- //TODO - A better prioritization algorithm
-
+
+ // Initially we are just using a simple prioritization algorithm:
+ // Highest priority refs always get returned first.
+ // This could cause starvation of lower priority refs.
+
+ // TODO - A better prioritization algorithm
+
for (int i = priorities - 1; i >= 0; i--)
{
LinkedList<T> ll = linkedLists.get(i);
-
+
if (!ll.isEmpty())
{
t = ll.removeFirst();
break;
- }
+ }
}
-
+
if (t != null)
{
size--;
}
-
- return t;
+
+ return t;
}
-
+
public T peekFirst()
{
T t = null;
-
+
for (int i = priorities - 1; i >= 0; i--)
{
LinkedList<T> ll = linkedLists.get(i);
@@ -120,38 +120,38 @@
break;
}
}
-
- return t;
+
+ return t;
}
-
+
public List<T> getAll()
{
List<T> all = new ArrayList<T>();
-
+
for (int i = priorities - 1; i >= 0; i--)
{
LinkedList<T> list = linkedLists.get(i);
all.addAll(list);
}
-
+
return all;
}
-
+
public void clear()
{
- for (LinkedList<T> list: linkedLists)
+ for (LinkedList<T> list : linkedLists)
{
list.clear();
}
-
- size = 0;
+
+ size = 0;
}
-
+
public int size()
{
return size;
}
-
+
public boolean isEmpty()
{
return size == 0;
@@ -161,17 +161,17 @@
{
return new PriorityLinkedListIterator();
}
-
+
private class PriorityLinkedListIterator implements Iterator<T>
- {
+ {
private int index;
-
+
private ListIterator<T> currentIter;
-
+
PriorityLinkedListIterator()
{
index = linkedLists.size() - 1;
-
+
currentIter = linkedLists.get(index).listIterator();
}
@@ -181,19 +181,19 @@
{
return true;
}
-
+
while (index >= 0)
- {
+ {
if (index == 0 || currentIter.hasNext())
{
break;
- }
-
+ }
+
index--;
-
+
currentIter = linkedLists.get(index).listIterator();
}
- return currentIter.hasNext();
+ return currentIter.hasNext();
}
public T next()
@@ -207,9 +207,9 @@
public void remove()
{
- currentIter.remove();
-
+ currentIter.remove();
+
size--;
}
- }
+ }
}
Modified: trunk/src/main/org/jboss/messaging/core/message/impl/MessageImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/message/impl/MessageImpl.java 2009-02-21 01:16:23 UTC (rev 5917)
+++ trunk/src/main/org/jboss/messaging/core/message/impl/MessageImpl.java 2009-02-21 14:22:13 UTC (rev 5918)
@@ -231,11 +231,8 @@
// TODO - this can be optimised
byte[] bytes = new byte[len];
buffer.getBytes(bytes);
- // body = new ByteBufferWrapper(ByteBuffer.wrap(bytes));
- // body.position(body.limit());
body = buffer.createNewBuffer(len);
body.putBytes(bytes);
-
}
public long getMessageID()
Modified: trunk/src/main/org/jboss/messaging/core/persistence/StorageManager.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/persistence/StorageManager.java 2009-02-21 01:16:23 UTC (rev 5917)
+++ trunk/src/main/org/jboss/messaging/core/persistence/StorageManager.java 2009-02-21 14:22:13 UTC (rev 5918)
@@ -40,6 +40,7 @@
import org.jboss.messaging.core.transaction.ResourceManager;
import org.jboss.messaging.util.Pair;
import org.jboss.messaging.util.SimpleString;
+import org.jboss.messaging.util.UUID;
/**
*
@@ -52,6 +53,8 @@
public interface StorageManager extends MessagingComponent
{
// Message related operations
+
+ UUID getPersistentID();
long generateUniqueID();
Modified: trunk/src/main/org/jboss/messaging/core/persistence/impl/journal/JournalStorageManager.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/persistence/impl/journal/JournalStorageManager.java 2009-02-21 01:16:23 UTC (rev 5917)
+++ trunk/src/main/org/jboss/messaging/core/persistence/impl/journal/JournalStorageManager.java 2009-02-21 14:22:13 UTC (rev 5918)
@@ -85,6 +85,8 @@
import org.jboss.messaging.util.Pair;
import org.jboss.messaging.util.SimpleString;
import org.jboss.messaging.util.TimeAndCounterIDGenerator;
+import org.jboss.messaging.util.UUID;
+import org.jboss.messaging.util.UUIDGenerator;
/**
*
@@ -104,6 +106,8 @@
public static final byte QUEUE_BINDING_RECORD = 21;
public static final byte DESTINATION_RECORD = 22;
+
+ public static final byte PERSISTENT_ID_RECORD = 23;
// type + expiration + timestamp + priority
public static final int SIZE_FIELDS = SIZE_INT + SIZE_LONG + SIZE_LONG + SIZE_BYTE;
@@ -125,6 +129,8 @@
public static final byte SET_SCHEDULED_DELIVERY_TIME = 36;
public static final byte DUPLICATE_ID = 37;
+
+ private UUID persistentID;
// This will produce a unique id **for this node only**
private final IDGenerator idGenerator = new TimeAndCounterIDGenerator();
@@ -227,6 +233,11 @@
this.bindingsJournal = bindingsJournal;
this.largeMessagesFactory = largeMessagesFactory;
}
+
+ public UUID getPersistentID()
+ {
+ return persistentID;
+ }
public long generateUniqueID()
{
@@ -959,11 +970,26 @@
destinations.add(destinationEncoding.destination);
}
+ else if (rec == PERSISTENT_ID_RECORD)
+ {
+ PersistentIDEncoding encoding = new PersistentIDEncoding();
+
+ encoding.decode(buffer);
+
+ persistentID = encoding.uuid;
+ }
else
{
throw new IllegalStateException("Invalid record type " + rec);
}
}
+
+ if (persistentID == null)
+ {
+ persistentID = UUIDGenerator.getInstance().generateUUID();
+
+ bindingsJournal.appendAddRecord(generateUniqueID(), PERSISTENT_ID_RECORD, new PersistentIDEncoding(persistentID), true);
+ }
}
// MessagingComponent implementation
@@ -1239,7 +1265,41 @@
}
}
+
+ private static class PersistentIDEncoding implements EncodingSupport
+ {
+ UUID uuid;
+ PersistentIDEncoding(final UUID uuid)
+ {
+ this.uuid = uuid;
+ }
+
+ PersistentIDEncoding()
+ {
+ }
+
+ public void decode(final MessagingBuffer buffer)
+ {
+ byte[] bytes = new byte[16];
+
+ buffer.getBytes(bytes);
+
+ uuid = new UUID(UUID.TYPE_TIME_BASED, bytes);
+ }
+
+ public void encode(final MessagingBuffer buffer)
+ {
+ buffer.putBytes(uuid.asBytes());
+ }
+
+ public int getEncodeSize()
+ {
+ return 16;
+ }
+
+ }
+
private static class LargeMessageEncoding implements EncodingSupport
{
private final LargeServerMessage message;
Modified: trunk/src/main/org/jboss/messaging/core/persistence/impl/nullpm/NullStorageManager.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/persistence/impl/nullpm/NullStorageManager.java 2009-02-21 01:16:23 UTC (rev 5917)
+++ trunk/src/main/org/jboss/messaging/core/persistence/impl/nullpm/NullStorageManager.java 2009-02-21 14:22:13 UTC (rev 5918)
@@ -43,6 +43,8 @@
import org.jboss.messaging.util.Pair;
import org.jboss.messaging.util.SimpleString;
import org.jboss.messaging.util.TimeAndCounterIDGenerator;
+import org.jboss.messaging.util.UUID;
+import org.jboss.messaging.util.UUIDGenerator;
/**
*
@@ -55,8 +57,15 @@
public class NullStorageManager implements StorageManager
{
private final IDGenerator idGenerator = new TimeAndCounterIDGenerator();
+
+ private final UUID id = UUIDGenerator.getInstance().generateUUID();
private volatile boolean started;
+
+ public UUID getPersistentID()
+ {
+ return id;
+ }
public void addQueueBinding(final Binding queueBinding) throws Exception
{
Modified: trunk/src/main/org/jboss/messaging/core/postoffice/impl/PostOfficeImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/postoffice/impl/PostOfficeImpl.java 2009-02-21 01:16:23 UTC (rev 5917)
+++ trunk/src/main/org/jboss/messaging/core/postoffice/impl/PostOfficeImpl.java 2009-02-21 14:22:13 UTC (rev 5918)
@@ -24,13 +24,11 @@
import java.nio.ByteBuffer;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.Timer;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledThreadPoolExecutor;
@@ -57,7 +55,6 @@
import org.jboss.messaging.core.postoffice.PostOffice;
import org.jboss.messaging.core.postoffice.QueueInfo;
import org.jboss.messaging.core.remoting.impl.ByteBufferWrapper;
-import org.jboss.messaging.core.server.MessageReference;
import org.jboss.messaging.core.server.Queue;
import org.jboss.messaging.core.server.QueueFactory;
import org.jboss.messaging.core.server.SendLock;
@@ -137,8 +134,6 @@
private final HierarchicalRepository<AddressSettings> addressSettingsRepository;
- private final String nodeID;
-
public PostOfficeImpl(final StorageManager storageManager,
final PagingManager pagingManager,
final QueueFactory bindableFactory,
@@ -151,8 +146,7 @@
final int idCacheSize,
final boolean persistIDCache,
final ExecutorFactory orderedExecutorFactory,
- HierarchicalRepository<AddressSettings> addressSettingsRepository,
- final String nodeID)
+ HierarchicalRepository<AddressSettings> addressSettingsRepository)
{
this.storageManager = storageManager;
@@ -186,10 +180,7 @@
this.redistributorExecutorFactory = orderedExecutorFactory;
- this.addressSettingsRepository = addressSettingsRepository;
-
- this.nodeID = nodeID;
-
+ this.addressSettingsRepository = addressSettingsRepository;
}
// MessagingComponent implementation ---------------------------------------
Modified: trunk/src/main/org/jboss/messaging/core/server/MessagingServer.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/server/MessagingServer.java 2009-02-21 01:16:23 UTC (rev 5917)
+++ trunk/src/main/org/jboss/messaging/core/server/MessagingServer.java 2009-02-21 14:22:13 UTC (rev 5918)
@@ -32,6 +32,7 @@
import org.jboss.messaging.core.transaction.ResourceManager;
import org.jboss.messaging.core.version.Version;
import org.jboss.messaging.util.SimpleString;
+import org.jboss.messaging.util.UUID;
/**
* This interface defines the internal interface of the Messaging Server exposed to other components of the server. The
@@ -118,4 +119,6 @@
QueueFactory getQueueFactory();
SimpleString getNodeID();
+
+ UUID getUUID();
}
Modified: trunk/src/main/org/jboss/messaging/core/server/cluster/impl/BridgeImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/server/cluster/impl/BridgeImpl.java 2009-02-21 01:16:23 UTC (rev 5917)
+++ trunk/src/main/org/jboss/messaging/core/server/cluster/impl/BridgeImpl.java 2009-02-21 14:22:13 UTC (rev 5918)
@@ -27,7 +27,6 @@
import static org.jboss.messaging.core.client.impl.ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_ACKNOWLEDGE;
import static org.jboss.messaging.core.client.impl.ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_NON_PERSISTENT_SEND;
import static org.jboss.messaging.core.client.impl.ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_PERSISTENT_SEND;
-import static org.jboss.messaging.core.client.impl.ClientSessionFactoryImpl.DEFAULT_CALL_TIMEOUT;
import static org.jboss.messaging.core.client.impl.ClientSessionFactoryImpl.DEFAULT_CONNECTION_LOAD_BALANCING_POLICY_CLASS_NAME;
import static org.jboss.messaging.core.client.impl.ClientSessionFactoryImpl.DEFAULT_CONNECTION_TTL;
import static org.jboss.messaging.core.client.impl.ClientSessionFactoryImpl.DEFAULT_CONSUMER_MAX_RATE;
@@ -78,6 +77,7 @@
import org.jboss.messaging.util.Future;
import org.jboss.messaging.util.Pair;
import org.jboss.messaging.util.SimpleString;
+import org.jboss.messaging.util.UUID;
import org.jboss.messaging.util.UUIDGenerator;
/**
@@ -97,6 +97,8 @@
// Attributes ----------------------------------------------------
+ private final UUID nodeUUID;
+
private final SimpleString name;
private final Queue queue;
@@ -151,7 +153,8 @@
// Public --------------------------------------------------------
- public BridgeImpl(final SimpleString name,
+ public BridgeImpl(final UUID nodeUUID,
+ final SimpleString name,
final Queue queue,
final Pair<TransportConfiguration, TransportConfiguration> connectorPair,
final Executor executor,
@@ -168,7 +171,8 @@
final SimpleString managementNotificationAddress,
final String clusterPassword) throws Exception
{
- this(name,
+ this(nodeUUID,
+ name,
queue,
connectorPair,
executor,
@@ -187,7 +191,8 @@
null);
}
- public BridgeImpl(final SimpleString name,
+ public BridgeImpl(final UUID nodeUUID,
+ final SimpleString name,
final Queue queue,
final Pair<TransportConfiguration, TransportConfiguration> connectorPair,
final Executor executor,
@@ -205,6 +210,8 @@
final String clusterPassword,
final MessageFlowRecord flowRecord) throws Exception
{
+ this.nodeUUID = nodeUUID;
+
this.name = name;
this.queue = queue;
@@ -645,11 +652,17 @@
if (useDuplicateDetection && !message.containsProperty(MessageImpl.HDR_DUPLICATE_DETECTION_ID))
{
- byte[] bytes = new byte[8];
+ //If we are using duplicate detection and there's not already a duplicate detection header, then
+ //we add a header composed of the persistent node id and the message id, which makes it globally unique
+ //between restarts.
+ //If you use a cluster connection then a guid based duplicate id will be used since it is added *before* the
+ //message goes into the store and forward queue.
+ //But with this technique it also works when the messages don't already have such a header in them in the queue.
+ byte[] bytes = new byte[24];
ByteBuffer bb = ByteBuffer.wrap(bytes);
- // TODO NEEDS to incluse server id
+ bb.put(nodeUUID.asBytes());
bb.putLong(message.getMessageID());
Modified: trunk/src/main/org/jboss/messaging/core/server/cluster/impl/ClusterConnectionImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/server/cluster/impl/ClusterConnectionImpl.java 2009-02-21 01:16:23 UTC (rev 5917)
+++ trunk/src/main/org/jboss/messaging/core/server/cluster/impl/ClusterConnectionImpl.java 2009-02-21 14:22:13 UTC (rev 5918)
@@ -31,7 +31,6 @@
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.atomic.AtomicInteger;
import org.jboss.messaging.core.client.ClientMessage;
import org.jboss.messaging.core.client.management.impl.ManagementHelper;
@@ -56,6 +55,7 @@
import org.jboss.messaging.util.ExecutorFactory;
import org.jboss.messaging.util.Pair;
import org.jboss.messaging.util.SimpleString;
+import org.jboss.messaging.util.UUID;
/**
*
@@ -76,7 +76,7 @@
private final StorageManager storageManager;
private final PostOffice postOffice;
-
+
private final ManagementService managementService;
private final SimpleString name;
@@ -90,7 +90,7 @@
private final int maxRetriesBeforeFailover;
private final int maxRetriesAfterFailover;
-
+
private final boolean useDuplicateDetection;
private final boolean routeWhenNoConsumers;
@@ -104,9 +104,9 @@
private final QueueFactory queueFactory;
private final int maxHops;
-
- private final SimpleString nodeID;
+ private final UUID nodeUUID;
+
private volatile boolean started;
/*
@@ -126,22 +126,22 @@
final ManagementService managementService,
final ScheduledExecutorService scheduledExecutor,
final QueueFactory queueFactory,
- final List<Pair<TransportConfiguration, TransportConfiguration>> connectors,
+ final List<Pair<TransportConfiguration, TransportConfiguration>> connectors,
final int maxHops,
- final SimpleString nodeID) throws Exception
+ final UUID nodeUUID) throws Exception
{
this.name = name;
this.address = address;
this.retryInterval = retryInterval;
-
+
this.retryIntervalMultiplier = retryIntervalMultiplier;
-
+
this.maxRetriesBeforeFailover = maxRetriesBeforeFailover;
-
+
this.maxRetriesAfterFailover = maxRetriesAfterFailover;
-
+
this.useDuplicateDetection = useDuplicateDetection;
this.routeWhenNoConsumers = routeWhenNoConsumers;
@@ -151,7 +151,7 @@
this.storageManager = storageManager;
this.postOffice = postOffice;
-
+
this.managementService = managementService;
this.discoveryGroup = null;
@@ -159,11 +159,11 @@
this.scheduledExecutor = scheduledExecutor;
this.queueFactory = queueFactory;
-
+
this.maxHops = maxHops;
-
- this.nodeID = nodeID;
+ this.nodeUUID = nodeUUID;
+
this.updateConnectors(connectors);
}
@@ -184,20 +184,20 @@
final ManagementService managementService,
final ScheduledExecutorService scheduledExecutor,
final QueueFactory queueFactory,
- final DiscoveryGroup discoveryGroup,
+ final DiscoveryGroup discoveryGroup,
final int maxHops,
- final SimpleString nodeID) throws Exception
+ final UUID nodeUUID) throws Exception
{
this.name = name;
this.address = address;
this.retryInterval = retryInterval;
-
+
this.retryIntervalMultiplier = retryIntervalMultiplier;
-
+
this.maxRetriesBeforeFailover = maxRetriesBeforeFailover;
-
+
this.maxRetriesAfterFailover = maxRetriesAfterFailover;
this.executorFactory = executorFactory;
@@ -205,7 +205,7 @@
this.storageManager = storageManager;
this.postOffice = postOffice;
-
+
this.managementService = managementService;
this.scheduledExecutor = scheduledExecutor;
@@ -219,8 +219,8 @@
this.routeWhenNoConsumers = routeWhenNoConsumers;
this.maxHops = maxHops;
-
- this.nodeID = nodeID;
+
+ this.nodeUUID = nodeUUID;
}
public synchronized void start() throws Exception
@@ -244,7 +244,7 @@
{
return;
}
-
+
if (discoveryGroup != null)
{
discoveryGroup.unregisterListener(this);
@@ -283,7 +283,7 @@
log.error("Failed to update connectors", e);
}
}
-
+
private void updateConnectors(final List<Pair<TransportConfiguration, TransportConfiguration>> connectors) throws Exception
{
Set<Pair<TransportConfiguration, TransportConfiguration>> connectorSet = new HashSet<Pair<TransportConfiguration, TransportConfiguration>>();
@@ -313,7 +313,7 @@
if (!records.containsKey(connectorPair))
{
SimpleString queueName = generateQueueName(name, connectorPair);
-
+
Binding queueBinding = postOffice.getBinding(queueName);
Queue queue;
@@ -329,19 +329,20 @@
// 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
- Binding storeBinding = new LocalQueueBinding(queue.getName(), queue, nodeID);
+ Binding storeBinding = new LocalQueueBinding(queue.getName(), queue, new SimpleString(nodeUUID.toString()));
storageManager.addQueueBinding(storeBinding);
}
MessageFlowRecordImpl record = new MessageFlowRecordImpl(queue);
- Bridge bridge = new BridgeImpl(queueName,
+ Bridge bridge = new BridgeImpl(nodeUUID,
+ queueName,
queue,
connectorPair,
- executorFactory.getExecutor(),
+ executorFactory.getExecutor(),
null,
- null,
+ null,
scheduledExecutor,
null,
retryInterval,
@@ -473,11 +474,12 @@
return;
}
- //TODO - optimised this by just passing int in header - but filter needs to be extended to support IN with a list of integers
+ // TODO - optimised this by just passing int in header - but filter needs to be extended to support IN with
+ // a list of integers
SimpleString type = (SimpleString)message.getProperty(ManagementHelper.HDR_NOTIFICATION_TYPE);
-
+
NotificationType ntype = NotificationType.valueOf(type.toString());
-
+
Integer distance = (Integer)message.getProperty(ManagementHelper.HDR_DISTANCE);
if (distance == null)
@@ -489,37 +491,37 @@
{
case NotificationType.BINDING_ADDED_INDEX:
{
-
+
SimpleString queueAddress = (SimpleString)message.getProperty(ManagementHelper.HDR_ADDRESS);
-
+
if (queueAddress == null)
{
throw new IllegalStateException("queueAddress is null");
}
-
+
SimpleString clusterName = (SimpleString)message.getProperty(ManagementHelper.HDR_CLUSTER_NAME);
-
+
if (clusterName == null)
{
throw new IllegalStateException("clusterName is null");
}
-
+
SimpleString routingName = (SimpleString)message.getProperty(ManagementHelper.HDR_ROUTING_NAME);
-
+
if (routingName == null)
{
throw new IllegalStateException("routingName is null");
}
-
+
SimpleString filterString = (SimpleString)message.getProperty(ManagementHelper.HDR_FILTERSTRING);
-
+
Integer queueID = (Integer)message.getProperty(ManagementHelper.HDR_BINDING_ID);
-
+
if (queueID == null)
{
throw new IllegalStateException("queueID is null");
}
-
+
RemoteQueueBinding binding = new RemoteQueueBindingImpl(queueAddress,
clusterName,
routingName,
@@ -529,103 +531,104 @@
useDuplicateDetection,
bridge.getName(),
distance + 1);
-
+
bindings.put(clusterName, binding);
-
+
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
-
- log.warn("Remoting queue binding " + clusterName + " has already been bound in the post office. Most likely cause for this is you have a loop " +
+ // 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
+
+ log.warn("Remoting 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");
-
+
return;
}
-
-
+
postOffice.addBinding(binding);
-
+
Bindings theBindings = postOffice.getBindingsForAddress(queueAddress);
-
+
theBindings.setRouteWhenNoConsumers(routeWhenNoConsumers);
-
+
break;
}
case NotificationType.BINDING_REMOVED_INDEX:
- {
+ {
SimpleString clusterName = (SimpleString)message.getProperty(ManagementHelper.HDR_CLUSTER_NAME);
-
+
if (clusterName == null)
{
throw new IllegalStateException("clusterName is null");
}
-
+
RemoteQueueBinding binding = bindings.remove(clusterName);
-
+
if (binding == null)
{
throw new IllegalStateException("Cannot find binding for queue " + clusterName);
}
-
+
postOffice.removeBinding(binding.getUniqueName());
-
+
break;
}
case NotificationType.CONSUMER_CREATED_INDEX:
- {
+ {
SimpleString clusterName = (SimpleString)message.getProperty(ManagementHelper.HDR_CLUSTER_NAME);
-
+
if (clusterName == null)
{
throw new IllegalStateException("clusterName is null");
}
-
+
SimpleString filterString = (SimpleString)message.getProperty(ManagementHelper.HDR_FILTERSTRING);
-
+
RemoteQueueBinding binding = bindings.get(clusterName);
-
+
if (binding == null)
{
throw new IllegalStateException("Cannot find binding for " + clusterName);
}
-
+
binding.addConsumer(filterString);
-
+
message.putIntProperty(ManagementHelper.HDR_DISTANCE, distance + 1);
-
- //Need to propagate the consumer add
+
+ // Need to propagate the consumer add
Notification notification = new Notification(ntype, message.getProperties());
-
+
managementService.sendNotification(notification);
-
+
break;
}
case NotificationType.CONSUMER_CLOSED_INDEX:
- {
+ {
SimpleString clusterName = (SimpleString)message.getProperty(ManagementHelper.HDR_CLUSTER_NAME);
-
+
if (clusterName == null)
{
throw new IllegalStateException("clusterName is null");
}
-
+
SimpleString filterString = (SimpleString)message.getProperty(ManagementHelper.HDR_FILTERSTRING);
-
+
RemoteQueueBinding binding = bindings.get(clusterName);
-
+
if (binding == null)
{
throw new IllegalStateException("Cannot find binding for " + clusterName);
}
-
+
binding.removeConsumer(filterString);
-
+
message.putIntProperty(ManagementHelper.HDR_DISTANCE, distance + 1);
-
- //Need to propagate the consumer close
+
+ // Need to propagate the consumer close
Notification notification = new Notification(ntype, message.getProperties());
-
+
managementService.sendNotification(notification);
break;
@@ -639,7 +642,7 @@
}
private void clearBindings() throws Exception
- {
+ {
for (RemoteQueueBinding binding : bindings.values())
{
postOffice.removeBinding(binding.getUniqueName());
Modified: trunk/src/main/org/jboss/messaging/core/server/cluster/impl/ClusterManagerImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/server/cluster/impl/ClusterManagerImpl.java 2009-02-21 01:16:23 UTC (rev 5917)
+++ trunk/src/main/org/jboss/messaging/core/server/cluster/impl/ClusterManagerImpl.java 2009-02-21 14:22:13 UTC (rev 5918)
@@ -57,6 +57,7 @@
import org.jboss.messaging.util.ExecutorFactory;
import org.jboss.messaging.util.Pair;
import org.jboss.messaging.util.SimpleString;
+import org.jboss.messaging.util.UUID;
/**
* A ClusterManagerImpl
@@ -93,7 +94,7 @@
private final QueueFactory queueFactory;
- private final SimpleString nodeID;
+ private final UUID nodeUUID;
private volatile boolean started;
@@ -104,7 +105,7 @@
final ManagementService managementService,
final Configuration configuration,
final QueueFactory queueFactory,
- final SimpleString nodeID)
+ final UUID nodeUUID)
{
this.executorFactory = executorFactory;
@@ -120,7 +121,7 @@
this.queueFactory = queueFactory;
- this.nodeID = nodeID;
+ this.nodeUUID = nodeUUID;
}
public synchronized void start() throws Exception
@@ -202,7 +203,7 @@
{
return new HashMap<String, Bridge>(bridges);
}
-
+
public Set<ClusterConnection> getClusterConnections()
{
return new HashSet<ClusterConnection>(clusters.values());
@@ -222,7 +223,7 @@
InetAddress groupAddress = InetAddress.getByName(config.getGroupAddress());
- BroadcastGroupImpl group = new BroadcastGroupImpl(nodeID.toString(),
+ BroadcastGroupImpl group = new BroadcastGroupImpl(nodeUUID.toString(),
config.getName(),
localBindAddress,
config.getLocalBindPort(),
@@ -291,7 +292,7 @@
InetAddress groupAddress = InetAddress.getByName(config.getGroupAddress());
- DiscoveryGroup group = new DiscoveryGroupImpl(nodeID.toString(),
+ DiscoveryGroup group = new DiscoveryGroupImpl(nodeUUID.toString(),
config.getName(),
groupAddress,
config.getGroupPort(),
@@ -381,7 +382,8 @@
Pair<TransportConfiguration, TransportConfiguration> pair = new Pair<TransportConfiguration, TransportConfiguration>(connector,
backupConnector);
- bridge = new BridgeImpl(new SimpleString(config.getName()),
+ bridge = new BridgeImpl(nodeUUID,
+ new SimpleString(config.getName()),
queue,
pair,
executorFactory.getExecutor(),
@@ -477,7 +479,7 @@
queueFactory,
connectors,
config.getMaxHops(),
- nodeID);
+ nodeUUID);
}
else
{
@@ -505,7 +507,7 @@
queueFactory,
dg,
config.getMaxHops(),
- nodeID);
+ nodeUUID);
}
managementService.registerCluster(clusterConnection, config);
Modified: trunk/src/main/org/jboss/messaging/core/server/impl/GroupingRoundRobinDistributor.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/server/impl/GroupingRoundRobinDistributor.java 2009-02-21 01:16:23 UTC (rev 5917)
+++ trunk/src/main/org/jboss/messaging/core/server/impl/GroupingRoundRobinDistributor.java 2009-02-21 14:22:13 UTC (rev 5918)
@@ -22,6 +22,7 @@
package org.jboss.messaging.core.server.impl;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import org.jboss.messaging.core.message.impl.MessageImpl;
import org.jboss.messaging.core.server.Consumer;
@@ -38,6 +39,7 @@
* so on.
*
* @author <a href="mailto:andy.taylor at jboss.org">Andy Taylor</a>
+ * @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
*/
public class GroupingRoundRobinDistributor extends RoundRobinDistributor
{
@@ -47,34 +49,42 @@
// Attributes ----------------------------------------------------
- private ConcurrentHashMap<SimpleString, Consumer> cons = new ConcurrentHashMap<SimpleString, Consumer>();
+ private ConcurrentMap<SimpleString, Consumer> cons = new ConcurrentHashMap<SimpleString, Consumer>();
-
// Distributor implementation ------------------------------------
- public HandleStatus distribute(MessageReference reference)
+ public HandleStatus distribute(final MessageReference reference)
{
- if (getConsumerCount() == 0)
+ if (consumers.isEmpty())
{
return HandleStatus.BUSY;
}
- final SimpleString groupId = (SimpleString) reference.getMessage().getProperty(MessageImpl.HDR_GROUP_ID);
+
+ final SimpleString groupId = (SimpleString)reference.getMessage().getProperty(MessageImpl.HDR_GROUP_ID);
+
if (groupId != null)
{
int startPos = pos;
+
boolean filterRejected = false;
while (true)
{
- Consumer consumer = cons.putIfAbsent(groupId, consumers.get(pos));
+ Consumer consumer = consumers.get(pos);
+
+ Consumer oldConsumer = cons.putIfAbsent(groupId, consumer);
- if (consumer == null)
+ if (oldConsumer == null)
{
incrementPosition();
- consumer = cons.get(groupId);
}
+ else
+ {
+ consumer = oldConsumer;
+ }
+
+ HandleStatus status = handle(reference, consumer);
- HandleStatus status = handle(reference, consumer);
if (status == HandleStatus.HANDLED)
{
return HandleStatus.HANDLED;
@@ -85,10 +95,10 @@
}
else if (status == HandleStatus.BUSY)
{
- //if we were previously bound, we can remove and try the next consumer
+ // if we were previously bound, we can remove and try the next consumer
return HandleStatus.BUSY;
}
- //if we've tried all of them
+ // if we've tried all of them
if (startPos == pos)
{
// Tried all of them
@@ -113,6 +123,7 @@
public synchronized boolean removeConsumer(Consumer consumer)
{
boolean removed = super.removeConsumer(consumer);
+
if (removed)
{
for (SimpleString group : cons.keySet())
Modified: trunk/src/main/org/jboss/messaging/core/server/impl/MessagingServerImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/server/impl/MessagingServerImpl.java 2009-02-21 01:16:23 UTC (rev 5917)
+++ trunk/src/main/org/jboss/messaging/core/server/impl/MessagingServerImpl.java 2009-02-21 14:22:13 UTC (rev 5918)
@@ -71,8 +71,8 @@
import org.jboss.messaging.core.server.cluster.Transformer;
import org.jboss.messaging.core.server.cluster.impl.ClusterManagerImpl;
import org.jboss.messaging.core.settings.HierarchicalRepository;
+import org.jboss.messaging.core.settings.impl.AddressSettings;
import org.jboss.messaging.core.settings.impl.HierarchicalObjectRepository;
-import org.jboss.messaging.core.settings.impl.AddressSettings;
import org.jboss.messaging.core.transaction.ResourceManager;
import org.jboss.messaging.core.transaction.impl.ResourceManagerImpl;
import org.jboss.messaging.core.version.Version;
@@ -81,7 +81,7 @@
import org.jboss.messaging.util.OrderedExecutorFactory;
import org.jboss.messaging.util.Pair;
import org.jboss.messaging.util.SimpleString;
-import org.jboss.messaging.util.UUIDGenerator;
+import org.jboss.messaging.util.UUID;
import org.jboss.messaging.util.VersionLoader;
/**
@@ -105,6 +105,8 @@
// -----------------------------------------------------------------------------------
private SimpleString nodeID;
+
+ private UUID uuid;
private final Version version;
@@ -174,8 +176,6 @@
return;
}
- nodeID = UUIDGenerator.getInstance().generateSimpleStringUUID();
-
asyncDeliveryPool = Executors.newCachedThreadPool(new JBMThreadFactory("JBM-async-session-delivery-threads"));
executorFactory = new OrderedExecutorFactory(asyncDeliveryPool);
@@ -247,8 +247,7 @@
configuration.getIDCacheSize(),
configuration.isPersistIDCache(),
executorFactory,
- addressSettingsRepository,
- nodeID.toString());
+ addressSettingsRepository);
securityRepository = new HierarchicalObjectRepository<Set<Role>>();
securityRepository.setDefault(new HashSet<Role>());
@@ -263,7 +262,18 @@
managementService.setManagementNotificationAddress(configuration.getManagementNotificationAddress());
managementService.setClusterPassword(configuration.getManagementClusterPassword());
managementService.setManagementRequestTimeout(configuration.getManagementRequestTimeout());
+
+ List<QueueBindingInfo> queueBindingInfos = new ArrayList<QueueBindingInfo>();
+ List<SimpleString> destinations = new ArrayList<SimpleString>();
+ storageManager.loadBindingJournal(queueBindingInfos, destinations);
+
+ uuid = storageManager.getPersistentID();
+
+ nodeID = new SimpleString(uuid.toString());
+
+ log.info("*** messaging server node id is " + nodeID);
+
serverManagement = managementService.registerServer(postOffice,
storageManager,
configuration,
@@ -274,11 +284,8 @@
this,
queueFactory);
- List<QueueBindingInfo> queueBindingInfos = new ArrayList<QueueBindingInfo>();
- List<SimpleString> destinations = new ArrayList<SimpleString>();
-
- storageManager.loadBindingJournal(queueBindingInfos, destinations);
-
+
+
// FIXME the destination corresponding to the notification address is always created
// so that queues can be created wether the address is allowable or not (to revisit later)
if (!postOffice.containsDestination(configuration.getManagementNotificationAddress()))
@@ -387,7 +394,7 @@
managementService,
configuration,
queueFactory,
- nodeID);
+ uuid);
clusterManager.start();
}
@@ -760,6 +767,11 @@
{
return nodeID;
}
+
+ public UUID getUUID()
+ {
+ return uuid;
+ }
// Public
// ---------------------------------------------------------------------------------------
Modified: trunk/src/main/org/jboss/messaging/core/server/impl/RoundRobinDistributor.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/server/impl/RoundRobinDistributor.java 2009-02-21 01:16:23 UTC (rev 5917)
+++ trunk/src/main/org/jboss/messaging/core/server/impl/RoundRobinDistributor.java 2009-02-21 14:22:13 UTC (rev 5918)
@@ -65,8 +65,11 @@
return HandleStatus.BUSY;
}
int startPos = pos;
+
boolean filterRejected = false;
+
HandleStatus status;
+
while (true)
{
status = handle(reference, getNextConsumer());
@@ -95,16 +98,19 @@
}
}
- protected synchronized Consumer getNextConsumer()
+ private final synchronized Consumer getNextConsumer()
{
Consumer consumer = consumers.get(pos);
+
incrementPosition();
+
return consumer;
}
protected void incrementPosition()
{
pos++;
+
if (pos == consumers.size())
{
pos = 0;
Modified: trunk/src/main/org/jboss/messaging/util/UUID.java
===================================================================
--- trunk/src/main/org/jboss/messaging/util/UUID.java 2009-02-21 01:16:23 UTC (rev 5917)
+++ trunk/src/main/org/jboss/messaging/util/UUID.java 2009-02-21 14:22:13 UTC (rev 5918)
@@ -88,7 +88,7 @@
* @param data
* 16 byte UUID contents
*/
- UUID(int type, byte[] data)
+ public UUID(int type, byte[] data)
{
for (int i = 0; i < 16; ++i)
{
Modified: trunk/tests/src/org/jboss/messaging/tests/integration/cluster/bridge/BridgeStartTest.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/tests/integration/cluster/bridge/BridgeStartTest.java 2009-02-21 01:16:23 UTC (rev 5917)
+++ trunk/tests/src/org/jboss/messaging/tests/integration/cluster/bridge/BridgeStartTest.java 2009-02-21 14:22:13 UTC (rev 5918)
@@ -196,7 +196,8 @@
service1.stop();
}
-
+
+
public void testTargetServerUpAndDown() throws Exception
{
//This test needs to use real files, since it requires duplicate detection, since when the target server is shutdown, messages will get resent when it is started, so the dup id cache needs
@@ -383,7 +384,7 @@
service1.stop();
}
}
-
+
public void testTargetServerNotAvailableNoReconnectTries() throws Exception
{
Map<String, Object> service0Params = new HashMap<String, Object>();
Modified: trunk/tests/src/org/jboss/messaging/tests/integration/management/DiscoveryGroupControlTest.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/tests/integration/management/DiscoveryGroupControlTest.java 2009-02-21 01:16:23 UTC (rev 5917)
+++ trunk/tests/src/org/jboss/messaging/tests/integration/management/DiscoveryGroupControlTest.java 2009-02-21 14:22:13 UTC (rev 5918)
@@ -103,39 +103,12 @@
discoveryGroupControl.stop();
assertFalse(discoveryGroupControl.isStarted());
- }
-
- public void testStoppedDiscoveryGroupCanNotBeRestarted() throws Exception
- {
- DiscoveryGroupConfiguration discoveryGroupConfig = new DiscoveryGroupConfiguration(randomString(), "231.7.7.7", 2000, randomPositiveLong());
-
- MBeanServer mbeanServer = MBeanServerFactory.createMBeanServer();
- Configuration conf = new ConfigurationImpl();
- conf.setSecurityEnabled(false);
- conf.setJMXManagementEnabled(true);
- conf.setClustered(true);
- conf.getDiscoveryGroupConfigurations().put(discoveryGroupConfig.getName(), discoveryGroupConfig);
- service = Messaging.newNullStorageMessagingService(conf, mbeanServer);
- service.start();
-
- DiscoveryGroupControlMBean discoveryGroupControl = createDiscoveryGroupControl(discoveryGroupConfig.getName(), mbeanServer);
-
- // started by the service
- assertTrue(discoveryGroupControl.isStarted());
-
- discoveryGroupControl.stop();
- assertFalse(discoveryGroupControl.isStarted());
- try
- {
- discoveryGroupControl.start();
- fail("once stopped, a discovery group can not be restarted");
- }
- catch (Exception e)
- {
- }
+ discoveryGroupControl.start();
+ assertTrue(discoveryGroupControl.isStarted());
}
+
// Package protected ---------------------------------------------
// Protected -----------------------------------------------------
More information about the jboss-cvs-commits
mailing list