3:43 a.m.
Author: clebert.suconic(a)jboss.com
Date: 2010-11-12 21:43:47 -0500 (Fri, 12 Nov 2010)
New Revision: 9884
Modified:
branches/Branch_New_Paging/src/main/org/hornetq/core/paging/cursor/impl/PageSubscriptionImpl.java
branches/Branch_New_Paging/src/main/org/hornetq/core/paging/impl/PageTransactionInfoImpl.java
branches/Branch_New_Paging/src/main/org/hornetq/core/server/impl/QueueImpl.java
Log:
tweaks
Modified:
branches/Branch_New_Paging/src/main/org/hornetq/core/paging/cursor/impl/PageSubscriptionImpl.java
===================================================================
---
branches/Branch_New_Paging/src/main/org/hornetq/core/paging/cursor/impl/PageSubscriptionImpl.java 2010-11-12
20:24:00 UTC (rev 9883)
+++
branches/Branch_New_Paging/src/main/org/hornetq/core/paging/cursor/impl/PageSubscriptionImpl.java 2010-11-13
02:43:47 UTC (rev 9884)
@@ -209,8 +209,8 @@
{
if (entry.getKey() == lastAckedPosition.getPageNr())
{
- PageSubscriptionImpl.trace("We can't clear page " +
entry.getKey() +
- " now since it's the current
page");
+ // PageSubscriptionImpl.trace("We can't clear page " +
entry.getKey() +
+ // " now since it's the current page");
}
else
{
@@ -1030,115 +1030,120 @@
/* (non-Javadoc)
* @see org.hornetq.core.paging.cursor.PageCursor#moveNext()
*/
- public synchronized PagedReference moveNext() throws Exception
+ public PagedReference moveNext() throws Exception
{
- boolean match = false;
+ synchronized (PageSubscriptionImpl.this)
+ {
+ boolean match = false;
- PagedReference message = null;
+ PagedReference message = null;
- PagePosition lastPosition = position;
- PagePosition tmpPosition = position;
+ PagePosition lastPosition = position;
+ PagePosition tmpPosition = position;
- do
- {
- synchronized (redeliveries)
+ do
{
- if (redeliveryIterator.hasNext())
+ synchronized (redeliveries)
{
- // There's a redelivery pending, we will get it out of that pool
instead
- isredelivery = true;
- return getReference(redeliveryIterator.next());
+ if (redeliveryIterator.hasNext())
+ {
+ // There's a redelivery pending, we will get it out of that pool
instead
+ isredelivery = true;
+ PagedReference redeliveredMsg =
getReference(redeliveryIterator.next());
+
+ return redeliveredMsg;
+ }
+ else
+ {
+ isredelivery = false;
+ }
+
+ message = internalGetNext(tmpPosition);
}
- else
+
+ if (message == null)
{
- isredelivery = false;
+ break;
}
- message = internalGetNext(tmpPosition);
- }
+ tmpPosition = message.getPosition();
- if (message == null)
- {
- break;
- }
+ boolean valid = true;
+ boolean ignored = false;
- tmpPosition = message.getPosition();
+ // Validate the scenarios where the message should be considered not valid
even to be considered
- boolean valid = true;
- boolean ignored = false;
+ // 1st... is it routed?
- // Validate the scenarios where the message should be considered not valid
even to be considered
+ valid = routed(message.getPagedMessage());
+ if (!valid)
+ ignored = true;
- // 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)
+ // 2nd ... if TX, is it committed?
+ if (valid && message.getPagedMessage().getTransactionID() != 0)
{
- log.warn("Couldn't locate page transaction " +
message.getPagedMessage().getTransactionID() +
- ", ignoring message on position " +
- message.getPosition());
- valid = false;
- ignored = true;
+ 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;
+ }
+ }
}
- else
+
+ // 3rd... was it previously removed?
+ if (valid)
{
- if (tx.deliverAfterCommit(PageSubscriptionImpl.this,
message.getPosition()))
+ // 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;
- 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()))
+ if (!ignored)
{
- valid = false;
+ position = message.getPosition();
}
- }
- if (!ignored)
- {
- position = message.getPosition();
- }
+ if (valid)
+ {
+ match = match(message.getMessage());
- if (valid)
- {
- match = match(message.getMessage());
-
- if (!match)
+ if (!match)
+ {
+ processACK(message.getPosition());
+ }
+ }
+ else if (ignored)
{
- processACK(message.getPosition());
+ positionIgnored(message.getPosition());
}
}
- else if (ignored)
+ while (message != null && !match);
+
+ if (message != null)
{
- positionIgnored(message.getPosition());
+ lastOperation = lastPosition;
}
- }
- while (message != null && !match);
- if (message != null)
- {
- lastOperation = lastPosition;
+ return message;
}
-
- return message;
}
/** QueueImpl::deliver could be calling hasNext while QueueImpl.depage could be
using next and hasNext as well.
Modified:
branches/Branch_New_Paging/src/main/org/hornetq/core/paging/impl/PageTransactionInfoImpl.java
===================================================================
---
branches/Branch_New_Paging/src/main/org/hornetq/core/paging/impl/PageTransactionInfoImpl.java 2010-11-12
20:24:00 UTC (rev 9883)
+++
branches/Branch_New_Paging/src/main/org/hornetq/core/paging/impl/PageTransactionInfoImpl.java 2010-11-13
02:43:47 UTC (rev 9884)
@@ -51,6 +51,8 @@
private volatile long recordID = -1;
private volatile boolean committed = false;
+
+ private volatile boolean useRedelivery = false;
private volatile boolean rolledback = false;
@@ -237,6 +239,12 @@
*/
public synchronized boolean deliverAfterCommit(PageSubscription cursor, PagePosition
cursorPos)
{
+ if (committed && useRedelivery)
+ {
+ cursor.redeliver(cursorPos);
+ return true;
+ }
+ else
if (committed)
{
return false;
@@ -249,6 +257,7 @@
}
else
{
+ useRedelivery = true;
if (lateDeliveries == null)
{
lateDeliveries = new LinkedList<Pair<PageSubscription,
PagePosition>>();
Modified: branches/Branch_New_Paging/src/main/org/hornetq/core/server/impl/QueueImpl.java
===================================================================
---
branches/Branch_New_Paging/src/main/org/hornetq/core/server/impl/QueueImpl.java 2010-11-12
20:24:00 UTC (rev 9883)
+++
branches/Branch_New_Paging/src/main/org/hornetq/core/server/impl/QueueImpl.java 2010-11-13
02:43:47 UTC (rev 9884)
@@ -1273,7 +1273,7 @@
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());
+ //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())
@@ -1283,12 +1283,12 @@
pageIterator.remove();
}
- System.out.println("Depaged " + nmessages);
+ //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());
- }
+// else
+// {
+// System.out.println("Depaging not being done now.. there are msgRef =
" + messageReferences.size() + " scheduled = " + getScheduledCount() +
" concurrentQueue.size() = " + concurrentQueue.size());
+// }
deliverAsync();
}