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

check messageId when removing ConsumerMessageRef (pesky redo logs can add/delete things - breaking the assumption we always get a nice ordered behaviour for durable subs) 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@515630 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Robert Davies 2007-03-07 16:15:08 +00:00
parent 0d9b413256
commit b3bbb9c2ea
5 changed files with 51 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/ConsumerMessageRef.java
View File

@ -14,6 +14,7 @@
package org.apache.activemq.store.kahadaptor; package org.apache.activemq.store.kahadaptor;
import org.apache.activemq.command.MessageId;
import org.apache.activemq.kaha.StoreEntry; import org.apache.activemq.kaha.StoreEntry;
/** /**
@ -23,6 +24,7 @@ import org.apache.activemq.kaha.StoreEntry;
*/ */
public class ConsumerMessageRef{ public class ConsumerMessageRef{
private MessageId messageId;
private StoreEntry messageEntry; private StoreEntry messageEntry;
private StoreEntry ackEntry; private StoreEntry ackEntry;
@ -54,5 +56,21 @@ public class ConsumerMessageRef{
this.messageEntry=messageEntry; this.messageEntry=messageEntry;
} }
/**
* @return the messageId
*/
public MessageId getMessageId(){
return this.messageId;
}
/**
* @param messageId the messageId to set
*/
public void setMessageId(MessageId messageId){
this.messageId=messageId;
}
} }

4
activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/ConsumerMessageRefMarshaller.java
View File

@ -20,6 +20,7 @@ package org.apache.activemq.store.kahadaptor;
import java.io.DataInput; import java.io.DataInput;
import java.io.DataOutput; import java.io.DataOutput;
import java.io.IOException; import java.io.IOException;
import org.apache.activemq.command.MessageId;
import org.apache.activemq.kaha.Marshaller; import org.apache.activemq.kaha.Marshaller;
import org.apache.activemq.kaha.impl.index.IndexItem; import org.apache.activemq.kaha.impl.index.IndexItem;
@ -39,6 +40,7 @@ public class ConsumerMessageRefMarshaller implements Marshaller{
*/ */
public void writePayload(Object object,DataOutput dataOut) throws IOException{ public void writePayload(Object object,DataOutput dataOut) throws IOException{
ConsumerMessageRef ref = (ConsumerMessageRef) object; ConsumerMessageRef ref = (ConsumerMessageRef) object;
dataOut.writeUTF(ref.getMessageId().toString());
IndexItem item = (IndexItem)ref.getMessageEntry(); IndexItem item = (IndexItem)ref.getMessageEntry();
dataOut.writeLong(item.getOffset()); dataOut.writeLong(item.getOffset());
item.write(dataOut); item.write(dataOut);
@ -46,6 +48,7 @@ public class ConsumerMessageRefMarshaller implements Marshaller{
dataOut.writeLong(item.getOffset()); dataOut.writeLong(item.getOffset());
item.write(dataOut); item.write(dataOut);
} }
/** /**
@ -56,6 +59,7 @@ public class ConsumerMessageRefMarshaller implements Marshaller{
*/ */
public Object readPayload(DataInput dataIn) throws IOException{ public Object readPayload(DataInput dataIn) throws IOException{
ConsumerMessageRef ref = new ConsumerMessageRef(); ConsumerMessageRef ref = new ConsumerMessageRef();
ref.setMessageId(new MessageId(dataIn.readUTF()));
IndexItem item = new IndexItem(); IndexItem item = new IndexItem();
item.setOffset(dataIn.readLong()); item.setOffset(dataIn.readLong());
item.read(dataIn); item.read(dataIn);

6
activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaTopicMessageStore.java
View File

@ -58,7 +58,8 @@ public class KahaTopicMessageStore extends KahaMessageStore implements TopicMess
public synchronized void addMessage(ConnectionContext context,Message message) throws IOException{ public synchronized void addMessage(ConnectionContext context,Message message) throws IOException{
int subscriberCount=subscriberMessages.size(); int subscriberCount=subscriberMessages.size();
if(subscriberCount>0){ if(subscriberCount>0){
StoreEntry messageEntry=messageContainer.place(message.getMessageId(),message); MessageId id = message.getMessageId();
StoreEntry messageEntry=messageContainer.place(id,message);
TopicSubAck tsa=new TopicSubAck(); TopicSubAck tsa=new TopicSubAck();
tsa.setCount(subscriberCount); tsa.setCount(subscriberCount);
tsa.setMessageEntry(messageEntry); tsa.setMessageEntry(messageEntry);
@ -68,6 +69,7 @@ public class KahaTopicMessageStore extends KahaMessageStore implements TopicMess
ConsumerMessageRef ref=new ConsumerMessageRef(); ConsumerMessageRef ref=new ConsumerMessageRef();
ref.setAckEntry(ackEntry); ref.setAckEntry(ackEntry);
ref.setMessageEntry(messageEntry); ref.setMessageEntry(messageEntry);
ref.setMessageId(id);
container.add(ref); container.add(ref);
} }
} }
@ -78,7 +80,7 @@ public class KahaTopicMessageStore extends KahaMessageStore implements TopicMess
String subcriberId=getSubscriptionKey(clientId,subscriptionName); String subcriberId=getSubscriptionKey(clientId,subscriptionName);
TopicSubContainer container=(TopicSubContainer)subscriberMessages.get(subcriberId); TopicSubContainer container=(TopicSubContainer)subscriberMessages.get(subcriberId);
if(container!=null){ if(container!=null){
ConsumerMessageRef ref=container.remove(); ConsumerMessageRef ref=container.remove(messageId);
if(container.isEmpty()){ if(container.isEmpty()){
container.reset(); container.reset();
} }

33
activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaTopicReferenceStore.java
View File

@ -68,30 +68,29 @@ public class KahaTopicReferenceStore extends KahaReferenceStore implements Topic
listener.recoverMessageReference(new MessageId(record.getMessageId())); listener.recoverMessageReference(new MessageId(record.getMessageId()));
} }
public void addMessageReference(ConnectionContext context,MessageId messageId,ReferenceData data) public void addMessageReference(final ConnectionContext context,final MessageId messageId,final ReferenceData data){
throws IOException{ final ReferenceRecord record=new ReferenceRecord(messageId.toString(),data);
ReferenceRecord record=new ReferenceRecord(messageId.toString(),data); final int subscriberCount=subscriberMessages.size();
int subscriberCount=subscriberMessages.size();
if(subscriberCount>0){ if(subscriberCount>0){
StoreEntry messageEntry=messageContainer.place(messageId,record); final StoreEntry messageEntry=messageContainer.place(messageId,record);
addInterest(record); addInterest(record);
TopicSubAck tsa=new TopicSubAck(); final TopicSubAck tsa=new TopicSubAck();
tsa.setCount(subscriberCount); tsa.setCount(subscriberCount);
tsa.setMessageEntry(messageEntry); tsa.setMessageEntry(messageEntry);
StoreEntry ackEntry=ackContainer.placeLast(tsa); final StoreEntry ackEntry=ackContainer.placeLast(tsa);
for(Iterator i=subscriberMessages.values().iterator();i.hasNext();){ for(final Iterator i=subscriberMessages.values().iterator();i.hasNext();){
TopicSubContainer container=(TopicSubContainer)i.next(); final TopicSubContainer container=(TopicSubContainer)i.next();
ConsumerMessageRef ref=new ConsumerMessageRef(); final ConsumerMessageRef ref=new ConsumerMessageRef();
ref.setAckEntry(ackEntry); ref.setAckEntry(ackEntry);
ref.setMessageEntry(messageEntry); ref.setMessageEntry(messageEntry);
StoreEntry listEntry = container.add(ref); ref.setMessageId(messageId);
container.add(ref);
} }
} }
} }
public ReferenceData getMessageReference(MessageId identity) throws IOException{ public ReferenceData getMessageReference(final MessageId identity) throws IOException{
ReferenceRecord result=messageContainer.get(identity); final ReferenceRecord result=messageContainer.get(identity);
if(result==null) if(result==null)
return null; return null;
return result.getData(); return result.getData();
@ -119,12 +118,10 @@ public class KahaTopicReferenceStore extends KahaReferenceStore implements Topic
public synchronized void acknowledge(ConnectionContext context,String clientId,String subscriptionName, public synchronized void acknowledge(ConnectionContext context,String clientId,String subscriptionName,
MessageId messageId) throws IOException{ MessageId messageId) throws IOException{
String key=getSubscriptionKey(clientId,subscriptionName); String key=getSubscriptionKey(clientId,subscriptionName);
TopicSubContainer container=(TopicSubContainer)subscriberMessages.get(key); TopicSubContainer container=(TopicSubContainer)subscriberMessages.get(key);
if(container!=null){ if(container!=null){
ConsumerMessageRef ref=container.remove(); ConsumerMessageRef ref=container.remove(messageId);
if(container.isEmpty()){
container.reset();
}
if(ref!=null){ if(ref!=null){
TopicSubAck tsa=(TopicSubAck)ackContainer.get(ref.getAckEntry()); TopicSubAck tsa=(TopicSubAck)ackContainer.get(ref.getAckEntry());
if(tsa!=null){ if(tsa!=null){

16
activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/TopicSubContainer.java
View File

@ -15,6 +15,7 @@
package org.apache.activemq.store.kahadaptor; package org.apache.activemq.store.kahadaptor;
import java.util.Iterator; import java.util.Iterator;
import org.apache.activemq.command.MessageId;
import org.apache.activemq.kaha.ListContainer; import org.apache.activemq.kaha.ListContainer;
import org.apache.activemq.kaha.StoreEntry; import org.apache.activemq.kaha.StoreEntry;
@ -58,14 +59,17 @@ import org.apache.activemq.kaha.StoreEntry;
return listContainer.placeLast(ref); return listContainer.placeLast(ref);
} }
public ConsumerMessageRef remove(){ public ConsumerMessageRef remove(MessageId id){
ConsumerMessageRef result=null; ConsumerMessageRef result=null;
if(!listContainer.isEmpty()){ if(!listContainer.isEmpty()){
StoreEntry entry=listContainer.getFirst(); for(StoreEntry entry=listContainer.getFirst();entry!=null;entry=listContainer.getNext(entry)){
if(entry!=null){ ConsumerMessageRef ref=(ConsumerMessageRef)listContainer.get(entry);
result=(ConsumerMessageRef)listContainer.removeFirst(); if(ref!=null&&ref.getMessageId().equals(id)){
if(listContainer.isEmpty()||(batchEntry!=null&&batchEntry.equals(entry))){ listContainer.remove(entry);
reset(); result=ref;
if(listContainer.isEmpty()||batchEntry.equals(entry)){
reset();
}
} }
} }
} }