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

https://issues.apache.org/jira/browse/AMQ-3576 - only update audit sequence on valid message, apply test to isolated class where broker side timeout is absent 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1196804 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Gary Tully 2011-11-02 21:17:41 +00:00
parent aa8bc42fd4
commit 5cf33b52c1
4 changed files with 38 additions and 64 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/ProducerBrokerExchange.java
View File

@ -138,7 +138,9 @@ public class ProducerBrokerExchange {
+ messageSend.getMessageId().getProducerSequenceId() + "] less than last stored: " + lastSendSequenceNumber); + messageSend.getMessageId().getProducerSequenceId() + "] less than last stored: " + lastSendSequenceNumber);
} }
} }
if (canDispatch) {
lastSendSequenceNumber.set(messageSend.getMessageId().getProducerSequenceId()); lastSendSequenceNumber.set(messageSend.getMessageId().getProducerSequenceId());
}
return canDispatch; return canDispatch;
} }

5
activemq-core/src/main/java/org/apache/activemq/state/ConnectionStateTracker.java
View File

@ -80,6 +80,9 @@ public class ConnectionStateTracker extends CommandVisitorAdapter {
} else if (eldest.getValue() instanceof MessagePull) { } else if (eldest.getValue() instanceof MessagePull) {
currentCacheSize -= MESSAGE_PULL_SIZE; currentCacheSize -= MESSAGE_PULL_SIZE;
} }
if (LOG.isTraceEnabled()) {
LOG.trace("removing tracked message: " + eldest.getKey());
}
} }
return result; return result;
} }
@ -167,7 +170,7 @@ public class ConnectionStateTracker extends CommandVisitorAdapter {
//now flush messages //now flush messages
for (Command msg:messageCache.values()) { for (Command msg:messageCache.values()) {
if (LOG.isDebugEnabled()) { if (LOG.isDebugEnabled()) {
LOG.debug("command: " + msg.getCommandId()); LOG.debug("command: " + (msg.isMessage() ? ((Message) msg).getMessageId() : msg));
} }
transport.oneway(msg); transport.oneway(msg);
} }

26
activemq-core/src/test/java/org/apache/activemq/transport/SoWriteTimeoutClientTest.java
View File

@ -28,30 +28,33 @@ import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.JmsTestSupport; import org.apache.activemq.JmsTestSupport;
import org.apache.activemq.broker.BrokerService; import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.command.ActiveMQQueue; import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter;
import org.apache.activemq.util.SocketProxy; import org.apache.activemq.util.SocketProxy;
import org.apache.activemq.util.URISupport; import org.apache.activemq.util.URISupport;
import org.apache.activemq.util.Wait;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
public class SoWriteTimeoutClientTest extends JmsTestSupport { public class SoWriteTimeoutClientTest extends JmsTestSupport {
private static final Logger LOG = LoggerFactory.getLogger(SoWriteTimeoutClientTest.class); private static final Logger LOG = LoggerFactory.getLogger(SoWriteTimeoutClientTest.class);
public String brokerTransportScheme = "tcp";
protected BrokerService createBroker() throws Exception { protected BrokerService createBroker() throws Exception {
BrokerService broker = new BrokerService(); BrokerService broker = new BrokerService();
broker.addConnector(brokerTransportScheme + "://localhost:0?wireFormat.maxInactivityDuration=0"); broker.setDeleteAllMessagesOnStartup(true);
KahaDBPersistenceAdapter adapter = new KahaDBPersistenceAdapter();
adapter.setConcurrentStoreAndDispatchQueues(false);
broker.setPersistenceAdapter(adapter);
broker.addConnector("tcp://localhost:0?wireFormat.maxInactivityDuration=0");
return broker; return broker;
} }
public void x_testSendWithClientWriteTimeout() throws Exception { public void testSendWithClientWriteTimeout() throws Exception {
final ActiveMQQueue dest = new ActiveMQQueue("testClientWriteTimeout"); final ActiveMQQueue dest = new ActiveMQQueue("testClientWriteTimeout");
messageTextPrefix = initMessagePrefix(80*1024); messageTextPrefix = initMessagePrefix(80*1024);
URI tcpBrokerUri = URISupport.removeQuery(broker.getTransportConnectors().get(0).getConnectUri()); URI tcpBrokerUri = URISupport.removeQuery(broker.getTransportConnectors().get(0).getConnectUri());
LOG.info("consuming using uri: " + tcpBrokerUri); LOG.info("consuming using uri: " + tcpBrokerUri);
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(tcpBrokerUri); ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(tcpBrokerUri);
Connection c = factory.createConnection(); Connection c = factory.createConnection();
c.start(); c.start();
@ -62,7 +65,7 @@ public class SoWriteTimeoutClientTest extends JmsTestSupport {
proxy.setTarget(tcpBrokerUri); proxy.setTarget(tcpBrokerUri);
proxy.open(); proxy.open();
ActiveMQConnectionFactory pFactory = new ActiveMQConnectionFactory("failover:(" + proxy.getUrl() + "?soWriteTimeout=500)?jms.useAsyncSend=true"); ActiveMQConnectionFactory pFactory = new ActiveMQConnectionFactory("failover:(" + proxy.getUrl() + "?soWriteTimeout=4000&sleep=500)?jms.useAsyncSend=true&trackMessages=true&maxCacheSize=6638400");
final Connection pc = pFactory.createConnection(); final Connection pc = pFactory.createConnection();
pc.start(); pc.start();
proxy.pause(); proxy.pause();
@ -81,11 +84,20 @@ public class SoWriteTimeoutClientTest extends JmsTestSupport {
}); });
// wait for timeout and reconnect // wait for timeout and reconnect
TimeUnit.SECONDS.sleep(20); TimeUnit.SECONDS.sleep(8);
proxy.goOn(); proxy.goOn();
for (int i=0; i<messageCount; i++) { for (int i=0; i<messageCount; i++) {
assertNotNull("Got message " + i + " after reconnect", consumer.receive(5000)); assertNotNull("Got message " + i + " after reconnect", consumer.receive(5000));
} }
assertTrue("no pending messages when done", Wait.waitFor(new Wait.Condition() {
@Override
public boolean isSatisified() throws Exception {
LOG.info("current total message count: " + broker.getAdminView().getTotalMessageCount());
return broker.getAdminView().getTotalMessageCount() == 0;
}
}));
} }
private String initMessagePrefix(int i) { private String initMessagePrefix(int i) {

67
activemq-core/src/test/java/org/apache/activemq/transport/SoWriteTimeoutTest.java
View File

@ -16,6 +16,15 @@
*/ */
package org.apache.activemq.transport; package org.apache.activemq.transport;
import java.net.Socket;
import java.net.SocketException;
import java.net.URI;
import java.util.concurrent.TimeUnit;
import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import junit.framework.Test; import junit.framework.Test;
import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.JmsTestSupport; import org.apache.activemq.JmsTestSupport;
@ -29,27 +38,20 @@ import org.apache.activemq.util.URISupport;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import javax.jms.*;
import java.net.Socket;
import java.net.SocketException;
import java.net.URI;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class SoWriteTimeoutTest extends JmsTestSupport { public class SoWriteTimeoutTest extends JmsTestSupport {
private static final Logger LOG = LoggerFactory.getLogger(SoWriteTimeoutTest.class); private static final Logger LOG = LoggerFactory.getLogger(SoWriteTimeoutTest.class);
final int receiveBufferSize = 16*1024; final int receiveBufferSize = 16*1024;
public String brokerTransportScheme = "tcp"; public String brokerTransportScheme = "nio";
protected BrokerService createBroker() throws Exception { protected BrokerService createBroker() throws Exception {
BrokerService broker = super.createBroker(); BrokerService broker = super.createBroker();
broker.setPersistent(true); broker.setPersistent(true);
broker.setDeleteAllMessagesOnStartup(true);
KahaDBPersistenceAdapter adapter = new KahaDBPersistenceAdapter(); KahaDBPersistenceAdapter adapter = new KahaDBPersistenceAdapter();
adapter.setConcurrentStoreAndDispatchQueues(false); adapter.setConcurrentStoreAndDispatchQueues(false);
broker.setPersistenceAdapter(adapter); broker.setPersistenceAdapter(adapter);
broker.addConnector(brokerTransportScheme + "://localhost:0?wireFormat.maxInactivityDuration=0"); broker.addConnector(brokerTransportScheme + "://localhost:0?wireFormat.maxInactivityDuration=0&transport.soWriteTimeout=1000&transport.sleep=1000");
if ("nio".equals(brokerTransportScheme)) { if ("nio".equals(brokerTransportScheme)) {
broker.addConnector("stomp+" + brokerTransportScheme + "://localhost:0?transport.soWriteTimeout=1000&transport.sleep=1000&socketBufferSize=" + receiveBufferSize + "&trace=true"); broker.addConnector("stomp+" + brokerTransportScheme + "://localhost:0?transport.soWriteTimeout=1000&transport.sleep=1000&socketBufferSize=" + receiveBufferSize + "&trace=true");
} }
@ -148,51 +150,6 @@ public class SoWriteTimeoutTest extends JmsTestSupport {
} }
} }
public void testClientWriteTimeout() throws Exception {
final ActiveMQQueue dest = new ActiveMQQueue("testClientWriteTimeout");
messageTextPrefix = initMessagePrefix(80*1024);
URI tcpBrokerUri = URISupport.removeQuery(broker.getTransportConnectors().get(0).getConnectUri());
LOG.info("consuming using uri: " + tcpBrokerUri);
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(tcpBrokerUri);
Connection c = factory.createConnection();
c.start();
Session session = c.createSession(false, Session.AUTO_ACKNOWLEDGE);
MessageConsumer consumer = session.createConsumer(dest);
SocketProxy proxy = new SocketProxy();
proxy.setTarget(tcpBrokerUri);
proxy.open();
ActiveMQConnectionFactory pFactory = new ActiveMQConnectionFactory("failover:(" + proxy.getUrl() + "?soWriteTimeout=500)?jms.useAsyncSend=true&trackMessages=true");
final Connection pc = pFactory.createConnection();
pc.start();
proxy.pause();
final int messageCount = 20;
ExecutorService executorService = Executors.newCachedThreadPool();
executorService.execute(new Runnable() {
@Override
public void run() {
try {
sendMessages(pc, dest, messageCount);
} catch (Exception ignored) {
ignored.printStackTrace();
}
}
});
// wait for timeout and reconnect
TimeUnit.SECONDS.sleep(7);
proxy.goOn();
for (int i=0; i<messageCount; i++) {
assertNotNull("Got message after reconnect", consumer.receive(5000));
}
//broker.getAdminView().get
}
private String initMessagePrefix(int i) { private String initMessagePrefix(int i) {
byte[] content = new byte[i]; byte[] content = new byte[i];
return new String(content); return new String(content);