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

store specific tests for setBatch http://issues.apache.org/activemq/browse/AMQ-2020 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@741061 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Gary Tully 2009-02-05 10:06:39 +00:00
parent 37c2a955a2
commit 9bb46820b2
11 changed files with 271 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/AbstractStoreCursor.java
View File

@ -158,10 +158,10 @@ public abstract class AbstractStoreCursor extends AbstractPendingMessageCursor i
lastCachedId = node.getMessageId();
} else {
if (cacheEnabled) {
cacheEnabled=false;
// sync with store on disabling the cache
setBatch(lastCachedId);
}
cacheEnabled=false;
}
size++;
}

5
activemq-core/src/main/java/org/apache/activemq/store/jdbc/JDBCMessageStore.java
View File

@ -249,4 +249,9 @@ public class JDBCMessageStore extends AbstractMessageStore {
}
@Override
public void setBatch(MessageId messageId) {
lastMessageId.set(messageId.getBrokerSequenceId());
}
}

2
activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/DefaultJDBCAdapter.java
View File

@ -720,7 +720,7 @@ public class DefaultJDBCAdapter implements JDBCAdapter {
s = c.getConnection().prepareStatement(this.statements.getFindNextMessagesStatement());
s.setMaxRows(maxReturned * 2);
s.setString(1, destination.getQualifiedName());
s.setLong(2, nextSeq - maxReturned);
s.setLong(2, nextSeq);
rs = s.executeQuery();
int count = 0;
if (this.statements.isUseExternalMessageReferences()) {

6
activemq-core/src/main/java/org/apache/activemq/store/journal/JournalMessageStore.java
View File

@ -411,4 +411,10 @@ public class JournalMessageStore extends AbstractMessageStore {
}
@Override
public void setBatch(MessageId messageId) {
peristenceAdapter.checkpoint(true, true);
longTermStore.setBatch(messageId);
}
}

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

@ -175,4 +175,10 @@ public class KahaMessageStore extends AbstractMessageStore {
public boolean isSupportForCursors() {
return true;
}
@Override
public void setBatch(MessageId messageId) {
batchEntry = messageContainer.getEntry(messageId);
}
}

5
activemq-core/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java
View File

@ -243,6 +243,11 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter {
cursorPos=0;
}
@Override
public void setBatch(MessageId messageId) {
}
public void setMemoryUsage(MemoryUsage memoeyUSage) {
}
public void start() throws Exception {

6
activemq-core/src/main/java/org/apache/activemq/store/memory/MemoryMessageStore.java
View File

@ -150,4 +150,10 @@ public class MemoryMessageStore extends AbstractMessageStore {
public void resetBatching() {
lastBatchId = null;
}
@Override
public void setBatch(MessageId messageId) {
lastBatchId = messageId;
}
}

37
activemq-core/src/test/java/org/apache/activemq/broker/region/cursors/StoreQueueCursorJDBCNoDuplicateTest.java Normal file
View File

@ -0,0 +1,37 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF 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.apache.activemq.broker.region.cursors;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.store.PersistenceAdapter;
import org.apache.activemq.store.jdbc.JDBCPersistenceAdapter;
/**
* @author gtully
* @see https://issues.apache.org/activemq/browse/AMQ-2020
**/
public class StoreQueueCursorJDBCNoDuplicateTest extends StoreQueueCursorNoDuplicateTest {
protected BrokerService createBroker() throws Exception {
BrokerService broker = super.createBroker();
PersistenceAdapter persistenceAdapter = new JDBCPersistenceAdapter();
broker.setPersistenceAdapter(persistenceAdapter);
return broker;
}
}

48
activemq-core/src/test/java/org/apache/activemq/broker/region/cursors/StoreQueueCursorJournalNoDuplicateTest.java Normal file
View File

@ -0,0 +1,48 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF 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.apache.activemq.broker.region.cursors;
import java.io.File;
import org.apache.activeio.journal.active.JournalImpl;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.store.journal.JournalPersistenceAdapter;
import org.apache.activemq.store.kahadaptor.KahaPersistenceAdapter;
/**
* @author gtully
* @see https://issues.apache.org/activemq/browse/AMQ-2020
**/
public class StoreQueueCursorJournalNoDuplicateTest extends StoreQueueCursorNoDuplicateTest {
@Override
protected BrokerService createBroker() throws Exception {
BrokerService broker = super.createBroker();
File dataFileDir = new File("target/activemq-data/StoreQueueCursorJournalNoDuplicateTest");
File journalDir = new File(dataFileDir, "journal").getCanonicalFile();
JournalImpl journal = new JournalImpl(journalDir, 3, 1024 * 1024 * 20);
KahaPersistenceAdapter kahaAdaptor = new KahaPersistenceAdapter();
kahaAdaptor.setDirectory(dataFileDir);
JournalPersistenceAdapter journalAdaptor = new JournalPersistenceAdapter(journal, kahaAdaptor, broker.getTaskRunnerFactory());
journalAdaptor.setMaxCheckpointWorkers(1);
broker.setPersistenceAdapter(journalAdaptor);
return broker;
}
}

33
activemq-core/src/test/java/org/apache/activemq/broker/region/cursors/StoreQueueCursorMemoryNoDuplicateTest.java Normal file
View File

@ -0,0 +1,33 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF 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.apache.activemq.broker.region.cursors;
import org.apache.activemq.broker.BrokerService;
/**
* @author gtully
* @see https://issues.apache.org/activemq/browse/AMQ-2020
**/
public class StoreQueueCursorMemoryNoDuplicateTest extends StoreQueueCursorNoDuplicateTest {
protected BrokerService createBroker() throws Exception {
BrokerService broker = super.createBroker();
broker.setPersistent(false);
return broker;
}
}

123
activemq-core/src/test/java/org/apache/activemq/broker/region/cursors/StoreQueueCursorNoDuplicateTest.java Normal file
View File

@ -0,0 +1,123 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF 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.apache.activemq.broker.region.cursors;
import junit.framework.TestCase;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.DestinationStatistics;
import org.apache.activemq.broker.region.MessageReference;
import org.apache.activemq.broker.region.Queue;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTextMessage;
import org.apache.activemq.command.ConsumerInfo;
import org.apache.activemq.command.MessageId;
import org.apache.activemq.store.MessageStore;
import org.apache.activemq.store.PersistenceAdapter;
import org.apache.activemq.usage.SystemUsage;
/**
* @author gtully
* @see https://issues.apache.org/activemq/browse/AMQ-2020
**/
public class StoreQueueCursorNoDuplicateTest extends TestCase {
ActiveMQQueue destination = new ActiveMQQueue("queue-"
+ StoreQueueCursorNoDuplicateTest.class.getSimpleName());
BrokerService brokerService;
final static String mesageIdRoot = "11111:22222:";
final int messageBytesSize = 1024;
final String text = new String(new byte[messageBytesSize]);
protected int count = 6;
public void setUp() throws Exception {
brokerService = createBroker();
brokerService.setUseJmx(false);
brokerService.deleteAllMessages();
brokerService.start();
}
protected BrokerService createBroker() throws Exception {
return new BrokerService();
}
public void tearDown() throws Exception {
brokerService.stop();
}
public void testNoDuplicateAfterCacheFullAndReadPast() throws Exception {
final PersistenceAdapter persistenceAdapter = brokerService
.getPersistenceAdapter();
final MessageStore queueMessageStore = persistenceAdapter
.createQueueMessageStore(destination);
final ConsumerInfo consumerInfo = new ConsumerInfo();
final DestinationStatistics destinationStatistics = new DestinationStatistics();
consumerInfo.setExclusive(true);
final Queue queue = new Queue(brokerService, destination,
queueMessageStore, destinationStatistics, null);
queueMessageStore.start();
QueueStorePrefetch underTest = new QueueStorePrefetch(queue);
SystemUsage systemUsage = new SystemUsage();
// ensure memory limit is reached
systemUsage.getMemoryUsage().setLimit(messageBytesSize * (count + 2));
underTest.setSystemUsage(systemUsage);
underTest.setEnableAudit(false);
underTest.start();
final ConnectionContext contextNotInTx = new ConnectionContext();
for (int i = 0; i < count; i++) {
ActiveMQTextMessage msg = getMessage(i);
msg.setMemoryUsage(systemUsage.getMemoryUsage());
queueMessageStore.addMessage(contextNotInTx, msg);
underTest.addMessageLast(msg);
}
int dequeueCount = 0;
underTest.setMaxBatchSize(2);
underTest.reset();
while (underTest.hasNext() && dequeueCount < count) {
MessageReference ref = underTest.next();
underTest.remove();
assertEquals(dequeueCount++, ref.getMessageId()
.getProducerSequenceId());
}
underTest.release();
assertEquals(count, dequeueCount);
}
private ActiveMQTextMessage getMessage(int i) throws Exception {
ActiveMQTextMessage message = new ActiveMQTextMessage();
MessageId id = new MessageId(mesageIdRoot + i);
id.setBrokerSequenceId(i);
id.setProducerSequenceId(i);
message.setMessageId(id);
message.setDestination(destination);
message.setPersistent(true);
message.setResponseRequired(true);
message.setText("Msg:" + i + " " + text);
assertEquals(message.getMessageId().getProducerSequenceId(), i);
return message;
}
}