[jboss-cvs] JBoss Messaging SVN: r5023 - in trunk: src/main/org/jboss/messaging/core/persistence and 6 other directories.
jboss-cvs-commits at lists.jboss.org
jboss-cvs-commits at lists.jboss.org
Wed Sep 24 19:08:48 EDT 2008
Author: clebert.suconic at jboss.com
Date: 2008-09-24 19:08:48 -0400 (Wed, 24 Sep 2008)
New Revision: 5023
Added:
trunk/src/main/org/jboss/messaging/util/SequenceGenerator.java
trunk/tests/src/org/jboss/messaging/tests/unit/util/SequenceGeneratorTest.java
Modified:
trunk/src/main/org/jboss/messaging/core/paging/impl/PageMessageImpl.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/server/impl/ServerSessionPacketHandler.java
trunk/tests/src/org/jboss/messaging/tests/unit/core/paging/impl/PageImplTestBase.java
trunk/tests/src/org/jboss/messaging/tests/unit/core/persistence/impl/journal/JournalStorageManagerTest.java
Log:
Adding SequenceGenerator
Modified: trunk/src/main/org/jboss/messaging/core/paging/impl/PageMessageImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/paging/impl/PageMessageImpl.java 2008-09-24 12:44:36 UTC (rev 5022)
+++ trunk/src/main/org/jboss/messaging/core/paging/impl/PageMessageImpl.java 2008-09-24 23:08:48 UTC (rev 5023)
@@ -82,19 +82,21 @@
public void decode(final MessagingBuffer buffer)
{
transactionID = buffer.getLong();
+ message.setMessageID(buffer.getLong());
message.decode(buffer);
}
public void encode(final MessagingBuffer buffer)
{
buffer.putLong(transactionID);
+ buffer.putLong(message.getMessageID());
message.encode(buffer);
}
public int getEncodeSize()
{
- return 8 + message.getEncodeSize();
+ return 8 * 2 + message.getEncodeSize();
}
// Package protected ---------------------------------------------
Modified: trunk/src/main/org/jboss/messaging/core/persistence/StorageManager.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/persistence/StorageManager.java 2008-09-24 12:44:36 UTC (rev 5022)
+++ trunk/src/main/org/jboss/messaging/core/persistence/StorageManager.java 2008-09-24 23:08:48 UTC (rev 5023)
@@ -18,22 +18,27 @@
* 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.persistence;
+import java.util.List;
+import java.util.Map;
+
+import javax.transaction.xa.Xid;
+
import org.jboss.messaging.core.paging.LastPageRecord;
import org.jboss.messaging.core.paging.PageTransactionInfo;
import org.jboss.messaging.core.postoffice.Binding;
import org.jboss.messaging.core.postoffice.PostOffice;
-import org.jboss.messaging.core.server.*;
+import org.jboss.messaging.core.server.MessageReference;
+import org.jboss.messaging.core.server.MessagingComponent;
+import org.jboss.messaging.core.server.Queue;
+import org.jboss.messaging.core.server.QueueFactory;
+import org.jboss.messaging.core.server.ServerMessage;
import org.jboss.messaging.core.transaction.ResourceManager;
import org.jboss.messaging.util.SimpleString;
-import javax.transaction.xa.Xid;
-import java.util.List;
-import java.util.Map;
-
/**
*
* A StorageManager
@@ -44,62 +49,51 @@
public interface StorageManager extends MessagingComponent
{
- // Message related operations
-
+ // Message related operations
+
long generateID();
-
- void setMaxID(long id);
-
+
long generateTransactionID();
-
-
+
void storeMessage(ServerMessage message) throws Exception;
-
+
void storeAcknowledge(long queueID, long messageID) throws Exception;
-
+
void storeDelete(long messageID) throws Exception;
-
-
+
void storeMessageTransactional(long txID, ServerMessage message) throws Exception;
-
+
void storeAcknowledgeTransactional(long txID, long queueID, long messageiD) throws Exception;
-
+
void storeDeleteMessageTransactional(long txID, long queueID, long messageID) throws Exception;
-
+
/** Used to delete non-messaging data (such as PageTransaction and LasPage) */
void storeDeleteTransactional(long txID, long recordID) throws Exception;
-
-
+
void prepare(long txID, Xid xid) throws Exception;
-
+
void commit(long txID) throws Exception;
-
+
void rollback(long txID) throws Exception;
-
-
+
void storePageTransaction(long txID, PageTransactionInfo pageTransaction) throws Exception;
-
void storeLastPage(long txID, LastPageRecord pageTransaction) throws Exception;
-
-
- void updateDeliveryCount(MessageReference ref) throws Exception;
-
+
+ void updateDeliveryCount(MessageReference ref) throws Exception;
+
void loadMessages(PostOffice postOffice, Map<Long, Queue> queues, ResourceManager resourceManager) throws Exception;
-
-
+
// Bindings related operations
-
+
void addBinding(Binding binding) throws Exception;
-
+
void deleteBinding(Binding binding) throws Exception;
-
+
boolean addDestination(SimpleString destination) throws Exception;
-
+
boolean deleteDestination(SimpleString destination) throws Exception;
-
-
- void loadBindings(QueueFactory queueFactory, List<Binding> bindings,
- List<SimpleString> destinations) throws Exception;
-
+
+ void loadBindings(QueueFactory queueFactory, List<Binding> bindings, List<SimpleString> destinations) throws Exception;
+
}
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 2008-09-24 12:44:36 UTC (rev 5022)
+++ trunk/src/main/org/jboss/messaging/core/persistence/impl/journal/JournalStorageManager.java 2008-09-24 23:08:48 UTC (rev 5023)
@@ -18,14 +18,29 @@
* 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.persistence.impl.journal;
+import java.io.File;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.transaction.xa.Xid;
+
import org.jboss.messaging.core.config.Configuration;
import org.jboss.messaging.core.filter.Filter;
import org.jboss.messaging.core.filter.impl.FilterImpl;
-import org.jboss.messaging.core.journal.*;
+import org.jboss.messaging.core.journal.EncodingSupport;
+import org.jboss.messaging.core.journal.Journal;
+import org.jboss.messaging.core.journal.PreparedTransactionInfo;
+import org.jboss.messaging.core.journal.RecordInfo;
+import org.jboss.messaging.core.journal.SequentialFileFactory;
import org.jboss.messaging.core.journal.impl.AIOSequentialFileFactory;
import org.jboss.messaging.core.journal.impl.JournalImpl;
import org.jboss.messaging.core.journal.impl.NIOSequentialFileFactory;
@@ -42,23 +57,18 @@
import org.jboss.messaging.core.remoting.impl.ByteBufferWrapper;
import org.jboss.messaging.core.remoting.impl.wireformat.XidCodecSupport;
import org.jboss.messaging.core.remoting.spi.MessagingBuffer;
-import org.jboss.messaging.core.server.*;
+import org.jboss.messaging.core.server.JournalType;
+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.ServerMessage;
import org.jboss.messaging.core.server.impl.ServerMessageImpl;
import org.jboss.messaging.core.transaction.ResourceManager;
import org.jboss.messaging.core.transaction.Transaction;
import org.jboss.messaging.core.transaction.impl.TransactionImpl;
+import org.jboss.messaging.util.SequenceGenerator;
import org.jboss.messaging.util.SimpleString;
-import javax.transaction.xa.Xid;
-import java.io.*;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.atomic.AtomicLong;
-
/**
*
* A JournalStorageManager
@@ -70,87 +80,86 @@
*/
public class JournalStorageManager implements StorageManager
{
- private static final Logger log = Logger.getLogger(JournalStorageManager.class);
-
- private static final int SIZE_LONG = 8;
-
+ private static final Logger log = Logger.getLogger(JournalStorageManager.class);
+
+ private static final int SIZE_LONG = 8;
+
private static final int SIZE_INT = 4;
-
+
private static final int SIZE_BYTE = 1;
-
+
// Bindings journal record type
-
- public static final byte BINDING_RECORD = 21;
-
- public static final byte DESTINATION_RECORD = 22;
-
+
+ public static final byte BINDING_RECORD = 21;
+
+ public static final byte DESTINATION_RECORD = 22;
+
// type + expiration + timestamp + priority
- public static final int SIZE_FIELDS = SIZE_INT + SIZE_LONG + SIZE_LONG + SIZE_BYTE;
-
+ public static final int SIZE_FIELDS = SIZE_INT + SIZE_LONG + SIZE_LONG + SIZE_BYTE;
+
// Message journal record types
-
+
public static final byte ADD_MESSAGE = 31;
-
+
public static final byte ACKNOWLEDGE_REF = 32;
-
+
public static final byte UPDATE_DELIVERY_COUNT = 33;
-
+
public static final byte PAGE_TRANSACTION = 34;
-
+
public static final byte LAST_PAGE = 35;
-
+
public static final byte SET_SCHEDULED_DELIVERY_TIME = 44;
-
- private final AtomicLong idSequence = new AtomicLong(0);
-
- private final AtomicLong bindingIDSequence = new AtomicLong(0);
-
- private final Journal messageJournal;
-
- private final Journal bindingsJournal;
-
- private final ConcurrentMap<SimpleString, Long> destinationIDMap = new ConcurrentHashMap<SimpleString, Long>();
-
- private volatile boolean started;
- public JournalStorageManager(final Configuration config)
- {
- if (config.getJournalType() != JournalType.NIO && config.getJournalType() != JournalType.ASYNCIO)
- {
- throw new IllegalArgumentException("Only NIO and AsyncIO are supported journals");
- }
-
- String bindingsDir = config.getBindingsDirectory();
-
- if (bindingsDir == null)
- {
- throw new NullPointerException("bindings-dir is null");
- }
-
- checkAndCreateDir(bindingsDir, config.isCreateBindingsDir());
-
- SequentialFileFactory bindingsFF = new NIOSequentialFileFactory(bindingsDir);
-
- bindingsJournal = new JournalImpl(1024 * 1024, 2, true, true, bindingsFF, "jbm-bindings", "bindings", 1, -1);
-
- String journalDir = config.getJournalDirectory();
-
- if (journalDir == null)
- {
- throw new NullPointerException("journal-dir is null");
- }
-
- checkAndCreateDir(journalDir, config.isCreateBindingsDir());
-
+ private final SequenceGenerator idSequence = new SequenceGenerator();
+
+ private final AtomicLong bindingIDSequence = new AtomicLong(0);
+
+ private final Journal messageJournal;
+
+ private final Journal bindingsJournal;
+
+ private final ConcurrentMap<SimpleString, Long> destinationIDMap = new ConcurrentHashMap<SimpleString, Long>();
+
+ private volatile boolean started;
+
+ public JournalStorageManager(final Configuration config)
+ {
+ if (config.getJournalType() != JournalType.NIO && config.getJournalType() != JournalType.ASYNCIO)
+ {
+ throw new IllegalArgumentException("Only NIO and AsyncIO are supported journals");
+ }
+
+ String bindingsDir = config.getBindingsDirectory();
+
+ if (bindingsDir == null)
+ {
+ throw new NullPointerException("bindings-dir is null");
+ }
+
+ checkAndCreateDir(bindingsDir, config.isCreateBindingsDir());
+
+ SequentialFileFactory bindingsFF = new NIOSequentialFileFactory(bindingsDir);
+
+ bindingsJournal = new JournalImpl(1024 * 1024, 2, true, true, bindingsFF, "jbm-bindings", "bindings", 1, -1);
+
+ String journalDir = config.getJournalDirectory();
+
+ if (journalDir == null)
+ {
+ throw new NullPointerException("journal-dir is null");
+ }
+
+ checkAndCreateDir(journalDir, config.isCreateBindingsDir());
+
SequentialFileFactory journalFF = null;
-
+
if (config.getJournalType() == JournalType.ASYNCIO)
{
log.info("AIO journal selected");
if (!AIOSequentialFileFactory.isSupported())
{
- log.warn("AIO wasn't located on this platform, it will fall back to using pure Java NIO. " +
- "If your platform is Linux, install LibAIO to enable the AIO journal");
+ log.warn("AIO wasn't located on this platform, it will fall back to using pure Java NIO. " + "If your platform is Linux, install LibAIO to enable the AIO journal");
journalFF = new NIOSequentialFileFactory(journalDir);
}
else
@@ -170,227 +179,229 @@
// Sanity check only... this is previously tested
throw new IllegalArgumentException("JDBC Journal is not supported yet");
}
-
- messageJournal = new JournalImpl(config.getJournalFileSize(),
- config.getJournalMinFiles(), config.isJournalSyncTransactional(),
- config.isJournalSyncNonTransactional(), journalFF,
- "jbm-data", "jbm", config.getJournalMaxAIO(), config.getJournalBufferReuseSize());
- }
-
- /* This constructor is only used for testing */
- public JournalStorageManager(final Journal messageJournal, final Journal bindingsJournal)
+
+ messageJournal = new JournalImpl(config.getJournalFileSize(),
+ config.getJournalMinFiles(),
+ config.isJournalSyncTransactional(),
+ config.isJournalSyncNonTransactional(),
+ journalFF,
+ "jbm-data",
+ "jbm",
+ config.getJournalMaxAIO(),
+ config.getJournalBufferReuseSize());
+ }
+
+ /* This constructor is only used for testing */
+ public JournalStorageManager(final Journal messageJournal, final Journal bindingsJournal)
{
- this.messageJournal = messageJournal;
- this.bindingsJournal = bindingsJournal;
+ this.messageJournal = messageJournal;
+ this.bindingsJournal = bindingsJournal;
}
-
- public long generateID()
- {
- return idSequence.getAndIncrement();
- }
-
- //Needed for replication
-
- //TODO can be optimised
- public synchronized void setMaxID(final long id)
- {
- if (1 + id > idSequence.get())
- {
- idSequence.set(id + 1);
- }
- }
-
- public long generateTransactionID()
- {
- return messageJournal.getTransactionID();
- }
-
- // Non transactional operations
-
- public void storeMessage(final ServerMessage message) throws Exception
- {
- messageJournal.appendAddRecord(message.getMessageID(), ADD_MESSAGE, message);
- }
- public void storeAcknowledge(final long queueID, final long messageID) throws Exception
- {
- messageJournal.appendUpdateRecord(messageID, ACKNOWLEDGE_REF, new ACKEncoding(queueID));
- }
-
- public void storeDelete(final long messageID) throws Exception
- {
- messageJournal.appendDeleteRecord(messageID);
- }
-
- // Transactional operations
-
- public void storeMessageTransactional(long txID, ServerMessage message) throws Exception
+ public long generateID()
{
+ return idSequence.generateID();
+ }
+
+ // Needed for replication
+
+ public long generateTransactionID()
+ {
+ return messageJournal.getTransactionID();
+ }
+
+ // Non transactional operations
+
+ public void storeMessage(final ServerMessage message) throws Exception
+ {
+ messageJournal.appendAddRecord(message.getMessageID(), ADD_MESSAGE, message);
+ }
+
+ public void storeAcknowledge(final long queueID, final long messageID) throws Exception
+ {
+ messageJournal.appendUpdateRecord(messageID, ACKNOWLEDGE_REF, new ACKEncoding(queueID));
+ }
+
+ public void storeDelete(final long messageID) throws Exception
+ {
+ messageJournal.appendDeleteRecord(messageID);
+ }
+
+ // Transactional operations
+
+ public void storeMessageTransactional(final long txID, final ServerMessage message) throws Exception
+ {
messageJournal.appendAddRecordTransactional(txID, message.getMessageID(), ADD_MESSAGE, message);
}
- public void storePageTransaction(long txID, PageTransactionInfo pageTransaction) throws Exception
+ public void storePageTransaction(final long txID, final PageTransactionInfo pageTransaction) throws Exception
{
if (pageTransaction.getRecordID() != 0)
{
- // Instead of updating the record, we delete the old one as that is better for reclaiming
+ // Instead of updating the record, we delete the old one as that is
+ // better for reclaiming
messageJournal.appendDeleteRecordTransactional(txID, pageTransaction.getRecordID(), null);
}
-
+
pageTransaction.setRecordID(generateID());
-
- messageJournal.appendAddRecordTransactional(txID, pageTransaction.getRecordID(), PAGE_TRANSACTION, pageTransaction);
+
+ messageJournal.appendAddRecordTransactional(txID,
+ pageTransaction.getRecordID(),
+ PAGE_TRANSACTION,
+ pageTransaction);
}
-
- public void storeLastPage(long txID, LastPageRecord lastPage) throws Exception
+
+ public void storeLastPage(final long txID, final LastPageRecord lastPage) throws Exception
{
if (lastPage.getRecordId() != 0)
{
- // To avoid linked list effect on reclaiming, we delete and add a new record, instead of simply updating it
+ // To avoid linked list effect on reclaiming, we delete and add a new
+ // record, instead of simply updating it
messageJournal.appendDeleteRecordTransactional(txID, lastPage.getRecordId(), null);
}
-
+
lastPage.setRecordId(generateID());
-
+
messageJournal.appendAddRecordTransactional(txID, lastPage.getRecordId(), LAST_PAGE, lastPage);
}
- public void storeAcknowledgeTransactional(long txID, long queueID, long messageID) throws Exception
+ public void storeAcknowledgeTransactional(final long txID, final long queueID, final long messageID) throws Exception
{
- messageJournal.appendUpdateRecordTransactional(txID, messageID, ACKNOWLEDGE_REF, new ACKEncoding(queueID));
+ messageJournal.appendUpdateRecordTransactional(txID, messageID, ACKNOWLEDGE_REF, new ACKEncoding(queueID));
}
-
- public void storeDeleteTransactional(long txID, long recordID) throws Exception
+
+ public void storeDeleteTransactional(final long txID, final long recordID) throws Exception
{
- messageJournal.appendDeleteRecordTransactional(txID, recordID, null);
+ messageJournal.appendDeleteRecordTransactional(txID, recordID, null);
}
-
- public void storeDeleteMessageTransactional(long txID, long queueID, long messageID) throws Exception
+
+ public void storeDeleteMessageTransactional(final long txID, final long queueID, final long messageID) throws Exception
{
messageJournal.appendDeleteRecordTransactional(txID, messageID, new DeleteEncoding(queueID));
}
-
- public void prepare(long txID, Xid xid) throws Exception
+
+ public void prepare(final long txID, final Xid xid) throws Exception
{
- messageJournal.appendPrepareRecord(txID, new XidEncoding(xid));
+ messageJournal.appendPrepareRecord(txID, new XidEncoding(xid));
}
-
- public void commit(long txID) throws Exception
+
+ public void commit(final long txID) throws Exception
{
- messageJournal.appendCommitRecord(txID);
+ messageJournal.appendCommitRecord(txID);
}
-
- public void rollback(long txID) throws Exception
+
+ public void rollback(final long txID) throws Exception
{
messageJournal.appendRollbackRecord(txID);
}
-
+
// Other operations
-
- public void updateDeliveryCount(final MessageReference ref) throws Exception
- {
- DeliveryCountUpdateEncoding updateInfo = new DeliveryCountUpdateEncoding(ref.getQueue().getPersistenceID(), ref.getDeliveryCount());
-
- messageJournal.appendUpdateRecord(ref.getMessage().getMessageID(), UPDATE_DELIVERY_COUNT, updateInfo);
- }
-
- public void loadMessages(final PostOffice postOffice, final Map<Long, Queue> queues, ResourceManager resourceManager) throws Exception
- {
- List<RecordInfo> records = new ArrayList<RecordInfo>();
-
- List<PreparedTransactionInfo> preparedTransactions = new ArrayList<PreparedTransactionInfo>();
-
- long maxID = messageJournal.load(records, preparedTransactions);
-
- idSequence.set(maxID + 1);
- for (RecordInfo record: records)
- {
- byte[] data = record.data;
-
- ByteBuffer bb = ByteBuffer.wrap(data);
+ public void updateDeliveryCount(final MessageReference ref) throws Exception
+ {
+ DeliveryCountUpdateEncoding updateInfo = new DeliveryCountUpdateEncoding(ref.getQueue().getPersistenceID(),
+ ref.getDeliveryCount());
- MessagingBuffer buff = new ByteBufferWrapper(bb);
-
- byte recordType = record.getUserRecordType();
-
- switch (recordType)
- {
- case ADD_MESSAGE:
- {
- ServerMessage message = new ServerMessageImpl(record.id);
-
- message.decode(buff);
+ messageJournal.appendUpdateRecord(ref.getMessage().getMessageID(), UPDATE_DELIVERY_COUNT, updateInfo);
+ }
- List<MessageReference> refs = postOffice.route(message);
+ public void loadMessages(final PostOffice postOffice,
+ final Map<Long, Queue> queues,
+ final ResourceManager resourceManager) throws Exception
+ {
+ List<RecordInfo> records = new ArrayList<RecordInfo>();
- for (MessageReference ref: refs)
- {
- ref.getQueue().addLast(ref);
- }
-
- break;
- }
- case ACKNOWLEDGE_REF:
- {
+ List<PreparedTransactionInfo> preparedTransactions = new ArrayList<PreparedTransactionInfo>();
+
+ messageJournal.load(records, preparedTransactions);
+
+ for (RecordInfo record : records)
+ {
+ byte[] data = record.data;
+
+ ByteBuffer bb = ByteBuffer.wrap(data);
+
+ MessagingBuffer buff = new ByteBufferWrapper(bb);
+
+ byte recordType = record.getUserRecordType();
+
+ switch (recordType)
+ {
+ case ADD_MESSAGE:
+ {
+ ServerMessage message = new ServerMessageImpl(record.id);
+
+ message.decode(buff);
+
+ List<MessageReference> refs = postOffice.route(message);
+
+ for (MessageReference ref : refs)
+ {
+ ref.getQueue().addLast(ref);
+ }
+
+ break;
+ }
+ case ACKNOWLEDGE_REF:
+ {
long messageID = record.id;
ACKEncoding encoding = new ACKEncoding();
-
- encoding.decode(buff);
-
- Queue queue = queues.get(encoding.queueID);
-
- if (queue == null)
- {
- throw new IllegalStateException("Cannot find queue with id " + encoding.queueID);
- }
-
- MessageReference removed = queue.removeReferenceWithID(messageID);
-
- if (removed == null)
- {
- throw new IllegalStateException("Failed to remove reference for " + messageID);
- }
-
- break;
- }
- case UPDATE_DELIVERY_COUNT:
- {
- long messageID = record.id;
-
- DeliveryCountUpdateEncoding deliveryUpdate = new DeliveryCountUpdateEncoding();
-
- deliveryUpdate.decode(buff);
-
- Queue queue = queues.get(deliveryUpdate.queueID);
-
- if (queue == null)
- {
- throw new IllegalStateException("Cannot find queue with id " + deliveryUpdate.queueID);
- }
-
- MessageReference reference = queue.getReference(messageID);
-
- if (reference == null)
- {
- throw new IllegalStateException("Failed to find reference for " + messageID);
- }
-
- reference.setDeliveryCount(deliveryUpdate.count);
-
- break;
- }
+
+ encoding.decode(buff);
+
+ Queue queue = queues.get(encoding.queueID);
+
+ if (queue == null)
+ {
+ throw new IllegalStateException("Cannot find queue with id " + encoding.queueID);
+ }
+
+ MessageReference removed = queue.removeReferenceWithID(messageID);
+
+ if (removed == null)
+ {
+ throw new IllegalStateException("Failed to remove reference for " + messageID);
+ }
+
+ break;
+ }
+ case UPDATE_DELIVERY_COUNT:
+ {
+ long messageID = record.id;
+
+ DeliveryCountUpdateEncoding deliveryUpdate = new DeliveryCountUpdateEncoding();
+
+ deliveryUpdate.decode(buff);
+
+ Queue queue = queues.get(deliveryUpdate.queueID);
+
+ if (queue == null)
+ {
+ throw new IllegalStateException("Cannot find queue with id " + deliveryUpdate.queueID);
+ }
+
+ MessageReference reference = queue.getReference(messageID);
+
+ if (reference == null)
+ {
+ throw new IllegalStateException("Failed to find reference for " + messageID);
+ }
+
+ reference.setDeliveryCount(deliveryUpdate.count);
+
+ break;
+ }
case PAGE_TRANSACTION:
- {
+ {
PageTransactionInfoImpl pageTransactionInfo = new PageTransactionInfoImpl();
-
+
pageTransactionInfo.decode(buff);
-
+
pageTransactionInfo.setRecordID(record.id);
-
+
PagingManager pagingManager = postOffice.getPagingManager();
-
+
pagingManager.addTransaction(pageTransactionInfo);
break;
@@ -398,235 +409,240 @@
case LAST_PAGE:
{
LastPageRecordImpl recordImpl = new LastPageRecordImpl();
-
+
recordImpl.setRecordId(record.id);
-
+
recordImpl.decode(buff);
-
+
PagingManager pagingManager = postOffice.getPagingManager();
-
+
pagingManager.setLastPage(recordImpl);
-
+
break;
}
- case SET_SCHEDULED_DELIVERY_TIME:
- {
- //TODO
- }
- default:
- {
- throw new IllegalStateException("Invalid record type " + recordType);
- }
- }
- }
-
- loadPreparedTransactions(postOffice, queues, resourceManager,preparedTransactions);
-
+ case SET_SCHEDULED_DELIVERY_TIME:
+ {
+ // TODO
+ }
+ default:
+ {
+ throw new IllegalStateException("Invalid record type " + recordType);
+ }
+ }
+ }
+
+ loadPreparedTransactions(postOffice, queues, resourceManager, preparedTransactions);
+
}
- //Bindings operations
-
- public void addBinding(Binding binding) throws Exception
- {
- Queue queue = binding.getQueue();
+ // Bindings operations
- //We generate the queue id here
-
- long queueID = bindingIDSequence.getAndIncrement();
+ public void addBinding(final Binding binding) throws Exception
+ {
+ Queue queue = binding.getQueue();
- queue.setPersistenceID(queueID);
-
- final SimpleString filterString;
-
- final Filter filter = queue.getFilter();
-
- if (filter != null)
- {
- filterString = filter.getFilterString();
- }
- else
- {
- filterString = null;
- }
-
- BindingEncoding bindingEncoding = new BindingEncoding(binding.getQueue().getName(), binding.getAddress(), filterString);
-
- bindingsJournal.appendAddRecord(queueID, BINDING_RECORD, bindingEncoding);
- }
-
- public void deleteBinding(Binding binding) throws Exception
- {
- long id = binding.getQueue().getPersistenceID();
-
- if (id == -1)
- {
- throw new IllegalArgumentException("Cannot delete binding, id is " + id);
- }
-
- bindingsJournal.appendDeleteRecord(id);
- }
-
- public boolean addDestination(final SimpleString destination) throws Exception
- {
- long destinationID = bindingIDSequence.getAndIncrement();
-
- if (destinationIDMap.putIfAbsent(destination, destinationID) != null)
- {
- //Already exists
- return false;
- }
- else
- {
- DestinationEncoding destinationEnc = new DestinationEncoding(destination);
-
- bindingsJournal.appendAddRecord(destinationID, DESTINATION_RECORD, destinationEnc);
-
- return true;
- }
- }
-
- public boolean deleteDestination(final SimpleString destination) throws Exception
- {
- Long destinationID = destinationIDMap.remove(destination);
-
- if (destinationID == null)
- {
- return false;
- }
- else
- {
- bindingsJournal.appendDeleteRecord(destinationID);
-
- return true;
- }
- }
-
- public void loadBindings(final QueueFactory queueFactory,
- final List<Binding> bindings, final List<SimpleString> destinations) throws Exception
- {
- List<RecordInfo> records = new ArrayList<RecordInfo>();
-
- List<PreparedTransactionInfo> preparedTransactions = new ArrayList<PreparedTransactionInfo>();
-
- long maxID = bindingsJournal.load(records, preparedTransactions);
+ // We generate the queue id here
- for (RecordInfo record: records)
- {
- long id = record.id;
-
+ long queueID = bindingIDSequence.getAndIncrement();
+
+ queue.setPersistenceID(queueID);
+
+ final SimpleString filterString;
+
+ final Filter filter = queue.getFilter();
+
+ if (filter != null)
+ {
+ filterString = filter.getFilterString();
+ }
+ else
+ {
+ filterString = null;
+ }
+
+ BindingEncoding bindingEncoding = new BindingEncoding(binding.getQueue().getName(),
+ binding.getAddress(),
+ filterString);
+
+ bindingsJournal.appendAddRecord(queueID, BINDING_RECORD, bindingEncoding);
+ }
+
+ public void deleteBinding(final Binding binding) throws Exception
+ {
+ long id = binding.getQueue().getPersistenceID();
+
+ if (id == -1)
+ {
+ throw new IllegalArgumentException("Cannot delete binding, id is " + id);
+ }
+
+ bindingsJournal.appendDeleteRecord(id);
+ }
+
+ public boolean addDestination(final SimpleString destination) throws Exception
+ {
+ long destinationID = bindingIDSequence.getAndIncrement();
+
+ if (destinationIDMap.putIfAbsent(destination, destinationID) != null)
+ {
+ // Already exists
+ return false;
+ }
+ else
+ {
+ DestinationEncoding destinationEnc = new DestinationEncoding(destination);
+
+ bindingsJournal.appendAddRecord(destinationID, DESTINATION_RECORD, destinationEnc);
+
+ return true;
+ }
+ }
+
+ public boolean deleteDestination(final SimpleString destination) throws Exception
+ {
+ Long destinationID = destinationIDMap.remove(destination);
+
+ if (destinationID == null)
+ {
+ return false;
+ }
+ else
+ {
+ bindingsJournal.appendDeleteRecord(destinationID);
+
+ return true;
+ }
+ }
+
+ public void loadBindings(final QueueFactory queueFactory,
+ final List<Binding> bindings,
+ final List<SimpleString> destinations) throws Exception
+ {
+ List<RecordInfo> records = new ArrayList<RecordInfo>();
+
+ List<PreparedTransactionInfo> preparedTransactions = new ArrayList<PreparedTransactionInfo>();
+
+ long maxID = bindingsJournal.load(records, preparedTransactions);
+
+ for (RecordInfo record : records)
+ {
+ long id = record.id;
+
MessagingBuffer buffer = new ByteBufferWrapper(ByteBuffer.wrap(record.data));
- byte rec = record.getUserRecordType();
-
- if (rec == BINDING_RECORD)
- {
- BindingEncoding encodeBinding = new BindingEncoding();
-
- encodeBinding.decode(buffer);
-
- Filter filter = null;
-
- if (encodeBinding.filter != null)
- {
- filter = new FilterImpl(encodeBinding.filter);
- }
-
- Queue queue = queueFactory.createQueue(id, encodeBinding.queueName, filter, true, false);
-
- Binding binding = new BindingImpl(encodeBinding.address, queue);
-
- bindings.add(binding);
- }
- else if (rec == DESTINATION_RECORD)
- {
- DestinationEncoding destEnc = new DestinationEncoding();
-
- destEnc.decode(buffer);
+ byte rec = record.getUserRecordType();
- destinationIDMap.put(destEnc.destination, id);
-
- destinations.add(destEnc.destination);
- }
- else
- {
- throw new IllegalStateException("Invalid record type " + rec);
- }
- }
-
- bindingIDSequence.set(maxID + 1);
- }
-
-
- // MessagingComponent implementation ------------------------------------------------------
+ if (rec == BINDING_RECORD)
+ {
+ BindingEncoding encodeBinding = new BindingEncoding();
- public synchronized void start() throws Exception
- {
- if (started)
- {
- return;
- }
-
- bindingsJournal.start();
-
- messageJournal.start();
-
- started = true;
- }
+ encodeBinding.decode(buffer);
- public synchronized void stop() throws Exception
- {
- if (!started)
- {
- return;
- }
-
- bindingsJournal.stop();
-
- messageJournal.stop();
-
- started = false;
- }
-
- public synchronized boolean isStarted()
- {
- return started;
- }
-
- // Public -----------------------------------------------------------------------------------
-
- public Journal getMessageJournal()
- {
- return messageJournal;
- }
-
- public Journal getBindingsJournal()
- {
- return bindingsJournal;
- }
-
- // Private ----------------------------------------------------------------------------------
-
-
+ Filter filter = null;
+
+ if (encodeBinding.filter != null)
+ {
+ filter = new FilterImpl(encodeBinding.filter);
+ }
+
+ Queue queue = queueFactory.createQueue(id, encodeBinding.queueName, filter, true, false);
+
+ Binding binding = new BindingImpl(encodeBinding.address, queue);
+
+ bindings.add(binding);
+ }
+ else if (rec == DESTINATION_RECORD)
+ {
+ DestinationEncoding destEnc = new DestinationEncoding();
+
+ destEnc.decode(buffer);
+
+ destinationIDMap.put(destEnc.destination, id);
+
+ destinations.add(destEnc.destination);
+ }
+ else
+ {
+ throw new IllegalStateException("Invalid record type " + rec);
+ }
+ }
+
+ bindingIDSequence.set(maxID + 1);
+ }
+
+ // MessagingComponent implementation
+ // ------------------------------------------------------
+
+ public synchronized void start() throws Exception
+ {
+ if (started)
+ {
+ return;
+ }
+
+ bindingsJournal.start();
+
+ messageJournal.start();
+
+ started = true;
+ }
+
+ public synchronized void stop() throws Exception
+ {
+ if (!started)
+ {
+ return;
+ }
+
+ bindingsJournal.stop();
+
+ messageJournal.stop();
+
+ started = false;
+ }
+
+ public synchronized boolean isStarted()
+ {
+ return started;
+ }
+
+ // Public
+ // -----------------------------------------------------------------------------------
+
+ public Journal getMessageJournal()
+ {
+ return messageJournal;
+ }
+
+ public Journal getBindingsJournal()
+ {
+ return bindingsJournal;
+ }
+
+ // Private
+ // ----------------------------------------------------------------------------------
+
private void loadPreparedTransactions(final PostOffice postOffice,
- final Map<Long, Queue> queues, ResourceManager resourceManager,
- List<PreparedTransactionInfo> preparedTransactions) throws Exception
+ final Map<Long, Queue> queues,
+ final ResourceManager resourceManager,
+ final List<PreparedTransactionInfo> preparedTransactions) throws Exception
{
- //recover prepared transactions
+ // recover prepared transactions
for (PreparedTransactionInfo preparedTransaction : preparedTransactions)
{
XidEncoding encodingXid = new XidEncoding(preparedTransaction.extraData);
-
+
Xid xid = encodingXid.xid;
Transaction tx = new TransactionImpl(preparedTransaction.id, xid, this, postOffice);
-
+
List<MessageReference> messages = new ArrayList<MessageReference>();
-
+
List<MessageReference> messagesToAck = new ArrayList<MessageReference>();
-
+
PageTransactionInfoImpl pageTransactionInfo = null;
-
- //first get any sent messages for this tx and recreate
+
+ // first get any sent messages for this tx and recreate
for (RecordInfo record : preparedTransaction.records)
{
byte[] data = record.data;
@@ -637,7 +653,7 @@
byte recordType = record.getUserRecordType();
- switch(recordType)
+ switch (recordType)
{
case ADD_MESSAGE:
{
@@ -646,9 +662,9 @@
message.decode(buff);
List<MessageReference> refs = postOffice.route(message);
-
+
messages.addAll(refs);
-
+
break;
}
case ACKNOWLEDGE_REF:
@@ -656,7 +672,7 @@
long messageID = record.id;
ACKEncoding encoding = new ACKEncoding();
-
+
encoding.decode(buff);
Queue queue = queues.get(encoding.queueID);
@@ -669,29 +685,30 @@
MessageReference removed = queue.removeReferenceWithID(messageID);
messagesToAck.add(removed);
-
+
if (removed == null)
{
throw new IllegalStateException("Failed to remove reference for " + messageID);
}
-
+
break;
}
case PAGE_TRANSACTION:
{
pageTransactionInfo = new PageTransactionInfoImpl();
-
+
pageTransactionInfo.decode(buff);
-
+
pageTransactionInfo.markIncomplete();
-
+
break;
}
default:
- log.warn("InternalError: Record type " + recordType + " not recognized. Maybe you're using journal files created on a different version" );
+ log.warn("InternalError: Record type " + recordType +
+ " not recognized. Maybe you're using journal files created on a different version");
}
}
-
+
for (RecordInfo record : preparedTransaction.recordsToDelete)
{
byte[] data = record.data;
@@ -703,7 +720,7 @@
long messageID = record.id;
DeleteEncoding encoding = new DeleteEncoding();
-
+
encoding.decode(buff);
Queue queue = queues.get(encoding.queueID);
@@ -716,72 +733,73 @@
MessageReference removed = queue.removeReferenceWithID(messageID);
messagesToAck.add(removed);
-
+
if (removed == null)
{
throw new IllegalStateException("Failed to remove reference for " + messageID);
}
}
-
- //now we recreate the state of the tx and add to the resource manager
+
+ // now we recreate the state of the tx and add to the resource manager
tx.replay(messages, messagesToAck, pageTransactionInfo, Transaction.State.PREPARED);
-
+
resourceManager.putTransaction(xid, tx);
}
}
-
- private void checkAndCreateDir(String dir, boolean create)
- {
- File f = new File(dir);
-
- if (!f.exists())
- {
- log.info("Directory " + dir + " does not already exists");
-
- if (create)
- {
- log.info("Creating it");
-
- if (!f.mkdirs())
- {
- throw new IllegalStateException("Failed to create directory " + dir);
- }
- }
- else
- {
- log.info("Not creating it");
-
- throw new IllegalArgumentException("Directory " + dir + " does not exist and will not create it");
- }
- }
- else
- {
- log.info("Directory " + dir + " already exists");
- }
- }
-
- // Inner Classes ----------------------------------------------------------------------------
-
- private static class XidEncoding implements EncodingSupport
- {
- final Xid xid;
-
- XidEncoding(Xid xid)
- {
- this.xid = xid;
- }
-
- XidEncoding(byte[] data)
- {
- xid = XidCodecSupport.decodeXid(new ByteBufferWrapper(ByteBuffer.wrap(data)));
- }
-
- public void decode(MessagingBuffer buffer)
+
+ private void checkAndCreateDir(final String dir, final boolean create)
+ {
+ File f = new File(dir);
+
+ if (!f.exists())
{
+ log.info("Directory " + dir + " does not already exists");
+
+ if (create)
+ {
+ log.info("Creating it");
+
+ if (!f.mkdirs())
+ {
+ throw new IllegalStateException("Failed to create directory " + dir);
+ }
+ }
+ else
+ {
+ log.info("Not creating it");
+
+ throw new IllegalArgumentException("Directory " + dir + " does not exist and will not create it");
+ }
+ }
+ else
+ {
+ log.info("Directory " + dir + " already exists");
+ }
+ }
+
+ // Inner Classes
+ // ----------------------------------------------------------------------------
+
+ private static class XidEncoding implements EncodingSupport
+ {
+ final Xid xid;
+
+ XidEncoding(final Xid xid)
+ {
+ this.xid = xid;
+ }
+
+ XidEncoding(final byte[] data)
+ {
+ xid = XidCodecSupport.decodeXid(new ByteBufferWrapper(ByteBuffer.wrap(data)));
+ }
+
+ public void decode(final MessagingBuffer buffer)
+ {
throw new IllegalStateException("Non Supported Operation");
}
- public void encode(MessagingBuffer buffer)
+ public void encode(final MessagingBuffer buffer)
{
XidCodecSupport.encodeXid(xid, buffer);
}
@@ -789,21 +807,22 @@
public int getEncodeSize()
{
return XidCodecSupport.getXidEncodeLength(xid);
- }
- }
+ }
+ }
private static class BindingEncoding implements EncodingSupport
- {
+ {
SimpleString queueName;
+
SimpleString address;
+
SimpleString filter;
public BindingEncoding()
- {
+ {
}
-
- public BindingEncoding(SimpleString queueName,
- SimpleString address, SimpleString filter)
+
+ public BindingEncoding(final SimpleString queueName, final SimpleString address, final SimpleString filter)
{
super();
this.queueName = queueName;
@@ -811,14 +830,14 @@
this.filter = filter;
}
- public void decode(MessagingBuffer buffer)
+ public void decode(final MessagingBuffer buffer)
{
queueName = buffer.getSimpleString();
address = buffer.getSimpleString();
- filter = buffer.getNullableSimpleString();
+ filter = buffer.getNullableSimpleString();
}
- public void encode(MessagingBuffer buffer)
+ public void encode(final MessagingBuffer buffer)
{
buffer.putSimpleString(queueName);
buffer.putSimpleString(address);
@@ -827,9 +846,7 @@
public int getEncodeSize()
{
- return SimpleString.sizeofString(queueName) +
- SimpleString.sizeofString(address) +
- 1 + // HasFilter?
+ return SimpleString.sizeofString(queueName) + SimpleString.sizeofString(address) + 1 + // HasFilter?
((filter != null) ? SimpleString.sizeofString(filter) : 0);
}
}
@@ -837,22 +854,22 @@
private static class DestinationEncoding implements EncodingSupport
{
SimpleString destination;
-
- DestinationEncoding(SimpleString destination)
+
+ DestinationEncoding(final SimpleString destination)
{
this.destination = destination;
}
-
+
DestinationEncoding()
{
}
-
- public void decode(MessagingBuffer buffer)
+
+ public void decode(final MessagingBuffer buffer)
{
- this.destination = buffer.getSimpleString();
+ destination = buffer.getSimpleString();
}
- public void encode(MessagingBuffer buffer)
+ public void encode(final MessagingBuffer buffer)
{
buffer.putSimpleString(destination);
}
@@ -861,33 +878,34 @@
{
return SimpleString.sizeofString(destination);
}
-
+
}
-
+
private static class DeliveryCountUpdateEncoding implements EncodingSupport
{
long queueID;
+
int count;
-
+
public DeliveryCountUpdateEncoding()
{
super();
}
-
- public DeliveryCountUpdateEncoding(long queueID, int count)
+
+ public DeliveryCountUpdateEncoding(final long queueID, final int count)
{
super();
this.queueID = queueID;
this.count = count;
}
- public void decode(MessagingBuffer buffer)
+ public void decode(final MessagingBuffer buffer)
{
queueID = buffer.getLong();
count = buffer.getInt();
}
- public void encode(MessagingBuffer buffer)
+ public void encode(final MessagingBuffer buffer)
{
buffer.putLong(queueID);
buffer.putInt(count);
@@ -896,14 +914,14 @@
public int getEncodeSize()
{
return 8 + 4;
- }
+ }
}
-
+
private static class QueueEncoding implements EncodingSupport
{
long queueID;
-
- public QueueEncoding(long queueID)
+
+ public QueueEncoding(final long queueID)
{
super();
this.queueID = queueID;
@@ -914,12 +932,12 @@
super();
}
- public void decode(MessagingBuffer buffer)
+ public void decode(final MessagingBuffer buffer)
{
- this.queueID = buffer.getLong();
+ queueID = buffer.getLong();
}
- public void encode(MessagingBuffer buffer)
+ public void encode(final MessagingBuffer buffer)
{
buffer.putLong(queueID);
}
@@ -927,9 +945,9 @@
public int getEncodeSize()
{
return 8;
- }
+ }
}
-
+
private static class DeleteEncoding extends QueueEncoding
{
public DeleteEncoding()
@@ -937,12 +955,12 @@
super();
}
- public DeleteEncoding(long queueID)
+ public DeleteEncoding(final long queueID)
{
super(queueID);
- }
+ }
}
-
+
private static class ACKEncoding extends QueueEncoding
{
public ACKEncoding()
@@ -950,9 +968,9 @@
super();
}
- public ACKEncoding(long queueID)
+ public ACKEncoding(final long queueID)
{
super(queueID);
}
- }
+ }
}
Modified: trunk/src/main/org/jboss/messaging/core/server/impl/ServerSessionPacketHandler.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/server/impl/ServerSessionPacketHandler.java 2008-09-24 12:44:36 UTC (rev 5022)
+++ trunk/src/main/org/jboss/messaging/core/server/impl/ServerSessionPacketHandler.java 2008-09-24 23:08:48 UTC (rev 5023)
@@ -73,7 +73,6 @@
import org.jboss.messaging.core.remoting.ChannelHandler;
import org.jboss.messaging.core.remoting.Packet;
import org.jboss.messaging.core.remoting.impl.wireformat.MessagingExceptionMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionAcknowledgeMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionAddDestinationMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionBindingQueryMessage;
@@ -186,7 +185,7 @@
doHandlePacket(packet);
}
else
- {
+ {
Runnable action = new Runnable()
{
public void run()
@@ -422,9 +421,10 @@
break;
}
case SESS_SEND:
- {
+ {
SessionSendMessage message = (SessionSendMessage)packet;
- // log.info("Got send " + message.getServerMessage().getMessageID());
+ // log.info("Got send " +
+ // message.getServerMessage().getMessageID());
session.sendProducerMessage(message.getProducerID(), message.getServerMessage());
if (message.isRequiresResponse())
{
@@ -433,10 +433,9 @@
break;
}
case SESS_REPLICATE_SEND:
- {
+ {
SessionReplicateSendMessage message = (SessionReplicateSendMessage)packet;
//log.info("Got replicated send " + message.getServerMessage().getMessageID());
- storageManager.setMaxID(message.getServerMessage().getMessageID());
session.sendProducerMessage(message.getProducerID(), message.getServerMessage());
break;
}
Added: trunk/src/main/org/jboss/messaging/util/SequenceGenerator.java
===================================================================
--- trunk/src/main/org/jboss/messaging/util/SequenceGenerator.java (rev 0)
+++ trunk/src/main/org/jboss/messaging/util/SequenceGenerator.java 2008-09-24 23:08:48 UTC (rev 5023)
@@ -0,0 +1,132 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2005-2008, Red Hat Middleware LLC, and individual contributors
+ * by the @authors tag. See the copyright.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * 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.util;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * A SequenceGenerator
+ *
+ * @author <a href="mailto:clebert.suconic at jboss.org">Clebert Suconic</a>
+ *
+ * Created Sep 24, 2008 11:54:10 AM
+ *
+ *
+ */
+public class SequenceGenerator
+{
+
+ // (0x7fffffff) We take one bit out, as we don't want negative numbers
+ // (take out the signal bit before merging the numbers)
+ private static final long MASK_TIME = Integer.MAX_VALUE;
+
+ // Attributes ----------------------------------------------------
+
+ /**
+ * Using a long just to avoid making cast conversions on every ID generated
+ */
+ private final AtomicLong counter = new AtomicLong(0);
+
+ private volatile long tmMark;
+
+ // Static --------------------------------------------------------
+
+ // Constructors --------------------------------------------------
+
+ public SequenceGenerator()
+ {
+ refresh();
+ }
+
+ // Public --------------------------------------------------------
+
+ public long generateID()
+ {
+
+ long value = counter.incrementAndGet();
+
+ if (value >= Integer.MAX_VALUE)
+ {
+ synchronized (this)
+ {
+ if (counter.get() >= Integer.MAX_VALUE)
+ {
+ refresh();
+ }
+ value = counter.incrementAndGet();
+ }
+ }
+
+ return tmMark | value;
+ }
+
+ public void setInternalID(final long id)
+ {
+ counter.set(id);
+ }
+
+ public synchronized void refresh()
+ {
+ long newTm = newTM();
+
+ // To avoid quick restarts.
+ // This shouldn't ever happen.
+ // I doubt any system will be able to generate more than Integer.MAX_VALUE
+ // ids per millisecond.
+ // This would be used only on testcases validating the logic of the class
+ while (newTm <= tmMark)
+ {
+ System.out.println("Equals!!!!");
+ try
+ {
+ Thread.sleep(20);
+ }
+ catch (InterruptedException e)
+ {
+ }
+ newTm = newTM();
+ }
+ tmMark = newTm;
+ counter.set(0);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "SequenceGenerator(tmMark=" + String.format("%1$X", tmMark) + ", counter = " + counter.get() + ")";
+ }
+
+ // Package protected ---------------------------------------------
+
+ // Protected -----------------------------------------------------
+
+ // Private -------------------------------------------------------
+
+ private long newTM()
+ {
+ return (System.currentTimeMillis() & MASK_TIME) << 32;
+ }
+
+ // Inner classes -------------------------------------------------
+
+}
Modified: trunk/tests/src/org/jboss/messaging/tests/unit/core/paging/impl/PageImplTestBase.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/tests/unit/core/paging/impl/PageImplTestBase.java 2008-09-24 12:44:36 UTC (rev 5022)
+++ trunk/tests/src/org/jboss/messaging/tests/unit/core/paging/impl/PageImplTestBase.java 2008-09-24 23:08:48 UTC (rev 5023)
@@ -119,7 +119,7 @@
for (int i = 0; i < msgs.length; i++)
{
- assertEquals(0, msgs[i].getMessage().getMessageID());
+ assertEquals(i, msgs[i].getMessage().getMessageID());
assertEquals(simpleDestination, msgs[i].getMessage().getDestination());
Modified: trunk/tests/src/org/jboss/messaging/tests/unit/core/persistence/impl/journal/JournalStorageManagerTest.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/tests/unit/core/persistence/impl/journal/JournalStorageManagerTest.java 2008-09-24 12:44:36 UTC (rev 5022)
+++ trunk/tests/src/org/jboss/messaging/tests/unit/core/persistence/impl/journal/JournalStorageManagerTest.java 2008-09-24 23:08:48 UTC (rev 5023)
@@ -405,8 +405,6 @@
EasyMock.verify(refs1.toArray());
EasyMock.verify(refs2.toArray());
EasyMock.verify(queue1, queue2, queue3);
-
- assertEquals(msg1ID + 1, jsm.generateID());
}
public void testAddBindingWithFilter() throws Exception
@@ -702,7 +700,11 @@
Queue queue1 = EasyMock.createStrictMock(Queue.class);
Queue queue2 = EasyMock.createStrictMock(Queue.class);
Queue queue3 = EasyMock.createStrictMock(Queue.class);
- EasyMock.expect(qf.createQueue(EasyMock.eq(0L), EasyMock.eq(squeue1), EasyMock.isA(Filter.class), EasyMock.eq(true), EasyMock.eq(false))).andReturn(queue1);
+ EasyMock.expect(qf.createQueue(EasyMock.eq(0L),
+ EasyMock.eq(squeue1),
+ EasyMock.isA(Filter.class),
+ EasyMock.eq(true),
+ EasyMock.eq(false))).andReturn(queue1);
EasyMock.expect(qf.createQueue(1L, squeue2, null, true, false)).andReturn(queue1);
EasyMock.expect(qf.createQueue(2L, squeue3, null, true, false)).andReturn(queue1);
@@ -783,18 +785,19 @@
public void testGenerateMessageID()
{
- long id = 0;
Journal messageJournal = EasyMock.createStrictMock(Journal.class);
Journal bindingsJournal = EasyMock.createStrictMock(Journal.class);
JournalStorageManager jsm = new JournalStorageManager(messageJournal, bindingsJournal);
- assertEquals(id++, jsm.generateID());
- assertEquals(id++, jsm.generateID());
- assertEquals(id++, jsm.generateID());
- assertEquals(id++, jsm.generateID());
- assertEquals(id++, jsm.generateID());
+ long id = jsm.generateID();
+
+ assertEquals(++id, jsm.generateID());
+ assertEquals(++id, jsm.generateID());
+ assertEquals(++id, jsm.generateID());
+ assertEquals(++id, jsm.generateID());
+ assertEquals(++id, jsm.generateID());
}
public void testGenerateTransactionID()
Added: trunk/tests/src/org/jboss/messaging/tests/unit/util/SequenceGeneratorTest.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/tests/unit/util/SequenceGeneratorTest.java (rev 0)
+++ trunk/tests/src/org/jboss/messaging/tests/unit/util/SequenceGeneratorTest.java 2008-09-24 23:08:48 UTC (rev 5023)
@@ -0,0 +1,164 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2005-2008, Red Hat Middleware LLC, and individual contributors
+ * by the @authors tag. See the copyright.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * 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.tests.unit.util;
+
+import java.util.concurrent.CountDownLatch;
+
+import org.jboss.messaging.tests.util.UnitTestCase;
+import org.jboss.messaging.util.ConcurrentHashSet;
+import org.jboss.messaging.util.SequenceGenerator;
+
+/**
+ * A SequenceGeneratorTest
+ *
+ * @author <a href="mailto:clebert.suconic at jboss.org">Clebert Suconic</a>
+ *
+ * Created 24-Sep-08 3:42:25 PM
+ *
+ *
+ */
+public class SequenceGeneratorTest extends UnitTestCase
+{
+
+ // Constants -----------------------------------------------------
+
+ // Attributes ----------------------------------------------------
+
+ // Static --------------------------------------------------------
+
+ // Constructors --------------------------------------------------
+
+ // Public --------------------------------------------------------
+
+ public void testCalculation()
+ {
+ SequenceGenerator seq = new SequenceGenerator();
+ long max = 100000;
+
+ long lastNr = 0;
+
+ for (long i = 0; i < max; i++)
+ {
+ if (i % 1 == 1000)
+ {
+ seq.refresh();
+ }
+
+ long seqNr = seq.generateID();
+
+ assertTrue("The sequence generator should aways generate crescent numbers", seqNr > lastNr);
+
+ lastNr = seqNr;
+ }
+
+ }
+
+ public void testCalculationOnMultiThread() throws Throwable
+ {
+
+ for (int i = 0; i < 10; i++)
+ {
+ internaltestCalculationOnMultiThread();
+ }
+ }
+
+ public void internaltestCalculationOnMultiThread() throws Throwable
+ {
+ final ConcurrentHashSet<Long> hashSet = new ConcurrentHashSet<Long>();
+
+ final SequenceGenerator seq = new SequenceGenerator();
+
+ seq.setInternalID(Integer.MAX_VALUE - 50);
+
+ final int NUMBER_OF_THREADS = 100;
+
+ final int NUMBER_OF_IDS = 10;
+
+ final CountDownLatch latchAlign = new CountDownLatch(NUMBER_OF_THREADS);
+
+ final CountDownLatch latchStart = new CountDownLatch(1);
+
+ class T1 extends Thread
+ {
+ Throwable e;
+
+ @Override
+ public void run()
+ {
+ try
+ {
+ latchAlign.countDown();
+ latchStart.await();
+
+ long lastValue = 0l;
+ for (int i = 0; i < NUMBER_OF_IDS; i++)
+ {
+ long value = seq.generateID();
+ assertTrue(hex(value) + " should be greater than " + hex(lastValue) + " on seq " + seq.toString(),
+ value > lastValue);
+ lastValue = value;
+
+ hashSet.add(value);
+ }
+ }
+ catch (Throwable e)
+ {
+ this.e = e;
+ }
+ }
+
+ };
+
+ T1[] arrays = new T1[NUMBER_OF_THREADS];
+
+ for (int i = 0; i < arrays.length; i++)
+ {
+ arrays[i] = new T1();
+ arrays[i].start();
+ }
+
+ latchAlign.await();
+
+ latchStart.countDown();
+
+ for (T1 t : arrays)
+ {
+ t.join();
+ if (t.e != null)
+ {
+ throw t.e;
+ }
+ }
+
+ assertEquals(NUMBER_OF_THREADS * NUMBER_OF_IDS, hashSet.size());
+
+ hashSet.clear();
+
+ }
+
+ private static String hex(final long value)
+ {
+ return String.format("%1$X", value);
+ }
+
+}
More information about the jboss-cvs-commits
mailing list