Author: clebert.suconic(a)jboss.com Date: 2011-05-28 01:06:01 -0400 (Sat, 28 May 2011) New Revision: 10751 Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/cursor/PagedReferenceImpl.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/cursor/impl/PageCursorProviderImpl.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/impl/PageImpl.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/impl/PagedMessageImpl.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/QueueImpl.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java branches/Branch_2_2_EAP/src/main/org/hornetq/utils/SoftValueHashMap.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/client/PagingTest.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/util/SoftValueMapTest.java Log: HORNETQ-706 - fixing performance issue on paging Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/cursor/PagedReferenceImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/cursor/PagedReferenceImpl.java 2011-05-27 15:41:56 UTC (rev 10750) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/cursor/PagedReferenceImpl.java 2011-05-28 05:06:01 UTC (rev 10751) @@ -216,7 +216,7 @@ { return "PagedReferenceImpl [position=" + position + ", message=" + - message + + getPagedMessage() + ", deliveryTime=" + deliveryTime + ", persistedCount=" + Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/cursor/impl/PageCursorProviderImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/cursor/impl/PageCursorProviderImpl.java 2011-05-27 15:41:56 UTC (rev 10750) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/cursor/impl/PageCursorProviderImpl.java 2011-05-28 05:06:01 UTC (rev 10751) @@ -49,6 +49,8 @@ // Constants ----------------------------------------------------- private static final Logger log = Logger.getLogger(PageCursorProviderImpl.class); + + boolean isTrace = log.isTraceEnabled(); // Attributes ---------------------------------------------------- @@ -160,6 +162,10 @@ // anyone reading from this cache will have to wait reading to finish first // we also want only one thread reading this cache cache.lock(); + if (isTrace) + { + log.trace("adding " + pageId + " into cursor = " + this.pagingStore.getAddress()); + } softCache.put(pageId, cache); } } @@ -411,8 +417,13 @@ PagedMessage[] pgdMessages; synchronized (softCache) { - cache = softCache.remove((long)depagedPage.getPageId()); + cache = softCache.get((long)depagedPage.getPageId()); } + + if (isTrace) + { + log.trace("Removing page " + depagedPage.getPageId() + " from page-cache"); + } if (cache == null) { @@ -430,6 +441,7 @@ } depagedPage.delete(pgdMessages); + synchronized (softCache) { softCache.remove((long)depagedPage.getPageId()); Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/impl/PageImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/impl/PageImpl.java 2011-05-27 15:41:56 UTC (rev 10750) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/impl/PageImpl.java 2011-05-28 05:06:01 UTC (rev 10751) @@ -41,6 +41,8 @@ // Constants ----------------------------------------------------- private static final Logger log = Logger.getLogger(PageImpl.class); + + private static final boolean isTrace = log.isTraceEnabled(); public static final int SIZE_RECORD = DataConstants.SIZE_BYTE + DataConstants.SIZE_INT + DataConstants.SIZE_BYTE; @@ -104,6 +106,11 @@ public List<PagedMessage> read(StorageManager storage) throws Exception { + if (isTrace) + { + log.trace("reading page " + this.pageId + " on address = " + storeName); + } + ArrayList<PagedMessage> messages = new ArrayList<PagedMessage>(); size.set((int)file.size()); @@ -142,6 +149,10 @@ throw new IllegalStateException("Internal error, it wasn't possible to locate END_BYTE " + b); } msg.initMessage(storage); + if (isTrace) + { + log.trace("Reading message " + msg + " on pageId=" + this.pageId + " for address=" + storeName); + } messages.add(msg); } else @@ -226,6 +237,11 @@ { storageManager.pageDeleted(storeName, pageId); } + + if (isTrace) + { + log.trace("Deleting pageId=" + pageId + " on store " + storeName); + } if (messages != null) { Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/impl/PagedMessageImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/impl/PagedMessageImpl.java 2011-05-27 15:41:56 UTC (rev 10750) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/impl/PagedMessageImpl.java 2011-05-28 05:06:01 UTC (rev 10751) @@ -13,6 +13,8 @@ package org.hornetq.core.paging.impl; +import java.util.Arrays; + import org.hornetq.api.core.HornetQBuffer; import org.hornetq.api.core.HornetQBuffers; import org.hornetq.core.logging.Logger; @@ -158,6 +160,22 @@ DataConstants.SIZE_INT + queueIDs.length * DataConstants.SIZE_LONG; } + /* (non-Javadoc) + * @see java.lang.Object#toString() + */ + @Override + public String toString() + { + return "PagedMessageImpl [queueIDs=" + Arrays.toString(queueIDs) + + ", transactionID=" + + transactionID + + ", message=" + + message + + "]"; + } + + + // Package protected --------------------------------------------- // Protected ----------------------------------------------------- Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java 2011-05-27 15:41:56 UTC (rev 10750) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java 2011-05-28 05:06:01 UTC (rev 10751) @@ -59,6 +59,7 @@ // -------------------------------------------------------------------------------------------------------------------------- private static final Logger log = Logger.getLogger(PagingManagerImpl.class); + private static boolean isTrace = log.isTraceEnabled(); // Constructors // -------------------------------------------------------------------------------------------------------------------- @@ -167,16 +168,28 @@ public void addTransaction(final PageTransactionInfo pageTransaction) { + if (isTrace) + { + log.trace("Adding pageTransaction " + pageTransaction.getTransactionID()); + } transactions.put(pageTransaction.getTransactionID(), pageTransaction); } public void removeTransaction(final long id) { + if (isTrace) + { + log.trace("Removing pageTransaction " +id); + } transactions.remove(id); } public PageTransactionInfo getTransaction(final long id) { + if (isTrace) + { + log.trace("looking up pageTX = " + id); + } return transactions.get(id); } Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2011-05-27 15:41:56 UTC (rev 10750) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2011-05-28 05:06:01 UTC (rev 10751) @@ -13,7 +13,6 @@ package org.hornetq.core.paging.impl; -import java.io.File; import java.text.DecimalFormat; import java.util.Collections; import java.util.HashSet; @@ -42,10 +41,7 @@ import org.hornetq.core.paging.cursor.PageCursorProvider; import org.hornetq.core.paging.cursor.impl.LivePageCacheImpl; import org.hornetq.core.paging.cursor.impl.PageCursorProviderImpl; -import org.hornetq.core.persistence.OperationContext; import org.hornetq.core.persistence.StorageManager; -import org.hornetq.core.persistence.impl.journal.OperationContextImpl; -import org.hornetq.core.postoffice.PostOffice; import org.hornetq.core.server.LargeServerMessage; import org.hornetq.core.server.MessageReference; import org.hornetq.core.server.RouteContextList; @@ -56,9 +52,7 @@ import org.hornetq.core.transaction.Transaction; import org.hornetq.core.transaction.Transaction.State; import org.hornetq.core.transaction.TransactionOperation; -import org.hornetq.core.transaction.TransactionOperationAbstract; import org.hornetq.core.transaction.TransactionPropertyIndexes; -import org.hornetq.utils.ExecutorFactory; import org.hornetq.utils.Future; /** @@ -882,7 +876,14 @@ } currentPage.write(pagedMessage); + + if (isTrace) + { + log.trace("Paging message " + pagedMessage + " on pageStore " + this.getStoreName() + + " pageId=" + currentPage.getPageId()); + } + if (tx != null) { installPageTransaction(tx, listCtx); Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/QueueImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/QueueImpl.java 2011-05-27 15:41:56 UTC (rev 10750) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/QueueImpl.java 2011-05-28 05:06:01 UTC (rev 10751) @@ -82,6 +82,11 @@ public static final int MAX_DELIVERIES_IN_LOOP = 1000; public static final int CHECK_QUEUE_SIZE_PERIOD = 100; + + /** If The system gets slow for any reason, this is the maximum time an Delivery or + or depage executor should be hanging on + */ + private static final int DELIVERY_TIMEOUT = 1000; private final long id; @@ -1461,7 +1466,7 @@ int handled = 0; - long timeout = System.currentTimeMillis() + 1000; + long timeout = System.currentTimeMillis() + DELIVERY_TIMEOUT; while (handled < numRefs) { @@ -1474,11 +1479,11 @@ return; } - if (pageSubscription != null && pageSubscription.isPaging() && System.currentTimeMillis() > timeout) + if (System.currentTimeMillis() > timeout) { if (isTrace) { - log.trace("Page delivery has been running for too long. Scheduling another delivery task now"); + log.warn("delivery has been running for too long. Scheduling another delivery task now"); } deliverAsync(); @@ -1523,6 +1528,11 @@ } Consumer groupConsumer = null; + + if (isTrace) + { + log.trace("Queue " + this.getName() + " is delivering reference " + ref); + } // If a group id is set, then this overrides the consumer chosen round-robin @@ -1584,7 +1594,7 @@ } } - if (pageIterator != null && messageReferences.size() == 0 && pageSubscription.isPaging() && pageIterator.hasNext()) + if (pageIterator != null && messageReferences.size() == 0 && pageSubscription.isPaging() && pageIterator.hasNext() && !depagePending) { scheduleDepage(); } @@ -1626,26 +1636,43 @@ } } - private void depage() + private synchronized void depage() { + depagePending = false; + if (paused || pageIterator == null) { return; } long maxSize = pageSubscription.getPagingStore().getPageSizeBytes(); + + + long timeout = System.currentTimeMillis() + DELIVERY_TIMEOUT; - // System.out.println("QueueMemorySize before depage = " + queueMemorySize.get()); + if (isTrace) + { + log.trace("QueueMemorySize before depage on queue=" + this.getName() + " is " + queueMemorySize.get()); + } + int depaged = 0; - while (queueMemorySize.get() < maxSize && pageIterator.hasNext()) + while (timeout > System.currentTimeMillis() && queueMemorySize.get() < maxSize && pageIterator.hasNext()) { depaged++; PagedReference reference = pageIterator.next(); + if (isTrace) + { + log.trace("Depaging reference " + reference + " on queue " + this.getName()); + } addTail(reference, false); pageIterator.remove(); } - log.debug("Queue Memory Size after depage on queue="+this.getName() + " is " + queueMemorySize.get() + " with maxSize = " + maxSize + ". Depaged " + depaged + " messages"); - + + if (isTrace) + { + log.trace("Queue Memory Size after depage on queue="+this.getName() + " is " + queueMemorySize.get() + " with maxSize = " + maxSize + ". Depaged " + depaged + " messages"); + } + deliverAsync(); } @@ -2215,7 +2242,6 @@ { try { - depagePending = false; depage(); } catch (Exception e) Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java 2011-05-27 15:41:56 UTC (rev 10750) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java 2011-05-28 05:06:01 UTC (rev 10751) @@ -61,6 +61,8 @@ // Constants ------------------------------------------------------------------------------------ private static final Logger log = Logger.getLogger(ServerConsumerImpl.class); + + private static boolean isTrace = log.isTraceEnabled(); // Static --------------------------------------------------------------------------------------- @@ -408,6 +410,10 @@ { for (MessageReference ref : deliveringRefs) { + if (isTrace) + { + log.trace("Cancelling reference for messageID = " + ref.getMessage().getMessageID() + ", ref = " + ref); + } if (performACK) { acknowledge(false, tx, ref.getMessage().getMessageID()); Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/utils/SoftValueHashMap.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/utils/SoftValueHashMap.java 2011-05-27 15:41:56 UTC (rev 10750) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/utils/SoftValueHashMap.java 2011-05-28 05:06:01 UTC (rev 10751) @@ -25,6 +25,8 @@ import java.util.TreeSet; import java.util.concurrent.atomic.AtomicLong; +import org.hornetq.core.logging.Logger; + /** * A SoftValueHashMap * @@ -34,14 +36,18 @@ */ public class SoftValueHashMap<K, V extends SoftValueHashMap.ValueCache> implements Map<K, V> { + private static final Logger log = Logger.getLogger(SoftValueHashMap.class); + + private final boolean isTrace = log.isTraceEnabled(); + // The soft references that are already good. // too bad there's no way to override the queue method on ReferenceQueue, so I wouldn't need this private final ReferenceQueue<V> refQueue = new ReferenceQueue<V>(); private final Map<K, AggregatedSoftReference> mapDelegate = new HashMap<K, AggregatedSoftReference>(); + + private final AtomicLong usedCounter = new AtomicLong(0); - private final AtomicLong nextId = new AtomicLong(0); - private int maxElements; // Constants ----------------------------------------------------- @@ -64,18 +70,17 @@ // Public -------------------------------------------------------- - public void setMaxElements(final int maxElements) { this.maxElements = maxElements; checkCacheSize(); } - + public int getMaxEelements() { return this.maxElements; } - + /** * @return * @see java.util.Map#size() @@ -156,11 +161,13 @@ public V put(final K key, final V value) { processQueue(); - AggregatedSoftReference refPut = mapDelegate.put(key, createReference(key, value)); + AggregatedSoftReference newRef = createReference(key, value); + AggregatedSoftReference oldRef = mapDelegate.put(key, newRef); checkCacheSize(); - if (refPut != null) + newRef.used(); + if (oldRef != null) { - return refPut.get(); + return oldRef.get(); } else { @@ -173,11 +180,11 @@ if (maxElements > 0 && mapDelegate.size() > maxElements) { TreeSet<AggregatedSoftReference> usedReferences = new TreeSet<AggregatedSoftReference>(new ComparatorAgregated()); - + for (AggregatedSoftReference ref : mapDelegate.values()) { V v = ref.get(); - + if (v != null && !v.isLive()) { usedReferences.add(ref); @@ -186,11 +193,19 @@ for (AggregatedSoftReference ref : usedReferences) { - mapDelegate.remove(ref.key); - - if (mapDelegate.size() <= maxElements) + if (ref.used > 0) { - break; + Object removed = mapDelegate.remove(ref.key); + + if (isTrace) + { + log.trace("Removing " + removed + " with id = " + ref.key + " from SoftValueHashMap"); + } + + if (mapDelegate.size() <= maxElements) + { + break; + } } } } @@ -210,14 +225,14 @@ { return -1; } - - k = o1.id - o2.id; - + + k = o1.hashCode() - o2.hashCode(); + if (k > 0) { return 1; } - else if (k < 0) + else if (k < 0) { return -1; } @@ -369,8 +384,6 @@ { final K key; - long id = nextId.incrementAndGet(); - long used = 0; public long getUsed() @@ -380,7 +393,7 @@ public void used() { - used++; + used = usedCounter.incrementAndGet(); } public AggregatedSoftReference(final K key, final V referent) @@ -388,6 +401,17 @@ super(referent, refQueue); this.key = key; } + + /* (non-Javadoc) + * @see java.lang.Object#toString() + */ + @Override + public String toString() + { + return "AggregatedSoftReference [key=" + key + ", used=" + used + "]"; + } + + } static final class EntryElement<K, V> implements Map.Entry<K, V> Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/client/PagingTest.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/client/PagingTest.java 2011-05-27 15:41:56 UTC (rev 10750) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/client/PagingTest.java 2011-05-28 05:06:01 UTC (rev 10751) @@ -1772,7 +1772,7 @@ } } } - + public void testOrderingNonTX() throws Exception { clearData(); @@ -1792,7 +1792,7 @@ final AtomicInteger errors = new AtomicInteger(0); - final int messageSize = 1024; // 1k + final int messageSize = 1024; final int numberOfMessages = 2000; try @@ -1836,7 +1836,7 @@ sessionProducer.commit(); - System.out.println("Producer gone"); + log.info("Producer gone"); } catch (Throwable e) @@ -1876,8 +1876,12 @@ { ClientMessage msg = consumer.receive(5000); assertNotNull(msg); - System.out.println("Received " + i); - assertEquals(i, msg.getIntProperty("count").intValue()); + log.info("Received " + i + " with property = " + msg.getIntProperty("count")); + if (i != msg.getIntProperty("count").intValue()) + { + log.info("###### different"); + } + //assertEquals(i, msg.getIntProperty("count").intValue()); msg.acknowledge(); } @@ -3529,7 +3533,7 @@ } } } - + public void testDLAOnLargeMessageAndPaging() throws Exception { clearData(); @@ -3550,17 +3554,20 @@ final int messageSize = 1024; + ServerLocator locator = null; + ClientSessionFactory sf = null; + ClientSession session = null; try { - ServerLocator locator = createInVMNonHALocator(); + locator = createInVMNonHALocator(); locator.setBlockOnNonDurableSend(true); locator.setBlockOnDurableSend(true); locator.setBlockOnAcknowledge(true); - ClientSessionFactory sf = locator.createSessionFactory(); + sf = locator.createSessionFactory(); - ClientSession session = sf.createSession(false, false, false); + session = sf.createSession(false, false, false); session.createQueue(ADDRESS, ADDRESS, true); @@ -3572,15 +3579,14 @@ ClientProducer producer = session.createProducer(PagingTest.ADDRESS); - ClientMessage message = null; for (int i = 0; i < 100; i++) { - log.info("send message #" + i); - message = session.createMessage(true); + log.debug("send message #" + i); + ClientMessage message = session.createMessage(true); message.putStringProperty("id", "str" + i); - + message.setBodyInputStream(createFakeLargeStream(messageSize)); producer.send(message); @@ -3596,14 +3602,12 @@ session.start(); ClientConsumer cons = session.createConsumer(ADDRESS); - - ClientMessage msg = null; for (int msgNr = 0 ; msgNr < 2; msgNr++) { for (int i = 0 ; i < 5; i++) { - msg = cons.receive(5000); + ClientMessage msg = cons.receive(5000); assertNotNull(msg); @@ -3624,12 +3628,12 @@ for (int i = 2; i < 100; i++) { - log.info("Received message " + i); - message = cons.receive(5000); - assertNotNull(message); + log.debug("Received message " + i); + ClientMessage message = cons.receive(5000); + assertNotNull("Message " + i + " wasn't received", message); message.acknowledge(); - - message.saveToOutputStream(new OutputStream() + + message.setOutputStream(new OutputStream() { @Override public void write(int b) throws IOException @@ -3638,6 +3642,12 @@ } }); + if (!message.waitOutputStreamCompletion(5000)) + { + log.info(threadDump("dump")); + fail("Couldn't finish large message sending"); + } + } assertNull(cons.receiveImmediate()); @@ -3646,6 +3656,8 @@ sf.close(); + session.close(); + locator.close(); server.stop(); @@ -3664,12 +3676,15 @@ for (int i = 2; i < 100; i++) { - log.info("Received message " + i); - message = cons.receive(5000); + log.debug("Received message " + i); + ClientMessage message = cons.receive(5000); assertNotNull(message); + + assertEquals("str" + i, message.getStringProperty("id")); + message.acknowledge(); - message.saveToOutputStream(new OutputStream() + message.setOutputStream(new OutputStream() { @Override public void write(int b) throws IOException @@ -3677,6 +3692,8 @@ } }); + + assertTrue(message.waitOutputStreamCompletion(5000)); } cons.close(); @@ -3685,7 +3702,7 @@ for (int msgNr = 0 ; msgNr < 2; msgNr++) { - msg = cons.receive(5000); + ClientMessage msg = cons.receive(10000); assertNotNull(msg); @@ -3723,11 +3740,11 @@ assertFalse(pgStoreAddress.isPaging()); session.commit(); - - session.close(); } finally { + session.close(); + sf.close(); locator.close(); try { Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/util/SoftValueMapTest.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/util/SoftValueMapTest.java 2011-05-27 15:41:56 UTC (rev 10750) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/util/SoftValueMapTest.java 2011-05-28 05:06:01 UTC (rev 10751) @@ -44,24 +44,6 @@ // each buffer will be 1/10th of the maxMemory int bufferSize = (int)(maxMemory / 100); - class Value implements SoftValueHashMap.ValueCache - { - byte[] payload; - - Value(byte[] payload) - { - this.payload = payload; - } - - /* (non-Javadoc) - * @see org.hornetq.utils.SoftValueHashMap.ValueCache#isLive() - */ - public boolean isLive() - { - return false; - } - } - SoftValueHashMap<Long, Value> softCache = new SoftValueHashMap<Long, Value>(100); final int MAX_ELEMENTS = 1000; @@ -83,31 +65,6 @@ { forceGC(); - class Value implements SoftValueHashMap.ValueCache - { - byte[] payload; - - boolean live; - - Value(byte[] payload) - { - this.payload = payload; - } - - /* (non-Javadoc) - * @see org.hornetq.utils.SoftValueHashMap.ValueCache#isLive() - */ - public boolean isLive() - { - return live; - } - - public void setLive(boolean live) - { - this.live = live; - } - } - SoftValueHashMap<Long, Value> softCache = new SoftValueHashMap<Long, Value>(200); for (long i = 0 ; i < 100; i++) @@ -144,7 +101,54 @@ System.out.println("Soft cache has " + softCache.size() + " elements"); } + + public void testEvictOldestElement() + { + Value one = new Value(new byte[100]); + Value two = new Value(new byte[100]); + Value three = new Value(new byte[100]); + + + SoftValueHashMap<Integer, Value> softCache = new SoftValueHashMap<Integer, Value>(2); + softCache.put(3, three); + softCache.put(2, two); + softCache.put(1, one); + + assertNull(softCache.get(3)); + assertEquals(two, softCache.get(2)); + assertEquals(one, softCache.get(1)); + + + + } + + class Value implements SoftValueHashMap.ValueCache + { + byte[] payload; + + boolean live; + Value(byte[] payload) + { + this.payload = payload; + } + + /* (non-Javadoc) + * @see org.hornetq.utils.SoftValueHashMap.ValueCache#isLive() + */ + public boolean isLive() + { + return live; + } + + public void setLive(boolean live) + { + this.live = live; + } + } + + + // Package protected --------------------------------------------- // Protected -----------------------------------------------------