November 2009 - hornetq-commits - Jboss List Archives

JBoss hornetq SVN: r8231 - trunk/tests/src/org/hornetq/tests/unit/core/paging/impl.

by do-not-reply＠jboss.org

Author: clebert.suconic(a)jboss.com Date: 2009-11-05 22:16:44 -0500 (Thu, 05 Nov 2009) New Revision: 8231 Modified: trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java Log: no more need for those repetitions Modified: trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java 2009-11-06 02:01:22 UTC (rev 8230) +++ trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java 2009-11-06 03:16:44 UTC (rev 8231) @@ -167,12 +167,8 @@ public void testPageWithNIO() throws Exception { - // This integration test could fail 1 in 100 due to race conditions. - for (int i = 0; i < 100; i++) - { - recreateDirectory(getTestDir()); - testConcurrentPaging(new NIOSequentialFileFactory(getTestDir()), 1); - } + recreateDirectory(getTestDir()); + testConcurrentPaging(new NIOSequentialFileFactory(getTestDir()), 1); } public void testStore() throws Exception @@ -495,9 +491,9 @@ long id = messageIdGenerator.incrementAndGet(); // Each thread will Keep paging until all the messages are depaged. - // This is possible because the depage thread is not actually reading the pages. + // This is possible because the depage thread is not actually reading the pages. // Just using the internal API to remove it from the page file system - ServerMessage msg = createMessage(storeImpl, destination, createRandomBuffer(id, 5)); + ServerMessage msg = createMessage(storeImpl, destination, createRandomBuffer(id, 5)); if (storeImpl.page(msg, false)) { buffers.put(id, msg); @@ -538,7 +534,7 @@ { // Wait every producer to produce at least one message latchStart.await(); - + while (aliveProducers.get() > 0) { Page page = storeImpl.depage(); @@ -614,7 +610,8 @@ { SequentialFile fileTmp = factory.createSequentialFile(file, 1); fileTmp.open(); - assertTrue("The page file size (" + fileTmp.size() + ") shouldn't be > " + MAX_SIZE, fileTmp.size() <= MAX_SIZE); + assertTrue("The page file size (" + fileTmp.size() + ") shouldn't be > " + MAX_SIZE, + fileTmp.size() <= MAX_SIZE); fileTmp.close(); } @@ -705,12 +702,14 @@ return new FakePostOffice(); } - private ServerMessage createMessage(final PagingStore store, final SimpleString destination, final HornetQBuffer buffer) + private ServerMessage createMessage(final PagingStore store, + final SimpleString destination, + final HornetQBuffer buffer) { ServerMessage msg = new ServerMessageImpl((byte)1, true, 0, System.currentTimeMillis(), (byte)0, buffer); msg.setDestination(destination); - + msg.setPagingStore(store); return msg;

16 years, 2 months

1
0
0 / 0

JBoss hornetq SVN: r8230 - in trunk: src/main/org/hornetq/core/journal/impl and 4 other directories.

by do-not-reply＠jboss.org

Author: clebert.suconic(a)jboss.com Date: 2009-11-05 21:01:22 -0500 (Thu, 05 Nov 2009) New Revision: 8230 Modified: trunk/src/main/org/hornetq/core/journal/SequentialFileFactory.java trunk/src/main/org/hornetq/core/journal/impl/AIOSequentialFileFactory.java trunk/src/main/org/hornetq/core/journal/impl/AbstractSequentialFactory.java trunk/src/main/org/hornetq/core/journal/impl/JournalCleaner.java trunk/src/main/org/hornetq/core/journal/impl/JournalImpl.java trunk/src/main/org/hornetq/core/journal/impl/NIOSequentialFileFactory.java trunk/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java trunk/tests/src/org/hornetq/tests/integration/client/PagingTest.java trunk/tests/src/org/hornetq/tests/unit/core/journal/impl/fakes/FakeSequentialFileFactory.java trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java Log: Fixing PageStoreTest and few other minor tweaks Modified: trunk/src/main/org/hornetq/core/journal/SequentialFileFactory.java =================================================================== --- trunk/src/main/org/hornetq/core/journal/SequentialFileFactory.java 2009-11-05 19:01:30 UTC (rev 8229) +++ trunk/src/main/org/hornetq/core/journal/SequentialFileFactory.java 2009-11-06 02:01:22 UTC (rev 8230) @@ -39,9 +39,9 @@ /** The factory may need to do some initialization before the file is activated. * this was added as a hook for AIO to initialize the Observer on TimedBuffer. * It could be eventually done the same on NIO if we implement TimedBuffer on NIO */ - void activate(SequentialFile file); + void activateBuffer(SequentialFile file); - void deactivate(SequentialFile file); + void deactivateBuffer(); // To be used in tests only ByteBuffer wrapBuffer(byte[] bytes); Modified: trunk/src/main/org/hornetq/core/journal/impl/AIOSequentialFileFactory.java =================================================================== --- trunk/src/main/org/hornetq/core/journal/impl/AIOSequentialFileFactory.java 2009-11-05 19:01:30 UTC (rev 8229) +++ trunk/src/main/org/hornetq/core/journal/impl/AIOSequentialFileFactory.java 2009-11-06 02:01:22 UTC (rev 8230) @@ -36,10 +36,10 @@ */ public class AIOSequentialFileFactory extends AbstractSequentialFactory { - + // Timeout used to wait executors to shutdown private static final int EXECUTOR_TIMEOUT = 60; - + private static final Logger log = Logger.getLogger(AIOSequentialFileFactory.class); private static final boolean trace = log.isTraceEnabled(); @@ -77,21 +77,22 @@ } public AIOSequentialFileFactory(final String journalDir, - int bufferSize, - long bufferTimeout, - boolean flushOnSync, - boolean logRates) + final int bufferSize, + final long bufferTimeout, + final boolean flushOnSync, + final boolean logRates) { super(journalDir); this.bufferSize = bufferSize; this.bufferTimeout = bufferTimeout; - this.timedBuffer = new TimedBuffer(bufferSize, bufferTimeout, flushOnSync, logRates); + timedBuffer = new TimedBuffer(bufferSize, bufferTimeout, flushOnSync, logRates); } /* (non-Javadoc) * @see org.hornetq.core.journal.SequentialFileFactory#activate(org.hornetq.core.journal.SequentialFile) */ - public void activate(SequentialFile file) + @Override + public void activateBuffer(final SequentialFile file) { final AIOSequentialFile sequentialFile = (AIOSequentialFile)file; timedBuffer.disableAutoFlush(); @@ -105,12 +106,14 @@ } } + @Override public void flush() { timedBuffer.flush(); } - public void deactivate(SequentialFile file) + @Override + public void deactivateBuffer() { timedBuffer.flush(); timedBuffer.setObserver(null); @@ -179,33 +182,37 @@ /* (non-Javadoc) * @see org.hornetq.core.journal.SequentialFileFactory#releaseBuffer(java.nio.ByteBuffer) */ - public void releaseBuffer(ByteBuffer buffer) + @Override + public void releaseBuffer(final ByteBuffer buffer) { AsynchronousFileImpl.destroyBuffer(buffer); } + @Override public void start() { timedBuffer.start(); - + writeExecutor = Executors.newSingleThreadExecutor(new HornetQThreadFactory("HornetQ-AIO-writer-pool" + System.identityHashCode(this), - true)); + true)); pollerExecutor = Executors.newCachedThreadPool(new HornetQThreadFactory("HornetQ-AIO-poller-pool" + System.identityHashCode(this), - true)); + true)); - } + @Override public void stop() { buffersControl.stop(); + timedBuffer.stop(); - - this.writeExecutor.shutdown(); + + writeExecutor.shutdown(); + try { - if (!this.writeExecutor.awaitTermination(EXECUTOR_TIMEOUT, TimeUnit.SECONDS)) + if (!writeExecutor.awaitTermination(EXECUTOR_TIMEOUT, TimeUnit.SECONDS)) { log.warn("Timed out on AIO writer shutdown", new Exception("Timed out on AIO writer shutdown")); } @@ -213,12 +220,12 @@ catch (InterruptedException e) { } - - this.pollerExecutor.shutdown(); + pollerExecutor.shutdown(); + try { - if (!this.pollerExecutor.awaitTermination(EXECUTOR_TIMEOUT, TimeUnit.SECONDS)) + if (!pollerExecutor.awaitTermination(EXECUTOR_TIMEOUT, TimeUnit.SECONDS)) { log.warn("Timed out on AIO poller shutdown", new Exception("Timed out on AIO writer shutdown")); } @@ -228,9 +235,10 @@ } } + @Override protected void finalize() { - this.stop(); + stop(); } /** Class that will control buffer-reuse */ @@ -255,7 +263,9 @@ if (bufferSize > 0 && System.currentTimeMillis() - bufferReuseLastTime > 10000) { if (trace) + { trace("Clearing reuse buffers queue with " + reuseBuffersQueue.size() + " elements"); + } bufferReuseLastTime = System.currentTimeMillis(); Modified: trunk/src/main/org/hornetq/core/journal/impl/AbstractSequentialFactory.java =================================================================== --- trunk/src/main/org/hornetq/core/journal/impl/AbstractSequentialFactory.java 2009-11-05 19:01:30 UTC (rev 8229) +++ trunk/src/main/org/hornetq/core/journal/impl/AbstractSequentialFactory.java 2009-11-06 02:01:22 UTC (rev 8230) @@ -52,7 +52,7 @@ { } - public void activate(SequentialFile file) + public void activateBuffer(SequentialFile file) { } @@ -60,7 +60,7 @@ { } - public void deactivate(SequentialFile file) + public void deactivateBuffer() { } Modified: trunk/src/main/org/hornetq/core/journal/impl/JournalCleaner.java =================================================================== --- trunk/src/main/org/hornetq/core/journal/impl/JournalCleaner.java 2009-11-05 19:01:30 UTC (rev 8229) +++ trunk/src/main/org/hornetq/core/journal/impl/JournalCleaner.java 2009-11-06 02:01:22 UTC (rev 8230) @@ -213,15 +213,15 @@ * Commits and rollbacks are also counted as negatives. We need to fix those also. * @param dependencies */ - public void fixDependencies(final JournalFile originalFile, final ArrayList<JournalFile> dependencies) throws Exception + public void fixDependencies(final JournalFile originalFile, final ArrayList<JournalFile> dependencies) throws Exception { for (JournalFile dependency : dependencies) { fixDependency(originalFile, dependency); } - + } - + // Package protected --------------------------------------------- // Protected ----------------------------------------------------- @@ -256,15 +256,17 @@ { JournalReaderCallback txfix = new JournalReaderCallbackAbstract() { - public void onReadCommitRecord(long transactionID, int numberOfRecords) throws Exception + @Override + public void onReadCommitRecord(final long transactionID, final int numberOfRecords) throws Exception { if (transactionCounter.containsKey(transactionID)) { dependency.incNegCount(originalFile); } } - - public void onReadRollbackRecord(long transactionID) throws Exception + + @Override + public void onReadRollbackRecord(final long transactionID) throws Exception { if (transactionCounter.containsKey(transactionID)) { @@ -272,11 +274,10 @@ } } }; - + JournalImpl.readJournalFile(fileFactory, dependency, txfix); } - // Inner classes ------------------------------------------------- } Modified: trunk/src/main/org/hornetq/core/journal/impl/JournalImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/journal/impl/JournalImpl.java 2009-11-05 19:01:30 UTC (rev 8229) +++ trunk/src/main/org/hornetq/core/journal/impl/JournalImpl.java 2009-11-06 02:01:22 UTC (rev 8230) @@ -2058,7 +2058,7 @@ openFile(currentFile, true); } - fileFactory.activate(currentFile.getFile()); + fileFactory.activateBuffer(currentFile.getFile()); pushOpenedFile(); @@ -2559,7 +2559,7 @@ log.warn("Couldn't stop journal executor after 60 seconds"); } - fileFactory.flush(); + fileFactory.deactivateBuffer(); if (currentFile != null && currentFile.getFile().isOpen()) { @@ -3037,7 +3037,7 @@ trace("moveNextFile: " + currentFile.getFile().getFileName() + " sync: " + synchronous); } - fileFactory.activate(currentFile.getFile()); + fileFactory.activateBuffer(currentFile.getFile()); } /** @@ -3171,7 +3171,7 @@ private void closeFile(final JournalFile file, final boolean synchronous) { - fileFactory.deactivate(file.getFile()); + fileFactory.deactivateBuffer(); pendingCloseFiles.add(file); Runnable run = new Runnable() Modified: trunk/src/main/org/hornetq/core/journal/impl/NIOSequentialFileFactory.java =================================================================== --- trunk/src/main/org/hornetq/core/journal/impl/NIOSequentialFileFactory.java 2009-11-05 19:01:30 UTC (rev 8229) +++ trunk/src/main/org/hornetq/core/journal/impl/NIOSequentialFileFactory.java 2009-11-06 02:01:22 UTC (rev 8230) @@ -30,14 +30,14 @@ public class NIOSequentialFileFactory extends AbstractSequentialFactory implements SequentialFileFactory { private static final Logger log = Logger.getLogger(NIOSequentialFileFactory.class); - + public NIOSequentialFileFactory(final String journalDir) { super(journalDir); - + if (journalDir == null) { - new Exception ("journalDir is null").printStackTrace(); + new Exception("journalDir is null").printStackTrace(); } } @@ -46,7 +46,7 @@ { return new NIOSequentialFile(journalDir, fileName); } - + public boolean isSupportsCallbacks() { return false; Modified: trunk/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2009-11-05 19:01:30 UTC (rev 8229) +++ trunk/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2009-11-06 02:01:22 UTC (rev 8230) @@ -60,7 +60,7 @@ public class PagingStoreImpl implements TestSupportPageStore { // Constants ----------------------------------------------------- - + private static final Logger log = Logger.getLogger(PagingStoreImpl.class); // Attributes ---------------------------------------------------- @@ -210,7 +210,7 @@ public boolean isPaging() { currentPageLock.readLock().lock(); - + try { if (addressFullMessagePolicy != AddressFullMessagePolicy.PAGE) @@ -271,7 +271,6 @@ checkReleaseProducerFlowControlCredits(-credits); } - public void addSize(final ServerMessage message, final boolean add) throws Exception { long size = message.getMemoryEstimate(); @@ -307,7 +306,7 @@ addSize(-size); } } - + public boolean page(final ServerMessage message, final long transactionID, final boolean duplicateDetection) throws Exception { // The sync on transactions is done on commit only @@ -320,7 +319,7 @@ // of crash return page(message, -1, syncNonTransactional && message.isDurable(), duplicateDetection); } - + public void sync() throws Exception { currentPageLock.readLock().lock(); @@ -482,6 +481,7 @@ currentPageLock.readLock().lock(); try { + // Already paging, nothing to be done if (currentPage != null) { return false; @@ -515,7 +515,6 @@ } } - public Page getCurrentPage() { return currentPage; @@ -597,18 +596,16 @@ { firstPageId = Integer.MAX_VALUE; - if (currentPage != null) + if (currentPage == null) { - returnPage = currentPage; - returnPage.close(); - currentPage = null; - } - else - { // sanity check... it shouldn't happen! throw new IllegalStateException("CurrentPage is null"); } + returnPage = currentPage; + returnPage.close(); + currentPage = null; + // The current page is empty... which means we reached the end of the pages if (returnPage.getNumberOfMessages() == 0) { @@ -679,7 +676,6 @@ } - private synchronized void checkReleaseProducerFlowControlCredits(final long size) { if (addressFullMessagePolicy == AddressFullMessagePolicy.BLOCK && maxSize != -1) @@ -702,7 +698,6 @@ } } - private void addSize(final long size) throws Exception { if (addressFullMessagePolicy != AddressFullMessagePolicy.PAGE) @@ -734,7 +729,6 @@ } else { - // When in Global mode, we use the default page size as the mark to start depage if (maxSize > 0 && currentPage != null && addressSize <= maxSize - pageSize && !depaging.get()) { if (startDepaging()) @@ -750,8 +744,11 @@ return; } } - - private boolean page(final ServerMessage message, final long transactionID, final boolean sync, final boolean duplicateDetection) throws Exception + + private boolean page(final ServerMessage message, + final long transactionID, + final boolean sync, + final boolean duplicateDetection) throws Exception { if (!running) { @@ -813,7 +810,7 @@ { // We set the duplicate detection header to prevent the message being depaged more than once in case of // failure during depage - + byte[] bytes = new byte[8]; ByteBuffer buff = ByteBuffer.wrap(bytes); @@ -823,8 +820,19 @@ message.putBytesProperty(MessageImpl.HDR_DUPLICATE_DETECTION_ID, bytes); } - int bytesToWrite = message.getEncodeSize() + PageImpl.SIZE_RECORD; + PagedMessage pagedMessage; + if (transactionID != -1) + { + pagedMessage = new PagedMessageImpl(message, transactionID); + } + else + { + pagedMessage = new PagedMessageImpl(message); + } + + int bytesToWrite = pagedMessage.getEncodeSize() + PageImpl.SIZE_RECORD; + if (currentPageSize.addAndGet(bytesToWrite) > pageSize && currentPage.getNumberOfMessages() > 0) { // Make sure nothing is currently validating or using currentPage @@ -846,32 +854,14 @@ try { - if (currentPage != null) - { - PagedMessage pagedMessage; - - if (transactionID != -1) - { - pagedMessage = new PagedMessageImpl(message, transactionID); - } - else - { - pagedMessage = new PagedMessageImpl(message); - } - - currentPage.write(pagedMessage); + currentPage.write(pagedMessage); - if (sync) - { - currentPage.sync(); - } - - return true; - } - else + if (sync) { - return false; + currentPage.sync(); } + + return true; } finally { @@ -884,7 +874,7 @@ } } - + /** * This method will remove files from the page system and and route them, doing it transactionally * Modified: trunk/tests/src/org/hornetq/tests/integration/client/PagingTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/PagingTest.java 2009-11-05 19:01:30 UTC (rev 8229) +++ trunk/tests/src/org/hornetq/tests/integration/client/PagingTest.java 2009-11-06 02:01:22 UTC (rev 8230) @@ -83,7 +83,7 @@ clearData(); Configuration config = createDefaultConfig(); - + HornetQServer server = createServer(true, config, PAGE_SIZE, PAGE_MAX, new HashMap<String, AddressSettings>()); server.start(); Modified: trunk/tests/src/org/hornetq/tests/unit/core/journal/impl/fakes/FakeSequentialFileFactory.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/journal/impl/fakes/FakeSequentialFileFactory.java 2009-11-05 19:01:30 UTC (rev 8229) +++ trunk/tests/src/org/hornetq/tests/unit/core/journal/impl/fakes/FakeSequentialFileFactory.java 2009-11-06 02:01:22 UTC (rev 8230) @@ -677,7 +677,7 @@ /* (non-Javadoc) * @see org.hornetq.core.journal.SequentialFileFactory#activate(org.hornetq.core.journal.SequentialFile) */ - public void activate(SequentialFile file) + public void activateBuffer(SequentialFile file) { } @@ -691,7 +691,7 @@ /* (non-Javadoc) * @see org.hornetq.core.journal.SequentialFileFactory#deactivate(org.hornetq.core.journal.SequentialFile) */ - public void deactivate(SequentialFile file) + public void deactivateBuffer() { } Modified: trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java 2009-11-05 19:01:30 UTC (rev 8229) +++ trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java 2009-11-06 02:01:22 UTC (rev 8230) @@ -33,7 +33,6 @@ import org.hornetq.core.journal.SequentialFile; import org.hornetq.core.journal.SequentialFileFactory; import org.hornetq.core.journal.impl.NIOSequentialFileFactory; -import org.hornetq.core.logging.Logger; import org.hornetq.core.paging.Page; import org.hornetq.core.paging.PageTransactionInfo; import org.hornetq.core.paging.PagedMessage; @@ -77,8 +76,6 @@ { // Constants ----------------------------------------------------- - - private static final Logger log = Logger.getLogger(PagingStoreImplTest.class); private final static SimpleString destinationTestName = new SimpleString("test"); @@ -168,15 +165,15 @@ } -// public void testPageWithNIO() throws Exception -// { -// // This integration test could fail 1 in 100 due to race conditions. -// for (int i = 0; i < 100; i++) -// { -// recreateDirectory(getTestDir()); -// testConcurrentPaging(new NIOSequentialFileFactory(getTestDir()), 1); -// } -// } + public void testPageWithNIO() throws Exception + { + // This integration test could fail 1 in 100 due to race conditions. + for (int i = 0; i < 100; i++) + { + recreateDirectory(getTestDir()); + testConcurrentPaging(new NIOSequentialFileFactory(getTestDir()), 1); + } + } public void testStore() throws Exception { @@ -431,13 +428,13 @@ } -// public void testConcurrentDepage() throws Exception -// { -// SequentialFileFactory factory = new FakeSequentialFileFactory(1, false); -// -// testConcurrentPaging(factory, 10); -// } + public void testConcurrentDepage() throws Exception + { + SequentialFileFactory factory = new FakeSequentialFileFactory(1, false); + testConcurrentPaging(factory, 10); + } + protected void testConcurrentPaging(final SequentialFileFactory factory, final int numberOfThreads) throws Exception, InterruptedException { @@ -474,28 +471,35 @@ assertEquals(0, storeImpl.getNumberOfPages()); + // Marked the store to be paged storeImpl.startPaging(); assertEquals(1, storeImpl.getNumberOfPages()); final SimpleString destination = new SimpleString("test"); - class ProducerThread extends Thread + class WriterThread extends Thread { + Exception e; @Override public void run() { + try { boolean firstTime = true; while (true) { long id = messageIdGenerator.incrementAndGet(); - ServerMessage msg = createMessage(storeImpl, destination, createRandomBuffer(id, 5)); - if (storeImpl.page(msg, true)) - { + + // Each thread will Keep paging until all the messages are depaged. + // This is possible because the depage thread is not actually reading the pages. + // Just using the internal API to remove it from the page file system + ServerMessage msg = createMessage(storeImpl, destination, createRandomBuffer(id, 5)); + if (storeImpl.page(msg, false)) + { buffers.put(id, msg); } else @@ -505,6 +509,7 @@ if (firstTime) { + // We have at least one data paged. So, we can start depaging now latchStart.countDown(); firstTime = false; } @@ -522,7 +527,7 @@ } } - class ConsumerThread extends Thread + class ReaderThread extends Thread { Exception e; @@ -533,12 +538,10 @@ { // Wait every producer to produce at least one message latchStart.await(); + while (aliveProducers.get() > 0) { Page page = storeImpl.depage(); - - //log.info("depaged " + page); - if (page != null) { readPages.add(page); @@ -553,15 +556,15 @@ } } - ProducerThread producerThread[] = new ProducerThread[numberOfThreads]; + WriterThread producerThread[] = new WriterThread[numberOfThreads]; for (int i = 0; i < numberOfThreads; i++) { - producerThread[i] = new ProducerThread(); + producerThread[i] = new WriterThread(); producerThread[i].start(); } - ConsumerThread consumer = new ConsumerThread(); + ReaderThread consumer = new ReaderThread(); consumer.start(); for (int i = 0; i < numberOfThreads; i++) @@ -611,7 +614,7 @@ { SequentialFile fileTmp = factory.createSequentialFile(file, 1); fileTmp.open(); - assertTrue(fileTmp.size() + " <= " + MAX_SIZE, fileTmp.size() <= MAX_SIZE); + assertTrue("The page file size (" + fileTmp.size() + ") shouldn't be > " + MAX_SIZE, fileTmp.size() <= MAX_SIZE); fileTmp.close(); }

16 years, 2 months

1
0
0 / 0

JBoss hornetq SVN: r8229 - trunk/src/main/org/hornetq/core/persistence/impl/journal.

by do-not-reply＠jboss.org

Author: clebert.suconic(a)jboss.com Date: 2009-11-05 14:01:30 -0500 (Thu, 05 Nov 2009) New Revision: 8229 Added: trunk/src/main/org/hornetq/core/persistence/impl/journal/FileLargeServerMessage.java Removed: trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalLargeServerMessage.java Modified: trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java Log: tweaks Copied: trunk/src/main/org/hornetq/core/persistence/impl/journal/FileLargeServerMessage.java (from rev 8213, trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalLargeServerMessage.java) =================================================================== --- trunk/src/main/org/hornetq/core/persistence/impl/journal/FileLargeServerMessage.java (rev 0) +++ trunk/src/main/org/hornetq/core/persistence/impl/journal/FileLargeServerMessage.java 2009-11-05 19:01:30 UTC (rev 8229) @@ -0,0 +1,400 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.core.persistence.impl.journal; + +import static org.hornetq.utils.DataConstants.SIZE_INT; + +import java.nio.ByteBuffer; +import java.util.concurrent.atomic.AtomicInteger; + +import org.hornetq.core.journal.SequentialFile; +import org.hornetq.core.logging.Logger; +import org.hornetq.core.paging.PagingStore; +import org.hornetq.core.remoting.spi.HornetQBuffer; +import org.hornetq.core.server.LargeServerMessage; +import org.hornetq.core.server.MessageReference; +import org.hornetq.core.server.ServerMessage; +import org.hornetq.core.message.LargeMessageEncodingContext; +import org.hornetq.core.server.impl.ServerMessageImpl; + +/** + * A JournalLargeServerMessage + * + * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * Created 30-Sep-08 12:02:45 PM + * + * + */ +public class FileLargeServerMessage extends ServerMessageImpl implements LargeServerMessage +{ + // Constants ----------------------------------------------------- + + private static final Logger log = Logger.getLogger(FileLargeServerMessage.class); + + private static boolean isTrace = log.isTraceEnabled(); + + // Attributes ---------------------------------------------------- + + private final JournalStorageManager storageManager; + + private LargeServerMessage linkMessage; + + // We should only use the NIO implementation on the Journal + private SequentialFile file; + + private long bodySize = -1; + + private final AtomicInteger delayDeletionCount = new AtomicInteger(0); + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + public FileLargeServerMessage(final JournalStorageManager storageManager) + { + this.storageManager = storageManager; + } + + /** + * Copy constructor + * @param copy + * @param fileCopy + */ + private FileLargeServerMessage(final FileLargeServerMessage copy, + final SequentialFile fileCopy, + final long newID) + { + super(copy); + this.linkMessage = copy; + storageManager = copy.storageManager; + file = fileCopy; + bodySize = copy.bodySize; + setMessageID(newID); + } + + // Public -------------------------------------------------------- + + /* (non-Javadoc) + * @see org.hornetq.core.server.LargeServerMessage#addBytes(byte[]) + */ + public synchronized void addBytes(final byte[] bytes) throws Exception + { + validateFile(); + + if (!file.isOpen()) + { + file.open(); + } + + storageManager.addBytesToLargeMessage(file, this.getMessageID(), bytes); + + bodySize += bytes.length; + } + + public void encodeBody(final HornetQBuffer bufferOut, LargeMessageEncodingContext context, int size) + { + try + { + // This could maybe be optimized (maybe reading directly into bufferOut) + ByteBuffer bufferRead = ByteBuffer.allocate(size); + + int bytesRead = context.write(bufferRead); + + bufferRead.flip(); + + if (bytesRead > 0) + { + bufferOut.writeBytes(bufferRead.array(), 0, bytesRead); + } + + } + catch (Exception e) + { + throw new RuntimeException(e.getMessage(), e); + } + } + + @Override + public synchronized int getBodySize() + { + try + { + validateFile(); + } + catch (Exception e) + { + throw new RuntimeException(e.getMessage(), e); + } + return (int)Math.min(bodySize, Integer.MAX_VALUE); + } + + @Override + public synchronized long getLargeBodySize() + { + try + { + validateFile(); + } + catch (Exception e) + { + throw new RuntimeException(e.getMessage(), e); + } + return bodySize; + } + + @Override + public synchronized int getEncodeSize() + { + return getHeadersAndPropertiesEncodeSize(); + } + + @Override + public void encode(final HornetQBuffer buffer) + { + encodeHeadersAndProperties(buffer); + } + + @Override + public void decode(final HornetQBuffer buffer) + { + file = null; + decodeHeadersAndProperties(buffer); + } + + public synchronized void incrementDelayDeletionCount() + { + this.delayDeletionCount.incrementAndGet(); + } + + public synchronized void decrementDelayDeletionCount() throws Exception + { + int count = this.delayDeletionCount.decrementAndGet(); + + if (count == 0) + { + checkDelete(); + } + } + + public LargeMessageEncodingContext createNewContext() + { + return new DecodingContext(); + } + + private void checkDelete() throws Exception + { + if (getRefCount() <= 0) + { + if (linkMessage != null) + { + // This file is linked to another message, deleting the reference where it belongs on this case + linkMessage.decrementDelayDeletionCount(); + } + else + { + if (isTrace) + { + log.trace("Deleting file " + file + " as the usage was complete"); + } + + try + { + deleteFile(); + } + catch (Exception e) + { + log.error(e.getMessage(), e); + } + } + } + } + + @Override + public synchronized int decrementRefCount(PagingStore pagingStore, MessageReference reference) throws Exception + { + int currentRefCount = super.decrementRefCount(pagingStore, reference); + + // We use <= as this could be used by load. + // because of a failure, no references were loaded, so we have 0... and we still need to delete the associated + // files + if (delayDeletionCount.get() <= 0) + { + checkDelete(); + } + + return currentRefCount; + } + + @Override + public boolean isLargeMessage() + { + return true; + } + + public synchronized void deleteFile() throws Exception + { + validateFile(); + releaseResources(); + storageManager.deleteFile(file); + } + + public boolean isFileExists() throws Exception + { + SequentialFile localfile = storageManager.createFileForLargeMessage(getMessageID(), durable); + return localfile.exists(); + } + + // We cache this + private volatile int memoryEstimate = -1; + + @Override + public synchronized int getMemoryEstimate() + { + if (memoryEstimate == -1) + { + // The body won't be on memory (aways on-file), so we don't consider this for paging + memoryEstimate = getHeadersAndPropertiesEncodeSize() + SIZE_INT + getEncodeSize() + (16 + 4) * 2 + 1; + } + + return memoryEstimate; + } + + public synchronized void releaseResources() + { + if (file != null && file.isOpen()) + { + try + { + file.close(); + } + catch (Exception e) + { + log.error(e.getMessage(), e); + } + } + } + + @Override + public synchronized ServerMessage copy(final long newID) throws Exception + { + incrementDelayDeletionCount(); + + long idToUse = messageID; + + if (linkMessage != null) + { + idToUse = linkMessage.getMessageID(); + } + + SequentialFile newfile = storageManager.createFileForLargeMessage(idToUse, durable); + + ServerMessage newMessage = new FileLargeServerMessage(linkMessage == null ? this + : (FileLargeServerMessage)linkMessage, + newfile, + newID); + + return newMessage; + } + + public SequentialFile getFile() + { + return file; + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + @Override + protected void finalize() throws Throwable + { + releaseResources(); + super.finalize(); + } + + // Private ------------------------------------------------------- + + private synchronized void validateFile() throws Exception + { + if (file == null) + { + if (messageID <= 0) + { + throw new RuntimeException("MessageID not set on LargeMessage"); + } + + file = storageManager.createFileForLargeMessage(getMessageID(), durable); + + file.open(); + + bodySize = file.size(); + + } + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.LargeServerMessage#setLinkedMessage(org.hornetq.core.server.LargeServerMessage) + */ + public void setLinkedMessage(LargeServerMessage message) + { + if (file != null) + { + // Sanity check.. it shouldn't happen + throw new IllegalStateException("LargeMessage file was already set"); + } + + this.linkMessage = message; + + file = storageManager.createFileForLargeMessage(message.getMessageID(), durable); + try + { + file.open(); + this.bodySize = file.size(); + file.close(); + } + catch (Exception e) + { + throw new RuntimeException("could not setup linked file", e); + } + } + + // Inner classes ------------------------------------------------- + + class DecodingContext implements LargeMessageEncodingContext + { + private SequentialFile cFile; + + public void open() throws Exception + { + cFile = file.copy(); + cFile.open(); + } + + public void close() throws Exception + { + cFile.close(); + } + + public int write(ByteBuffer bufferRead) throws Exception + { + return cFile.read(bufferRead); + } + + public int write(HornetQBuffer bufferOut, int size) + { + return -1; + } + } +} Deleted: trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalLargeServerMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalLargeServerMessage.java 2009-11-05 17:57:42 UTC (rev 8228) +++ trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalLargeServerMessage.java 2009-11-05 19:01:30 UTC (rev 8229) @@ -1,403 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.core.persistence.impl.journal; - -import static org.hornetq.utils.DataConstants.SIZE_INT; - -import java.nio.ByteBuffer; -import java.util.concurrent.atomic.AtomicInteger; - -import org.hornetq.core.journal.SequentialFile; -import org.hornetq.core.logging.Logger; -import org.hornetq.core.paging.PagingStore; -import org.hornetq.core.remoting.spi.HornetQBuffer; -import org.hornetq.core.server.LargeServerMessage; -import org.hornetq.core.server.MessageReference; -import org.hornetq.core.server.ServerMessage; -import org.hornetq.core.message.LargeMessageEncodingContext; -import org.hornetq.core.server.impl.ServerMessageImpl; - -/** - * A JournalLargeServerMessage - * - * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> - * - * Created 30-Sep-08 12:02:45 PM - * - * - */ - - -//FIXME - this class should be renamed to just large message -public class JournalLargeServerMessage extends ServerMessageImpl implements LargeServerMessage -{ - // Constants ----------------------------------------------------- - - private static final Logger log = Logger.getLogger(JournalLargeServerMessage.class); - - private static boolean isTrace = log.isTraceEnabled(); - - // Attributes ---------------------------------------------------- - - private final JournalStorageManager storageManager; - - private LargeServerMessage linkMessage; - - // We should only use the NIO implementation on the Journal - private SequentialFile file; - - private long bodySize = -1; - - private final AtomicInteger delayDeletionCount = new AtomicInteger(0); - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - public JournalLargeServerMessage(final JournalStorageManager storageManager) - { - this.storageManager = storageManager; - } - - /** - * Copy constructor - * @param copy - * @param fileCopy - */ - private JournalLargeServerMessage(final JournalLargeServerMessage copy, - final SequentialFile fileCopy, - final long newID) - { - super(copy); - this.linkMessage = copy; - storageManager = copy.storageManager; - file = fileCopy; - bodySize = copy.bodySize; - setMessageID(newID); - } - - // Public -------------------------------------------------------- - - /* (non-Javadoc) - * @see org.hornetq.core.server.LargeServerMessage#addBytes(byte[]) - */ - public synchronized void addBytes(final byte[] bytes) throws Exception - { - validateFile(); - - if (!file.isOpen()) - { - file.open(); - } - - storageManager.addBytesToLargeMessage(file, this.getMessageID(), bytes); - - bodySize += bytes.length; - } - - public void encodeBody(final HornetQBuffer bufferOut, LargeMessageEncodingContext context, int size) - { - try - { - // This could maybe be optimized (maybe reading directly into bufferOut) - ByteBuffer bufferRead = ByteBuffer.allocate(size); - - int bytesRead = context.write(bufferRead); - - bufferRead.flip(); - - if (bytesRead > 0) - { - bufferOut.writeBytes(bufferRead.array(), 0, bytesRead); - } - - } - catch (Exception e) - { - throw new RuntimeException(e.getMessage(), e); - } - } - - @Override - public synchronized int getBodySize() - { - try - { - validateFile(); - } - catch (Exception e) - { - throw new RuntimeException(e.getMessage(), e); - } - return (int)Math.min(bodySize, Integer.MAX_VALUE); - } - - @Override - public synchronized long getLargeBodySize() - { - try - { - validateFile(); - } - catch (Exception e) - { - throw new RuntimeException(e.getMessage(), e); - } - return bodySize; - } - - @Override - public synchronized int getEncodeSize() - { - return getHeadersAndPropertiesEncodeSize(); - } - - @Override - public void encode(final HornetQBuffer buffer) - { - encodeHeadersAndProperties(buffer); - } - - @Override - public void decode(final HornetQBuffer buffer) - { - file = null; - decodeHeadersAndProperties(buffer); - } - - public synchronized void incrementDelayDeletionCount() - { - this.delayDeletionCount.incrementAndGet(); - } - - public synchronized void decrementDelayDeletionCount() throws Exception - { - int count = this.delayDeletionCount.decrementAndGet(); - - if (count == 0) - { - checkDelete(); - } - } - - public LargeMessageEncodingContext createNewContext() - { - return new DecodingContext(); - } - - private void checkDelete() throws Exception - { - if (getRefCount() <= 0) - { - if (linkMessage != null) - { - // This file is linked to another message, deleting the reference where it belongs on this case - linkMessage.decrementDelayDeletionCount(); - } - else - { - if (isTrace) - { - log.trace("Deleting file " + file + " as the usage was complete"); - } - - try - { - deleteFile(); - } - catch (Exception e) - { - log.error(e.getMessage(), e); - } - } - } - } - - @Override - public synchronized int decrementRefCount(PagingStore pagingStore, MessageReference reference) throws Exception - { - int currentRefCount = super.decrementRefCount(pagingStore, reference); - - // We use <= as this could be used by load. - // because of a failure, no references were loaded, so we have 0... and we still need to delete the associated - // files - if (delayDeletionCount.get() <= 0) - { - checkDelete(); - } - - return currentRefCount; - } - - @Override - public boolean isLargeMessage() - { - return true; - } - - public synchronized void deleteFile() throws Exception - { - validateFile(); - releaseResources(); - storageManager.deleteFile(file); - } - - public boolean isFileExists() throws Exception - { - SequentialFile localfile = storageManager.createFileForLargeMessage(getMessageID(), durable); - return localfile.exists(); - } - - // We cache this - private volatile int memoryEstimate = -1; - - @Override - public synchronized int getMemoryEstimate() - { - if (memoryEstimate == -1) - { - // The body won't be on memory (aways on-file), so we don't consider this for paging - memoryEstimate = getHeadersAndPropertiesEncodeSize() + SIZE_INT + getEncodeSize() + (16 + 4) * 2 + 1; - } - - return memoryEstimate; - } - - public synchronized void releaseResources() - { - if (file != null && file.isOpen()) - { - try - { - file.close(); - } - catch (Exception e) - { - log.error(e.getMessage(), e); - } - } - } - - @Override - public synchronized ServerMessage copy(final long newID) throws Exception - { - incrementDelayDeletionCount(); - - long idToUse = messageID; - - if (linkMessage != null) - { - idToUse = linkMessage.getMessageID(); - } - - SequentialFile newfile = storageManager.createFileForLargeMessage(idToUse, durable); - - ServerMessage newMessage = new JournalLargeServerMessage(linkMessage == null ? this - : (JournalLargeServerMessage)linkMessage, - newfile, - newID); - - return newMessage; - } - - public SequentialFile getFile() - { - return file; - } - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - @Override - protected void finalize() throws Throwable - { - releaseResources(); - super.finalize(); - } - - // Private ------------------------------------------------------- - - private synchronized void validateFile() throws Exception - { - if (file == null) - { - if (messageID <= 0) - { - throw new RuntimeException("MessageID not set on LargeMessage"); - } - - file = storageManager.createFileForLargeMessage(getMessageID(), durable); - - file.open(); - - bodySize = file.size(); - - } - } - - /* (non-Javadoc) - * @see org.hornetq.core.server.LargeServerMessage#setLinkedMessage(org.hornetq.core.server.LargeServerMessage) - */ - public void setLinkedMessage(LargeServerMessage message) - { - if (file != null) - { - // Sanity check.. it shouldn't happen - throw new IllegalStateException("LargeMessage file was already set"); - } - - this.linkMessage = message; - - file = storageManager.createFileForLargeMessage(message.getMessageID(), durable); - try - { - file.open(); - this.bodySize = file.size(); - file.close(); - } - catch (Exception e) - { - throw new RuntimeException("could not setup linked file", e); - } - } - - // Inner classes ------------------------------------------------- - - class DecodingContext implements LargeMessageEncodingContext - { - private SequentialFile cFile; - - public void open() throws Exception - { - cFile = file.copy(); - cFile.open(); - } - - public void close() throws Exception - { - cFile.close(); - } - - public int write(ByteBuffer bufferRead) throws Exception - { - return cFile.read(bufferRead); - } - - public int write(HornetQBuffer bufferOut, int size) - { - return -1; - } - } -} Modified: trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java =================================================================== --- trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2009-11-05 17:57:42 UTC (rev 8228) +++ trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2009-11-05 19:01:30 UTC (rev 8229) @@ -398,7 +398,7 @@ public LargeServerMessage createLargeMessage() { - return new JournalLargeServerMessage(this); + return new FileLargeServerMessage(this); } public void addBytesToLargeMessage(SequentialFile file, long messageId, final byte[] bytes) throws Exception @@ -420,7 +420,7 @@ replicator.largeMessageBegin(id); } - JournalLargeServerMessage largeMessage = (JournalLargeServerMessage)createLargeMessage(); + FileLargeServerMessage largeMessage = (FileLargeServerMessage)createLargeMessage(); HornetQBuffer headerBuffer = ChannelBuffers.wrappedBuffer(header);

16 years, 2 months

1
0
0 / 0

JBoss hornetq SVN: r8228 - trunk/tests/src/org/hornetq/tests/integration/client.

by do-not-reply＠jboss.org

Author: clebert.suconic(a)jboss.com Date: 2009-11-05 12:57:42 -0500 (Thu, 05 Nov 2009) New Revision: 8228 Modified: trunk/tests/src/org/hornetq/tests/integration/client/ConsumerWindowSizeTest.java Log: tweak / typo Modified: trunk/tests/src/org/hornetq/tests/integration/client/ConsumerWindowSizeTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/ConsumerWindowSizeTest.java 2009-11-05 16:57:44 UTC (rev 8227) +++ trunk/tests/src/org/hornetq/tests/integration/client/ConsumerWindowSizeTest.java 2009-11-05 17:57:42 UTC (rev 8228) @@ -51,7 +51,7 @@ protected boolean isNetty() { - return true; + return false; } private int getMessageEncodeSize(final SimpleString address) throws Exception

16 years, 2 months

1
0
0 / 0

JBoss hornetq SVN: r8227 - trunk/src/main/org/hornetq/jms/server/config/impl.

by do-not-reply＠jboss.org

Author: timfox Date: 2009-11-05 11:57:44 -0500 (Thu, 05 Nov 2009) New Revision: 8227 Modified: trunk/src/main/org/hornetq/jms/server/config/impl/ConnectionFactoryConfigurationImpl.java Log: made attributes private on connectionfactoryconfiguration Modified: trunk/src/main/org/hornetq/jms/server/config/impl/ConnectionFactoryConfigurationImpl.java =================================================================== --- trunk/src/main/org/hornetq/jms/server/config/impl/ConnectionFactoryConfigurationImpl.java 2009-11-05 15:38:12 UTC (rev 8226) +++ trunk/src/main/org/hornetq/jms/server/config/impl/ConnectionFactoryConfigurationImpl.java 2009-11-05 16:57:44 UTC (rev 8227) @@ -34,71 +34,71 @@ // Attributes ---------------------------------------------------- - public final String[] bindings; + private final String[] bindings; - public final String name; + private final String name; - public String discoveryAddress; + private String discoveryAddress; - public int discoveryPort; + private int discoveryPort; - public List<Pair<TransportConfiguration, TransportConfiguration>> connectorConfigs; + private List<Pair<TransportConfiguration, TransportConfiguration>> connectorConfigs; - public String clientID = null; + private String clientID = null; - public long discoveryRefreshTimeout = ClientSessionFactoryImpl.DEFAULT_DISCOVERY_REFRESH_TIMEOUT; + private long discoveryRefreshTimeout = ClientSessionFactoryImpl.DEFAULT_DISCOVERY_REFRESH_TIMEOUT; - public long clientFailureCheckPeriod = ClientSessionFactoryImpl.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD; + private long clientFailureCheckPeriod = ClientSessionFactoryImpl.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD; - public long connectionTTL = ClientSessionFactoryImpl.DEFAULT_CONNECTION_TTL; + private long connectionTTL = ClientSessionFactoryImpl.DEFAULT_CONNECTION_TTL; - public long callTimeout = ClientSessionFactoryImpl.DEFAULT_CALL_TIMEOUT; + private long callTimeout = ClientSessionFactoryImpl.DEFAULT_CALL_TIMEOUT; - public boolean cacheLargeMessagesClient = ClientSessionFactoryImpl.DEFAULT_CACHE_LARGE_MESSAGE_CLIENT; + private boolean cacheLargeMessagesClient = ClientSessionFactoryImpl.DEFAULT_CACHE_LARGE_MESSAGE_CLIENT; - public int minLargeMessageSize = ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE; + private int minLargeMessageSize = ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE; - public int consumerWindowSize = ClientSessionFactoryImpl.DEFAULT_CONSUMER_WINDOW_SIZE; + private int consumerWindowSize = ClientSessionFactoryImpl.DEFAULT_CONSUMER_WINDOW_SIZE; - public int consumerMaxRate = ClientSessionFactoryImpl.DEFAULT_CONSUMER_MAX_RATE; + private int consumerMaxRate = ClientSessionFactoryImpl.DEFAULT_CONSUMER_MAX_RATE; - public int confirmationWindowSize = ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE; + private int confirmationWindowSize = ClientSessionFactoryImpl.DEFAULT_CONFIRMATION_WINDOW_SIZE; - public int producerMaxRate = ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE; + private int producerMaxRate = ClientSessionFactoryImpl.DEFAULT_PRODUCER_MAX_RATE; - public boolean blockOnAcknowledge = ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_ACKNOWLEDGE; + private boolean blockOnAcknowledge = ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_ACKNOWLEDGE; - public boolean blockOnPersistentSend = ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_PERSISTENT_SEND; + private boolean blockOnPersistentSend = ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_PERSISTENT_SEND; - public boolean blockOnNonPersistentSend = ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_NON_PERSISTENT_SEND; + private boolean blockOnNonPersistentSend = ClientSessionFactoryImpl.DEFAULT_BLOCK_ON_NON_PERSISTENT_SEND; - public boolean autoGroup = ClientSessionFactoryImpl.DEFAULT_AUTO_GROUP; + private boolean autoGroup = ClientSessionFactoryImpl.DEFAULT_AUTO_GROUP; - public boolean preAcknowledge = ClientSessionFactoryImpl.DEFAULT_PRE_ACKNOWLEDGE; + private boolean preAcknowledge = ClientSessionFactoryImpl.DEFAULT_PRE_ACKNOWLEDGE; - public String loadBalancingPolicyClassName = ClientSessionFactoryImpl.DEFAULT_CONNECTION_LOAD_BALANCING_POLICY_CLASS_NAME; + private String loadBalancingPolicyClassName = ClientSessionFactoryImpl.DEFAULT_CONNECTION_LOAD_BALANCING_POLICY_CLASS_NAME; - public int transactionBatchSize = ClientSessionFactoryImpl.DEFAULT_ACK_BATCH_SIZE; + private int transactionBatchSize = ClientSessionFactoryImpl.DEFAULT_ACK_BATCH_SIZE; - public int dupsOKBatchSize = ClientSessionFactoryImpl.DEFAULT_ACK_BATCH_SIZE; + private int dupsOKBatchSize = ClientSessionFactoryImpl.DEFAULT_ACK_BATCH_SIZE; - public long initialWaitTimeout = ClientSessionFactoryImpl.DEFAULT_DISCOVERY_INITIAL_WAIT_TIMEOUT; + private long initialWaitTimeout = ClientSessionFactoryImpl.DEFAULT_DISCOVERY_INITIAL_WAIT_TIMEOUT; - public boolean useGlobalPools = ClientSessionFactoryImpl.DEFAULT_USE_GLOBAL_POOLS; + private boolean useGlobalPools = ClientSessionFactoryImpl.DEFAULT_USE_GLOBAL_POOLS; - public int scheduledThreadPoolMaxSize = ClientSessionFactoryImpl.DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE; + private int scheduledThreadPoolMaxSize = ClientSessionFactoryImpl.DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE; - public int threadPoolMaxSize = ClientSessionFactoryImpl.DEFAULT_THREAD_POOL_MAX_SIZE; + private int threadPoolMaxSize = ClientSessionFactoryImpl.DEFAULT_THREAD_POOL_MAX_SIZE; - public long retryInterval = ClientSessionFactoryImpl.DEFAULT_RETRY_INTERVAL; + private long retryInterval = ClientSessionFactoryImpl.DEFAULT_RETRY_INTERVAL; - public double retryIntervalMultiplier = ClientSessionFactoryImpl.DEFAULT_RETRY_INTERVAL_MULTIPLIER; + private double retryIntervalMultiplier = ClientSessionFactoryImpl.DEFAULT_RETRY_INTERVAL_MULTIPLIER; - public long maxRetryInterval = ClientSessionFactoryImpl.DEFAULT_MAX_RETRY_INTERVAL; + private long maxRetryInterval = ClientSessionFactoryImpl.DEFAULT_MAX_RETRY_INTERVAL; - public int reconnectAttempts = ClientSessionFactoryImpl.DEFAULT_RECONNECT_ATTEMPTS; + private int reconnectAttempts = ClientSessionFactoryImpl.DEFAULT_RECONNECT_ATTEMPTS; - public boolean failoverOnServerShutdown = ClientSessionFactoryImpl.DEFAULT_FAILOVER_ON_SERVER_SHUTDOWN; + private boolean failoverOnServerShutdown = ClientSessionFactoryImpl.DEFAULT_FAILOVER_ON_SERVER_SHUTDOWN; // Static --------------------------------------------------------

16 years, 2 months

1
0
0 / 0

JBoss hornetq SVN: r8226 - in trunk: tests/src/org/hornetq/tests/unit/core/paging/impl and 1 other directory.

by do-not-reply＠jboss.org

Author: timfox Date: 2009-11-05 10:38:12 -0500 (Thu, 05 Nov 2009) New Revision: 8226 Modified: trunk/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java Log: temporary commenting out strange tests Modified: trunk/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2009-11-05 15:28:20 UTC (rev 8225) +++ trunk/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2009-11-05 15:38:12 UTC (rev 8226) @@ -60,6 +60,7 @@ public class PagingStoreImpl implements TestSupportPageStore { // Constants ----------------------------------------------------- + private static final Logger log = Logger.getLogger(PagingStoreImpl.class); // Attributes ---------------------------------------------------- @@ -864,6 +865,7 @@ { currentPage.sync(); } + return true; } else Modified: trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java 2009-11-05 15:28:20 UTC (rev 8225) +++ trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java 2009-11-05 15:38:12 UTC (rev 8226) @@ -33,6 +33,7 @@ import org.hornetq.core.journal.SequentialFile; import org.hornetq.core.journal.SequentialFileFactory; import org.hornetq.core.journal.impl.NIOSequentialFileFactory; +import org.hornetq.core.logging.Logger; import org.hornetq.core.paging.Page; import org.hornetq.core.paging.PageTransactionInfo; import org.hornetq.core.paging.PagedMessage; @@ -76,6 +77,8 @@ { // Constants ----------------------------------------------------- + + private static final Logger log = Logger.getLogger(PagingStoreImplTest.class); private final static SimpleString destinationTestName = new SimpleString("test"); @@ -165,15 +168,15 @@ } - public void testPageWithNIO() throws Exception - { - // This integration test could fail 1 in 100 due to race conditions. - for (int i = 0; i < 100; i++) - { - recreateDirectory(getTestDir()); - testConcurrentPaging(new NIOSequentialFileFactory(getTestDir()), 1); - } - } +// public void testPageWithNIO() throws Exception +// { +// // This integration test could fail 1 in 100 due to race conditions. +// for (int i = 0; i < 100; i++) +// { +// recreateDirectory(getTestDir()); +// testConcurrentPaging(new NIOSequentialFileFactory(getTestDir()), 1); +// } +// } public void testStore() throws Exception { @@ -428,13 +431,13 @@ } - public void testConcurrentDepage() throws Exception - { - SequentialFileFactory factory = new FakeSequentialFileFactory(1, false); +// public void testConcurrentDepage() throws Exception +// { +// SequentialFileFactory factory = new FakeSequentialFileFactory(1, false); +// +// testConcurrentPaging(factory, 10); +// } - testConcurrentPaging(factory, 10); - } - protected void testConcurrentPaging(final SequentialFileFactory factory, final int numberOfThreads) throws Exception, InterruptedException { @@ -479,13 +482,11 @@ class ProducerThread extends Thread { - Exception e; @Override public void run() { - try { boolean firstTime = true; @@ -494,7 +495,7 @@ long id = messageIdGenerator.incrementAndGet(); ServerMessage msg = createMessage(storeImpl, destination, createRandomBuffer(id, 5)); if (storeImpl.page(msg, true)) - { + { buffers.put(id, msg); } else @@ -535,6 +536,9 @@ while (aliveProducers.get() > 0) { Page page = storeImpl.depage(); + + //log.info("depaged " + page); + if (page != null) { readPages.add(page);

16 years, 2 months

1
0
0 / 0

JBoss hornetq SVN: r8225 - trunk/src/main/org/hornetq/integration/transports/netty.

by do-not-reply＠jboss.org

Author: jmesnil Date: 2009-11-05 10:28:20 -0500 (Thu, 05 Nov 2009) New Revision: 8225 Removed: trunk/src/main/org/hornetq/integration/transports/netty/VirtualExecutorService.java Modified: trunk/src/main/org/hornetq/integration/transports/netty/NettyAcceptor.java trunk/src/main/org/hornetq/integration/transports/netty/NettyConnector.java Log: removed duplicated class VirtualExecutorService and use Netty's original instead this class was added when HornetQ neede the functionality and a Netty version with it had not been released. Switch back to Netty class before they diverge Modified: trunk/src/main/org/hornetq/integration/transports/netty/NettyAcceptor.java =================================================================== --- trunk/src/main/org/hornetq/integration/transports/netty/NettyAcceptor.java 2009-11-05 14:38:23 UTC (rev 8224) +++ trunk/src/main/org/hornetq/integration/transports/netty/NettyAcceptor.java 2009-11-05 15:28:20 UTC (rev 8225) @@ -64,6 +64,7 @@ import org.jboss.netty.handler.codec.http.HttpResponseEncoder; import org.jboss.netty.handler.ssl.SslHandler; import org.jboss.netty.util.Version; +import org.jboss.netty.util.VirtualExecutorService; /** * A Netty TCP Acceptor that supports SSL Modified: trunk/src/main/org/hornetq/integration/transports/netty/NettyConnector.java =================================================================== --- trunk/src/main/org/hornetq/integration/transports/netty/NettyConnector.java 2009-11-05 14:38:23 UTC (rev 8224) +++ trunk/src/main/org/hornetq/integration/transports/netty/NettyConnector.java 2009-11-05 15:28:20 UTC (rev 8225) @@ -72,6 +72,7 @@ import org.jboss.netty.handler.codec.http.HttpVersion; import org.jboss.netty.handler.ssl.SslHandler; import org.jboss.netty.util.Version; +import org.jboss.netty.util.VirtualExecutorService; /** * A NettyConnector Deleted: trunk/src/main/org/hornetq/integration/transports/netty/VirtualExecutorService.java =================================================================== --- trunk/src/main/org/hornetq/integration/transports/netty/VirtualExecutorService.java 2009-11-05 14:38:23 UTC (rev 8224) +++ trunk/src/main/org/hornetq/integration/transports/netty/VirtualExecutorService.java 2009-11-05 15:28:20 UTC (rev 8225) @@ -1,226 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ -package org.hornetq.integration.transports.netty; - -import java.util.Collections; -import java.util.IdentityHashMap; -import java.util.List; -import java.util.Set; -import java.util.concurrent.AbstractExecutorService; -import java.util.concurrent.Executor; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.RejectedExecutionException; -import java.util.concurrent.TimeUnit; - -import org.hornetq.core.logging.Logger; -import org.jboss.netty.util.internal.MapBackedSet; - -/** - * A delegating {@link ExecutorService} with its own termination management. - * <p> - * {@link VirtualExecutorService} is used when you want to inject an - * {@link ExecutorService} but you do not want to allow the explicit termination - * of threads on shutdown request. It is particularly useful when the - * {@link ExecutorService} to inject is shared by different components and - * the life cycle of the components depend on the termination of the injected - * {@link ExecutorService}. - * - * <pre> - * ExecutorService globalExecutor = ...; - * ExecutorService virtualExecutor = new VirtualExecutorService(globalExecutor); - * - * ChannelFactory factory = - * new NioServerSocketChannelFactory(virtualExecutor, virtualExecutor); - * ... - * - * // ChannelFactory.releaseExternalResources() shuts down the executor and - * // interrupts the I/O threads to terminate all I/O tasks and to release all - * // resources acquired by ChannelFactory. - * factory.releaseExternalResources(); - * - * // Note that globalExecutor is not shut down because VirtualExecutorService - * // implements its own termination management. All threads which were acquired - * // by ChannelFactory via VirtualExecutorService are returned to the pool. - * assert !globalExecutor.isShutdown(); - * </pre> - * - * <h3>The differences from an ordinary {@link ExecutorService}</h3> - * - * A shutdown request ({@link #shutdown()} or {@link #shutdownNow()}) does not - * shut down its parent {@link Executor} but simply sets its internal flag to - * reject further execution request. - * <p> - * {@link #shutdownNow()} interrupts only the thread which is executing the - * task executed via {@link VirtualExecutorService}. - * <p> - * {@link #awaitTermination(long, TimeUnit)} does not wait for real thread - * termination but wait until {@link VirtualExecutorService} is shut down and - * its active tasks are finished and the threads are returned to the parent - * {@link Executor}. - * - * @author The Netty Project (netty-dev(a)lists.jboss.org) - * @author Trustin Lee (tlee(a)redhat.com) - * @version $Rev: 1315 $, $Date: 2009-06-03 03:48:23 -0400 (Wed, 03 Jun 2009) $ - */ -public class VirtualExecutorService extends AbstractExecutorService -{ - private static final Logger log = Logger.getLogger(VirtualExecutorService.class); - - private final Executor e; - - private final ExecutorService s; - - final Object startStopLock = new Object(); - - volatile boolean shutdown; - - Set<Thread> activeThreads = new MapBackedSet<Thread>(new IdentityHashMap<Thread, Boolean>()); - - /** - * Creates a new instance with the specified parent {@link Executor}. - */ - public VirtualExecutorService(Executor parent) - { - if (parent == null) - { - throw new NullPointerException("parent"); - } - - if (parent instanceof ExecutorService) - { - e = null; - s = (ExecutorService)parent; - } - else - { - e = parent; - s = null; - } - } - - public boolean isShutdown() - { - synchronized (startStopLock) - { - return shutdown; - } - } - - public boolean isTerminated() - { - synchronized (startStopLock) - { - return shutdown && activeThreads.isEmpty(); - } - } - - public void shutdown() - { - synchronized (startStopLock) - { - if (shutdown) - { - return; - } - shutdown = true; - } - } - - public List<Runnable> shutdownNow() - { - synchronized (startStopLock) - { - if (!isTerminated()) - { - shutdown(); - for (Thread t : activeThreads) - { - t.interrupt(); - } - } - } - - return Collections.emptyList(); - } - - public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException - { - synchronized (startStopLock) - { - while (!isTerminated()) - { - startStopLock.wait(TimeUnit.MILLISECONDS.convert(timeout, unit)); - } - - return isTerminated(); - } - } - - public void execute(Runnable command) - { - if (command == null) - { - throw new NullPointerException("command"); - } - - if (shutdown) - { - throw new RejectedExecutionException(); - } - - if (s != null) - { - s.execute(new ChildExecutorRunnable(command)); - } - else - { - e.execute(new ChildExecutorRunnable(command)); - } - } - - private class ChildExecutorRunnable implements Runnable - { - - private final Runnable runnable; - - ChildExecutorRunnable(Runnable runnable) - { - this.runnable = runnable; - } - - public void run() - { - Thread thread = Thread.currentThread(); - synchronized (startStopLock) - { - activeThreads.add(thread); - } - try - { - runnable.run(); - } - finally - { - synchronized (startStopLock) - { - boolean removed = activeThreads.remove(thread); - assert removed; - if (isTerminated()) - { - startStopLock.notifyAll(); - } - } - } - } - } -} \ No newline at end of file

16 years, 2 months

1
0
0 / 0

JBoss hornetq SVN: r8224 - in trunk: examples/core/embedded-remote/src/org/hornetq/core/example and 11 other directories.

by do-not-reply＠jboss.org

Author: jmesnil Date: 2009-11-05 09:38:23 -0500 (Thu, 05 Nov 2009) New Revision: 8224 Added: trunk/src/main/org/hornetq/core/message/PropertyConversionException.java trunk/tests/src/org/hornetq/tests/unit/core/message/impl/MessagePropertyConversionTest.java Modified: trunk/examples/core/embedded-remote/src/org/hornetq/core/example/EmbeddedRemoteExample.java trunk/examples/core/embedded/src/org/hornetq/core/example/EmbeddedExample.java trunk/examples/core/microcontainer/src/org/hornetq/core/example/EmbeddedMicroContainerExample.java trunk/src/main/org/hornetq/core/message/Message.java trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java trunk/src/main/org/hornetq/jms/client/HornetQMessage.java trunk/tests/src/org/hornetq/tests/integration/InterceptorTest.java trunk/tests/src/org/hornetq/tests/integration/client/ConsumerTest.java trunk/tests/src/org/hornetq/tests/integration/cluster/failover/AsynchronousFailoverTest.java trunk/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java trunk/tests/src/org/hornetq/tests/util/ServiceTestBase.java Log: https://jira.jboss.org/jira/browse/HORNETQ-176: typed property getters on the core Message class * moved type conversion for properties from HornetQMessage to MessageImpl Core message props conversion has the same semantics than JMS props Modified: trunk/examples/core/embedded/src/org/hornetq/core/example/EmbeddedExample.java =================================================================== --- trunk/examples/core/embedded/src/org/hornetq/core/example/EmbeddedExample.java 2009-11-05 14:23:20 UTC (rev 8223) +++ trunk/examples/core/embedded/src/org/hornetq/core/example/EmbeddedExample.java 2009-11-05 14:38:23 UTC (rev 8224) @@ -94,7 +94,7 @@ // Step 8. Receive the message. ClientMessage messageReceived = messageConsumer.receive(1000); - System.out.println("Received TextMessage:" + messageReceived.getProperty(propName)); + System.out.println("Received TextMessage:" + messageReceived.getObjectProperty(propName)); } finally { Modified: trunk/examples/core/embedded-remote/src/org/hornetq/core/example/EmbeddedRemoteExample.java =================================================================== --- trunk/examples/core/embedded-remote/src/org/hornetq/core/example/EmbeddedRemoteExample.java 2009-11-05 14:23:20 UTC (rev 8223) +++ trunk/examples/core/embedded-remote/src/org/hornetq/core/example/EmbeddedRemoteExample.java 2009-11-05 14:38:23 UTC (rev 8224) @@ -86,7 +86,7 @@ // Step 9. Receive the message. ClientMessage messageReceived = messageConsumer.receive(1000); - System.out.println("Received TextMessage:" + messageReceived.getProperty(propName)); + System.out.println("Received TextMessage:" + messageReceived.getObjectProperty(propName)); } finally { Modified: trunk/examples/core/microcontainer/src/org/hornetq/core/example/EmbeddedMicroContainerExample.java =================================================================== --- trunk/examples/core/microcontainer/src/org/hornetq/core/example/EmbeddedMicroContainerExample.java 2009-11-05 14:23:20 UTC (rev 8223) +++ trunk/examples/core/microcontainer/src/org/hornetq/core/example/EmbeddedMicroContainerExample.java 2009-11-05 14:38:23 UTC (rev 8224) @@ -87,7 +87,7 @@ // Step 7. Receive the message. ClientMessage messageReceived = messageConsumer.receive(1000); - System.out.println("Received TextMessage:" + messageReceived.getProperty(propName)); + System.out.println("Received TextMessage:" + messageReceived.getObjectProperty(propName)); } finally { Modified: trunk/src/main/org/hornetq/core/message/Message.java =================================================================== --- trunk/src/main/org/hornetq/core/message/Message.java 2009-11-05 14:23:20 UTC (rev 8223) +++ trunk/src/main/org/hornetq/core/message/Message.java 2009-11-05 14:38:23 UTC (rev 8224) @@ -18,7 +18,6 @@ import java.util.Set; import org.hornetq.core.remoting.spi.HornetQBuffer; -import org.hornetq.core.message.LargeMessageEncodingContext; import org.hornetq.utils.SimpleString; import org.hornetq.utils.TypedProperties; @@ -35,132 +34,152 @@ * $Id: Message.java 3341 2007-11-19 14:34:57Z timfox $ */ public interface Message -{ +{ long getMessageID(); - + SimpleString getDestination(); - + void setDestination(SimpleString destination); - - byte getType(); - + + byte getType(); + boolean isDurable(); - + void setDurable(boolean durable); - + long getExpiration(); boolean isExpired(); - + void setExpiration(long expiration); - + long getTimestamp(); - + void setTimestamp(long timestamp); - + byte getPriority(); - + void setPriority(byte priority); - + int getEncodeSize(); void encode(HornetQBuffer buffer); - + void decode(HornetQBuffer buffer); - - + int getHeadersAndPropertiesEncodeSize(); - + void encodeHeadersAndProperties(HornetQBuffer buffer); - + void decodeHeadersAndProperties(HornetQBuffer buffer); - + int getBodySize(); boolean isLargeMessage(); - + long getLargeBodySize(); - + // Used on Message chunk void encodeBody(final HornetQBuffer bufferOut, LargeMessageEncodingContext context, int size); - + /** Set the InputStream used on a message that will be sent over a producer */ void setBodyInputStream(InputStream stream); - + /** Get the InputStream used on a message that will be sent over a producer */ InputStream getBodyInputStream(); - - + void encodeBody(HornetQBuffer buffer); - + void decodeBody(HornetQBuffer buffer); - + // Properties // ------------------------------------------------------------------ - + TypedProperties getProperties(); - + void putBooleanProperty(SimpleString key, boolean value); - + void putByteProperty(SimpleString key, byte value); - + void putBytesProperty(SimpleString key, byte[] value); - + void putShortProperty(SimpleString key, short value); - + void putIntProperty(SimpleString key, int value); - + void putLongProperty(SimpleString key, long value); - + void putFloatProperty(SimpleString key, float value); - + void putDoubleProperty(SimpleString key, double value); - + void putStringProperty(SimpleString key, SimpleString value); - + + void putObjectProperty(SimpleString key, Object value) throws PropertyConversionException; + void putBooleanProperty(String key, boolean value); - + void putByteProperty(String key, byte value); - + void putBytesProperty(String key, byte[] value); - + void putShortProperty(String key, short value); - + void putIntProperty(String key, int value); - + void putLongProperty(String key, long value); - + void putFloatProperty(String key, float value); - + void putDoubleProperty(String key, double value); - + void putStringProperty(String key, String value); - + + void putObjectProperty(String key, Object value) throws PropertyConversionException; + void putTypedProperties(TypedProperties properties); - // TODO - should have typed property getters and do conversions herein - Object getProperty(SimpleString key); - + Object removeProperty(SimpleString key); - + boolean containsProperty(SimpleString key); - - Object getProperty(String key); - + + Object getObjectProperty(String key); + + Boolean getBooleanProperty(SimpleString key) throws PropertyConversionException; + + Byte getByteProperty(SimpleString key) throws PropertyConversionException; + + Double getDoubleProperty(SimpleString key) throws PropertyConversionException; + + Integer getIntProperty(SimpleString key) throws PropertyConversionException; + + Long getLongProperty(SimpleString key) throws PropertyConversionException; + + Object getObjectProperty(SimpleString key); + + Short getShortProperty(SimpleString key) throws PropertyConversionException; + + Float getFloatProperty(SimpleString key) throws PropertyConversionException; + + String getStringProperty(SimpleString key) throws PropertyConversionException; + + byte[] getBytesProperty(SimpleString key) throws PropertyConversionException; + Object removeProperty(String key); - + boolean containsProperty(String key); - + Set<SimpleString> getPropertyNames(); - + Map<String, Object> toMap(); - + // Body // --------------------------------------------------------------------------------- - + HornetQBuffer getBody(); - + void setBody(HornetQBuffer body); - + } Added: trunk/src/main/org/hornetq/core/message/PropertyConversionException.java =================================================================== --- trunk/src/main/org/hornetq/core/message/PropertyConversionException.java (rev 0) +++ trunk/src/main/org/hornetq/core/message/PropertyConversionException.java 2009-11-05 14:38:23 UTC (rev 8224) @@ -0,0 +1,51 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.core.message; + +/** + * A PropertyConversionException + * + * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a> + * + * + */ +public class PropertyConversionException extends Exception +{ + + private static final long serialVersionUID = -3010008708334904332L; + + public PropertyConversionException(String message) + { + super(message); + } + + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- + +} Modified: trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java 2009-11-05 14:23:20 UTC (rev 8223) +++ trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java 2009-11-05 14:38:23 UTC (rev 8224) @@ -25,9 +25,10 @@ import org.hornetq.core.buffers.ChannelBuffers; import org.hornetq.core.logging.Logger; +import org.hornetq.core.message.LargeMessageEncodingContext; import org.hornetq.core.message.Message; +import org.hornetq.core.message.PropertyConversionException; import org.hornetq.core.remoting.spi.HornetQBuffer; -import org.hornetq.core.message.LargeMessageEncodingContext; import org.hornetq.utils.SimpleString; import org.hornetq.utils.TypedProperties; @@ -60,15 +61,15 @@ public static final SimpleString HDR_GROUP_ID = new SimpleString("_HQ_GROUP_ID"); public static final SimpleString HDR_SCHEDULED_DELIVERY_TIME = new SimpleString("_HQ_SCHED_DELIVERY"); - + public static final SimpleString HDR_DUPLICATE_DETECTION_ID = new SimpleString("_HQ_DUPL_ID"); public static final SimpleString HDR_ROUTE_TO_IDS = new SimpleString("_HQ_ROUTE_TO"); - + public static final SimpleString HDR_FROM_CLUSTER = new SimpleString("_HQ_FROM_CLUSTER"); public static final SimpleString HDR_LAST_VALUE_NAME = new SimpleString("_HQ_LVQ_NAME"); - + // Attributes ---------------------------------------------------- protected long messageID; @@ -92,7 +93,7 @@ /** Used on LargeMessages */ private InputStream bodyInputStream; - + // Constructors -------------------------------------------------- protected MessageImpl() @@ -141,7 +142,7 @@ this.properties = new TypedProperties(other.properties); this.body = other.body; } - + /* * Copy constructor */ @@ -158,7 +159,6 @@ this.properties = new TypedProperties(other.getProperties()); this.body = other.getBody(); } - protected MessageImpl(final long messageID) { @@ -174,7 +174,7 @@ buffer.writeInt(getBodySize()); encodeBody(buffer); } - + public int getEncodeSize() { return getHeadersAndPropertiesEncodeSize() + SIZE_INT + getBodySize(); @@ -246,7 +246,7 @@ buffer.readBytes(bytes); // Reuse the same body on the initial body created - body = ChannelBuffers.dynamicBuffer(bytes); + body = ChannelBuffers.dynamicBuffer(bytes); } public long getMessageID() @@ -318,7 +318,7 @@ return System.currentTimeMillis() - expiration >= 0; } - + /** * @return the bodyInputStream */ @@ -338,7 +338,7 @@ public Map<String, Object> toMap() { Map<String, Object> map = new HashMap<String, Object>(); - + map.put("messageID", messageID); map.put("destination", destination.toString()); map.put("type", type); @@ -346,14 +346,13 @@ map.put("expiration", expiration); map.put("timestamp", timestamp); map.put("priority", priority); - for (SimpleString propName: properties.getPropertyNames()) + for (SimpleString propName : properties.getPropertyNames()) { map.put(propName.toString(), properties.getProperty(propName)); } - return map; + return map; } - - + // Properties // --------------------------------------------------------------------------------------- @@ -402,6 +401,57 @@ properties.putStringProperty(key, value); } + public void putObjectProperty(SimpleString key, Object value) throws PropertyConversionException + { + if (value == null) + { + // This is ok - when we try to read the same key it will return null too + return; + } + + if (value instanceof Boolean) + { + properties.putBooleanProperty(key, (Boolean)value); + } + else if (value instanceof Byte) + { + properties.putByteProperty(key, (Byte)value); + } + else if (value instanceof Short) + { + properties.putShortProperty(key, (Short)value); + } + else if (value instanceof Integer) + { + properties.putIntProperty(key, (Integer)value); + } + else if (value instanceof Long) + { + properties.putLongProperty(key, (Long)value); + } + else if (value instanceof Float) + { + properties.putFloatProperty(key, (Float)value); + } + else if (value instanceof Double) + { + properties.putDoubleProperty(key, (Double)value); + } + else if (value instanceof String) + { + properties.putStringProperty(key, new SimpleString((String)value)); + } + else + { + throw new PropertyConversionException(value.getClass() + " is not a valid property type"); + } + } + + public void putObjectProperty(String key, Object value) throws PropertyConversionException + { + putObjectProperty(new SimpleString(key), value); + } + public void putBooleanProperty(final String key, final boolean value) { properties.putBooleanProperty(new SimpleString(key), value); @@ -456,17 +506,269 @@ { return properties.getProperty(key); } + + public Boolean getBooleanProperty(SimpleString key) throws PropertyConversionException + { + final Object value = getObjectProperty(key); + if (value == null) + { + return Boolean.valueOf(null); + } + + if (value instanceof Boolean) + { + return (Boolean)value; + } + else if (value instanceof SimpleString) + { + return Boolean.valueOf(((SimpleString)value).toString()).booleanValue(); + } + else + { + throw new PropertyConversionException("Invalid conversion"); + } + } + + public Byte getByteProperty(SimpleString key) throws PropertyConversionException + { + final Object value = properties.getProperty(key); + + if (value == null) + { + return Byte.valueOf(null); + } + + if (value instanceof Byte) + { + return (Byte)value; + } + else if (value instanceof SimpleString) + { + return Byte.parseByte(((SimpleString)value).toString()); + } + else + { + throw new PropertyConversionException("Invalid conversion"); + } + } - public Object getProperty(final String key) + public byte[] getBytesProperty(SimpleString key) throws PropertyConversionException { + final Object value = properties.getProperty(key); + + if (value == null) + { + throw new IllegalArgumentException(key + " property is not defined"); + } + + if (value instanceof byte[]) + { + return (byte[])value; + } + else + { + throw new PropertyConversionException("Invalid conversion"); + } + } + + public Double getDoubleProperty(SimpleString key) throws PropertyConversionException + { + Object value = getProperty(key); + if (value == null) + { + return Double.valueOf(null); + } + + if (value instanceof Float) + { + return ((Float)value).doubleValue(); + } + else if (value instanceof Double) + { + return (Double)value; + } + else if (value instanceof SimpleString) + { + return Double.parseDouble(((SimpleString)value).toString()); + } + else + { + throw new PropertyConversionException("Invalid conversion"); + } + } + + public Integer getIntProperty(SimpleString key) throws PropertyConversionException + { + Object value = getObjectProperty(key); + if (value == null) + { + return Integer.valueOf(null); + } + else if (value instanceof Integer) + { + return (Integer)value; + } + else if (value instanceof Byte) + { + return ((Byte)value).intValue(); + } + else if (value instanceof Short) + { + return ((Short)value).intValue(); + } + else if (value instanceof SimpleString) + { + return Integer.parseInt(((SimpleString)value).toString()); + } + else + { + throw new PropertyConversionException("Invalid conversion"); + } + } + + public Long getLongProperty(SimpleString key) throws PropertyConversionException + { + Object value = getObjectProperty(key); + if (value == null) + { + return Long.valueOf(null); + } + else if (value instanceof Long) + { + return (Long)value; + } + else if (value instanceof Byte) + { + return ((Byte)value).longValue(); + } + else if (value instanceof Short) + { + return ((Short)value).longValue(); + } + else if (value instanceof Integer) + { + return ((Integer)value).longValue(); + } + else if (value instanceof SimpleString) + { + return Long.parseLong(((SimpleString)value).toString()); + } + else + { + throw new PropertyConversionException("Invalid conversion"); + } + } + + public Short getShortProperty(SimpleString key) throws PropertyConversionException + { + Object value = getObjectProperty(key); + if (value == null) + { + return Short.valueOf(null); + } + else if (value instanceof Byte) + { + return ((Byte)value).shortValue(); + } + else if (value instanceof Short) + { + return (Short)value; + } + else if (value instanceof SimpleString) + { + return Short.parseShort(((SimpleString)value).toString()); + } + else + { + throw new PropertyConversionException("Invalid Conversion."); + } + + } + + public Float getFloatProperty(SimpleString key) throws PropertyConversionException + { + Object value = getProperty(key); + if (value == null) + { + return Float.valueOf(null); + } + + if (value instanceof Float) + { + return ((Float)value).floatValue(); + } + else if (value instanceof SimpleString) + { + return Float.parseFloat(((SimpleString)value).toString()); + } + else + { + throw new PropertyConversionException("Invalid conversion"); + } + } + + public String getStringProperty(SimpleString key) throws PropertyConversionException + { + Object value = properties.getProperty(key); + + if (value == null) + { + return null; + } + + if (value instanceof SimpleString) + { + return ((SimpleString)value).toString(); + } + else if (value instanceof Boolean) + { + return value.toString(); + } + else if (value instanceof Byte) + { + return value.toString(); + } + else if (value instanceof Short) + { + return value.toString(); + } + else if (value instanceof Integer) + { + return value.toString(); + } + else if (value instanceof Long) + { + return value.toString(); + } + else if (value instanceof Float) + { + return value.toString(); + } + else if (value instanceof Double) + { + return value.toString(); + } + else + { + throw new PropertyConversionException("Invalid conversion"); + } + } + + public Object getObjectProperty(final String key) + { return properties.getProperty(new SimpleString(key)); } + public Object getObjectProperty(final SimpleString key) + { + return properties.getProperty(key); + } + public Object removeProperty(final SimpleString key) { return properties.removeProperty(key); } - + public Object removeProperty(final String key) { return properties.removeProperty(new SimpleString(key)); @@ -476,7 +778,7 @@ { return properties.containsProperty(key); } - + public boolean containsProperty(final String key) { return properties.containsProperty(new SimpleString(key)); @@ -491,7 +793,7 @@ { return this.properties; } - + // Body // ------------------------------------------------------------------------------------- Modified: trunk/src/main/org/hornetq/jms/client/HornetQMessage.java =================================================================== --- trunk/src/main/org/hornetq/jms/client/HornetQMessage.java 2009-11-05 14:23:20 UTC (rev 8223) +++ trunk/src/main/org/hornetq/jms/client/HornetQMessage.java 2009-11-05 14:38:23 UTC (rev 8224) @@ -39,6 +39,7 @@ import org.hornetq.core.client.impl.ClientMessageImpl; import org.hornetq.core.exception.HornetQException; import org.hornetq.core.logging.Logger; +import org.hornetq.core.message.PropertyConversionException; import org.hornetq.core.message.impl.MessageImpl; import org.hornetq.core.remoting.spi.HornetQBuffer; import org.hornetq.jms.HornetQDestination; @@ -604,72 +605,38 @@ public boolean getBooleanProperty(final String name) throws JMSException { - Object value = message.getProperty(new SimpleString(name)); - if (value == null) + try { - return Boolean.valueOf(null).booleanValue(); + return message.getBooleanProperty(new SimpleString(name)); } - - if (value instanceof Boolean) + catch (PropertyConversionException e) { - return ((Boolean)value).booleanValue(); + throw new MessageFormatException(e.getMessage()); } - else if (value instanceof SimpleString) - { - return Boolean.valueOf(((SimpleString)value).toString()).booleanValue(); - } - else - { - throw new MessageFormatException("Invalid conversion"); - } } public byte getByteProperty(final String name) throws JMSException { - Object value = message.getProperty(new SimpleString(name)); - if (value == null) + try { - throw new NumberFormatException("Message property '" + name + "' not set."); + return message.getByteProperty(new SimpleString(name)); } - - if (value instanceof Byte) + catch (PropertyConversionException e) { - return ((Byte)value).byteValue(); + throw new MessageFormatException(e.getMessage()); } - else if (value instanceof SimpleString) - { - return Byte.parseByte(((SimpleString)value).toString()); - } - else - { - throw new MessageFormatException("Invalid conversion"); - } } public short getShortProperty(final String name) throws JMSException { - Object value = message.getProperty(new SimpleString(name)); - if (value == null) + try { - throw new NumberFormatException("Message property '" + name + "' not set."); + return message.getShortProperty(new SimpleString(name)); } - - if (value instanceof Byte) + catch (PropertyConversionException e) { - return ((Byte)value).shortValue(); + throw new MessageFormatException(e.getMessage()); } - else if (value instanceof Short) - { - return ((Short)value).shortValue(); - } - else if (value instanceof SimpleString) - { - return Short.parseShort(((SimpleString)value).toString()); - } - else - { - throw new MessageFormatException("Invalid conversion"); - } } public int getIntProperty(final String name) throws JMSException @@ -679,33 +646,14 @@ return message.getDeliveryCount(); } - Object value = message.getProperty(new SimpleString(name)); - - if (value == null) + try { - return Integer.valueOf(null); + return message.getIntProperty(new SimpleString(name)); } - - if (value instanceof Byte) + catch (PropertyConversionException e) { - return ((Byte)value).intValue(); + throw new MessageFormatException(e.getMessage()); } - else if (value instanceof Short) - { - return ((Short)value).intValue(); - } - else if (value instanceof Integer) - { - return ((Integer)value).intValue(); - } - else if (value instanceof SimpleString) - { - return Integer.parseInt(((SimpleString)value).toString()); - } - else - { - throw new MessageFormatException("Invalid conversion"); - } } public long getLongProperty(final String name) throws JMSException @@ -715,85 +663,38 @@ return message.getDeliveryCount(); } - Object value = message.getProperty(new SimpleString(name)); - - if (value == null) + try { - return Long.valueOf(null); + return message.getLongProperty(new SimpleString(name)); } - - if (value instanceof Byte) + catch (PropertyConversionException e) { - return ((Byte)value).longValue(); + throw new MessageFormatException(e.getMessage()); } - else if (value instanceof Short) - { - return ((Short)value).longValue(); - } - else if (value instanceof Integer) - { - return ((Integer)value).longValue(); - } - else if (value instanceof Long) - { - return ((Long)value).longValue(); - } - else if (value instanceof SimpleString) - { - return Long.parseLong(((SimpleString)value).toString()); - } - else - { - throw new MessageFormatException("Invalid conversion"); - } } public float getFloatProperty(final String name) throws JMSException { - Object value = message.getProperty(new SimpleString(name)); - if (value == null) + try { - return Float.valueOf(null).floatValue(); + return message.getFloatProperty(new SimpleString(name)); } - - if (value instanceof Float) + catch (PropertyConversionException e) { - return ((Float)value).floatValue(); + throw new MessageFormatException(e.getMessage()); } - else if (value instanceof SimpleString) - { - return Float.parseFloat(((SimpleString)value).toString()); - } - else - { - throw new MessageFormatException("Invalid conversion"); - } } public double getDoubleProperty(final String name) throws JMSException { - Object value = message.getProperty(new SimpleString(name)); - if (value == null) + try { - return Double.valueOf(null).doubleValue(); + return message.getDoubleProperty(new SimpleString(name)); } - - if (value instanceof Float) + catch (PropertyConversionException e) { - return ((Float)value).doubleValue(); + throw new MessageFormatException(e.getMessage()); } - else if (value instanceof Double) - { - return ((Double)value).doubleValue(); - } - else if (value instanceof SimpleString) - { - return Double.parseDouble(((SimpleString)value).toString()); - } - else - { - throw new MessageFormatException("Invalid conversion"); - } } public String getStringProperty(final String name) throws JMSException @@ -802,56 +703,22 @@ { return String.valueOf(message.getDeliveryCount()); } - Object value; - if (JMSXGROUPID.equals(name)) - { - value = message.getProperty(MessageImpl.HDR_GROUP_ID); - } - else - { - value = message.getProperty(new SimpleString(name)); - } - if (value == null) - { - return null; - } - if (value instanceof SimpleString) + try { - return ((SimpleString)value).toString(); + if (JMSXGROUPID.equals(name)) + { + return message.getStringProperty(MessageImpl.HDR_GROUP_ID); + } + else + { + return message.getStringProperty(new SimpleString(name)); + } } - else if (value instanceof Boolean) + catch (PropertyConversionException e) { - return value.toString(); + throw new MessageFormatException(e.getMessage()); } - else if (value instanceof Byte) - { - return value.toString(); - } - else if (value instanceof Short) - { - return value.toString(); - } - else if (value instanceof Integer) - { - return value.toString(); - } - else if (value instanceof Long) - { - return value.toString(); - } - else if (value instanceof Float) - { - return value.toString(); - } - else if (value instanceof Double) - { - return value.toString(); - } - else - { - throw new MessageFormatException("Invalid conversion"); - } } public Object getObjectProperty(final String name) throws JMSException @@ -892,51 +759,44 @@ public void setBooleanProperty(final String name, final boolean value) throws JMSException { - Boolean b = Boolean.valueOf(value); - checkProperty(name, b); - message.putBooleanProperty(new SimpleString(name), b); + checkProperty(name, value); + message.putBooleanProperty(new SimpleString(name), value); } public void setByteProperty(final String name, final byte value) throws JMSException { - Byte b = new Byte(value); - checkProperty(name, b); + checkProperty(name, value); message.putByteProperty(new SimpleString(name), value); } public void setShortProperty(final String name, final short value) throws JMSException { - Short s = new Short(value); - checkProperty(name, s); + checkProperty(name, value); message.putShortProperty(new SimpleString(name), value); } public void setIntProperty(final String name, final int value) throws JMSException { - Integer i = new Integer(value); - checkProperty(name, i); + checkProperty(name, value); message.putIntProperty(new SimpleString(name), value); } public void setLongProperty(final String name, final long value) throws JMSException { - Long l = new Long(value); - checkProperty(name, l); + checkProperty(name, value); message.putLongProperty(new SimpleString(name), value); } public void setFloatProperty(final String name, final float value) throws JMSException { - Float f = new Float(value); - checkProperty(name, f); - message.putFloatProperty(new SimpleString(name), f); + checkProperty(name, value); + message.putFloatProperty(new SimpleString(name), value); } public void setDoubleProperty(final String name, final double value) throws JMSException { - Double d = new Double(value); - checkProperty(name, d); - message.putDoubleProperty(new SimpleString(name), d); + checkProperty(name, value); + message.putDoubleProperty(new SimpleString(name), value); } public void setStringProperty(final String name, final String value) throws JMSException @@ -989,44 +849,14 @@ return; } - SimpleString key = new SimpleString(name); - - if (value instanceof Boolean) + try { - message.putBooleanProperty(key, (Boolean)value); + message.putObjectProperty(new SimpleString(name), value); } - else if (value instanceof Byte) + catch (PropertyConversionException e) { - message.putByteProperty(key, (Byte)value); + throw new MessageFormatException(e.getMessage()); } - else if (value instanceof Short) - { - message.putShortProperty(key, (Short)value); - } - else if (value instanceof Integer) - { - message.putIntProperty(key, (Integer)value); - } - else if (value instanceof Long) - { - message.putLongProperty(key, (Long)value); - } - else if (value instanceof Float) - { - message.putFloatProperty(key, (Float)value); - } - else if (value instanceof Double) - { - message.putDoubleProperty(key, (Double)value); - } - else if (value instanceof String) - { - message.putStringProperty(key, new SimpleString((String)value)); - } - else - { - throw new MessageFormatException("Invalid property type"); - } } public void acknowledge() throws JMSException Modified: trunk/tests/src/org/hornetq/tests/integration/InterceptorTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/InterceptorTest.java 2009-11-05 14:23:20 UTC (rev 8223) +++ trunk/tests/src/org/hornetq/tests/integration/InterceptorTest.java 2009-11-05 14:38:23 UTC (rev 8224) @@ -272,7 +272,7 @@ { ClientMessage message = consumer.receive(1000); - assertEquals("orange", ((SimpleString)message.getProperty(key)).toString()); + assertEquals("orange", ((SimpleString)message.getObjectProperty(key)).toString()); } server.getRemotingService().removeInterceptor(interceptor); @@ -290,7 +290,7 @@ { ClientMessage message = consumer.receive(1000); - assertEquals("apple", ((SimpleString)message.getProperty(key)).toString()); + assertEquals("apple", ((SimpleString)message.getObjectProperty(key)).toString()); } session.close(); @@ -365,7 +365,7 @@ { ClientMessage message = consumer.receive(1000); - assertEquals("orange", ((SimpleString)message.getProperty(key)).toString()); + assertEquals("orange", ((SimpleString)message.getObjectProperty(key)).toString()); } sf.removeInterceptor(interceptor); @@ -383,7 +383,7 @@ { ClientMessage message = consumer.receive(1000); - assertEquals("apple", ((SimpleString)message.getProperty(key)).toString()); + assertEquals("apple", ((SimpleString)message.getObjectProperty(key)).toString()); } session.close(); @@ -460,10 +460,10 @@ { ClientMessage message = consumer.receive(1000); - assertEquals(1, ((Integer)message.getProperty("a")).intValue()); - assertEquals(2, ((Integer)message.getProperty("b")).intValue()); - assertEquals(3, ((Integer)message.getProperty("c")).intValue()); - assertEquals(4, ((Integer)message.getProperty("d")).intValue()); + assertEquals(1, ((Integer)message.getObjectProperty("a")).intValue()); + assertEquals(2, ((Integer)message.getObjectProperty("b")).intValue()); + assertEquals(3, ((Integer)message.getObjectProperty("c")).intValue()); + assertEquals(4, ((Integer)message.getObjectProperty("d")).intValue()); } server.getRemotingService().removeInterceptor(interceptor2); @@ -479,10 +479,10 @@ { ClientMessage message = consumer.receive(1000); - assertEquals(1, ((Integer)message.getProperty("a")).intValue()); - assertNull(message.getProperty("b")); - assertEquals(3, ((Integer)message.getProperty("c")).intValue()); - assertEquals(4, ((Integer)message.getProperty("d")).intValue()); + assertEquals(1, ((Integer)message.getObjectProperty("a")).intValue()); + assertNull(message.getObjectProperty("b")); + assertEquals(3, ((Integer)message.getObjectProperty("c")).intValue()); + assertEquals(4, ((Integer)message.getObjectProperty("d")).intValue()); } @@ -549,10 +549,10 @@ { ClientMessage message = consumer.receive(1000); - assertEquals(1, ((Integer)message.getProperty("a")).intValue()); - assertEquals(2, ((Integer)message.getProperty("b")).intValue()); - assertEquals(3, ((Integer)message.getProperty("c")).intValue()); - assertEquals(4, ((Integer)message.getProperty("d")).intValue()); + assertEquals(1, ((Integer)message.getObjectProperty("a")).intValue()); + assertEquals(2, ((Integer)message.getObjectProperty("b")).intValue()); + assertEquals(3, ((Integer)message.getObjectProperty("c")).intValue()); + assertEquals(4, ((Integer)message.getObjectProperty("d")).intValue()); } sf.removeInterceptor(interceptor2); @@ -568,10 +568,10 @@ { ClientMessage message = consumer.receive(1000); - assertEquals(1, ((Integer)message.getProperty("a")).intValue()); - assertNull(message.getProperty("b")); - assertEquals(3, ((Integer)message.getProperty("c")).intValue()); - assertEquals(4, ((Integer)message.getProperty("d")).intValue()); + assertEquals(1, ((Integer)message.getObjectProperty("a")).intValue()); + assertNull(message.getObjectProperty("b")); + assertEquals(3, ((Integer)message.getObjectProperty("c")).intValue()); + assertEquals(4, ((Integer)message.getObjectProperty("d")).intValue()); } Modified: trunk/tests/src/org/hornetq/tests/integration/client/ConsumerTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/ConsumerTest.java 2009-11-05 14:23:20 UTC (rev 8223) +++ trunk/tests/src/org/hornetq/tests/integration/client/ConsumerTest.java 2009-11-05 14:38:23 UTC (rev 8224) @@ -365,7 +365,7 @@ { ClientMessage message = (ClientMessage)consumer.receive(); - int count = (Integer)message.getProperty("count"); + int count = (Integer)message.getObjectProperty("count"); boolean redelivered = message.getDeliveryCount() > 1; @@ -425,7 +425,7 @@ { ClientMessage message = (ClientMessage)consumer.receive(); - int count = (Integer)message.getProperty("count"); + int count = (Integer)message.getObjectProperty("count"); boolean redelivered = message.getDeliveryCount() > 1; Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/failover/AsynchronousFailoverTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/failover/AsynchronousFailoverTest.java 2009-11-05 14:23:20 UTC (rev 8223) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/failover/AsynchronousFailoverTest.java 2009-11-05 14:38:23 UTC (rev 8224) @@ -260,7 +260,7 @@ // messages must remain ordered but there could be a "jump" if messages // are missing or duplicated - int count = (Integer)message.getProperty("counter"); + int count = (Integer)message.getObjectProperty("counter"); counts.add(count); if (count != lastCount + 1) { @@ -379,7 +379,7 @@ // There may be some missing or duplicate messages - but the order should be correct - int count = (Integer)message.getProperty("counter"); + int count = (Integer)message.getObjectProperty("counter"); assertTrue("count:" + count + " last count:" + lastCount, count >= lastCount); Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java 2009-11-05 14:23:20 UTC (rev 8223) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java 2009-11-05 14:38:23 UTC (rev 8224) @@ -155,7 +155,7 @@ assertMessageBody(i, message); - assertEquals(i, message.getProperty("counter")); + assertEquals(i, message.getObjectProperty("counter")); message.acknowledge(); } @@ -227,7 +227,7 @@ assertMessageBody(i, message); - assertEquals(i, message.getProperty("counter")); + assertEquals(i, message.getObjectProperty("counter")); message.acknowledge(); } @@ -393,7 +393,7 @@ assertMessageBody(i, message); - assertEquals(i, message.getProperty("counter")); + assertEquals(i, message.getObjectProperty("counter")); message.acknowledge(); } @@ -482,7 +482,7 @@ assertMessageBody(i, message); - assertEquals(i, message.getProperty("counter")); + assertEquals(i, message.getObjectProperty("counter")); message.acknowledge(); } @@ -553,7 +553,7 @@ assertMessageBody(i, message); - assertEquals(i, message.getProperty("counter")); + assertEquals(i, message.getObjectProperty("counter")); message.acknowledge(); } @@ -636,7 +636,7 @@ assertMessageBody(i, message); - assertEquals(i, message.getProperty("counter")); + assertEquals(i, message.getObjectProperty("counter")); message.acknowledge(); } @@ -659,7 +659,7 @@ assertMessageBody(i, message); - assertEquals(i, message.getProperty("counter")); + assertEquals(i, message.getObjectProperty("counter")); message.acknowledge(); } @@ -964,7 +964,7 @@ assertMessageBody(i, message); - assertEquals(i, message.getProperty("counter")); + assertEquals(i, message.getObjectProperty("counter")); message.acknowledge(); } @@ -1041,7 +1041,7 @@ assertMessageBody(i, message); - assertEquals(i, message.getProperty("counter")); + assertEquals(i, message.getObjectProperty("counter")); message.acknowledge(); } @@ -1127,7 +1127,7 @@ assertMessageBody(i, message); - assertEquals(i, message.getProperty("counter")); + assertEquals(i, message.getObjectProperty("counter")); message.acknowledge(); } @@ -1228,7 +1228,7 @@ assertMessageBody(i, message); - assertEquals(i, message.getProperty("counter")); + assertEquals(i, message.getObjectProperty("counter")); message.acknowledge(); } @@ -1393,7 +1393,7 @@ assertMessageBody(i, message); - assertEquals(i, message.getProperty("counter")); + assertEquals(i, message.getObjectProperty("counter")); message.acknowledge(); } @@ -1465,7 +1465,7 @@ assertMessageBody(i, message); - assertEquals(i, message.getProperty("counter")); + assertEquals(i, message.getObjectProperty("counter")); } fail(session, latch); @@ -1482,7 +1482,7 @@ assertMessageBody(i, message); - assertEquals(i, message.getProperty("counter")); + assertEquals(i, message.getObjectProperty("counter")); message.acknowledge(); } @@ -1545,7 +1545,7 @@ assertMessageBody(i, message); - assertEquals(i, message.getProperty("counter")); + assertEquals(i, message.getObjectProperty("counter")); } fail(session, latch); @@ -1564,7 +1564,7 @@ assertMessageBody(i, message); - assertEquals(i, message.getProperty("counter")); + assertEquals(i, message.getObjectProperty("counter")); message.acknowledge(); } @@ -1628,7 +1628,7 @@ assertMessageBody(i, message); - assertEquals(i, message.getProperty("counter")); + assertEquals(i, message.getObjectProperty("counter")); message.acknowledge(); } @@ -1658,7 +1658,7 @@ assertMessageBody(i, message); - assertEquals(i, message.getProperty("counter")); + assertEquals(i, message.getObjectProperty("counter")); message.acknowledge(); } @@ -1882,7 +1882,7 @@ assertMessageBody(i, message); - assertEquals(i, message.getProperty("counter")); + assertEquals(i, message.getObjectProperty("counter")); message.acknowledge(); } @@ -2022,7 +2022,7 @@ assertMessageBody(i, message); - assertEquals(i, message.getProperty("counter")); + assertEquals(i, message.getObjectProperty("counter")); message.acknowledge(); } Modified: trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java 2009-11-05 14:23:20 UTC (rev 8223) +++ trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java 2009-11-05 14:38:23 UTC (rev 8224) @@ -50,7 +50,7 @@ * * */ -public class LargeMessageTestBase extends ServiceTestBase +public abstract class LargeMessageTestBase extends ServiceTestBase { // Constants ----------------------------------------------------- Added: trunk/tests/src/org/hornetq/tests/unit/core/message/impl/MessagePropertyConversionTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/message/impl/MessagePropertyConversionTest.java (rev 0) +++ trunk/tests/src/org/hornetq/tests/unit/core/message/impl/MessagePropertyConversionTest.java 2009-11-05 14:38:23 UTC (rev 8224) @@ -0,0 +1,350 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.tests.unit.core.message.impl; + +import static org.hornetq.tests.util.RandomUtil.randomBoolean; +import static org.hornetq.tests.util.RandomUtil.randomByte; +import static org.hornetq.tests.util.RandomUtil.randomDouble; +import static org.hornetq.tests.util.RandomUtil.randomFloat; +import static org.hornetq.tests.util.RandomUtil.randomInt; +import static org.hornetq.tests.util.RandomUtil.randomShort; +import static org.hornetq.tests.util.RandomUtil.randomSimpleString; +import static org.hornetq.tests.util.UnitTestCase.assertEqualsByteArrays; +import junit.framework.TestCase; + +import org.hornetq.core.client.impl.ClientMessageImpl; +import org.hornetq.core.message.Message; +import org.hornetq.core.message.PropertyConversionException; +import org.hornetq.tests.util.RandomUtil; +import org.hornetq.utils.SimpleString; + +/** + * A MessagePropertyConversionTest + * + * @author jmesnil + * + * + */ +public class MessagePropertyConversionTest extends TestCase +{ + + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + private Message msg; + + private SimpleString key; + + private SimpleString unknownKey = new SimpleString("this.key.is.never.used"); + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + + @Override + protected void setUp() throws Exception + { + super.setUp(); + + key = randomSimpleString(); + msg = new ClientMessageImpl(false); + } + + @Override + protected void tearDown() throws Exception + { + key = null; + msg = null; + + super.tearDown(); + } + + public void testBooleanProperty() throws Exception + { + Boolean val = randomBoolean(); + msg.putBooleanProperty(key, val); + + assertEquals(val, msg.getBooleanProperty(key)); + assertEquals(Boolean.toString(val), msg.getStringProperty(key)); + + msg.putStringProperty(key, new SimpleString(Boolean.toString(val))); + assertEquals(val, msg.getBooleanProperty(key)); + + try + { + msg.putByteProperty(key, randomByte()); + msg.getBooleanProperty(key); + fail(); + } + catch (PropertyConversionException e) + { + } + + assertFalse(msg.getBooleanProperty(unknownKey)); + } + + public void testByteProperty() throws Exception + { + Byte val = randomByte(); + msg.putByteProperty(key, val); + + assertEquals(val, msg.getByteProperty(key)); + assertEquals(Byte.toString(val), msg.getStringProperty(key)); + + msg.putStringProperty(key, new SimpleString(Byte.toString(val))); + assertEquals(val, msg.getByteProperty(key)); + + try + { + msg.putBooleanProperty(key, randomBoolean()); + msg.getByteProperty(key); + fail(); + } + catch (PropertyConversionException e) + { + } + + try + { + msg.getByteProperty(unknownKey); + fail(); + } + catch (NumberFormatException e) + { + } + } + + public void testIntProperty() throws Exception + { + Integer val = randomInt(); + msg.putIntProperty(key, val); + + assertEquals(val, msg.getIntProperty(key)); + assertEquals(Integer.toString(val), msg.getStringProperty(key)); + + msg.putStringProperty(key, new SimpleString(Integer.toString(val))); + assertEquals(val, msg.getIntProperty(key)); + + Byte byteVal = randomByte(); + msg.putByteProperty(key, byteVal); + assertEquals(Integer.valueOf(byteVal), msg.getIntProperty(key)); + + try + { + msg.putBooleanProperty(key, randomBoolean()); + msg.getIntProperty(key); + fail(); + } + catch (PropertyConversionException e) + { + } + + try + { + msg.getIntProperty(unknownKey); + fail(); + } + catch (NumberFormatException e) + { + } + } + + public void testLongProperty() throws Exception + { + Long val = RandomUtil.randomLong(); + msg.putLongProperty(key, val); + + assertEquals(val, msg.getLongProperty(key)); + assertEquals(Long.toString(val), msg.getStringProperty(key)); + + msg.putStringProperty(key, new SimpleString(Long.toString(val))); + assertEquals(val, msg.getLongProperty(key)); + + Byte byteVal = randomByte(); + msg.putByteProperty(key, byteVal); + assertEquals(Long.valueOf(byteVal), msg.getLongProperty(key)); + + Short shortVal = randomShort(); + msg.putShortProperty(key, shortVal); + assertEquals(Long.valueOf(shortVal), msg.getLongProperty(key)); + + Integer intVal = randomInt(); + msg.putIntProperty(key, intVal); + assertEquals(Long.valueOf(intVal), msg.getLongProperty(key)); + + try + { + msg.putBooleanProperty(key, randomBoolean()); + msg.getLongProperty(key); + fail(); + } + catch (PropertyConversionException e) + { + } + + try + { + msg.getLongProperty(unknownKey); + fail(); + } + catch (NumberFormatException e) + { + } + } + + public void testDoubleProperty() throws Exception + { + Double val = randomDouble(); + msg.putDoubleProperty(key, val); + + assertEquals(val, msg.getDoubleProperty(key)); + assertEquals(Double.toString(val), msg.getStringProperty(key)); + + msg.putStringProperty(key, new SimpleString(Double.toString(val))); + assertEquals(val, msg.getDoubleProperty(key)); + + try + { + msg.putBooleanProperty(key, randomBoolean()); + msg.getDoubleProperty(key); + fail(); + } + catch (PropertyConversionException e) + { + } + + try + { + msg.getDoubleProperty(unknownKey); + fail(); + } + catch (Exception e) + { + } + } + + public void testFloatProperty() throws Exception + { + Float val = randomFloat(); + msg.putFloatProperty(key, val); + + assertEquals(val, msg.getFloatProperty(key)); + assertEquals(Double.valueOf(val), msg.getDoubleProperty(key)); + assertEquals(Float.toString(val), msg.getStringProperty(key)); + + msg.putStringProperty(key, new SimpleString(Float.toString(val))); + assertEquals(val, msg.getFloatProperty(key)); + + try + { + msg.putBooleanProperty(key, randomBoolean()); + msg.getFloatProperty(key); + fail(); + } + catch (PropertyConversionException e) + { + } + + try + { + msg.getFloatProperty(unknownKey); + fail(); + } + catch (Exception e) + { + } + } + + public void testShortProperty() throws Exception + { + Short val = randomShort(); + msg.putShortProperty(key, val); + + assertEquals(val, msg.getShortProperty(key)); + assertEquals(Integer.valueOf(val), msg.getIntProperty(key)); + assertEquals(Short.toString(val), msg.getStringProperty(key)); + + msg.putStringProperty(key, new SimpleString(Short.toString(val))); + assertEquals(val, msg.getShortProperty(key)); + + Byte byteVal = randomByte(); + msg.putByteProperty(key, byteVal); + assertEquals(Short.valueOf(byteVal), msg.getShortProperty(key)); + + try + { + msg.putBooleanProperty(key, randomBoolean()); + msg.getShortProperty(key); + fail(); + } + catch (PropertyConversionException e) + { + } + + try + { + msg.getShortProperty(unknownKey); + fail(); + } + catch (NumberFormatException e) + { + } + } + + public void testStringProperty() throws Exception + { + SimpleString strVal = randomSimpleString(); + msg.putStringProperty(key, strVal); + assertEquals(strVal.toString(), msg.getStringProperty(key)); + } + + public void testBytesProperty() throws Exception + { + byte[] val = RandomUtil.randomBytes(); + msg.putBytesProperty(key, val); + + assertEqualsByteArrays(val, msg.getBytesProperty(key)); + + try + { + msg.putBooleanProperty(key, randomBoolean()); + msg.getBytesProperty(key); + fail(); + } + catch (PropertyConversionException e) + { + } + + try + { + msg.getBytesProperty(unknownKey); + fail(); + } + catch (Exception e) + { + } + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- + +} Modified: trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java 2009-11-05 14:23:20 UTC (rev 8223) +++ trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java 2009-11-05 14:38:23 UTC (rev 8224) @@ -22,16 +22,21 @@ import org.hornetq.core.exception.HornetQException; import org.hornetq.core.filter.Filter; +import org.hornetq.core.message.LargeMessageEncodingContext; +import org.hornetq.core.message.PropertyConversionException; import org.hornetq.core.paging.PagingStore; import org.hornetq.core.postoffice.Binding; import org.hornetq.core.postoffice.BindingType; import org.hornetq.core.postoffice.impl.BindingsImpl; import org.hornetq.core.remoting.spi.HornetQBuffer; -import org.hornetq.core.server.*; +import org.hornetq.core.server.Bindable; +import org.hornetq.core.server.MessageReference; +import org.hornetq.core.server.Queue; +import org.hornetq.core.server.RoutingContext; +import org.hornetq.core.server.ServerMessage; import org.hornetq.core.server.impl.RoutingContextImpl; import org.hornetq.core.transaction.Transaction; import org.hornetq.core.transaction.TransactionOperation; -import org.hornetq.core.message.LargeMessageEncodingContext; import org.hornetq.tests.util.UnitTestCase; import org.hornetq.utils.SimpleString; import org.hornetq.utils.TypedProperties; @@ -289,7 +294,7 @@ public void setOriginalHeaders(ServerMessage other, boolean expiry) { // TODO Auto-generated method stub - + } public Map<String, Object> toMap() @@ -455,7 +460,7 @@ public void encodeBody(HornetQBuffer bufferOut, LargeMessageEncodingContext context, int size) { - //To change body of implemented methods use File | Settings | File Templates. + // To change body of implemented methods use File | Settings | File Templates. } /* (non-Javadoc) @@ -680,6 +685,24 @@ { } + + /* (non-Javadoc) + * @see org.hornetq.core.message.Message#putObjectProperty(org.hornetq.utils.SimpleString, java.lang.Object) + */ + public void putObjectProperty(SimpleString key, Object value) + { + // TODO Auto-generated method stub + + } + + /* (non-Javadoc) + * @see org.hornetq.core.message.Message#putObjectProperty(java.lang.String, java.lang.Object) + */ + public void putObjectProperty(String key, Object value) + { + // TODO Auto-generated method stub + + } /* (non-Javadoc) * @see org.hornetq.core.message.Message#putTypedProperties(org.hornetq.utils.TypedProperties) @@ -774,16 +797,16 @@ */ public boolean containsProperty(String key) { - + return false; } /* (non-Javadoc) * @see org.hornetq.core.message.Message#getProperty(java.lang.String) */ - public Object getProperty(String key) + public Object getObjectProperty(String key) { - + return null; } @@ -792,8 +815,7 @@ */ public void putBooleanProperty(String key, boolean value) { - - + } /* (non-Javadoc) @@ -801,8 +823,7 @@ */ public void putByteProperty(String key, byte value) { - - + } /* (non-Javadoc) @@ -810,8 +831,7 @@ */ public void putBytesProperty(String key, byte[] value) { - - + } /* (non-Javadoc) @@ -819,8 +839,7 @@ */ public void putDoubleProperty(String key, double value) { - - + } /* (non-Javadoc) @@ -828,8 +847,7 @@ */ public void putFloatProperty(String key, float value) { - - + } /* (non-Javadoc) @@ -837,8 +855,7 @@ */ public void putIntProperty(String key, int value) { - - + } /* (non-Javadoc) @@ -846,8 +863,7 @@ */ public void putLongProperty(String key, long value) { - - + } /* (non-Javadoc) @@ -855,8 +871,7 @@ */ public void putShortProperty(String key, short value) { - - + } /* (non-Javadoc) @@ -864,8 +879,7 @@ */ public void putStringProperty(String key, String value) { - - + } /* (non-Javadoc) @@ -873,11 +887,101 @@ */ public Object removeProperty(String key) { - + return null; } /* (non-Javadoc) + * @see org.hornetq.core.message.Message#getStringProperty(org.hornetq.utils.SimpleString) + */ + public String getStringProperty(SimpleString key) + { + // TODO Auto-generated method stub + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.message.Message#getBooleanProperty(org.hornetq.utils.SimpleString) + */ + public Boolean getBooleanProperty(SimpleString key) + { + // TODO Auto-generated method stub + return false; + } + + /* (non-Javadoc) + * @see org.hornetq.core.message.Message#getByteProperty(org.hornetq.utils.SimpleString) + */ + public Byte getByteProperty(SimpleString key) + { + // TODO Auto-generated method stub + return 0; + } + + /* (non-Javadoc) + * @see org.hornetq.core.message.Message#getBytesProperty(org.hornetq.utils.SimpleString) + */ + public byte[] getBytesProperty(SimpleString key) throws PropertyConversionException + { + // TODO Auto-generated method stub + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.message.Message#getDoubleProperty(org.hornetq.utils.SimpleString) + */ + public Double getDoubleProperty(SimpleString key) + { + // TODO Auto-generated method stub + return 0.0; + } + + /* (non-Javadoc) + * @see org.hornetq.core.message.Message#getFloatProperty(org.hornetq.utils.SimpleString) + */ + public Float getFloatProperty(SimpleString key) + { + // TODO Auto-generated method stub + return 0f; + } + + /* (non-Javadoc) + * @see org.hornetq.core.message.Message#getIntProperty(org.hornetq.utils.SimpleString) + */ + public Integer getIntProperty(SimpleString key) + { + // TODO Auto-generated method stub + return 0; + } + + /* (non-Javadoc) + * @see org.hornetq.core.message.Message#getLongProperty(org.hornetq.utils.SimpleString) + */ + public Long getLongProperty(SimpleString key) + { + // TODO Auto-generated method stub + return 0L; + } + + /* (non-Javadoc) + * @see org.hornetq.core.message.Message#getObjectProperty(org.hornetq.utils.SimpleString) + */ + public Object getObjectProperty(SimpleString key) + { + // TODO Auto-generated method stub + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.message.Message#getShortProperty(org.hornetq.utils.SimpleString) + */ + public Short getShortProperty(SimpleString key) + { + // TODO Auto-generated method stub + return 0; + } + + /* (non-Javadoc) * @see org.hornetq.core.server.ServerMessage#decrementRefCount(org.hornetq.core.paging.PagingStore, org.hornetq.core.server.MessageReference) */ public int decrementRefCount(PagingStore pagingStore, MessageReference reference) throws Exception @@ -1058,7 +1162,7 @@ public void route(ServerMessage message, RoutingContext context) throws Exception { // TODO Auto-generated method stub - + } } Modified: trunk/tests/src/org/hornetq/tests/util/ServiceTestBase.java =================================================================== --- trunk/tests/src/org/hornetq/tests/util/ServiceTestBase.java 2009-11-05 14:23:20 UTC (rev 8223) +++ trunk/tests/src/org/hornetq/tests/util/ServiceTestBase.java 2009-11-05 14:38:23 UTC (rev 8224) @@ -47,7 +47,7 @@ * @author <a href="mailto:clebert.suconic@jboss.com">Clebert Suconic</a> * */ -public class ServiceTestBase extends UnitTestCase +public abstract class ServiceTestBase extends UnitTestCase { // Constants -----------------------------------------------------

16 years, 2 months

1
0
0 / 0

JBoss hornetq SVN: r8223 - in trunk: src/main/org/hornetq/core/paging and 10 other directories.

by do-not-reply＠jboss.org

Author: timfox Date: 2009-11-05 09:23:20 -0500 (Thu, 05 Nov 2009) New Revision: 8223 Modified: trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java trunk/src/main/org/hornetq/core/paging/PagingManager.java trunk/src/main/org/hornetq/core/paging/PagingStore.java trunk/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java trunk/src/main/org/hornetq/core/paging/impl/PagingStoreFactoryNIO.java trunk/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java trunk/src/main/org/hornetq/core/paging/impl/TestSupportPageStore.java trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java trunk/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java trunk/src/main/org/hornetq/core/server/ServerMessage.java trunk/src/main/org/hornetq/core/server/impl/DeliveryImpl.java trunk/src/main/org/hornetq/core/server/impl/DistributorImpl.java trunk/src/main/org/hornetq/core/server/impl/DivertImpl.java trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java trunk/src/main/org/hornetq/core/server/impl/LastValueQueue.java trunk/src/main/org/hornetq/core/server/impl/MemoryManagerImpl.java trunk/src/main/org/hornetq/core/server/impl/MessageReferenceImpl.java trunk/src/main/org/hornetq/core/server/impl/QueueFactoryImpl.java trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java trunk/src/main/org/hornetq/core/server/impl/RoundRobinDistributor.java trunk/src/main/org/hornetq/core/server/impl/RoutingContextImpl.java trunk/src/main/org/hornetq/core/server/impl/ScheduledDeliveryHandlerImpl.java trunk/src/main/org/hornetq/core/server/impl/ServerInfo.java trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java trunk/src/main/org/hornetq/core/server/impl/ServerProducerCreditManager.java trunk/src/main/org/hornetq/core/server/impl/ServerProducerCreditManagerImpl.java trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java trunk/src/main/org/hornetq/core/server/impl/ServerSessionPacketHandler.java trunk/tests/src/org/hornetq/tests/integration/client/PagingTest.java trunk/tests/src/org/hornetq/tests/integration/paging/PageCrashTest.java trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingManagerImplTest.java trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java Log: refactor and cleanup of paging code Modified: trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -73,7 +73,7 @@ protected long messageID; - private SimpleString destination; + protected SimpleString destination; protected byte type; Modified: trunk/src/main/org/hornetq/core/paging/PagingManager.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/PagingManager.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/paging/PagingManager.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -13,12 +13,9 @@ package org.hornetq.core.paging; -import java.util.Collection; - import org.hornetq.core.journal.SequentialFile; import org.hornetq.core.postoffice.PostOffice; import org.hornetq.core.server.HornetQComponent; -import org.hornetq.core.server.ServerMessage; import org.hornetq.utils.SimpleString; /** @@ -57,35 +54,6 @@ void resumeDepages() throws Exception; /** - * To be used by transactions only. - * If you're sure you will page if isPaging, just call the method page and look at its return. - * @param destination - * @return - */ - boolean isPaging(SimpleString destination) throws Exception; - - /** - * Page, only if destination is in page mode. - * @param message - * @param sync - Sync should be called right after the write - * @return false if destination is not on page mode - */ - - // FIXME - why are these methods still on PagingManager??? - // The current code is doing a lookup every time through this class just to call page store!! - boolean page(ServerMessage message, boolean duplicateDetection) throws Exception; - - /** - * Page, only if destination is in page mode. - * @param message - * @return false if destination is not on page mode - */ - - // FIXME - why are these methods still on PagingManager??? - // The current code is doing a lookup every time through this class just to call page store!! - boolean page(ServerMessage message, long transactionId, boolean duplicateDetection) throws Exception; - - /** * Point to inform/restoring Transactions used when the messages were added into paging * */ void addTransaction(PageTransactionInfo pageTransaction); @@ -95,9 +63,6 @@ * */ PageTransactionInfo getTransaction(long transactionID); - /** Sync current-pages on disk for these destinations */ - void sync(Collection<SimpleString> destinationsToSync) throws Exception; - /** * @param transactionID */ Modified: trunk/src/main/org/hornetq/core/paging/PagingStore.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/PagingStore.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/paging/PagingStore.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -33,32 +33,26 @@ */ public interface PagingStore extends HornetQComponent { + SimpleString getAddress(); + int getNumberOfPages(); SimpleString getStoreName(); - /** Maximum number of bytes allowed in memory */ - long getMaxSizeBytes(); - AddressFullMessagePolicy getAddressFullMessagePolicy(); long getPageSizeBytes(); long getAddressSize(); - /** @return true if paging was started, or false if paging was already started before this call */ - boolean startPaging() throws Exception; - boolean isPaging(); void sync() throws Exception; - boolean page(PagedMessage message, boolean sync, boolean duplicateDetection) throws Exception; + boolean page(ServerMessage message, long transactionId, boolean duplicateDetection) throws Exception; - public boolean readPage() throws Exception; + boolean page(ServerMessage message, boolean duplicateDetection) throws Exception; - Page getCurrentPage(); - Page createPage(final int page) throws Exception; /** Modified: trunk/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -13,7 +13,6 @@ package org.hornetq.core.paging.impl; -import java.util.Collection; import java.util.List; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; @@ -27,7 +26,6 @@ import org.hornetq.core.paging.PagingStoreFactory; import org.hornetq.core.persistence.StorageManager; import org.hornetq.core.postoffice.PostOffice; -import org.hornetq.core.server.ServerMessage; import org.hornetq.core.settings.HierarchicalRepository; import org.hornetq.core.settings.impl.AddressSettings; import org.hornetq.utils.SimpleString; @@ -47,8 +45,6 @@ private volatile boolean started = false; - // private volatile boolean backup; - private final AtomicLong totalMemoryBytes = new AtomicLong(0); private final ConcurrentMap<SimpleString, PagingStore> stores = new ConcurrentHashMap<SimpleString, PagingStore>(); @@ -59,8 +55,6 @@ private final StorageManager storageManager; - private final boolean syncNonTransactional; - private final ConcurrentMap</*TransactionID*/Long, PageTransactionInfo> transactions = new ConcurrentHashMap<Long, PageTransactionInfo>(); // Static @@ -73,13 +67,11 @@ public PagingManagerImpl(final PagingStoreFactory pagingSPI, final StorageManager storageManager, - final HierarchicalRepository<AddressSettings> addressSettingsRepository, - final boolean syncNonTransactional) + final HierarchicalRepository<AddressSettings> addressSettingsRepository) { pagingStoreFactory = pagingSPI; this.addressSettingsRepository = addressSettingsRepository; this.storageManager = storageManager; - this.syncNonTransactional = syncNonTransactional; } // Public @@ -150,28 +142,6 @@ pagingStoreFactory.setPostOffice(postOffice); } - public boolean isPaging(final SimpleString destination) throws Exception - { - return getPageStore(destination).isPaging(); - } - - public boolean page(final ServerMessage message, final long transactionId, final boolean duplicateDetection) throws Exception - { - // The sync on transactions is done on commit only - return getPageStore(message.getDestination()).page(new PagedMessageImpl(message, transactionId), - false, - duplicateDetection); - } - - public boolean page(final ServerMessage message, final boolean duplicateDetection) throws Exception - { - // If non Durable, there is no need to sync as there is no requirement for persistence for those messages in case - // of crash - return getPageStore(message.getDestination()).page(new PagedMessageImpl(message), - syncNonTransactional && message.isDurable(), - duplicateDetection); - } - public void addTransaction(final PageTransactionInfo pageTransaction) { transactions.put(pageTransaction.getTransactionID(), pageTransaction); @@ -187,14 +157,6 @@ return transactions.get(id); } - public void sync(final Collection<SimpleString> destinationsToSync) throws Exception - { - for (SimpleString destination : destinationsToSync) - { - getPageStore(destination).sync(); - } - } - // HornetQComponent implementation // ------------------------------------------------------------------------------------------------ Modified: trunk/src/main/org/hornetq/core/paging/impl/PagingStoreFactoryNIO.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/impl/PagingStoreFactoryNIO.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/paging/impl/PagingStoreFactoryNIO.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -57,6 +57,8 @@ private final String directory; private final ExecutorFactory executorFactory; + + protected final boolean syncNonTransactional; private PagingManager pagingManager; @@ -68,11 +70,14 @@ // Constructors -------------------------------------------------- - public PagingStoreFactoryNIO(final String directory, final ExecutorFactory executorFactory) + public PagingStoreFactoryNIO(final String directory, final ExecutorFactory executorFactory, + final boolean syncNonTransactional) { this.directory = directory; this.executorFactory = executorFactory; + + this.syncNonTransactional = syncNonTransactional; } // Public -------------------------------------------------------- @@ -81,17 +86,19 @@ { } - public synchronized PagingStore newStore(final SimpleString destinationName, final AddressSettings settings) throws Exception + public synchronized PagingStore newStore(final SimpleString address, final AddressSettings settings) throws Exception { - return new PagingStoreImpl(pagingManager, + return new PagingStoreImpl(address, + pagingManager, storageManager, postOffice, null, this, - destinationName, + address, settings, - executorFactory.getExecutor()); + executorFactory.getExecutor(), + syncNonTransactional); } /** @@ -168,31 +175,33 @@ BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(addressFile))); - String destination; + String addressString; try { - destination = reader.readLine(); + addressString = reader.readLine(); } finally { reader.close(); } - SimpleString destinationName = new SimpleString(destination); + SimpleString address = new SimpleString(addressString); SequentialFileFactory factory = newFileFactory(guid); - AddressSettings settings = addressSettingsRepository.getMatch(destinationName.toString()); + AddressSettings settings = addressSettingsRepository.getMatch(address.toString()); - PagingStore store = new PagingStoreImpl(pagingManager, + PagingStore store = new PagingStoreImpl(address, + pagingManager, storageManager, postOffice, factory, this, - destinationName, + address, settings, - executorFactory.getExecutor()); + executorFactory.getExecutor(), + syncNonTransactional); storesReturn.add(store); } Modified: trunk/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -64,6 +64,8 @@ // Attributes ---------------------------------------------------- + private final SimpleString address; + private final StorageManager storageManager; private final PostOffice postOffice; @@ -110,15 +112,17 @@ * We need to perform checks on currentPage with minimal locking * */ private final ReadWriteLock currentPageLock = new ReentrantReadWriteLock(); - + private final ServerProducerCreditManager creditManager; - + private boolean exceededAvailableCredits; - + private volatile boolean running = false; - + private final AtomicLong availableProducerCredits = new AtomicLong(0); - + + protected final boolean syncNonTransactional; + // Static -------------------------------------------------------- private static final boolean isTrace = log.isTraceEnabled(); @@ -133,20 +137,24 @@ // Constructors -------------------------------------------------- - public PagingStoreImpl(final PagingManager pagingManager, + public PagingStoreImpl(final SimpleString address, + final PagingManager pagingManager, final StorageManager storageManager, final PostOffice postOffice, final SequentialFileFactory fileFactory, final PagingStoreFactory storeFactory, final SimpleString storeName, final AddressSettings addressSettings, - final Executor executor) + final Executor executor, + final boolean syncNonTransactional) { if (pagingManager == null) { throw new IllegalStateException("Paging Manager can't be null"); } + this.address = address; + this.storageManager = storageManager; this.postOffice = postOffice; @@ -166,25 +174,26 @@ this.fileFactory = fileFactory; this.storeFactory = storeFactory; - + this.availableProducerCredits.set(maxSize); - - this.creditManager = new ServerProducerCreditManagerImpl(this); + + this.creditManager = new ServerProducerCreditManagerImpl(this); + + this.syncNonTransactional = syncNonTransactional; } // Public -------------------------------------------------------- // PagingStore implementation ------------------------------------ - public long getAddressSize() + public SimpleString getAddress() { - return sizeInBytes.get(); + return address; } - /** Maximum number of bytes allowed in memory */ - public long getMaxSizeBytes() + public long getAddressSize() { - return maxSize; + return sizeInBytes.get(); } public AddressFullMessagePolicy getAddressFullMessagePolicy() @@ -200,6 +209,7 @@ public boolean isPaging() { currentPageLock.readLock().lock(); + try { if (addressFullMessagePolicy != AddressFullMessagePolicy.PAGE) @@ -226,27 +236,27 @@ { return storeName; } - + public boolean isExceededAvailableCredits() { return exceededAvailableCredits; } - + public synchronized int getAvailableProducerCredits(final int credits) - { + { if (maxSize != -1 && addressFullMessagePolicy == AddressFullMessagePolicy.BLOCK) - { + { long avail = availableProducerCredits.get(); - + if (avail > 0) { long take = Math.min(avail, credits); - + availableProducerCredits.addAndGet(-take); - + return (int)take; } - + return 0; } else @@ -254,243 +264,62 @@ return credits; } } - + public void returnProducerCredits(final int credits) { checkReleaseProducerFlowControlCredits(-credits); } + - private synchronized void checkReleaseProducerFlowControlCredits(final long size) + public void addSize(final ServerMessage message, final boolean add) throws Exception { - if (addressFullMessagePolicy == AddressFullMessagePolicy.BLOCK && maxSize != -1) - { - long avail = availableProducerCredits.addAndGet(-size); - - if (avail > 0) - { - int used = creditManager.creditsReleased((int)avail); - - long num = availableProducerCredits.addAndGet(-used); - - if (num < 0) - { - log.warn("Available credits has gone negative"); - - exceededAvailableCredits = true; - } - } - } - } - - public void addSize(final ServerMessage message, final boolean add) throws Exception - { long size = message.getMemoryEstimate(); - + if (add) { checkReleaseProducerFlowControlCredits(size); - + addSize(size); } else { checkReleaseProducerFlowControlCredits(-size); - + addSize(-size); } } - + public void addSize(final MessageReference reference, final boolean add) throws Exception { long size = reference.getMemoryEstimate(); - + if (add) { checkReleaseProducerFlowControlCredits(size); - + addSize(size); } else { checkReleaseProducerFlowControlCredits(-size); - + addSize(-size); } } - private void addSize(final long size) throws Exception - { - if (addressFullMessagePolicy != AddressFullMessagePolicy.PAGE) - { - addAddressSize(size); - - pagingManager.addSize(size); - - return; - } - else - { - pagingManager.addSize(size); - - final long addressSize = addAddressSize(size); - - if (size > 0) - { - if (maxSize > 0 && addressSize > maxSize) - { - if (startPaging()) - { - if (isTrace) - { - trace("Starting paging on " + getStoreName() + ", size = " + addressSize + ", maxSize=" + maxSize); - } - } - } - } - else - { - // When in Global mode, we use the default page size as the mark to start depage - if (maxSize > 0 && currentPage != null && addressSize <= maxSize - pageSize && !depaging.get()) - { - if (startDepaging()) - { - if (isTrace) - { - trace("Starting depaging Thread, size = " + addressSize); - } - } - } - } - - return; - } + public boolean page(final ServerMessage message, final long transactionID, final boolean duplicateDetection) throws Exception + { + // The sync on transactions is done on commit only + return page(message, transactionID, false, duplicateDetection); } - // TODO all of this can be simplified - public boolean page(final PagedMessage message, final boolean sync, final boolean duplicateDetection) throws Exception + public boolean page(final ServerMessage message, final boolean duplicateDetection) throws Exception { - if (!running) - { - throw new IllegalStateException("PagingStore(" + getStoreName() + ") not initialized"); - } - - boolean full = isFull(); - - if (addressFullMessagePolicy == AddressFullMessagePolicy.DROP) - { - if (full) - { - if (!printedDropMessagesWarning) - { - printedDropMessagesWarning = true; - - log.warn("Messages are being dropped on address " + getStoreName()); - } - - // Address is full, we just pretend we are paging, and drop the data - return true; - } - else - { - return false; - } - } - else if (addressFullMessagePolicy == AddressFullMessagePolicy.BLOCK) - { - return false; - } - - // We need to ensure a read lock, as depage could change the paging state - currentPageLock.readLock().lock(); - - try - { - // First check done concurrently, to avoid synchronization and increase throughput - if (currentPage == null) - { - return false; - } - } - finally - { - currentPageLock.readLock().unlock(); - } - - writeLock.lock(); - - try - { - if (currentPage == null) - { - return false; - } - - if (duplicateDetection) - { - // We set the duplicate detection header to prevent the message being depaged more than once in case of - // failure during depage - - ServerMessage msg = message.getMessage(storageManager); - - byte[] bytes = new byte[8]; - - ByteBuffer buff = ByteBuffer.wrap(bytes); - - buff.putLong(msg.getMessageID()); - - msg.putBytesProperty(MessageImpl.HDR_DUPLICATE_DETECTION_ID, bytes); - } - - int bytesToWrite = message.getEncodeSize() + PageImpl.SIZE_RECORD; - - if (currentPageSize.addAndGet(bytesToWrite) > pageSize && currentPage.getNumberOfMessages() > 0) - { - // Make sure nothing is currently validating or using currentPage - currentPageLock.writeLock().lock(); - try - { - openNewPage(); - - // openNewPage will set currentPageSize to zero, we need to set it again - currentPageSize.addAndGet(bytesToWrite); - } - finally - { - currentPageLock.writeLock().unlock(); - } - } - - currentPageLock.readLock().lock(); - - try - { - if (currentPage != null) - { - currentPage.write(message); - - if (sync) - { - currentPage.sync(); - } - return true; - } - else - { - return false; - } - } - finally - { - currentPageLock.readLock().unlock(); - } - } - finally - { - writeLock.unlock(); - } - + // If non Durable, there is no need to sync as there is no requirement for persistence for those messages in case + // of crash + return page(message, -1, syncNonTransactional && message.isDurable(), duplicateDetection); } - + public void sync() throws Exception { currentPageLock.readLock().lock(); @@ -685,43 +514,12 @@ } } - /** - * Depage one page-file, read it and send it to the pagingManager / postoffice - * @return - * @throws Exception - */ - public boolean readPage() throws Exception - { - Page page = depage(); - - if (page == null) - { - return false; - } - - page.open(); - - List<PagedMessage> messages = page.read(); - - if (onDepage(page.getPageId(), storeName, messages)) - { - page.delete(); - - return true; - } - else - { - return false; - } - - } - + public Page getCurrentPage() { return currentPage; } - public Page createPage(final int page) throws Exception { String fileName = createFileName(page); @@ -748,13 +546,12 @@ return new PageImpl(this.storeName, storageManager, fileFactory, file, page); } - + public ServerProducerCreditManager getProducerCreditManager() { return creditManager; } - // TestSupportPageStore ------------------------------------------ public void forceAnotherPage() throws Exception @@ -851,6 +648,242 @@ // Private ------------------------------------------------------- /** + * Depage one page-file, read it and send it to the pagingManager / postoffice + * @return + * @throws Exception + */ + private boolean readPage() throws Exception + { + Page page = depage(); + + if (page == null) + { + return false; + } + + page.open(); + + List<PagedMessage> messages = page.read(); + + if (onDepage(page.getPageId(), storeName, messages)) + { + page.delete(); + + return true; + } + else + { + return false; + } + + } + + + private synchronized void checkReleaseProducerFlowControlCredits(final long size) + { + if (addressFullMessagePolicy == AddressFullMessagePolicy.BLOCK && maxSize != -1) + { + long avail = availableProducerCredits.addAndGet(-size); + + if (avail > 0) + { + int used = creditManager.creditsReleased((int)avail); + + long num = availableProducerCredits.addAndGet(-used); + + if (num < 0) + { + log.warn("Available credits has gone negative"); + + exceededAvailableCredits = true; + } + } + } + } + + + private void addSize(final long size) throws Exception + { + if (addressFullMessagePolicy != AddressFullMessagePolicy.PAGE) + { + addAddressSize(size); + + pagingManager.addSize(size); + + return; + } + else + { + pagingManager.addSize(size); + + final long addressSize = addAddressSize(size); + + if (size > 0) + { + if (maxSize > 0 && addressSize > maxSize) + { + if (startPaging()) + { + if (isTrace) + { + trace("Starting paging on " + getStoreName() + ", size = " + addressSize + ", maxSize=" + maxSize); + } + } + } + } + else + { + // When in Global mode, we use the default page size as the mark to start depage + if (maxSize > 0 && currentPage != null && addressSize <= maxSize - pageSize && !depaging.get()) + { + if (startDepaging()) + { + if (isTrace) + { + trace("Starting depaging Thread, size = " + addressSize); + } + } + } + } + + return; + } + } + + private boolean page(final ServerMessage message, final long transactionID, final boolean sync, final boolean duplicateDetection) throws Exception + { + if (!running) + { + throw new IllegalStateException("PagingStore(" + getStoreName() + ") not initialized"); + } + + boolean full = isFull(); + + if (addressFullMessagePolicy == AddressFullMessagePolicy.DROP) + { + if (full) + { + if (!printedDropMessagesWarning) + { + printedDropMessagesWarning = true; + + log.warn("Messages are being dropped on address " + getStoreName()); + } + + // Address is full, we just pretend we are paging, and drop the data + return true; + } + else + { + return false; + } + } + else if (addressFullMessagePolicy == AddressFullMessagePolicy.BLOCK) + { + return false; + } + + // We need to ensure a read lock, as depage could change the paging state + currentPageLock.readLock().lock(); + + try + { + // First check done concurrently, to avoid synchronization and increase throughput + if (currentPage == null) + { + return false; + } + } + finally + { + currentPageLock.readLock().unlock(); + } + + writeLock.lock(); + + try + { + if (currentPage == null) + { + return false; + } + + if (duplicateDetection) + { + // We set the duplicate detection header to prevent the message being depaged more than once in case of + // failure during depage + + byte[] bytes = new byte[8]; + + ByteBuffer buff = ByteBuffer.wrap(bytes); + + buff.putLong(message.getMessageID()); + + message.putBytesProperty(MessageImpl.HDR_DUPLICATE_DETECTION_ID, bytes); + } + + int bytesToWrite = message.getEncodeSize() + PageImpl.SIZE_RECORD; + + if (currentPageSize.addAndGet(bytesToWrite) > pageSize && currentPage.getNumberOfMessages() > 0) + { + // Make sure nothing is currently validating or using currentPage + currentPageLock.writeLock().lock(); + try + { + openNewPage(); + + // openNewPage will set currentPageSize to zero, we need to set it again + currentPageSize.addAndGet(bytesToWrite); + } + finally + { + currentPageLock.writeLock().unlock(); + } + } + + currentPageLock.readLock().lock(); + + try + { + if (currentPage != null) + { + PagedMessage pagedMessage; + + if (transactionID != -1) + { + pagedMessage = new PagedMessageImpl(message, transactionID); + } + else + { + pagedMessage = new PagedMessageImpl(message); + } + + currentPage.write(pagedMessage); + + if (sync) + { + currentPage.sync(); + } + return true; + } + else + { + return false; + } + } + finally + { + currentPageLock.readLock().unlock(); + } + } + finally + { + writeLock.unlock(); + } + + } + + /** * This method will remove files from the page system and and route them, doing it transactionally * * If persistent messages are also used, it will update eventual PageTransactions @@ -868,13 +901,12 @@ // nothing to be done on this case. return true; } - // Depage has to be done atomically, in case of failure it should be // back to where it was Transaction depageTransaction = new TransactionImpl(storageManager); - + depageTransaction.putProperty(TransactionPropertyIndexes.IS_DEPAGE, Boolean.valueOf(true)); HashSet<PageTransactionInfo> pageTransactionsToUpdate = new HashSet<PageTransactionInfo>(); @@ -882,13 +914,14 @@ for (PagedMessage pagedMessage : pagedMessages) { ServerMessage message = pagedMessage.getMessage(storageManager); - + if (message.isLargeMessage()) { LargeServerMessage largeMsg = (LargeServerMessage)message; if (!largeMsg.isFileExists()) { - log.warn("File for large message " + largeMsg.getMessageID() + " doesn't exist, so ignoring depage for this large message"); + log.warn("File for large message " + largeMsg.getMessageID() + + " doesn't exist, so ignoring depage for this large message"); continue; } } @@ -931,7 +964,7 @@ if (isTrace) { trace("Rollback was called after prepare, ignoring message " + message); - } + } continue; } @@ -970,8 +1003,8 @@ } depageTransaction.commit(); - - // StorageManager does the check: if (replicated) -> do the proper cleanup already + + // StorageManager does the check: if (replicated) -> do the proper cleanup already storageManager.completeReplication(); if (isTrace) @@ -987,7 +1020,7 @@ */ private boolean isAddressFull(final long nextPageSize) { - return getMaxSizeBytes() > 0 && getAddressSize() + nextPageSize > getMaxSizeBytes(); + return maxSize > 0 && getAddressSize() + nextPageSize > maxSize; } private long addAddressSize(final long delta) @@ -1012,7 +1045,7 @@ " addressSize = " + this.getAddressSize() + " addressMax " + - this.getMaxSizeBytes() + + maxSize + " isPaging = " + isPaging() + " addressFull = " + @@ -1050,9 +1083,8 @@ { currentPage.close(); } - + currentPage = createPage(currentPageId); - currentPageSize.set(0); @@ -1084,7 +1116,7 @@ // To be used on isDropMessagesWhenFull private boolean isFull() { - return getMaxSizeBytes() > 0 && getAddressSize() > getMaxSizeBytes(); + return maxSize > 0 && getAddressSize() > maxSize; } // Inner classes ------------------------------------------------- Modified: trunk/src/main/org/hornetq/core/paging/impl/TestSupportPageStore.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/impl/TestSupportPageStore.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/paging/impl/TestSupportPageStore.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -37,4 +37,9 @@ void forceAnotherPage() throws Exception; boolean isExceededAvailableCredits(); + + /** @return true if paging was started, or false if paging was already started before this call */ + boolean startPaging() throws Exception; + + Page getCurrentPage(); } Modified: trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -20,6 +20,7 @@ import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -598,18 +599,16 @@ if (context.getTransaction() == null) { - if (pagingManager.page(message, true)) + if (message.page(true)) { return; } } else { - SimpleString destination = message.getDestination(); - boolean depage = context.getTransaction().getProperty(TransactionPropertyIndexes.IS_DEPAGE) != null; - if (!depage && pagingManager.isPaging(destination)) + if (!depage && message.storeIsPaging()) { getPageOperation(context.getTransaction()).addMessageToPage(message); @@ -1149,20 +1148,20 @@ boolean pagingPersistent = false; - HashSet<SimpleString> pagedDestinationsToSync = new HashSet<SimpleString>(); + Set<PagingStore> pagingStoresToSync = new HashSet<PagingStore>(); // We only need to add the dupl id header once per transaction boolean first = true; for (ServerMessage message : messagesToPage) { - if (pagingManager.page(message, tx.getID(), first)) + if (message.page(tx.getID(), first)) { if (message.isDurable()) { // We only create pageTransactions if using persistent messages pageTransaction.increment(); pagingPersistent = true; - pagedDestinationsToSync.add(message.getDestination()); + pagingStoresToSync.add(message.getPagingStore()); } } else @@ -1179,9 +1178,13 @@ { tx.putProperty(TransactionPropertyIndexes.CONTAINS_PERSISTENT, true); - if (!pagedDestinationsToSync.isEmpty()) + if (!pagingStoresToSync.isEmpty()) { - pagingManager.sync(pagedDestinationsToSync); + for (PagingStore store: pagingStoresToSync) + { + store.sync(); + } + storageManager.storePageTransaction(tx.getID(), pageTransaction); } } Modified: trunk/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -200,10 +200,10 @@ journalLoadInformation = storage.loadInternalOnly(); pageManager = new PagingManagerImpl(new PagingStoreFactoryNIO(config.getPagingDirectory(), - server.getExecutorFactory()), + server.getExecutorFactory(), + config.isJournalSyncNonTransactional()), storage, - server.getAddressSettingsRepository(), - false); + server.getAddressSettingsRepository()); pageManager.start(); Modified: trunk/src/main/org/hornetq/core/server/ServerMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/server/ServerMessage.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/server/ServerMessage.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -52,4 +52,14 @@ ServerMessage makeCopyForExpiryOrDLA(long newID, boolean expiry) throws Exception; void setOriginalHeaders(ServerMessage other, boolean expiry); + + void setPagingStore(PagingStore store); + + PagingStore getPagingStore(); + + boolean page(boolean duplicateDetection) throws Exception; + + boolean page(long transactionID, boolean duplicateDetection) throws Exception; + + boolean storeIsPaging(); } Modified: trunk/src/main/org/hornetq/core/server/impl/DeliveryImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/DeliveryImpl.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/server/impl/DeliveryImpl.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -9,7 +9,7 @@ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or * implied. See the License for the specific language governing * permissions and limitations under the License. - */ + */ package org.hornetq.core.server.impl; @@ -24,25 +24,24 @@ * */ public class DeliveryImpl implements Delivery -{ +{ private final long consumerID; - + private final MessageReference reference; - - public DeliveryImpl(final long consumerID, - final MessageReference reference) - { + + public DeliveryImpl(final long consumerID, final MessageReference reference) + { this.consumerID = consumerID; - this.reference = reference; + this.reference = reference; } public long getConsumerID() { return consumerID; } - + public MessageReference getReference() { return reference; - } + } } Modified: trunk/src/main/org/hornetq/core/server/impl/DistributorImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/DistributorImpl.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/server/impl/DistributorImpl.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -28,12 +28,12 @@ protected final List<Consumer> consumers = new ArrayList<Consumer>(); - public void addConsumer(Consumer consumer) + public void addConsumer(final Consumer consumer) { consumers.add(consumer); } - public boolean removeConsumer(Consumer consumer) + public boolean removeConsumer(final Consumer consumer) { return consumers.remove(consumer); } Modified: trunk/src/main/org/hornetq/core/server/impl/DivertImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/DivertImpl.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/server/impl/DivertImpl.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -50,7 +50,7 @@ private final Filter filter; private final Transformer transformer; - + public DivertImpl(final SimpleString forwardAddress, final SimpleString uniqueName, final SimpleString routingName, @@ -74,19 +74,19 @@ this.postOffice = postOffice; } - public void route(ServerMessage message, final RoutingContext context) throws Exception - { + public void route(final ServerMessage message, final RoutingContext context) throws Exception + { SimpleString originalDestination = message.getDestination(); - - //We must make a copy of the message, otherwise things like returning credits to the page won't work - //properly on ack, since the original destination will be overwritten - - //TODO we can optimise this so it doesn't copy if it's not routed anywhere else - + + // We must make a copy of the message, otherwise things like returning credits to the page won't work + // properly on ack, since the original destination will be overwritten + + // TODO we can optimise this so it doesn't copy if it's not routed anywhere else + ServerMessage copy = message.copy(); - + copy.setDestination(forwardAddress); - + copy.putStringProperty(HDR_ORIGINAL_DESTINATION, originalDestination); if (transformer != null) @@ -94,7 +94,7 @@ copy = transformer.transform(copy); } - postOffice.route(copy, context.getTransaction()); + postOffice.route(copy, context.getTransaction()); } public SimpleString getRoutingName() @@ -111,7 +111,7 @@ { return exclusive; } - + public Filter getFilter() { return filter; Modified: trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -60,9 +60,9 @@ import org.hornetq.core.paging.PagingManager; import org.hornetq.core.paging.impl.PagingManagerImpl; import org.hornetq.core.paging.impl.PagingStoreFactoryNIO; +import org.hornetq.core.persistence.GroupingInfo; import org.hornetq.core.persistence.QueueBindingInfo; import org.hornetq.core.persistence.StorageManager; -import org.hornetq.core.persistence.GroupingInfo; import org.hornetq.core.persistence.impl.journal.JournalStorageManager; import org.hornetq.core.persistence.impl.nullpm.NullStorageManager; import org.hornetq.core.postoffice.Binding; @@ -93,14 +93,14 @@ import org.hornetq.core.server.Queue; import org.hornetq.core.server.QueueFactory; import org.hornetq.core.server.ServerSession; +import org.hornetq.core.server.cluster.ClusterManager; +import org.hornetq.core.server.cluster.Transformer; +import org.hornetq.core.server.cluster.impl.ClusterManagerImpl; import org.hornetq.core.server.group.GroupingHandler; +import org.hornetq.core.server.group.impl.GroupBinding; import org.hornetq.core.server.group.impl.GroupingHandlerConfiguration; import org.hornetq.core.server.group.impl.LocalGroupingHandler; import org.hornetq.core.server.group.impl.RemoteGroupingHandler; -import org.hornetq.core.server.group.impl.GroupBinding; -import org.hornetq.core.server.cluster.ClusterManager; -import org.hornetq.core.server.cluster.Transformer; -import org.hornetq.core.server.cluster.impl.ClusterManagerImpl; import org.hornetq.core.settings.HierarchicalRepository; import org.hornetq.core.settings.impl.AddressSettings; import org.hornetq.core.settings.impl.HierarchicalObjectRepository; @@ -221,7 +221,7 @@ this(configuration, null, null); } - public HornetQServerImpl(final Configuration configuration, MBeanServer mbeanServer) + public HornetQServerImpl(final Configuration configuration, final MBeanServer mbeanServer) { this(configuration, mbeanServer, null); } @@ -256,7 +256,7 @@ this.securityManager = securityManager; - this.addressSettingsRepository = new HierarchicalObjectRepository<AddressSettings>(); + addressSettingsRepository = new HierarchicalObjectRepository<AddressSettings>(); addressSettingsRepository.setDefault(new AddressSettings()); } @@ -289,8 +289,8 @@ { if (!configuration.isSharedStore()) { - this.replicationEndpoint = new ReplicationEndpointImpl(this); - this.replicationEndpoint.start(); + replicationEndpoint = new ReplicationEndpointImpl(this); + replicationEndpoint.start(); } // We defer actually initialisation until the live node has contacted the backup log.info("Backup server initialised"); @@ -529,8 +529,8 @@ } public ReattachSessionResponseMessage reattachSession(final RemotingConnection connection, - final String name, - final int lastReceivedCommandID) throws Exception + final String name, + final int lastReceivedCommandID) throws Exception { if (!started) { @@ -580,17 +580,17 @@ } public CreateSessionResponseMessage createSession(final String name, - final long channelID, - final String username, - final String password, - final int minLargeMessageSize, - final int incrementingVersion, - final RemotingConnection connection, - final boolean autoCommitSends, - final boolean autoCommitAcks, - final boolean preAcknowledge, - final boolean xa, - final int sendWindowSize) throws Exception + final long channelID, + final String username, + final String password, + final int minLargeMessageSize, + final int incrementingVersion, + final RemotingConnection connection, + final boolean autoCommitSends, + final boolean autoCommitAcks, + final boolean preAcknowledge, + final boolean xa, + final int sendWindowSize) throws Exception { if (!started) { @@ -607,14 +607,16 @@ ". " + "Please ensure all clients and servers are upgraded to the same version for them to " + "interoperate properly"); - throw new HornetQException(HornetQException.INCOMPATIBLE_CLIENT_SERVER_VERSIONS, "Server and client versions incompatible"); + throw new HornetQException(HornetQException.INCOMPATIBLE_CLIENT_SERVER_VERSIONS, + "Server and client versions incompatible"); } if (!checkActivate()) { // Backup server is not ready to accept connections - throw new HornetQException(HornetQException.SESSION_CREATION_REJECTED, "Server will not accept create session requests"); + throw new HornetQException(HornetQException.SESSION_CREATION_REJECTED, + "Server will not accept create session requests"); } if (securityStore != null) @@ -674,11 +676,11 @@ if (replicationEndpoint.getChannel() != null) { - throw new HornetQException(HornetQException.ILLEGAL_STATE, "Backup replication server is already connected to another server"); + throw new HornetQException(HornetQException.ILLEGAL_STATE, + "Backup replication server is already connected to another server"); } - + replicationEndpoint.setChannel(channel); - return replicationEndpoint; } @@ -713,7 +715,7 @@ return new HashSet<ServerSession>(sessions.values()); } - //TODO - should this really be here?? It's only used in tests + // TODO - should this really be here?? It's only used in tests public boolean isInitialised() { synchronized (initialiseLock) @@ -748,19 +750,19 @@ } public Queue createQueue(final SimpleString address, - final SimpleString queueName, - final SimpleString filterString, - final boolean durable, - final boolean temporary) throws Exception + final SimpleString queueName, + final SimpleString filterString, + final boolean durable, + final boolean temporary) throws Exception { return createQueue(address, queueName, filterString, durable, temporary, false); } public Queue deployQueue(final SimpleString address, - final SimpleString queueName, - final SimpleString filterString, - final boolean durable, - final boolean temporary) throws Exception + final SimpleString queueName, + final SimpleString filterString, + final boolean durable, + final boolean temporary) throws Exception { return createQueue(address, queueName, filterString, durable, temporary, true); } @@ -864,10 +866,11 @@ protected PagingManager createPagingManager() { - return new PagingManagerImpl(new PagingStoreFactoryNIO(configuration.getPagingDirectory(), executorFactory), + return new PagingManagerImpl(new PagingStoreFactoryNIO(configuration.getPagingDirectory(), + executorFactory, + configuration.isJournalSyncNonTransactional()), storageManager, - addressSettingsRepository, - configuration.isJournalSyncNonTransactional()); + addressSettingsRepository); } /** for use on sub-classes */ @@ -912,7 +915,8 @@ replicationFailoverManager = createBackupConnection(backupConnector, threadPool, scheduledPool); - this.replicationManager = new ReplicationManagerImpl(replicationFailoverManager, configuration.getBackupWindowSize()); + replicationManager = new ReplicationManagerImpl(replicationFailoverManager, + configuration.getBackupWindowSize()); replicationManager.start(); } } @@ -1008,7 +1012,7 @@ startReplication(); - this.storageManager = createStorageManager(); + storageManager = createStorageManager(); securityRepository = new HierarchicalObjectRepository<Set<Role>>(); securityRepository.setDefault(new HashSet<Role>()); @@ -1093,13 +1097,12 @@ } } deployGroupingHandlerConfiguration(configuration.getGroupingHandlerConfiguration()); - + // Load the journal and populate queues, transactions and caches in memory JournalLoadInformation[] journalInfo = loadJournals(); - + compareJournals(journalInfo); - // Deploy any queues in the Configuration class - if there's no file deployment we still need // to load those deployQueuesFromConfiguration(); @@ -1163,7 +1166,7 @@ /** * @param journalInfo */ - private void compareJournals(JournalLoadInformation[] journalInfo) throws Exception + private void compareJournals(final JournalLoadInformation[] journalInfo) throws Exception { if (replicationManager != null) { @@ -1185,7 +1188,7 @@ private JournalLoadInformation[] loadJournals() throws Exception { JournalLoadInformation[] journalInfo = new JournalLoadInformation[2]; - + List<QueueBindingInfo> queueBindingInfos = new ArrayList<QueueBindingInfo>(); List<GroupingInfo> groupingInfos = new ArrayList<GroupingInfo>(); @@ -1230,7 +1233,11 @@ Map<SimpleString, List<Pair<byte[], Long>>> duplicateIDMap = new HashMap<SimpleString, List<Pair<byte[], Long>>>(); - journalInfo[1] = storageManager.loadMessageJournal(postOffice, pagingManager, resourceManager, queues, duplicateIDMap); + journalInfo[1] = storageManager.loadMessageJournal(postOffice, + pagingManager, + resourceManager, + queues, + duplicateIDMap); for (Map.Entry<SimpleString, List<Pair<byte[], Long>>> entry : duplicateIDMap.entrySet()) { @@ -1243,7 +1250,7 @@ cache.load(entry.getValue()); } } - + return journalInfo; } @@ -1405,10 +1412,10 @@ config.getName(), config.getAddress(), config.getTimeout()); - } - + } + this.groupingHandler = groupingHandler; - + managementService.addNotificationListener(groupingHandler); } } Modified: trunk/src/main/org/hornetq/core/server/impl/LastValueQueue.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/LastValueQueue.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/server/impl/LastValueQueue.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -67,6 +67,7 @@ addressSettingsRepository); } + @Override public synchronized void add(final MessageReference ref, final boolean first) { SimpleString prop = (SimpleString)ref.getMessage().getProperty(MessageImpl.HDR_LAST_VALUE_NAME); @@ -76,7 +77,7 @@ HolderReference hr = map.get(prop); if (!first) - { + { if (hr != null) { // We need to overwrite the old ref with the new one and ack the old one @@ -169,7 +170,7 @@ this.ref = ref; } - public MessageReference copy(Queue queue) + public MessageReference copy(final Queue queue) { return ref.copy(queue); } @@ -209,12 +210,12 @@ ref.incrementDeliveryCount(); } - public void setDeliveryCount(int deliveryCount) + public void setDeliveryCount(final int deliveryCount) { ref.setDeliveryCount(deliveryCount); } - public void setScheduledDeliveryTime(long scheduledDeliveryTime) + public void setScheduledDeliveryTime(final long scheduledDeliveryTime) { ref.setScheduledDeliveryTime(scheduledDeliveryTime); } Modified: trunk/src/main/org/hornetq/core/server/impl/MemoryManagerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/MemoryManagerImpl.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/server/impl/MemoryManagerImpl.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -28,79 +28,80 @@ public class MemoryManagerImpl implements MemoryManager { private static final Logger log = Logger.getLogger(MemoryManagerImpl.class); - - private Runtime runtime; - - private long measureInterval; - - private int memoryWarningThreshold; - + + private final Runtime runtime; + + private final long measureInterval; + + private final int memoryWarningThreshold; + private volatile boolean started; - + private Thread thread; - + private volatile boolean low; - - public MemoryManagerImpl(int memoryWarningThreshold, long measureInterval) + + public MemoryManagerImpl(final int memoryWarningThreshold, final long measureInterval) { runtime = Runtime.getRuntime(); - + this.measureInterval = measureInterval; - - this.memoryWarningThreshold = memoryWarningThreshold; + + this.memoryWarningThreshold = memoryWarningThreshold; } - + public boolean isMemoryLow() { return low; } - + public synchronized boolean isStarted() { return started; } - + public synchronized void start() { - log.debug("Starting MemoryManager with MEASURE_INTERVAL: " + measureInterval - + " FREE_MEMORY_PERCENT: " + memoryWarningThreshold); - + log.debug("Starting MemoryManager with MEASURE_INTERVAL: " + measureInterval + + " FREE_MEMORY_PERCENT: " + + memoryWarningThreshold); + if (started) { - //Already started + // Already started return; } - + started = true; - + thread = new Thread(new MemoryRunnable()); - + thread.setDaemon(true); - + thread.start(); } - + public synchronized void stop() - { + { if (!started) { - //Already stopped + // Already stopped return; } - + started = false; - + thread.interrupt(); - + try { thread.join(); } catch (InterruptedException ignore) - { + { } } - + private class MemoryRunnable implements Runnable { public void run() @@ -113,7 +114,7 @@ { break; } - + Thread.sleep(measureInterval); } catch (InterruptedException ignore) @@ -123,17 +124,17 @@ break; } } - + long maxMemory = runtime.maxMemory(); - + long totalMemory = runtime.totalMemory(); - + long freeMemory = runtime.freeMemory(); - - long availableMemory = freeMemory + (maxMemory - totalMemory); - - double availableMemoryPercent = 100.0 * (double)availableMemory / maxMemory; - + + long availableMemory = freeMemory + maxMemory - totalMemory; + + double availableMemoryPercent = 100.0 * availableMemory / maxMemory; + String info = ""; info += String.format("free memory: %s\n", SizeFormatterUtil.sizeof(freeMemory)); info += String.format("max memory: %s\n", SizeFormatterUtil.sizeof(maxMemory)); @@ -144,21 +145,22 @@ { log.debug(info); } - + if (availableMemoryPercent <= memoryWarningThreshold) { - log.warn("Less than " + memoryWarningThreshold + "%\n" - + info + + log.warn("Less than " + memoryWarningThreshold + + "%\n" + + info + "\nYou are in danger of running out of RAM. Have you set paging parameters " + "on your addresses? (See user manual \"Paging\" chapter)"); - + low = true; } else { low = false; } - + } } } Modified: trunk/src/main/org/hornetq/core/server/impl/MessageReferenceImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/MessageReferenceImpl.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/server/impl/MessageReferenceImpl.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -47,11 +47,11 @@ public MessageReferenceImpl(final MessageReferenceImpl other, final Queue queue) { - this.deliveryCount = other.deliveryCount; + deliveryCount = other.deliveryCount; - this.scheduledDeliveryTime = other.scheduledDeliveryTime; + scheduledDeliveryTime = other.scheduledDeliveryTime; - this.message = other.message; + message = other.message; this.queue = queue; } @@ -94,7 +94,7 @@ { deliveryCount++; } - + public void decrementDeliveryCount() { deliveryCount--; @@ -119,7 +119,7 @@ { return queue; } - + public void handled() { queue.referenceHandled(); @@ -127,6 +127,7 @@ // Public -------------------------------------------------------- + @Override public String toString() { return "Reference[" + getMessage().getMessageID() + Modified: trunk/src/main/org/hornetq/core/server/impl/QueueFactoryImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/QueueFactoryImpl.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/server/impl/QueueFactoryImpl.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -72,15 +72,15 @@ if (addressSettings.isLastValueQueue()) { queue = new LastValueQueue(persistenceID, - address, - name, - filter, - durable, - temporary, - scheduledExecutor, - postOffice, - storageManager, - addressSettingsRepository); + address, + name, + filter, + durable, + temporary, + scheduledExecutor, + postOffice, + storageManager, + addressSettingsRepository); } else { Modified: trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -123,7 +123,7 @@ private final ScheduledExecutorService scheduledExecutor; - private SimpleString address; + private final SimpleString address; private Redistributor redistributor; @@ -137,7 +137,7 @@ private final Map<Consumer, Iterator<MessageReference>> iterators = new HashMap<Consumer, Iterator<MessageReference>>(); - private ConcurrentMap<SimpleString, Consumer> groups = new ConcurrentHashMap<SimpleString, Consumer>(); + private final ConcurrentMap<SimpleString, Consumer> groups = new ConcurrentHashMap<SimpleString, Consumer>(); private volatile SimpleString expiryAddress; @@ -877,7 +877,7 @@ @Override public String toString() { - return "QueueImpl(name=" + this.name.toString() + ")"; + return "QueueImpl(name=" + name.toString() + ")"; } // Private @@ -1402,7 +1402,8 @@ // ack isn't committed, then the server crashes and on // recovery the message is deleted even though the other ack never committed - //also note then when this happens as part of a trasaction its the tx commt of the ack that is important not this + // also note then when this happens as part of a trasaction its the tx commt of the ack that is important + // not this try { storageManager.deleteMessage(message.getMessageID()); @@ -1434,7 +1435,7 @@ } - void postRollback(LinkedList<MessageReference> refs) throws Exception + void postRollback(final LinkedList<MessageReference> refs) throws Exception { synchronized (this) { @@ -1449,7 +1450,7 @@ } } - private synchronized void initPagingStore(SimpleString destination) + private synchronized void initPagingStore(final SimpleString destination) { // PagingManager would be null only on testcases if (pagingStore == null && pagingManager != null) @@ -1496,14 +1497,14 @@ // Must be set to false *before* executing to avoid race waitingToDeliver.set(false); - QueueImpl.this.lockDelivery(); + lockDelivery(); try { deliver(); } finally { - QueueImpl.this.unlockDelivery(); + unlockDelivery(); } } } Modified: trunk/src/main/org/hornetq/core/server/impl/RoundRobinDistributor.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/RoundRobinDistributor.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/server/impl/RoundRobinDistributor.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -43,6 +43,7 @@ return super.removeConsumer(consumer); } + @Override public synchronized int getConsumerCount() { return super.getConsumerCount(); @@ -54,11 +55,11 @@ incrementPosition(); return consumer; } - + private synchronized void incrementPosition() { pos++; - + if (pos == consumers.size()) { pos = 0; Modified: trunk/src/main/org/hornetq/core/server/impl/RoutingContextImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/RoutingContextImpl.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/server/impl/RoutingContextImpl.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -29,19 +29,19 @@ */ public class RoutingContextImpl implements RoutingContext { - private List<Queue> queues = new ArrayList<Queue>(); - + private final List<Queue> queues = new ArrayList<Queue>(); + private Transaction transaction; - + private int depth; - + public RoutingContextImpl(final Transaction transaction) { this.transaction = transaction; } public void addQueue(final Queue queue) - { + { queues.add(queue); } @@ -49,12 +49,12 @@ { return transaction; } - + public void setTransaction(final Transaction tx) { - this.transaction = tx; + transaction = tx; } - + public List<Queue> getQueues() { return queues; Modified: trunk/src/main/org/hornetq/core/server/impl/ScheduledDeliveryHandlerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ScheduledDeliveryHandlerImpl.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/server/impl/ScheduledDeliveryHandlerImpl.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -41,7 +41,7 @@ private final ScheduledExecutorService scheduledExecutor; private final Map<Long, ScheduledDeliveryRunnable> scheduledRunnables = new LinkedHashMap<Long, ScheduledDeliveryRunnable>(); - + private boolean rescheduled; public ScheduledDeliveryHandlerImpl(final ScheduledExecutorService scheduledExecutor) @@ -67,13 +67,13 @@ scheduledRunnables.put(ref.getMessage().getMessageID(), runnable); } - scheduleDelivery(runnable, deliveryTime); + scheduleDelivery(runnable, deliveryTime); return true; } return false; } - + public void reSchedule() { synchronized (scheduledRunnables) @@ -84,7 +84,7 @@ { scheduleDelivery(runnable, runnable.getReference().getScheduledDeliveryTime()); } - + rescheduled = true; } } @@ -98,7 +98,7 @@ public List<MessageReference> getScheduledReferences() { List<MessageReference> refs = new ArrayList<MessageReference>(); - + synchronized (scheduledRunnables) { for (ScheduledDeliveryRunnable scheduledRunnable : scheduledRunnables.values()) @@ -112,13 +112,13 @@ public List<MessageReference> cancel() { List<MessageReference> refs = new ArrayList<MessageReference>(); - + synchronized (scheduledRunnables) { for (ScheduledDeliveryRunnable runnable : scheduledRunnables.values()) { runnable.cancel(); - + refs.add(runnable.getReference()); } @@ -126,8 +126,8 @@ } return refs; } - - public MessageReference removeReferenceWithID(long id) + + public MessageReference removeReferenceWithID(final long id) { synchronized (scheduledRunnables) { Modified: trunk/src/main/org/hornetq/core/server/impl/ServerInfo.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerInfo.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/server/impl/ServerInfo.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -34,7 +34,7 @@ { private final HornetQServer server; - private PagingManager pagingManager; + private final PagingManager pagingManager; // Constants ----------------------------------------------------- @@ -57,8 +57,8 @@ long maxMemory = Runtime.getRuntime().maxMemory(); long totalMemory = Runtime.getRuntime().totalMemory(); long freeMemory = Runtime.getRuntime().freeMemory(); - long availableMemory = freeMemory + (maxMemory - totalMemory); - double availableMemoryPercent = 100.0 * (double)availableMemory / maxMemory; + long availableMemory = freeMemory + maxMemory - totalMemory; + double availableMemoryPercent = 100.0 * availableMemory / maxMemory; ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean(); String info = "\n**** Server Dump ****\n"; @@ -90,7 +90,9 @@ try { pageStore = pagingManager.getPageStore(storeName); - info += String.format("\t%s: %s\n", storeName, SizeFormatterUtil.sizeof(pageStore.getPageSizeBytes() * pageStore.getNumberOfPages())); + info += String.format("\t%s: %s\n", + storeName, + SizeFormatterUtil.sizeof(pageStore.getPageSizeBytes() * pageStore.getNumberOfPages())); } catch (Exception e) { Modified: trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -45,6 +45,8 @@ // We cache this private volatile int memoryEstimate = -1; + private PagingStore pagingStore; + /* * Constructor for when reading from network */ @@ -88,7 +90,7 @@ messageID = id; } - public void setType(byte type) + public void setType(final byte type) { this.type = type; } @@ -103,34 +105,34 @@ public int incrementRefCount(final PagingStore pagingStore, final MessageReference reference) throws Exception { int count = refCount.incrementAndGet(); - + if (pagingStore != null) { if (count == 1) { pagingStore.addSize(this, true); } - + pagingStore.addSize(reference, true); } - + return count; } - + public int decrementRefCount(final PagingStore pagingStore, final MessageReference reference) throws Exception { int count = refCount.decrementAndGet(); - + if (pagingStore != null) { if (count == 0) { pagingStore.addSize(this, false); } - + pagingStore.addSize(reference, false); } - + return count; } @@ -144,8 +146,6 @@ return durableRefCount.decrementAndGet(); } - - public int getRefCount() { return refCount.get(); @@ -158,7 +158,7 @@ public long getLargeBodySize() { - return (long)getBodySize(); + return getBodySize(); } public int getMemoryEstimate() @@ -200,18 +200,18 @@ */ ServerMessage copy = copy(newID); - + copy.setOriginalHeaders(this, expiry); return copy; } - + public void setOriginalHeaders(final ServerMessage other, final boolean expiry) { if (other.getProperty(HDR_ORIG_MESSAGE_ID) != null) { putStringProperty(HDR_ORIGINAL_DESTINATION, (SimpleString)other.getProperty(HDR_ORIGINAL_DESTINATION)); - + putLongProperty(HDR_ORIG_MESSAGE_ID, (Long)other.getProperty(HDR_ORIG_MESSAGE_ID)); } else @@ -222,18 +222,68 @@ putLongProperty(HDR_ORIG_MESSAGE_ID, other.getMessageID()); } - + // reset expiry setExpiration(0); - + if (expiry) - { + { long actualExpiryTime = System.currentTimeMillis(); putLongProperty(HDR_ACTUAL_EXPIRY_TIME, actualExpiryTime); } } + public void setPagingStore(final PagingStore pagingStore) + { + this.pagingStore = pagingStore; + + // On the server side, we reset the address to point to the instance of address in the paging store + // Otherwise each message would have its own copy of the address String which would take up more memory + destination = pagingStore.getAddress(); + } + + public PagingStore getPagingStore() + { + return pagingStore; + } + + public boolean page(final boolean duplicateDetection) throws Exception + { + if (pagingStore != null) + { + return pagingStore.page(this, duplicateDetection); + } + else + { + return false; + } + } + + public boolean page(final long transactionID, final boolean duplicateDetection) throws Exception + { + if (pagingStore != null) + { + return pagingStore.page(this, transactionID, duplicateDetection); + } + else + { + return false; + } + } + + public boolean storeIsPaging() + { + if (pagingStore != null) + { + return pagingStore.isPaging(); + } + else + { + return false; + } + } + @Override public String toString() { Modified: trunk/src/main/org/hornetq/core/server/impl/ServerProducerCreditManager.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerProducerCreditManager.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/server/impl/ServerProducerCreditManager.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -15,7 +15,6 @@ import org.hornetq.core.paging.PagingStore; - /** * A ServerProducerCreditManager * @@ -26,10 +25,10 @@ public interface ServerProducerCreditManager { int creditsReleased(int credits); - + int acquireCredits(int credits, CreditsAvailableRunnable runnable); - + int waitingEntries(); - + PagingStore getStore(); } Modified: trunk/src/main/org/hornetq/core/server/impl/ServerProducerCreditManagerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerProducerCreditManagerImpl.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/server/impl/ServerProducerCreditManagerImpl.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -91,7 +91,7 @@ entry.credits -= credits; boolean sent = sendCredits(entry.waiting, credits); - + if (sent) { credits = 0; Modified: trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -168,6 +168,8 @@ private boolean closed; + private final Map<SimpleString, CreditManagerHolder> creditManagerHolders = new HashMap<SimpleString, CreditManagerHolder>(); + // Constructors --------------------------------------------------------------------------------- public ServerSessionImpl(final String name, @@ -190,7 +192,7 @@ final HornetQServer server, final SimpleString managementAddress) throws Exception { - this.id = channel.getID(); + id = channel.getID(); this.username = username; @@ -332,7 +334,7 @@ } } - public void promptDelivery(final Queue queue, boolean async) + public void promptDelivery(final Queue queue, final boolean async) { if (async) { @@ -368,17 +370,17 @@ Filter filter = FilterImpl.createFilter(filterString);; ServerConsumer consumer = new ServerConsumerImpl(packet.getID(), - this, - (QueueBinding)binding, - filter, - started, - browseOnly, - storageManager, - channel, - preAcknowledge, - updateDeliveries, - executor, - managementService); + this, + (QueueBinding)binding, + filter, + started, + browseOnly, + storageManager, + channel, + preAcknowledge, + updateDeliveries, + executor, + managementService); consumers.put(consumer.getID(), consumer); @@ -635,7 +637,7 @@ sendResponse(packet, response, false, false); } - public void handleForceConsumerDelivery(SessionForceConsumerDelivery message) + public void handleForceConsumerDelivery(final SessionForceConsumerDelivery message) { try { @@ -1401,7 +1403,7 @@ } catch (Exception e) { - log.error("Failed to receive credits " + this.server.getConfiguration().isBackup(), e); + log.error("Failed to receive credits " + server.getConfiguration().isBackup(), e); } sendResponse(packet, null, false, false); @@ -1412,13 +1414,10 @@ // need to create the LargeMessage before continue long id = storageManager.generateUniqueID(); - final LargeServerMessage msg = doCreateLargeMessage(id, packet); + LargeServerMessage msg = doCreateLargeMessage(id, packet); if (msg != null) { - // With a send we must make sure it is replicated to backup before being processed on live - // or can end up with delivery being processed on backup before original send - if (currentLargeMessage != null) { log.warn("Replacing incomplete LargeMessage with ID=" + currentLargeMessage.getMessageID()); @@ -1438,6 +1437,8 @@ try { + setPagingStore(message); + long id = storageManager.generateUniqueID(); message.setMessageID(id); @@ -1499,22 +1500,22 @@ { throw new HornetQException(HornetQException.ILLEGAL_STATE, "large-message not initialized on server"); } - - //Immediately release the credits for the continuations- these don't contrinute to the in-memory size - //of the message - + + // Immediately release the credits for the continuations- these don't contrinute to the in-memory size + // of the message + releaseOutStanding(currentLargeMessage, packet.getRequiredBufferSize()); - + currentLargeMessage.addBytes(packet.getBody()); if (!packet.isContinues()) - { + { currentLargeMessage.releaseResources(); send(currentLargeMessage); releaseOutStanding(currentLargeMessage, currentLargeMessage.getEncodeSize()); - + currentLargeMessage = null; } @@ -1543,40 +1544,6 @@ sendResponse(packet, response, false, false); } - private static final class CreditManagerHolder - { - CreditManagerHolder(final PagingStore store) - { - this.store = store; - - this.manager = store.getProducerCreditManager(); - } - - final PagingStore store; - - final ServerProducerCreditManager manager; - - volatile int outstandingCredits; - } - - private Map<SimpleString, CreditManagerHolder> creditManagerHolders = new HashMap<SimpleString, CreditManagerHolder>(); - - private CreditManagerHolder getCreditManagerHolder(final SimpleString address) throws Exception - { - CreditManagerHolder holder = creditManagerHolders.get(address); - - if (holder == null) - { - PagingStore store = postOffice.getPagingManager().getPageStore(address); - - holder = new CreditManagerHolder(store); - - creditManagerHolders.put(address, holder); - } - - return holder; - } - public void handleRequestProducerCredits(final SessionRequestProducerCreditsMessage packet) throws Exception { final SimpleString address = packet.getAddress(); @@ -1587,7 +1554,7 @@ int gotCredits = holder.manager.acquireCredits(credits, new CreditsAvailableRunnable() { - public boolean run(int credits) + public boolean run(final int credits) { synchronized (ServerSessionImpl.this) { @@ -1613,22 +1580,13 @@ sendResponse(packet, null, false, false); } - private void sendProducerCredits(final CreditManagerHolder holder, final int credits, final SimpleString address) - { - holder.outstandingCredits += credits; - - Packet packet = new SessionProducerCreditsMessage(credits, address); - - channel.send(packet); - } - public int transferConnection(final RemotingConnection newConnection, final int lastReceivedCommandID) { - boolean wasStarted = this.started; + boolean wasStarted = started; if (wasStarted) { - this.setStarted(false); + setStarted(false); } remotingConnection.removeFailureListener(this); @@ -1652,11 +1610,11 @@ int serverLastReceivedCommandID = channel.getLastReceivedCommandID(); - channel.replayCommands(lastReceivedCommandID, this.id); + channel.replayCommands(lastReceivedCommandID, id); if (wasStarted) { - this.setStarted(true); + setStarted(true); } return serverLastReceivedCommandID; @@ -1804,11 +1762,15 @@ * @param packet * @throws Exception */ - private LargeServerMessage doCreateLargeMessage(long id, final SessionSendLargeMessage packet) + private LargeServerMessage doCreateLargeMessage(final long id, final SessionSendLargeMessage packet) { try { - return createLargeMessageStorage(id, packet.getLargeMessageHeader()); + LargeServerMessage msg = createLargeMessageStorage(id, packet.getLargeMessageHeader()); + + setPagingStore(msg); + + return msg; } catch (Exception e) { @@ -1888,7 +1850,7 @@ } } - private void rollback(boolean lastMessageAsDelived) throws Exception + private void rollback(final boolean lastMessageAsDelived) throws Exception { if (tx == null) { @@ -1915,15 +1877,66 @@ */ private void releaseOutStanding(final ServerMessage message, final int credits) throws Exception { - CreditManagerHolder holder = getCreditManagerHolder(message.getDestination()); + CreditManagerHolder holder = getCreditManagerHolder(message); holder.outstandingCredits -= credits; holder.store.returnProducerCredits(credits); } - + + // TODO can we combine these two methods.... + private CreditManagerHolder getCreditManagerHolder(final SimpleString address) throws Exception + { + CreditManagerHolder holder = creditManagerHolders.get(address); + + if (holder == null) + { + PagingStore store = postOffice.getPagingManager().getPageStore(address); + + holder = new CreditManagerHolder(store); + + creditManagerHolders.put(address, holder); + } + + return holder; + } + + private CreditManagerHolder getCreditManagerHolder(final ServerMessage message) throws Exception + { + SimpleString address = message.getDestination(); + + CreditManagerHolder holder = creditManagerHolders.get(address); + + if (holder == null) + { + holder = new CreditManagerHolder(message.getPagingStore()); + + creditManagerHolders.put(address, holder); + } + + return holder; + } + + private void setPagingStore(final ServerMessage message) throws Exception + { + PagingStore store = postOffice.getPagingManager().getPageStore(message.getDestination()); + + message.setPagingStore(store); + } + + private void sendProducerCredits(final CreditManagerHolder holder, final int credits, final SimpleString address) + { + holder.outstandingCredits += credits; + + Packet packet = new SessionProducerCreditsMessage(credits, address); + + channel.send(packet); + } + private void send(final ServerMessage msg) throws Exception { + // Look up the paging store + // check the user has write access to this address. try { @@ -1947,4 +1960,20 @@ postOffice.route(msg, tx); } } + + private static final class CreditManagerHolder + { + CreditManagerHolder(final PagingStore store) + { + this.store = store; + + manager = store.getProducerCreditManager(); + } + + final PagingStore store; + + final ServerProducerCreditManager manager; + + volatile int outstandingCredits; + } } Modified: trunk/src/main/org/hornetq/core/server/impl/ServerSessionPacketHandler.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerSessionPacketHandler.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/src/main/org/hornetq/core/server/impl/ServerSessionPacketHandler.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -20,10 +20,10 @@ import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_CLOSE; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_COMMIT; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_CONSUMER_CLOSE; -import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_FORCE_CONSUMER_DELIVERY; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_CREATECONSUMER; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_EXPIRED; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_FLOWTOKEN; +import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_FORCE_CONSUMER_DELIVERY; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_QUEUEQUERY; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_ROLLBACK; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.SESS_SEND; @@ -53,11 +53,11 @@ import org.hornetq.core.remoting.impl.wireformat.SessionAcknowledgeMessage; import org.hornetq.core.remoting.impl.wireformat.SessionBindingQueryMessage; import org.hornetq.core.remoting.impl.wireformat.SessionConsumerCloseMessage; -import org.hornetq.core.remoting.impl.wireformat.SessionForceConsumerDelivery; import org.hornetq.core.remoting.impl.wireformat.SessionConsumerFlowCreditMessage; import org.hornetq.core.remoting.impl.wireformat.SessionCreateConsumerMessage; import org.hornetq.core.remoting.impl.wireformat.SessionDeleteQueueMessage; import org.hornetq.core.remoting.impl.wireformat.SessionExpiredMessage; +import org.hornetq.core.remoting.impl.wireformat.SessionForceConsumerDelivery; import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryMessage; import org.hornetq.core.remoting.impl.wireformat.SessionRequestProducerCreditsMessage; import org.hornetq.core.remoting.impl.wireformat.SessionSendContinuationMessage; @@ -275,12 +275,11 @@ { SessionForceConsumerDelivery message = (SessionForceConsumerDelivery)packet; session.handleForceConsumerDelivery(message); - break; + break; } case PacketImpl.SESS_PRODUCER_REQUEST_CREDITS: { - SessionRequestProducerCreditsMessage message = (SessionRequestProducerCreditsMessage) - packet; + SessionRequestProducerCreditsMessage message = (SessionRequestProducerCreditsMessage)packet; session.handleRequestProducerCredits(message); break; } Modified: trunk/tests/src/org/hornetq/tests/integration/client/PagingTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/PagingTest.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/tests/src/org/hornetq/tests/integration/client/PagingTest.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -27,6 +27,7 @@ import org.hornetq.core.config.Configuration; import org.hornetq.core.logging.Logger; import org.hornetq.core.message.impl.MessageImpl; +import org.hornetq.core.paging.impl.TestSupportPageStore; import org.hornetq.core.remoting.spi.HornetQBuffer; import org.hornetq.core.server.HornetQServer; import org.hornetq.core.server.Queue; @@ -47,17 +48,17 @@ */ public class PagingTest extends ServiceTestBase { - + public PagingTest(String name) { super(name); } - + public PagingTest() { super(); } - + // Constants ----------------------------------------------------- private static final Logger log = Logger.getLogger(PagingTest.class); @@ -160,7 +161,7 @@ ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT); assertNotNull(message2); - + assertEquals(i, ((Integer)message2.getProperty(new SimpleString("id"))).intValue()); message2.acknowledge(); @@ -244,7 +245,7 @@ message.setBody(bodyLocal); // Stop sending message as soon as we start paging - if (server.getPostOffice().getPagingManager().isPaging(ADDRESS)) + if (server.getPostOffice().getPagingManager().getPageStore(ADDRESS).isPaging()) { break; } @@ -253,7 +254,7 @@ producer.send(message); } - assertTrue(server.getPostOffice().getPagingManager().isPaging(ADDRESS)); + assertTrue(server.getPostOffice().getPagingManager().getPageStore(ADDRESS).isPaging()); session.start(); @@ -389,8 +390,12 @@ message.setBody(ChannelBuffers.wrappedBuffer(body)); message.putIntProperty(new SimpleString("id"), i); + TestSupportPageStore store = (TestSupportPageStore)server.getPostOffice() + .getPagingManager() + .getPageStore(ADDRESS); + // Worse scenario possible... only schedule what's on pages - if (server.getPostOffice().getPagingManager().getPageStore(ADDRESS).getCurrentPage() != null) + if (store.getCurrentPage() != null) { message.putLongProperty(MessageImpl.HDR_SCHEDULED_DELIVERY_TIME, scheduledTime); } @@ -1112,7 +1117,7 @@ // Package protected --------------------------------------------- // Protected ----------------------------------------------------- - + protected Configuration createDefaultConfig() { Configuration config = super.createDefaultConfig(); Modified: trunk/tests/src/org/hornetq/tests/integration/paging/PageCrashTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/paging/PageCrashTest.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/tests/src/org/hornetq/tests/integration/paging/PageCrashTest.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -242,10 +242,11 @@ @Override protected PagingManager createPagingManager() { - return new PagingManagerImpl(new FailurePagingStoreFactoryNIO(super.getConfiguration().getPagingDirectory()), + return new PagingManagerImpl(new FailurePagingStoreFactoryNIO(super.getConfiguration().getPagingDirectory(), + super.getConfiguration() + .isJournalSyncNonTransactional()), super.getStorageManager(), - super.getAddressSettingsRepository(), - super.getConfiguration().isJournalSyncNonTransactional()); + super.getAddressSettingsRepository()); } class FailurePagingStoreFactoryNIO extends PagingStoreFactoryNIO @@ -255,9 +256,9 @@ * @param directory * @param maxThreads */ - public FailurePagingStoreFactoryNIO(final String directory) + public FailurePagingStoreFactoryNIO(final String directory, final boolean syncNonTransactional) { - super(directory, new OrderedExecutorFactory(Executors.newCachedThreadPool())); + super(directory, new OrderedExecutorFactory(Executors.newCachedThreadPool()), syncNonTransactional); } // Constants ----------------------------------------------------- @@ -277,7 +278,7 @@ factoryField.setAccessible(true); OrderedExecutorFactory factory = (org.hornetq.utils.OrderedExecutorFactory)factoryField.get(this); - return new FailingPagingStore(destinationName, settings, factory.getExecutor()); + return new FailingPagingStore(destinationName, settings, factory.getExecutor(), syncNonTransactional); } // Package protected --------------------------------------------- @@ -297,16 +298,19 @@ */ public FailingPagingStore(final SimpleString storeName, final AddressSettings addressSettings, - final Executor executor) + final Executor executor, + final boolean syncNonTransactional) { - super(getPostOffice().getPagingManager(), + super(storeName, + getPostOffice().getPagingManager(), getStorageManager(), getPostOffice(), null, FailurePagingStoreFactoryNIO.this, storeName, addressSettings, - executor); + executor, + syncNonTransactional); } @Override Modified: trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -109,7 +109,8 @@ try { - ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); + ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, + ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); manager.start(); manager.stop(); } @@ -134,11 +135,13 @@ try { - ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); + ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, + ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); manager.start(); try { - manager.compareJournals(new JournalLoadInformation[]{new JournalLoadInformation(2,2), new JournalLoadInformation(2,2)}); + manager.compareJournals(new JournalLoadInformation[] { new JournalLoadInformation(2, 2), + new JournalLoadInformation(2, 2) }); fail("Exception was expected"); } catch (HornetQException e) @@ -147,7 +150,8 @@ assertEquals(HornetQException.ILLEGAL_STATE, e.getCode()); } - manager.compareJournals(new JournalLoadInformation[]{new JournalLoadInformation(), new JournalLoadInformation()}); + manager.compareJournals(new JournalLoadInformation[] { new JournalLoadInformation(), + new JournalLoadInformation() }); manager.stop(); } @@ -173,13 +177,15 @@ try { - ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); + ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, + ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); manager.start(); try { - ReplicationManagerImpl manager2 = new ReplicationManagerImpl(failoverManager, ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); + ReplicationManagerImpl manager2 = new ReplicationManagerImpl(failoverManager, + ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); manager2.start(); fail("Exception was expected"); @@ -212,7 +218,8 @@ try { - ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); + ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, + ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); try { @@ -246,7 +253,8 @@ try { - ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); + ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, + ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); manager.start(); Journal replicatedJournal = new ReplicatedJournal((byte)1, new FakeJournal(), manager); @@ -363,7 +371,8 @@ try { - ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); + ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, + ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); manager.start(); Journal replicatedJournal = new ReplicatedJournal((byte)1, new FakeJournal(), manager); @@ -425,7 +434,8 @@ try { - ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); + ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, + ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); manager.start(); fail("Exception expected"); } @@ -450,7 +460,8 @@ try { - ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); + ReplicationManagerImpl manager = new ReplicationManagerImpl(failoverManager, + ConfigurationImpl.DEFAULT_BACKUP_WINDOW_SIZE); manager.start(); Journal replicatedJournal = new ReplicatedJournal((byte)1, new FakeJournal(), manager); @@ -580,10 +591,10 @@ { PagingManager paging = new PagingManagerImpl(new PagingStoreFactoryNIO(configuration.getPagingDirectory(), - executorFactory), + executorFactory, + false), storageManager, - addressSettingsRepository, - false); + addressSettingsRepository); paging.start(); return paging; Modified: trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingManagerImplTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingManagerImplTest.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingManagerImplTest.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -64,10 +64,9 @@ addressSettings.setDefault(settings); - PagingManagerImpl managerImpl = new PagingManagerImpl(new PagingStoreFactoryNIO(getPageDir(), new OrderedExecutorFactory(Executors.newCachedThreadPool())), + PagingManagerImpl managerImpl = new PagingManagerImpl(new PagingStoreFactoryNIO(getPageDir(), new OrderedExecutorFactory(Executors.newCachedThreadPool()), true), new NullStorageManager(), - addressSettings, - true); + addressSettings); managerImpl.start(); @@ -75,11 +74,11 @@ ServerMessage msg = createMessage(1l, new SimpleString("simple-test"), createRandomBuffer(10)); - assertFalse(store.page(new PagedMessageImpl(msg), true, true)); + assertFalse(store.page(msg, true)); store.startPaging(); - assertTrue(store.page(new PagedMessageImpl(msg), true, true)); + assertTrue(store.page(msg, true)); Page page = store.depage(); @@ -97,7 +96,7 @@ assertNull(store.depage()); - assertFalse(store.page(new PagedMessageImpl(msg), true, true)); + assertFalse(store.page(msg, true)); } Modified: trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -18,6 +18,7 @@ import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; @@ -39,12 +40,11 @@ import org.hornetq.core.paging.PagingStore; import org.hornetq.core.paging.PagingStoreFactory; import org.hornetq.core.paging.impl.PageTransactionInfoImpl; -import org.hornetq.core.paging.impl.PagedMessageImpl; import org.hornetq.core.paging.impl.PagingStoreImpl; import org.hornetq.core.paging.impl.TestSupportPageStore; +import org.hornetq.core.persistence.GroupingInfo; import org.hornetq.core.persistence.QueueBindingInfo; import org.hornetq.core.persistence.StorageManager; -import org.hornetq.core.persistence.GroupingInfo; import org.hornetq.core.postoffice.Binding; import org.hornetq.core.postoffice.PostOffice; import org.hornetq.core.remoting.spi.HornetQBuffer; @@ -56,8 +56,8 @@ import org.hornetq.core.server.group.impl.GroupBinding; import org.hornetq.core.server.impl.ServerMessageImpl; import org.hornetq.core.settings.HierarchicalRepository; +import org.hornetq.core.settings.impl.AddressFullMessagePolicy; import org.hornetq.core.settings.impl.AddressSettings; -import org.hornetq.core.settings.impl.AddressFullMessagePolicy; import org.hornetq.core.transaction.ResourceManager; import org.hornetq.tests.unit.core.journal.impl.fakes.FakeSequentialFileFactory; import org.hornetq.tests.unit.core.server.impl.fakes.FakePostOffice; @@ -142,14 +142,17 @@ AddressSettings addressSettings = new AddressSettings(); addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE); - PagingStore storeImpl = new PagingStoreImpl(createMockManager(), + + PagingStore storeImpl = new PagingStoreImpl(destinationTestName, + createMockManager(), createStorageManagerMock(), createPostOfficeMock(), factory, null, destinationTestName, addressSettings, - executor); + executor, + true); storeImpl.start(); @@ -180,14 +183,16 @@ AddressSettings addressSettings = new AddressSettings(); addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE); - PagingStore storeImpl = new PagingStoreImpl(createMockManager(), - createStorageManagerMock(), - createPostOfficeMock(), - factory, - storeFactory, - destinationTestName, - addressSettings, - executor); + TestSupportPageStore storeImpl = new PagingStoreImpl(destinationTestName, + createMockManager(), + createStorageManagerMock(), + createPostOfficeMock(), + factory, + storeFactory, + destinationTestName, + addressSettings, + executor, + true); storeImpl.start(); @@ -204,24 +209,26 @@ buffers.add(buffer); SimpleString destination = new SimpleString("test"); - PagedMessageImpl msg = createMessage(destination, buffer); + ServerMessage msg = createMessage(storeImpl, destination, buffer); assertTrue(storeImpl.isPaging()); - assertTrue(storeImpl.page(msg, true, true)); + assertTrue(storeImpl.page(msg, true)); assertEquals(1, storeImpl.getNumberOfPages()); storeImpl.sync(); - storeImpl = new PagingStoreImpl(createMockManager(), + storeImpl = new PagingStoreImpl(destinationTestName, + createMockManager(), createStorageManagerMock(), createPostOfficeMock(), factory, null, destinationTestName, addressSettings, - executor); + executor, + true); storeImpl.start(); @@ -239,14 +246,16 @@ AddressSettings addressSettings = new AddressSettings(); addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE); - TestSupportPageStore storeImpl = new PagingStoreImpl(createMockManager(), + TestSupportPageStore storeImpl = new PagingStoreImpl(destinationTestName, + createMockManager(), createStorageManagerMock(), createPostOfficeMock(), factory, storeFactory, destinationTestName, addressSettings, - executor); + executor, + true); storeImpl.start(); @@ -263,9 +272,9 @@ buffers.add(buffer); - PagedMessageImpl msg = createMessage(destination, buffer); + ServerMessage msg = createMessage(storeImpl, destination, buffer); - assertTrue(storeImpl.page(msg, true, true)); + assertTrue(storeImpl.page(msg, true)); } assertEquals(1, storeImpl.getNumberOfPages()); @@ -304,14 +313,16 @@ AddressSettings addressSettings = new AddressSettings(); addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE); - TestSupportPageStore storeImpl = new PagingStoreImpl(createMockManager(), + TestSupportPageStore storeImpl = new PagingStoreImpl(destinationTestName, + createMockManager(), createStorageManagerMock(), createPostOfficeMock(), factory, storeFactory, destinationTestName, addressSettings, - executor); + executor, + true); storeImpl.start(); @@ -335,9 +346,9 @@ storeImpl.forceAnotherPage(); } - PagedMessageImpl msg = createMessage(destination, buffer); + ServerMessage msg = createMessage(storeImpl, destination, buffer); - assertTrue(storeImpl.page(msg, true, true)); + assertTrue(storeImpl.page(msg, true)); } assertEquals(2, storeImpl.getNumberOfPages()); @@ -367,9 +378,9 @@ assertTrue(storeImpl.isPaging()); - PagedMessageImpl msg = createMessage(destination, buffers.get(0)); + ServerMessage msg = createMessage(storeImpl, destination, buffers.get(0)); - assertTrue(storeImpl.page(msg, true, true)); + assertTrue(storeImpl.page(msg, true)); Page newPage = storeImpl.depage(); @@ -387,11 +398,11 @@ assertFalse(storeImpl.isPaging()); - assertFalse(storeImpl.page(msg, true, true)); + assertFalse(storeImpl.page(msg, true)); storeImpl.startPaging(); - assertTrue(storeImpl.page(msg, true, true)); + assertTrue(storeImpl.page(msg, true)); Page page = storeImpl.depage(); @@ -427,7 +438,6 @@ protected void testConcurrentPaging(final SequentialFileFactory factory, final int numberOfThreads) throws Exception, InterruptedException { - PagingStoreFactory storeFactory = new FakeStoreFactory(factory); final int MAX_SIZE = 1024 * 10; @@ -438,7 +448,7 @@ final CountDownLatch latchStart = new CountDownLatch(numberOfThreads); - final ConcurrentHashMap<Long, PagedMessageImpl> buffers = new ConcurrentHashMap<Long, PagedMessageImpl>(); + final ConcurrentHashMap<Long, ServerMessage> buffers = new ConcurrentHashMap<Long, ServerMessage>(); final ArrayList<Page> readPages = new ArrayList<Page>(); @@ -446,14 +456,16 @@ settings.setPageSizeBytes(MAX_SIZE); settings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE); - final TestSupportPageStore storeImpl = new PagingStoreImpl(createMockManager(), + final TestSupportPageStore storeImpl = new PagingStoreImpl(destinationTestName, + createMockManager(), createStorageManagerMock(), createPostOfficeMock(), factory, storeFactory, new SimpleString("test"), settings, - executor); + executor, + true); storeImpl.start(); @@ -480,8 +492,8 @@ while (true) { long id = messageIdGenerator.incrementAndGet(); - PagedMessageImpl msg = createMessage(destination, createRandomBuffer(id, 5)); - if (storeImpl.page(msg, false, true)) + ServerMessage msg = createMessage(storeImpl, destination, createRandomBuffer(id, 5)); + if (storeImpl.page(msg, true)) { buffers.put(id, msg); } @@ -564,7 +576,7 @@ throw consumer.e; } - final ConcurrentHashMap<Long, PagedMessage> buffers2 = new ConcurrentHashMap<Long, PagedMessage>(); + final ConcurrentMap<Long, ServerMessage> buffers2 = new ConcurrentHashMap<Long, ServerMessage>(); for (Page page : readPages) { @@ -577,13 +589,11 @@ long id = msg.getMessage(null).getBody().readLong(); msg.getMessage(null).getBody().resetReaderIndex(); - PagedMessageImpl msgWritten = buffers.remove(id); - buffers2.put(id, msg); + ServerMessage msgWritten = buffers.remove(id); + buffers2.put(id, msg.getMessage(null)); assertNotNull(msgWritten); - assertEquals(msg.getMessage(null).getDestination(), msgWritten.getMessage(null).getDestination()); - assertEqualsByteArrays(msgWritten.getMessage(null).getBody().array(), msg.getMessage(null) - .getBody() - .array()); + assertEquals(msg.getMessage(null).getDestination(), msgWritten.getDestination()); + assertEqualsByteArrays(msgWritten.getBody().array(), msg.getMessage(null).getBody().array()); } } @@ -601,14 +611,16 @@ fileTmp.close(); } - TestSupportPageStore storeImpl2 = new PagingStoreImpl(createMockManager(), + TestSupportPageStore storeImpl2 = new PagingStoreImpl(destinationTestName, + createMockManager(), createStorageManagerMock(), createPostOfficeMock(), factory, storeFactory, new SimpleString("test"), settings, - executor); + executor, + true); storeImpl2.start(); int numberOfPages = storeImpl2.getNumberOfPages(); @@ -621,9 +633,9 @@ assertEquals(numberOfPages, storeImpl2.getNumberOfPages()); long lastMessageId = messageIdGenerator.incrementAndGet(); - PagedMessage lastMsg = createMessage(destination, createRandomBuffer(lastMessageId, 5)); + ServerMessage lastMsg = createMessage(storeImpl, destination, createRandomBuffer(lastMessageId, 5)); - storeImpl2.page(lastMsg, false, true); + storeImpl2.page(lastMsg, true); buffers2.put(lastMessageId, lastMsg); Page lastPage = null; @@ -647,12 +659,10 @@ { long id = msg.getMessage(null).getBody().readLong(); - PagedMessage msgWritten = buffers2.remove(id); + ServerMessage msgWritten = buffers2.remove(id); assertNotNull(msgWritten); - assertEquals(msg.getMessage(null).getDestination(), msgWritten.getMessage(null).getDestination()); - assertEqualsByteArrays(msgWritten.getMessage(null).getBody().array(), msg.getMessage(null) - .getBody() - .array()); + assertEquals(msg.getMessage(null).getDestination(), msgWritten.getDestination()); + assertEqualsByteArrays(msgWritten.getBody().array(), msg.getMessage(null).getBody().array()); } } @@ -663,14 +673,11 @@ lastMessages.get(0).getMessage(null).getBody().resetReaderIndex(); assertEquals(lastMessages.get(0).getMessage(null).getBody().readLong(), lastMessageId); - assertEqualsByteArrays(lastMessages.get(0).getMessage(null).getBody().array(), lastMsg.getMessage(null) - .getBody() - .array()); + assertEqualsByteArrays(lastMessages.get(0).getMessage(null).getBody().array(), lastMsg.getBody().array()); assertEquals(0, buffers2.size()); assertEquals(0, storeImpl.getAddressSize()); - } /** @@ -691,12 +698,15 @@ return new FakePostOffice(); } - private PagedMessageImpl createMessage(final SimpleString destination, final HornetQBuffer buffer) + private ServerMessage createMessage(final PagingStore store, final SimpleString destination, final HornetQBuffer buffer) { ServerMessage msg = new ServerMessageImpl((byte)1, true, 0, System.currentTimeMillis(), (byte)0, buffer); msg.setDestination(destination); - return new PagedMessageImpl(msg); + + msg.setPagingStore(store); + + return msg; } private HornetQBuffer createRandomBuffer(final long id, final int size) @@ -947,29 +957,30 @@ /* (non-Javadoc) * @see org.hornetq.core.persistence.StorageManager#loadBindingJournal(java.util.List) */ - public JournalLoadInformation loadBindingJournal(final List<QueueBindingInfo> queueBindingInfos, List<GroupingInfo> groupingInfos) throws Exception + public JournalLoadInformation loadBindingJournal(final List<QueueBindingInfo> queueBindingInfos, + List<GroupingInfo> groupingInfos) throws Exception { return new JournalLoadInformation(); } public void addGrouping(GroupBinding groupBinding) throws Exception { - //To change body of implemented methods use File | Settings | File Templates. + // To change body of implemented methods use File | Settings | File Templates. } public void deleteGrouping(GroupBinding groupBinding) throws Exception { - //To change body of implemented methods use File | Settings | File Templates. + // To change body of implemented methods use File | Settings | File Templates. } /* (non-Javadoc) * @see org.hornetq.core.persistence.StorageManager#loadMessageJournal(org.hornetq.core.paging.PagingManager, java.util.Map, org.hornetq.core.transaction.ResourceManager, java.util.Map) */ public JournalLoadInformation loadMessageJournal(PostOffice postOffice, - PagingManager pagingManager, - ResourceManager resourceManager, - Map<Long, Queue> queues, - Map<SimpleString, List<Pair<byte[], Long>>> duplicateIDMap) throws Exception + PagingManager pagingManager, + ResourceManager resourceManager, + Map<Long, Queue> queues, + Map<SimpleString, List<Pair<byte[], Long>>> duplicateIDMap) throws Exception { return new JournalLoadInformation(); } @@ -1136,8 +1147,7 @@ */ public void afterReplicated(Runnable run) { - - + } /* (non-Javadoc) @@ -1145,8 +1155,7 @@ */ public void completeReplication() { - - + } /* (non-Javadoc) @@ -1154,7 +1163,7 @@ */ public LargeServerMessage createLargeMessage(long messageId, byte[] header) { - + return null; } @@ -1163,7 +1172,7 @@ */ public boolean isReplicated() { - + return false; } @@ -1172,7 +1181,7 @@ */ public JournalLoadInformation[] loadInternalOnly() throws Exception { - return null; + return null; } /* (non-Javadoc) @@ -1180,8 +1189,7 @@ */ public void pageClosed(SimpleString storeName, int pageNumber) { - - + } /* (non-Javadoc) @@ -1189,8 +1197,7 @@ */ public void pageDeleted(SimpleString storeName, int pageNumber) { - - + } /* (non-Javadoc) @@ -1198,8 +1205,7 @@ */ public void pageWrite(PagedMessage message, int pageNumber) { - - + } /* (non-Javadoc) @@ -1293,8 +1299,6 @@ { } - - } } Modified: trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java 2009-11-05 13:30:49 UTC (rev 8222) +++ trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java 2009-11-05 14:23:20 UTC (rev 8223) @@ -895,6 +895,36 @@ return 0; } + public PagingStore getPagingStore() + { + // TODO Auto-generated method stub + return null; + } + + public void setPagingStore(PagingStore store) + { + // TODO Auto-generated method stub + + } + + public boolean page(boolean duplicateDetection) throws Exception + { + // TODO Auto-generated method stub + return false; + } + + public boolean page(long transactionID, boolean duplicateDetection) throws Exception + { + // TODO Auto-generated method stub + return false; + } + + public boolean storeIsPaging() + { + // TODO Auto-generated method stub + return false; + } + } class FakeFilter implements Filter

16 years, 2 months

1
0
0 / 0

JBoss hornetq SVN: r8222 - in trunk: tests/src/org/hornetq/tests/integration/cluster/distribution and 1 other directories.

by do-not-reply＠jboss.org

Author: ataylor Date: 2009-11-05 08:30:49 -0500 (Thu, 05 Nov 2009) New Revision: 8222 Modified: trunk/src/main/org/hornetq/core/server/group/impl/RemoteGroupingHandler.java trunk/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusteredGroupingTest.java trunk/tests/src/org/hornetq/tests/integration/cluster/failover/GroupingFailoverTestBase.java Log: test tweaks + fix Modified: trunk/src/main/org/hornetq/core/server/group/impl/RemoteGroupingHandler.java =================================================================== --- trunk/src/main/org/hornetq/core/server/group/impl/RemoteGroupingHandler.java 2009-11-05 08:42:21 UTC (rev 8221) +++ trunk/src/main/org/hornetq/core/server/group/impl/RemoteGroupingHandler.java 2009-11-05 13:30:49 UTC (rev 8222) @@ -175,7 +175,6 @@ { SimpleString clusterName = (SimpleString)notification.getProperties() .getProperty(ManagementHelper.HDR_CLUSTER_NAME); - groupMap.remove(clusterName); List<SimpleString> list = groupMap.remove(clusterName); if (list != null) { Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusteredGroupingTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusteredGroupingTest.java 2009-11-05 08:42:21 UTC (rev 8221) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusteredGroupingTest.java 2009-11-05 13:30:49 UTC (rev 8222) @@ -16,12 +16,15 @@ import java.util.concurrent.TimeUnit; import org.hornetq.core.management.Notification; +import org.hornetq.core.management.NotificationListener; +import org.hornetq.core.management.NotificationType; import org.hornetq.core.message.impl.MessageImpl; import org.hornetq.core.server.group.GroupingHandler; import org.hornetq.core.server.group.impl.GroupBinding; import org.hornetq.core.server.group.impl.GroupingHandlerConfiguration; import org.hornetq.core.server.group.impl.Proposal; import org.hornetq.core.server.group.impl.Response; +import org.hornetq.core.client.management.impl.ManagementHelper; import org.hornetq.utils.SimpleString; /** @@ -49,7 +52,7 @@ startServers(0, 1, 2); try - { + { setupSessionFactory(0, isNetty()); setupSessionFactory(1, isNetty()); @@ -100,7 +103,7 @@ setupClusterConnection("cluster2", "queues", false, 1, isNetty(), 2, 0, 1); setUpGroupHandler(GroupingHandlerConfiguration.TYPE.REMOTE, 1, 1); - + setUpGroupHandler(GroupingHandlerConfiguration.TYPE.REMOTE, 2, 1); startServers(0, 1, 2); @@ -623,22 +626,52 @@ waitForBindings(1, "queues.testaddress", 2, 0, false); waitForBindings(2, "queues.testaddress", 2, 1, false); - sendInRange(1, "queues.testaddress", 0, 10, false, MessageImpl.HDR_GROUP_ID, new SimpleString("id1")); + sendInRange(1, "queues.testaddress", 0, 10, true, MessageImpl.HDR_GROUP_ID, new SimpleString("id1")); - verifyReceiveAllInRange(0, 10, 0); + verifyReceiveAllInRange(true, 0, 10, 0); - stopServers(1); + closeAllConsumers(); - sendInRange(2, "queues.testaddress", 10, 20, false, MessageImpl.HDR_GROUP_ID, new SimpleString("id1")); - startServers(1); + final CountDownLatch latch = new CountDownLatch(4); + NotificationListener listener = new NotificationListener() + { + public void onNotification(Notification notification) + { + if(NotificationType.BINDING_REMOVED == notification.getType()) + { + if(notification.getProperties().getProperty(ManagementHelper.HDR_ADDRESS).toString().equals("queues.testaddress")) + { + latch.countDown(); + } + } + else if(NotificationType.BINDING_ADDED == notification.getType()) + { + if(notification.getProperties().getProperty(ManagementHelper.HDR_ADDRESS).toString().equals("queues.testaddress")) + { + latch.countDown(); + } + } + } + }; + getServer(0).getManagementService().addNotificationListener(listener); + getServer(2).getManagementService().addNotificationListener(listener); - addConsumer(1, 1, "queue0", null); - waitForBindings(1, "queues.testaddress", 1, 1, true); + stopServers(1); + startServers(1); + assertTrue("timed out waiting for bindings to be removed and added back",latch.await(5, TimeUnit.SECONDS)); + getServer(0).getManagementService().removeNotificationListener(listener); + getServer(2).getManagementService().removeNotificationListener(listener); + addConsumer(1, 2, "queue0", null); + waitForBindings(2, "queues.testaddress", 1, 1, true); + waitForBindings(1, "queues.testaddress", 2, 1, false); + waitForBindings(0, "queues.testaddress", 2, 1, false); + sendInRange(2, "queues.testaddress", 10, 20, true, MessageImpl.HDR_GROUP_ID, new SimpleString("id1")); verifyReceiveAllInRange(10, 20, 1); + System.out.println("*****************************************************************************"); } finally @@ -697,17 +730,42 @@ closeAllConsumers(); sendInRange(2, "queues.testaddress", 10, 20, true, MessageImpl.HDR_GROUP_ID, new SimpleString("id1")); + final CountDownLatch latch = new CountDownLatch(4); + NotificationListener listener = new NotificationListener() + { + public void onNotification(Notification notification) + { + if(NotificationType.BINDING_REMOVED == notification.getType()) + { + if(notification.getProperties().getProperty(ManagementHelper.HDR_ADDRESS).toString().equals("queues.testaddress")) + { + latch.countDown(); + } + } + else if(NotificationType.BINDING_ADDED == notification.getType()) + { + if(notification.getProperties().getProperty(ManagementHelper.HDR_ADDRESS).toString().equals("queues.testaddress")) + { + latch.countDown(); + } + } + } + }; + getServer(0).getManagementService().addNotificationListener(listener); + getServer(2).getManagementService().addNotificationListener(listener); - stopServers(1); startServers(1); - + assertTrue("timed out waiting for bindings to be removed and added back",latch.await(5, TimeUnit.SECONDS)); + getServer(0).getManagementService().removeNotificationListener(listener); + getServer(2).getManagementService().removeNotificationListener(listener); addConsumer(1, 1, "queue0", null); waitForBindings(1, "queues.testaddress", 1, 1, true); + waitForBindings(0, "queues.testaddress", 2, 1, false); + waitForBindings(2, "queues.testaddress", 2, 1, false); - verifyReceiveAllInRange(10, 20, 1); @@ -766,17 +824,40 @@ sendInRange(1, "queues.testaddress", 0, 10, false, MessageImpl.HDR_GROUP_ID, new SimpleString("id1")); verifyReceiveAllInRange(0, 10, 0); - + final CountDownLatch latch = new CountDownLatch(4); + NotificationListener listener = new NotificationListener() + { + public void onNotification(Notification notification) + { + if(NotificationType.BINDING_REMOVED == notification.getType()) + { + if(notification.getProperties().getProperty(ManagementHelper.HDR_ADDRESS).toString().equals("queues.testaddress")) + { + latch.countDown(); + } + } + else if(NotificationType.BINDING_ADDED == notification.getType()) + { + if(notification.getProperties().getProperty(ManagementHelper.HDR_ADDRESS).toString().equals("queues.testaddress")) + { + latch.countDown(); + } + } + } + }; + getServer(0).getManagementService().addNotificationListener(listener); + getServer(2).getManagementService().addNotificationListener(listener); stopServers(1); - startServers(1); - - sendInRange(2, "queues.testaddress", 10, 20, false, MessageImpl.HDR_GROUP_ID, new SimpleString("id1")); - - + assertTrue("timed out waiting for bindings to be removed and added back",latch.await(5, TimeUnit.SECONDS)); + getServer(0).getManagementService().removeNotificationListener(listener); + getServer(2).getManagementService().removeNotificationListener(listener); addConsumer(1, 1, "queue0", null); waitForBindings(1, "queues.testaddress", 1, 1, true); + waitForBindings(0, "queues.testaddress", 2, 1, false); + waitForBindings(2, "queues.testaddress", 2, 1, false); + sendInRange(2, "queues.testaddress", 10, 20, false, MessageImpl.HDR_GROUP_ID, new SimpleString("id1")); verifyReceiveAllInRange(10, 20, 1); Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/failover/GroupingFailoverTestBase.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/failover/GroupingFailoverTestBase.java 2009-11-05 08:42:21 UTC (rev 8221) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/failover/GroupingFailoverTestBase.java 2009-11-05 13:30:49 UTC (rev 8222) @@ -65,12 +65,12 @@ createQueue(0, "queues.testaddress", "queue0", null, true); createQueue(1, "queues.testaddress", "queue0", null, true); + waitForBindings(0, "queues.testaddress", 1, 0, true); + waitForBindings(1, "queues.testaddress", 1, 0, true); + addConsumer(0, 0, "queue0", null); addConsumer(1, 1, "queue0", null); - waitForBindings(0, "queues.testaddress", 1, 1, true); - waitForBindings(1, "queues.testaddress", 1, 1, true); - waitForBindings(0, "queues.testaddress", 1, 1, false); waitForBindings(1, "queues.testaddress", 1, 1, false); @@ -168,11 +168,13 @@ createQueue(0, "queues.testaddress", "queue0", null, true); createQueue(1, "queues.testaddress", "queue0", null, true); + + waitForBindings(0, "queues.testaddress", 1, 0, true); + waitForBindings(1, "queues.testaddress", 1, 0, true); + addConsumer(0, 0, "queue0", null); addConsumer(1, 1, "queue0", null); - waitForBindings(0, "queues.testaddress", 1, 1, true); - waitForBindings(1, "queues.testaddress", 1, 1, true); waitForBindings(0, "queues.testaddress", 1, 1, false); waitForBindings(1, "queues.testaddress", 1, 1, false);

16 years, 2 months

1
0
0 / 0

2026

2025

2024

2023

2022

2021

2020

2019

2018

2017

2016

2015

2014

2013

2012

2011

2010

2009

hornetq-commits November 2009