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

Patch applied from Vadim: https://issues.apache.org/activemq/browse/AMQ-855 

git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@439804 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Hiram R. Chirino 2006-09-03 17:09:51 +00:00
parent 7d1e6bcdee
commit a7891c3dca
3 changed files with 32 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
activemq-core/src/main/java/org/apache/activemq/broker/region/EndOfBrowseMarkerQueueMessageReference.java → activemq-core/src/main/java/org/apache/activemq/broker/region/NullMessageReference.java
View File

@ -27,7 +27,7 @@ import org.apache.activemq.command.MessageId;
/** /**
* Only used by the {@link QueueMessageReference#NULL_MESSAGE} * Only used by the {@link QueueMessageReference#NULL_MESSAGE}
*/ */
final class EndOfBrowseMarkerQueueMessageReference implements final class NullMessageReference implements
QueueMessageReference { QueueMessageReference {
private ActiveMQMessage message = new ActiveMQMessage(); private ActiveMQMessage message = new ActiveMQMessage();
@ -50,7 +50,7 @@ final class EndOfBrowseMarkerQueueMessageReference implements
} }
public boolean lock(LockOwner subscription) { public boolean lock(LockOwner subscription) {
throw new RuntimeException("not implemented"); return true;
} }
public void setAcked(boolean b) { public void setAcked(boolean b) {
@ -58,7 +58,6 @@ final class EndOfBrowseMarkerQueueMessageReference implements
} }
public void unlock() { public void unlock() {
throw new RuntimeException("not implemented");
} }
public int decrementReferenceCount() { public int decrementReferenceCount() {
@ -70,11 +69,11 @@ final class EndOfBrowseMarkerQueueMessageReference implements
} }
public String getGroupID() { public String getGroupID() {
throw new RuntimeException("not implemented"); return null;
} }
public int getGroupSequence() { public int getGroupSequence() {
throw new RuntimeException("not implemented"); return 0;
} }
public Message getMessage() throws IOException { public Message getMessage() throws IOException {

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

@ -25,7 +25,7 @@ package org.apache.activemq.broker.region;
*/ */
public interface QueueMessageReference extends MessageReference { public interface QueueMessageReference extends MessageReference {
public static final QueueMessageReference NULL_MESSAGE = new EndOfBrowseMarkerQueueMessageReference(); public static final QueueMessageReference NULL_MESSAGE = new NullMessageReference();
public boolean isAcked(); public boolean isAcked();

34
activemq-core/src/test/java/org/apache/activemq/ZeroPrefetchConsumerTest.java
View File

@ -21,14 +21,7 @@ import org.apache.activemq.spring.SpringConsumer;
import org.apache.commons.logging.Log; import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory; import org.apache.commons.logging.LogFactory;
import javax.jms.Connection; import javax.jms.*;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
/** /**
* *
@ -65,6 +58,31 @@ public class ZeroPrefetchConsumerTest extends EmbeddedBrokerTestSupport {
MessageConsumer consumer = session.createConsumer(queue); MessageConsumer consumer = session.createConsumer(queue);
Message answer = consumer.receive(5000); Message answer = consumer.receive(5000);
assertNotNull("Should have received a message!", answer); assertNotNull("Should have received a message!", answer);
// check if method will return at all and will return a null
answer = consumer.receive(1000);
assertNull("Should have not received a message!", answer);
answer = consumer.receiveNoWait();
assertNull("Should have not received a message!", answer);
}
public void testIdleConsumer() throws Exception {
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
MessageProducer producer = session.createProducer(queue);
producer.send(session.createTextMessage("Msg1"));
producer.send(session.createTextMessage("Msg2"));
// now lets receive it
MessageConsumer consumer = session.createConsumer(queue);
//noinspection UNUSED_SYMBOL
MessageConsumer idleConsumer = session.createConsumer(queue);
TextMessage answer = (TextMessage) consumer.receive(5000);
assertEquals("Should have received a message!", answer.getText(), "Msg1");
// this call would return null if prefetchSize > 0
answer = (TextMessage) consumer.receive(5000);
assertEquals("Should have not received a message!", answer.getText(), "Msg2");
answer = (TextMessage) consumer.receiveNoWait();
assertNull("Should have not received a message!", answer);
} }
protected void setUp() throws Exception { protected void setUp() throws Exception {