Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

related to AMQ-4296 : Fixes leveldb store cursoring. It was recovering too many messages and sometimes not the right messages. 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1485810 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Hiram R. Chirino 2013-05-23 18:33:06 +00:00
parent 5b0749c4ba
commit 148909357f
2 changed files with 23 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala
View File

@ -667,17 +667,24 @@ class DBManager(val parent:LevelDBStore) {
} }
def cursorMessages(key:Long, listener:MessageRecoveryListener, startPos:Long) = { def cursorMessages(key:Long, listener:MessageRecoveryListener, startPos:Long) = {
var nextPos = startPos; var lastmsgid:MessageId = null
client.queueCursor(key, nextPos) { msg => client.queueCursor(key, startPos) { msg =>
if( listener.hasSpace ) { if( listener.hasSpace ) {
listener.recoverMessage(msg) if( listener.recoverMessage(msg) ) {
nextPos += 1 lastmsgid = msg.getMessageId
true true
} else { } else {
false false
} }
} else {
false
}
}
if( lastmsgid==null ) {
startPos
} else {
lastmsgid.getEntryLocator.asInstanceOf[EntryLocator].seq+1
} }
nextPos
} }
def getXAActions(key:Long) = { def getXAActions(key:Long) = {

17
activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala
View File

@ -678,33 +678,36 @@ class LevelDBStore extends LockableServiceSupport with BrokerServiceAware with P
} }
def recover(listener: MessageRecoveryListener): Unit = { def recover(listener: MessageRecoveryListener): Unit = {
cursorPosition = db.cursorMessages(key, preparedExcluding(listener), 0) cursorPosition = db.cursorMessages(key, PreparedExcluding(listener), 0)
} }
def preparedExcluding(listener: MessageRecoveryListener) = new MessageRecoveryListener { case class PreparedExcluding(listener: MessageRecoveryListener) extends MessageRecoveryListener {
def isDuplicate(ref: MessageId) = listener.isDuplicate(ref) def isDuplicate(ref: MessageId) = listener.isDuplicate(ref)
def hasSpace = listener.hasSpace def hasSpace = listener.hasSpace
def recoverMessageReference(ref: MessageId) = { def recoverMessageReference(ref: MessageId) = {
if (!preparedAcks.contains(ref)) { if (!preparedAcks.contains(ref)) {
listener.recoverMessageReference(ref) listener.recoverMessageReference(ref)
} } else {
true true
} }
}
def recoverMessage(message: Message) = { def recoverMessage(message: Message) = {
if (!preparedAcks.contains(message.getMessageId)) { if (!preparedAcks.contains(message.getMessageId)) {
listener.recoverMessage(message) listener.recoverMessage(message)
} } else {
true true
} }
} }
}
def resetBatching: Unit = { def resetBatching: Unit = {
cursorPosition = 0 cursorPosition = 0
} }
def recoverNextMessages(maxReturned: Int, listener: MessageRecoveryListener): Unit = { def recoverNextMessages(maxReturned: Int, listener: MessageRecoveryListener): Unit = {
cursorPosition = db.cursorMessages(key, preparedExcluding(LimitingRecoveryListener(maxReturned, listener)), cursorPosition) val excluding = PreparedExcluding(LimitingRecoveryListener(maxReturned, listener))
cursorPosition = db.cursorMessages(key, excluding, cursorPosition)
} }
override def setBatch(id: MessageId): Unit = { override def setBatch(id: MessageId): Unit = {
@ -714,7 +717,7 @@ class LevelDBStore extends LockableServiceSupport with BrokerServiceAware with P
} }
case class LimitingRecoveryListener(max: Int, listener: MessageRecoveryListener) extends MessageRecoveryListener { case class LimitingRecoveryListener(max: Int, listener: MessageRecoveryListener) extends MessageRecoveryListener {
private var recovered: Int = 0 var recovered: Int = 0
def hasSpace = recovered < max def hasSpace = recovered < max
def recoverMessage(message: Message) = { def recoverMessage(message: Message) = {
recovered += 1; recovered += 1;
@ -849,7 +852,7 @@ class LevelDBStore extends LockableServiceSupport with BrokerServiceAware with P
def recoverNextMessages(clientId: String, subscriptionName: String, maxReturned: Int, listener: MessageRecoveryListener): Unit = { def recoverNextMessages(clientId: String, subscriptionName: String, maxReturned: Int, listener: MessageRecoveryListener): Unit = {
lookup(clientId, subscriptionName).foreach { sub => lookup(clientId, subscriptionName).foreach { sub =>
sub.cursorPosition = db.cursorMessages(key, preparedExcluding(LimitingRecoveryListener(maxReturned, listener)), sub.cursorPosition.max(sub.lastAckPosition+1)) sub.cursorPosition = db.cursorMessages(key, PreparedExcluding(LimitingRecoveryListener(maxReturned, listener)), sub.cursorPosition.max(sub.lastAckPosition+1))
} }
} }