[hornetq-commits] JBoss hornetq SVN: r9905 - in trunk: src/main/org/hornetq/core/paging and 29 other directories.

Author: clebert.suconic(a)jboss.com Date: 2010-11-17 00:07:27 -0500 (Wed, 17 Nov 2010) New Revision: 9905 Added: trunk/src/main/org/hornetq/core/paging/cursor/ trunk/src/main/org/hornetq/core/paging/cursor/LivePageCache.java trunk/src/main/org/hornetq/core/paging/cursor/PageCache.java trunk/src/main/org/hornetq/core/paging/cursor/PageCursorProvider.java trunk/src/main/org/hornetq/core/paging/cursor/PagePosition.java trunk/src/main/org/hornetq/core/paging/cursor/PageSubscription.java trunk/src/main/org/hornetq/core/paging/cursor/PagedReference.java trunk/src/main/org/hornetq/core/paging/cursor/PagedReferenceImpl.java trunk/src/main/org/hornetq/core/paging/cursor/impl/ trunk/src/main/org/hornetq/core/paging/cursor/impl/LivePageCacheImpl.java trunk/src/main/org/hornetq/core/paging/cursor/impl/PageCacheImpl.java trunk/src/main/org/hornetq/core/paging/cursor/impl/PageCursorProviderImpl.java trunk/src/main/org/hornetq/core/paging/cursor/impl/PagePositionImpl.java trunk/src/main/org/hornetq/core/paging/cursor/impl/PageSubscriptionImpl.java trunk/src/main/org/hornetq/core/server/RouteContextList.java trunk/src/main/org/hornetq/core/transaction/TransactionOperationAbstract.java trunk/src/main/org/hornetq/utils/SoftValueHashMap.java trunk/tests/src/org/hornetq/tests/integration/paging/PagePositionTest.java trunk/tests/src/org/hornetq/tests/stress/paging/PageCursorStressTest.java trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagePositionTest.java trunk/tests/src/org/hornetq/tests/unit/util/SoftValueMapTest.java Removed: trunk/tests/src/org/hornetq/tests/integration/paging/PageCrashTest.java Modified: trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java trunk/src/main/org/hornetq/core/paging/Page.java trunk/src/main/org/hornetq/core/paging/PageTransactionInfo.java trunk/src/main/org/hornetq/core/paging/PagedMessage.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/PagingStoreFactory.java trunk/src/main/org/hornetq/core/paging/impl/PageImpl.java trunk/src/main/org/hornetq/core/paging/impl/PageTransactionInfoImpl.java trunk/src/main/org/hornetq/core/paging/impl/PagedMessageImpl.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/persistence/StorageManager.java trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java trunk/src/main/org/hornetq/core/persistence/impl/journal/OperationContextImpl.java trunk/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java trunk/src/main/org/hornetq/core/postoffice/AddressManager.java trunk/src/main/org/hornetq/core/postoffice/Bindings.java trunk/src/main/org/hornetq/core/postoffice/BindingsFactory.java trunk/src/main/org/hornetq/core/postoffice/PostOffice.java trunk/src/main/org/hornetq/core/postoffice/impl/BindingsImpl.java trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java trunk/src/main/org/hornetq/core/postoffice/impl/SimpleAddressManager.java trunk/src/main/org/hornetq/core/postoffice/impl/WildcardAddressManager.java trunk/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java trunk/src/main/org/hornetq/core/server/MessageReference.java trunk/src/main/org/hornetq/core/server/Queue.java trunk/src/main/org/hornetq/core/server/QueueFactory.java trunk/src/main/org/hornetq/core/server/RoutingContext.java trunk/src/main/org/hornetq/core/server/ServerMessage.java trunk/src/main/org/hornetq/core/server/ServerSession.java trunk/src/main/org/hornetq/core/server/cluster/impl/RemoteQueueBindingImpl.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/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/RoutingContextImpl.java trunk/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java trunk/src/main/org/hornetq/core/transaction/TransactionPropertyIndexes.java trunk/tests/jms-tests/src/org/hornetq/jms/tests/BrowserTest.java trunk/tests/src/org/hornetq/tests/concurrent/server/impl/QueueTest.java trunk/tests/src/org/hornetq/tests/integration/client/PagingTest.java trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControl2Test.java trunk/tests/src/org/hornetq/tests/integration/persistence/DeleteMessagesOnStartupTest.java trunk/tests/src/org/hornetq/tests/integration/persistence/RestartSMTest.java trunk/tests/src/org/hornetq/tests/integration/persistence/StorageManagerTestBase.java trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java trunk/tests/src/org/hornetq/tests/integration/server/LVQTest.java trunk/tests/src/org/hornetq/tests/timing/core/server/impl/QueueImplTest.java trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PageImplTest.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 trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/DuplicateDetectionUnitTest.java trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/FakeQueue.java trunk/tests/src/org/hornetq/tests/unit/core/server/impl/fakes/FakeQueueFactory.java trunk/tests/src/org/hornetq/tests/util/RandomUtil.java trunk/tests/src/org/hornetq/tests/util/ServiceTestBase.java trunk/tests/src/org/hornetq/tests/util/UnitTestCase.java Log: https://jira.jboss.org/browse/HORNETQ-325 Page at queue through cursors, soft caches and journal persistence holding the acks Modified: trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -321,7 +321,19 @@ public ClientMessage receive(final long timeout) throws HornetQException { - return receive(timeout, false); + if (isBrowseOnly()) + { + ClientMessage msg = receive(timeout, false); + if (msg == null) + { + msg = receive(0, true); + } + return msg; + } + else + { + return receive(timeout, false); + } } public ClientMessage receive() throws HornetQException Modified: trunk/src/main/org/hornetq/core/paging/Page.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/Page.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/paging/Page.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -15,6 +15,8 @@ import java.util.List; +import org.hornetq.core.paging.cursor.LivePageCache; + /** * * @see PagingManager @@ -28,6 +30,8 @@ void write(PagedMessage message) throws Exception; List<PagedMessage> read() throws Exception; + + void setLiveCache(LivePageCache pageCache); int getSize(); Modified: trunk/src/main/org/hornetq/core/paging/PageTransactionInfo.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/PageTransactionInfo.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/paging/PageTransactionInfo.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -14,6 +14,8 @@ package org.hornetq.core.paging; import org.hornetq.core.journal.EncodingSupport; +import org.hornetq.core.paging.cursor.PageSubscription; +import org.hornetq.core.paging.cursor.PagePosition; import org.hornetq.core.persistence.StorageManager; import org.hornetq.core.transaction.Transaction; @@ -24,8 +26,6 @@ */ public interface PageTransactionInfo extends EncodingSupport { - boolean waitCompletion(int timeoutMilliSeconds) throws Exception; - boolean isCommit(); boolean isRollback(); @@ -42,14 +42,25 @@ void store(StorageManager storageManager, PagingManager pagingManager, Transaction tx) throws Exception; - void storeUpdate(StorageManager storageManager, PagingManager pagingManager, Transaction tx, int depages) throws Exception; + void storeUpdate(StorageManager storageManager, PagingManager pagingManager, Transaction tx) throws Exception; + + void storeUpdate(StorageManager storageManager, PagingManager pagingManager) throws Exception; // To be used after the update was stored or reload - void update(int update, StorageManager storageManager, PagingManager pagingManager); + void onUpdate(int update, StorageManager storageManager, PagingManager pagingManager); void increment(); + + void increment(int size); int getNumberOfMessages(); - void markIncomplete(); + /** + * This method will hold the position to be delivered later in case this transaction is pending. + * If the tx is not pending, it will return false, so the caller can deliver it right away + * @param cursor + * @param cursorPos + * @return true if the message will be delivered later, false if it should be delivered right away + */ + boolean deliverAfterCommit(PageSubscription cursor, PagePosition cursorPos); } Modified: trunk/src/main/org/hornetq/core/paging/PagedMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/PagedMessage.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/paging/PagedMessage.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -28,7 +28,12 @@ */ public interface PagedMessage extends EncodingSupport { - ServerMessage getMessage(StorageManager storageManager); + ServerMessage getMessage(); + + /** The queues that were routed during paging */ + long[] getQueueIDs(); + + void initMessage(StorageManager storageManager); long getTransactionID(); } Modified: trunk/src/main/org/hornetq/core/paging/PagingManager.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/PagingManager.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/paging/PagingManager.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -81,4 +81,6 @@ SimpleString[] getStoreNames(); void deletePageStore(SimpleString storeName) throws Exception; + + void processReload() throws Exception; } Modified: trunk/src/main/org/hornetq/core/paging/PagingStore.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/PagingStore.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/paging/PagingStore.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -13,10 +13,11 @@ package org.hornetq.core.paging; -import java.util.List; - import org.hornetq.api.core.SimpleString; +import org.hornetq.core.paging.cursor.PageCursorProvider; import org.hornetq.core.server.HornetQComponent; +import org.hornetq.core.server.RouteContextList; +import org.hornetq.core.server.RoutingContext; import org.hornetq.core.server.ServerMessage; import org.hornetq.core.settings.impl.AddressFullMessagePolicy; @@ -36,10 +37,17 @@ SimpleString getAddress(); int getNumberOfPages(); + + // The current page in which the system is writing files + int getCurrentWritingPage(); SimpleString getStoreName(); AddressFullMessagePolicy getAddressFullMessagePolicy(); + + long getFirstPage(); + + long getTopPage(); long getPageSizeBytes(); @@ -51,19 +59,61 @@ void sync() throws Exception; - boolean page(List<ServerMessage> messages, long transactionId) throws Exception; + boolean page(ServerMessage message, RoutingContext ctx) throws Exception; - boolean page(ServerMessage message) throws Exception; + boolean page(ServerMessage message, RoutingContext ctx, RouteContextList listCtx) throws Exception; Page createPage(final int page) throws Exception; + + PagingManager getPagingManager(); + + PageCursorProvider getCursorProvier(); + + void processReload() throws Exception; + + /** + * Remove the first page from the Writing Queue. + * The file will still exist until Page.delete is called, + * So, case the system is reloaded the same Page will be loaded back if delete is not called. + * + * @throws Exception + * + * Note: This should still be part of the interface, even though HornetQ only uses through the + */ + Page depage() throws Exception; + + void forceAnotherPage() throws Exception; + + Page getCurrentPage(); + + /** * @return false if a thread was already started, or if not in page mode * @throws Exception */ boolean startDepaging(); + + /** @return true if paging was started, or false if paging was already started before this call */ + boolean startPaging() throws Exception; + void stopPaging() throws Exception; + void addSize(int size); void executeRunnableWhenMemoryAvailable(Runnable runnable); + + /** This method will hold and producer, but it wait operations to finish before locking (write lock) */ + void lock(); + + /** + * + * Call this method using the same thread used by the last call of {@link PagingStore#lock()} + * + */ + void unlock(); + + /** This is used mostly by tests. + * We will wait any pending runnable to finish its execution */ + void flushExecutors(); } Modified: trunk/src/main/org/hornetq/core/paging/PagingStoreFactory.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/PagingStoreFactory.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/paging/PagingStoreFactory.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -30,7 +30,7 @@ */ public interface PagingStoreFactory { - PagingStore newStore(SimpleString address, AddressSettings addressSettings) throws Exception; + PagingStore newStore(SimpleString address, AddressSettings addressSettings); void stop() throws InterruptedException; Added: trunk/src/main/org/hornetq/core/paging/cursor/LivePageCache.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/cursor/LivePageCache.java (rev 0) +++ trunk/src/main/org/hornetq/core/paging/cursor/LivePageCache.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -0,0 +1,29 @@ +/* + * Copyright 2010 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.paging.cursor; + +import org.hornetq.core.paging.PagedMessage; + +/** + * A LivePageCache + * + * @author clebertsuconic + * + * + */ +public interface LivePageCache extends PageCache +{ + + void addLiveMessage(PagedMessage message); +} Added: trunk/src/main/org/hornetq/core/paging/cursor/PageCache.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/cursor/PageCache.java (rev 0) +++ trunk/src/main/org/hornetq/core/paging/cursor/PageCache.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -0,0 +1,62 @@ +/* + * Copyright 2010 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.paging.cursor; + +import org.hornetq.core.paging.Page; +import org.hornetq.core.paging.PagedMessage; + +/** + * A PageCache + * + * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * + */ +public interface PageCache +{ + Page getPage(); + + long getPageId(); + + int getNumberOfMessages(); + + void setMessages(PagedMessage[] messages); + + /** + * If this cache is still being updated + * @return + */ + boolean isLive(); + + /** + * + * @param messageNumber The order of the message on the page + * @return + */ + PagedMessage getMessage(int messageNumber); + + /** + * When the cache is being created, + * We need to first read the files before other threads can get messages from this. + */ + void lock(); + + /** + * You have to call this method within the same thread you called lock + */ + void unlock(); + + void close(); + +} Added: trunk/src/main/org/hornetq/core/paging/cursor/PageCursorProvider.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/cursor/PageCursorProvider.java (rev 0) +++ trunk/src/main/org/hornetq/core/paging/cursor/PageCursorProvider.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -0,0 +1,89 @@ +/* + * Copyright 2010 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.paging.cursor; + +import org.hornetq.core.filter.Filter; +import org.hornetq.core.paging.PagedMessage; +import org.hornetq.core.paging.PagingStore; + +/** + * The provider of Cursor for a given Address + * + * @author <a href="mailto:clebert.suconic@jboss.com">Clebert Suconic</a> + * + * + */ +public interface PageCursorProvider +{ + + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + + PageCache getPageCache(PagePosition pos); + + PagedReference newReference(final PagePosition pos, final PagedMessage msg, PageSubscription sub); + + void addPageCache(PageCache cache); + + PagingStore getAssociatedStore(); + + /** + * + * @param queueId The cursorID should be the same as the queueId associated for persistance + * @return + */ + PageSubscription getSubscription(long queueId); + + PageSubscription createSubscription(long queueId, Filter filter, boolean durable); + + PagedMessage getMessage(PagePosition pos) throws Exception; + + void processReload() throws Exception; + + void stop(); + + void flushExecutors(); + + void scheduleCleanup(); + + // Perform the cleanup at the caller's thread (for startup and recovery) + void cleanup(); + + /** + * @param pageCursorImpl + */ + void close(PageSubscription pageCursorImpl); + + // to be used on tests ------------------------------------------- + + int getCacheSize(); + + void printDebug(); + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- + +} Added: trunk/src/main/org/hornetq/core/paging/cursor/PagePosition.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/cursor/PagePosition.java (rev 0) +++ trunk/src/main/org/hornetq/core/paging/cursor/PagePosition.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -0,0 +1,54 @@ +/* + * Copyright 2010 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.paging.cursor; + + + +/** + * A PagePosition + * + * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * + */ +public interface PagePosition extends Comparable<PagePosition> +{ + + // The recordID associated during ack + long getRecordID(); + + // The recordID associated during ack + void setRecordID(long recordID); + + long getPageNr(); + + int getMessageNr(); + + void setPageCache(PageCache pageCache); + + /** + * PagePosition will hold the page with a weak reference. + * So, this could be eventually null case soft-cache was released + * @return + */ + PageCache getPageCache(); + + PagePosition nextMessage(); + + PagePosition nextPage(); + + /** This will just test if the current position is the immediate next to the parameter position */ + boolean isRightAfter(PagePosition previous); + +} Added: trunk/src/main/org/hornetq/core/paging/cursor/PageSubscription.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/cursor/PageSubscription.java (rev 0) +++ trunk/src/main/org/hornetq/core/paging/cursor/PageSubscription.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -0,0 +1,122 @@ +/* + * Copyright 2010 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.paging.cursor; + +import org.hornetq.core.paging.PagedMessage; +import org.hornetq.core.server.Queue; +import org.hornetq.core.transaction.Transaction; +import org.hornetq.utils.LinkedListIterator; + +/** + * A PageCursor + * + * @author <a href="mailto:clebert.suconic@jboss.com">Clebert Suconic</a> + * + * + */ +public interface PageSubscription +{ + + // Cursor query operations -------------------------------------- + + // To be called before the server is down + void stop(); + + void bookmark(PagePosition position) throws Exception; + + long getId(); + + boolean isPersistent(); + + /** Used as a delegate method to pageStore.isPaging() */ + boolean isPaging(); + + public LinkedListIterator<PagedReference> iterator(); + + // To be called when the cursor is closed for good. Most likely when the queue is deleted + void close() throws Exception; + + void scheduleCleanupCheck(); + + void cleanupEntries() throws Exception; + + void disableAutoCleanup(); + + void enableAutoCleanup(); + + void ack(PagedReference ref) throws Exception; + + // for internal (cursor) classes + void ack(PagePosition ref) throws Exception; + + void ackTx(Transaction tx, PagedReference position) throws Exception; + + // for internal (cursor) classes + void ackTx(Transaction tx, PagePosition position) throws Exception; + + /** + * + * @return the first page in use or MAX_LONG if none is in use + */ + long getFirstPage(); + + // Reload operations + + /** + * @param position + */ + void reloadACK(PagePosition position); + + /** + * To be called when the cursor decided to ignore a position. + * @param position + */ + void positionIgnored(PagePosition position); + + /** + * To be used to avoid a redelivery of a prepared ACK after load + * @param position + */ + void reloadPreparedACK(Transaction tx, PagePosition position); + + void processReload() throws Exception; + + /** + * To be used on redeliveries + * @param position + */ + void redeliver(PagePosition position); + + void printDebug(); + + /** + * @param minPage + * @return + */ + boolean isComplete(long page); + + /** wait all the scheduled runnables to finish their current execution */ + void flushExecutors(); + + void setQueue(Queue queue); + + Queue getQueue(); + + /** + * To be used to requery the reference case the Garbage Collection removed it from the PagedReference as it's using WeakReferences + * @param pos + * @return + */ + PagedMessage queryMessage(PagePosition pos); +} Added: trunk/src/main/org/hornetq/core/paging/cursor/PagedReference.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/cursor/PagedReference.java (rev 0) +++ trunk/src/main/org/hornetq/core/paging/cursor/PagedReference.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -0,0 +1,31 @@ +/* + * Copyright 2010 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.paging.cursor; + +import org.hornetq.core.paging.PagedMessage; +import org.hornetq.core.server.MessageReference; + +/** + * A PagedReference + * + * @author clebert + * + * + */ +public interface PagedReference extends MessageReference +{ + PagePosition getPosition(); + + PagedMessage getPagedMessage(); +} Added: trunk/src/main/org/hornetq/core/paging/cursor/PagedReferenceImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/cursor/PagedReferenceImpl.java (rev 0) +++ trunk/src/main/org/hornetq/core/paging/cursor/PagedReferenceImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -0,0 +1,188 @@ +/* + * Copyright 2010 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.paging.cursor; + +import java.lang.ref.WeakReference; + +import org.hornetq.api.core.Message; +import org.hornetq.core.paging.PagedMessage; +import org.hornetq.core.server.MessageReference; +import org.hornetq.core.server.Queue; +import org.hornetq.core.server.ServerMessage; +import org.hornetq.core.transaction.Transaction; + +/** + * A InternalReference + * + * @author clebert + * + * + */ +public class PagedReferenceImpl implements PagedReference +{ + + private static final long serialVersionUID = -8640232251318264710L; + + private final PagePosition position; + + private WeakReference<PagedMessage> message; + + private Long deliveryTime = null; + + private final PageSubscription subscription; + + public ServerMessage getMessage() + { + return getPagedMessage().getMessage(); + } + + public synchronized PagedMessage getPagedMessage() + { + PagedMessage returnMessage = message.get(); + + // We only keep a few references on the Queue from paging... + // Besides those references are SoftReferenced on page cache... + // So, this will unlikely be null, + // unless the Queue has stalled for some time after paging + if (returnMessage == null) + { + // reference is gone, we will reconstruct it + returnMessage = subscription.queryMessage(position); + message = new WeakReference<PagedMessage>(returnMessage); + } + return returnMessage; + } + + public PagePosition getPosition() + { + return position; + } + + public PagedReferenceImpl(final PagePosition position, final PagedMessage message, final PageSubscription subscription) + { + this.position = position; + this.message = new WeakReference<PagedMessage>(message); + this.subscription = subscription; + } + + public boolean isPaged() + { + return true; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.MessageReference#copy(org.hornetq.core.server.Queue) + */ + public MessageReference copy(final Queue queue) + { + // TODO Auto-generated method stub + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.MessageReference#getScheduledDeliveryTime() + */ + public long getScheduledDeliveryTime() + { + if (deliveryTime == null) + { + ServerMessage msg = getMessage(); + if (msg.containsProperty(Message.HDR_SCHEDULED_DELIVERY_TIME)) + { + deliveryTime = getMessage().getLongProperty(Message.HDR_SCHEDULED_DELIVERY_TIME); + } + else + { + deliveryTime = 0l; + } + } + return deliveryTime; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.MessageReference#setScheduledDeliveryTime(long) + */ + public void setScheduledDeliveryTime(final long scheduledDeliveryTime) + { + deliveryTime = scheduledDeliveryTime; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.MessageReference#getDeliveryCount() + */ + public int getDeliveryCount() + { + // TODO Auto-generated method stub + return 0; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.MessageReference#setDeliveryCount(int) + */ + public void setDeliveryCount(final int deliveryCount) + { + // TODO Auto-generated method stub + + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.MessageReference#incrementDeliveryCount() + */ + public void incrementDeliveryCount() + { + // TODO Auto-generated method stub + + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.MessageReference#decrementDeliveryCount() + */ + public void decrementDeliveryCount() + { + // TODO Auto-generated method stub + + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.MessageReference#getQueue() + */ + public Queue getQueue() + { + return subscription.getQueue(); + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.MessageReference#handled() + */ + public void handled() + { + getQueue().referenceHandled(); + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.MessageReference#acknowledge() + */ + public void acknowledge() throws Exception + { + subscription.ack(this); + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.MessageReference#acknowledge(org.hornetq.core.transaction.Transaction) + */ + public void acknowledge(final Transaction tx) throws Exception + { + subscription.ackTx(tx, this); + } +} Added: trunk/src/main/org/hornetq/core/paging/cursor/impl/LivePageCacheImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/cursor/impl/LivePageCacheImpl.java (rev 0) +++ trunk/src/main/org/hornetq/core/paging/cursor/impl/LivePageCacheImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -0,0 +1,155 @@ +/* + * Copyright 2010 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.paging.cursor.impl; + +import java.util.LinkedList; +import java.util.List; + +import org.hornetq.core.paging.Page; +import org.hornetq.core.paging.PagedMessage; +import org.hornetq.core.paging.cursor.LivePageCache; + +/** + * This is the same as PageCache, however this is for the page that's being currently written. + * + * @author clebertsuconic + * + * + */ +public class LivePageCacheImpl implements LivePageCache +{ + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + private final List<PagedMessage> messages = new LinkedList<PagedMessage>(); + + private final Page page; + + private boolean isLive = true; + + public String toString() + { + return "LivePacheCacheImpl::page=" + page.getPageId() + " number of messages=" + messages.size() + " isLive = " + isLive; + } + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + public LivePageCacheImpl(final Page page) + { + this.page = page; + } + + // Public -------------------------------------------------------- + + /* (non-Javadoc) + * @see org.hornetq.core.paging.cursor.PageCache#getPage() + */ + public Page getPage() + { + return page; + } + + public long getPageId() + { + return page.getPageId(); + } + + /* (non-Javadoc) + * @see org.hornetq.core.paging.cursor.PageCache#getNumberOfMessages() + */ + public synchronized int getNumberOfMessages() + { + return messages.size(); + } + + /* (non-Javadoc) + * @see org.hornetq.core.paging.cursor.PageCache#setMessages(org.hornetq.core.server.ServerMessage[]) + */ + public synchronized void setMessages(PagedMessage[] messages) + { + // This method shouldn't be called on liveCache, but we will provide the implementation for it anyway + for (PagedMessage msg : messages) + { + addLiveMessage(msg); + } + } + + /* (non-Javadoc) + * @see org.hornetq.core.paging.cursor.PageCache#getMessage(int) + */ + public synchronized PagedMessage getMessage(int messageNumber) + { + if (messageNumber < messages.size()) + { + return messages.get(messageNumber); + } + else + { + return null; + } + } + + /* (non-Javadoc) + * @see org.hornetq.core.paging.cursor.PageCache#lock() + */ + public void lock() + { + // nothing to be done on live cache + } + + /* (non-Javadoc) + * @see org.hornetq.core.paging.cursor.PageCache#unlock() + */ + public void unlock() + { + // nothing to be done on live cache + } + + /* (non-Javadoc) + * @see org.hornetq.core.paging.cursor.PageCache#isLive() + */ + public synchronized boolean isLive() + { + return isLive; + } + + /* (non-Javadoc) + * @see org.hornetq.core.paging.cursor.LivePageCache#addLiveMessage(org.hornetq.core.server.ServerMessage) + */ + public synchronized void addLiveMessage(PagedMessage message) + { + this.messages.add(message); + } + + /* (non-Javadoc) + * @see org.hornetq.core.paging.cursor.LivePageCache#close() + */ + public synchronized void close() + { + this.isLive = false; + } + + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- + +} Added: trunk/src/main/org/hornetq/core/paging/cursor/impl/PageCacheImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/cursor/impl/PageCacheImpl.java (rev 0) +++ trunk/src/main/org/hornetq/core/paging/cursor/impl/PageCacheImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -0,0 +1,144 @@ +/* + * Copyright 2010 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.paging.cursor.impl; + +import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; + +import org.hornetq.core.paging.Page; +import org.hornetq.core.paging.PagedMessage; +import org.hornetq.core.paging.cursor.PageCache; + +/** + * The caching associated to a single page. + * + * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * + */ +public class PageCacheImpl implements PageCache +{ + + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + private final ReadWriteLock lock = new ReentrantReadWriteLock(); + + private PagedMessage[] messages; + + private final Page page; + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + public PageCacheImpl(final Page page) + { + this.page = page; + } + + // Public -------------------------------------------------------- + + /* (non-Javadoc) + * @see org.hornetq.core.paging.cursor.PageCache#getPage() + */ + public Page getPage() + { + return page; + } + + /* (non-Javadoc) + * @see org.hornetq.core.paging.cursor.PageCache#getMessage(int) + */ + public PagedMessage getMessage(final int messageNumber) + { + lock.readLock().lock(); + try + { + if (messageNumber < messages.length) + { + return messages[messageNumber]; + } + else + { + return null; + } + } + finally + { + lock.readLock().unlock(); + } + } + + public long getPageId() + { + return page.getPageId(); + } + + public void lock() + { + lock.writeLock().lock(); + } + + public void unlock() + { + lock.writeLock().unlock(); + } + + public void setMessages(final PagedMessage[] messages) + { + this.messages = messages; + } + + public int getNumberOfMessages() + { + lock.readLock().lock(); + try + { + return messages.length; + } + finally + { + lock.readLock().unlock(); + } + } + + public void close() + { + } + + /* (non-Javadoc) + * @see org.hornetq.core.paging.cursor.PageCache#isLive() + */ + public boolean isLive() + { + return false; + } + + @Override + public String toString() + { + return "PageCacheImpl::page=" + page.getPageId() + " numberOfMessages = " + messages.length; + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- + +} Added: trunk/src/main/org/hornetq/core/paging/cursor/impl/PageCursorProviderImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/cursor/impl/PageCursorProviderImpl.java (rev 0) +++ trunk/src/main/org/hornetq/core/paging/cursor/impl/PageCursorProviderImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -0,0 +1,468 @@ +/* + * Copyright 2010 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.paging.cursor.impl; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.Executor; + +import org.hornetq.core.filter.Filter; +import org.hornetq.core.logging.Logger; +import org.hornetq.core.paging.Page; +import org.hornetq.core.paging.PagedMessage; +import org.hornetq.core.paging.PagingStore; +import org.hornetq.core.paging.cursor.PageCache; +import org.hornetq.core.paging.cursor.PageCursorProvider; +import org.hornetq.core.paging.cursor.PagePosition; +import org.hornetq.core.paging.cursor.PageSubscription; +import org.hornetq.core.paging.cursor.PagedReference; +import org.hornetq.core.paging.cursor.PagedReferenceImpl; +import org.hornetq.core.persistence.StorageManager; +import org.hornetq.utils.ExecutorFactory; +import org.hornetq.utils.Future; +import org.hornetq.utils.SoftValueHashMap; +import org.jboss.netty.util.internal.ConcurrentHashMap; + +/** + * A PageProviderIMpl + * + * TODO: this may be moved entirely into PagingStore as there's an one-to-one relationship here + * However I want to keep this isolated as much as possible during development + * + * @author <a href="mailto:clebert.suconic@jboss.com">Clebert Suconic</a> + * + * + */ +public class PageCursorProviderImpl implements PageCursorProvider +{ + // Constants ----------------------------------------------------- + + private static final Logger log = Logger.getLogger(PageCursorProviderImpl.class); + + // Attributes ---------------------------------------------------- + + private final PagingStore pagingStore; + + private final StorageManager storageManager; + + private final ExecutorFactory executorFactory; + + private final Executor executor; + + private Map<Long, PageCache> softCache = new SoftValueHashMap<Long, PageCache>(); + + private ConcurrentMap<Long, PageSubscription> activeCursors = new ConcurrentHashMap<Long, PageSubscription>(); + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + public PageCursorProviderImpl(final PagingStore pagingStore, + final StorageManager storageManager, + final ExecutorFactory executorFactory) + { + this.pagingStore = pagingStore; + this.storageManager = storageManager; + this.executorFactory = executorFactory; + this.executor = executorFactory.getExecutor(); + } + + // Public -------------------------------------------------------- + + public PagingStore getAssociatedStore() + { + return pagingStore; + } + + public synchronized PageSubscription createSubscription(long cursorID, Filter filter, boolean persistent) + { + PageSubscription activeCursor = activeCursors.get(cursorID); + if (activeCursor != null) + { + throw new IllegalStateException("Cursor " + cursorID + " had already been created"); + } + + activeCursor = new PageSubscriptionImpl(this, + pagingStore, + storageManager, + executorFactory.getExecutor(), + filter, + cursorID, + persistent); + activeCursors.put(cursorID, activeCursor); + return activeCursor; + } + + /* (non-Javadoc) + * @see org.hornetq.core.paging.cursor.PageCursorProvider#createCursor() + */ + public synchronized PageSubscription getSubscription(long cursorID) + { + return activeCursors.get(cursorID); + } + + public PagedMessage getMessage(final PagePosition pos) throws Exception + { + PageCache cache = getPageCache(pos); + + if (pos.getMessageNr() >= cache.getNumberOfMessages()) + { + // sanity check, this should never happen unless there's a bug + throw new IllegalStateException("Invalid messageNumber passed = " + pos); + } + + return cache.getMessage(pos.getMessageNr()); + } + + public PagedReference newReference(final PagePosition pos, final PagedMessage msg, final PageSubscription subscription) + { + return new PagedReferenceImpl(pos, msg, subscription); + } + + /** + * No need to synchronize this method since the private getPageCache will have a synchronized call + */ + public PageCache getPageCache(PagePosition pos) + { + PageCache cache = pos.getPageCache(); + if (cache == null) + { + cache = getPageCache(pos.getPageNr()); + pos.setPageCache(cache); + } + return cache; + } + + public void addPageCache(PageCache cache) + { + synchronized (softCache) + { + softCache.put(cache.getPageId(), cache); + } + } + + public int getCacheSize() + { + synchronized (softCache) + { + return softCache.size(); + } + } + + public void processReload() throws Exception + { + for (PageSubscription cursor : this.activeCursors.values()) + { + cursor.processReload(); + } + + cleanup(); + + } + + public void stop() + { + for (PageSubscription cursor : activeCursors.values()) + { + cursor.stop(); + } + + Future future = new Future(); + + executor.execute(future); + + while (!future.await(10000)) + { + log.warn("Waiting cursor provider " + this + " to finish executors"); + } + + } + + public void flushExecutors() + { + for (PageSubscription cursor : activeCursors.values()) + { + cursor.flushExecutors(); + } + + Future future = new Future(); + + executor.execute(future); + + while (!future.await(10000)) + { + log.warn("Waiting cursor provider " + this + " to finish executors"); + } + + } + + public void close(PageSubscription cursor) + { + activeCursors.remove(cursor.getId()); + + scheduleCleanup(); + } + + /* (non-Javadoc) + * @see org.hornetq.core.paging.cursor.PageCursorProvider#scheduleCleanup() + */ + public void scheduleCleanup() + { + + executor.execute(new Runnable() + { + public void run() + { + cleanup(); + } + }); + } + + public void cleanup() + { + ArrayList<Page> depagedPages = new ArrayList<Page>(); + + pagingStore.lock(); + + synchronized (this) + { + try + { + if (!pagingStore.isStarted()) + { + return; + } + + if (pagingStore.getNumberOfPages() == 0) + { + return; + } + + ArrayList<PageSubscription> cursorList = new ArrayList<PageSubscription>(); + cursorList.addAll(activeCursors.values()); + + long minPage = checkMinPage(cursorList); + + if (minPage == pagingStore.getCurrentWritingPage() && pagingStore.getCurrentPage().getNumberOfMessages() > 0) + { + boolean complete = true; + + for (PageSubscription cursor : cursorList) + { + if (!cursor.isComplete(minPage)) + { + complete = false; + break; + } + } + + if (complete) + { + + System.out.println("Disabling depage!"); + pagingStore.forceAnotherPage(); + + Page currentPage = pagingStore.getCurrentPage(); + + try + { + // First step: Move every cursor to the next bookmarked page (that was just created) + for (PageSubscription cursor : cursorList) + { + cursor.ack(new PagePositionImpl(currentPage.getPageId(), -1)); + } + + storageManager.waitOnOperations(); + } + finally + { + for (PageSubscription cursor : cursorList) + { + cursor.enableAutoCleanup(); + } + } + + pagingStore.stopPaging(); + + // This has to be called after we stopped paging + for (PageSubscription cursor : cursorList) + { + cursor.scheduleCleanupCheck(); + } + + } + } + + for (long i = pagingStore.getFirstPage(); i < minPage; i++) + { + Page page = pagingStore.depage(); + if (page == null) + { + break; + } + depagedPages.add(page); + } + + if (pagingStore.getNumberOfPages() == 0 || pagingStore.getNumberOfPages() == 1 && + pagingStore.getCurrentPage().getNumberOfMessages() == 0) + { + pagingStore.stopPaging(); + } + } + catch (Exception ex) + { + log.warn("Couldn't complete cleanup on paging", ex); + return; + } + finally + { + pagingStore.unlock(); + } + } + + try + { + for (Page depagedPage : depagedPages) + { + depagedPage.delete(); + synchronized (softCache) + { + softCache.remove((long)depagedPage.getPageId()); + } + } + } + catch (Exception ex) + { + log.warn("Couldn't complete cleanup on paging", ex); + return; + } + + } + + public void printDebug() + { + System.out.println("Debug information for PageCursorProviderImpl:"); + for (PageCache cache : softCache.values()) + { + System.out.println("Cache " + cache); + } + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + /* Protected as we may let test cases to instrument the test */ + protected PageCacheImpl createPageCache(final long pageId) throws Exception + { + return new PageCacheImpl(pagingStore.createPage((int)pageId)); + } + + // Private ------------------------------------------------------- + + /** + * This method is synchronized because we want it to be atomic with the cursors being used + */ + private long checkMinPage(List<PageSubscription> cursorList) + { + long minPage = Long.MAX_VALUE; + + for (PageSubscription cursor : cursorList) + { + long firstPage = cursor.getFirstPage(); + if (firstPage < minPage) + { + minPage = firstPage; + } + } + + return minPage; + + } + + private PageCache getPageCache(final long pageId) + { + try + { + boolean needToRead = false; + PageCache cache = null; + synchronized (softCache) + { + if (pageId > pagingStore.getCurrentWritingPage()) + { + return null; + } + + cache = softCache.get(pageId); + if (cache == null) + { + cache = createPageCache(pageId); + needToRead = true; + // anyone reading from this cache will have to wait reading to finish first + // we also want only one thread reading this cache + cache.lock(); + softCache.put(pageId, cache); + } + } + + // Reading is done outside of the synchronized block, however + // the page stays locked until the entire reading is finished + if (needToRead) + { + Page page = null; + try + { + page = pagingStore.createPage((int)pageId); + + page.open(); + + List<PagedMessage> pgdMessages = page.read(); + + for (PagedMessage pdgMessage : pgdMessages) + { + pdgMessage.initMessage(storageManager); + } + + cache.setMessages(pgdMessages.toArray(new PagedMessage[pgdMessages.size()])); + + } + finally + { + try + { + if (page != null) + { + page.close(); + } + } + catch (Throwable ignored) + { + } + cache.unlock(); + } + } + + return cache; + } + catch (Exception e) + { + throw new RuntimeException("Couldn't complete paging due to an IO Exception on Paging - " + e.getMessage(), e); + } + } + + // Inner classes ------------------------------------------------- + +} Added: trunk/src/main/org/hornetq/core/paging/cursor/impl/PagePositionImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/cursor/impl/PagePositionImpl.java (rev 0) +++ trunk/src/main/org/hornetq/core/paging/cursor/impl/PagePositionImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -0,0 +1,210 @@ +/* + * Copyright 2010 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.paging.cursor.impl; + +import java.lang.ref.WeakReference; + +import org.hornetq.core.paging.cursor.PageCache; +import org.hornetq.core.paging.cursor.PagePosition; + +/** + * A PagePosition + * + * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * + */ +public class PagePositionImpl implements PagePosition +{ + private long pageNr; + + private int messageNr; + + /** ID used for storage */ + private long recordID; + + private volatile WeakReference<PageCache> cacheReference; + + /** + * @param pageNr + * @param messageNr + */ + public PagePositionImpl(long pageNr, int messageNr) + { + super(); + this.pageNr = pageNr; + this.messageNr = messageNr; + } + + public PagePositionImpl(long pageNr, int messageNr, PageCache pageCache) + { + this(pageNr, messageNr); + this.setPageCache(pageCache); + } + + /** + * @param pageNr + * @param messageNr + */ + public PagePositionImpl() + { + + } + + /** + * The cached page associaed with this position + * @return + */ + public PageCache getPageCache() + { + if (cacheReference == null) + { + return null; + } + else + { + return cacheReference.get(); + } + } + + public void setPageCache(final PageCache cache) + { + if (cache != null) + { + this.cacheReference = new WeakReference<PageCache>(cache); + } + } + + + /** + * @return the recordID + */ + public long getRecordID() + { + return recordID; + } + + /** + * @param recordID the recordID to set + */ + public void setRecordID(long recordID) + { + this.recordID = recordID; + } + + /** + * @return the pageNr + */ + public long getPageNr() + { + return pageNr; + } + + /** + * @return the messageNr + */ + public int getMessageNr() + { + return messageNr; + } + + public boolean isRightAfter(final PagePosition previous) + { + return this.pageNr == previous.getPageNr() && this.messageNr == previous.getMessageNr() + 1; + } + + /* (non-Javadoc) + * @see java.lang.Comparable#compareTo(java.lang.Object) + */ + public int compareTo(PagePosition o) + { + if (pageNr > o.getPageNr()) + { + return 1; + } + else if (pageNr < o.getPageNr()) + { + return -1; + } + else if (recordID > o.getRecordID()) + { + return 1; + } + else if (recordID < o.getRecordID()) + { + return -1; + } + else + { + return 0; + } + } + + public PagePosition nextMessage() + { + return new PagePositionImpl(this.pageNr, this.messageNr + 1, this.getPageCache()); + } + + public PagePosition nextPage() + { + return new PagePositionImpl(this.pageNr + 1, 0); + } + + public boolean isNextSequenceOf(PagePosition pos) + { + return this.pageNr == pos.getPageNr() && this.getRecordID() - pos.getRecordID() == 1; + } + + /* (non-Javadoc) + * @see java.lang.Object#hashCode() + */ + @Override + public int hashCode() + { + final int prime = 31; + int result = 1; + result = prime * result + messageNr; + result = prime * result + (int)(pageNr ^ (pageNr >>> 32)); + return result; + } + + /* (non-Javadoc) + * @see java.lang.Object#equals(java.lang.Object) + */ + @Override + public boolean equals(Object obj) + { + if (this == obj) + return true; + if (obj == null) + return false; + if (getClass() != obj.getClass()) + return false; + PagePositionImpl other = (PagePositionImpl)obj; + if (messageNr != other.messageNr) + return false; + if (pageNr != other.pageNr) + return false; + return true; + } + + @Override + public String toString() + { + return "PagePositionImpl [pageNr=" + pageNr + ", messageNr=" + messageNr + ", recordID=" + recordID + "]"; + } + + + +} Added: trunk/src/main/org/hornetq/core/paging/cursor/impl/PageSubscriptionImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/cursor/impl/PageSubscriptionImpl.java (rev 0) +++ trunk/src/main/org/hornetq/core/paging/cursor/impl/PageSubscriptionImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -0,0 +1,1201 @@ +/* + * Copyright 2010 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.paging.cursor.impl; + +import java.lang.ref.WeakReference; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Set; +import java.util.SortedMap; +import java.util.TreeMap; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.Executor; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import org.hornetq.core.filter.Filter; +import org.hornetq.core.journal.IOAsyncTask; +import org.hornetq.core.logging.Logger; +import org.hornetq.core.paging.PageTransactionInfo; +import org.hornetq.core.paging.PagedMessage; +import org.hornetq.core.paging.PagingStore; +import org.hornetq.core.paging.cursor.PageCache; +import org.hornetq.core.paging.cursor.PageCursorProvider; +import org.hornetq.core.paging.cursor.PagePosition; +import org.hornetq.core.paging.cursor.PageSubscription; +import org.hornetq.core.paging.cursor.PagedReference; +import org.hornetq.core.persistence.StorageManager; +import org.hornetq.core.server.MessageReference; +import org.hornetq.core.server.Queue; +import org.hornetq.core.server.ServerMessage; +import org.hornetq.core.transaction.Transaction; +import org.hornetq.core.transaction.TransactionOperationAbstract; +import org.hornetq.core.transaction.TransactionPropertyIndexes; +import org.hornetq.core.transaction.impl.TransactionImpl; +import org.hornetq.utils.ConcurrentHashSet; +import org.hornetq.utils.Future; +import org.hornetq.utils.LinkedListIterator; + +/** + * A PageCursorImpl + * + * A page cursor will always store its + * @author <a href="mailto:clebert.suconic@jboss.com">Clebert Suconic</a> + * + * + */ +public class PageSubscriptionImpl implements PageSubscription +{ + // Constants ----------------------------------------------------- + private static final Logger log = Logger.getLogger(PageSubscriptionImpl.class); + + // Attributes ---------------------------------------------------- + + private final boolean isTrace = false; // PageCursorImpl.log.isTraceEnabled(); + + private static void trace(final String message) + { + // PageCursorImpl.log.info(message); + System.out.println(message); + } + + private volatile boolean autoCleanup = true; + + private final StorageManager store; + + private final long cursorId; + + private Queue queue; + + private final boolean persistent; + + private final Filter filter; + + private final PagingStore pageStore; + + private final PageCursorProvider cursorProvider; + + private final Executor executor; + + private volatile PagePosition lastAckedPosition; + + private List<PagePosition> recoveredACK; + + private final SortedMap<Long, PageCursorInfo> consumedPages = Collections.synchronizedSortedMap(new TreeMap<Long, PageCursorInfo>()); + + // We only store the position for redeliveries. They will be read from the SoftCache again during delivery. + private final ConcurrentLinkedQueue<PagePosition> redeliveries = new ConcurrentLinkedQueue<PagePosition>(); + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + public PageSubscriptionImpl(final PageCursorProvider cursorProvider, + final PagingStore pageStore, + final StorageManager store, + final Executor executor, + final Filter filter, + final long cursorId, + final boolean persistent) + { + this.pageStore = pageStore; + this.store = store; + this.cursorProvider = cursorProvider; + this.cursorId = cursorId; + this.executor = executor; + this.filter = filter; + this.persistent = persistent; + } + + // Public -------------------------------------------------------- + + public Queue getQueue() + { + return queue; + } + + public boolean isPaging() + { + return pageStore.isPaging(); + } + + public void setQueue(Queue queue) + { + this.queue = queue; + } + + public void disableAutoCleanup() + { + autoCleanup = false; + } + + public void enableAutoCleanup() + { + autoCleanup = true; + } + + public PageCursorProvider getProvider() + { + return cursorProvider; + } + + public void bookmark(PagePosition position) throws Exception + { + PageCursorInfo cursorInfo = getPageInfo(position); + + if (position.getMessageNr() > 0) + { + cursorInfo.confirmed.addAndGet(position.getMessageNr()); + } + + ack(position); + } + + public void scheduleCleanupCheck() + { + if (autoCleanup) + { + executor.execute(new Runnable() + { + + public void run() + { + try + { + cleanupEntries(); + } + catch (Exception e) + { + PageSubscriptionImpl.log.warn("Error on cleaning up cursor pages", e); + } + } + }); + } + } + + /** + * It will cleanup all the records for completed pages + * */ + public void cleanupEntries() throws Exception + { + Transaction tx = new TransactionImpl(store); + + boolean persist = false; + + final ArrayList<PageCursorInfo> completedPages = new ArrayList<PageCursorInfo>(); + + // First get the completed pages using a lock + synchronized (this) + { + for (Entry<Long, PageCursorInfo> entry : consumedPages.entrySet()) + { + PageCursorInfo info = entry.getValue(); + if (info.isDone() && !info.isPendingDelete() && lastAckedPosition != null) + { + if (entry.getKey() == lastAckedPosition.getPageNr()) + { + // PageSubscriptionImpl.trace("We can't clear page " + entry.getKey() + + // " now since it's the current page"); + } + else + { + info.setPendingDelete(); + completedPages.add(entry.getValue()); + } + } + } + } + + for (int i = 0; i < completedPages.size(); i++) + { + PageCursorInfo info = completedPages.get(i); + + for (PagePosition pos : info.acks) + { + if (pos.getRecordID() > 0) + { + store.deleteCursorAcknowledgeTransactional(tx.getID(), pos.getRecordID()); + if (!persist) + { + // only need to set it once + tx.setContainsPersistent(); + persist = true; + } + } + } + } + + tx.addOperation(new TransactionOperationAbstract() + { + + @Override + public void afterCommit(final Transaction tx) + { + executor.execute(new Runnable() + { + + public void run() + { + synchronized (PageSubscriptionImpl.this) + { + for (PageCursorInfo completePage : completedPages) + { + if (isTrace) + { + PageSubscriptionImpl.trace("Removing page " + completePage.getPageId()); + } + if (consumedPages.remove(completePage.getPageId()) == null) + { + PageSubscriptionImpl.log.warn("Couldn't remove page " + completePage.getPageId() + + " from consumed pages on cursor for address " + + pageStore.getAddress()); + } + } + } + + cursorProvider.scheduleCleanup(); + } + }); + } + }); + + tx.commit(); + + } + + private PagedReference getReference(PagePosition pos) throws Exception + { + return cursorProvider.newReference(pos, cursorProvider.getMessage(pos), this); + } + + /* (non-Javadoc) + * @see org.hornetq.core.paging.cursor.PageCursor#iterator() + */ + public LinkedListIterator<PagedReference> iterator() + { + return new CursorIterator(); + } + + private PagedReference internalGetNext(final PagePosition pos) + { + PagePosition retPos = pos.nextMessage(); + + PageCache cache = cursorProvider.getPageCache(pos); + + if (cache == null) + { + return null; + } + + if (!cache.isLive() && retPos.getMessageNr() >= cache.getNumberOfMessages()) + { + retPos = pos.nextPage(); + + cache = cursorProvider.getPageCache(retPos); + + if (cache == null) + { + return null; + } + + if (retPos.getMessageNr() >= cache.getNumberOfMessages()) + { + return null; + } + } + + PagedMessage serverMessage = cache.getMessage(retPos.getMessageNr()); + + if (serverMessage != null) + { + return cursorProvider.newReference(retPos, serverMessage, this); + } + else + { + return null; + } + } + + private boolean routed(PagedMessage message) + { + long id = getId(); + + for (long qid : message.getQueueIDs()) + { + if (qid == id) + { + return true; + } + } + return false; + } + + /** + * + */ + private synchronized PagePosition getStartPosition() + { + // Get the first page not marked for deletion + // It's important to verify if it's not marked for deletion as you may have a pending request on the queue + for (Map.Entry<Long, PageCursorInfo> entry : consumedPages.entrySet()) + { + if (!entry.getValue().isPendingDelete()) + { + if (entry.getValue().acks.isEmpty()) + { + return new PagePositionImpl(entry.getKey(), -1); + } + else + { + // The list is not ordered... + // This is only done at creation of the queue, so we just scan instead of keeping the list ordened + PagePosition retValue = null; + + for (PagePosition pos : entry.getValue().acks) + { + System.out.println("Analizing " + pos); + if (retValue == null || retValue.getMessageNr() > pos.getMessageNr()) + { + retValue = pos; + } + } + + System.out.println("Returning initial position " + retValue); + + return retValue; + } + } + } + + return new PagePositionImpl(pageStore.getFirstPage(), -1); + } + + public void ackTx(final Transaction tx, final PagePosition position) throws Exception + { + // if the cursor is persistent + if (persistent) + { + store.storeCursorAcknowledgeTransactional(tx.getID(), cursorId, position); + } + installTXCallback(tx, position); + + } + + public void ackTx(final Transaction tx, final PagedReference reference) throws Exception + { + ackTx(tx, reference.getPosition()); + + PageTransactionInfo txInfo = getPageTransaction(reference); + if (txInfo != null) + { + txInfo.storeUpdate(store, pageStore.getPagingManager(), tx); + } + } + + /* (non-Javadoc) + * @see org.hornetq.core.paging.cursor.PageCursor#confirm(org.hornetq.core.paging.cursor.PagePosition) + */ + public void ack(final PagedReference reference) throws Exception + { + ack(reference.getPosition()); + PageTransactionInfo txInfo = getPageTransaction(reference); + if (txInfo != null) + { + txInfo.storeUpdate(this.store, pageStore.getPagingManager()); + } + } + + public void ack(final PagePosition position) throws Exception + { + // if we are dealing with a persistent cursor + if (persistent) + { + store.storeCursorAcknowledge(cursorId, position); + } + + store.afterCompleteOperations(new IOAsyncTask() + { + + public void onError(final int errorCode, final String errorMessage) + { + } + + public void done() + { + processACK(position); + } + }); + } + + /* (non-Javadoc) + * @see org.hornetq.core.paging.cursor.PageCursor#getFirstPage() + */ + public long getFirstPage() + { + if (consumedPages.isEmpty()) + { + return 0; + } + else + { + return consumedPages.firstKey(); + } + } + + /* (non-Javadoc) + * @see org.hornetq.core.paging.cursor.PageCursor#returnElement(org.hornetq.core.paging.cursor.PagePosition) + */ + public void redeliver(final PagePosition position) + { + synchronized (redeliveries) + { + redeliveries.add(position); + } + } + + /* (non-Javadoc) + * @see org.hornetq.core.paging.cursor.PageSubscription#queryMessage(org.hornetq.core.paging.cursor.PagePosition) + */ + public PagedMessage queryMessage(PagePosition pos) + { + try + { + return cursorProvider.getMessage(pos); + } + catch (Exception e) + { + throw new RuntimeException(e.getMessage(), e); + } + } + + /** + * Theres no need to synchronize this method as it's only called from journal load on startup + */ + public void reloadACK(final PagePosition position) + { + if (recoveredACK == null) + { + recoveredACK = new LinkedList<PagePosition>(); + } + + recoveredACK.add(position); + } + + /* (non-Javadoc) + * @see org.hornetq.core.paging.cursor.PageCursor#recoverPreparedACK(org.hornetq.core.paging.cursor.PagePosition) + */ + public void reloadPreparedACK(final Transaction tx, final PagePosition position) + { + installTXCallback(tx, position); + } + + /* (non-Javadoc) + * @see org.hornetq.core.paging.cursor.PageCursor#positionIgnored(org.hornetq.core.paging.cursor.PagePosition) + */ + public void positionIgnored(final PagePosition position) + { + processACK(position); + } + + /* (non-Javadoc) + * @see org.hornetq.core.paging.cursor.PageCursor#isComplete(long) + */ + public boolean isComplete(long page) + { + PageCursorInfo info = consumedPages.get(page); + return info != null && info.isDone(); + } + + /** + * All the data associated with the cursor should go away here + */ + public void close() throws Exception + { + final long tx = store.generateUniqueID(); + + final ArrayList<Exception> ex = new ArrayList<Exception>(); + + final AtomicBoolean isPersistent = new AtomicBoolean(false); + + // We can't delete the records at the caller's thread + // because an executor may be holding the synchronized on PageCursorImpl + // what would lead to a dead lock + // so, we delete it inside the executor also + // and wait for the result + // The caller will be treating eventual IO exceptions and dispatching to the original thread's caller + executor.execute(new Runnable() + { + + public void run() + { + try + { + synchronized (PageSubscriptionImpl.this) + { + for (PageCursorInfo cursor : consumedPages.values()) + { + for (PagePosition info : cursor.acks) + { + if (info.getRecordID() != 0) + { + isPersistent.set(true); + store.deleteCursorAcknowledgeTransactional(tx, info.getRecordID()); + } + } + } + } + } + catch (Exception e) + { + ex.add(e); + PageSubscriptionImpl.log.warn(e.getMessage(), e); + } + } + }); + + Future future = new Future(); + + executor.execute(future); + + while (!future.await(5000)) + { + PageSubscriptionImpl.log.warn("Timeout on waiting cursor " + this + " to be closed"); + } + + if (isPersistent.get()) + { + // Another reason to perform the commit at the main thread is because the OperationContext may only send the + // result to the client when + // the IO on commit is done + if (ex.size() == 0) + { + store.commit(tx); + } + else + { + store.rollback(tx); + throw ex.get(0); + } + } + + cursorProvider.close(this); + } + + /* (non-Javadoc) + * @see org.hornetq.core.paging.cursor.PageCursor#getId() + */ + public long getId() + { + return cursorId; + } + + public boolean isPersistent() + { + return persistent; + } + + public void processReload() throws Exception + { + if (recoveredACK != null) + { + if (isTrace) + { + PageSubscriptionImpl.trace("********** processing reload!!!!!!!"); + } + Collections.sort(recoveredACK); + + boolean first = true; + + for (PagePosition pos : recoveredACK) + { + lastAckedPosition = pos; + PageCursorInfo positions = getPageInfo(pos); + if (first) + { + first = false; + if (pos.getMessageNr() > 0) + { + positions.confirmed.addAndGet(pos.getMessageNr()); + } + } + + positions.addACK(pos); + } + + recoveredACK.clear(); + recoveredACK = null; + } + } + + public void flushExecutors() + { + Future future = new Future(); + executor.execute(future); + while (!future.await(1000)) + { + PageSubscriptionImpl.log.warn("Waiting page cursor to finish executors - " + this); + } + } + + public void stop() + { + flushExecutors(); + } + + public void printDebug() + { + printDebug(toString()); + } + + public void printDebug(final String msg) + { + System.out.println("Debug information on PageCurorImpl- " + msg); + for (PageCursorInfo info : consumedPages.values()) + { + System.out.println(info); + } + } + + private synchronized PageCursorInfo getPageInfo(final PagePosition pos) + { + return getPageInfo(pos, true); + } + + /** + * @param page + * @return + */ + private synchronized PageCursorInfo getPageInfo(final PagePosition pos, boolean create) + { + PageCursorInfo pageInfo = consumedPages.get(pos.getPageNr()); + + if (create && pageInfo == null) + { + PageCache cache = cursorProvider.getPageCache(pos); + pageInfo = new PageCursorInfo(pos.getPageNr(), cache.getNumberOfMessages(), cache); + consumedPages.put(pos.getPageNr(), pageInfo); + } + + return pageInfo; + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + protected boolean match(final ServerMessage message) + { + if (filter == null) + { + return true; + } + else + { + return filter.match(message); + } + } + + // Private ------------------------------------------------------- + + // To be called only after the ACK has been processed and guaranteed to be on storae + // The only exception is on non storage events such as not matching messages + private void processACK(final PagePosition pos) + { + if (lastAckedPosition == null || pos.compareTo(lastAckedPosition) > 0) + { + if (lastAckedPosition != null && lastAckedPosition.getPageNr() != pos.getPageNr()) + { + // there's a different page being acked, we will do the check right away + if (autoCleanup) + { + scheduleCleanupCheck(); + } + } + lastAckedPosition = pos; + } + PageCursorInfo info = getPageInfo(pos); + + info.addACK(pos); + } + + /** + * @param tx + * @param position + */ + private void installTXCallback(final Transaction tx, final PagePosition position) + { + if (position.getRecordID() > 0) + { + // It needs to persist, otherwise the cursor will return to the fist page position + tx.setContainsPersistent(); + } + + PageCursorTX cursorTX = (PageCursorTX)tx.getProperty(TransactionPropertyIndexes.PAGE_CURSOR_POSITIONS); + + if (cursorTX == null) + { + cursorTX = new PageCursorTX(); + tx.putProperty(TransactionPropertyIndexes.PAGE_CURSOR_POSITIONS, cursorTX); + tx.addOperation(cursorTX); + } + + cursorTX.addPositionConfirmation(this, position); + + } + + private PageTransactionInfo getPageTransaction(final PagedReference reference) + { + if (reference.getPagedMessage().getTransactionID() != 0) + { + return pageStore.getPagingManager().getTransaction(reference.getPagedMessage().getTransactionID()); + } + else + { + return null; + } + } + + /** + * A callback from the PageCursorInfo. It will be called when all the messages on a page have been acked + * @param info + */ + private void onPageDone(final PageCursorInfo info) + { + if (autoCleanup) + { + scheduleCleanupCheck(); + } + } + + // Inner classes ------------------------------------------------- + + /** + * This will hold information about the pending ACKs towards a page. + * This instance will be released as soon as the entire page is consumed, releasing the memory at that point + * The ref counts are increased also when a message is ignored for any reason. + * */ + private class PageCursorInfo + { + // Number of messages existent on this page + private final int numberOfMessages; + + private final long pageId; + + // Confirmed ACKs on this page + private final List<PagePosition> acks = Collections.synchronizedList(new LinkedList<PagePosition>()); + + private WeakReference<PageCache> cache; + + private Set<PagePosition> removedReferences = new ConcurrentHashSet<PagePosition>(); + + // The page was live at the time of the creation + private final boolean wasLive; + + // There's a pending delete on the async IO pipe + // We're holding this object to avoid delete the pages before the IO is complete, + // however we can't delete these records again + private boolean pendingDelete; + + // We need a separate counter as the cursor may be ignoring certain values because of incomplete transactions or + // expressions + private final AtomicInteger confirmed = new AtomicInteger(0); + + @Override + public String toString() + { + return "PageCursorInfo::PageID=" + pageId + + " numberOfMessage = " + + numberOfMessages + + ", confirmed = " + + confirmed; + } + + public PageCursorInfo(final long pageId, final int numberOfMessages, final PageCache cache) + { + this.pageId = pageId; + this.numberOfMessages = numberOfMessages; + wasLive = cache.isLive(); + if (wasLive) + { + this.cache = new WeakReference<PageCache>(cache); + } + } + + public boolean isDone() + { + return getNumberOfMessages() == confirmed.get(); + } + + public boolean isPendingDelete() + { + return pendingDelete; + } + + public void setPendingDelete() + { + pendingDelete = true; + } + + /** + * @return the pageId + */ + public long getPageId() + { + return pageId; + } + + public boolean isRemoved(final PagePosition pos) + { + return removedReferences.contains(pos); + } + + public void remove(final PagePosition position) + { + removedReferences.add(position); + } + + public void addACK(final PagePosition posACK) + { + removedReferences.add(posACK); + acks.add(posACK); + + if (isTrace) + { + PageSubscriptionImpl.trace("numberOfMessages = " + getNumberOfMessages() + + " confirmed = " + + (confirmed.get() + 1) + + ", page = " + + pageId); + } + + // Negative could mean a bookmark on the first element for the page (example -1) + if (posACK.getMessageNr() >= 0) + { + if (getNumberOfMessages() == confirmed.incrementAndGet()) + { + onPageDone(this); + } + } + } + + private int getNumberOfMessages() + { + if (wasLive) + { + PageCache cache = this.cache.get(); + if (cache != null) + { + return cache.getNumberOfMessages(); + } + else + { + cache = cursorProvider.getPageCache(new PagePositionImpl(pageId, 0)); + this.cache = new WeakReference<PageCache>(cache); + return cache.getNumberOfMessages(); + } + } + else + { + return numberOfMessages; + } + } + + } + + static class PageCursorTX extends TransactionOperationAbstract + { + HashMap<PageSubscriptionImpl, List<PagePosition>> pendingPositions = new HashMap<PageSubscriptionImpl, List<PagePosition>>(); + + public void addPositionConfirmation(final PageSubscriptionImpl cursor, final PagePosition position) + { + List<PagePosition> list = pendingPositions.get(cursor); + + if (list == null) + { + list = new LinkedList<PagePosition>(); + pendingPositions.put(cursor, list); + } + + list.add(position); + } + + /* (non-Javadoc) + * @see org.hornetq.core.transaction.TransactionOperation#afterCommit(org.hornetq.core.transaction.Transaction) + */ + @Override + public void afterCommit(final Transaction tx) + { + for (Entry<PageSubscriptionImpl, List<PagePosition>> entry : pendingPositions.entrySet()) + { + PageSubscriptionImpl cursor = entry.getKey(); + + List<PagePosition> positions = entry.getValue(); + + for (PagePosition confirmed : positions) + { + cursor.processACK(confirmed); + } + + } + } + + /* (non-Javadoc) + * @see org.hornetq.core.transaction.TransactionOperation#getRelatedMessageReferences() + */ + public List<MessageReference> getRelatedMessageReferences() + { + return Collections.emptyList(); + } + + + } + + class CursorIterator implements LinkedListIterator<PagedReference> + { + private PagePosition position = null; + + private PagePosition lastOperation = null; + + private volatile boolean isredelivery = false; + + private volatile PagedReference lastRedelivery = null; + + /** next element taken on hasNext test. + * it has to be delivered on next next operation */ + private volatile PagedReference cachedNext; + + public CursorIterator() + { + } + + public void repeat() + { + if (isredelivery) + { + synchronized (redeliveries) + { + cachedNext = lastRedelivery; + } + } + else + { + if (lastOperation == null) + { + position = null; + } + else + { + position = lastOperation; + } + } + } + + /* (non-Javadoc) + * @see java.util.Iterator#next() + */ + public synchronized PagedReference next() + { + + if (cachedNext != null) + { + PagedReference retPos = cachedNext; + cachedNext = null; + return retPos; + } + + try + { + if (position == null) + { + position = getStartPosition(); + } + + return moveNext(); + } + catch (Exception e) + { + throw new RuntimeException(e.getMessage(), e); + } + } + + /* (non-Javadoc) + * @see org.hornetq.core.paging.cursor.PageCursor#moveNext() + */ + public PagedReference moveNext() throws Exception + { + synchronized (PageSubscriptionImpl.this) + { + boolean match = false; + + PagedReference message = null; + + PagePosition lastPosition = position; + PagePosition tmpPosition = position; + + do + { + synchronized (redeliveries) + { + PagePosition redelivery = redeliveries.poll(); + + if (redelivery != null) + { + // There's a redelivery pending, we will get it out of that pool instead + isredelivery = true; + PagedReference redeliveredMsg = getReference(redelivery); + lastRedelivery = redeliveredMsg; + + return redeliveredMsg; + } + else + { + lastRedelivery = null; + isredelivery = false; + } + + message = internalGetNext(tmpPosition); + } + + if (message == null) + { + break; + } + + tmpPosition = message.getPosition(); + + boolean valid = true; + boolean ignored = false; + + // Validate the scenarios where the message should be considered not valid even to be considered + + // 1st... is it routed? + + valid = routed(message.getPagedMessage()); + if (!valid) + { + ignored = true; + } + + // 2nd ... if TX, is it committed? + if (valid && message.getPagedMessage().getTransactionID() != 0) + { + PageTransactionInfo tx = pageStore.getPagingManager().getTransaction(message.getPagedMessage() + .getTransactionID()); + if (tx == null) + { + log.warn("Couldn't locate page transaction " + message.getPagedMessage().getTransactionID() + + ", ignoring message on position " + + message.getPosition()); + valid = false; + ignored = true; + } + else + { + if (tx.deliverAfterCommit(PageSubscriptionImpl.this, message.getPosition())) + { + valid = false; + ignored = false; + } + } + } + + // 3rd... was it previously removed? + if (valid) + { + // We don't create a PageCursorInfo unless we are doing a write operation (ack or removing) + // Say you have a Browser that will only read the files... there's no need to control PageCursors is + // nothing + // is being changed. That's why the false is passed as a parameter here + PageCursorInfo info = getPageInfo(message.getPosition(), false); + if (info != null && info.isRemoved(message.getPosition())) + { + valid = false; + } + } + + if (!ignored) + { + position = message.getPosition(); + } + + if (valid) + { + match = match(message.getMessage()); + + if (!match) + { + processACK(message.getPosition()); + } + } + else if (ignored) + { + positionIgnored(message.getPosition()); + } + } + while (message != null && !match); + + if (message != null) + { + lastOperation = lastPosition; + } + + return message; + } + } + + /** QueueImpl::deliver could be calling hasNext while QueueImpl.depage could be using next and hasNext as well. + * It would be a rare race condition but I would prefer avoiding that scenario */ + public synchronized boolean hasNext() + { + // if an unbehaved program called hasNext twice before next, we only cache it once. + if (cachedNext != null) + { + return true; + } + + if (!pageStore.isPaging()) + { + return false; + } + + cachedNext = next(); + + return cachedNext != null; + } + + /* (non-Javadoc) + * @see java.util.Iterator#remove() + */ + public void remove() + { + if (!isredelivery) + { + PageSubscriptionImpl.this.getPageInfo(position).remove(position); + } + } + + /* (non-Javadoc) + * @see org.hornetq.utils.LinkedListIterator#close() + */ + public void close() + { + } + } + +} Modified: trunk/src/main/org/hornetq/core/paging/impl/PageImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/impl/PageImpl.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/paging/impl/PageImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -13,7 +13,6 @@ package org.hornetq.core.paging.impl; -import java.nio.Buffer; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.List; @@ -27,6 +26,7 @@ import org.hornetq.core.logging.Logger; import org.hornetq.core.paging.Page; import org.hornetq.core.paging.PagedMessage; +import org.hornetq.core.paging.cursor.LivePageCache; import org.hornetq.core.persistence.StorageManager; import org.hornetq.utils.DataConstants; @@ -35,7 +35,7 @@ * @author <a href="mailto:clebert.suconic@jboss.com">Clebert Suconic</a> * */ -public class PageImpl implements Page +public class PageImpl implements Page, Comparable<Page> { // Constants ----------------------------------------------------- @@ -58,6 +58,11 @@ private final SequentialFile file; private final SequentialFileFactory fileFactory; + + /** + * The page cache that will be filled with data as we write more data + */ + private volatile LivePageCache pageCache; private final AtomicInteger size = new AtomicInteger(0); @@ -90,13 +95,19 @@ { return pageId; } + + public void setLiveCache(LivePageCache pageCache) + { + this.pageCache = pageCache; + } public List<PagedMessage> read() throws Exception { ArrayList<PagedMessage> messages = new ArrayList<PagedMessage>(); + size.set((int)file.size()); // Using direct buffer, as described on https://jira.jboss.org/browse/HORNETQ-467 - ByteBuffer buffer2 = ByteBuffer.allocateDirect((int)file.size()); + ByteBuffer buffer2 = ByteBuffer.allocateDirect(size.get()); file.position(0); file.read(buffer2); @@ -168,6 +179,11 @@ buffer.rewind(); file.writeDirect(buffer, false); + + if (pageCache != null) + { + pageCache.addLiveMessage(message); + } numberOfMessages.incrementAndGet(); size.addAndGet(buffer.limit()); @@ -193,6 +209,12 @@ { storageManager.pageClosed(storeName, pageId); } + if (pageCache != null) + { + pageCache.close(); + // leave it to the soft cache to decide when to release it now + pageCache = null; + } file.close(); } @@ -241,13 +263,54 @@ { return "PageImpl::pageID=" + this.pageId + ", file=" + this.file; } + + /* (non-Javadoc) + * @see java.lang.Comparable#compareTo(java.lang.Object) + */ + public int compareTo(Page otherPage) + { + return otherPage.getPageId() - this.pageId; + } + + + /* (non-Javadoc) + * @see java.lang.Object#hashCode() + */ + @Override + public int hashCode() + { + final int prime = 31; + int result = 1; + result = prime * result + pageId; + return result; + } + + // Package protected --------------------------------------------- // Protected ----------------------------------------------------- // Private ------------------------------------------------------- + /* (non-Javadoc) + * @see java.lang.Object#equals(java.lang.Object) + */ + @Override + public boolean equals(Object obj) + { + if (this == obj) + return true; + if (obj == null) + return false; + if (getClass() != obj.getClass()) + return false; + PageImpl other = (PageImpl)obj; + if (pageId != other.pageId) + return false; + return true; + } + /** * @param position * @param msgNumber Modified: trunk/src/main/org/hornetq/core/paging/impl/PageTransactionInfoImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/impl/PageTransactionInfoImpl.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/paging/impl/PageTransactionInfoImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -13,19 +13,25 @@ package org.hornetq.core.paging.impl; +import java.util.HashMap; +import java.util.LinkedList; import java.util.List; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; +import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.api.core.Pair; +import org.hornetq.core.journal.IOAsyncTask; import org.hornetq.core.logging.Logger; import org.hornetq.core.paging.PageTransactionInfo; import org.hornetq.core.paging.PagingManager; +import org.hornetq.core.paging.cursor.PagePosition; +import org.hornetq.core.paging.cursor.PageSubscription; import org.hornetq.core.persistence.StorageManager; import org.hornetq.core.server.MessageReference; import org.hornetq.core.transaction.Transaction; -import org.hornetq.core.transaction.TransactionOperation; +import org.hornetq.core.transaction.TransactionOperationAbstract; +import org.hornetq.core.transaction.TransactionPropertyIndexes; import org.hornetq.utils.DataConstants; /** @@ -45,13 +51,15 @@ private volatile long recordID = -1; - private volatile CountDownLatch countDownCompleted; + private volatile boolean committed = false; + + private volatile boolean useRedelivery = false; - private volatile boolean committed; + private volatile boolean rolledback = false; - private volatile boolean rolledback; - private AtomicInteger numberOfMessages = new AtomicInteger(0); + + private List<Pair<PageSubscription, PagePosition>> lateDeliveries; // Static -------------------------------------------------------- @@ -61,7 +69,6 @@ { this(); this.transactionID = transactionID; - countDownCompleted = new CountDownLatch(1); } public PageTransactionInfoImpl() @@ -85,7 +92,7 @@ return transactionID; } - public void update(final int update, final StorageManager storageManager, PagingManager pagingManager) + public void onUpdate(final int update, final StorageManager storageManager, PagingManager pagingManager) { int sizeAfterUpdate = numberOfMessages.addAndGet(-update); if (sizeAfterUpdate == 0 && storageManager != null) @@ -98,10 +105,7 @@ { log.warn("Can't delete page transaction id=" + this.recordID); } - } - - if (sizeAfterUpdate == 0 && pagingManager != null) - { + pagingManager.removeTransaction(this.transactionID); } } @@ -110,6 +114,11 @@ { numberOfMessages.incrementAndGet(); } + + public void increment(final int size) + { + numberOfMessages.addAndGet(size); + } public int getNumberOfMessages() { @@ -122,7 +131,6 @@ { transactionID = buffer.readLong(); numberOfMessages.set(buffer.readInt()); - countDownCompleted = null; committed = true; } @@ -137,25 +145,18 @@ return DataConstants.SIZE_LONG + DataConstants.SIZE_INT; } - public void commit() + public synchronized void commit() { - committed = true; - /** - * this is to avoid a race condition where the transaction still being committed while another thread is depaging messages - */ - countDownCompleted.countDown(); - } - - public boolean waitCompletion(final int timeoutMilliseconds) throws InterruptedException - { - if (countDownCompleted == null) + if (lateDeliveries != null) { - return true; + for (Pair<PageSubscription, PagePosition> pos : lateDeliveries) + { + pos.a.redeliver(pos.b); + } + lateDeliveries.clear(); } - else - { - return countDownCompleted.await(timeoutMilliseconds, TimeUnit.MILLISECONDS); - } + committed = true; + lateDeliveries = null; } public void store(final StorageManager storageManager, PagingManager pagingManager, final Transaction tx) throws Exception @@ -166,38 +167,35 @@ /* (non-Javadoc) * @see org.hornetq.core.paging.PageTransactionInfo#storeUpdate(org.hornetq.core.persistence.StorageManager, org.hornetq.core.transaction.Transaction, int) */ - public void storeUpdate(final StorageManager storageManager, final PagingManager pagingManager, final Transaction tx, final int depages) throws Exception + public void storeUpdate(final StorageManager storageManager, final PagingManager pagingManager, final Transaction tx) throws Exception { - storageManager.updatePageTransaction(tx.getID(), this, depages); + UpdatePageTXOperation pgtxUpdate = (UpdatePageTXOperation)tx.getProperty(TransactionPropertyIndexes.PAGE_TRANSACTION_UPDATE); - final PageTransactionInfo pgToUpdate = this; + if (pgtxUpdate == null) + { + pgtxUpdate = new UpdatePageTXOperation(storageManager, pagingManager); + tx.putProperty(TransactionPropertyIndexes.PAGE_TRANSACTION_UPDATE, pgtxUpdate); + tx.addOperation(pgtxUpdate); + } - tx.addOperation(new TransactionOperation() + tx.setContainsPersistent(); + + pgtxUpdate.addUpdate(this); + } + + public void storeUpdate(final StorageManager storageManager, final PagingManager pagingManager) throws Exception + { + storageManager.updatePageTransaction(this, 1); + storageManager.afterCompleteOperations(new IOAsyncTask() { - public void beforeRollback(Transaction tx) throws Exception + public void onError(int errorCode, String errorMessage) { } - public void beforePrepare(Transaction tx) throws Exception + public void done() { + PageTransactionInfoImpl.this.onUpdate(1, storageManager, pagingManager); } - - public void beforeCommit(Transaction tx) throws Exception - { - } - - public void afterRollback(Transaction tx) - { - } - - public void afterPrepare(Transaction tx) - { - } - - public void afterCommit(Transaction tx) - { - pgToUpdate.update(depages, storageManager, pagingManager); - } public List<MessageReference> getRelatedMessageReferences() { @@ -205,6 +203,8 @@ } }); } + + public boolean isCommit() { @@ -216,19 +216,18 @@ return rolledback; } - public void rollback() + public synchronized void rollback() { rolledback = true; committed = false; - countDownCompleted.countDown(); - } - public void markIncomplete() - { - committed = false; - rolledback = false; - - countDownCompleted = new CountDownLatch(1); + if (lateDeliveries != null) + { + for (Pair<PageSubscription, PagePosition> pos : lateDeliveries) + { + pos.a.positionIgnored(pos.b); + } + } } public String toString() @@ -241,6 +240,39 @@ ")"; } + /* (non-Javadoc) + * @see org.hornetq.core.paging.PageTransactionInfo#deliverAfterCommit(org.hornetq.core.paging.cursor.PageCursor, org.hornetq.core.paging.cursor.PagePosition) + */ + public synchronized boolean deliverAfterCommit(PageSubscription cursor, PagePosition cursorPos) + { + if (committed && useRedelivery) + { + cursor.redeliver(cursorPos); + return true; + } + else + if (committed) + { + return false; + } + else + if (rolledback) + { + cursor.positionIgnored(cursorPos); + return true; + } + else + { + useRedelivery = true; + if (lateDeliveries == null) + { + lateDeliveries = new LinkedList<Pair<PageSubscription, PagePosition>>(); + } + lateDeliveries.add(new Pair<PageSubscription, PagePosition>(cursor, cursorPos)); + return true; + } + } + // Package protected --------------------------------------------- // Protected ----------------------------------------------------- @@ -248,4 +280,68 @@ // Private ------------------------------------------------------- // Inner classes ------------------------------------------------- + + + static class UpdatePageTXOperation extends TransactionOperationAbstract + { + private HashMap<PageTransactionInfo, AtomicInteger> countsToUpdate = new HashMap<PageTransactionInfo, AtomicInteger>(); + + private boolean stored = false; + + private final StorageManager storageManager; + + private final PagingManager pagingManager; + + public UpdatePageTXOperation(final StorageManager storageManager, final PagingManager pagingManager) + { + this.storageManager = storageManager; + this.pagingManager = pagingManager; + } + + public void addUpdate(PageTransactionInfo info) + { + AtomicInteger counter = countsToUpdate.get(info); + + if (counter == null) + { + counter = new AtomicInteger(0); + countsToUpdate.put(info, counter); + } + + counter.incrementAndGet(); + } + + public void beforePrepare(Transaction tx) throws Exception + { + storeUpdates(tx); + } + + public void beforeCommit(Transaction tx) throws Exception + { + storeUpdates(tx); + } + + public void afterCommit(Transaction tx) + { + for (Map.Entry<PageTransactionInfo, AtomicInteger> entry : countsToUpdate.entrySet()) + { + entry.getKey().onUpdate(entry.getValue().intValue(), storageManager, pagingManager); + } + } + + private void storeUpdates(Transaction tx) throws Exception + { + if (!stored) + { + stored = true; + for (Map.Entry<PageTransactionInfo, AtomicInteger> entry : countsToUpdate.entrySet()) + { + storageManager.updatePageTransaction(tx.getID(), entry.getKey(), entry.getValue().get()); + } + } + } + + + + } } Modified: trunk/src/main/org/hornetq/core/paging/impl/PagedMessageImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/impl/PagedMessageImpl.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/paging/impl/PagedMessageImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -49,17 +49,20 @@ private byte[] largeMessageLazyData; private ServerMessage message; + + private long queueIDs[]; - private long transactionID = -1; + private long transactionID = 0; - public PagedMessageImpl(final ServerMessage message, final long transactionID) + public PagedMessageImpl(final ServerMessage message, final long[] queueIDs, final long transactionID) { - this.message = message; + this(message, queueIDs); this.transactionID = transactionID; } - public PagedMessageImpl(final ServerMessage message) + public PagedMessageImpl(final ServerMessage message, final long[] queueIDs) { + this.queueIDs = queueIDs; this.message = message; } @@ -67,8 +70,13 @@ { } - public ServerMessage getMessage(final StorageManager storage) + public ServerMessage getMessage() { + return message; + } + + public void initMessage(StorageManager storage) + { if (largeMessageLazyData != null) { message = storage.createLargeMessage(); @@ -76,13 +84,17 @@ message.decodeHeadersAndProperties(buffer); largeMessageLazyData = null; } - return message; } public long getTransactionID() { return transactionID; } + + public long[] getQueueIDs() + { + return queueIDs; + } // EncodingSupport implementation -------------------------------- @@ -108,6 +120,15 @@ message.decode(buffer); } + + int queueIDsSize = buffer.readInt(); + + queueIDs = new long[queueIDsSize]; + + for (int i = 0 ; i < queueIDsSize; i++) + { + queueIDs[i] = buffer.readLong(); + } } public void encode(final HornetQBuffer buffer) @@ -119,11 +140,19 @@ buffer.writeInt(message.getEncodeSize()); message.encode(buffer); + + buffer.writeInt(queueIDs.length); + + for (int i = 0 ; i < queueIDs.length; i++) + { + buffer.writeLong(queueIDs[i]); + } } public int getEncodeSize() { - return DataConstants.SIZE_LONG + DataConstants.SIZE_BYTE + DataConstants.SIZE_INT + message.getEncodeSize(); + return DataConstants.SIZE_LONG + DataConstants.SIZE_BYTE + DataConstants.SIZE_INT + message.getEncodeSize() + + DataConstants.SIZE_INT + queueIDs.length * DataConstants.SIZE_LONG; } // Package protected --------------------------------------------- Modified: trunk/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -228,14 +228,23 @@ } } } + + public void processReload() throws Exception + { + for (PagingStore store: stores.values()) + { + store.processReload(); + } + } + // Package protected --------------------------------------------- // Protected ----------------------------------------------------- // Private ------------------------------------------------------- - protected PagingStore newStore(final SimpleString address) throws Exception + protected PagingStore newStore(final SimpleString address) { return pagingStoreFactory.newStore(address, addressSettingsRepository.getMatch(address.toString())); Modified: trunk/src/main/org/hornetq/core/paging/impl/PagingStoreFactoryNIO.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/impl/PagingStoreFactoryNIO.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/paging/impl/PagingStoreFactoryNIO.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -87,7 +87,7 @@ { } - public synchronized PagingStore newStore(final SimpleString address, final AddressSettings settings) throws Exception + public synchronized PagingStore newStore(final SimpleString address, final AddressSettings settings) { return new PagingStoreImpl(address, @@ -98,7 +98,7 @@ this, address, settings, - executorFactory.getExecutor(), + executorFactory, syncNonTransactional); } @@ -202,7 +202,7 @@ this, address, settings, - executorFactory.getExecutor(), + executorFactory, 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 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -14,20 +14,15 @@ package org.hornetq.core.paging.impl; import java.text.DecimalFormat; -import java.util.Arrays; -import java.util.HashMap; +import java.util.Collections; import java.util.List; -import java.util.Map; import java.util.Queue; import java.util.concurrent.ConcurrentLinkedQueue; -import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; -import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.ReadWriteLock; -import java.util.concurrent.locks.ReentrantLock; import java.util.concurrent.locks.ReentrantReadWriteLock; import org.hornetq.api.core.SimpleString; @@ -40,17 +35,25 @@ import org.hornetq.core.paging.PagingManager; import org.hornetq.core.paging.PagingStore; import org.hornetq.core.paging.PagingStoreFactory; +import org.hornetq.core.paging.cursor.LivePageCache; +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.StorageManager; import org.hornetq.core.postoffice.DuplicateIDCache; import org.hornetq.core.postoffice.PostOffice; -import org.hornetq.core.server.LargeServerMessage; +import org.hornetq.core.server.MessageReference; +import org.hornetq.core.server.RouteContextList; +import org.hornetq.core.server.RoutingContext; import org.hornetq.core.server.ServerMessage; import org.hornetq.core.settings.impl.AddressFullMessagePolicy; import org.hornetq.core.settings.impl.AddressSettings; import org.hornetq.core.transaction.Transaction; -import org.hornetq.core.transaction.TransactionPropertyIndexes; import org.hornetq.core.transaction.Transaction.State; -import org.hornetq.core.transaction.impl.TransactionImpl; +import org.hornetq.core.transaction.TransactionOperation; +import org.hornetq.core.transaction.TransactionPropertyIndexes; +import org.hornetq.utils.ExecutorFactory; +import org.hornetq.utils.Future; /** * @@ -109,17 +112,16 @@ private volatile int currentPageId; private volatile Page currentPage; + + private volatile boolean paging = false; - private final ReentrantLock writeLock = new ReentrantLock(); - /** duplicate cache used at this address */ private final DuplicateIDCache duplicateCache; - /** - * We need to perform checks on currentPage with minimal locking - * */ - private final ReadWriteLock currentPageLock = new ReentrantReadWriteLock(); + private final PageCursorProvider cursorProvider; + private final ReadWriteLock lock = new ReentrantReadWriteLock(); + private volatile boolean running = false; protected final boolean syncNonTransactional; @@ -146,7 +148,7 @@ final PagingStoreFactory storeFactory, final SimpleString storeName, final AddressSettings addressSettings, - final Executor executor, + final ExecutorFactory executorFactory, final boolean syncNonTransactional) { if (pagingManager == null) @@ -178,7 +180,7 @@ pageSize); } - this.executor = executor; + this.executor = executorFactory.getExecutor(); this.pagingManager = pagingManager; @@ -188,6 +190,8 @@ this.syncNonTransactional = syncNonTransactional; + this.cursorProvider = new PageCursorProviderImpl(this, this.storageManager, executorFactory); + // Post office could be null on the backup node if (postOffice == null) { @@ -202,8 +206,38 @@ // Public -------------------------------------------------------- + public String toString() + { + return "PagingStoreImpl(" + this.address + ")"; + } + // PagingStore implementation ------------------------------------ + public void lock() + { + lock.writeLock().lock(); + } + + public void unlock() + { + lock.writeLock().unlock(); + } + + public PageCursorProvider getCursorProvier() + { + return cursorProvider; + } + + public long getFirstPage() + { + return firstPageId; + } + + public long getTopPage() + { + return currentPageId; + } + public SimpleString getAddress() { return address; @@ -231,7 +265,7 @@ public boolean isPaging() { - currentPageLock.readLock().lock(); + lock.readLock().lock(); try { @@ -245,12 +279,12 @@ } else { - return currentPage != null; + return paging; } } finally { - currentPageLock.readLock().unlock(); + lock.readLock().unlock(); } } @@ -259,27 +293,31 @@ return numberOfPages; } + public int getCurrentWritingPage() + { + return currentPageId; + } + public SimpleString getStoreName() { return storeName; } - - public boolean page(final List<ServerMessage> message, final long transactionID) throws Exception + + public boolean page(final ServerMessage message, final RoutingContext ctx) throws Exception { - // The sync on transactions is done on commit only - return page(message, transactionID, false); + return page(message, ctx, ctx.getContextListing(storeName)); } - public boolean page(final ServerMessage message) throws Exception + public boolean page(final ServerMessage message, final RoutingContext ctx, RouteContextList listCtx) 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 page(Arrays.asList(message), -1, syncNonTransactional && message.isDurable()); + // The sync on transactions is done on commit only + // TODO: sync on paging + return page(message, ctx, listCtx, false); } public void sync() throws Exception { - currentPageLock.readLock().lock(); + lock.readLock().lock(); try { @@ -290,12 +328,18 @@ } finally { - currentPageLock.readLock().unlock(); + lock.readLock().unlock(); } } public boolean startDepaging() { + + // Disabled for now + + return false; + + /* if (!running) { return false; @@ -332,12 +376,22 @@ finally { currentPageLock.readLock().unlock(); - } + } */ } + public void processReload() throws Exception + { + cursorProvider.processReload(); + } + + public PagingManager getPagingManager() + { + return pagingManager; + } + // HornetQComponent implementation - public synchronized boolean isStarted() + public boolean isStarted() { return running; } @@ -346,22 +400,12 @@ { if (running) { - running = false; - final CountDownLatch latch = new CountDownLatch(1); + cursorProvider.stop(); - executor.execute(new Runnable() - { - public void run() - { - latch.countDown(); - } - }); + running = false; - if (!latch.await(60, TimeUnit.SECONDS)) - { - PagingStoreImpl.log.warn("Timed out on waiting PagingStore " + address + " to shutdown"); - } + flushExecutors(); if (currentPage != null) { @@ -370,10 +414,24 @@ } } } + + public void flushExecutors() + { + cursorProvider.flushExecutors(); + + Future future = new Future(); + executor.execute(future); + + if (!future.await(60000)) + { + PagingStoreImpl.log.warn("Timed out on waiting PagingStore " + address + " to shutdown"); + } + } + public void start() throws Exception { - writeLock.lock(); + lock.writeLock().lock(); try { @@ -389,57 +447,78 @@ } else { - currentPageLock.writeLock().lock(); + running = true; + firstPageId = Integer.MAX_VALUE; - try + // There are no files yet on this Storage. We will just return it empty + if (fileFactory != null) { - running = true; - firstPageId = Integer.MAX_VALUE; - // There are no files yet on this Storage. We will just return it empty - if (fileFactory != null) - { + currentPageId = 0; + currentPage = null; - currentPageId = 0; - currentPage = null; + List<String> files = fileFactory.listFiles("page"); - List<String> files = fileFactory.listFiles("page"); + numberOfPages = files.size(); - numberOfPages = files.size(); + for (String fileName : files) + { + final int fileId = PagingStoreImpl.getPageIdFromFileName(fileName); - for (String fileName : files) + if (fileId > currentPageId) { - final int fileId = PagingStoreImpl.getPageIdFromFileName(fileName); + currentPageId = fileId; + } - if (fileId > currentPageId) - { - currentPageId = fileId; - } - - if (fileId < firstPageId) - { - firstPageId = fileId; - } + if (fileId < firstPageId) + { + firstPageId = fileId; } + } - if (numberOfPages != 0) + if (currentPageId != 0) + { + currentPage = createPage(currentPageId); + currentPage.open(); + + List<PagedMessage> messages = currentPage.read(); + + LivePageCache pageCache = new LivePageCacheImpl(currentPage); + + for (PagedMessage msg : messages) { - startPaging(); + msg.initMessage(storageManager); + pageCache.addLiveMessage(msg); } + + currentPage.setLiveCache(pageCache); + + currentPageSize.set(currentPage.getSize()); + + cursorProvider.addPageCache(pageCache); } + + // We will not mark it for paging if there's only a single empty file + if (currentPage != null && !(numberOfPages == 1 && currentPage.getSize() == 0)) + { + startPaging(); + } } - finally - { - currentPageLock.writeLock().unlock(); - } } } finally { - writeLock.unlock(); + lock.writeLock().unlock(); } } + + public void stopPaging() + { + lock.writeLock().lock(); + paging = false; + lock.writeLock().unlock(); + } public boolean startPaging() { @@ -448,28 +527,30 @@ return false; } - // First check without any global locks. - // (Faster) - currentPageLock.readLock().lock(); + lock.readLock().lock(); try { - // Already paging, nothing to be done - if (currentPage != null) + if (paging) { return false; } } finally { - currentPageLock.readLock().unlock(); + lock.readLock().unlock(); } // if the first check failed, we do it again under a global currentPageLock // (writeLock) this time - writeLock.lock(); + lock.writeLock().lock(); try { + if (paging) + { + return false; + } + if (currentPage == null) { try @@ -483,17 +564,15 @@ PagingStoreImpl.log.warn("IO Error, impossible to start paging", e); return false; } - - return true; } - else - { - return false; - } + + paging = true; + + return true; } finally { - writeLock.unlock(); + lock.writeLock().unlock(); } } @@ -502,9 +581,9 @@ return currentPage; } - public Page createPage(final int page) throws Exception + public Page createPage(final int pageNumber) throws Exception { - String fileName = createFileName(page); + String fileName = createFileName(pageNumber); if (fileFactory == null) { @@ -513,17 +592,18 @@ SequentialFile file = fileFactory.createSequentialFile(fileName, 1000); + Page page = new PageImpl(storeName, storageManager, fileFactory, file, pageNumber); + + // To create the file file.open(); file.position(0); file.close(); - return new PageImpl(storeName, storageManager, fileFactory, file, page); + return page; } - // TestSupportPageStore ------------------------------------------ - public void forceAnotherPage() throws Exception { openNewPage(); @@ -539,9 +619,7 @@ * */ public Page depage() throws Exception { - writeLock.lock(); - - currentPageLock.writeLock().lock(); // Make sure no checks are done on currentPage while we are depaging + lock.writeLock().lock(); // Make sure no checks are done on currentPage while we are depaging try { if (!running) @@ -578,6 +656,7 @@ // The current page is empty... which means we reached the end of the pages if (returnPage.getNumberOfMessages() == 0) { + stopPaging(); returnPage.open(); returnPage.delete(); @@ -603,8 +682,7 @@ } finally { - currentPageLock.writeLock().unlock(); - writeLock.unlock(); + lock.writeLock().unlock(); } } @@ -620,59 +698,6 @@ * @return * @throws Exception */ - protected boolean readPage() throws Exception - { - Page page = depage(); - - // It's important that only depage should happen while locked - // or we would be holding a lock for a long time - // The reading (IO part) should happen outside of any locks - - if (page == null) - { - return false; - } - - page.open(); - - List<PagedMessage> messages = null; - - try - { - messages = page.read(); - } - finally - { - try - { - page.close(); - } - catch (Throwable ignored) - { - } - } - - if (onDepage(page.getPageId(), storeName, messages)) - { - if (page.delete()) - { - // DuplicateCache could be null during replication - // however the deletes on the journal will happen through replicated journal - if (duplicateCache != null) - { - duplicateCache.deleteFromCache(generateDuplicateID(page.getPageId())); - } - } - - return true; - } - else - { - return false; - } - - } - private Queue<OurRunnable> onMemoryFreedRunnables = new ConcurrentLinkedQueue<OurRunnable>(); private class MemoryFreedRunnablesExecutor implements Runnable @@ -802,7 +827,7 @@ } - protected boolean page(final List<ServerMessage> messages, final long transactionID, final boolean sync) throws Exception + protected boolean page(ServerMessage message, final RoutingContext ctx, RouteContextList listCtx, final boolean sync) throws Exception { if (!running) { @@ -836,252 +861,182 @@ } // We need to ensure a read lock, as depage could change the paging state - currentPageLock.readLock().lock(); + lock.readLock().lock(); try { // First check done concurrently, to avoid synchronization and increase throughput - if (currentPage == null) + if (!paging) { return false; } } finally { - currentPageLock.readLock().unlock(); + lock.readLock().unlock(); } - writeLock.lock(); + lock.writeLock().lock(); try { - if (currentPage == null) + if (!paging) { return false; } - for (ServerMessage message : messages) + PagedMessage pagedMessage; + + if (!message.isDurable()) { - PagedMessage pagedMessage; + // The address should never be transient when paging (even for non-persistent messages when paging) + // This will force everything to be persisted + message.bodyChanged(); + } + + Transaction tx = ctx.getTransaction(); - if (!message.isDurable()) - { - // The address should never be transient when paging (even for non-persistent messages when paging) - // This will force everything to be persisted - message.bodyChanged(); - } + pagedMessage = new PagedMessageImpl(message, getQueueIDs(listCtx), getTransactionID(tx, listCtx)); - if (transactionID != -1) - { - pagedMessage = new PagedMessageImpl(message, transactionID); - } - else - { - pagedMessage = new PagedMessageImpl(message); - } + int bytesToWrite = pagedMessage.getEncodeSize() + PageImpl.SIZE_RECORD; - int bytesToWrite = pagedMessage.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 - { - currentPage.write(pagedMessage); - - if (sync) - { - currentPage.sync(); - } - } - finally - { - currentPageLock.readLock().unlock(); - } + if (currentPageSize.addAndGet(bytesToWrite) > pageSize && currentPage.getNumberOfMessages() > 0) + { + // Make sure nothing is currently validating or using currentPage + openNewPage(); } + currentPage.write(pagedMessage); + return true; } finally { - writeLock.unlock(); + lock.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 - */ - - private boolean onDepage(final int pageId, final SimpleString address, final List<PagedMessage> pagedMessages) throws Exception + private long[] getQueueIDs(RouteContextList ctx) { - if (PagingStoreImpl.isTrace) + List<org.hornetq.core.server.Queue> durableQueues = ctx.getDurableQueues(); + List<org.hornetq.core.server.Queue> nonDurableQueues = ctx.getNonDurableQueues(); + long ids[] = new long [durableQueues.size() + nonDurableQueues.size()]; + int i = 0; + + for (org.hornetq.core.server.Queue q : durableQueues) { - PagingStoreImpl.trace("Depaging...."); + ids[i++] = q.getID(); } - - if (pagedMessages.size() == 0) + + for (org.hornetq.core.server.Queue q : nonDurableQueues) { - // nothing to be done on this case. - return true; + ids[i++] = q.getID(); } - - // Depage has to be done atomically, in case of failure it should be - // back to where it was - - byte[] duplicateIdForPage = generateDuplicateID(pageId); - - Transaction depageTransaction = new TransactionImpl(storageManager); - - // DuplicateCache could be null during replication - if (duplicateCache != null) + return ids; + } + + private long getTransactionID(final Transaction tx, final RouteContextList listCtx) throws Exception + { + if (tx == null) { - if (duplicateCache.contains(duplicateIdForPage)) + return 0l; + } + else + { + PageTransactionInfo pgTX = (PageTransactionInfo) tx.getProperty(TransactionPropertyIndexes.PAGE_TRANSACTION); + if (pgTX == null) { - log.warn("Page " + pageId + - " had been processed already but the file wasn't removed as a crash happened. Ignoring this page"); - return true; + pgTX = new PageTransactionInfoImpl(tx.getID()); + System.out.println("Creating pageTransaction " + pgTX.getTransactionID()); + pagingManager.addTransaction(pgTX); + tx.putProperty(TransactionPropertyIndexes.PAGE_TRANSACTION, pgTX); + tx.addOperation(new FinishPageMessageOperation(pgTX)); } - - duplicateCache.addToCache(duplicateIdForPage, depageTransaction); + + pgTX.increment(listCtx.getNumberOfQueues()); + + return tx.getID(); } + } - depageTransaction.putProperty(TransactionPropertyIndexes.IS_DEPAGE, Boolean.valueOf(true)); + + private class FinishPageMessageOperation implements TransactionOperation + { + private final PageTransactionInfo pageTransaction; + + private boolean stored = false; - HashMap<PageTransactionInfo, AtomicInteger> pageTransactionsToUpdate = new HashMap<PageTransactionInfo, AtomicInteger>(); - - for (PagedMessage pagedMessage : pagedMessages) + public FinishPageMessageOperation(final PageTransactionInfo pageTransaction) { - ServerMessage message = pagedMessage.getMessage(storageManager); + this.pageTransaction = pageTransaction; + } + + public void afterCommit(final Transaction tx) + { + // If part of the transaction goes to the queue, and part goes to paging, we can't let depage start for the + // transaction until all the messages were added to the queue + // or else we could deliver the messages out of order - if (message.isLargeMessage()) + if (pageTransaction != null) { - LargeServerMessage largeMsg = (LargeServerMessage)message; - if (!largeMsg.isFileExists()) - { - PagingStoreImpl.log.warn("File for large message " + largeMsg.getMessageID() + - " doesn't exist, so ignoring depage for this large message"); - continue; - } + pageTransaction.commit(); } + } - final long transactionIdDuringPaging = pagedMessage.getTransactionID(); + public void afterPrepare(final Transaction tx) + { + } - PageTransactionInfo pageUserTransaction = null; - AtomicInteger countPageTX = null; - - if (transactionIdDuringPaging >= 0) + public void afterRollback(final Transaction tx) + { + if (tx.getState() == State.PREPARED && pageTransaction != null) { - pageUserTransaction = pagingManager.getTransaction(transactionIdDuringPaging); - - if (pageUserTransaction == null) - { - // This is not supposed to happen - PagingStoreImpl.log.warn("Transaction " + pagedMessage.getTransactionID() + - " used during paging not found"); - continue; - } - else - { - countPageTX = pageTransactionsToUpdate.get(pageUserTransaction); - if (countPageTX == null) - { - countPageTX = new AtomicInteger(); - pageTransactionsToUpdate.put(pageUserTransaction, countPageTX); - } - - // This is to avoid a race condition where messages are depaged - // before the commit arrived - - while (running && !pageUserTransaction.waitCompletion(500)) - { - // This is just to give us a chance to interrupt the process.. - // if we start a shutdown in the middle of transactions, the commit/rollback may never come, delaying - // the shutdown of the server - if (PagingStoreImpl.isTrace) - { - PagingStoreImpl.trace("Waiting pageTransaction to complete"); - } - } - - if (!running) - { - break; - } - - if (!pageUserTransaction.isCommit()) - { - if (PagingStoreImpl.isTrace) - { - PagingStoreImpl.trace("Rollback was called after prepare, ignoring message " + message); - } - continue; - } - } - + pageTransaction.rollback(); } + } - postOffice.route(message, depageTransaction, false); + public void beforeCommit(final Transaction tx) throws Exception + { + storePageTX(tx); + } - // This means the page is duplicated. So we need to ignore this - if (depageTransaction.getState() == State.ROLLBACK_ONLY) + public void beforePrepare(final Transaction tx) throws Exception + { + storePageTX(tx); + } + + private void storePageTX(final Transaction tx) throws Exception + { + if (!stored) { - break; + tx.setContainsPersistent(); + pageTransaction.store(storageManager, pagingManager, tx); + stored = true; } - - // Update information about transactions - // This needs to be done after routing because of duplication detection - if (pageUserTransaction != null && message.isDurable()) - { - countPageTX.incrementAndGet(); - } } - if (!running) + public void beforeRollback(final Transaction tx) throws Exception { - depageTransaction.rollback(); - return false; } - for (Map.Entry<PageTransactionInfo, AtomicInteger> entry : pageTransactionsToUpdate.entrySet()) + /* (non-Javadoc) + * @see org.hornetq.core.transaction.TransactionOperation#getRelatedMessageReferences() + */ + public List<MessageReference> getRelatedMessageReferences() { - // This will set the journal transaction to commit; - depageTransaction.setContainsPersistent(); - - entry.getKey().storeUpdate(storageManager, this.pagingManager, depageTransaction, entry.getValue().intValue()); + return Collections.emptyList(); } - depageTransaction.commit(); - - storageManager.waitOnOperations(); - - if (PagingStoreImpl.isTrace) - { - PagingStoreImpl.trace("Depage committed, running = " + running); - } - - return true; } /** + * 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 + */ + + /** * @param pageId * @return */ @@ -1091,77 +1046,45 @@ return duplicateIdForPage; } - /** - * @return - */ - private boolean isAddressFull(final long nextPageSize) - { - return maxSize > 0 && getAddressSize() + nextPageSize > maxSize; - } + - /** - * startDepaging and clearDepage needs to be atomic. - * We can't use writeLock to this operation as writeLock would still be used by another thread, and still being a valid usage - * @return true if the depage status was cleared - */ - private synchronized boolean clearDepage() - { - final boolean addressFull = isAddressFull(getPageSizeBytes()); - - if (PagingStoreImpl.isTrace) - { - PagingStoreImpl.trace("Clear Depage on Address = " + getStoreName() + - " addressSize = " + - getAddressSize() + - " addressMax " + - maxSize + - " isPaging = " + - isPaging() + - " addressFull = " + - addressFull); - } - - // It should stop the executor when the address is full or when there is nothing else to be depaged - if (addressFull || !isPaging()) - { - depaging.set(false); - return true; - } - else - { - return false; - } - } - private void openNewPage() throws Exception { - currentPageLock.writeLock().lock(); + lock.writeLock().lock(); try { numberOfPages++; - currentPageId++; + int tmpCurrentPageId = currentPageId + 1; - if (currentPageId < firstPageId) - { - firstPageId = currentPageId; - } - if (currentPage != null) { currentPage.close(); } - currentPage = createPage(currentPageId); + currentPage = createPage(tmpCurrentPageId); + LivePageCache pageCache = new LivePageCacheImpl(currentPage); + + currentPage.setLiveCache(pageCache); + + cursorProvider.addPageCache(pageCache); + currentPageSize.set(0); currentPage.open(); + + currentPageId = tmpCurrentPageId; + + if (currentPageId < firstPageId) + { + firstPageId = currentPageId; + } } finally { - currentPageLock.writeLock().unlock(); + lock.writeLock().unlock(); } } @@ -1190,39 +1113,39 @@ // Inner classes ------------------------------------------------- - private class DepageRunnable implements Runnable - { - private final Executor followingExecutor; - - public DepageRunnable(final Executor followingExecutor) + /* private class DepageRunnable implements Runnable { - this.followingExecutor = followingExecutor; - } + private final Executor followingExecutor; - public void run() - { - try + public DepageRunnable(final Executor followingExecutor) { - if (running) + this.followingExecutor = followingExecutor; + } + + public void run() + { + try { - if (!isAddressFull(getPageSizeBytes())) + if (running) { - readPage(); - } + if (!isAddressFull(getPageSizeBytes())) + { + readPage(); + } - // Note: clearDepage is an atomic operation, it needs to be done even if readPage was not executed - // however clearDepage shouldn't be executed if the page-store is being stopped, as stop will be holding - // the lock and this would dead lock - if (running && !clearDepage()) - { - followingExecutor.execute(this); + // Note: clearDepage is an atomic operation, it needs to be done even if readPage was not executed + // however clearDepage shouldn't be executed if the page-store is being stopped, as stop will be holding + // the lock and this would dead lock + if (running && !clearDepage()) + { + followingExecutor.execute(this); + } } } + catch (Throwable e) + { + PagingStoreImpl.log.error(e, e); + } } - catch (Throwable e) - { - PagingStoreImpl.log.error(e, e); - } - } - } + } */ } Modified: trunk/src/main/org/hornetq/core/paging/impl/TestSupportPageStore.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/impl/TestSupportPageStore.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/paging/impl/TestSupportPageStore.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -13,7 +13,6 @@ package org.hornetq.core.paging.impl; -import org.hornetq.core.paging.Page; import org.hornetq.core.paging.PagingStore; /** @@ -23,21 +22,4 @@ */ public interface TestSupportPageStore extends PagingStore { - /** - * Remove the first page from the Writing Queue. - * The file will still exist until Page.delete is called, - * So, case the system is reloaded the same Page will be loaded back if delete is not called. - * - * @throws Exception - * - * Note: This should still be part of the interface, even though HornetQ only uses through the - */ - Page depage() throws Exception; - - void forceAnotherPage() throws Exception; - - /** @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/persistence/StorageManager.java =================================================================== --- trunk/src/main/org/hornetq/core/persistence/StorageManager.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/persistence/StorageManager.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -13,7 +13,6 @@ package org.hornetq.core.persistence; -import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.concurrent.Executor; @@ -27,6 +26,7 @@ import org.hornetq.core.paging.PageTransactionInfo; import org.hornetq.core.paging.PagedMessage; import org.hornetq.core.paging.PagingManager; +import org.hornetq.core.paging.cursor.PagePosition; import org.hornetq.core.persistence.config.PersistedAddressSetting; import org.hornetq.core.persistence.config.PersistedRoles; import org.hornetq.core.postoffice.Binding; @@ -98,6 +98,8 @@ void deleteMessage(long messageID) throws Exception; void storeAcknowledge(long queueID, long messageID) throws Exception; + + void storeCursorAcknowledge(long queueID, PagePosition position) throws Exception; void updateDeliveryCount(MessageReference ref) throws Exception; @@ -113,6 +115,10 @@ void storeAcknowledgeTransactional(long txID, long queueID, long messageID) throws Exception; + void storeCursorAcknowledgeTransactional(long txID, long queueID, PagePosition position) throws Exception; + + void deleteCursorAcknowledgeTransactional(long txID, long ackID) throws Exception; + void updateScheduledDeliveryTimeTransactional(long txID, MessageReference ref) throws Exception; void deleteMessageTransactional(long txID, long queueID, long messageID) throws Exception; @@ -136,6 +142,8 @@ void storePageTransaction(long txID, PageTransactionInfo pageTransaction) throws Exception; void updatePageTransaction(long txID, PageTransactionInfo pageTransaction, int depage) throws Exception; + + void updatePageTransaction(PageTransactionInfo pageTransaction, int depage) throws Exception; void deletePageTransactional(long recordID) throws Exception; @@ -147,6 +155,7 @@ final PagingManager pagingManager, final ResourceManager resourceManager, final Map<Long, Queue> queues, + Map<Long, QueueBindingInfo> queueInfos, final Map<SimpleString, List<Pair<byte[], Long>>> duplicateIDMap) throws Exception; long storeHeuristicCompletion(Xid xid, boolean isCommit) throws Exception; Modified: trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java =================================================================== --- trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -50,6 +50,10 @@ import org.hornetq.core.paging.PageTransactionInfo; import org.hornetq.core.paging.PagedMessage; import org.hornetq.core.paging.PagingManager; +import org.hornetq.core.paging.PagingStore; +import org.hornetq.core.paging.cursor.PageSubscription; +import org.hornetq.core.paging.cursor.PagePosition; +import org.hornetq.core.paging.cursor.impl.PagePositionImpl; import org.hornetq.core.paging.impl.PageTransactionInfoImpl; import org.hornetq.core.persistence.GroupingInfo; import org.hornetq.core.persistence.OperationContext; @@ -134,6 +138,8 @@ public static final byte HEURISTIC_COMPLETION = 38; + public static final byte ACKNOWLEDGE_CURSOR = 39; + private UUID persistentID; private final BatchingIDGenerator idGenerator; @@ -508,7 +514,16 @@ syncNonTransactional, getContext(syncNonTransactional)); } + + public void storeCursorAcknowledge(long queueID, PagePosition position) throws Exception + { + long ackID = idGenerator.generateID(); + position.setRecordID(ackID); + messageJournal.appendAddRecord(ackID, ACKNOWLEDGE_CURSOR, new CursorAckRecordEncoding(queueID, position), syncNonTransactional, getContext(syncNonTransactional)); + } + + public void deleteMessage(final long messageID) throws Exception { // Messages are deleted on postACK, one after another. @@ -591,6 +606,16 @@ depages)); } + public void updatePageTransaction(final PageTransactionInfo pageTransaction, final int depages) throws Exception + { + messageJournal.appendUpdateRecord(pageTransaction.getRecordID(), + JournalStorageManager.PAGE_TRANSACTION, + new PageUpdateTXEncoding(pageTransaction.getTransactionID(), + depages), + syncNonTransactional, + getContext(syncNonTransactional)); + } + public void storeReferenceTransactional(final long txID, final long queueID, final long messageID) throws Exception { messageJournal.appendUpdateRecordTransactional(txID, @@ -607,6 +632,26 @@ new RefEncoding(queueID)); } + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#storeCursorAcknowledgeTransactional(long, long, org.hornetq.core.paging.cursor.PagePosition) + */ + public void storeCursorAcknowledgeTransactional(long txID, long queueID, PagePosition position) throws Exception + { + long ackID = idGenerator.generateID(); + position.setRecordID(ackID); + messageJournal.appendAddRecordTransactional(txID, ackID, ACKNOWLEDGE_CURSOR, new CursorAckRecordEncoding(queueID, position)); + } + + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#deleteCursorAcknowledgeTransactional(long, long) + */ + public void deleteCursorAcknowledgeTransactional(long txID, long ackID) throws Exception + { + messageJournal.appendDeleteRecordTransactional(txID, ackID); + } + + public long storeHeuristicCompletion(final Xid xid, final boolean isCommit) throws Exception { long id = generateUniqueID(); @@ -763,6 +808,7 @@ final PagingManager pagingManager, final ResourceManager resourceManager, final Map<Long, Queue> queues, + Map<Long, QueueBindingInfo> queueInfos, final Map<SimpleString, List<Pair<byte[], Long>>> duplicateIDMap) throws Exception { List<RecordInfo> records = new ArrayList<RecordInfo>(); @@ -912,7 +958,7 @@ PageTransactionInfo pageTX = pagingManager.getTransaction(pageUpdate.pageTX); - pageTX.update(pageUpdate.recods, null, null); + pageTX.onUpdate(pageUpdate.recods, null, null); } else { @@ -979,6 +1025,29 @@ resourceManager.putHeuristicCompletion(record.id, encoding.xid, encoding.isCommit); break; } + case ACKNOWLEDGE_CURSOR: + { + CursorAckRecordEncoding encoding = new CursorAckRecordEncoding(); + encoding.decode(buff); + + encoding.position.setRecordID(record.id); + + QueueBindingInfo queueInfo = queueInfos.get(encoding.queueID); + + if (queueInfo != null) + { + SimpleString address = queueInfo.getAddress(); + PagingStore store = pagingManager.getPageStore(address); + PageSubscription cursor = store.getCursorProvier().getSubscription(encoding.queueID); + cursor.reloadACK(encoding.position); + } + else + { + log.warn("Can't find queue " + encoding.queueID + " while reloading ACKNOWLEDGE_CURSOR"); + } + + break; + } default: { throw new IllegalStateException("Invalid record type " + recordType); @@ -1475,8 +1544,6 @@ pageTransactionInfo.decode(buff); - pageTransactionInfo.markIncomplete(); - tx.putProperty(TransactionPropertyIndexes.PAGE_TRANSACTION, pageTransactionInfo); pagingManager.addTransaction(pageTransactionInfo); @@ -1513,6 +1580,12 @@ break; } + case ACKNOWLEDGE_CURSOR: + { + // TODO: implement and test this case + // and make sure the rollback will work well also + break; + } default: { JournalStorageManager.log.warn("InternalError: Record type " + recordType + @@ -2201,6 +2274,7 @@ } } + private static final class AddMessageRecord { @@ -2214,8 +2288,53 @@ long scheduledDeliveryTime; int deliveryCount; + } - boolean referenced = false; + private static final class CursorAckRecordEncoding implements EncodingSupport + { + public CursorAckRecordEncoding(final long queueID, final PagePosition position) + { + this.queueID = queueID; + this.position = position; + } + + public CursorAckRecordEncoding() + { + this.position = new PagePositionImpl(); + } + + long queueID; + + PagePosition position; + + /* (non-Javadoc) + * @see org.hornetq.core.journal.EncodingSupport#getEncodeSize() + */ + public int getEncodeSize() + { + return DataConstants.SIZE_LONG + DataConstants.SIZE_LONG + DataConstants.SIZE_INT; + } + + /* (non-Javadoc) + * @see org.hornetq.core.journal.EncodingSupport#encode(org.hornetq.api.core.HornetQBuffer) + */ + public void encode(HornetQBuffer buffer) + { + buffer.writeLong(queueID); + buffer.writeLong(position.getPageNr()); + buffer.writeInt(position.getMessageNr()); + } + + /* (non-Javadoc) + * @see org.hornetq.core.journal.EncodingSupport#decode(org.hornetq.api.core.HornetQBuffer) + */ + public void decode(HornetQBuffer buffer) + { + queueID = buffer.readLong(); + long pageNR = buffer.readLong(); + int messageNR = buffer.readInt(); + this.position = new PagePositionImpl(pageNR, messageNR); + } } private class LargeMessageTXFailureCallback implements TransactionFailureCallback @@ -2254,5 +2373,4 @@ } } - } Modified: trunk/src/main/org/hornetq/core/persistence/impl/journal/OperationContextImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/persistence/impl/journal/OperationContextImpl.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/persistence/impl/journal/OperationContextImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -280,8 +280,12 @@ if (timeout == 0) { waitCallback.waitCompletion(); + return true; } - return waitCallback.waitCompletion(timeout); + else + { + return waitCallback.waitCompletion(timeout); + } } } Modified: trunk/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java =================================================================== --- trunk/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -31,6 +31,7 @@ import org.hornetq.core.paging.PageTransactionInfo; import org.hornetq.core.paging.PagedMessage; import org.hornetq.core.paging.PagingManager; +import org.hornetq.core.paging.cursor.PagePosition; import org.hornetq.core.persistence.GroupingInfo; import org.hornetq.core.persistence.OperationContext; import org.hornetq.core.persistence.QueueBindingInfo; @@ -267,6 +268,7 @@ final PagingManager pagingManager, final ResourceManager resourceManager, final Map<Long, Queue> queues, + Map<Long, QueueBindingInfo> queueInfos, final Map<SimpleString, List<Pair<byte[], Long>>> duplicateIDMap) throws Exception { return new JournalLoadInformation(); @@ -450,4 +452,40 @@ { } + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#storeCursorAcknowledge(long, org.hornetq.core.paging.cursor.PagePosition) + */ + public void storeCursorAcknowledge(long queueID, PagePosition position) + { + // TODO Auto-generated method stub + + } + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#storeCursorAcknowledgeTransactional(long, long, org.hornetq.core.paging.cursor.PagePosition) + */ + public void storeCursorAcknowledgeTransactional(long txID, long queueID, PagePosition position) + { + // TODO Auto-generated method stub + + } + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#deleteCursorAcknowledgeTransactional(long, long) + */ + public void deleteCursorAcknowledgeTransactional(long txID, long ackID) throws Exception + { + // TODO Auto-generated method stub + + } + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#updatePageTransaction(org.hornetq.core.paging.PageTransactionInfo, int) + */ + public void updatePageTransaction(PageTransactionInfo pageTransaction, int depage) throws Exception + { + // TODO Auto-generated method stub + + } + } Modified: trunk/src/main/org/hornetq/core/postoffice/AddressManager.java =================================================================== --- trunk/src/main/org/hornetq/core/postoffice/AddressManager.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/postoffice/AddressManager.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -25,13 +25,13 @@ */ public interface AddressManager { - boolean addBinding(Binding binding); + boolean addBinding(Binding binding) throws Exception; - Binding removeBinding(SimpleString uniqueName); + Binding removeBinding(SimpleString uniqueName) throws Exception; - Bindings getBindingsForRoutingAddress(SimpleString address); + Bindings getBindingsForRoutingAddress(SimpleString address) throws Exception; - Bindings getMatchingBindings(SimpleString address); + Bindings getMatchingBindings(SimpleString address) throws Exception; void clear(); Modified: trunk/src/main/org/hornetq/core/postoffice/Bindings.java =================================================================== --- trunk/src/main/org/hornetq/core/postoffice/Bindings.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/postoffice/Bindings.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -15,6 +15,7 @@ import java.util.Collection; +import org.hornetq.core.paging.PagingStore; import org.hornetq.core.server.Queue; import org.hornetq.core.server.RoutingContext; import org.hornetq.core.server.ServerMessage; @@ -41,4 +42,6 @@ boolean redistribute(ServerMessage message, Queue originatingQueue, RoutingContext context) throws Exception; void route(ServerMessage message, RoutingContext context) throws Exception; + + PagingStore getPagingStore(); } Modified: trunk/src/main/org/hornetq/core/postoffice/BindingsFactory.java =================================================================== --- trunk/src/main/org/hornetq/core/postoffice/BindingsFactory.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/postoffice/BindingsFactory.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -12,6 +12,8 @@ */ package org.hornetq.core.postoffice; +import org.hornetq.api.core.SimpleString; + /** * A factory for creating bindings * @@ -19,5 +21,5 @@ */ public interface BindingsFactory { - Bindings createBindings(); + Bindings createBindings(SimpleString address) throws Exception; } Modified: trunk/src/main/org/hornetq/core/postoffice/PostOffice.java =================================================================== --- trunk/src/main/org/hornetq/core/postoffice/PostOffice.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/postoffice/PostOffice.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -47,7 +47,7 @@ Binding getBinding(SimpleString uniqueName); - Bindings getMatchingBindings(SimpleString address); + Bindings getMatchingBindings(SimpleString address) throws Exception; void route(ServerMessage message, boolean direct) throws Exception; Modified: trunk/src/main/org/hornetq/core/postoffice/impl/BindingsImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/postoffice/impl/BindingsImpl.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/postoffice/impl/BindingsImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -27,6 +27,7 @@ import org.hornetq.core.filter.Filter; import org.hornetq.core.logging.Logger; import org.hornetq.core.message.impl.MessageImpl; +import org.hornetq.core.paging.PagingStore; import org.hornetq.core.postoffice.Binding; import org.hornetq.core.postoffice.Bindings; import org.hornetq.core.server.Queue; @@ -60,10 +61,13 @@ private volatile boolean routeWhenNoConsumers; private final GroupingHandler groupingHandler; + + private final PagingStore pageStore; - public BindingsImpl(final GroupingHandler groupingHandler) + public BindingsImpl(final GroupingHandler groupingHandler, final PagingStore pageStore) { this.groupingHandler = groupingHandler; + this.pageStore = pageStore; } public void setRouteWhenNoConsumers(final boolean routeWhenNoConsumers) @@ -218,7 +222,12 @@ return false; } } - + + public PagingStore getPagingStore() + { + return pageStore; + } + public void route(final ServerMessage message, final RoutingContext context) throws Exception { boolean routed = false; Modified: trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -14,10 +14,13 @@ package org.hornetq.core.postoffice.impl; import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -47,6 +50,7 @@ import org.hornetq.core.server.MessageReference; import org.hornetq.core.server.Queue; import org.hornetq.core.server.QueueFactory; +import org.hornetq.core.server.RouteContextList; import org.hornetq.core.server.RoutingContext; import org.hornetq.core.server.ServerMessage; import org.hornetq.core.server.impl.RoutingContextImpl; @@ -58,6 +62,7 @@ import org.hornetq.core.settings.impl.AddressSettings; import org.hornetq.core.transaction.Transaction; import org.hornetq.core.transaction.TransactionOperation; +import org.hornetq.core.transaction.TransactionOperationAbstract; import org.hornetq.core.transaction.TransactionPropertyIndexes; import org.hornetq.core.transaction.Transaction.State; import org.hornetq.core.transaction.impl.TransactionImpl; @@ -508,13 +513,13 @@ return binding; } - public Bindings getBindingsForAddress(final SimpleString address) + public Bindings getBindingsForAddress(final SimpleString address) throws Exception { Bindings bindings = addressManager.getBindingsForRoutingAddress(address); if (bindings == null) { - bindings = createBindings(); + bindings = createBindings(address); } return bindings; @@ -525,7 +530,7 @@ return addressManager.getBinding(name); } - public Bindings getMatchingBindings(final SimpleString address) + public Bindings getMatchingBindings(final SimpleString address) throws Exception { return addressManager.getMatchingBindings(address); } @@ -606,47 +611,13 @@ cache.addToCache(duplicateIDBytes, context.getTransaction()); } - if (context.getTransaction() == null) - { - if (message.page()) - { - return; - } - } - else - { - Transaction tx = context.getTransaction(); - - boolean depage = tx.getProperty(TransactionPropertyIndexes.IS_DEPAGE) != null; - - // if the TX paged at least one message on a give address, all the other addresses should also go towards - // paging cache now - boolean alreadyPaging = false; - - if (tx.isPaging()) - { - alreadyPaging = getPageOperation(tx).isPaging(message.getAddress()); - } - - if (!depage && message.storeIsPaging() || alreadyPaging) - { - tx.setPaging(true); - getPageOperation(tx).addMessageToPage(message); - if (startedTx) - { - tx.commit(); - } - - return; - } - } - Bindings bindings = addressManager.getBindingsForRoutingAddress(address); if (bindings != null) { bindings.route(message, context); } + if (context.getQueueCount() == 0) { // Send to DLA if appropriate @@ -867,89 +838,133 @@ processRoute(message, context, false); } } - - private void processRoute(final ServerMessage message, final RoutingContext context, final boolean direct) throws Exception + + + private class PageDelivery extends TransactionOperationAbstract { - final List<MessageReference> refs = new ArrayList<MessageReference>(); - - Transaction tx = context.getTransaction(); - - for (Queue queue : context.getNonDurableQueues()) + private Set<Queue> queues = new HashSet<Queue>(); + + public void addQueues(List<Queue> queueList) { - MessageReference reference = message.createReference(queue); - - refs.add(reference); - - if (message.containsProperty(Message.HDR_SCHEDULED_DELIVERY_TIME)) + queues.addAll(queueList); + } + + public void afterCommit(Transaction tx) + { + // We need to try delivering async after paging, or nothing may start a delivery after paging since nothing is going towards the queues + // The queue will try to depage case it's empty + for (Queue queue : queues) { - Long scheduledDeliveryTime = message.getLongProperty(Message.HDR_SCHEDULED_DELIVERY_TIME); - - reference.setScheduledDeliveryTime(scheduledDeliveryTime); + queue.deliverAsync(); } - - message.incrementRefCount(); } - Iterator<Queue> iter = context.getDurableQueues().iterator(); - - while (iter.hasNext()) + /* (non-Javadoc) + * @see org.hornetq.core.transaction.TransactionOperation#getRelatedMessageReferences() + */ + public List<MessageReference> getRelatedMessageReferences() { - Queue queue = iter.next(); + return Collections.emptyList(); + } + + } - MessageReference reference = message.createReference(queue); + private void processRoute(final ServerMessage message, final RoutingContext context, final boolean direct) throws Exception + { + final List<MessageReference> refs = new ArrayList<MessageReference>(); - refs.add(reference); - - if (message.containsProperty(Message.HDR_SCHEDULED_DELIVERY_TIME)) + Transaction tx = context.getTransaction(); + + + for (Map.Entry<SimpleString, RouteContextList> entry: context.getContexListing().entrySet()) + { + PagingStore store = pagingManager.getPageStore(entry.getKey()); + + if (store.page(message, context, entry.getValue())) { - Long scheduledDeliveryTime = message.getLongProperty(Message.HDR_SCHEDULED_DELIVERY_TIME); - - reference.setScheduledDeliveryTime(scheduledDeliveryTime); + + // We need to kick delivery so the Queues may check for the cursors case they are empty + schedulePageDelivery(tx, entry); + continue; } - - if (message.isDurable()) + + for (Queue queue : entry.getValue().getNonDurableQueues()) { - int durableRefCount = message.incrementDurableRefCount(); - - if (durableRefCount == 1) + MessageReference reference = message.createReference(queue); + + refs.add(reference); + + if (message.containsProperty(Message.HDR_SCHEDULED_DELIVERY_TIME)) { - if (tx != null) - { - storageManager.storeMessageTransactional(tx.getID(), message); - } - else - { - storageManager.storeMessage(message); - } + Long scheduledDeliveryTime = message.getLongProperty(Message.HDR_SCHEDULED_DELIVERY_TIME); + + reference.setScheduledDeliveryTime(scheduledDeliveryTime); } - - if (tx != null) + + message.incrementRefCount(); + } + + Iterator<Queue> iter = entry.getValue().getDurableQueues().iterator(); + + while (iter.hasNext()) + { + Queue queue = iter.next(); + + MessageReference reference = message.createReference(queue); + + refs.add(reference); + + if (message.containsProperty(Message.HDR_SCHEDULED_DELIVERY_TIME)) { - storageManager.storeReferenceTransactional(tx.getID(), queue.getID(), message.getMessageID()); - - tx.setContainsPersistent(); + Long scheduledDeliveryTime = message.getLongProperty(Message.HDR_SCHEDULED_DELIVERY_TIME); + + reference.setScheduledDeliveryTime(scheduledDeliveryTime); } - else + + if (message.isDurable()) { - storageManager.storeReference(queue.getID(), message.getMessageID(), !iter.hasNext()); - } - - if (message.containsProperty(Message.HDR_SCHEDULED_DELIVERY_TIME)) - { + int durableRefCount = message.incrementDurableRefCount(); + + if (durableRefCount == 1) + { + if (tx != null) + { + storageManager.storeMessageTransactional(tx.getID(), message); + } + else + { + storageManager.storeMessage(message); + } + } + if (tx != null) { - storageManager.updateScheduledDeliveryTimeTransactional(tx.getID(), reference); + storageManager.storeReferenceTransactional(tx.getID(), queue.getID(), message.getMessageID()); + + tx.setContainsPersistent(); } else { - storageManager.updateScheduledDeliveryTime(reference); + storageManager.storeReference(queue.getID(), message.getMessageID(), !iter.hasNext()); } + + if (message.containsProperty(Message.HDR_SCHEDULED_DELIVERY_TIME)) + { + if (tx != null) + { + storageManager.updateScheduledDeliveryTimeTransactional(tx.getID(), reference); + } + else + { + storageManager.updateScheduledDeliveryTime(reference); + } + } } + + message.incrementRefCount(); } - - message.incrementRefCount(); } - + if (tx != null) { tx.addOperation(new AddOperation(refs)); @@ -976,6 +991,56 @@ } /** + * This will kick a delivery async on the queue, so the queue may have a chance to depage messages + * @param tx + * @param entry + */ + private void schedulePageDelivery(Transaction tx, Map.Entry<SimpleString, RouteContextList> entry) + { + if (tx != null) + { + PageDelivery delivery = (PageDelivery)tx.getProperty(TransactionPropertyIndexes.PAGE_DELIVERY); + if (delivery == null) + { + delivery = new PageDelivery(); + tx.putProperty(TransactionPropertyIndexes.PAGE_DELIVERY, delivery); + tx.addOperation(delivery); + } + + delivery.addQueues(entry.getValue().getDurableQueues()); + delivery.addQueues(entry.getValue().getNonDurableQueues()); + } + else + { + + List<Queue> durableQueues = entry.getValue().getDurableQueues(); + List<Queue> nonDurableQueues = entry.getValue().getNonDurableQueues(); + + final List<Queue> queues = new ArrayList<Queue>(durableQueues.size() + nonDurableQueues.size()); + + queues.addAll(durableQueues); + queues.addAll(nonDurableQueues); + + storageManager.afterCompleteOperations(new IOAsyncTask() + { + + public void onError(int errorCode, String errorMessage) + { + } + + public void done() + { + for (Queue queue : queues) + { + // in case of paging, we need to kick asynchronous delivery to try delivering + queue.deliverAsync(); + } + } + }); + } + } + + /** * @param refs */ private void addReferences(final List<MessageReference> refs, final boolean direct) @@ -1014,27 +1079,6 @@ return message; } - private final PageMessageOperation getPageOperation(final Transaction tx) - { - // you could have races on the case two sessions using the same XID - // so this whole operation needs to be atomic per TX - synchronized (tx) - { - PageMessageOperation oper = (PageMessageOperation)tx.getProperty(TransactionPropertyIndexes.PAGE_MESSAGES_OPERATION); - - if (oper == null) - { - oper = new PageMessageOperation(); - - tx.putProperty(TransactionPropertyIndexes.PAGE_MESSAGES_OPERATION, oper); - - tx.addOperation(oper); - } - - return oper; - } - } - private class Reaper implements Runnable { private volatile boolean closed = false; @@ -1114,189 +1158,6 @@ } } - private class PageMessageOperation implements TransactionOperation - { - private final HashMap<SimpleString, Pair<PagingStore, List<ServerMessage>>> pagingData = new HashMap<SimpleString, Pair<PagingStore, List<ServerMessage>>>(); - - private Transaction subTX = null; - - void addMessageToPage(final ServerMessage message) - { - Pair<PagingStore, List<ServerMessage>> pagePair = pagingData.get(message.getAddress()); - if (pagePair == null) - { - pagePair = new Pair<PagingStore, List<ServerMessage>>(message.getPagingStore(), - new ArrayList<ServerMessage>()); - pagingData.put(message.getAddress(), pagePair); - } - - pagePair.b.add(message); - } - - boolean isPaging(final SimpleString address) - { - return pagingData.get(address) != null; - } - - public void afterCommit(final Transaction tx) - { - // If part of the transaction goes to the queue, and part goes to paging, we can't let depage start for the - // transaction until all the messages were added to the queue - // or else we could deliver the messages out of order - - PageTransactionInfo pageTransaction = (PageTransactionInfo)tx.getProperty(TransactionPropertyIndexes.PAGE_TRANSACTION); - - if (pageTransaction != null) - { - pageTransaction.commit(); - } - - if (subTX != null) - { - subTX.afterCommit(); - } - } - - public void afterPrepare(final Transaction tx) - { - if (subTX != null) - { - subTX.afterPrepare(); - } - } - - public void afterRollback(final Transaction tx) - { - PageTransactionInfo pageTransaction = (PageTransactionInfo)tx.getProperty(TransactionPropertyIndexes.PAGE_TRANSACTION); - - if (tx.getState() == State.PREPARED && pageTransaction != null) - { - pageTransaction.rollback(); - } - - if (subTX != null) - { - subTX.afterRollback(); - } - } - - public void beforeCommit(final Transaction tx) throws Exception - { - if (tx.getState() != Transaction.State.PREPARED) - { - pageMessages(tx); - } - - if (subTX != null) - { - subTX.beforeCommit(); - } - - } - - public void beforePrepare(final Transaction tx) throws Exception - { - pageMessages(tx); - - if (subTX != null) - { - subTX.beforePrepare(); - } - } - - public void beforeRollback(final Transaction tx) throws Exception - { - if (subTX != null) - { - subTX.beforeRollback(); - } - } - - public List<MessageReference> getRelatedMessageReferences() - { - return null; - } - - private void pageMessages(final Transaction tx) throws Exception - { - if (!pagingData.isEmpty()) - { - PageTransactionInfo pageTransaction = (PageTransactionInfo)tx.getProperty(TransactionPropertyIndexes.PAGE_TRANSACTION); - - if (pageTransaction == null) - { - pageTransaction = new PageTransactionInfoImpl(tx.getID()); - - tx.putProperty(TransactionPropertyIndexes.PAGE_TRANSACTION, pageTransaction); - - // To avoid a race condition where depage happens before the transaction is completed, we need to inform - // the pager about this transaction is being processed - pagingManager.addTransaction(pageTransaction); - } - - boolean pagingPersistent = false; - - ArrayList<ServerMessage> nonPagedMessages = null; - - for (Pair<PagingStore, List<ServerMessage>> pair : pagingData.values()) - { - - if (!pair.a.page(pair.b, tx.getID())) - { - if (nonPagedMessages == null) - { - nonPagedMessages = new ArrayList<ServerMessage>(); - } - nonPagedMessages.addAll(pair.b); - } - - for (ServerMessage msg : pair.b) - { - if (msg.isDurable()) - { - pageTransaction.increment(); - pagingPersistent = true; - } - } - } - - if (nonPagedMessages != null) - { - for (ServerMessage message : nonPagedMessages) - { - // This could happen when the PageStore left the pageState - // we create a copy of the transaction so that messages are routed with the same tx ID. - // but we can not use directly the tx as it has already its own set of TransactionOperations - if (subTX == null) - { - subTX = tx.copy(); - } - - route(message, subTX, false); - - if (subTX.isContainsPersistent()) - { - // The route wouldn't be able to update the persistent flag on the main TX - // If we don't do this we would eventually miss a commit record - tx.setContainsPersistent(); - } - } - } - - if (pagingPersistent) - { - tx.setContainsPersistent(); - for (Pair<PagingStore, List<ServerMessage>> pair : pagingData.values()) - { - pair.a.sync(); - } - - pageTransaction.store(storageManager, pagingManager, tx); - } - } - } - } - private class AddOperation implements TransactionOperation { private final List<MessageReference> refs; @@ -1353,8 +1214,8 @@ } } - public Bindings createBindings() + public Bindings createBindings(final SimpleString address) throws Exception { - return new BindingsImpl(server.getGroupingHandler()); + return new BindingsImpl(server.getGroupingHandler(), pagingManager.getPageStore(address)); } } Modified: trunk/src/main/org/hornetq/core/postoffice/impl/SimpleAddressManager.java =================================================================== --- trunk/src/main/org/hornetq/core/postoffice/impl/SimpleAddressManager.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/postoffice/impl/SimpleAddressManager.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -46,7 +46,7 @@ this.bindingsFactory = bindingsFactory; } - public boolean addBinding(final Binding binding) + public boolean addBinding(final Binding binding) throws Exception { if (nameMap.putIfAbsent(binding.getUniqueName(), binding) != null) { @@ -56,7 +56,7 @@ return addMappingInternal(binding.getAddress(), binding); } - public Binding removeBinding(final SimpleString uniqueName) + public Binding removeBinding(final SimpleString uniqueName) throws Exception { Binding binding = nameMap.remove(uniqueName); @@ -70,7 +70,7 @@ return binding; } - public Bindings getBindingsForRoutingAddress(final SimpleString address) + public Bindings getBindingsForRoutingAddress(final SimpleString address) throws Exception { return mappings.get(address); } @@ -85,11 +85,11 @@ return nameMap; } - public Bindings getMatchingBindings(final SimpleString address) + public Bindings getMatchingBindings(final SimpleString address) throws Exception { Address add = new AddressImpl(address); - Bindings bindings = bindingsFactory.createBindings(); + Bindings bindings = bindingsFactory.createBindings(address); for (Binding binding : nameMap.values()) { @@ -149,7 +149,7 @@ return theBinding; } - protected boolean addMappingInternal(final SimpleString address, final Binding binding) + protected boolean addMappingInternal(final SimpleString address, final Binding binding) throws Exception { Bindings bindings = mappings.get(address); @@ -157,7 +157,7 @@ if (bindings == null) { - bindings = bindingsFactory.createBindings(); + bindings = bindingsFactory.createBindings(address); prevBindings = mappings.putIfAbsent(address, bindings); Modified: trunk/src/main/org/hornetq/core/postoffice/impl/WildcardAddressManager.java =================================================================== --- trunk/src/main/org/hornetq/core/postoffice/impl/WildcardAddressManager.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/postoffice/impl/WildcardAddressManager.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -57,7 +57,7 @@ } @Override - public Bindings getBindingsForRoutingAddress(final SimpleString address) + public Bindings getBindingsForRoutingAddress(final SimpleString address) throws Exception { Bindings bindings = super.getBindingsForRoutingAddress(address); @@ -93,7 +93,7 @@ * @return true if the address was a new mapping */ @Override - public boolean addBinding(final Binding binding) + public boolean addBinding(final Binding binding) throws Exception { boolean exists = super.addBinding(binding); if (!exists) @@ -129,7 +129,7 @@ * @return true if this was the last mapping for a specific address */ @Override - public Binding removeBinding(final SimpleString uniqueName) + public Binding removeBinding(final SimpleString uniqueName) throws Exception { Binding binding = super.removeBinding(uniqueName); if (binding != null) @@ -239,7 +239,7 @@ } } - private synchronized void removeAndUpdateAddressMap(final Address address) + private synchronized void removeAndUpdateAddressMap(final Address address) throws Exception { // we only remove if there are no bindings left Bindings bindings = super.getBindingsForRoutingAddress(address.getAddress()); Modified: trunk/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -548,7 +548,8 @@ private void handlePageWrite(final ReplicationPageWriteMessage packet) throws Exception { PagedMessage pgdMessage = packet.getPagedMessage(); - ServerMessage msg = pgdMessage.getMessage(storage); + pgdMessage.initMessage(storage); + ServerMessage msg = pgdMessage.getMessage(); Page page = getPage(msg.getAddress(), packet.getPageNumber()); page.write(pgdMessage); } Modified: trunk/src/main/org/hornetq/core/server/MessageReference.java =================================================================== --- trunk/src/main/org/hornetq/core/server/MessageReference.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/server/MessageReference.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -13,6 +13,8 @@ package org.hornetq.core.server; +import org.hornetq.core.transaction.Transaction; + /** * A reference to a message. * @@ -26,6 +28,9 @@ */ public interface MessageReference { + + boolean isPaged(); + ServerMessage getMessage(); MessageReference copy(Queue queue); @@ -48,6 +53,11 @@ void decrementDeliveryCount(); Queue getQueue(); + + void acknowledge() throws Exception; + + void acknowledge(final Transaction tx) throws Exception; + void handled(); } Modified: trunk/src/main/org/hornetq/core/server/Queue.java =================================================================== --- trunk/src/main/org/hornetq/core/server/Queue.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/server/Queue.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -20,6 +20,7 @@ import org.hornetq.api.core.SimpleString; import org.hornetq.core.filter.Filter; +import org.hornetq.core.paging.cursor.PageSubscription; import org.hornetq.core.transaction.Transaction; /** @@ -38,6 +39,8 @@ long getID(); Filter getFilter(); + + PageSubscription getPageSubscription(); boolean isDurable(); Modified: trunk/src/main/org/hornetq/core/server/QueueFactory.java =================================================================== --- trunk/src/main/org/hornetq/core/server/QueueFactory.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/server/QueueFactory.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -15,6 +15,7 @@ import org.hornetq.api.core.SimpleString; import org.hornetq.core.filter.Filter; +import org.hornetq.core.paging.cursor.PageSubscription; import org.hornetq.core.postoffice.PostOffice; /** @@ -33,6 +34,7 @@ final SimpleString address, SimpleString name, Filter filter, + PageSubscription pageSubscription, boolean durable, boolean temporary); Added: trunk/src/main/org/hornetq/core/server/RouteContextList.java =================================================================== --- trunk/src/main/org/hornetq/core/server/RouteContextList.java (rev 0) +++ trunk/src/main/org/hornetq/core/server/RouteContextList.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -0,0 +1,34 @@ +/* + * Copyright 2010 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.server; + +import java.util.List; + +/** + * This is a simple datatype containing the list of a routing context + * + * @author clebertsuconic + * + * + */ +public interface RouteContextList +{ + + int getNumberOfQueues(); + + List<Queue> getDurableQueues(); + + List<Queue> getNonDurableQueues(); + +} Modified: trunk/src/main/org/hornetq/core/server/RoutingContext.java =================================================================== --- trunk/src/main/org/hornetq/core/server/RoutingContext.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/server/RoutingContext.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -14,13 +14,17 @@ package org.hornetq.core.server; import java.util.List; +import java.util.Map; +import org.hornetq.api.core.Pair; +import org.hornetq.api.core.SimpleString; import org.hornetq.core.transaction.Transaction; /** * A RoutingContext * * @author Tim Fox + * @author Clebert Suconic * * */ @@ -30,14 +34,20 @@ void setTransaction(Transaction transaction); - void addQueue(Queue queue); + void addQueue(SimpleString address, Queue queue); - List<Queue> getNonDurableQueues(); + Map<SimpleString, RouteContextList> getContexListing(); + + RouteContextList getContextListing(SimpleString address); + + List<Queue> getNonDurableQueues(SimpleString address); - List<Queue> getDurableQueues(); + List<Queue> getDurableQueues(SimpleString address); int getQueueCount(); void clear(); + + } Modified: trunk/src/main/org/hornetq/core/server/ServerMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/server/ServerMessage.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/server/ServerMessage.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -56,10 +56,6 @@ PagingStore getPagingStore(); - boolean page() throws Exception; - - boolean page(long transactionID) throws Exception; - boolean storeIsPaging(); void encodeMessageIDToBuffer(); Modified: trunk/src/main/org/hornetq/core/server/ServerSession.java =================================================================== --- trunk/src/main/org/hornetq/core/server/ServerSession.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/server/ServerSession.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -92,7 +92,7 @@ QueueQueryResult executeQueueQuery(SimpleString name) throws Exception; - BindingQueryResult executeBindingQuery(SimpleString address); + BindingQueryResult executeBindingQuery(SimpleString address) throws Exception; void closeConsumer(long consumerID) throws Exception; Modified: trunk/src/main/org/hornetq/core/server/cluster/impl/RemoteQueueBindingImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/cluster/impl/RemoteQueueBindingImpl.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/server/cluster/impl/RemoteQueueBindingImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -16,6 +16,7 @@ import java.nio.ByteBuffer; import java.util.HashMap; import java.util.HashSet; +import java.util.List; import java.util.Map; import java.util.Set; @@ -199,12 +200,14 @@ buff.putLong(remoteQueueID); message.putBytesProperty(idsHeaderName, ids); + + List<Queue> durableQueuesOnContext = context.getDurableQueues(address); - if (!context.getDurableQueues().contains(storeAndForwardQueue)) + if (!durableQueuesOnContext.contains(storeAndForwardQueue)) { // There can be many remote bindings for the same node, we only want to add the message once to // the s & f queue for that node - context.addQueue(storeAndForwardQueue); + context.addQueue(address, storeAndForwardQueue); } } Modified: trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -21,8 +21,8 @@ import java.util.HashSet; import java.util.List; import java.util.Map; + import java.util.Map.Entry; import java.util.Set; -import java.util.Map.Entry; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -61,6 +61,7 @@ import org.hornetq.core.logging.Logger; import org.hornetq.core.management.impl.HornetQServerControlImpl; import org.hornetq.core.paging.PagingManager; +import org.hornetq.core.paging.cursor.PageSubscription; import org.hornetq.core.paging.impl.PagingManagerImpl; import org.hornetq.core.paging.impl.PagingStoreFactoryNIO; import org.hornetq.core.persistence.GroupingInfo; @@ -163,7 +164,7 @@ private volatile QueueFactory queueFactory; private volatile PagingManager pagingManager; - + private volatile PostOffice postOffice; private volatile ExecutorService threadPool; @@ -712,6 +713,8 @@ } Queue queue = (Queue)binding.getBindable(); + + queue.getPageSubscription().close(); if (queue.getConsumerCount() != 0) { @@ -1113,6 +1116,8 @@ deploymentManager.start(); } + pagingManager.processReload(); + pagingManager.resumeDepages(); final ServerInfo dumper = new ServerInfo(this, pagingManager); @@ -1188,15 +1193,21 @@ setNodeID(); Map<Long, Queue> queues = new HashMap<Long, Queue>(); + Map<Long, QueueBindingInfo> queueBindingInfosMap = new HashMap<Long, QueueBindingInfo>(); for (QueueBindingInfo queueBindingInfo : queueBindingInfos) { + queueBindingInfosMap.put(queueBindingInfo.getId(), queueBindingInfo); + Filter filter = FilterImpl.createFilter(queueBindingInfo.getFilterString()); + PageSubscription subscription = pagingManager.getPageStore(queueBindingInfo.getAddress()).getCursorProvier().createSubscription(queueBindingInfo.getId(), filter, true); + Queue queue = queueFactory.createQueue(queueBindingInfo.getId(), queueBindingInfo.getAddress(), queueBindingInfo.getQueueName(), filter, + subscription, true, false); @@ -1208,6 +1219,8 @@ managementService.registerAddress(queueBindingInfo.getAddress()); managementService.registerQueue(queue, queueBindingInfo.getAddress(), storageManager); + + } for (GroupingInfo groupingInfo : groupingInfos) @@ -1226,6 +1239,7 @@ pagingManager, resourceManager, queues, + queueBindingInfosMap, duplicateIDMap); for (Map.Entry<SimpleString, List<Pair<byte[], Long>>> entry : duplicateIDMap.entrySet()) @@ -1328,11 +1342,16 @@ } Filter filter = FilterImpl.createFilter(filterString); + + long queueID = storageManager.generateUniqueID(); - final Queue queue = queueFactory.createQueue(storageManager.generateUniqueID(), + PageSubscription pageSubscription = pagingManager.getPageStore(address).getCursorProvier().createSubscription(queueID, filter, durable); + + final Queue queue = queueFactory.createQueue(queueID, address, queueName, filter, + pageSubscription, durable, temporary); Modified: trunk/src/main/org/hornetq/core/server/impl/LastValueQueue.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/LastValueQueue.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/server/impl/LastValueQueue.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -21,6 +21,7 @@ import org.hornetq.api.core.SimpleString; import org.hornetq.core.filter.Filter; import org.hornetq.core.logging.Logger; +import org.hornetq.core.paging.cursor.PageSubscription; import org.hornetq.core.persistence.StorageManager; import org.hornetq.core.postoffice.PostOffice; import org.hornetq.core.server.MessageReference; @@ -28,6 +29,7 @@ import org.hornetq.core.server.ServerMessage; import org.hornetq.core.settings.HierarchicalRepository; import org.hornetq.core.settings.impl.AddressSettings; +import org.hornetq.core.transaction.Transaction; /** * A queue that will discard messages if a newer message with the same MessageImpl.HDR_LAST_VALUE_NAME property value. @@ -49,6 +51,7 @@ final SimpleString address, final SimpleString name, final Filter filter, + final PageSubscription pageSubscription, final boolean durable, final boolean temporary, final ScheduledExecutorService scheduledExecutor, @@ -61,6 +64,7 @@ address, name, filter, + pageSubscription, durable, temporary, scheduledExecutor, @@ -89,7 +93,7 @@ try { - super.acknowledge(oldRef); + oldRef.acknowledge(); } catch (Exception e) { @@ -225,5 +229,26 @@ { ref.setScheduledDeliveryTime(scheduledDeliveryTime); } + + public boolean isPaged() + { + return false; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.MessageReference#acknowledge(org.hornetq.core.server.MessageReference) + */ + public void acknowledge() throws Exception + { + ref.getQueue().acknowledge(this); + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.MessageReference#acknowledge(org.hornetq.core.transaction.Transaction, org.hornetq.core.server.MessageReference) + */ + public void acknowledge(Transaction tx) throws Exception + { + ref.getQueue().acknowledge(tx, this); + } } } Modified: trunk/src/main/org/hornetq/core/server/impl/MessageReferenceImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/MessageReferenceImpl.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/server/impl/MessageReferenceImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -17,6 +17,7 @@ import org.hornetq.core.server.MessageReference; import org.hornetq.core.server.Queue; import org.hornetq.core.server.ServerMessage; +import org.hornetq.core.transaction.Transaction; import org.hornetq.utils.MemorySize; /** @@ -144,7 +145,29 @@ { queue.referenceHandled(); } + + public boolean isPaged() + { + return false; + } + /* (non-Javadoc) + * @see org.hornetq.core.server.MessageReference#acknowledge(org.hornetq.core.server.MessageReference) + */ + public void acknowledge() throws Exception + { + queue.acknowledge(this); + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.MessageReference#acknowledge(org.hornetq.core.transaction.Transaction, org.hornetq.core.server.MessageReference) + */ + public void acknowledge(Transaction tx) throws Exception + { + queue.acknowledge(tx, this); + } + + // Public -------------------------------------------------------- @Override @@ -154,7 +177,6 @@ "]:" + (getMessage().isDurable() ? "RELIABLE" : "NON-RELIABLE"); } - // Package protected --------------------------------------------- // Protected ----------------------------------------------------- Modified: trunk/src/main/org/hornetq/core/server/impl/QueueFactoryImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/QueueFactoryImpl.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/server/impl/QueueFactoryImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -17,6 +17,7 @@ import org.hornetq.api.core.SimpleString; import org.hornetq.core.filter.Filter; +import org.hornetq.core.paging.cursor.PageSubscription; import org.hornetq.core.persistence.StorageManager; import org.hornetq.core.postoffice.PostOffice; import org.hornetq.core.server.Queue; @@ -69,6 +70,7 @@ final SimpleString address, final SimpleString name, final Filter filter, + final PageSubscription pageSubscription, final boolean durable, final boolean temporary) { @@ -81,6 +83,7 @@ address, name, filter, + pageSubscription, durable, temporary, scheduledExecutor, @@ -95,6 +98,7 @@ address, name, filter, + pageSubscription, durable, temporary, scheduledExecutor, Modified: trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -32,6 +32,8 @@ import org.hornetq.api.core.SimpleString; import org.hornetq.core.filter.Filter; import org.hornetq.core.logging.Logger; +import org.hornetq.core.paging.cursor.PageSubscription; +import org.hornetq.core.paging.cursor.PagedReference; import org.hornetq.core.persistence.StorageManager; import org.hornetq.core.postoffice.Bindings; import org.hornetq.core.postoffice.PostOffice; @@ -89,6 +91,10 @@ private final boolean temporary; private final PostOffice postOffice; + + private final PageSubscription pageSubscription; + + private final LinkedListIterator<PagedReference> pageIterator; private final ConcurrentLinkedQueue<MessageReference> concurrentQueue = new ConcurrentLinkedQueue<MessageReference>(); @@ -106,6 +112,8 @@ private final Runnable deliverRunner = new DeliverRunner(); + private final Runnable depageRunner = new DepageRunner(); + private final StorageManager storageManager; private final HierarchicalRepository<AddressSettings> addressSettingsRepository; @@ -141,11 +149,39 @@ private volatile boolean checkDirect; private volatile boolean directDeliver = true; + + public QueueImpl(final long id, + final SimpleString address, + final SimpleString name, + final Filter filter, + final boolean durable, + final boolean temporary, + final ScheduledExecutorService scheduledExecutor, + final PostOffice postOffice, + final StorageManager storageManager, + final HierarchicalRepository<AddressSettings> addressSettingsRepository, + final Executor executor) + { + this(id, + address, + name, + filter, + null, + durable, + temporary, + scheduledExecutor, + postOffice, + storageManager, + addressSettingsRepository, + executor); + } + public QueueImpl(final long id, final SimpleString address, final SimpleString name, final Filter filter, + final PageSubscription pageSubscription, final boolean durable, final boolean temporary, final ScheduledExecutorService scheduledExecutor, @@ -161,6 +197,8 @@ this.name = name; this.filter = filter; + + this.pageSubscription = pageSubscription; this.durable = durable; @@ -184,6 +222,16 @@ { expiryAddress = null; } + + if (pageSubscription != null) + { + pageSubscription.setQueue(this); + this.pageIterator = pageSubscription.iterator(); + } + else + { + this.pageIterator = null; + } this.executor = executor; @@ -219,7 +267,7 @@ public void route(final ServerMessage message, final RoutingContext context) throws Exception { - context.addQueue(this); + context.addQueue(address, this); } // Queue implementation ---------------------------------------------------------------------------------------- @@ -249,6 +297,11 @@ return id; } + public PageSubscription getPageSubscription() + { + return pageSubscription; + } + public Filter getFilter() { return filter; @@ -301,7 +354,7 @@ // We don't recompute it on every delivery since executing isEmpty is expensive for a ConcurrentQueue if (checkDirect) { - if (direct && !directDeliver && concurrentQueue.isEmpty() && messageReferences.isEmpty()) + if (direct && !directDeliver && concurrentQueue.isEmpty() && messageReferences.isEmpty() && !pageIterator.hasNext() && !pageSubscription.isPaging()) { // We must block on the executor to ensure any async deliveries have completed or we might get out of order // deliveries @@ -627,32 +680,46 @@ public void acknowledge(final MessageReference ref) throws Exception { - ServerMessage message = ref.getMessage(); - - boolean durableRef = message.isDurable() && durable; - - if (durableRef) + if (ref.isPaged()) { - storageManager.storeAcknowledge(id, message.getMessageID()); + pageSubscription.ack((PagedReference)ref); } + else + { + ServerMessage message = ref.getMessage(); + + boolean durableRef = message.isDurable() && durable; + + if (durableRef) + { + storageManager.storeAcknowledge(id, message.getMessageID()); + } + postAcknowledge(ref); + } - postAcknowledge(ref); } public void acknowledge(final Transaction tx, final MessageReference ref) throws Exception { - ServerMessage message = ref.getMessage(); - - boolean durableRef = message.isDurable() && durable; - - if (durableRef) + if (ref.isPaged()) { - storageManager.storeAcknowledgeTransactional(tx.getID(), id, message.getMessageID()); - - tx.setContainsPersistent(); + pageSubscription.ackTx(tx, (PagedReference)ref); } - - getRefsOperation(tx).addAck(ref); + else + { + ServerMessage message = ref.getMessage(); + + boolean durableRef = message.isDurable() && durable; + + if (durableRef) + { + storageManager.storeAcknowledgeTransactional(tx.getID(), id, message.getMessageID()); + + tx.setContainsPersistent(); + } + + getRefsOperation(tx).addAck(ref); + } } public void reacknowledge(final Transaction tx, final MessageReference ref) throws Exception @@ -1187,7 +1254,49 @@ pos = 0; } } + + if (pageIterator != null && messageReferences.size() == 0 && pageIterator.hasNext()) + { + scheduleDepage(); + } } + + private void scheduleDepage() + { + executor.execute(depageRunner); + } + + private void depage() + { + if (paused || consumerList.isEmpty()) + { + return; + } + + int msgsToDeliver = MAX_DELIVERIES_IN_LOOP - (messageReferences.size() + getScheduledCount() + concurrentQueue.size()); + + if (msgsToDeliver > 0) + { + //System.out.println("Depaging " + msgsToDeliver + " messages"); + //System.out.println("Depage " + msgsToDeliver + " now.. there are msgRef = " + messageReferences.size() + " scheduled = " + getScheduledCount() + " concurrentQueue.size() = " + concurrentQueue.size()); + + int nmessages = 0; + while (nmessages < msgsToDeliver && pageIterator.hasNext()) + { + nmessages ++; + addTail(pageIterator.next(), false); + pageIterator.remove(); + } + + //System.out.println("Depaged " + nmessages); + } +// else +// { +// System.out.println("Depaging not being done now.. there are msgRef = " + messageReferences.size() + " scheduled = " + getScheduledCount() + " concurrentQueue.size() = " + concurrentQueue.size()); +// } + + deliverAsync(); + } private void internalAddRedistributor(final Executor executor) { @@ -1212,7 +1321,8 @@ { ServerMessage message = reference.getMessage(); - if (message.isDurable() && durable) + // TODO: DeliveryCount on paging + if (message.isDurable() && durable && !reference.isPaged()) { storageManager.updateDeliveryCount(reference); } @@ -1682,6 +1792,21 @@ } } + private class DepageRunner implements Runnable + { + public void run() + { + try + { + depage(); + } + catch (Exception e) + { + log.error("Failed to deliver", e); + } + } + } + private class ConcurrentPoller implements Runnable { public void run() Modified: trunk/src/main/org/hornetq/core/server/impl/RoutingContextImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/RoutingContextImpl.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/server/impl/RoutingContextImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -14,9 +14,15 @@ package org.hornetq.core.server.impl; import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; import java.util.List; +import java.util.Map; +import org.hornetq.api.core.Pair; +import org.hornetq.api.core.SimpleString; import org.hornetq.core.server.Queue; +import org.hornetq.core.server.RouteContextList; import org.hornetq.core.server.RoutingContext; import org.hornetq.core.transaction.Transaction; @@ -29,10 +35,10 @@ */ public class RoutingContextImpl implements RoutingContext { - private final List<Queue> nonDurableQueues = new ArrayList<Queue>(1); + + // The pair here is Durable and NonDurable + private Map<SimpleString, RouteContextList> map = new HashMap<SimpleString, RouteContextList>(); - private final List<Queue> durableQueues = new ArrayList<Queue>(1); - private Transaction transaction; private int queueCount; @@ -41,35 +47,42 @@ { this.transaction = transaction; } - + public void clear() { transaction = null; - nonDurableQueues.clear(); - - durableQueues.clear(); - + map.clear(); + queueCount = 0; } - public void addQueue(final Queue queue) + public void addQueue(final SimpleString address, final Queue queue) { + + RouteContextList listing = getContextListing(address); + if (queue.isDurable()) { - durableQueues.add(queue); + listing.getDurableQueues().add(queue); } else { - nonDurableQueues.add(queue); + listing.getNonDurableQueues().add(queue); } queueCount++; } - - public void addDurableQueue(final Queue queue) + + public RouteContextList getContextListing(SimpleString address) { - durableQueues.add(queue); + RouteContextList listing = map.get(address); + if (listing == null) + { + listing = new ContextListing(); + map.put(address, listing); + } + return listing; } public Transaction getTransaction() @@ -82,14 +95,14 @@ transaction = tx; } - public List<Queue> getNonDurableQueues() + public List<Queue> getNonDurableQueues(SimpleString address) { - return nonDurableQueues; + return getContextListing(address).getNonDurableQueues(); } - public List<Queue> getDurableQueues() + public List<Queue> getDurableQueues(SimpleString address) { - return durableQueues; + return getContextListing(address).getDurableQueues(); } public int getQueueCount() @@ -97,4 +110,41 @@ return queueCount; } + /* (non-Javadoc) + * @see org.hornetq.core.server.RoutingContext#getAddresses() + */ + public Map<SimpleString, RouteContextList> getContexListing() + { + return this.map; + } + + + private class ContextListing implements RouteContextList + { + private List<Queue> durableQueue = new ArrayList<Queue>(1); + + private List<Queue> nonDurableQueue = new ArrayList<Queue>(1); + + public int getNumberOfQueues() + { + return durableQueue.size() + nonDurableQueue.size(); + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.RouteContextList#getDurableQueues() + */ + public List<Queue> getDurableQueues() + { + return durableQueue; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.RouteContextList#getNonDurableQueues() + */ + public List<Queue> getNonDurableQueues() + { + return nonDurableQueue; + } + } + } Modified: trunk/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -553,11 +553,11 @@ if (autoCommitAcks || tx == null) { - ref.getQueue().acknowledge(ref); + ref.acknowledge(); } else { - ref.getQueue().acknowledge(tx, ref); + ref.acknowledge(tx); } } while (ref.getMessage().getMessageID() != messageID); Modified: trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -249,30 +249,6 @@ return pagingStore; } - public boolean page() throws Exception - { - if (pagingStore != null) - { - return pagingStore.page(this); - } - else - { - return false; - } - } - - public boolean page(final long transactionID) throws Exception - { - if (pagingStore != null) - { - return pagingStore.page(Arrays.asList((ServerMessage)this), transactionID); - } - else - { - return false; - } - } - public boolean storeIsPaging() { if (pagingStore != null) Modified: trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -141,6 +141,7 @@ private Map<String, String> metaData; + // Session's usage should be by definition single threaded, hence it's not needed to use a concurrentHashMap here private Map<SimpleString, UUID> targetAddressInfos = new HashMap<SimpleString, UUID>(); private long creationTime = System.currentTimeMillis(); @@ -485,7 +486,7 @@ return response; } - public BindingQueryResult executeBindingQuery(final SimpleString address) + public BindingQueryResult executeBindingQuery(final SimpleString address) throws Exception { if (address == null) { Added: trunk/src/main/org/hornetq/core/transaction/TransactionOperationAbstract.java =================================================================== --- trunk/src/main/org/hornetq/core/transaction/TransactionOperationAbstract.java (rev 0) +++ trunk/src/main/org/hornetq/core/transaction/TransactionOperationAbstract.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -0,0 +1,87 @@ +/* + * Copyright 2010 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.transaction; + +import java.util.Collections; +import java.util.List; + +import org.hornetq.core.server.MessageReference; + +/** + * Just a helper, when you don't want to implement all the methods on a transaction operation. + * + * @author clebertsuconic + * + * + */ +public abstract class TransactionOperationAbstract implements TransactionOperation +{ + + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + public void beforePrepare(Transaction tx) throws Exception + { + + } + + /** After prepare shouldn't throw any exception. Any verification has to be done on before prepare */ + public void afterPrepare(Transaction tx) + { + + } + + public void beforeCommit(Transaction tx) throws Exception + { + } + + /** After commit shouldn't throw any exception. Any verification has to be done on before commit */ + public void afterCommit(Transaction tx) + { + } + + public void beforeRollback(Transaction tx) throws Exception + { + } + + /** After rollback shouldn't throw any exception. Any verification has to be done on before rollback */ + public void afterRollback(Transaction tx) + { + } + + /* (non-Javadoc) + * @see org.hornetq.core.transaction.TransactionOperation#getRelatedMessageReferences() + */ + public List<MessageReference> getRelatedMessageReferences() + { + return Collections.emptyList(); + } + + + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- + +} Modified: trunk/src/main/org/hornetq/core/transaction/TransactionPropertyIndexes.java =================================================================== --- trunk/src/main/org/hornetq/core/transaction/TransactionPropertyIndexes.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/src/main/org/hornetq/core/transaction/TransactionPropertyIndexes.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -24,11 +24,14 @@ */ public class TransactionPropertyIndexes { - public static final int IS_DEPAGE = 3; + public static final int PAGE_TRANSACTION_UPDATE = 4; + public static final int PAGE_TRANSACTION = 5; public static final int REFS_OPERATION = 6; - public static final int PAGE_MESSAGES_OPERATION = 7; + public static final int PAGE_DELIVERY = 7; + + public static final int PAGE_CURSOR_POSITIONS = 8; } Added: trunk/src/main/org/hornetq/utils/SoftValueHashMap.java =================================================================== --- trunk/src/main/org/hornetq/utils/SoftValueHashMap.java (rev 0) +++ trunk/src/main/org/hornetq/utils/SoftValueHashMap.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -0,0 +1,326 @@ +/* + * Copyright 2010 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.utils; + +import java.lang.ref.ReferenceQueue; +import java.lang.ref.SoftReference; +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +/** + * A SoftValueConcurrentHashMap + * + * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * + */ +public class SoftValueHashMap<K, V> implements Map<K, V> +{ + // 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>(); + + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + + /** + * @return + * @see java.util.Map#size() + */ + public int size() + { + processQueue(); + return mapDelegate.size(); + } + + /** + * @return + * @see java.util.Map#isEmpty() + */ + public boolean isEmpty() + { + processQueue(); + return mapDelegate.isEmpty(); + } + + /** + * @param key + * @return + * @see java.util.Map#containsKey(java.lang.Object) + */ + public boolean containsKey(final Object key) + { + processQueue(); + return mapDelegate.containsKey(key); + } + + /** + * @param value + * @return + * @see java.util.Map#containsValue(java.lang.Object) + */ + public boolean containsValue(final Object value) + { + processQueue(); + for (AggregatedSoftReference valueIter : mapDelegate.values()) + { + V valueElement = valueIter.get(); + if (valueElement != null && value.equals(valueElement)) + { + return true; + } + + } + return false; + } + + /** + * @param key + * @return + * @see java.util.Map#get(java.lang.Object) + */ + public V get(final Object key) + { + processQueue(); + AggregatedSoftReference value = mapDelegate.get(key); + if (value != null) + { + return value.get(); + } + else + { + return null; + } + } + + /** + * @param key + * @param value + * @return + * @see java.util.Map#put(java.lang.Object, java.lang.Object) + */ + public V put(final K key, final V value) + { + processQueue(); + AggregatedSoftReference refPut = mapDelegate.put(key, createReference(key, value)); + if (refPut != null) + { + return refPut.get(); + } + else + { + return null; + } + } + + /** + * @param key + * @return + * @see java.util.Map#remove(java.lang.Object) + */ + public V remove(final Object key) + { + processQueue(); + AggregatedSoftReference ref = mapDelegate.remove(key); + if (ref != null) + { + return ref.get(); + } + else + { + return null; + } + } + + /** + * @param m + * @see java.util.Map#putAll(java.util.Map) + */ + public void putAll(final Map<? extends K, ? extends V> m) + { + processQueue(); + for (Map.Entry<? extends K, ? extends V> e : m.entrySet()) + { + put(e.getKey(), e.getValue()); + } + } + + /** + * + * @see java.util.Map#clear() + */ + public void clear() + { + mapDelegate.clear(); + } + + /** + * @return + * @see java.util.Map#keySet() + */ + public Set<K> keySet() + { + processQueue(); + return mapDelegate.keySet(); + } + + /** + * @return + * @see java.util.Map#values() + */ + public Collection<V> values() + { + processQueue(); + ArrayList<V> list = new ArrayList<V>(); + + for (AggregatedSoftReference refs : mapDelegate.values()) + { + V value = refs.get(); + if (value != null) + { + list.add(value); + } + } + + return list; + } + + /** + * @return + * @see java.util.Map#entrySet() + */ + public Set<java.util.Map.Entry<K, V>> entrySet() + { + processQueue(); + HashSet<Map.Entry<K, V>> set = new HashSet<Map.Entry<K, V>>(); + for (Map.Entry<K, AggregatedSoftReference> pair : mapDelegate.entrySet()) + { + V value = pair.getValue().get(); + if (value != null) + { + set.add(new EntryElement<K,V>(pair.getKey(), value)); + } + } + return set; + } + + /** + * @param o + * @return + * @see java.util.Map#equals(java.lang.Object) + */ + @Override + public boolean equals(final Object o) + { + processQueue(); + return mapDelegate.equals(o); + } + + /** + * @return + * @see java.util.Map#hashCode() + */ + @Override + public int hashCode() + { + return mapDelegate.hashCode(); + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + @SuppressWarnings("unchecked") + private void processQueue() + { + AggregatedSoftReference ref = null; + while ((ref = (AggregatedSoftReference)this.refQueue.poll()) != null) + { + mapDelegate.remove(ref.key); + } + } + + private AggregatedSoftReference createReference(final K key, final V value) + { + AggregatedSoftReference ref = new AggregatedSoftReference(key, value); + return ref; + } + + // Inner classes ------------------------------------------------- + + class AggregatedSoftReference extends SoftReference<V> + { + final K key; + + public AggregatedSoftReference(final K key, final V referent) + { + super(referent, refQueue); + this.key = key; + } + } + + static final class EntryElement<K, V> implements Map.Entry<K, V> + { + final K key; + + volatile V value; + + EntryElement(final K key, final V value) + { + this.key = key; + this.value = value; + } + + /* (non-Javadoc) + * @see java.util.Map.Entry#getKey() + */ + public K getKey() + { + return key; + } + + /* (non-Javadoc) + * @see java.util.Map.Entry#getValue() + */ + public V getValue() + { + return value; + } + + /* (non-Javadoc) + * @see java.util.Map.Entry#setValue(java.lang.Object) + */ + public V setValue(final V value) + { + this.value = value; + return value; + } + } + +} Modified: trunk/tests/jms-tests/src/org/hornetq/jms/tests/BrowserTest.java =================================================================== --- trunk/tests/jms-tests/src/org/hornetq/jms/tests/BrowserTest.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/tests/jms-tests/src/org/hornetq/jms/tests/BrowserTest.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -26,6 +26,12 @@ import javax.jms.Session; import javax.jms.TextMessage; +import org.hornetq.api.core.HornetQException; +import org.hornetq.api.core.client.ClientConsumer; +import org.hornetq.api.core.client.ClientMessage; +import org.hornetq.api.core.client.ClientSession; +import org.hornetq.api.core.client.MessageHandler; +import org.hornetq.jms.client.HornetQConnectionFactory; import org.hornetq.jms.tests.util.ProxyAssertSupport; /** @@ -109,8 +115,8 @@ } } } - - public void testBrowse() throws Exception + + public void testBrowse2() throws Exception { Connection conn = null; @@ -121,59 +127,75 @@ Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); MessageProducer producer = session.createProducer(HornetQServerTestCase.queue1); + + HornetQConnectionFactory cf = (HornetQConnectionFactory) getConnectionFactory(); - final int numMessages = 10; + ClientSession coreSession = cf.getCoreFactory().createSession(true, true); - for (int i = 0; i < numMessages; i++) + coreSession.start(); + + ClientConsumer browser = coreSession.createConsumer("jms.queue.Queue1", true); + + conn.start(); + + Message m = session.createMessage(); + m.setIntProperty("cnt", 0); + producer.send(m); + + + assertNotNull(browser.receive(5000)); + + Thread.sleep(5000); + + coreSession.close(); + + + System.out.println("Draining destination..."); + drainDestination(getConnectionFactory(), queue1); + + } + finally + { + if (conn != null) { - Message m = session.createMessage(); - m.setIntProperty("cnt", i); - producer.send(m); + conn.close(); } + } + } + public void testBrowse() throws Exception + { + Connection conn = null; + + try + { + conn = getConnectionFactory().createConnection(); + + Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); + + MessageProducer producer = session.createProducer(HornetQServerTestCase.queue1); + QueueBrowser browser = session.createBrowser(HornetQServerTestCase.queue1); ProxyAssertSupport.assertEquals(browser.getQueue(), HornetQServerTestCase.queue1); ProxyAssertSupport.assertNull(browser.getMessageSelector()); - Enumeration en = browser.getEnumeration(); + Enumeration<Message> en = (Enumeration<Message>)browser.getEnumeration(); - int count = 0; - while (en.hasMoreElements()) - { - en.nextElement(); - count++; - } - - ProxyAssertSupport.assertEquals(numMessages, count); - - MessageConsumer mc = session.createConsumer(HornetQServerTestCase.queue1); - conn.start(); - for (int i = 0; i < numMessages; i++) - { - Message m = mc.receive(); - ProxyAssertSupport.assertNotNull(m); - } - - browser = session.createBrowser(HornetQServerTestCase.queue1); - en = browser.getEnumeration(); - - log.info("browsing"); - - count = 0; - while (en.hasMoreElements()) - { - Message mess = (Message)en.nextElement(); - log.info("message:" + mess); - count++; - } - - log.trace("Received " + count + " messages"); - - ProxyAssertSupport.assertEquals(0, count); + Message m = session.createMessage(); + m.setIntProperty("cnt", 0); + producer.send(m); + Message m2 = en.nextElement(); + + assertNotNull(m2); + + + System.out.println("Draining destination..."); + drainDestination(getConnectionFactory(), queue1); + } finally { @@ -204,19 +226,6 @@ m.setIntProperty("test_counter", i + 1); producer.send(m); } - - QueueBrowser browser = session.createBrowser(HornetQServerTestCase.queue1, "test_counter > 30"); - - Enumeration en = browser.getEnumeration(); - int count = 0; - while (en.hasMoreElements()) - { - Message m = (Message)en.nextElement(); - int testCounter = m.getIntProperty("test_counter"); - ProxyAssertSupport.assertTrue(testCounter > 30); - count++; - } - ProxyAssertSupport.assertEquals(70, count); } finally { Modified: trunk/tests/src/org/hornetq/tests/concurrent/server/impl/QueueTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/concurrent/server/impl/QueueTest.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/tests/src/org/hornetq/tests/concurrent/server/impl/QueueTest.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -65,6 +65,7 @@ new SimpleString("address1"), new SimpleString("queue1"), null, + null, false, false); Modified: trunk/tests/src/org/hornetq/tests/integration/client/PagingTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/PagingTest.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/tests/src/org/hornetq/tests/integration/client/PagingTest.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -17,12 +17,10 @@ import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.HashMap; -import java.util.List; import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; import java.util.concurrent.Executors; -import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -42,11 +40,9 @@ import org.hornetq.core.config.DivertConfiguration; import org.hornetq.core.journal.SequentialFileFactory; import org.hornetq.core.logging.Logger; -import org.hornetq.core.paging.Page; import org.hornetq.core.paging.PagingManager; import org.hornetq.core.paging.PagingStore; import org.hornetq.core.paging.PagingStoreFactory; -import org.hornetq.core.paging.impl.PageImpl; import org.hornetq.core.paging.impl.PagingManagerImpl; import org.hornetq.core.paging.impl.PagingStoreFactoryNIO; import org.hornetq.core.paging.impl.PagingStoreImpl; @@ -55,7 +51,6 @@ import org.hornetq.core.postoffice.PostOffice; import org.hornetq.core.server.HornetQServer; import org.hornetq.core.server.Queue; -import org.hornetq.core.server.ServerMessage; import org.hornetq.core.server.impl.HornetQServerImpl; import org.hornetq.core.settings.impl.AddressFullMessagePolicy; import org.hornetq.core.settings.impl.AddressSettings; @@ -118,6 +113,16 @@ public void testWithDiverts() throws Exception { + internalMultiQueuesTest(true); + } + + public void testWithMultiQueues() throws Exception + { + internalMultiQueuesTest(false); + } + + public void internalMultiQueuesTest(final boolean divert) throws Exception + { clearData(); Configuration config = createDefaultConfig(); @@ -128,28 +133,31 @@ PagingTest.PAGE_MAX, new HashMap<String, AddressSettings>()); - DivertConfiguration divert1 = new DivertConfiguration("dv1", - "nm1", - PagingTest.ADDRESS.toString(), - PagingTest.ADDRESS.toString() + "-1", - true, - null, - null); + if (divert) + { + DivertConfiguration divert1 = new DivertConfiguration("dv1", + "nm1", + PagingTest.ADDRESS.toString(), + PagingTest.ADDRESS.toString() + "-1", + true, + null, + null); + + DivertConfiguration divert2 = new DivertConfiguration("dv2", + "nm2", + PagingTest.ADDRESS.toString(), + PagingTest.ADDRESS.toString() + "-2", + true, + null, + null); + + ArrayList<DivertConfiguration> divertList = new ArrayList<DivertConfiguration>(); + divertList.add(divert1); + divertList.add(divert2); + + config.setDivertConfigurations(divertList); + } - DivertConfiguration divert2 = new DivertConfiguration("dv2", - "nm2", - PagingTest.ADDRESS.toString(), - PagingTest.ADDRESS.toString() + "-2", - true, - null, - null); - - ArrayList<DivertConfiguration> divertList = new ArrayList<DivertConfiguration>(); - divertList.add(divert1); - divertList.add(divert2); - - config.setDivertConfigurations(divertList); - server.start(); final int messageSize = 1024; @@ -176,10 +184,19 @@ ClientSession session = sf.createSession(false, false, false); - session.createQueue(PagingTest.ADDRESS + "-1", PagingTest.ADDRESS + "-1", null, true); + if (divert) + { + session.createQueue(PagingTest.ADDRESS + "-1", PagingTest.ADDRESS + "-1", null, true); - session.createQueue(PagingTest.ADDRESS + "-2", PagingTest.ADDRESS + "-2", null, true); + session.createQueue(PagingTest.ADDRESS + "-2", PagingTest.ADDRESS + "-2", null, true); + } + else + { + session.createQueue(PagingTest.ADDRESS.toString(), PagingTest.ADDRESS + "-1", null, true); + session.createQueue(PagingTest.ADDRESS.toString(), PagingTest.ADDRESS + "-2", null, true); + } + ClientProducer producer = session.createProducer(PagingTest.ADDRESS); ClientMessage message = null; @@ -319,6 +336,8 @@ private void internaltestSendReceivePaging(final boolean persistentMessages) throws Exception { + + System.out.println("PageDir:" + getPageDir()); clearData(); Configuration config = createDefaultConfig(); @@ -448,7 +467,7 @@ UnitTestCase.assertEqualsByteArrays(body, other); } - + /** * - Make a destination in page mode * - Add stuff to a transaction @@ -903,7 +922,7 @@ msg.putIntProperty("count", i); producer.send(msg); - if (i % 50 == 0 && i != 0) + if (i % 100 == 0 && i != 0) { sessionProducer.commit(); // Thread.sleep(500); @@ -947,17 +966,17 @@ ClientConsumer consumer = session.createConsumer(PagingTest.ADDRESS); for (int i = 0; i < numberOfMessages; i++) - { - ClientMessage msg = consumer.receive(500000); + { + ClientMessage msg = consumer.receive(5000); assertNotNull(msg); assertEquals(i, msg.getIntProperty("count").intValue()); msg.acknowledge(); if (i > 0 && i % 10 == 0) { - // session.commit(); + session.commit(); } } - // session.commit(); + session.commit(); session.close(); @@ -979,7 +998,7 @@ } // This test will force a depage thread as soon as the first message hits the page - public void testDepageOnTX5() throws Exception + public void testDepageDuringTransaction5() throws Exception { clearData(); @@ -999,7 +1018,7 @@ final PagingStoreFactory storeFactory, final SimpleString storeName, final AddressSettings addressSettings, - final Executor executor, + final ExecutorFactory executorFactory, final boolean syncNonTransactional) { super(address, @@ -1010,41 +1029,10 @@ storeFactory, storeName, addressSettings, - executor, + executorFactory, syncNonTransactional); } - protected boolean page(final List<ServerMessage> messages, final long transactionID, final boolean sync) throws Exception - { - boolean paged = super.page(messages, transactionID, sync); - - if (paged) - { - - if (countDepage.incrementAndGet() == 1) - { - countDepage.set(0); - - executor.execute(new Runnable() - { - public void run() - { - try - { - while (isStarted() && readPage()); - } - catch (Exception e) - { - e.printStackTrace(); - } - } - }); - } - } - - return paged; - } - public boolean startDepaging() { // do nothing, we are hacking depage right in between paging @@ -1062,7 +1050,7 @@ super(directory, executorFactory, syncNonTransactional); } - public synchronized PagingStore newStore(final SimpleString address, final AddressSettings settings) throws Exception + public synchronized PagingStore newStore(final SimpleString address, final AddressSettings settings) { return new HackPagingStore(address, @@ -1073,7 +1061,7 @@ this, address, settings, - getExecutorFactory().getExecutor(), + getExecutorFactory(), syncNonTransactional); } @@ -1308,8 +1296,9 @@ for (int i = 0; i < numberOfMessages; i++) { - ClientMessage msg = consumer.receive(500000); + ClientMessage msg = consumer.receive(5000); assertNotNull(msg); + System.out.println("Received " + i); assertEquals(i, msg.getIntProperty("count").intValue()); msg.acknowledge(); } Modified: trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControl2Test.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControl2Test.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControl2Test.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -47,6 +47,7 @@ import org.hornetq.core.remoting.impl.netty.NettyConnectorFactory; import org.hornetq.core.server.HornetQServer; import org.hornetq.core.server.HornetQServers; +import org.hornetq.jms.client.HornetQMessage; import org.hornetq.jms.client.HornetQQueue; import org.hornetq.jms.server.impl.JMSServerManagerImpl; import org.hornetq.tests.integration.management.ManagementControlHelper; @@ -359,6 +360,8 @@ System.out.println("queueName is: " + queueName); + Connection connection = null; + try { startHornetQServer(NettyAcceptorFactory.class.getName()); @@ -372,16 +375,20 @@ ConnectionFactory cf1 = JMSUtil.createFactory(NettyConnectorFactory.class.getName(), JMSServerControl2Test.CONNECTION_TTL, JMSServerControl2Test.PING_PERIOD); - Connection connection = cf1.createConnection(); + connection = cf1.createConnection(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); MessageProducer producer = session.createProducer(queue); + TextMessage msgSent = null; for (int i = 0; i < 10; i++) { - TextMessage msg = session.createTextMessage("mymessage-" + i); - producer.send(msg); + msgSent = session.createTextMessage("mymessage-" + i); + producer.send(msgSent); + System.out.println("sending msgID " + msgSent.getJMSMessageID()); } + + connection.start(); @@ -392,12 +399,15 @@ for (int i = 0; i < 10; i++) { receivedMsg = (TextMessage)consumer.receive(3000); + assertNotNull(receivedMsg); System.out.println("receiveMsg: " + receivedMsg); } - String lastMsgID = receivedMsg.getJMSMessageID(); - System.out.println("Last mid: " + lastMsgID); + assertEquals(msgSent.getJMSMessageID(), receivedMsg.getJMSMessageID()); + HornetQMessage jmsMessage = (HornetQMessage)receivedMsg; + String lastMsgID = jmsMessage.getCoreMessage().getUserID().toString(); + String jsonStr = control.listConnectionsAsJSON(); JMSConnectionInfo[] infos = JMSConnectionInfo.from(jsonStr); @@ -434,11 +444,23 @@ } finally { - if (serverManager != null) + try { - serverManager.destroyQueue(queueName); - serverManager.stop(); + if (connection != null) + { + connection.close(); + } + + if (serverManager != null) + { + serverManager.destroyQueue(queueName); + serverManager.stop(); + } } + catch (Throwable ignored) + { + ignored.printStackTrace(); + } if (server != null) { Deleted: trunk/tests/src/org/hornetq/tests/integration/paging/PageCrashTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/paging/PageCrashTest.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/tests/src/org/hornetq/tests/integration/paging/PageCrashTest.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -1,442 +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.tests.integration.paging; - -import java.io.File; -import java.lang.management.ManagementFactory; -import java.lang.reflect.Field; -import java.util.HashMap; -import java.util.List; -import java.util.concurrent.Executor; -import java.util.concurrent.Executors; - -import junit.framework.Assert; - -import org.hornetq.api.core.SimpleString; -import org.hornetq.api.core.client.ClientConsumer; -import org.hornetq.api.core.client.ClientMessage; -import org.hornetq.api.core.client.ClientProducer; -import org.hornetq.api.core.client.ClientSession; -import org.hornetq.api.core.client.ClientSessionFactory; -import org.hornetq.core.config.Configuration; -import org.hornetq.core.paging.Page; -import org.hornetq.core.paging.PagedMessage; -import org.hornetq.core.paging.PagingManager; -import org.hornetq.core.paging.PagingStore; -import org.hornetq.core.paging.impl.PagingManagerImpl; -import org.hornetq.core.paging.impl.PagingStoreFactoryNIO; -import org.hornetq.core.paging.impl.PagingStoreImpl; -import org.hornetq.core.server.HornetQServer; -import org.hornetq.core.server.impl.HornetQServerImpl; -import org.hornetq.core.settings.impl.AddressSettings; -import org.hornetq.spi.core.security.HornetQSecurityManager; -import org.hornetq.spi.core.security.HornetQSecurityManagerImpl; -import org.hornetq.tests.util.ServiceTestBase; -import org.hornetq.utils.OrderedExecutorFactory; - -/** - * This test will make sure that a failing depage won't cause duplicated messages - * - * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> - * - * Created Jan 7, 2009 6:19:43 PM - * - * - */ -public class PageCrashTest extends ServiceTestBase -{ - - // Constants ----------------------------------------------------- - - public static final SimpleString ADDRESS = new SimpleString("SimpleAddress"); - - // Attributes ---------------------------------------------------- - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - // Public -------------------------------------------------------- - - public void testCrashDuringDeleteFile() throws Exception - { - doTestCrashDuringDeleteFile(false); - } - - public void testCrashDuringDeleteFileTransacted() throws Exception - { - doTestCrashDuringDeleteFile(true); - } - - public void doTestCrashDuringDeleteFile(final boolean transacted) throws Exception - { - pageAndFail(transacted); - - File pageDir = new File(getPageDir()); - - File directories[] = pageDir.listFiles(); - - Assert.assertEquals(1, directories.length); - - if (!transacted) - { - // When depage happened, a new empty page was supposed to be opened, what will create 3 files - Assert.assertEquals("Missing a file, supposed to have address.txt, 1st page and 2nd page", - 3, - directories[0].list().length); - } - - Configuration config = createDefaultConfig(); - - HornetQServer messagingService = createServer(true, - config, - 10 * 1024, - 100 * 1024, - new HashMap<String, AddressSettings>()); - - messagingService.start(); - - try - { - ClientSessionFactory sf = createInVMFactory(); - - ClientSession session = sf.createSession(null, null, false, true, true, false, 0); - - session.start(); - - ClientConsumer consumer = session.createConsumer(PageCrashTest.ADDRESS); - - Assert.assertNull(consumer.receiveImmediate()); - - session.close(); - } - finally - { - messagingService.stop(); - } - - } - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - // Private ------------------------------------------------------- - - /** This method will leave garbage on paging. - * It will not delete page files as if the server crashed right after commit, - * and before removing the file*/ - private void pageAndFail(final boolean transacted) throws Exception - { - clearData(); - Configuration config = createDefaultConfig(); - - HornetQServer server = newHornetQServer(config); - - server.start(); - - try - { - ClientSessionFactory sf = createInVMFactory(); - - // Making it synchronous, just because we want to stop sending messages as soon as the page-store becomes in - // page mode - // and we could only guarantee that by setting it to synchronous - sf.setBlockOnNonDurableSend(true); - sf.setBlockOnDurableSend(true); - sf.setBlockOnAcknowledge(true); - - ClientSession session = sf.createSession(null, null, false, !transacted, !transacted, false, 0); - - session.createQueue(PageCrashTest.ADDRESS, PageCrashTest.ADDRESS, null, true); - - ClientProducer producer = session.createProducer(PageCrashTest.ADDRESS); - - ClientMessage message = session.createMessage(true); - message.getBodyBuffer().writeBytes(new byte[1024]); - - PagingStore store = server.getPostOffice().getPagingManager().getPageStore(PageCrashTest.ADDRESS); - - int messages = 0; - while (!store.isPaging()) - { - producer.send(message); - messages++; - if (transacted && messages % 100 == 0) - { - session.commit(); - } - } - - for (int i = 0; i < 2; i++) - { - messages++; - producer.send(message); - } - - session.commit(); - - session.close(); - - session = sf.createSession(null, null, false, true, true, false, 0); - - ClientConsumer consumer = session.createConsumer(PageCrashTest.ADDRESS); - - session.start(); - - for (int i = 0; i < messages; i++) - { - ClientMessage message2 = consumer.receive(10000); - - Assert.assertNotNull(message2); - - message2.acknowledge(); - } - - consumer.close(); - - session.close(); - } - finally - { - try - { - server.stop(); - } - catch (Throwable ignored) - { - } - } - } - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - // Private ------------------------------------------------------- - - private HornetQServer newHornetQServer(final Configuration configuration) - { - HornetQSecurityManager securityManager = new HornetQSecurityManagerImpl(); - - HornetQServer server = new FailingHornetQServer(configuration, securityManager); - - AddressSettings defaultSetting = new AddressSettings(); - defaultSetting.setPageSizeBytes(10 * 1024); - defaultSetting.setMaxSizeBytes(100 * 1024); - - server.getAddressSettingsRepository().addMatch("#", defaultSetting); - - return server; - } - - // Inner classes ------------------------------------------------- - - /** This is hacking HornetQServerImpl, - * to make sure the server will fail right - * before the page-file was removed */ - class FailingHornetQServer extends HornetQServerImpl - { - FailingHornetQServer(final Configuration config, final HornetQSecurityManager securityManager) - { - super(config, ManagementFactory.getPlatformMBeanServer(), securityManager); - } - - @Override - protected PagingManager createPagingManager() - { - return new PagingManagerImpl(new FailurePagingStoreFactoryNIO(super.getConfiguration().getPagingDirectory(), - super.getConfiguration() - .isJournalSyncNonTransactional()), - super.getStorageManager(), - super.getAddressSettingsRepository()); - } - - class FailurePagingStoreFactoryNIO extends PagingStoreFactoryNIO - - { - /** - * @param directory - * @param maxThreads - */ - public FailurePagingStoreFactoryNIO(final String directory, final boolean syncNonTransactional) - { - super(directory, new OrderedExecutorFactory(Executors.newCachedThreadPool()), syncNonTransactional); - } - - // Constants ----------------------------------------------------- - - // Attributes ---------------------------------------------------- - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - // Public -------------------------------------------------------- - - @Override - public synchronized PagingStore newStore(final SimpleString destinationName, final AddressSettings settings) throws Exception - { - Field factoryField = PagingStoreFactoryNIO.class.getDeclaredField("executorFactory"); - factoryField.setAccessible(true); - - OrderedExecutorFactory factory = (org.hornetq.utils.OrderedExecutorFactory)factoryField.get(this); - return new FailingPagingStore(destinationName, settings, factory.getExecutor(), syncNonTransactional); - } - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - // Private ------------------------------------------------------- - - // Inner classes ------------------------------------------------- - class FailingPagingStore extends PagingStoreImpl - { - - /** - * @param storeName - * @param addressSettings - * @param executor - */ - public FailingPagingStore(final SimpleString storeName, - final AddressSettings addressSettings, - final Executor executor, - final boolean syncNonTransactional) - { - super(storeName, - getPostOffice().getPagingManager(), - getStorageManager(), - getPostOffice(), - null, - FailurePagingStoreFactoryNIO.this, - storeName, - addressSettings, - executor, - syncNonTransactional); - } - - @Override - public Page createPage(final int page) throws Exception - { - - Page originalPage = super.createPage(page); - - return new FailingPage(originalPage); - } - - } - - } - - class FailingPage implements Page - { - Page delegatedPage; - - /** - * @throws Exception - * @see org.hornetq.core.paging.Page#close() - */ - public void close() throws Exception - { - delegatedPage.close(); - } - - /** - * @throws Exception - * @see org.hornetq.core.paging.Page#delete() - */ - public boolean delete() throws Exception - { - - System.out.println("Won't delete"); - return false; - } - - /** - * @return - * @see org.hornetq.core.paging.Page#getNumberOfMessages() - */ - public int getNumberOfMessages() - { - return delegatedPage.getNumberOfMessages(); - } - - /** - * @return - * @see org.hornetq.core.paging.Page#getPageId() - */ - public int getPageId() - { - return delegatedPage.getPageId(); - } - - /** - * @return - * @see org.hornetq.core.paging.Page#getSize() - */ - public int getSize() - { - return delegatedPage.getSize(); - } - - /** - * @throws Exception - * @see org.hornetq.core.paging.Page#open() - */ - public void open() throws Exception - { - delegatedPage.open(); - } - - /** - * @return - * @throws Exception - * @see org.hornetq.core.paging.Page#read() - */ - public List<PagedMessage> read() throws Exception - { - return delegatedPage.read(); - } - - /** - * @throws Exception - * @see org.hornetq.core.paging.Page#sync() - */ - public void sync() throws Exception - { - delegatedPage.sync(); - } - - /** - * @param message - * @throws Exception - * @see org.hornetq.core.paging.Page#write(org.hornetq.core.paging.PagedMessage) - */ - public void write(final PagedMessage message) throws Exception - { - delegatedPage.write(message); - } - - public FailingPage(final Page delegatePage) - { - delegatedPage = delegatePage; - } - } - - } - - // Inner classes ------------------------------------------------- - -} Added: trunk/tests/src/org/hornetq/tests/integration/paging/PagePositionTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/paging/PagePositionTest.java (rev 0) +++ trunk/tests/src/org/hornetq/tests/integration/paging/PagePositionTest.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -0,0 +1,67 @@ +/* + * Copyright 2010 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.integration.paging; + +import org.hornetq.tests.util.ServiceTestBase; + +/** + * A PageCursorTest + * + * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * + */ +public class PagePositionTest extends ServiceTestBase +{ + + // Test what would happen on redelivery situations + public void testRedeliverLike() + { + + } + + public void testRedeliverPersistence() + { + + } + + public void testDeletePagesAfterRedelivery() + { + + } + + public void testNextAfterPosition() + { + + } + + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- + +} Modified: trunk/tests/src/org/hornetq/tests/integration/persistence/DeleteMessagesOnStartupTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/persistence/DeleteMessagesOnStartupTest.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/tests/src/org/hornetq/tests/integration/persistence/DeleteMessagesOnStartupTest.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -76,7 +76,7 @@ journal.loadBindingJournal(new ArrayList<QueueBindingInfo>(), new ArrayList<GroupingInfo>()); - journal.loadMessageJournal(new FakePostOffice(), null, null, queues, null); + journal.loadMessageJournal(new FakePostOffice(), null, null, queues, null, null); assertEquals(98, deletedMessage.size()); Modified: trunk/tests/src/org/hornetq/tests/integration/persistence/RestartSMTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/persistence/RestartSMTest.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/tests/src/org/hornetq/tests/integration/persistence/RestartSMTest.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -101,7 +101,7 @@ Map<Long, Queue> queues = new HashMap<Long, Queue>(); - journal.loadMessageJournal(postOffice, null, null, queues, null); + journal.loadMessageJournal(postOffice, null, null, queues, null, null); journal.stop(); @@ -111,7 +111,7 @@ queues = new HashMap<Long, Queue>(); - journal.loadMessageJournal(postOffice, null, null, queues, null); + journal.loadMessageJournal(postOffice, null, null, queues, null, null); queueBindingInfos = new ArrayList<QueueBindingInfo>(); Modified: trunk/tests/src/org/hornetq/tests/integration/persistence/StorageManagerTestBase.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/persistence/StorageManagerTestBase.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/tests/src/org/hornetq/tests/integration/persistence/StorageManagerTestBase.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -128,7 +128,7 @@ Map<Long, Queue> queues = new HashMap<Long, Queue>(); - journal.loadMessageJournal(new FakePostOffice(), null, null, queues, null); + journal.loadMessageJournal(new FakePostOffice(), null, null, queues, null, null); } /** Modified: trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -288,7 +288,7 @@ replicatedJournal.appendAddRecordTransactional(23, 24, (byte)1, new FakeData()); - PagedMessage pgmsg = new PagedMessageImpl(msg, -1); + PagedMessage pgmsg = new PagedMessageImpl(msg, new long[0]); manager.pageWrite(pgmsg, 1); manager.pageWrite(pgmsg, 2); manager.pageWrite(pgmsg, 3); @@ -303,7 +303,7 @@ PagingStore store = pagingManager.getPageStore(dummy); store.start(); - Assert.assertEquals(5, store.getNumberOfPages()); + Assert.assertEquals(4, store.getNumberOfPages()); store.stop(); manager.pageDeleted(dummy, 1); Modified: trunk/tests/src/org/hornetq/tests/integration/server/LVQTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/server/LVQTest.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/tests/src/org/hornetq/tests/integration/server/LVQTest.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -289,6 +289,26 @@ Assert.assertEquals(m.getBodyBuffer().readString(), "m4"); } + public void testSingleTXRollback() throws Exception + { + ClientProducer producer = clientSessionTxReceives.createProducer(address); + ClientConsumer consumer = clientSessionTxReceives.createConsumer(qName1); + SimpleString messageId1 = new SimpleString("SMID1"); + ClientMessage m1 = createTextMessage("m1", clientSession); + m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, messageId1); + producer.send(m1); + clientSessionTxReceives.start(); + ClientMessage m = consumer.receive(1000); + Assert.assertNotNull(m); + m.acknowledge(); + clientSessionTxReceives.rollback(); + m = consumer.receive(1000); + Assert.assertNotNull(m); + m.acknowledge(); + Assert.assertEquals(m.getBodyBuffer().readString(), "m1"); + Assert.assertNull(consumer.receiveImmediate()); + } + public void testMultipleMessagesInTxSend() throws Exception { ClientProducer producer = clientSessionTxSends.createProducer(address); Added: trunk/tests/src/org/hornetq/tests/stress/paging/PageCursorStressTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/stress/paging/PageCursorStressTest.java (rev 0) +++ trunk/tests/src/org/hornetq/tests/stress/paging/PageCursorStressTest.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -0,0 +1,1330 @@ +/* + * Copyright 2010 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.stress.paging; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; + +import junit.framework.Assert; + +import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.api.core.SimpleString; +import org.hornetq.api.core.client.ClientSession; +import org.hornetq.api.core.client.ClientSessionFactory; +import org.hornetq.core.config.Configuration; +import org.hornetq.core.filter.Filter; +import org.hornetq.core.paging.PagedMessage; +import org.hornetq.core.paging.cursor.PageCache; +import org.hornetq.core.paging.cursor.PageCursorProvider; +import org.hornetq.core.paging.cursor.PagePosition; +import org.hornetq.core.paging.cursor.PageSubscription; +import org.hornetq.core.paging.cursor.PagedReference; +import org.hornetq.core.paging.cursor.impl.PageCursorProviderImpl; +import org.hornetq.core.paging.cursor.impl.PagePositionImpl; +import org.hornetq.core.paging.cursor.impl.PageSubscriptionImpl; +import org.hornetq.core.paging.impl.PagingStoreImpl; +import org.hornetq.core.persistence.StorageManager; +import org.hornetq.core.persistence.impl.journal.OperationContextImpl; +import org.hornetq.core.server.HornetQServer; +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.server.impl.ServerMessageImpl; +import org.hornetq.core.settings.impl.AddressSettings; +import org.hornetq.core.transaction.Transaction; +import org.hornetq.core.transaction.impl.TransactionImpl; +import org.hornetq.tests.unit.core.postoffice.impl.FakeQueue; +import org.hornetq.tests.util.RandomUtil; +import org.hornetq.tests.util.ServiceTestBase; +import org.hornetq.utils.LinkedListIterator; + +/** + * A PageCursorTest + * + * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * + */ +public class PageCursorStressTest extends ServiceTestBase +{ + + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + private SimpleString ADDRESS = new SimpleString("test-add"); + + private HornetQServer server; + + private Queue queue; + + private List<Queue> queueList; + + private static final int PAGE_MAX = -1; + + private static final int PAGE_SIZE = 10 * 1024 * 1024; + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + + // Read more cache than what would fit on the memory, and validate if the memory would be cleared through soft-caches + public void testReadCache() throws Exception + { + + final int NUM_MESSAGES = 1000; + + int numberOfPages = addMessages(NUM_MESSAGES, 1024 * 1024); + + System.out.println("NumberOfPages = " + numberOfPages); + + PageCursorProviderImpl cursorProvider = new PageCursorProviderImpl(lookupPageStore(ADDRESS), + server.getStorageManager(), + server.getExecutorFactory()); + + for (int i = 0; i < numberOfPages; i++) + { + PageCache cache = cursorProvider.getPageCache(new PagePositionImpl(i + 1, 0)); + System.out.println("Page " + i + " had " + cache.getNumberOfMessages() + " messages"); + + } + + forceGC(); + + assertTrue(cursorProvider.getCacheSize() < numberOfPages); + + System.out.println("Cache size = " + cursorProvider.getCacheSize()); + } + + public void testSimpleCursor() throws Exception + { + + final int NUM_MESSAGES = 100; + + PageSubscription cursor = lookupPageStore(ADDRESS).getCursorProvier().getSubscription(queue.getID()); + + Iterator<PagedReference> iterEmpty = cursor.iterator(); + + int numberOfPages = addMessages(NUM_MESSAGES, 1024 * 1024); + + System.out.println("NumberOfPages = " + numberOfPages); + + PagedReference msg; + + LinkedListIterator<PagedReference> iterator = cursor.iterator(); + int key = 0; + while ((msg = iterator.next()) != null) + { + assertEquals(key++, msg.getMessage().getIntProperty("key").intValue()); + cursor.ack(msg.getPosition()); + } + assertEquals(NUM_MESSAGES, key); + + server.getStorageManager().waitOnOperations(); + + waitCleanup(); + + assertFalse(lookupPageStore(ADDRESS).isPaging()); + + assertEquals(1, lookupPageStore(ADDRESS).getNumberOfPages()); + + forceGC(); + + server.stop(); + createServer(); + waitCleanup(); + assertEquals(1, lookupPageStore(ADDRESS).getNumberOfPages()); + + } + + public void testSimpleCursorWithFilter() throws Exception + { + + final int NUM_MESSAGES = 100; + + PageSubscription cursorEven = createNonPersistentCursor(new Filter() + { + + public boolean match(ServerMessage message) + { + Boolean property = message.getBooleanProperty("even"); + if (property == null) + { + return false; + } + else + { + return property.booleanValue(); + } + } + + public SimpleString getFilterString() + { + return new SimpleString("even=true"); + } + + }); + + PageSubscription cursorOdd = createNonPersistentCursor(new Filter() + { + + public boolean match(ServerMessage message) + { + Boolean property = message.getBooleanProperty("even"); + if (property == null) + { + return false; + } + else + { + return !property.booleanValue(); + } + } + + public SimpleString getFilterString() + { + return new SimpleString("even=true"); + } + + }); + + int numberOfPages = addMessages(NUM_MESSAGES, 1024 * 1024); + + System.out.println("NumberOfPages = " + numberOfPages); + + queue.getPageSubscription().close(); + + PagedReference msg; + + LinkedListIterator<PagedReference> iteratorEven = cursorEven.iterator(); + + LinkedListIterator<PagedReference> iteratorOdd = cursorOdd.iterator(); + + int key = 0; + while ((msg = iteratorEven.next()) != null) + { + System.out.println("Received" + msg); + assertEquals(key, msg.getMessage().getIntProperty("key").intValue()); + assertTrue(msg.getMessage().getBooleanProperty("even").booleanValue()); + key += 2; + cursorEven.ack(msg.getPosition()); + } + assertEquals(NUM_MESSAGES, key); + + key = 1; + while ((msg = iteratorOdd.next()) != null) + { + assertEquals(key, msg.getMessage().getIntProperty("key").intValue()); + assertFalse(msg.getMessage().getBooleanProperty("even").booleanValue()); + key += 2; + cursorOdd.ack(msg.getPosition()); + } + assertEquals(NUM_MESSAGES + 1, key); + + forceGC(); + + // assertTrue(lookupCursorProvider().getCacheSize() < numberOfPages); + + server.stop(); + createServer(); + waitCleanup(); + assertEquals(1, lookupPageStore(ADDRESS).getNumberOfPages()); + + } + + public void testReadNextPage() throws Exception + { + + final int NUM_MESSAGES = 1; + + int numberOfPages = addMessages(NUM_MESSAGES, 1024); + + System.out.println("NumberOfPages = " + numberOfPages); + + PageCursorProvider cursorProvider = lookupCursorProvider(); + + PageCache cache = cursorProvider.getPageCache(new PagePositionImpl(2, 0)); + + assertNull(cache); + } + + public void testRestart() throws Exception + { + final int NUM_MESSAGES = 1000; + + int numberOfPages = addMessages(NUM_MESSAGES, 100 * 1024); + + System.out.println("Number of pages = " + numberOfPages); + + PageCursorProvider cursorProvider = lookupCursorProvider(); + + PageSubscription cursor = this.server.getPagingManager() + .getPageStore(ADDRESS) + .getCursorProvier() + .getSubscription(queue.getID()); + + PageCache firstPage = cursorProvider.getPageCache(new PagePositionImpl(server.getPagingManager() + .getPageStore(ADDRESS) + .getFirstPage(), 0)); + + int firstPageSize = firstPage.getNumberOfMessages(); + + firstPage = null; + + System.out.println("Cursor: " + cursor); + cursorProvider.printDebug(); + + LinkedListIterator<PagedReference> iterator = cursor.iterator(); + + for (int i = 0; i < 1000; i++) + { + System.out.println("Reading Msg : " + i); + PagedReference msg = iterator.next(); + assertNotNull(msg); + assertEquals(i, msg.getMessage().getIntProperty("key").intValue()); + + if (i < firstPageSize) + { + cursor.ack(msg); + } + } + cursorProvider.printDebug(); + + server.getStorageManager().waitOnOperations(); + lookupPageStore(ADDRESS).flushExecutors(); + + // needs to clear the context since we are using the same thread over two distinct servers + // otherwise we will get the old executor on the factory + OperationContextImpl.clearContext(); + + server.stop(); + + server.start(); + + cursor = this.server.getPagingManager().getPageStore(ADDRESS).getCursorProvier().getSubscription(queue.getID()); + + iterator = cursor.iterator(); + + for (int i = firstPageSize; i < NUM_MESSAGES; i++) + { + System.out.println("Received " + i); + PagedReference msg = iterator.next(); + assertNotNull(msg); + assertEquals(i, msg.getMessage().getIntProperty("key").intValue()); + + cursor.ack(msg); + + OperationContextImpl.getContext(null).waitCompletion(); + + } + + OperationContextImpl.getContext(null).waitCompletion(); + + lookupPageStore(ADDRESS).flushExecutors(); + + assertFalse(lookupPageStore(ADDRESS).isPaging()); + + server.stop(); + createServer(); + assertFalse(lookupPageStore(ADDRESS).isPaging()); + waitCleanup(); + assertEquals(1, lookupPageStore(ADDRESS).getNumberOfPages()); + + } + + public void testRestartWithHoleOnAck() throws Exception + { + + final int NUM_MESSAGES = 1000; + + int numberOfPages = addMessages(NUM_MESSAGES, 10 * 1024); + + System.out.println("Number of pages = " + numberOfPages); + + PageCursorProvider cursorProvider = this.server.getPagingManager().getPageStore(ADDRESS).getCursorProvier(); + System.out.println("cursorProvider = " + cursorProvider); + + PageSubscription cursor = this.server.getPagingManager() + .getPageStore(ADDRESS) + .getCursorProvier() + .getSubscription(queue.getID()); + + System.out.println("Cursor: " + cursor); + LinkedListIterator<PagedReference> iterator = cursor.iterator(); + for (int i = 0; i < 100; i++) + { + PagedReference msg = iterator.next(); + assertEquals(i, msg.getMessage().getIntProperty("key").intValue()); + if (i < 10 || i > 20) + { + cursor.ack(msg); + } + } + + server.getStorageManager().waitOnOperations(); + + server.stop(); + + OperationContextImpl.clearContext(); + + server.start(); + + cursor = this.server.getPagingManager().getPageStore(ADDRESS).getCursorProvier().getSubscription(queue.getID()); + iterator = cursor.iterator(); + + for (int i = 10; i <= 20; i++) + { + PagedReference msg = iterator.next(); + assertEquals(i, msg.getMessage().getIntProperty("key").intValue()); + cursor.ack(msg); + } + + for (int i = 100; i < NUM_MESSAGES; i++) + { + PagedReference msg = iterator.next(); + assertEquals(i, msg.getMessage().getIntProperty("key").intValue()); + cursor.ack(msg); + } + + server.stop(); + createServer(); + waitCleanup(); + assertEquals(1, lookupPageStore(ADDRESS).getNumberOfPages()); + + } + + public void testRestartWithHoleOnAckAndTransaction() throws Exception + { + final int NUM_MESSAGES = 1000; + + int numberOfPages = addMessages(NUM_MESSAGES, 10 * 1024); + + System.out.println("Number of pages = " + numberOfPages); + + PageCursorProvider cursorProvider = this.server.getPagingManager().getPageStore(ADDRESS).getCursorProvier(); + System.out.println("cursorProvider = " + cursorProvider); + + PageSubscription cursor = this.server.getPagingManager() + .getPageStore(ADDRESS) + .getCursorProvier() + .getSubscription(queue.getID()); + + System.out.println("Cursor: " + cursor); + + Transaction tx = new TransactionImpl(server.getStorageManager(), 60 * 1000); + + LinkedListIterator<PagedReference> iterator = cursor.iterator(); + + for (int i = 0; i < 100; i++) + { + PagedReference msg = iterator.next(); + assertEquals(i, msg.getMessage().getIntProperty("key").intValue()); + if (i < 10 || i > 20) + { + cursor.ackTx(tx, msg); + } + } + + tx.commit(); + + server.stop(); + + OperationContextImpl.clearContext(); + + server.start(); + + cursor = this.server.getPagingManager().getPageStore(ADDRESS).getCursorProvier().getSubscription(queue.getID()); + + tx = new TransactionImpl(server.getStorageManager(), 60 * 1000); + iterator = cursor.iterator(); + + for (int i = 10; i <= 20; i++) + { + PagedReference msg = iterator.next(); + assertEquals(i, msg.getMessage().getIntProperty("key").intValue()); + cursor.ackTx(tx, msg); + } + + for (int i = 100; i < NUM_MESSAGES; i++) + { + PagedReference msg = iterator.next(); + assertEquals(i, msg.getMessage().getIntProperty("key").intValue()); + cursor.ackTx(tx, msg); + } + + tx.commit(); + + server.stop(); + createServer(); + waitCleanup(); + assertEquals(1, lookupPageStore(ADDRESS).getNumberOfPages()); + + } + + public void testConsumeLivePage() throws Exception + { + PagingStoreImpl pageStore = lookupPageStore(ADDRESS); + + pageStore.startPaging(); + + final int NUM_MESSAGES = 100; + + final int messageSize = 1024 * 1024; + + PageCursorProvider cursorProvider = this.server.getPagingManager().getPageStore(ADDRESS).getCursorProvier(); + System.out.println("cursorProvider = " + cursorProvider); + + PageSubscription cursor = this.server.getPagingManager() + .getPageStore(ADDRESS) + .getCursorProvier() + .getSubscription(queue.getID()); + + System.out.println("Cursor: " + cursor); + + RoutingContextImpl ctx = generateCTX(); + + LinkedListIterator<PagedReference> iterator = cursor.iterator(); + + for (int i = 0; i < NUM_MESSAGES; i++) + { + // if (i % 100 == 0) + System.out.println("read/written " + i); + + HornetQBuffer buffer = RandomUtil.randomBuffer(messageSize, i + 1l); + + ServerMessage msg = new ServerMessageImpl(i, buffer.writerIndex()); + msg.putIntProperty("key", i); + + msg.getBodyBuffer().writeBytes(buffer, 0, buffer.writerIndex()); + + Assert.assertTrue(pageStore.page(msg, ctx, ctx.getContextListing(ADDRESS))); + + PagedReference readMessage = iterator.next(); + + assertNotNull(readMessage); + + assertEquals(i, readMessage.getMessage().getIntProperty("key").intValue()); + + assertNull(iterator.next()); + } + + server.stop(); + + OperationContextImpl.clearContext(); + + createServer(); + + pageStore = lookupPageStore(ADDRESS); + + cursor = this.server.getPagingManager().getPageStore(ADDRESS).getCursorProvier().getSubscription(queue.getID()); + iterator = cursor.iterator(); + + for (int i = 0; i < NUM_MESSAGES * 2; i++) + { + if (i % 100 == 0) + System.out.println("Paged " + i); + + if (i >= NUM_MESSAGES) + { + + HornetQBuffer buffer = RandomUtil.randomBuffer(messageSize, i + 1l); + + ServerMessage msg = new ServerMessageImpl(i, buffer.writerIndex()); + msg.putIntProperty("key", i); + + msg.getBodyBuffer().writeBytes(buffer, 0, buffer.writerIndex()); + + Assert.assertTrue(pageStore.page(msg, ctx, ctx.getContextListing(ADDRESS))); + } + + PagedReference readMessage = iterator.next(); + + assertNotNull(readMessage); + + assertEquals(i, readMessage.getMessage().getIntProperty("key").intValue()); + } + + server.stop(); + + OperationContextImpl.clearContext(); + + createServer(); + + pageStore = lookupPageStore(ADDRESS); + + cursor = this.server.getPagingManager().getPageStore(ADDRESS).getCursorProvier().getSubscription(queue.getID()); + iterator = cursor.iterator(); + + for (int i = 0; i < NUM_MESSAGES * 3; i++) + { + if (i % 100 == 0) + System.out.println("Paged " + i); + + if (i >= NUM_MESSAGES * 2 - 1) + { + + HornetQBuffer buffer = RandomUtil.randomBuffer(messageSize, i + 1l); + + ServerMessage msg = new ServerMessageImpl(i, buffer.writerIndex()); + msg.putIntProperty("key", i + 1); + + msg.getBodyBuffer().writeBytes(buffer, 0, buffer.writerIndex()); + + Assert.assertTrue(pageStore.page(msg, ctx, ctx.getContextListing(ADDRESS))); + } + + PagedReference readMessage = iterator.next(); + + assertNotNull(readMessage); + + cursor.ack(readMessage); + + assertEquals(i, readMessage.getMessage().getIntProperty("key").intValue()); + } + + PagedReference readMessage = iterator.next(); + + assertEquals(NUM_MESSAGES * 3, readMessage.getMessage().getIntProperty("key").intValue()); + + cursor.ack(readMessage); + + server.getStorageManager().waitOnOperations(); + + pageStore.flushExecutors(); + + assertFalse(pageStore.isPaging()); + + server.stop(); + createServer(); + + assertFalse(pageStore.isPaging()); + + waitCleanup(); + + assertFalse(lookupPageStore(ADDRESS).isPaging()); + + } + + + public void testConsumeLivePageMultiThread() throws Exception + { + final PagingStoreImpl pageStore = lookupPageStore(ADDRESS); + + pageStore.startPaging(); + + final int NUM_TX = 100; + + final int MSGS_TX = 100; + + final int TOTAL_MSG = NUM_TX * MSGS_TX; + + final int messageSize = 1024; + + PageCursorProvider cursorProvider = this.server.getPagingManager().getPageStore(ADDRESS).getCursorProvier(); + System.out.println("cursorProvider = " + cursorProvider); + + PageSubscription cursor = this.server.getPagingManager() + .getPageStore(ADDRESS) + .getCursorProvier() + .getSubscription(queue.getID()); + + System.out.println("Cursor: " + cursor); + + final StorageManager storage = this.server.getStorageManager(); + + final AtomicInteger exceptions = new AtomicInteger(0); + + Thread t1 = new Thread() + { + public void run() + { + try + { + int count = 0; + + for (int txCount = 0; txCount < NUM_TX; txCount++) + { + + Transaction tx = null; + + if (txCount % 2 == 0) + { + tx = new TransactionImpl(storage); + } + + RoutingContext ctx = generateCTX(tx); + + for (int i = 0 ; i < MSGS_TX; i++) + { + //System.out.println("Sending " + count); + HornetQBuffer buffer = RandomUtil.randomBuffer(messageSize, count); + + ServerMessage msg = new ServerMessageImpl(i, buffer.writerIndex()); + msg.putIntProperty("key", count++); + + msg.getBodyBuffer().writeBytes(buffer, 0, buffer.writerIndex()); + + Assert.assertTrue(pageStore.page(msg, ctx, ctx.getContextListing(ADDRESS))); + } + + if (tx != null) + { + tx.commit(); + } + + } + } + catch (Throwable e) + { + e.printStackTrace(); + exceptions.incrementAndGet(); + } + } + }; + + t1.start(); + + + LinkedListIterator<PagedReference> iterator = cursor.iterator(); + + for (int i = 0 ; i < TOTAL_MSG; i++ ) + { + assertEquals(0, exceptions.get()); + PagedReference ref = null; + for (int repeat = 0 ; repeat < 5; repeat++) + { + ref = iterator.next(); + if (ref == null) + { + Thread.sleep(1000); + } + else + { + break; + } + } + assertNotNull(ref); + + ref.acknowledge(); + assertNotNull(ref); + + System.out.println("Consuming " + ref.getMessage().getIntProperty("key")); + //assertEquals(i, ref.getMessage().getIntProperty("key").intValue()); + } + + assertEquals(0, exceptions.get()); + } + + private RoutingContextImpl generateCTX() + { + return generateCTX(null); + } + + private RoutingContextImpl generateCTX(Transaction tx) + { + RoutingContextImpl ctx = new RoutingContextImpl(tx); + ctx.addQueue(ADDRESS, queue); + + for (Queue q : this.queueList) + { + ctx.addQueue(ADDRESS, q); + } + + return ctx; + } + + /** + * @throws Exception + * @throws InterruptedException + */ + private void waitCleanup() throws Exception, InterruptedException + { + // The cleanup is done asynchronously, so we need to wait some time + long timeout = System.currentTimeMillis() + 10000; + + while (System.currentTimeMillis() < timeout && lookupPageStore(ADDRESS).getNumberOfPages() != 1) + { + Thread.sleep(100); + } + + assertTrue("expected " + lookupPageStore(ADDRESS).getNumberOfPages(), + lookupPageStore(ADDRESS).getNumberOfPages() <= 2); + } + + public void testPrepareScenarios() throws Exception + { + PagingStoreImpl pageStore = lookupPageStore(ADDRESS); + + pageStore.startPaging(); + + final int NUM_MESSAGES = 100; + + final int messageSize = 100 * 1024; + + PageCursorProvider cursorProvider = this.server.getPagingManager().getPageStore(ADDRESS).getCursorProvier(); + System.out.println("cursorProvider = " + cursorProvider); + + PageSubscription cursor = this.server.getPagingManager() + .getPageStore(ADDRESS) + .getCursorProvier() + .getSubscription(queue.getID()); + LinkedListIterator<PagedReference> iterator = cursor.iterator(); + + System.out.println("Cursor: " + cursor); + + StorageManager storage = this.server.getStorageManager(); + + long pgtxRollback = storage.generateUniqueID(); + long pgtxForgotten = storage.generateUniqueID(); + long pgtxCommit = storage.generateUniqueID(); + + Transaction txRollback = pgMessages(storage, pageStore, pgtxRollback, 0, NUM_MESSAGES, messageSize); + pageStore.forceAnotherPage(); + Transaction txForgotten = pgMessages(storage, pageStore, pgtxForgotten, 100, NUM_MESSAGES, messageSize); + pageStore.forceAnotherPage(); + Transaction txCommit = pgMessages(storage, pageStore, pgtxCommit, 200, NUM_MESSAGES, messageSize); + pageStore.forceAnotherPage(); + + addMessages(300, NUM_MESSAGES, messageSize); + + System.out.println("Number of pages - " + pageStore.getNumberOfPages()); + + // First consume what's already there without any tx as nothing was committed + for (int i = 300; i < 400; i++) + { + PagedReference pos = iterator.next(); + assertNotNull("Null at position " + i, pos); + assertEquals(i, pos.getMessage().getIntProperty("key").intValue()); + cursor.ack(pos); + } + + assertNull(iterator.next()); + + cursor.printDebug(); + + txCommit.commit(); + + txRollback.rollback(); + + storage.waitOnOperations(); + + // Second:after pgtxCommit was done + for (int i = 200; i < 300; i++) + { + PagedReference pos = iterator.next(); + assertNotNull(pos); + assertEquals(i, pos.getMessage().getIntProperty("key").intValue()); + cursor.ack(pos); + } + + assertNull(iterator.next()); + + server.getStorageManager().waitOnOperations(); + + server.stop(); + createServer(); + + long timeout = System.currentTimeMillis() + 10000; + + while (System.currentTimeMillis() < timeout && lookupPageStore(ADDRESS).getNumberOfPages() != 1) + { + Thread.sleep(500); + } + assertEquals(1, lookupPageStore(ADDRESS).getNumberOfPages()); + + } + + + public void testLazyCommit() throws Exception + { + PagingStoreImpl pageStore = lookupPageStore(ADDRESS); + + pageStore.startPaging(); + + final int NUM_MESSAGES = 100; + + final int messageSize = 100 * 1024; + + PageCursorProvider cursorProvider = this.server.getPagingManager().getPageStore(ADDRESS).getCursorProvier(); + System.out.println("cursorProvider = " + cursorProvider); + + PageSubscription cursor = this.server.getPagingManager() + .getPageStore(ADDRESS) + .getCursorProvier() + .getSubscription(queue.getID()); + LinkedListIterator<PagedReference> iterator = cursor.iterator(); + + System.out.println("Cursor: " + cursor); + + StorageManager storage = this.server.getStorageManager(); + + long pgtxLazy = storage.generateUniqueID(); + + Transaction txLazy = pgMessages(storage, pageStore, pgtxLazy, 0, NUM_MESSAGES, messageSize); + + addMessages(100, NUM_MESSAGES, messageSize); + + System.out.println("Number of pages - " + pageStore.getNumberOfPages()); + + // First consume what's already there without any tx as nothing was committed + for (int i = 100; i < 200; i++) + { + PagedReference pos = iterator.next(); + assertNotNull("Null at position " + i, pos); + assertEquals(i, pos.getMessage().getIntProperty("key").intValue()); + cursor.ack(pos); + } + + assertNull(iterator.next()); + + txLazy.commit(); + + storage.waitOnOperations(); + + for (int i = 0; i < 100; i++) + { + PagedReference pos = iterator.next(); + assertNotNull("Null at position " + i, pos); + assertEquals(i, pos.getMessage().getIntProperty("key").intValue()); + cursor.ack(pos); + } + + assertNull(iterator.next()); + + waitCleanup(); + + server.stop(); + createServer(); + waitCleanup(); + assertEquals(1, lookupPageStore(ADDRESS).getNumberOfPages()); + + } + + public void testCloseNonPersistentConsumer() throws Exception + { + + final int NUM_MESSAGES = 100; + + PageCursorProvider cursorProvider = lookupCursorProvider(); + + PageSubscription cursor = cursorProvider.createSubscription(11, null, false); + PageSubscriptionImpl cursor2 = (PageSubscriptionImpl)cursorProvider.createSubscription(12, null, false); + + this.queueList.add(new FakeQueue(new SimpleString("a"), 11)); + + this.queueList.add(new FakeQueue(new SimpleString("b"), 12)); + + int numberOfPages = addMessages(NUM_MESSAGES, 1024 * 1024); + + System.out.println("NumberOfPages = " + numberOfPages); + + queue.getPageSubscription().close(); + + PagedReference msg; + LinkedListIterator<PagedReference> iterator = cursor.iterator(); + LinkedListIterator<PagedReference> iterator2 = cursor2.iterator(); + + cursor2.bookmark(new PagePositionImpl(1, -1)); + + int key = 0; + while ((msg = iterator.next()) != null) + { + System.out.println("key = " + key); + assertEquals(key++, msg.getMessage().getIntProperty("key").intValue()); + cursor.ack(msg); + } + assertEquals(NUM_MESSAGES, key); + + forceGC(); + + for (int i = 0; i < 10; i++) + { + assertTrue(iterator2.hasNext()); + msg = iterator2.next(); + assertEquals(i, msg.getMessage().getIntProperty("key").intValue()); + } + + assertSame(cursor2.getProvider(), cursorProvider); + + cursor2.close(); + + lookupPageStore(ADDRESS).flushExecutors(); + + server.stop(); + createServer(); + waitCleanup(); + assertEquals(1, lookupPageStore(ADDRESS).getNumberOfPages()); + + } + + public void testNoCursors() throws Exception + { + + final int NUM_MESSAGES = 100; + + int numberOfPages = addMessages(NUM_MESSAGES, 1024 * 1024); + + ClientSessionFactory sf = createInVMFactory(); + ClientSession session = sf.createSession(); + session.deleteQueue(ADDRESS); + + System.out.println("NumberOfPages = " + numberOfPages); + + server.stop(); + createServer(); + waitCleanup(); + assertEquals(0, lookupPageStore(ADDRESS).getNumberOfPages()); + + } + + public void testFirstMessageInTheMiddle() throws Exception + { + + final int NUM_MESSAGES = 100; + + PageCursorProvider cursorProvider = lookupCursorProvider(); + + PageSubscription cursor = cursorProvider.createSubscription(2, null, false); + + queueList.add(new FakeQueue(new SimpleString("tmp"), 2)); + + int numberOfPages = addMessages(NUM_MESSAGES, 1024 * 1024); + + System.out.println("NumberOfPages = " + numberOfPages); + + PageCache cache = cursorProvider.getPageCache(new PagePositionImpl(5, 0)); + + queue.getPageSubscription().close(); + + PagePosition startingPos = new PagePositionImpl(5, cache.getNumberOfMessages() / 2); + cursor.bookmark(startingPos); + PagedMessage msg = cache.getMessage(startingPos.getMessageNr() + 1); + msg.initMessage(server.getStorageManager()); + int key = msg.getMessage().getIntProperty("key").intValue(); + + msg = null; + + cache = null; + LinkedListIterator<PagedReference> iterator = cursor.iterator(); + + PagedReference msgCursor = null; + while ((msgCursor = iterator.next()) != null) + { + assertEquals(key++, msgCursor.getMessage().getIntProperty("key").intValue()); + cursor.ack(msgCursor); + } + assertEquals(NUM_MESSAGES, key); + + forceGC(); + + // assertTrue(cursorProvider.getCacheSize() < numberOfPages); + + server.stop(); + createServer(); + waitCleanup(); + assertEquals(1, lookupPageStore(ADDRESS).getNumberOfPages()); + } + + public void testFirstMessageInTheMiddlePersistent() throws Exception + { + + final int NUM_MESSAGES = 100; + + int numberOfPages = addMessages(NUM_MESSAGES, 1024 * 1024); + + System.out.println("NumberOfPages = " + numberOfPages); + + PageCursorProvider cursorProvider = lookupCursorProvider(); + + PageCache cache = cursorProvider.getPageCache(new PagePositionImpl(5, 0)); + + PageSubscription cursor = cursorProvider.getSubscription(queue.getID()); + PagePosition startingPos = new PagePositionImpl(5, cache.getNumberOfMessages() / 2); + cursor.bookmark(startingPos); + + // We can't proceed until the operation has finished + server.getStorageManager().waitOnOperations(); + + PagedMessage msg = cache.getMessage(startingPos.getMessageNr() + 1); + msg.initMessage(server.getStorageManager()); + int initialKey = msg.getMessage().getIntProperty("key").intValue(); + int key = initialKey; + + msg = null; + + cache = null; + + LinkedListIterator<PagedReference> iterator = cursor.iterator(); + + PagedReference msgCursor = null; + while ((msgCursor = iterator.next()) != null) + { + assertEquals(key++, msgCursor.getMessage().getIntProperty("key").intValue()); + } + assertEquals(NUM_MESSAGES, key); + + server.stop(); + + OperationContextImpl.clearContext(); + + createServer(); + + cursorProvider = lookupCursorProvider(); + cursor = cursorProvider.getSubscription(queue.getID()); + key = initialKey; + iterator = cursor.iterator(); + while ((msgCursor = iterator.next()) != null) + { + assertEquals(key++, msgCursor.getMessage().getIntProperty("key").intValue()); + cursor.ack(msgCursor); + } + + forceGC(); + + assertTrue(cursorProvider.getCacheSize() < numberOfPages); + + server.stop(); + createServer(); + waitCleanup(); + assertEquals(1, lookupPageStore(ADDRESS).getNumberOfPages()); + + } + + private int tstProperty(ServerMessage msg) + { + return msg.getIntProperty("key").intValue(); + } + + public void testMultipleIterators() throws Exception + { + + final int NUM_MESSAGES = 10; + + int numberOfPages = addMessages(NUM_MESSAGES, 1024 * 1024); + + System.out.println("NumberOfPages = " + numberOfPages); + + PageCursorProvider cursorProvider = lookupCursorProvider(); + + PageSubscription cursor = cursorProvider.getSubscription(queue.getID()); + + LinkedListIterator<PagedReference> iter = cursor.iterator(); + + LinkedListIterator<PagedReference> iter2 = cursor.iterator(); + + assertTrue(iter.hasNext()); + + PagedReference msg1 = iter.next(); + + PagedReference msg2 = iter2.next(); + + assertEquals(tstProperty(msg1.getMessage()), tstProperty(msg2.getMessage())); + + System.out.println("property = " + tstProperty(msg1.getMessage())); + + msg1 = iter.next(); + + assertEquals(1, tstProperty(msg1.getMessage())); + + iter.remove(); + + msg2 = iter2.next(); + + assertEquals(2, tstProperty(msg2.getMessage())); + + iter2.repeat(); + + msg2 = iter2.next(); + + assertEquals(2, tstProperty(msg2.getMessage())); + + iter2.repeat(); + + assertEquals(2, tstProperty(msg2.getMessage())); + + msg1 = iter.next(); + + assertEquals(2, tstProperty(msg1.getMessage())); + + iter.repeat(); + + msg1 = iter.next(); + + assertEquals(2, tstProperty(msg1.getMessage())); + + assertTrue(iter2.hasNext()); + + + } + + private int addMessages(final int numMessages, final int messageSize) throws Exception + { + return addMessages(0, numMessages, messageSize); + } + + /** + * @param numMessages + * @param pageStore + * @throws Exception + */ + private int addMessages(final int start, final int numMessages, final int messageSize) throws Exception + { + PagingStoreImpl pageStore = lookupPageStore(ADDRESS); + + pageStore.startPaging(); + + RoutingContext ctx = generateCTX(); + + for (int i = start; i < start + numMessages; i++) + { + if (i % 100 == 0) + System.out.println("Paged " + i); + HornetQBuffer buffer = RandomUtil.randomBuffer(messageSize, i + 1l); + + ServerMessage msg = new ServerMessageImpl(i, buffer.writerIndex()); + msg.putIntProperty("key", i); + // to be used on tests that are validating filters + msg.putBooleanProperty("even", i % 2 == 0); + + msg.getBodyBuffer().writeBytes(buffer, 0, buffer.writerIndex()); + + Assert.assertTrue(pageStore.page(msg, ctx, ctx.getContextListing(ADDRESS))); + } + + return pageStore.getNumberOfPages(); + } + + /** + * @return + * @throws Exception + */ + private PagingStoreImpl lookupPageStore(SimpleString address) throws Exception + { + return (PagingStoreImpl)server.getPagingManager().getPageStore(address); + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + protected void tearDown() throws Exception + { + server.stop(); + server = null; + queue = null; + queueList = null; + super.tearDown(); + } + + protected void setUp() throws Exception + { + super.setUp(); + OperationContextImpl.clearContext(); + System.out.println("Tmp:" + getTemporaryDir()); + + queueList = new ArrayList<Queue>(); + + createServer(); + } + + /** + * @throws Exception + */ + private void createServer() throws Exception + { + OperationContextImpl.clearContext(); + + Configuration config = createDefaultConfig(); + + config.setJournalSyncNonTransactional(true); + + server = createServer(true, config, PAGE_SIZE, PAGE_MAX, new HashMap<String, AddressSettings>()); + + server.start(); + + try + { + queue = server.createQueue(ADDRESS, ADDRESS, null, true, false); + queue.pause(); + } + catch (Exception ignored) + { + } + } + + /** + * @return + * @throws Exception + */ + private PageSubscription createNonPersistentCursor(Filter filter) throws Exception + { + long id = server.getStorageManager().generateUniqueID(); + queueList.add(new FakeQueue(new SimpleString(filter.toString()), id)); + return lookupCursorProvider().createSubscription(id, filter, false); + } + + /** + * @return + * @throws Exception + */ + private PageCursorProvider lookupCursorProvider() throws Exception + { + return lookupPageStore(ADDRESS).getCursorProvier(); + } + + /** + * @param storage + * @param pageStore + * @param pgParameter + * @param start + * @param NUM_MESSAGES + * @param messageSize + * @throws Exception + */ + private Transaction pgMessages(StorageManager storage, + PagingStoreImpl pageStore, + long pgParameter, + int start, + final int NUM_MESSAGES, + final int messageSize) throws Exception + { + + TransactionImpl txImpl = new TransactionImpl(pgParameter, null, storage); + + RoutingContext ctx = generateCTX(txImpl); + + for (int i = start; i < start + NUM_MESSAGES; i++) + { + HornetQBuffer buffer = RandomUtil.randomBuffer(messageSize, i + 1l); + ServerMessage msg = new ServerMessageImpl(storage.generateUniqueID(), buffer.writerIndex()); + msg.getBodyBuffer().writeBytes(buffer, 0, buffer.writerIndex()); + msg.putIntProperty("key", i); + pageStore.page(msg, ctx, ctx.getContextListing(ADDRESS)); + } + + return txImpl; + + } + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- + +} Modified: trunk/tests/src/org/hornetq/tests/timing/core/server/impl/QueueImplTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/timing/core/server/impl/QueueImplTest.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/tests/src/org/hornetq/tests/timing/core/server/impl/QueueImplTest.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -70,6 +70,7 @@ new SimpleString("address1"), new SimpleString("queue1"), null, + null, false, true, scheduledExecutor, @@ -145,6 +146,7 @@ new SimpleString("address1"), new SimpleString("queue1"), null, + null, false, true, scheduledExecutor, @@ -253,6 +255,7 @@ new SimpleString("address1"), QueueImplTest.queue1, null, + null, false, true, scheduledExecutor, Modified: trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PageImplTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PageImplTest.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PageImplTest.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -106,10 +106,10 @@ for (int i = 0; i < msgs.size(); i++) { - Assert.assertEquals(simpleDestination, msgs.get(i).getMessage(null).getAddress()); + Assert.assertEquals(simpleDestination, msgs.get(i).getMessage().getAddress()); UnitTestCase.assertEqualsByteArrays(buffers.get(i).toByteBuffer().array(), msgs.get(i) - .getMessage(null) + .getMessage() .getBodyBuffer() .toByteBuffer() .array()); @@ -178,10 +178,10 @@ for (int i = 0; i < msgs.size(); i++) { - Assert.assertEquals(simpleDestination, msgs.get(i).getMessage(null).getAddress()); + Assert.assertEquals(simpleDestination, msgs.get(i).getMessage().getAddress()); UnitTestCase.assertEqualsByteArrays(buffers.get(i).toByteBuffer().array(), msgs.get(i) - .getMessage(null) + .getMessage() .getBodyBuffer() .toByteBuffer() .array()); @@ -223,7 +223,7 @@ msg.setAddress(simpleDestination); - page.write(new PagedMessageImpl(msg)); + page.write(new PagedMessageImpl(msg, new long [0])); Assert.assertEquals(initialNumberOfMessages + i + 1, page.getNumberOfMessages()); } Added: trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagePositionTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagePositionTest.java (rev 0) +++ trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagePositionTest.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -0,0 +1,51 @@ +/* + * Copyright 2010 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.paging.impl; + +import org.hornetq.tests.util.UnitTestCase; + +/** + * A PagePositionTest + * + * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * + */ +public class PagePositionTest extends UnitTestCase +{ + + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + + public void testNextSequenceOf() + { + + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- + +} 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 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingManagerImplTest.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -28,6 +28,7 @@ import org.hornetq.core.paging.impl.TestSupportPageStore; import org.hornetq.core.persistence.impl.nullpm.NullStorageManager; import org.hornetq.core.server.ServerMessage; +import org.hornetq.core.server.impl.RoutingContextImpl; import org.hornetq.core.server.impl.ServerMessageImpl; import org.hornetq.core.settings.HierarchicalRepository; import org.hornetq.core.settings.impl.AddressFullMessagePolicy; @@ -81,11 +82,11 @@ ServerMessage msg = createMessage(1l, new SimpleString("simple-test"), createRandomBuffer(10)); - Assert.assertFalse(store.page(msg)); + Assert.assertFalse(store.page(msg, new RoutingContextImpl(null))); store.startPaging(); - Assert.assertTrue(store.page(msg)); + Assert.assertTrue(store.page(msg, new RoutingContextImpl(null))); Page page = store.depage(); @@ -98,7 +99,7 @@ Assert.assertEquals(1, msgs.size()); UnitTestCase.assertEqualsByteArrays(msg.getBodyBuffer().writerIndex(), msg.getBodyBuffer().toByteBuffer().array(), msgs.get(0) - .getMessage(null) + .getMessage() .getBodyBuffer() .toByteBuffer() .array()); @@ -107,7 +108,7 @@ Assert.assertNull(store.depage()); - Assert.assertFalse(store.page(msg)); + Assert.assertFalse(store.page(msg, new RoutingContextImpl(null))); } // Package protected --------------------------------------------- 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 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -46,6 +46,7 @@ import org.hornetq.core.paging.PagingManager; import org.hornetq.core.paging.PagingStore; import org.hornetq.core.paging.PagingStoreFactory; +import org.hornetq.core.paging.cursor.PagePosition; import org.hornetq.core.paging.impl.PageTransactionInfoImpl; import org.hornetq.core.paging.impl.PagingStoreImpl; import org.hornetq.core.paging.impl.TestSupportPageStore; @@ -63,6 +64,7 @@ import org.hornetq.core.server.Queue; import org.hornetq.core.server.ServerMessage; import org.hornetq.core.server.group.impl.GroupBinding; +import org.hornetq.core.server.impl.RoutingContextImpl; import org.hornetq.core.server.impl.ServerMessageImpl; import org.hornetq.core.settings.HierarchicalRepository; import org.hornetq.core.settings.impl.AddressFullMessagePolicy; @@ -72,6 +74,7 @@ import org.hornetq.tests.unit.core.server.impl.fakes.FakePostOffice; import org.hornetq.tests.util.RandomUtil; import org.hornetq.tests.util.UnitTestCase; +import org.hornetq.utils.ExecutorFactory; import org.hornetq.utils.UUID; /** @@ -142,7 +145,7 @@ null, PagingStoreImplTest.destinationTestName, addressSettings, - executor, + getExecutorFactory(), true); storeImpl.start(); @@ -178,7 +181,7 @@ storeFactory, PagingStoreImplTest.destinationTestName, addressSettings, - executor, + getExecutorFactory(), true); storeImpl.start(); @@ -200,7 +203,7 @@ Assert.assertTrue(storeImpl.isPaging()); - Assert.assertTrue(storeImpl.page(msg)); + Assert.assertTrue(storeImpl.page(msg, new RoutingContextImpl(null))); Assert.assertEquals(1, storeImpl.getNumberOfPages()); @@ -214,12 +217,12 @@ null, PagingStoreImplTest.destinationTestName, addressSettings, - executor, + getExecutorFactory(), true); storeImpl.start(); - Assert.assertEquals(2, storeImpl.getNumberOfPages()); + Assert.assertEquals(1, storeImpl.getNumberOfPages()); } @@ -241,7 +244,7 @@ storeFactory, PagingStoreImplTest.destinationTestName, addressSettings, - executor, + getExecutorFactory(), true); storeImpl.start(); @@ -263,7 +266,7 @@ ServerMessage msg = createMessage(i, storeImpl, destination, buffer); - Assert.assertTrue(storeImpl.page(msg)); + Assert.assertTrue(storeImpl.page(msg, new RoutingContextImpl(null))); } Assert.assertEquals(1, storeImpl.getNumberOfPages()); @@ -288,7 +291,7 @@ for (int i = 0; i < numMessages; i++) { HornetQBuffer horn1 = buffers.get(i); - HornetQBuffer horn2 = msg.get(i).getMessage(null).getBodyBuffer(); + HornetQBuffer horn2 = msg.get(i).getMessage().getBodyBuffer(); horn1.resetReaderIndex(); horn2.resetReaderIndex(); for (int j = 0; j < horn1.writerIndex(); j++) @@ -316,7 +319,7 @@ storeFactory, PagingStoreImplTest.destinationTestName, addressSettings, - executor, + getExecutorFactory(), true); storeImpl.start(); @@ -343,7 +346,7 @@ ServerMessage msg = createMessage(i, storeImpl, destination, buffer); - Assert.assertTrue(storeImpl.page(msg)); + Assert.assertTrue(storeImpl.page(msg, new RoutingContextImpl(null))); } Assert.assertEquals(2, storeImpl.getNumberOfPages()); @@ -355,6 +358,8 @@ for (int pageNr = 0; pageNr < 2; pageNr++) { Page page = storeImpl.depage(); + + System.out.println("numberOfPages = " + storeImpl.getNumberOfPages()); page.open(); @@ -366,10 +371,8 @@ for (int i = 0; i < 5; i++) { - Assert.assertEquals(sequence++, msg.get(i).getMessage(null).getMessageID()); - UnitTestCase.assertEqualsBuffers(18, buffers.get(pageNr * 5 + i), msg.get(i) - .getMessage(null) - .getBodyBuffer()); + Assert.assertEquals(sequence++, msg.get(i).getMessage().getMessageID()); + UnitTestCase.assertEqualsBuffers(18, buffers.get(pageNr * 5 + i), msg.get(i).getMessage().getBodyBuffer()); } } @@ -379,7 +382,7 @@ ServerMessage msg = createMessage(1, storeImpl, destination, buffers.get(0)); - Assert.assertTrue(storeImpl.page(msg)); + Assert.assertTrue(storeImpl.page(msg, new RoutingContextImpl(null))); Page newPage = storeImpl.depage(); @@ -397,11 +400,11 @@ Assert.assertFalse(storeImpl.isPaging()); - Assert.assertFalse(storeImpl.page(msg)); + Assert.assertFalse(storeImpl.page(msg, new RoutingContextImpl(null))); storeImpl.startPaging(); - Assert.assertTrue(storeImpl.page(msg)); + Assert.assertTrue(storeImpl.page(msg, new RoutingContextImpl(null))); Page page = storeImpl.depage(); @@ -411,9 +414,9 @@ Assert.assertEquals(1, msgs.size()); - Assert.assertEquals(1l, msgs.get(0).getMessage(null).getMessageID()); + Assert.assertEquals(1l, msgs.get(0).getMessage().getMessageID()); - UnitTestCase.assertEqualsBuffers(18, buffers.get(0), msgs.get(0).getMessage(null).getBodyBuffer()); + UnitTestCase.assertEqualsBuffers(18, buffers.get(0), msgs.get(0).getMessage().getBodyBuffer()); Assert.assertEquals(1, storeImpl.getNumberOfPages()); @@ -463,7 +466,7 @@ storeFactory, new SimpleString("test"), settings, - executor, + getExecutorFactory(), true); storeImpl.start(); @@ -497,7 +500,7 @@ // 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(id, storeImpl, destination, createRandomBuffer(id, 5)); - if (storeImpl.page(msg)) + if (storeImpl.page(msg, new RoutingContextImpl(null))) { buffers.put(id, msg); } @@ -592,14 +595,14 @@ for (PagedMessage msg : msgs) { - long id = msg.getMessage(null).getBodyBuffer().readLong(); - msg.getMessage(null).getBodyBuffer().resetReaderIndex(); + long id = msg.getMessage().getBodyBuffer().readLong(); + msg.getMessage().getBodyBuffer().resetReaderIndex(); ServerMessage msgWritten = buffers.remove(id); - buffers2.put(id, msg.getMessage(null)); + buffers2.put(id, msg.getMessage()); Assert.assertNotNull(msgWritten); - Assert.assertEquals(msg.getMessage(null).getAddress(), msgWritten.getAddress()); - UnitTestCase.assertEqualsBuffers(10, msgWritten.getBodyBuffer(), msg.getMessage(null).getBodyBuffer()); + Assert.assertEquals(msg.getMessage().getAddress(), msgWritten.getAddress()); + UnitTestCase.assertEqualsBuffers(10, msgWritten.getBodyBuffer(), msg.getMessage().getBodyBuffer()); } } @@ -626,7 +629,7 @@ storeFactory, new SimpleString("test"), settings, - executor, + getExecutorFactory(), true); storeImpl2.start(); @@ -641,8 +644,10 @@ long lastMessageId = messageIdGenerator.incrementAndGet(); ServerMessage lastMsg = createMessage(lastMessageId, storeImpl, destination, createRandomBuffer(lastMessageId, 5)); + + storeImpl2.forceAnotherPage(); - storeImpl2.page(lastMsg); + storeImpl2.page(lastMsg, new RoutingContextImpl(null)); buffers2.put(lastMessageId, lastMsg); Page lastPage = null; @@ -665,13 +670,13 @@ for (PagedMessage msg : msgs) { - long id = msg.getMessage(null).getBodyBuffer().readLong(); + long id = msg.getMessage().getBodyBuffer().readLong(); ServerMessage msgWritten = buffers2.remove(id); Assert.assertNotNull(msgWritten); - Assert.assertEquals(msg.getMessage(null).getAddress(), msgWritten.getAddress()); + Assert.assertEquals(msg.getMessage().getAddress(), msgWritten.getAddress()); UnitTestCase.assertEqualsByteArrays(msgWritten.getBodyBuffer().writerIndex(), msgWritten.getBodyBuffer().toByteBuffer().array(), - msg.getMessage(null).getBodyBuffer().toByteBuffer().array()); + msg.getMessage().getBodyBuffer().toByteBuffer().array()); } } @@ -680,14 +685,54 @@ lastPage.close(); Assert.assertEquals(1, lastMessages.size()); - lastMessages.get(0).getMessage(null).getBodyBuffer().resetReaderIndex(); - Assert.assertEquals(lastMessages.get(0).getMessage(null).getBodyBuffer().readLong(), lastMessageId); + lastMessages.get(0).getMessage().getBodyBuffer().resetReaderIndex(); + Assert.assertEquals(lastMessages.get(0).getMessage().getBodyBuffer().readLong(), lastMessageId); Assert.assertEquals(0, buffers2.size()); Assert.assertEquals(0, storeImpl.getAddressSize()); } + public void testRestartPage() throws Throwable + { + clearData(); + SequentialFileFactory factory = new NIOSequentialFileFactory(this.getPageDir()); + + PagingStoreFactory storeFactory = new FakeStoreFactory(factory); + + final int MAX_SIZE = 1024 * 10; + + AddressSettings settings = new AddressSettings(); + settings.setPageSizeBytes(MAX_SIZE); + settings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE); + + final TestSupportPageStore storeImpl = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, + createMockManager(), + createStorageManagerMock(), + createPostOfficeMock(), + factory, + storeFactory, + new SimpleString("test"), + settings, + getExecutorFactory(), + true); + + storeImpl.start(); + + Assert.assertEquals(0, storeImpl.getNumberOfPages()); + + // Marked the store to be paged + storeImpl.startPaging(); + + storeImpl.depage(); + + assertNull(storeImpl.getCurrentPage()); + + storeImpl.startPaging(); + + assertNotNull(storeImpl.getCurrentPage()); + } + public void testOrderOnPaging() throws Throwable { clearData(); @@ -709,7 +754,7 @@ storeFactory, new SimpleString("test"), settings, - executor, + getExecutorFactory(), true); storeImpl.start(); @@ -750,7 +795,7 @@ // Just using the internal API to remove it from the page file system ServerMessage msg = createMessage(i, storeImpl, destination, createRandomBuffer(i, 1024)); msg.putLongProperty("count", i); - while (!storeImpl.page(msg)) + while (!storeImpl.page(msg, new RoutingContextImpl(null))) { storeImpl.startPaging(); } @@ -791,22 +836,22 @@ { page.open(); List<PagedMessage> messages = page.read(); - + for (PagedMessage pgmsg : messages) { - ServerMessage msg = pgmsg.getMessage(null); + ServerMessage msg = pgmsg.getMessage(); assertEquals(msgsRead++, msg.getMessageID()); assertEquals(msg.getMessageID(), msg.getLongProperty("count").longValue()); } - + page.close(); page.delete(); } else { - System.out.println("Depaged!!!!"); + System.out.println("Depaged!!!! numerOfMessages = " + msgsRead + " of " + NUMBER_OF_MESSAGES); Thread.sleep(500); } } @@ -830,7 +875,7 @@ storeImpl.stop(); - for (Throwable e: errors) + for (Throwable e : errors) { throw e; } @@ -854,6 +899,18 @@ return new FakePostOffice(); } + private ExecutorFactory getExecutorFactory() + { + return new ExecutorFactory() + { + + public Executor getExecutor() + { + return executor; + } + }; + } + private ServerMessage createMessage(final long id, final PagingStore store, final SimpleString destination, @@ -873,18 +930,9 @@ return msg; } - private HornetQBuffer createRandomBuffer(final long id, final int size) + protected HornetQBuffer createRandomBuffer(final long id, final int size) { - HornetQBuffer buffer = HornetQBuffers.fixedBuffer(size + 8); - - buffer.writeLong(id); - - for (int j = 8; j < buffer.capacity(); j++) - { - buffer.writeByte((byte)66); - } - - return buffer; + return RandomUtil.randomBuffer(size, id); } // Protected ---------------------------------------------------- @@ -1035,6 +1083,15 @@ return null; } + /* (non-Javadoc) + * @see org.hornetq.core.paging.PagingManager#processReload() + */ + public void processReload() + { + // TODO Auto-generated method stub + + } + } class FakeStorageManager implements StorageManager @@ -1162,6 +1219,7 @@ final PagingManager pagingManager, final ResourceManager resourceManager, final Map<Long, Queue> queues, + Map<Long, QueueBindingInfo> queueInfos, final Map<SimpleString, List<Pair<byte[], Long>>> duplicateIDMap) throws Exception { return new JournalLoadInformation(); @@ -1510,6 +1568,42 @@ { } + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#storeCursorAcknowledge(long, org.hornetq.core.paging.cursor.PagePosition) + */ + public void storeCursorAcknowledge(long queueID, PagePosition position) + { + // TODO Auto-generated method stub + + } + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#storeCursorAcknowledgeTransactional(long, long, org.hornetq.core.paging.cursor.PagePosition) + */ + public void storeCursorAcknowledgeTransactional(long txID, long queueID, PagePosition position) + { + // TODO Auto-generated method stub + + } + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#deleteCursorAcknowledgeTransactional(long, long) + */ + public void deleteCursorAcknowledgeTransactional(long txID, long ackID) throws Exception + { + // TODO Auto-generated method stub + + } + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#updatePageTransaction(org.hornetq.core.paging.PageTransactionInfo, int) + */ + public void updatePageTransaction(PageTransactionInfo pageTransaction, int depage) throws Exception + { + // TODO Auto-generated method stub + + } + } class FakeStoreFactory implements PagingStoreFactory @@ -1546,7 +1640,7 @@ /* (non-Javadoc) * @see org.hornetq.core.paging.PagingStoreFactory#newStore(org.hornetq.utils.SimpleString, org.hornetq.core.settings.impl.AddressSettings) */ - public PagingStore newStore(final SimpleString destinationName, final AddressSettings addressSettings) throws Exception + public PagingStore newStore(final SimpleString destinationName, final AddressSettings addressSettings) { return null; } 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 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -87,7 +87,7 @@ { final FakeBinding fake = new FakeBinding(new SimpleString("a")); - final BindingsImpl bind = new BindingsImpl(null); + final BindingsImpl bind = new BindingsImpl(null, null); bind.addBinding(fake); bind.addBinding(new FakeBinding(new SimpleString("a"))); bind.addBinding(new FakeBinding(new SimpleString("a"))); Modified: trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/DuplicateDetectionUnitTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/DuplicateDetectionUnitTest.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/DuplicateDetectionUnitTest.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -111,6 +111,7 @@ new FakePagingManager(), new ResourceManagerImpl(0, 0, scheduledThreadPool), new HashMap<Long, Queue>(), + null, mapDups); Assert.assertEquals(0, mapDups.size()); @@ -132,6 +133,7 @@ new FakePagingManager(), new ResourceManagerImpl(0, 0, scheduledThreadPool), new HashMap<Long, Queue>(), + null, mapDups); Assert.assertEquals(1, mapDups.size()); @@ -160,6 +162,7 @@ new FakePagingManager(), new ResourceManagerImpl(0, 0, scheduledThreadPool), new HashMap<Long, Queue>(), + null, mapDups); Assert.assertEquals(1, mapDups.size()); @@ -320,6 +323,16 @@ return null; } + + + + /* (non-Javadoc) + * @see org.hornetq.core.paging.PagingManager#processReload() + */ + public void processReload() + { + } + } } Modified: trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/FakeQueue.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/FakeQueue.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/FakeQueue.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -20,6 +20,7 @@ import org.hornetq.api.core.SimpleString; import org.hornetq.core.filter.Filter; +import org.hornetq.core.paging.cursor.PageSubscription; import org.hornetq.core.server.Consumer; import org.hornetq.core.server.MessageReference; import org.hornetq.core.server.Queue; @@ -91,10 +92,18 @@ } private final SimpleString name; + + private final long id; public FakeQueue(final SimpleString name) { + this(name, 0); + } + + public FakeQueue(final SimpleString name, final long id) + { this.name = name; + this.id = id; } /* (non-Javadoc) @@ -359,8 +368,7 @@ */ public long getID() { - // TODO Auto-generated method stub - return 0; + return id; } /* (non-Javadoc) @@ -597,4 +605,13 @@ } + /* (non-Javadoc) + * @see org.hornetq.core.server.Queue#getPageSubscription() + */ + public PageSubscription getPageSubscription() + { + // TODO Auto-generated method stub + return null; + } + } \ No newline at end of file Modified: trunk/tests/src/org/hornetq/tests/unit/core/server/impl/fakes/FakeQueueFactory.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/server/impl/fakes/FakeQueueFactory.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/tests/src/org/hornetq/tests/unit/core/server/impl/fakes/FakeQueueFactory.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -19,6 +19,7 @@ import org.hornetq.api.core.SimpleString; import org.hornetq.core.filter.Filter; +import org.hornetq.core.paging.cursor.PageSubscription; import org.hornetq.core.postoffice.PostOffice; import org.hornetq.core.server.Queue; import org.hornetq.core.server.QueueFactory; @@ -43,6 +44,7 @@ final SimpleString address, final SimpleString name, final Filter filter, + final PageSubscription subscription, final boolean durable, final boolean temporary) { @@ -50,6 +52,7 @@ address, name, filter, + subscription, durable, temporary, scheduledExecutor, Added: trunk/tests/src/org/hornetq/tests/unit/util/SoftValueMapTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/util/SoftValueMapTest.java (rev 0) +++ trunk/tests/src/org/hornetq/tests/unit/util/SoftValueMapTest.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -0,0 +1,72 @@ +/* + * Copyright 2010 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.util; + +import org.hornetq.tests.util.UnitTestCase; +import org.hornetq.utils.SoftValueHashMap; + +/** + * A SoftValueMapTest + * + * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * + */ +public class SoftValueMapTest extends UnitTestCase +{ + + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + + public void testEvictions() + { + forceGC(); + long maxMemory = Runtime.getRuntime().maxMemory() - Runtime.getRuntime().freeMemory(); + + // each buffer will be 1/10th of the maxMemory + int bufferSize = (int)(maxMemory / 100); + + SoftValueHashMap<Long, byte[]> softCache = new SoftValueHashMap<Long, byte[]>(); + + final int MAX_ELEMENTS = 1000; + + for (long i = 0 ; i < MAX_ELEMENTS; i++) + { + softCache.put(i, new byte[bufferSize]); + } + + + assertTrue(softCache.size() < MAX_ELEMENTS); + + System.out.println("Soft cache has " + softCache.size() + " elements"); + } + + + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- + +} Modified: trunk/tests/src/org/hornetq/tests/util/RandomUtil.java =================================================================== --- trunk/tests/src/org/hornetq/tests/util/RandomUtil.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/tests/src/org/hornetq/tests/util/RandomUtil.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -18,6 +18,8 @@ import javax.transaction.xa.Xid; +import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.api.core.HornetQBuffers; import org.hornetq.api.core.SimpleString; import org.hornetq.core.transaction.impl.XidImpl; @@ -72,7 +74,27 @@ return Math.abs(RandomUtil.randomInt()); } + + public static HornetQBuffer randomBuffer(final int size, final long... data) + { + HornetQBuffer buffer = HornetQBuffers.fixedBuffer(size + 8 * data.length); + + for (long d : data) + { + buffer.writeLong(d); + } + + for (int i = 0 ; i < size; i++) + { + buffer.writeByte((byte)randomByte()); + } + + return buffer; + } + + + public static int randomInterval(final int min, final int max) { return min + randomMax(max - min); Modified: trunk/tests/src/org/hornetq/tests/util/ServiceTestBase.java =================================================================== --- trunk/tests/src/org/hornetq/tests/util/ServiceTestBase.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/tests/src/org/hornetq/tests/util/ServiceTestBase.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -25,6 +25,7 @@ import org.hornetq.api.core.TransportConfiguration; import org.hornetq.api.core.client.ClientMessage; import org.hornetq.api.core.client.ClientSession; +import org.hornetq.api.core.client.ClientSessionFactory; import org.hornetq.api.core.client.HornetQClient; import org.hornetq.core.client.impl.ClientSessionFactoryImpl; import org.hornetq.core.config.Configuration; @@ -334,6 +335,15 @@ return createInVMFactory(); } } + + protected void createQueue(String address, String queue) throws Exception + { + ClientSessionFactory sf = createInVMFactory(); + ClientSession session = sf.createSession(); + session.createQueue(address, queue); + session.close(); + sf.close(); + } protected ClientSessionFactoryImpl createInVMFactory() { Modified: trunk/tests/src/org/hornetq/tests/util/UnitTestCase.java =================================================================== --- trunk/tests/src/org/hornetq/tests/util/UnitTestCase.java 2010-11-17 03:00:47 UTC (rev 9904) +++ trunk/tests/src/org/hornetq/tests/util/UnitTestCase.java 2010-11-17 05:07:27 UTC (rev 9905) @@ -26,6 +26,7 @@ import java.io.OutputStream; import java.io.PrintWriter; import java.io.StringWriter; +import java.lang.ref.Reference; import java.lang.ref.WeakReference; import java.net.ServerSocket; import java.nio.ByteBuffer; @@ -168,7 +169,7 @@ } } - public static void forceGC(WeakReference<?> ref, long timeout) + public static void forceGC(Reference<?> ref, long timeout) { long waitUntil = System.currentTimeMillis() + timeout; // A loop that will wait GC, using the minimal time as possible