[hornetq-commits] JBoss hornetq SVN: r10049 - in trunk: src/main/org/hornetq/api/core/client and 19 other directories.

Author: clebert.suconic(a)jboss.com Date: 2010-12-16 21:20:26 -0500 (Thu, 16 Dec 2010) New Revision: 10049 Modified: trunk/examples/jms/large-message/src/org/hornetq/jms/example/LargeMessageExample.java trunk/src/main/org/hornetq/api/core/client/ClientSession.java trunk/src/main/org/hornetq/api/core/management/QueueControl.java trunk/src/main/org/hornetq/api/jms/management/DestinationControl.java trunk/src/main/org/hornetq/api/jms/management/JMSQueueControl.java trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java trunk/src/main/org/hornetq/core/management/impl/QueueControlImpl.java trunk/src/main/org/hornetq/core/messagecounter/MessageCounter.java trunk/src/main/org/hornetq/core/paging/cursor/PageSubscription.java trunk/src/main/org/hornetq/core/paging/cursor/impl/PageCursorProviderImpl.java trunk/src/main/org/hornetq/core/paging/cursor/impl/PageSubscriptionImpl.java trunk/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionQueueQueryResponseMessage.java trunk/src/main/org/hornetq/core/server/Queue.java trunk/src/main/org/hornetq/core/server/QueueQueryResult.java trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java trunk/src/main/org/hornetq/jms/management/impl/JMSQueueControlImpl.java trunk/src/main/org/hornetq/jms/management/impl/JMSTopicControlImpl.java trunk/tests/jms-tests/src/org/hornetq/jms/tests/HornetQServerTestCase.java trunk/tests/jms-tests/src/org/hornetq/jms/tests/TransactedSessionTest.java trunk/tests/jms-tests/src/org/hornetq/jms/tests/tools/container/LocalTestServer.java trunk/tests/jms-tests/src/org/hornetq/jms/tests/tools/container/Server.java trunk/tests/src/org/hornetq/tests/integration/jms/bridge/BridgeTestBase.java trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSQueueControlUsingJMSTest.java trunk/tests/src/org/hornetq/tests/integration/management/QueueControlUsingCoreTest.java trunk/tests/src/org/hornetq/tests/stress/paging/PageCursorStressTest.java trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/FakeQueue.java Log: more PageCounter changes Modified: trunk/examples/jms/large-message/src/org/hornetq/jms/example/LargeMessageExample.java =================================================================== --- trunk/examples/jms/large-message/src/org/hornetq/jms/example/LargeMessageExample.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/examples/jms/large-message/src/org/hornetq/jms/example/LargeMessageExample.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -49,9 +49,9 @@ // This may take some considerable time to create, send and consume - if it takes too long or you don't have // enough disk space just reduce the file size here - // private final long FILE_SIZE = 256L * 1024 * 1024; + private final long FILE_SIZE = 256L * 1024 * 1024; - private final long FILE_SIZE = 10L * 1024 * 1024 * 1024; // 10 GiB message + //private final long FILE_SIZE = 10L * 1024 * 1024 * 1024; // 10 GiB message @Override public boolean runExample() throws Exception Modified: trunk/src/main/org/hornetq/api/core/client/ClientSession.java =================================================================== --- trunk/src/main/org/hornetq/api/core/client/ClientSession.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/src/main/org/hornetq/api/core/client/ClientSession.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -74,7 +74,7 @@ /** * Returns the number of messages in the queue. */ - int getMessageCount(); + long getMessageCount(); /** * Returns the queue's filter string (or <code>null</code> if the queue has no filter). Modified: trunk/src/main/org/hornetq/api/core/management/QueueControl.java =================================================================== --- trunk/src/main/org/hornetq/api/core/management/QueueControl.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/src/main/org/hornetq/api/core/management/QueueControl.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -60,7 +60,7 @@ /** * Returns the number of messages currently in this queue. */ - int getMessageCount(); + long getMessageCount(); /** * Returns the number of scheduled messages in this queue. @@ -142,7 +142,7 @@ * Using {@code null} or an empty filter will count <em>all</em> messages from this queue. */ @Operation(desc = "Returns the number of the messages in the queue matching the given filter", impact = MBeanOperationInfo.INFO) - int countMessages(@Parameter(name = "filter", desc = "A message filter (can be empty)") String filter) throws Exception; + long countMessages(@Parameter(name = "filter", desc = "A message filter (can be empty)") String filter) throws Exception; /** * Removes the message corresponding to the specified message ID. Modified: trunk/src/main/org/hornetq/api/jms/management/DestinationControl.java =================================================================== --- trunk/src/main/org/hornetq/api/jms/management/DestinationControl.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/src/main/org/hornetq/api/jms/management/DestinationControl.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -45,7 +45,7 @@ /** * Returns the number of messages currently in this destination. */ - int getMessageCount() throws Exception; + long getMessageCount() throws Exception; /** * Returns the number of messages that this queue is currently delivering to its consumers. Modified: trunk/src/main/org/hornetq/api/jms/management/JMSQueueControl.java =================================================================== --- trunk/src/main/org/hornetq/api/jms/management/JMSQueueControl.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/src/main/org/hornetq/api/jms/management/JMSQueueControl.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -103,7 +103,7 @@ * Using {@code null} or an empty filter will count <em>all</em> messages from this queue. */ @Operation(desc = "Returns the number of the messages in the queue matching the given filter", impact = MBeanOperationInfo.INFO) - int countMessages(@Parameter(name = "filter", desc = "A JMS message filter (can be empty)") String filter) throws Exception; + long countMessages(@Parameter(name = "filter", desc = "A JMS message filter (can be empty)") String filter) throws Exception; /** * Removes the message corresponding to the specified message ID. Modified: trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -1768,7 +1768,7 @@ private final boolean durable; - private final int messageCount; + private final long messageCount; private final SimpleString filterString; @@ -1778,7 +1778,7 @@ public QueueQueryImpl(final boolean durable, final int consumerCount, - final int messageCount, + final long messageCount, final SimpleString filterString, final SimpleString address, final boolean exists) @@ -1807,7 +1807,7 @@ return filterString; } - public int getMessageCount() + public long getMessageCount() { return messageCount; } Modified: trunk/src/main/org/hornetq/core/management/impl/QueueControlImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/management/impl/QueueControlImpl.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/src/main/org/hornetq/core/management/impl/QueueControlImpl.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -167,7 +167,7 @@ } } - public int getMessageCount() + public long getMessageCount() { checkStarted(); @@ -434,7 +434,7 @@ } } - public int countMessages(final String filterStr) throws Exception + public long countMessages(final String filterStr) throws Exception { checkStarted(); Modified: trunk/src/main/org/hornetq/core/messagecounter/MessageCounter.java =================================================================== --- trunk/src/main/org/hornetq/core/messagecounter/MessageCounter.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/src/main/org/hornetq/core/messagecounter/MessageCounter.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -66,7 +66,7 @@ private long countTotalLast; - private int depthLast; + private long depthLast; private long timeLastUpdate; @@ -188,7 +188,7 @@ * Gets the current message count of pending messages * within the destination waiting for dispatch */ - public int getMessageCount() + public long getMessageCount() { return destQueue.getMessageCount(); } @@ -197,10 +197,10 @@ * Gets the message count delta of pending messages * since last method call. */ - public int getMessageCountDelta() + public long getMessageCountDelta() { - int current = destQueue.getMessageCount(); - int delta = current - depthLast; + long current = destQueue.getMessageCount(); + int delta = (int)(current - depthLast); depthLast = current; Modified: trunk/src/main/org/hornetq/core/paging/cursor/PageSubscription.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/cursor/PageSubscription.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/src/main/org/hornetq/core/paging/cursor/PageSubscription.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -61,12 +61,12 @@ void ack(PagedReference ref) throws Exception; // for internal (cursor) classes - void ack(PagePosition ref) throws Exception; + void confirmPosition(PagePosition ref) throws Exception; void ackTx(Transaction tx, PagedReference position) throws Exception; // for internal (cursor) classes - void ackTx(Transaction tx, PagePosition position) throws Exception; + void confirmPosition(Transaction tx, PagePosition position) throws Exception; /** * Modified: trunk/src/main/org/hornetq/core/paging/cursor/impl/PageCursorProviderImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/cursor/impl/PageCursorProviderImpl.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/src/main/org/hornetq/core/paging/cursor/impl/PageCursorProviderImpl.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -282,7 +282,7 @@ // 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)); + cursor.confirmPosition(new PagePositionImpl(currentPage.getPageId(), -1)); } storageManager.waitOnOperations(); Modified: trunk/src/main/org/hornetq/core/paging/cursor/impl/PageSubscriptionImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/cursor/impl/PageSubscriptionImpl.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/src/main/org/hornetq/core/paging/cursor/impl/PageSubscriptionImpl.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -168,7 +168,7 @@ cursorInfo.confirmed.addAndGet(position.getMessageNr()); } - ack(position); + confirmPosition(position); } public PageSubscriptionCounter getCounter() @@ -395,7 +395,7 @@ return new PagePositionImpl(pageStore.getFirstPage(), -1); } - public void ackTx(final Transaction tx, final PagePosition position) throws Exception + public void confirmPosition(final Transaction tx, final PagePosition position) throws Exception { // if the cursor is persistent if (persistent) @@ -408,7 +408,9 @@ public void ackTx(final Transaction tx, final PagedReference reference) throws Exception { - ackTx(tx, reference.getPosition()); + confirmPosition(tx, reference.getPosition()); + + counter.increment(tx, -1); PageTransactionInfo txInfo = getPageTransaction(reference); if (txInfo != null) @@ -422,15 +424,13 @@ */ public void ack(final PagedReference reference) throws Exception { - ack(reference.getPosition()); - PageTransactionInfo txInfo = getPageTransaction(reference); - if (txInfo != null) - { - txInfo.storeUpdate(this.store, pageStore.getPagingManager()); - } + // Need to do the ACK and counter atomically (inside a TX) or the counter could get out of sync + Transaction tx = new TransactionImpl(this.store); + ackTx(tx, reference); + tx.commit(); } - public void ack(final PagePosition position) throws Exception + public void confirmPosition(final PagePosition position) throws Exception { // if we are dealing with a persistent cursor if (persistent) Modified: trunk/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -896,7 +896,7 @@ Transaction tx = ctx.getTransaction(); - pagedMessage = new PagedMessageImpl(message, getQueueIDs(listCtx), getTransactionID(tx, listCtx)); + pagedMessage = new PagedMessageImpl(message, routeQueues(tx, listCtx), getTransactionID(tx, listCtx)); int bytesToWrite = pagedMessage.getEncodeSize() + PageImpl.SIZE_RECORD; @@ -917,20 +917,22 @@ } - private long[] getQueueIDs(RouteContextList ctx) + private long[] routeQueues(Transaction tx, RouteContextList ctx) throws Exception { 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) { + q.getPageSubscription().getCounter().increment(tx, 1); ids[i++] = q.getID(); } for (org.hornetq.core.server.Queue q : nonDurableQueues) { + q.getPageSubscription().getCounter().increment(tx, 1); ids[i++] = q.getID(); } return ids; Modified: trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionQueueQueryResponseMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionQueueQueryResponseMessage.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionQueueQueryResponseMessage.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -35,7 +35,7 @@ private int consumerCount; - private int messageCount; + private long messageCount; private SimpleString filterString; @@ -60,7 +60,7 @@ final boolean temporary, final SimpleString filterString, final int consumerCount, - final int messageCount, + final long messageCount, final boolean exists) { super(PacketImpl.SESS_QUEUEQUERY_RESP); @@ -103,7 +103,7 @@ return consumerCount; } - public int getMessageCount() + public long getMessageCount() { return messageCount; } @@ -135,7 +135,7 @@ buffer.writeBoolean(durable); buffer.writeBoolean(temporary); buffer.writeInt(consumerCount); - buffer.writeInt(messageCount); + buffer.writeLong(messageCount); buffer.writeNullableSimpleString(filterString); buffer.writeNullableSimpleString(address); buffer.writeNullableSimpleString(name); @@ -148,7 +148,7 @@ durable = buffer.readBoolean(); temporary = buffer.readBoolean(); consumerCount = buffer.readInt(); - messageCount = buffer.readInt(); + messageCount = buffer.readLong(); filterString = buffer.readNullableSimpleString(); address = buffer.readNullableSimpleString(); name = buffer.readNullableSimpleString(); Modified: trunk/src/main/org/hornetq/core/server/Queue.java =================================================================== --- trunk/src/main/org/hornetq/core/server/Queue.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/src/main/org/hornetq/core/server/Queue.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -72,7 +72,7 @@ void deliverAsync(); - int getMessageCount(); + long getMessageCount(); int getDeliveringCount(); Modified: trunk/src/main/org/hornetq/core/server/QueueQueryResult.java =================================================================== --- trunk/src/main/org/hornetq/core/server/QueueQueryResult.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/src/main/org/hornetq/core/server/QueueQueryResult.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -32,7 +32,7 @@ private int consumerCount; - private int messageCount; + private long messageCount; private SimpleString filterString; @@ -46,7 +46,7 @@ final boolean temporary, final SimpleString filterString, final int consumerCount, - final int messageCount) + final long messageCount) { this(name, address, durable, temporary, filterString, consumerCount, messageCount, true); } @@ -62,7 +62,7 @@ final boolean temporary, final SimpleString filterString, final int consumerCount, - final int messageCount, + final long messageCount, final boolean exists) { this.durable = durable; @@ -97,7 +97,7 @@ return consumerCount; } - public int getMessageCount() + public long getMessageCount() { return messageCount; } Modified: trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -653,13 +653,13 @@ return null; } - public int getMessageCount() + public long getMessageCount() { blockOnExecutorFuture(); synchronized (this) { - return messageReferences.size() + getScheduledCount() + getDeliveringCount(); + return messageReferences.size() + getScheduledCount() + getDeliveringCount() + pageSubscription.getCounter().getValue(); } } Modified: trunk/src/main/org/hornetq/jms/management/impl/JMSQueueControlImpl.java =================================================================== --- trunk/src/main/org/hornetq/jms/management/impl/JMSQueueControlImpl.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/src/main/org/hornetq/jms/management/impl/JMSQueueControlImpl.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -115,7 +115,7 @@ return managedQueue.isTemporary(); } - public int getMessageCount() + public long getMessageCount() { return coreQueueControl.getMessageCount(); } @@ -225,7 +225,7 @@ return JMSQueueControlImpl.toJSON(listMessages(filter)); } - public int countMessages(final String filterStr) throws Exception + public long countMessages(final String filterStr) throws Exception { String filter = JMSQueueControlImpl.createFilterFromJMSSelector(filterStr); return coreQueueControl.countMessages(filter); Modified: trunk/src/main/org/hornetq/jms/management/impl/JMSTopicControlImpl.java =================================================================== --- trunk/src/main/org/hornetq/jms/management/impl/JMSTopicControlImpl.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/src/main/org/hornetq/jms/management/impl/JMSTopicControlImpl.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -113,7 +113,7 @@ return managedTopic.getAddress(); } - public int getMessageCount() + public long getMessageCount() { return getMessageCount(DurabilityType.ALL); } Modified: trunk/tests/jms-tests/src/org/hornetq/jms/tests/HornetQServerTestCase.java =================================================================== --- trunk/tests/jms-tests/src/org/hornetq/jms/tests/HornetQServerTestCase.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/tests/jms-tests/src/org/hornetq/jms/tests/HornetQServerTestCase.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -384,7 +384,7 @@ public boolean checkEmpty(final Queue queue) throws Exception { - Integer messageCount = HornetQServerTestCase.servers.get(0).getMessageCountForQueue(queue.getQueueName()); + Long messageCount = HornetQServerTestCase.servers.get(0).getMessageCountForQueue(queue.getQueueName()); if (messageCount > 0) { removeAllMessages(queue.getQueueName(), true); @@ -409,7 +409,7 @@ protected boolean assertRemainingMessages(final int expected) throws Exception { - Integer messageCount = HornetQServerTestCase.servers.get(0).getMessageCountForQueue("Queue1"); + Long messageCount = HornetQServerTestCase.servers.get(0).getMessageCountForQueue("Queue1"); ProxyAssertSupport.assertEquals(expected, messageCount.intValue()); return expected == messageCount.intValue(); @@ -501,7 +501,7 @@ return HornetQServerTestCase.servers.get(0).listAllSubscribersForTopic(s); } - protected Integer getMessageCountForQueue(final String s) throws Exception + protected Long getMessageCountForQueue(final String s) throws Exception { return HornetQServerTestCase.servers.get(0).getMessageCountForQueue(s); } Modified: trunk/tests/jms-tests/src/org/hornetq/jms/tests/TransactedSessionTest.java =================================================================== --- trunk/tests/jms-tests/src/org/hornetq/jms/tests/TransactedSessionTest.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/tests/jms-tests/src/org/hornetq/jms/tests/TransactedSessionTest.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -73,7 +73,7 @@ conn.close(); - Integer i = getMessageCountForQueue("Queue1"); + Long i = getMessageCountForQueue("Queue1"); ProxyAssertSupport.assertEquals(1, i.intValue()); } Modified: trunk/tests/jms-tests/src/org/hornetq/jms/tests/tools/container/LocalTestServer.java =================================================================== --- trunk/tests/jms-tests/src/org/hornetq/jms/tests/tools/container/LocalTestServer.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/tests/jms-tests/src/org/hornetq/jms/tests/tools/container/LocalTestServer.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -396,7 +396,7 @@ } - public Integer getMessageCountForQueue(final String queueName) throws Exception + public Long getMessageCountForQueue(final String queueName) throws Exception { JMSQueueControl queue = (JMSQueueControl)getHornetQServer().getManagementService() .getResource(ResourceNames.JMS_QUEUE + queueName); @@ -406,7 +406,7 @@ } else { - return -1; + return -1l; } } Modified: trunk/tests/jms-tests/src/org/hornetq/jms/tests/tools/container/Server.java =================================================================== --- trunk/tests/jms-tests/src/org/hornetq/jms/tests/tools/container/Server.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/tests/jms-tests/src/org/hornetq/jms/tests/tools/container/Server.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -176,7 +176,7 @@ void removeAllMessages(String destination, boolean isQueue) throws Exception; - Integer getMessageCountForQueue(String queueName) throws Exception; + Long getMessageCountForQueue(String queueName) throws Exception; List<String> listAllSubscribersForTopic(String s) throws Exception; Modified: trunk/tests/src/org/hornetq/tests/integration/jms/bridge/BridgeTestBase.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/bridge/BridgeTestBase.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/tests/src/org/hornetq/tests/integration/jms/bridge/BridgeTestBase.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -522,7 +522,7 @@ } JMSQueueControl queueControl = (JMSQueueControl)managementService.getResource(ResourceNames.JMS_QUEUE + queue.getQueueName()); - Integer messageCount = queueControl.getMessageCount(); + Long messageCount = queueControl.getMessageCount(); if (messageCount > 0) { Modified: trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSQueueControlUsingJMSTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSQueueControlUsingJMSTest.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSQueueControlUsingJMSTest.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -98,9 +98,9 @@ return (Integer)proxy.invokeOperation("changeMessagesPriority", filter, newPriority); } - public int countMessages(final String filter) throws Exception + public long countMessages(final String filter) throws Exception { - return (Integer)proxy.invokeOperation("countMessages", filter); + return (Long)proxy.invokeOperation("countMessages", filter); } public boolean expireMessage(final String messageID) throws Exception @@ -133,9 +133,9 @@ return (String)proxy.retrieveAttributeValue("expiryAddress"); } - public int getMessageCount() + public long getMessageCount() { - return (Integer)proxy.retrieveAttributeValue("messageCount"); + return (Long)proxy.retrieveAttributeValue("messageCount"); } public long getMessagesAdded() Modified: trunk/tests/src/org/hornetq/tests/integration/management/QueueControlUsingCoreTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/management/QueueControlUsingCoreTest.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/tests/src/org/hornetq/tests/integration/management/QueueControlUsingCoreTest.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -69,9 +69,9 @@ return (Integer)proxy.invokeOperation("changeMessagesPriority", filter, newPriority); } - public int countMessages(final String filter) throws Exception + public long countMessages(final String filter) throws Exception { - return (Integer)proxy.invokeOperation("countMessages", filter); + return (Long)proxy.invokeOperation("countMessages", filter); } public boolean expireMessage(final long messageID) throws Exception @@ -114,9 +114,9 @@ return (String)proxy.retrieveAttributeValue("filter"); } - public int getMessageCount() + public long getMessageCount() { - return (Integer)proxy.retrieveAttributeValue("messageCount"); + return (Long)proxy.retrieveAttributeValue("messageCount"); } public long getMessagesAdded() Modified: trunk/tests/src/org/hornetq/tests/stress/paging/PageCursorStressTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/stress/paging/PageCursorStressTest.java 2010-12-16 23:20:10 UTC (rev 10048) +++ trunk/tests/src/org/hornetq/tests/stress/paging/PageCursorStressTest.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -134,7 +134,7 @@ while ((msg = iterator.next()) != null) { assertEquals(key++, msg.getMessage().getIntProperty("key").intValue()); - cursor.ack(msg.getPosition()); + cursor.confirmPosition(msg.getPosition()); } assertEquals(NUM_MESSAGES, key); @@ -225,7 +225,7 @@ assertEquals(key, msg.getMessage().getIntProperty("key").intValue()); assertTrue(msg.getMessage().getBooleanProperty("even").booleanValue()); key += 2; - cursorEven.ack(msg.getPosition()); + cursorEven.confirmPosition(msg.getPosition()); } assertEquals(NUM_MESSAGES, key); @@ -235,7 +235,7 @@ assertEquals(key, msg.getMessage().getIntProperty("key").intValue()); assertFalse(msg.getMessage().getBooleanProperty("even").booleanValue()); key += 2; - cursorOdd.ack(msg.getPosition()); + cursorOdd.confirmPosition(msg.getPosition()); } assertEquals(NUM_MESSAGES + 1, key); 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-12-16 23:20:10 UTC (rev 10048) +++ trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/FakeQueue.java 2010-12-17 02:20:26 UTC (rev 10049) @@ -334,7 +334,7 @@ /* (non-Javadoc) * @see org.hornetq.core.server.Queue#getMessageCount() */ - public int getMessageCount() + public long getMessageCount() { // TODO Auto-generated method stub return 0;