You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ha...@apache.org on 2014/03/12 19:46:57 UTC
[01/17] git commit: fix for:
https://issues.apache.org/jira/browse/AMQ-4893
Repository: activemq
Updated Branches:
refs/heads/activemq-5.9 48a9edc8c -> 687aa9265
fix for: https://issues.apache.org/jira/browse/AMQ-4893
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/fcc773a0
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/fcc773a0
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/fcc773a0
Branch: refs/heads/activemq-5.9
Commit: fcc773a05bc55b74d4d09cbd48dd66ea1a764e2d
Parents: 48a9edc
Author: Timothy Bish <ta...@gmai.com>
Authored: Mon Nov 25 14:48:38 2013 -0500
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Wed Mar 12 13:07:48 2014 -0400
----------------------------------------------------------------------
.../activemq/command/ActiveMQMessage.java | 5 ++
activemq-unit-tests/.gitignore | 3 +
.../org/apache/activemq/bugs/AMQ4893Test.java | 69 ++++++++++++++++++++
3 files changed, 77 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/fcc773a0/activemq-client/src/main/java/org/apache/activemq/command/ActiveMQMessage.java
----------------------------------------------------------------------
diff --git a/activemq-client/src/main/java/org/apache/activemq/command/ActiveMQMessage.java b/activemq-client/src/main/java/org/apache/activemq/command/ActiveMQMessage.java
index e673d96..eef6c11 100755
--- a/activemq-client/src/main/java/org/apache/activemq/command/ActiveMQMessage.java
+++ b/activemq-client/src/main/java/org/apache/activemq/command/ActiveMQMessage.java
@@ -38,6 +38,7 @@ import org.apache.activemq.state.CommandVisitor;
import org.apache.activemq.util.Callback;
import org.apache.activemq.util.JMSExceptionSupport;
import org.apache.activemq.util.TypeConversionSupport;
+import org.fusesource.hawtbuf.UTF8Buffer;
/**
*
@@ -484,6 +485,10 @@ public class ActiveMQMessage extends Message implements org.apache.activemq.Mess
throw new IllegalArgumentException("Property name cannot be empty or null");
}
+ if (value instanceof UTF8Buffer) {
+ value = value.toString();
+ }
+
checkValidObject(value);
value = convertScheduled(name, value);
PropertySetter setter = JMS_PROPERTY_SETERS.get(name);
http://git-wip-us.apache.org/repos/asf/activemq/blob/fcc773a0/activemq-unit-tests/.gitignore
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/.gitignore b/activemq-unit-tests/.gitignore
new file mode 100644
index 0000000..d2b46e9
--- /dev/null
+++ b/activemq-unit-tests/.gitignore
@@ -0,0 +1,3 @@
+/createData
+/derbydb_15
+/testJdbcConfig
http://git-wip-us.apache.org/repos/asf/activemq/blob/fcc773a0/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4893Test.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4893Test.java b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4893Test.java
new file mode 100644
index 0000000..d78000d
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4893Test.java
@@ -0,0 +1,69 @@
+package org.apache.activemq.bugs;
+
+import java.io.IOException;
+import java.util.Map;
+
+import javax.jms.JMSException;
+
+import org.apache.activemq.command.ActiveMQObjectMessage;
+import org.apache.activemq.openwire.OpenWireFormat;
+import org.apache.activemq.util.ByteSequence;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class AMQ4893Test {
+
+ private static final transient Logger LOG = LoggerFactory.getLogger(AMQ4893Test.class);
+
+ @Test
+ public void testPropertiesInt() throws Exception {
+ ActiveMQObjectMessage message = new ActiveMQObjectMessage();
+ message.setIntProperty("TestProp", 333);
+ fakeUnmarshal(message);
+ roundTripProperties(message);
+ }
+
+ @Test
+ public void testPropertiesString() throws Exception {
+ ActiveMQObjectMessage message = new ActiveMQObjectMessage();
+ message.setStringProperty("TestProp", "Value");
+ fakeUnmarshal(message);
+ roundTripProperties(message);
+ }
+
+ @Test
+ public void testPropertiesObject() throws Exception {
+ ActiveMQObjectMessage message = new ActiveMQObjectMessage();
+ message.setObjectProperty("TestProp", "Value");
+ fakeUnmarshal(message);
+ roundTripProperties(message);
+ }
+
+ @Test
+ public void testPropertiesObjectNoMarshalling() throws Exception {
+ ActiveMQObjectMessage message = new ActiveMQObjectMessage();
+ message.setObjectProperty("TestProp", "Value");
+ roundTripProperties(message);
+ }
+
+ private void roundTripProperties(ActiveMQObjectMessage message) throws IOException, JMSException {
+ ActiveMQObjectMessage copy = new ActiveMQObjectMessage();
+ for (Map.Entry<String, Object> prop : message.getProperties().entrySet()) {
+ LOG.debug("{} -> {}", prop.getKey(), prop.getValue().getClass());
+ copy.setObjectProperty(prop.getKey(), prop.getValue());
+ }
+ }
+
+ private void fakeUnmarshal(ActiveMQObjectMessage message) throws IOException {
+ // we need to force the unmarshalled property field to be set so it
+ // gives us a hawtbuffer for the string
+ OpenWireFormat format = new OpenWireFormat();
+ message.beforeMarshall(format);
+ message.afterMarshall(format);
+
+ ByteSequence seq = message.getMarshalledProperties();
+ message.clearProperties();
+ message.setMarshalledProperties(seq);
+ }
+}
\ No newline at end of file
[11/17] git commit: Fix for
https://issues.apache.org/jira/browse/AMQ-4916
Posted by ha...@apache.org.
Fix for https://issues.apache.org/jira/browse/AMQ-4916
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/f6ed548b
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/f6ed548b
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/f6ed548b
Branch: refs/heads/activemq-5.9
Commit: f6ed548b82d4c4472b378871d646d27e9f519d0d
Parents: 2e5c9d5
Author: Rob Davies <ra...@gmail.com>
Authored: Tue Dec 3 17:15:02 2013 +0000
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Wed Mar 12 13:13:02 2014 -0400
----------------------------------------------------------------------
.../org/apache/activemq/broker/jmx/ProducerView.java | 12 ++++++++++++
.../apache/activemq/broker/jmx/ProducerViewMBean.java | 9 ++++++++-
.../apache/activemq/broker/jmx/SubscriptionView.java | 14 ++++++++++++++
.../activemq/broker/jmx/SubscriptionViewMBean.java | 7 +++++++
.../apache/activemq/broker/region/AbstractRegion.java | 3 +++
.../activemq/broker/region/AbstractSubscription.java | 7 +++++++
.../apache/activemq/broker/region/Subscription.java | 3 +++
.../org/apache/activemq/command/ProducerInfo.java | 6 ++++++
.../broker/region/QueueDuplicatesFromStoreTest.java | 6 ++++++
9 files changed, 66 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/f6ed548b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/ProducerView.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/ProducerView.java b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/ProducerView.java
index 1596d5e..6905c72 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/ProducerView.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/ProducerView.java
@@ -184,4 +184,16 @@ public class ProducerView implements ProducerViewMBean {
producerBrokerExchange.resetFlowControl();
}
}
+
+ @Override
+ public void resetStatistics() {
+ if (info != null){
+ info.getSentCount().reset();
+ }
+ }
+
+ @Override
+ public long getSentCount() {
+ return info != null ? info.getSentCount().getCount() :0;
+ }
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/f6ed548b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/ProducerViewMBean.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/ProducerViewMBean.java b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/ProducerViewMBean.java
index da357c1..14c2073 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/ProducerViewMBean.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/ProducerViewMBean.java
@@ -96,6 +96,13 @@ public interface ProducerViewMBean {
@MBeanInfo("percentage of sends Producer Blocked for Flow Control")
int getPercentageBlocked();
- @MBeanInfo("reset flow control stata")
+ @MBeanInfo("reset flow control state")
void resetFlowControlStats();
+
+ @MBeanInfo("Resets statistics.")
+ void resetStatistics();
+
+ @MBeanInfo("Messages consumed")
+ long getSentCount();
+
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/f6ed548b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/SubscriptionView.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/SubscriptionView.java b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/SubscriptionView.java
index 8201737..443a266 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/SubscriptionView.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/SubscriptionView.java
@@ -88,6 +88,8 @@ public class SubscriptionView implements SubscriptionViewMBean {
return result;
}
+
+
private ObjectName createConnectionQuery(ManagementContext ctx, String brokerName) throws IOException {
try {
return BrokerMBeanSupport.createConnectionQuery(ctx.getJmxDomainName(), brokerName, clientId);
@@ -415,4 +417,16 @@ public class SubscriptionView implements SubscriptionViewMBean {
public String getUserName() {
return userName;
}
+
+ @Override
+ public void resetStatistics() {
+ if (subscription != null){
+ subscription.getConsumedCount().reset();
+ }
+ }
+
+ @Override
+ public long getConsumedCount() {
+ return subscription != null ? subscription.getConsumedCount().getCount() : 0;
+ }
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/f6ed548b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/SubscriptionViewMBean.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/SubscriptionViewMBean.java b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/SubscriptionViewMBean.java
index 9bbedc1..3c3aab3 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/SubscriptionViewMBean.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/SubscriptionViewMBean.java
@@ -242,4 +242,11 @@ public interface SubscriptionViewMBean {
@MBeanInfo("ObjectName of the Connection that created this Subscription")
ObjectName getConnection();
+
+ @MBeanInfo("Resets statistics.")
+ void resetStatistics();
+
+ @MBeanInfo("Messages consumed")
+ long getConsumedCount();
+
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/f6ed548b/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java b/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java
index 168bd96..16deed4 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractRegion.java
@@ -392,6 +392,9 @@ public abstract class AbstractRegion implements Region {
}
producerExchange.getRegionDestination().send(producerExchange, messageSend);
+ if (producerExchange.getProducerState() != null && producerExchange.getProducerState().getInfo() != null){
+ producerExchange.getProducerState().getInfo().getSentCount().increment();
+ }
}
public void acknowledge(ConsumerBrokerExchange consumerExchange, MessageAck ack) throws Exception {
http://git-wip-us.apache.org/repos/asf/activemq/blob/f6ed548b/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractSubscription.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractSubscription.java b/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractSubscription.java
index 06a44bf..b2ff01c 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractSubscription.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractSubscription.java
@@ -36,6 +36,7 @@ import org.apache.activemq.filter.DestinationFilter;
import org.apache.activemq.filter.LogicExpression;
import org.apache.activemq.filter.MessageEvaluationContext;
import org.apache.activemq.filter.NoLocalExpression;
+import org.apache.activemq.management.CountStatisticImpl;
import org.apache.activemq.selector.SelectorParser;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -53,6 +54,7 @@ public abstract class AbstractSubscription implements Subscription {
private int cursorMemoryHighWaterMark = 70;
private boolean slowConsumer;
private long lastAckTime;
+ private CountStatisticImpl consumedCount = new CountStatisticImpl("consumed","The number of messages consumed");
public AbstractSubscription(Broker broker,ConnectionContext context, ConsumerInfo info) throws InvalidSelectorException {
this.broker = broker;
@@ -88,6 +90,7 @@ public abstract class AbstractSubscription implements Subscription {
@Override
public synchronized void acknowledge(final ConnectionContext context, final MessageAck ack) throws Exception {
this.lastAckTime = System.currentTimeMillis();
+ this.consumedCount.increment();
}
@Override
@@ -276,4 +279,8 @@ public abstract class AbstractSubscription implements Subscription {
public void setTimeOfLastMessageAck(long value) {
this.lastAckTime = value;
}
+
+ public CountStatisticImpl getConsumedCount(){
+ return consumedCount;
+ }
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/f6ed548b/activemq-broker/src/main/java/org/apache/activemq/broker/region/Subscription.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/Subscription.java b/activemq-broker/src/main/java/org/apache/activemq/broker/region/Subscription.java
index dfd427d..b79b37e 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/Subscription.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/Subscription.java
@@ -30,6 +30,7 @@ import org.apache.activemq.command.MessageDispatchNotification;
import org.apache.activemq.command.MessagePull;
import org.apache.activemq.command.Response;
import org.apache.activemq.filter.MessageEvaluationContext;
+import org.apache.activemq.management.CountStatisticImpl;
/**
*
@@ -234,4 +235,6 @@ public interface Subscription extends SubscriptionRecovery {
*/
long getTimeOfLastMessageAck();
+ CountStatisticImpl getConsumedCount();
+
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/f6ed548b/activemq-client/src/main/java/org/apache/activemq/command/ProducerInfo.java
----------------------------------------------------------------------
diff --git a/activemq-client/src/main/java/org/apache/activemq/command/ProducerInfo.java b/activemq-client/src/main/java/org/apache/activemq/command/ProducerInfo.java
index 9854c5e..05ef3a4 100755
--- a/activemq-client/src/main/java/org/apache/activemq/command/ProducerInfo.java
+++ b/activemq-client/src/main/java/org/apache/activemq/command/ProducerInfo.java
@@ -16,6 +16,7 @@
*/
package org.apache.activemq.command;
+import org.apache.activemq.management.CountStatisticImpl;
import org.apache.activemq.state.CommandVisitor;
/**
@@ -32,6 +33,7 @@ public class ProducerInfo extends BaseCommand {
protected BrokerId[] brokerPath;
protected boolean dispatchAsync;
protected int windowSize;
+ protected CountStatisticImpl sentCount = new CountStatisticImpl("sentCount","number of messages sent to a broker");
public ProducerInfo() {
}
@@ -135,4 +137,8 @@ public class ProducerInfo extends BaseCommand {
this.windowSize = windowSize;
}
+ public CountStatisticImpl getSentCount(){
+ return sentCount;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/f6ed548b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/QueueDuplicatesFromStoreTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/QueueDuplicatesFromStoreTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/QueueDuplicatesFromStoreTest.java
index 6fe28fa..9da839d 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/QueueDuplicatesFromStoreTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/QueueDuplicatesFromStoreTest.java
@@ -44,6 +44,7 @@ import org.apache.activemq.command.MessagePull;
import org.apache.activemq.command.ProducerInfo;
import org.apache.activemq.command.Response;
import org.apache.activemq.filter.MessageEvaluationContext;
+import org.apache.activemq.management.CountStatisticImpl;
import org.apache.activemq.state.ProducerState;
import org.apache.activemq.store.MessageStore;
import org.apache.activemq.store.PersistenceAdapter;
@@ -339,6 +340,11 @@ public class QueueDuplicatesFromStoreTest extends TestCase {
public long getTimeOfLastMessageAck() {
return 0;
}
+
+ @Override
+ public CountStatisticImpl getConsumedCount() {
+ return null;
+ }
};
queue.addSubscription(contextNotInTx, subscription);
[03/17] git commit: https://issues.apache.org/jira/browse/AMQ-4905 -
resolve by embedding spring xsd
Posted by ha...@apache.org.
https://issues.apache.org/jira/browse/AMQ-4905 - resolve by embedding spring xsd
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/3444c89f
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/3444c89f
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/3444c89f
Branch: refs/heads/activemq-5.9
Commit: 3444c89fd8cddfde9f08cb1e101c65f6824b3f27
Parents: 77b4d70
Author: gtully <ga...@gmail.com>
Authored: Wed Nov 27 16:04:14 2013 +0000
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Wed Mar 12 13:10:50 2014 -0400
----------------------------------------------------------------------
activemq-runtime-config/pom.xml | 21 ++++++++++++++++++++
.../plugin/RuntimeConfigurationBroker.java | 17 ++--------------
2 files changed, 23 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/3444c89f/activemq-runtime-config/pom.xml
----------------------------------------------------------------------
diff --git a/activemq-runtime-config/pom.xml b/activemq-runtime-config/pom.xml
index 6e3ea16..a49b3a2 100755
--- a/activemq-runtime-config/pom.xml
+++ b/activemq-runtime-config/pom.xml
@@ -115,6 +115,7 @@
<artifactItem>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-spring</artifactId>
+ <version>${project.version}</version>
<type>jar</type>
<overWrite>true</overWrite>
<includes>activemq.xsd</includes>
@@ -123,6 +124,26 @@
<outputDirectory>src/main/resources</outputDirectory>
</configuration>
</execution>
+ <execution>
+ <id>unpack-spring-xsd</id>
+ <phase>generate-sources</phase>
+ <goals>
+ <goal>unpack</goal>
+ </goals>
+ <configuration>
+ <artifactItems>
+ <artifactItem>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-beans</artifactId>
+ <version>${spring-version}</version>
+ <type>jar</type>
+ <overWrite>true</overWrite>
+ <includes>org/springframework/beans/factory/xml/spring-beans-3.0.xsd</includes>
+ </artifactItem>
+ </artifactItems>
+ <outputDirectory>src/main/resources</outputDirectory>
+ </configuration>
+ </execution>
</executions>
</plugin>
<plugin>
http://git-wip-us.apache.org/repos/asf/activemq/blob/3444c89f/activemq-runtime-config/src/main/java/org/apache/activemq/plugin/RuntimeConfigurationBroker.java
----------------------------------------------------------------------
diff --git a/activemq-runtime-config/src/main/java/org/apache/activemq/plugin/RuntimeConfigurationBroker.java b/activemq-runtime-config/src/main/java/org/apache/activemq/plugin/RuntimeConfigurationBroker.java
index 82a5ee3..1879b1c 100644
--- a/activemq-runtime-config/src/main/java/org/apache/activemq/plugin/RuntimeConfigurationBroker.java
+++ b/activemq-runtime-config/src/main/java/org/apache/activemq/plugin/RuntimeConfigurationBroker.java
@@ -809,7 +809,7 @@ public class RuntimeConfigurationBroker extends BrokerFilter {
Object springBean = getClass().getClassLoader().loadClass(value).newInstance();
if (springBean instanceof FactoryBean) {
// can't access the factory or created properties from spring context so we got to recreate
- initialProperties.putAll((Properties) FactoryBean.class.getMethod("getObject", null).invoke(springBean));
+ initialProperties.putAll((Properties) FactoryBean.class.getMethod("getObject", (Class<?>[]) null).invoke(springBean));
}
} catch (Throwable e) {
LOG.debug("unexpected exception processing properties bean class: " + propertiesClazzes, e);
@@ -856,20 +856,7 @@ public class RuntimeConfigurationBroker extends BrokerFilter {
ArrayList<StreamSource> schemas = new ArrayList<StreamSource>();
schemas.add(new StreamSource(getClass().getResource("/activemq.xsd").toExternalForm()));
-
- // avoid going to the net to pull down the spring schema,
- // REVISIT may need to be smarter in osgi
- final PluggableSchemaResolver springResolver =
- new PluggableSchemaResolver(getClass().getClassLoader());
- final InputSource beanInputSource =
- springResolver.resolveEntity(
- "http://www.springframework.org/schema/beans",
- "http://www.springframework.org/schema/beans/spring-beans.xsd");
- if (beanInputSource != null) {
- schemas.add(new StreamSource(beanInputSource.getByteStream()));
- } else {
- schemas.add(new StreamSource("http://www.springframework.org/schema/beans/spring-beans.xsd"));
- }
+ schemas.add(new StreamSource(getClass().getResource("/org/springframework/beans/factory/xml/spring-beans-3.0.xsd").toExternalForm()));
schema = schemaFactory.newSchema(schemas.toArray(new Source[]{}));
}
return schema;
[02/17] git commit: https://issues.apache.org/jira/browse/AMQ-4904 -
ensure transports are not restarted without a store lock
Posted by ha...@apache.org.
https://issues.apache.org/jira/browse/AMQ-4904 - ensure transports are not restarted without a store lock
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/77b4d700
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/77b4d700
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/77b4d700
Branch: refs/heads/activemq-5.9
Commit: 77b4d700b139793dd3489f54bbf54386ce5258b1
Parents: fcc773a
Author: gtully <ga...@gmail.com>
Authored: Wed Nov 27 12:48:16 2013 +0000
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Wed Mar 12 13:10:35 2014 -0400
----------------------------------------------------------------------
.../util/DefaultIOExceptionHandler.java | 8 +-
.../store/jdbc/JDBCIOExceptionHandler.java | 4 +
.../jdbc/JDBCIOExceptionHandlerMockeryTest.java | 108 +++++++++++++++++++
3 files changed, 117 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/77b4d700/activemq-broker/src/main/java/org/apache/activemq/util/DefaultIOExceptionHandler.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/util/DefaultIOExceptionHandler.java b/activemq-broker/src/main/java/org/apache/activemq/util/DefaultIOExceptionHandler.java
index ab35800..c65ec65 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/util/DefaultIOExceptionHandler.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/util/DefaultIOExceptionHandler.java
@@ -94,10 +94,12 @@ import org.slf4j.LoggerFactory;
LOG.info("waiting for broker persistence adapter checkpoint to succeed before restarting transports");
TimeUnit.MILLISECONDS.sleep(resumeCheckSleepPeriod);
}
- broker.startAllConnectors();
- LOG.info("Successfully restarted transports on " + broker);
+ if (hasLockOwnership()) {
+ broker.startAllConnectors();
+ LOG.info("Successfully restarted transports on " + broker);
+ }
} catch (Exception e) {
- LOG.warn("Stopping " + broker + " due to failure while restarting transports", e);
+ LOG.warn("Stopping " + broker + " due to failure restarting transports", e);
stopBroker(e);
} finally {
handlingException.compareAndSet(true, false);
http://git-wip-us.apache.org/repos/asf/activemq/blob/77b4d700/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandler.java
----------------------------------------------------------------------
diff --git a/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandler.java b/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandler.java
index a92856c..d0ea276 100644
--- a/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandler.java
+++ b/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandler.java
@@ -21,11 +21,14 @@ import java.io.IOException;
import org.apache.activemq.broker.Locker;
import org.apache.activemq.broker.SuppressReplyException;
import org.apache.activemq.util.DefaultIOExceptionHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* @org.apache.xbean.XBean
*/
public class JDBCIOExceptionHandler extends DefaultIOExceptionHandler {
+ private static final Logger LOG = LoggerFactory.getLogger(JDBCIOExceptionHandler.class);
public JDBCIOExceptionHandler() {
setIgnoreSQLExceptions(false);
@@ -49,6 +52,7 @@ public class JDBCIOExceptionHandler extends DefaultIOExceptionHandler {
}
if (!hasLock) {
+ LOG.warn("Lock keepAlive failed, no longer lock owner with: {}", locker);
throw new IOException("Lock keepAlive failed, no longer lock owner with: " + locker);
}
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/77b4d700/activemq-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandlerMockeryTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandlerMockeryTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandlerMockeryTest.java
new file mode 100644
index 0000000..dd5d506
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandlerMockeryTest.java
@@ -0,0 +1,108 @@
+/**
+ * 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.store.jdbc;
+
+import java.io.IOException;
+import java.util.HashMap;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.Locker;
+import org.apache.activemq.broker.SuppressReplyException;
+import org.apache.activemq.util.ServiceStopper;
+import org.apache.activemq.util.Wait;
+import org.jmock.Expectations;
+import org.jmock.Mockery;
+import org.jmock.States;
+import org.jmock.lib.legacy.ClassImposteriser;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class JDBCIOExceptionHandlerMockeryTest {
+
+ private static final Logger LOG = LoggerFactory.getLogger(JDBCIOExceptionHandlerMockeryTest.class);
+ private HashMap<Thread, Throwable> exceptions = new HashMap<Thread, Throwable>();
+
+ @Test
+ public void testShutdownWithoutTransportRestart() throws Exception {
+
+ Mockery context = new Mockery() {{
+ setImposteriser(ClassImposteriser.INSTANCE);
+ }};
+
+ Thread.setDefaultUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
+ @Override
+ public void uncaughtException(Thread t, Throwable e) {
+ LOG.error("unexpected exception {} on thread {}", e, t);
+ exceptions.put(t, e);
+ }
+ });
+
+ final BrokerService brokerService = context.mock(BrokerService.class);
+ final JDBCPersistenceAdapter jdbcPersistenceAdapter = context.mock(JDBCPersistenceAdapter.class);
+ final Locker locker = context.mock(Locker.class);
+
+ final States jdbcConn = context.states("jdbc").startsAs("down");
+ final States broker = context.states("broker").startsAs("started");
+
+ // simulate jdbc up between hasLock and checkpoint, so hasLock fails to verify
+ context.checking(new Expectations() {{
+ allowing(brokerService).isRestartAllowed();
+ will(returnValue(false));
+ allowing(brokerService).stopAllConnectors(with(any(ServiceStopper.class)));
+ allowing(brokerService).getPersistenceAdapter();
+ will(returnValue(jdbcPersistenceAdapter));
+ allowing(jdbcPersistenceAdapter).getLocker();
+ will(returnValue(locker));
+ allowing(locker).keepAlive();
+ when(jdbcConn.is("down"));
+ will(returnValue(true));
+ allowing(locker).keepAlive();
+ when(jdbcConn.is("up"));
+ will(returnValue(false));
+
+ allowing(jdbcPersistenceAdapter).checkpoint(with(true));
+ then(jdbcConn.is("up"));
+ allowing(brokerService).stop();
+ then(broker.is("stopped"));
+
+ }});
+
+ JDBCIOExceptionHandler underTest = new JDBCIOExceptionHandler();
+ underTest.setBrokerService(brokerService);
+
+ try {
+ underTest.handle(new IOException());
+ fail("except suppress reply ex");
+ } catch (SuppressReplyException expected) {
+ }
+
+ assertTrue("broker stopped state triggered", Wait.waitFor(new Wait.Condition() {
+ @Override
+ public boolean isSatisified() throws Exception {
+ LOG.info("broker state {}", broker);
+ return broker.is("stopped").isActive();
+ }
+ }));
+ context.assertIsSatisfied();
+
+ assertTrue("no exceptions: " + exceptions, exceptions.isEmpty());
+ }
+}
[13/17] git commit: Fixing
https://issues.apache.org/jira/browse/AMQ-4917 : LevelDB store can fail when
using durable subs.
Posted by ha...@apache.org.
Fixing https://issues.apache.org/jira/browse/AMQ-4917 : LevelDB store can fail when using durable subs.
We were browsing durable sub entries which had been concurrently GCed causing leveldb store failures which then caused the broker to restart.
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/3faddcd2
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/3faddcd2
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/3faddcd2
Branch: refs/heads/activemq-5.9
Commit: 3faddcd2d19de634eac53e2584bebc856e24d6d3
Parents: 720b8ac
Author: Hiram Chirino <hi...@hiramchirino.com>
Authored: Tue Dec 3 19:20:11 2013 -0500
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Wed Mar 12 13:13:33 2014 -0400
----------------------------------------------------------------------
.../org/apache/activemq/leveldb/DBManager.scala | 12 +-
.../apache/activemq/leveldb/LevelDBClient.scala | 117 +++++++++----------
.../apache/activemq/leveldb/LevelDBStore.scala | 10 +-
.../org/apache/activemq/leveldb/RecordLog.scala | 61 +++++-----
4 files changed, 104 insertions(+), 96 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/3faddcd2/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala
----------------------------------------------------------------------
diff --git a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala
index 6b575ee..722d932 100644
--- a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala
+++ b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala
@@ -37,7 +37,9 @@ import org.apache.activemq.ActiveMQMessageAuditNoSync
import org.fusesource.hawtdispatch
case class EntryLocator(qid:Long, seq:Long)
-case class DataLocator(store:LevelDBStore, pos:Long, len:Int)
+case class DataLocator(store:LevelDBStore, pos:Long, len:Int) {
+ override def toString: String = "DataLocator(%x, %d)".format(pos, len)
+}
case class MessageRecord(store:LevelDBStore, id:MessageId, data:Buffer, syncNeeded:Boolean) {
var locator:DataLocator = _
}
@@ -860,8 +862,12 @@ class DBManager(val parent:LevelDBStore) {
def getMessage(x: MessageId):Message = {
val id = Option(pendingStores.get(x)).flatMap(_.headOption).map(_.id).getOrElse(x)
val locator = id.getDataLocator()
- val msg = client.getMessageWithRetry(locator)
- msg.setMessageId(id)
+ val msg = client.getMessage(locator)
+ if( msg!=null ) {
+ msg.setMessageId(id)
+ } else {
+ LevelDBStore.warn("Could not load messages for: "+x+" at: "+locator)
+ }
msg
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/3faddcd2/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBClient.scala
----------------------------------------------------------------------
diff --git a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBClient.scala b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBClient.scala
index fe29012..7872ff8 100755
--- a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBClient.scala
+++ b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBClient.scala
@@ -21,7 +21,6 @@ import java.{lang=>jl}
import java.{util=>ju}
import java.util.concurrent.locks.ReentrantReadWriteLock
-import java.util.concurrent.atomic.AtomicBoolean
import collection.immutable.TreeMap
import collection.mutable.{HashMap, ListBuffer}
import org.iq80.leveldb._
@@ -37,20 +36,13 @@ import org.apache.activemq.command.{MessageAck, Message}
import org.apache.activemq.util.{IOExceptionSupport, ByteSequence}
import java.text.SimpleDateFormat
import java.util.{Date, Collections}
-import org.apache.activemq.leveldb.util.TimeMetric
-import org.apache.activemq.leveldb.RecordLog.LogInfo
import org.fusesource.leveldbjni.internal.JniDB
import org.apache.activemq.ActiveMQMessageAuditNoSync
-import java.util.zip.CRC32
import org.apache.activemq.leveldb.util.TimeMetric
import org.fusesource.hawtbuf.ByteArrayInputStream
import org.apache.activemq.leveldb.RecordLog.LogInfo
import scala.Some
import scala.Serializable
-import org.apache.activemq.leveldb.XaAckRecord
-import org.apache.activemq.leveldb.MessageRecord
-import org.apache.activemq.leveldb.EntryLocator
-import org.apache.activemq.leveldb.DataLocator
import org.fusesource.hawtbuf.ByteArrayOutputStream
import org.apache.activemq.broker.SuppressReplyException
@@ -772,19 +764,20 @@ class LevelDBClient(store: LevelDBStore) {
val index_record = new EntryRecord.Bean()
index_record.setValueLocation(record.getValueLocation)
- index_record.setValueLength(record.getValueLength)
- val index_value = encodeEntryRecord(index_record.freeze()).toByteArray
+ if( record.hasValueLength ) {
+ index_record.setValueLength(record.getValueLength)
+ }
+ val index_value = encodeEntryRecord(index_record.freeze()).toByteArray
replay_write_batch.put(encodeEntryKey(ENTRY_PREFIX, record.getCollectionKey, record.getEntryKey), index_value)
if( kind==LOG_ADD_ENTRY ) {
- if ( record.hasValueLocation ) {
- logRefIncrement(record.getValueLocation)
- }
+ logRefIncrement(record.getValueLocation)
collectionIncrementSize(record.getCollectionKey, record.getEntryKey.toByteArray)
+ trace("Replay of LOG_ADD_ENTRY at %d, collection: %s, entry: %s", indexRecoveryPosition, record.getCollectionKey, record.getEntryKey)
+ } else {
+ trace("Replay of LOG_UPDATE_ENTRY at %d, collection: %s, entry: %s", indexRecoveryPosition, record.getCollectionKey, record.getEntryKey)
}
- trace("Replay of LOG_ADD_ENTRY at %d, collection: %s, entry: %s", indexRecoveryPosition, record.getCollectionKey, record.getEntryKey)
-
case LOG_REMOVE_ENTRY =>
val record = decodeEntryRecord(data)
@@ -834,10 +827,9 @@ class LevelDBClient(store: LevelDBStore) {
private def logRefDecrement(pos: Long) {
for( key <- logRefKey(pos) ) {
- logRefs.get(key).foreach { counter =>
- if (counter.decrementAndGet() == 0) {
- logRefs.remove(key)
- }
+ logRefs.get(key) match {
+ case Some(counter) => counter.decrementAndGet() == 0
+ case None => warn("invalid: logRefDecrement: "+pos)
}
}
}
@@ -1252,11 +1244,16 @@ class LevelDBClient(store: LevelDBStore) {
collectionCursor(collectionKey, encodeLong(seq)) { (key, value) =>
val seq = decodeLong(key)
var locator = DataLocator(store, value.getValueLocation, value.getValueLength)
- val msg = getMessageWithRetry(locator)
- msg.getMessageId().setEntryLocator(EntryLocator(collectionKey, seq))
- msg.getMessageId().setDataLocator(locator)
- msg.setRedeliveryCounter(decodeQueueEntryMeta(value))
- func(msg)
+ val msg = getMessage(locator)
+ if( msg !=null ) {
+ msg.getMessageId().setEntryLocator(EntryLocator(collectionKey, seq))
+ msg.getMessageId().setDataLocator(locator)
+ msg.setRedeliveryCounter(decodeQueueEntryMeta(value))
+ func(msg)
+ } else {
+ warn("Could not load message seq: "+seq+" from "+locator)
+ true
+ }
}
}
@@ -1278,10 +1275,15 @@ class LevelDBClient(store: LevelDBStore) {
func(XaAckRecord(collectionKey, seq, ack, sub))
} else {
var locator = DataLocator(store, value.getValueLocation, value.getValueLength)
- val msg = getMessageWithRetry(locator)
- msg.getMessageId().setEntryLocator(EntryLocator(collectionKey, seq))
- msg.getMessageId().setDataLocator(locator)
- func(msg)
+ val msg = getMessage(locator)
+ if( msg !=null ) {
+ msg.getMessageId().setEntryLocator(EntryLocator(collectionKey, seq))
+ msg.getMessageId().setDataLocator(locator)
+ func(msg)
+ } else {
+ warn("Could not load XA message seq: "+seq+" from "+locator)
+ true
+ }
}
}
}
@@ -1295,22 +1297,6 @@ class LevelDBClient(store: LevelDBStore) {
}
}
- def getMessageWithRetry(locator:AnyRef):Message = {
- var retry = 0
- var rc = getMessage(locator);
- while( rc == null ) {
- if( retry > 10 )
- return null;
- Thread.sleep(retry*10)
- rc = getMessage(locator);
- retry+=1
- }
- if( retry > 0 ) {
- info("Recovered from 'failed getMessage' on retry: "+retry)
- }
- rc
- }
-
def getMessage(locator:AnyRef):Message = {
assert(locator!=null)
val buffer = locator match {
@@ -1487,9 +1473,7 @@ class LevelDBClient(store: LevelDBStore) {
batch.put(key, index_data)
if( kind==LOG_ADD_ENTRY ) {
- for (key <- logRefKey(dataLocator.pos, log_info)) {
- logRefs.getOrElseUpdate(key, new LongCounter()).incrementAndGet()
- }
+ logRefIncrement(dataLocator.pos)
collectionIncrementSize(entry.queueKey, log_record.getEntryKey.toByteArray)
}
@@ -1537,11 +1521,11 @@ class LevelDBClient(store: LevelDBStore) {
log_record.setCollectionKey(entry.subKey)
log_record.setEntryKey(ACK_POSITION)
log_record.setValueLocation(entry.ackPosition)
- appender.append(LOG_ADD_ENTRY, encodeEntryRecord(log_record.freeze()))
+ appender.append(LOG_UPDATE_ENTRY, encodeEntryRecord(log_record.freeze()))
val index_record = new EntryRecord.Bean()
index_record.setValueLocation(entry.ackPosition)
- batch.put(key, encodeEntryRecord(log_record.freeze()).toByteArray)
+ batch.put(key, encodeEntryRecord(index_record.freeze()).toByteArray)
}
if (uow.syncNeeded) {
@@ -1625,17 +1609,6 @@ class LevelDBClient(store: LevelDBStore) {
def gc(topicPositions:Seq[(Long, Long)]):Unit = {
- detect_if_compact_needed
-
- // Lets compact the leveldb index if it looks like we need to.
- if( index.compact_needed ) {
- debug("Compacting the leveldb index at: %s", dirtyIndexFile)
- val start = System.nanoTime()
- index.compact
- val duration = System.nanoTime() - start;
- info("Compacted the leveldb index at: %s in %.2f ms", dirtyIndexFile, (duration / 1000000.0))
- }
-
// Delete message refs for topics who's consumers have advanced..
if( !topicPositions.isEmpty ) {
might_fail_using_index {
@@ -1646,6 +1619,7 @@ class LevelDBClient(store: LevelDBStore) {
ro.verifyChecksums(verifyChecksums)
val start = encodeEntryKey(ENTRY_PREFIX, topic, 0)
val end = encodeEntryKey(ENTRY_PREFIX, topic, first)
+ debug("Topic: %d GC to seq: %d", topic, first)
index.cursorRange(start, end, ro) { case (key, value) =>
val entry = EntryRecord.FACTORY.parseUnframed(value)
batch.delete(key)
@@ -1657,8 +1631,30 @@ class LevelDBClient(store: LevelDBStore) {
}
}
+ detect_if_compact_needed
+
+ // Lets compact the leveldb index if it looks like we need to.
+ if( index.compact_needed ) {
+ val start = System.nanoTime()
+ index.compact
+ val duration = System.nanoTime() - start;
+ info("Compacted the leveldb index at: %s in %.2f ms", dirtyIndexFile, (duration / 1000000.0))
+ }
+
import collection.JavaConversions._
lastIndexSnapshotPos
+
+ // drop the logs that are no longer referenced.
+ for( (x,y) <- logRefs.toSeq ) {
+ if( y.get() <= 0 ) {
+ if( y.get() < 0 ) {
+ warn("Found a negative log reference for log: "+x)
+ }
+ debug("Log no longer referenced: %x", x)
+ logRefs.remove(x)
+ }
+ }
+
val emptyJournals = log.log_infos.keySet.toSet -- logRefs.keySet
// We don't want to delete any journals that the index has not snapshot'ed or
@@ -1669,6 +1665,7 @@ class LevelDBClient(store: LevelDBStore) {
emptyJournals.foreach { id =>
if ( id < deleteLimit ) {
+ debug("Deleting log at %x", id)
log.delete(id)
}
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/3faddcd2/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala
----------------------------------------------------------------------
diff --git a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala
index 43238a9..7b90b0c 100644
--- a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala
+++ b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala
@@ -669,6 +669,8 @@ class LevelDBStore extends LockableServiceSupport with BrokerServiceAware with P
lastSeq.set(db.getLastQueueEntrySeq(key))
+ def cursorResetPosition = 0L
+
def doAdd(uow: DelayableUOW, message: Message, delay:Boolean): CountDownFuture[AnyRef] = {
check_running
val seq = lastSeq.incrementAndGet()
@@ -731,7 +733,7 @@ class LevelDBStore extends LockableServiceSupport with BrokerServiceAware with P
def removeAllMessages(context: ConnectionContext): Unit = {
check_running
db.collectionEmpty(key)
- cursorPosition = 0
+ cursorPosition = cursorResetPosition
}
def getMessageCount: Int = {
@@ -744,11 +746,11 @@ class LevelDBStore extends LockableServiceSupport with BrokerServiceAware with P
def recover(listener: MessageRecoveryListener): Unit = {
check_running
- cursorPosition = db.cursorMessages(preparedAcks, key, listener, 0)
+ cursorPosition = db.cursorMessages(preparedAcks, key, listener, cursorResetPosition)
}
def resetBatching: Unit = {
- cursorPosition = 0
+ cursorPosition = cursorResetPosition
}
def recoverNextMessages(maxReturned: Int, listener: MessageRecoveryListener): Unit = {
@@ -789,6 +791,8 @@ class LevelDBStore extends LockableServiceSupport with BrokerServiceAware with P
val subscriptions = collection.mutable.HashMap[(String, String), DurableSubscription]()
var firstSeq = 0L
+ override def cursorResetPosition = firstSeq
+
def subscription_with_key(key:Long) = subscriptions.find(_._2.subKey == key).map(_._2)
override def asyncAddQueueMessage(context: ConnectionContext, message: Message, delay: Boolean): Future[AnyRef] = {
http://git-wip-us.apache.org/repos/asf/activemq/blob/3faddcd2/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/RecordLog.scala
----------------------------------------------------------------------
diff --git a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/RecordLog.scala b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/RecordLog.scala
index 28e1be1..08f18a7 100644
--- a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/RecordLog.scala
+++ b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/RecordLog.scala
@@ -517,42 +517,43 @@ case class RecordLog(directory: File, logSuffix:String) {
log_infos.map(_._2.position).toArray
}
- private def get_reader[T](record_position:Long)(func: (LogReader)=>T) = {
-
- val lookup = log_mutex.synchronized {
- val info = log_info(record_position)
- info.map { info=>
- if(info.position == current_appender.position) {
- current_appender.retain()
- (info, current_appender)
- } else {
- (info, null)
- }
+ private def get_reader[T](record_position:Long)(func: (LogReader)=>T):Option[T] = {
+
+ val (info, appender) = log_mutex.synchronized {
+ log_info(record_position) match {
+ case None =>
+ warn("No reader available for position: %x, log_infos: %s", record_position, log_infos)
+ return None
+ case Some(info) =>
+ if(info.position == current_appender.position) {
+ current_appender.retain()
+ (info, current_appender)
+ } else {
+ (info, null)
+ }
}
}
- lookup.map { case (info, appender) =>
- val reader = if( appender!=null ) {
- // read from the current appender.
- appender
- } else {
- // Checkout a reader from the cache...
- reader_cache.synchronized {
- var reader = reader_cache.get(info.file)
- if(reader==null) {
- reader = LogReader(info.file, info.position)
- reader_cache.put(info.file, reader)
- }
- reader.retain()
- reader
+ val reader = if( appender!=null ) {
+ // read from the current appender.
+ appender
+ } else {
+ // Checkout a reader from the cache...
+ reader_cache.synchronized {
+ var reader = reader_cache.get(info.file)
+ if(reader==null) {
+ reader = LogReader(info.file, info.position)
+ reader_cache.put(info.file, reader)
}
+ reader.retain()
+ reader
}
+ }
- try {
- func(reader)
- } finally {
- reader.release
- }
+ try {
+ Some(func(reader))
+ } finally {
+ reader.release
}
}
[17/17] git commit: Fix for
https://issues.apache.org/jira/browse/AMQ-4918
Posted by ha...@apache.org.
Fix for https://issues.apache.org/jira/browse/AMQ-4918
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/687aa926
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/687aa926
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/687aa926
Branch: refs/heads/activemq-5.9
Commit: 687aa9265ed7ca04b7e364fe1e74733cee3a2d5e
Parents: 3474f37
Author: rajdavies <ra...@gmail.com>
Authored: Thu Dec 5 08:55:56 2013 +0000
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Wed Mar 12 13:14:28 2014 -0400
----------------------------------------------------------------------
.../activemq/broker/TransportConnection.java | 2 +-
.../activemq/broker/jmx/NetworkBridgeView.java | 1 +
.../broker/jmx/NetworkDestinationView.java | 6 ++
.../network/DemandForwardingBridgeSupport.java | 5 ++
.../network/DiscoveryNetworkConnector.java | 2 +-
.../network/MBeanBridgeDestination.java | 68 ++++++++++++++++++--
.../activemq/network/MBeanNetworkListener.java | 16 +++--
.../apache/activemq/network/NetworkBridge.java | 2 +
.../network/NetworkBridgeConfiguration.java | 25 ++++++-
.../org/apache/activemq/bugs/AMQ4160Test.java | 4 ++
10 files changed, 116 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/687aa926/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java b/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java
index 1ce75a5..997b323 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java
@@ -1297,7 +1297,7 @@ public class TransportConnection implements Connection, Task, CommandVisitor {
if (duplexName.contains("#")) {
duplexName = duplexName.substring(duplexName.lastIndexOf("#"));
}
- MBeanNetworkListener listener = new MBeanNetworkListener(broker.getBrokerService(), broker.getBrokerService().createDuplexNetworkConnectorObjectName(duplexName));
+ MBeanNetworkListener listener = new MBeanNetworkListener(broker.getBrokerService(), config, broker.getBrokerService().createDuplexNetworkConnectorObjectName(duplexName));
listener.setCreatedByDuplex(true);
duplexBridge = NetworkBridgeFactory.createBridge(config, localTransport, remoteBridgeTransport, listener);
duplexBridge.setBrokerService(broker.getBrokerService());
http://git-wip-us.apache.org/repos/asf/activemq/blob/687aa926/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java
index 795de3a..47f167a 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java
@@ -72,6 +72,7 @@ public class NetworkBridgeView implements NetworkBridgeViewMBean {
}
public void resetStats(){
+ bridge.resetStats();
for (NetworkDestinationView networkDestinationView:networkDestinationViewList){
networkDestinationView.resetStats();
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/687aa926/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkDestinationView.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkDestinationView.java b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkDestinationView.java
index 26177d5..1dceee4 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkDestinationView.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkDestinationView.java
@@ -73,7 +73,13 @@ public class NetworkDestinationView implements NetworkDestinationViewMBean {
lastTime=currentTime;
}
+ public long getLastAccessTime(){
+ return timeStatistic.getLastSampleTime();
+ }
+
public void close(){
networkBridgeView.removeNetworkDestinationView(this);
}
+
+
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/687aa926/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java b/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
index bf0b4f6..9fa38a4 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
@@ -1538,6 +1538,11 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
return mbeanObjectName;
}
+ public void resetStats(){
+ enqueueCounter.set(0);
+ dequeueCounter.set(0);
+ }
+
/*
* Used to allow for async tasks to await receipt of the BrokerInfo from the local and
* remote sides of the network bridge.
http://git-wip-us.apache.org/repos/asf/activemq/blob/687aa926/activemq-broker/src/main/java/org/apache/activemq/network/DiscoveryNetworkConnector.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/DiscoveryNetworkConnector.java b/activemq-broker/src/main/java/org/apache/activemq/network/DiscoveryNetworkConnector.java
index ed71b7b..d9cb0c6 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/network/DiscoveryNetworkConnector.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/network/DiscoveryNetworkConnector.java
@@ -231,7 +231,7 @@ public class DiscoveryNetworkConnector extends NetworkConnector implements Disco
class DiscoverNetworkBridgeListener extends MBeanNetworkListener {
public DiscoverNetworkBridgeListener(BrokerService brokerService, ObjectName connectorName) {
- super(brokerService, connectorName);
+ super(brokerService, DiscoveryNetworkConnector.this, connectorName);
}
@Override
http://git-wip-us.apache.org/repos/asf/activemq/blob/687aa926/activemq-broker/src/main/java/org/apache/activemq/network/MBeanBridgeDestination.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/MBeanBridgeDestination.java b/activemq-broker/src/main/java/org/apache/activemq/network/MBeanBridgeDestination.java
index c718063..583fab7 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/network/MBeanBridgeDestination.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/network/MBeanBridgeDestination.java
@@ -16,6 +16,7 @@
*/
package org.apache.activemq.network;
+import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@@ -27,6 +28,7 @@ import org.apache.activemq.broker.jmx.NetworkBridgeView;
import org.apache.activemq.broker.jmx.NetworkDestinationView;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.Message;
+import org.apache.activemq.thread.Scheduler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -35,14 +37,24 @@ public class MBeanBridgeDestination {
private final BrokerService brokerService;
private final NetworkBridge bridge;
private final NetworkBridgeView networkBridgeView;
+ private final NetworkBridgeConfiguration networkBridgeConfiguration;
+ private final Scheduler scheduler;
+ private final Runnable purgeInactiveDestinationViewTask;
private Map<ActiveMQDestination, ObjectName> destinationObjectNameMap = new ConcurrentHashMap<ActiveMQDestination, ObjectName>();
private Map<ActiveMQDestination, NetworkDestinationView> outboundDestinationViewMap = new ConcurrentHashMap<ActiveMQDestination, NetworkDestinationView>();
private Map<ActiveMQDestination, NetworkDestinationView> inboundDestinationViewMap = new ConcurrentHashMap<ActiveMQDestination, NetworkDestinationView>();
- public MBeanBridgeDestination(BrokerService brokerService, NetworkBridge bridge, NetworkBridgeView networkBridgeView) {
+ public MBeanBridgeDestination(BrokerService brokerService, NetworkBridgeConfiguration networkBridgeConfiguration, NetworkBridge bridge, NetworkBridgeView networkBridgeView) {
this.brokerService = brokerService;
+ this.networkBridgeConfiguration = networkBridgeConfiguration;
this.bridge = bridge;
this.networkBridgeView = networkBridgeView;
+ this.scheduler = brokerService.getScheduler();
+ purgeInactiveDestinationViewTask = new Runnable() {
+ public void run() {
+ purgeInactiveDestinationViews();
+ }
+ };
}
@@ -55,7 +67,7 @@ public class MBeanBridgeDestination {
ObjectName bridgeObjectName = bridge.getMbeanObjectName();
try {
ObjectName objectName = BrokerMBeanSupport.createNetworkOutBoundDestinationObjectName(bridgeObjectName, destination);
- networkDestinationView = new NetworkDestinationView(networkBridgeView,destination.getPhysicalName());
+ networkDestinationView = new NetworkDestinationView(networkBridgeView, destination.getPhysicalName());
AnnotatedMBean.registerMBean(brokerService.getManagementContext(), networkDestinationView, objectName);
destinationObjectNameMap.put(destination, objectName);
outboundDestinationViewMap.put(destination, networkDestinationView);
@@ -79,7 +91,7 @@ public class MBeanBridgeDestination {
ObjectName bridgeObjectName = bridge.getMbeanObjectName();
try {
ObjectName objectName = BrokerMBeanSupport.createNetworkInBoundDestinationObjectName(bridgeObjectName, destination);
- networkDestinationView= new NetworkDestinationView(networkBridgeView,destination.getPhysicalName());
+ networkDestinationView = new NetworkDestinationView(networkBridgeView, destination.getPhysicalName());
networkBridgeView.addNetworkDestinationView(networkDestinationView);
AnnotatedMBean.registerMBean(brokerService.getManagementContext(), networkDestinationView, objectName);
destinationObjectNameMap.put(destination, objectName);
@@ -93,11 +105,21 @@ public class MBeanBridgeDestination {
networkDestinationView.messageSent();
}
- public void close() {
+ public void start() {
+ if (networkBridgeConfiguration.isGcDestinationViews()) {
+ long period = networkBridgeConfiguration.getGcSweepTime();
+ if (period > 0) {
+ scheduler.executePeriodically(purgeInactiveDestinationViewTask, period);
+ }
+ }
+ }
+
+ public void stop() {
if (!brokerService.isUseJmx()) {
return;
}
+ scheduler.cancel(purgeInactiveDestinationViewTask);
for (ObjectName objectName : destinationObjectNameMap.values()) {
try {
if (objectName != null) {
@@ -112,4 +134,42 @@ public class MBeanBridgeDestination {
inboundDestinationViewMap.clear();
}
+ private void purgeInactiveDestinationViews() {
+ if (!brokerService.isUseJmx()) {
+ return;
+ }
+ purgeInactiveDestinationView(inboundDestinationViewMap);
+ purgeInactiveDestinationView(outboundDestinationViewMap);
+ }
+
+ private void purgeInactiveDestinationView(Map<ActiveMQDestination, NetworkDestinationView> map) {
+ long time = System.currentTimeMillis() - networkBridgeConfiguration.getGcSweepTime();
+ Map<ActiveMQDestination, NetworkDestinationView> gc = null;
+ for (Map.Entry<ActiveMQDestination, NetworkDestinationView> entry : map.entrySet()) {
+ if (entry.getValue().getLastAccessTime() <= time) {
+ if (gc == null) {
+ gc = new HashMap<ActiveMQDestination, NetworkDestinationView>();
+ }
+ gc.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ if (gc != null) {
+ for (Map.Entry<ActiveMQDestination, NetworkDestinationView> entry : gc.entrySet()) {
+ map.remove(entry.getKey());
+ ObjectName objectName = destinationObjectNameMap.get(entry.getKey());
+ if (objectName != null) {
+ try {
+ if (objectName != null) {
+ brokerService.getManagementContext().unregisterMBean(objectName);
+ }
+ } catch (Throwable e) {
+ LOG.debug("Network bridge could not be unregistered in JMX: {}", e.getMessage(), e);
+ }
+ }
+ entry.getValue().close();
+ }
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/687aa926/activemq-broker/src/main/java/org/apache/activemq/network/MBeanNetworkListener.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/MBeanNetworkListener.java b/activemq-broker/src/main/java/org/apache/activemq/network/MBeanNetworkListener.java
index c877ecd..bf8facf 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/network/MBeanNetworkListener.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/network/MBeanNetworkListener.java
@@ -33,12 +33,15 @@ public class MBeanNetworkListener implements NetworkBridgeListener {
private static final Logger LOG = LoggerFactory.getLogger(MBeanNetworkListener.class);
- BrokerService brokerService;
- ObjectName connectorName;
- boolean createdByDuplex = false;
+ private final BrokerService brokerService;
+ private final ObjectName connectorName;
+ private final NetworkBridgeConfiguration networkBridgeConfiguration;
+ private boolean createdByDuplex = false;
private Map<NetworkBridge,MBeanBridgeDestination> destinationObjectNameMap = new ConcurrentHashMap<NetworkBridge,MBeanBridgeDestination>();
- public MBeanNetworkListener(BrokerService brokerService, ObjectName connectorName) {
+
+ public MBeanNetworkListener(BrokerService brokerService, NetworkBridgeConfiguration networkBridgeConfiguration, ObjectName connectorName) {
this.brokerService = brokerService;
+ this.networkBridgeConfiguration = networkBridgeConfiguration;
this.connectorName = connectorName;
}
@@ -57,8 +60,9 @@ public class MBeanNetworkListener implements NetworkBridgeListener {
ObjectName objectName = createNetworkBridgeObjectName(bridge);
AnnotatedMBean.registerMBean(brokerService.getManagementContext(), view, objectName);
bridge.setMbeanObjectName(objectName);
- MBeanBridgeDestination mBeanBridgeDestination = new MBeanBridgeDestination(brokerService,bridge,view);
+ MBeanBridgeDestination mBeanBridgeDestination = new MBeanBridgeDestination(brokerService,networkBridgeConfiguration,bridge,view);
destinationObjectNameMap.put(bridge,mBeanBridgeDestination);
+ mBeanBridgeDestination.start();
LOG.debug("registered: {} as: {}", bridge, objectName);
} catch (Throwable e) {
LOG.debug("Network bridge could not be registered in JMX: {}", e.getMessage(), e);
@@ -77,7 +81,7 @@ public class MBeanNetworkListener implements NetworkBridgeListener {
}
MBeanBridgeDestination mBeanBridgeDestination = destinationObjectNameMap.remove(bridge);
if (mBeanBridgeDestination != null){
- mBeanBridgeDestination.close();
+ mBeanBridgeDestination.stop();
}
} catch (Throwable e) {
LOG.debug("Network bridge could not be unregistered in JMX: {}", e.getMessage(), e);
http://git-wip-us.apache.org/repos/asf/activemq/blob/687aa926/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridge.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridge.java b/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridge.java
index 9ba3c9f..95d0477 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridge.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridge.java
@@ -83,4 +83,6 @@ public interface NetworkBridge extends Service {
* @return the MBean name used to identify this bridge in the MBean server.
*/
ObjectName getMbeanObjectName();
+
+ void resetStats();
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/687aa926/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeConfiguration.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeConfiguration.java b/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeConfiguration.java
index 597625c..bfff94e 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeConfiguration.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeConfiguration.java
@@ -16,13 +16,13 @@
*/
package org.apache.activemq.network;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
import org.apache.activemq.advisory.AdvisorySupport;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ConsumerInfo;
-import java.util.List;
-import java.util.concurrent.CopyOnWriteArrayList;
-
/**
* Configuration for a NetworkBridge
*/
@@ -59,6 +59,8 @@ public class NetworkBridgeConfiguration {
private boolean useCompression = false;
private boolean advisoryForFailedForward = false;
private boolean useBrokerNamesAsIdSeed = true;
+ private boolean gcDestinationViews = true;
+ private long gcSweepTime = 60 * 1000;
/**
* @return the conduitSubscriptions
@@ -421,4 +423,21 @@ public class NetworkBridgeConfiguration {
public void setUseBrokerNameAsIdSees(boolean val) {
useBrokerNamesAsIdSeed = val;
}
+
+ public boolean isGcDestinationViews() {
+ return gcDestinationViews;
+ }
+
+ public void setGcDestinationViews(boolean gcDestinationViews) {
+ this.gcDestinationViews = gcDestinationViews;
+ }
+
+ public long getGcSweepTime() {
+ return gcSweepTime;
+ }
+
+ public void setGcSweepTime(long gcSweepTime) {
+ this.gcSweepTime = gcSweepTime;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/687aa926/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4160Test.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4160Test.java b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4160Test.java
index c1dadd9..34bff2d 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4160Test.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4160Test.java
@@ -331,6 +331,10 @@ public class AMQ4160Test extends JmsMultipleBrokersTestSupport {
public ObjectName getMbeanObjectName() {
return next.getMbeanObjectName();
}
+
+ public void resetStats(){
+ next.resetStats();
+ }
};
}
};
[09/17] git commit: Adds several test cases which show some JMS
client issues, most are resolved by moving on to v 0.26-SNAPSHOT
Posted by ha...@apache.org.
Adds several test cases which show some JMS client issues, most are
resolved by moving on to v 0.26-SNAPSHOT
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/71b55233
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/71b55233
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/71b55233
Branch: refs/heads/activemq-5.9
Commit: 71b55233248b006e51b645f01b25c19e5277fc31
Parents: 0bf2233
Author: Timothy Bish <ta...@gmai.com>
Authored: Mon Dec 2 15:07:13 2013 -0500
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Wed Mar 12 13:12:37 2014 -0400
----------------------------------------------------------------------
.../activemq/transport/amqp/JMSClientTest.java | 135 +++++++++++++++++++
1 file changed, 135 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/71b55233/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientTest.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientTest.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientTest.java
index 16fde4e..e35127d 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientTest.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientTest.java
@@ -18,9 +18,12 @@ package org.apache.activemq.transport.amqp;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
import java.util.Enumeration;
+import java.util.concurrent.TimeUnit;
import javax.jms.Connection;
import javax.jms.DeliveryMode;
@@ -229,6 +232,138 @@ public class JMSClientTest extends AmqpTestSupport {
connection.close();
}
+ //should through exception IllegalStateException:The session is closed
+ @Test(timeout=30000)
+ public void testBrokerRestartPersistentQueueException() throws Exception {
+ QueueImpl queue = new QueueImpl("queue://" + name);
+
+ Connection connection = createConnection();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ connection.start();
+
+ MessageProducer producer = session.createProducer(queue);
+ producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+
+ Message m = session.createTextMessage("Sample text");
+ producer.send(m);
+
+ restartBroker();
+
+ try {
+ session.createConsumer(queue);
+ fail("Should have thrown an IllegalStateException");
+ } catch (Exception ex) {
+ LOG.info("Caught exception on receive: {}", ex);
+ }
+ }
+
+ @Test(timeout=30000)
+ public void testProducerThrowsWhenBrokerRestarted() throws Exception {
+ QueueImpl queue = new QueueImpl("queue://" + name);
+
+ Connection connection = createConnection();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ connection.start();
+
+ MessageProducer producer = session.createProducer(queue);
+ producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+
+ Message m = session.createTextMessage("Sample text");
+
+ Thread restart = new Thread(new Runnable() {
+
+ @Override
+ public void run() {
+ try {
+ TimeUnit.SECONDS.sleep(5);
+ restartBroker();
+ } catch (Exception ex) {}
+ }
+ });
+ restart.start();
+
+ try {
+ for (int i = 0; i < 10; ++i) {
+ producer.send(m);
+ TimeUnit.SECONDS.sleep(1);
+ }
+ fail("Should have thrown an IllegalStateException");
+ } catch (Exception ex) {
+ LOG.info("Caught exception on send: {}", ex);
+ }
+ }
+
+ @Test(timeout=30000)
+ public void testBrokerRestartWontHangConnectionClose() throws Exception {
+ QueueImpl queue = new QueueImpl("queue://" + name);
+
+ Connection connection = createConnection();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ connection.start();
+
+ MessageProducer producer = session.createProducer(queue);
+ producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+
+ Message m = session.createTextMessage("Sample text");
+ producer.send(m);
+
+ restartBroker();
+
+ try {
+ connection.close();
+ } catch (Exception ex) {
+ LOG.error("Should not thrown on disconnected connection close(): {}", ex);
+ fail("Should not have thrown an exception.");
+ }
+ }
+
+ @Test(timeout=120000)
+ public void testProduceAndConsumeLargeNumbersOfMessages() throws JMSException {
+
+ int count = 2000;
+
+ QueueImpl queue = new QueueImpl("queue://" + name);
+ Connection connection = createConnection();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ connection.start();
+
+ MessageProducer producer= session.createProducer(queue);
+ for (int i = 0; i < count; i++) {
+ Message m=session.createTextMessage("Test-Message:"+i);
+ producer.send(m);
+ }
+
+ MessageConsumer consumer=session.createConsumer(queue);
+ for(int i = 0; i < count; i++) {
+ Message message = consumer.receive(5000);
+ assertNotNull(message);
+ System.out.println(((TextMessage) message).getText());
+ assertEquals("Test-Message:" + i,((TextMessage) message).getText());
+ }
+
+ Message message = consumer.receive(5000);
+ assertNull(message);
+ }
+
+ @Test(timeout=30000)
+ public void testTTL() throws Exception {
+ QueueImpl queue = new QueueImpl("queue://" + name);
+ Connection connection = createConnection();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ connection.start();
+ MessageProducer producer = session.createProducer(queue);
+ producer.setTimeToLive(1000);
+ Message toSend = session.createTextMessage("Sample text");
+ producer.send(toSend);
+ MessageConsumer consumer = session.createConsumer(queue);
+ Message received = consumer.receive(5000);
+ assertNotNull(received);
+ producer.setTimeToLive(100);
+ producer.send(toSend);
+ TimeUnit.SECONDS.sleep(1);
+ assertNull(consumer.receive(5000));
+ }
+
private Connection createConnection() throws JMSException {
final ConnectionFactoryImpl factory = new ConnectionFactoryImpl("localhost", port, "admin", "password");
final Connection connection = factory.createConnection();
[07/17] git commit: Test to reproduce AMQ-4914 with the failing part
@Ignored for now
Posted by ha...@apache.org.
Test to reproduce AMQ-4914 with the failing part @Ignored for now
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/30352d01
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/30352d01
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/30352d01
Branch: refs/heads/activemq-5.9
Commit: 30352d0135e18c0d7f349068136c97e8e93616db
Parents: ead3d16
Author: Kevin Earls <ke...@kevinearls.com>
Authored: Mon Dec 2 17:52:30 2013 +0100
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Wed Mar 12 13:12:07 2014 -0400
----------------------------------------------------------------------
.../transport/amqp/bugs/AMQ4914Test.java | 119 +++++++++++++++++++
1 file changed, 119 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/30352d01/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/bugs/AMQ4914Test.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/bugs/AMQ4914Test.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/bugs/AMQ4914Test.java
new file mode 100644
index 0000000..1dde725
--- /dev/null
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/bugs/AMQ4914Test.java
@@ -0,0 +1,119 @@
+/**
+ * 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.transport.amqp.bugs;
+
+import org.apache.activemq.transport.amqp.AmqpTestSupport;
+import org.apache.qpid.amqp_1_0.jms.impl.ConnectionFactoryImpl;
+import org.apache.qpid.amqp_1_0.jms.impl.QueueImpl;
+import org.junit.Ignore;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TestName;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.jms.Connection;
+import javax.jms.ExceptionListener;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+
+import static org.junit.Assert.*;
+
+public class AMQ4914Test extends AmqpTestSupport {
+ @Rule
+ public TestName testName = new TestName();
+
+ protected static final Logger LOG = LoggerFactory.getLogger(AMQ4914Test.class);
+ private final static String QUEUE_NAME="queue://ENTMQ476TestQueue";
+
+ /**
+ *
+ * @param sizeInBytes
+ * @return
+ */
+ private String createLargeString(int sizeInBytes) {
+ byte[] base = {1, 2, 3, 4, 5, 6, 7, 8, 9, 0};
+ StringBuilder builder = new StringBuilder();
+ for (int i = 0; i < sizeInBytes; i++) {
+ builder.append(base[i % base.length]);
+ }
+
+ LOG.debug("Created string with size : " + builder.toString().getBytes().length + " bytes");
+ return builder.toString();
+ }
+
+ @Test(timeout = 2 * 60 * 1000)
+ public void testSendSmallerMessages() throws JMSException {
+ for (int i = 512; i <= (16 * 1024); i += 512) {
+ doTestSendLargeMessage(i);
+ }
+ }
+
+ @Ignore("AMQ-4914")
+ @Test(timeout = 2 * 60 * 1000)
+ public void testSendLargeMessages() throws JMSException {
+ //for (int i = 32000; i < (32 *1024); i++) {
+ doTestSendLargeMessage(32604); // Fails at 32614; or 32604 with my changes to AmqpProtocolBuffer
+ //}
+ }
+
+ public void doTestSendLargeMessage(int expectedSize) throws JMSException{
+ LOG.info("doTestSendLargeMessage called with expectedSize " + expectedSize);
+ String payload = createLargeString(expectedSize);
+ assertEquals(expectedSize, payload.getBytes().length);
+
+ Connection connection = createAMQPConnection(port, false);
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ QueueImpl queue = new QueueImpl(QUEUE_NAME);
+ MessageProducer producer = session.createProducer(queue);
+ TextMessage message = session.createTextMessage();
+ message.setText(payload);
+ producer.send(message);
+ LOG.debug("Returned from send");
+
+ MessageConsumer consumer = session.createConsumer(queue);
+ connection.start();
+ LOG.debug("Calling receive");
+ Message receivedMessage = consumer.receive();
+ assertNotNull(receivedMessage);
+ assertTrue(receivedMessage instanceof TextMessage);
+ TextMessage receivedTextMessage = (TextMessage) receivedMessage;
+ String receivedText = receivedTextMessage.getText();
+ assertEquals(expectedSize, receivedText.getBytes().length);
+ assertEquals(payload, receivedText);
+ connection.close();
+ }
+
+ private Connection createAMQPConnection(int testPort, boolean useSSL) throws JMSException {
+ LOG.debug("In createConnection using port {} ssl? {}", testPort, useSSL);
+ final ConnectionFactoryImpl connectionFactory = new ConnectionFactoryImpl("localhost", testPort, "admin", "password", null, useSSL);
+ final Connection connection = connectionFactory.createConnection();
+ connection.setExceptionListener(new ExceptionListener() {
+ @Override
+ public void onException(JMSException exception) {
+ exception.printStackTrace();
+ }
+ });
+ connection.start();
+ return connection;
+ }
+}
[10/17] git commit: Adding assertions to make sure that we only
append to the log from the write thread. Found a code path that was appending
to the log from a different thread. This might have been affecting
https://issues.apache.org/jira/browse/AMQ-4
Posted by ha...@apache.org.
Adding assertions to make sure that we only append to the log from the write thread. Found a code path that was appending to the log from a different thread. This might have been affecting https://issues.apache.org/jira/browse/AMQ-4882
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/2e5c9d59
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/2e5c9d59
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/2e5c9d59
Branch: refs/heads/activemq-5.9
Commit: 2e5c9d5929ffc7a1a832c831faf2467b410f98f3
Parents: 71b5523
Author: Hiram Chirino <hi...@hiramchirino.com>
Authored: Tue Dec 3 12:04:40 2013 -0500
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Wed Mar 12 13:12:50 2014 -0400
----------------------------------------------------------------------
.../org/apache/activemq/leveldb/DBManager.scala | 2 +-
.../apache/activemq/leveldb/LevelDBClient.scala | 18 +++++++++++++-----
2 files changed, 14 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/2e5c9d59/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala
----------------------------------------------------------------------
diff --git a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala
index 00260d9..6b575ee 100644
--- a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala
+++ b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala
@@ -798,7 +798,7 @@ class DBManager(val parent:LevelDBStore) {
def createTransactionContainer(id:XATransactionId) =
createCollection(buffer(parent.wireFormat.marshal(id)), TRANSACTION_COLLECTION_TYPE)
- def removeTransactionContainer(key:Long) = { // writeExecutor.sync {
+ def removeTransactionContainer(key:Long) = writeExecutor.sync {
client.removeCollection(key)
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/2e5c9d59/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBClient.scala
----------------------------------------------------------------------
diff --git a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBClient.scala b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBClient.scala
index c0cedce..fe29012 100755
--- a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBClient.scala
+++ b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBClient.scala
@@ -59,6 +59,10 @@ import org.apache.activemq.broker.SuppressReplyException
*/
object LevelDBClient extends Log {
+ class WriteThread(r:Runnable) extends Thread(r) {
+ setDaemon(true)
+ }
+
final val STORE_SCHEMA_PREFIX = "activemq_leveldb_store:"
final val STORE_SCHEMA_VERSION = 1
@@ -512,6 +516,7 @@ class LevelDBClient(store: LevelDBStore) {
}
def storeTrace(ascii:String, force:Boolean=false) = {
+ assert_write_thread_executing
val time = new SimpleDateFormat("dd/MMM/yyyy:HH:mm::ss Z").format(new Date)
log.appender { appender =>
appender.append(LOG_TRACE, new AsciiBuffer("%s: %s".format(time, ascii)))
@@ -566,6 +571,8 @@ class LevelDBClient(store: LevelDBStore) {
replay_write_batch = null;
}
+ def assert_write_thread_executing = assert(Thread.currentThread().getClass == classOf[WriteThread])
+
def init() ={
// Lets check store compatibility...
@@ -590,11 +597,7 @@ class LevelDBClient(store: LevelDBStore) {
version_file.writeText(STORE_SCHEMA_PREFIX + STORE_SCHEMA_VERSION)
writeExecutor = Executors.newFixedThreadPool(1, new ThreadFactory() {
- def newThread(r: Runnable) = {
- val rc = new Thread(r, "LevelDB store io write")
- rc.setDaemon(true)
- rc
- }
+ def newThread(r: Runnable) = new WriteThread(r)
})
val factoryNames = store.indexFactory
@@ -1125,6 +1128,8 @@ class LevelDBClient(store: LevelDBStore) {
}
def addCollection(record: CollectionRecord.Buffer) = {
+ assert_write_thread_executing
+
val key = encodeLongKey(COLLECTION_PREFIX, record.getKey)
val value = record.toUnframedBuffer
might_fail_using_index {
@@ -1153,6 +1158,7 @@ class LevelDBClient(store: LevelDBStore) {
}
def removeCollection(collectionKey: Long) = {
+ assert_write_thread_executing
val key = encodeLongKey(COLLECTION_PREFIX, collectionKey)
val value = encodeVLong(collectionKey)
val entryKeyPrefix = encodeLongKey(ENTRY_PREFIX, collectionKey)
@@ -1181,6 +1187,7 @@ class LevelDBClient(store: LevelDBStore) {
}
def collectionEmpty(collectionKey: Long) = {
+ assert_write_thread_executing
val key = encodeLongKey(COLLECTION_PREFIX, collectionKey)
val value = encodeVLong(collectionKey)
val entryKeyPrefix = encodeLongKey(ENTRY_PREFIX, collectionKey)
@@ -1366,6 +1373,7 @@ class LevelDBClient(store: LevelDBStore) {
val max_index_write_latency = TimeMetric()
def store(uows: Array[DelayableUOW]) {
+ assert_write_thread_executing
might_fail_using_index {
log.appender { appender =>
val syncNeeded = index.write(new WriteOptions, max_index_write_latency) { batch =>
[14/17] git commit: Fixing bug where the DataByteArrayInputStream was
not throwing EOFExceptions when reads were going past the end of the byte
array.
Posted by ha...@apache.org.
Fixing bug where the DataByteArrayInputStream was not throwing EOFExceptions when reads were going past the end of the byte array.
This would make it harder to figure out when things went wrong.
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/37e1fd76
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/37e1fd76
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/37e1fd76
Branch: refs/heads/activemq-5.9
Commit: 37e1fd760fa55d624f128bc61923a3568c124b17
Parents: 3faddcd
Author: Hiram Chirino <hi...@hiramchirino.com>
Authored: Wed Dec 4 12:08:28 2013 -0500
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Wed Mar 12 13:13:46 2014 -0400
----------------------------------------------------------------------
.../activemq/command/XATransactionId.java | 4 +-
.../activemq/util/DataByteArrayInputStream.java | 58 +++++++++++---------
2 files changed, 35 insertions(+), 27 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/37e1fd76/activemq-client/src/main/java/org/apache/activemq/command/XATransactionId.java
----------------------------------------------------------------------
diff --git a/activemq-client/src/main/java/org/apache/activemq/command/XATransactionId.java b/activemq-client/src/main/java/org/apache/activemq/command/XATransactionId.java
index e963c87..5f786e5 100755
--- a/activemq-client/src/main/java/org/apache/activemq/command/XATransactionId.java
+++ b/activemq-client/src/main/java/org/apache/activemq/command/XATransactionId.java
@@ -50,7 +50,7 @@ public class XATransactionId extends TransactionId implements Xid, Comparable {
this.branchQualifier = xid.getBranchQualifier();
}
- public XATransactionId(byte[] encodedBytes) {
+ public XATransactionId(byte[] encodedBytes) throws IOException {
encodedXidBytes = encodedBytes;
initFromEncodedBytes();
}
@@ -61,7 +61,7 @@ public class XATransactionId extends TransactionId implements Xid, Comparable {
final int XID_PREFIX_SIZE = 16;
//+|-,(long)lastAck,(byte)priority,(int)formatid,(short)globalLength....
- private void initFromEncodedBytes() {
+ private void initFromEncodedBytes() throws IOException {
DataByteArrayInputStream inputStream = new DataByteArrayInputStream(encodedXidBytes);
inputStream.skipBytes(10);
formatId = inputStream.readInt();
http://git-wip-us.apache.org/repos/asf/activemq/blob/37e1fd76/activemq-client/src/main/java/org/apache/activemq/util/DataByteArrayInputStream.java
----------------------------------------------------------------------
diff --git a/activemq-client/src/main/java/org/apache/activemq/util/DataByteArrayInputStream.java b/activemq-client/src/main/java/org/apache/activemq/util/DataByteArrayInputStream.java
index 9bf229d..0a13aa0 100755
--- a/activemq-client/src/main/java/org/apache/activemq/util/DataByteArrayInputStream.java
+++ b/activemq-client/src/main/java/org/apache/activemq/util/DataByteArrayInputStream.java
@@ -16,10 +16,7 @@
*/
package org.apache.activemq.util;
-import java.io.DataInput;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.UTFDataFormatException;
+import java.io.*;
/**
* Optimized ByteArrayInputStream that can be used more than once
@@ -124,6 +121,14 @@ public final class DataByteArrayInputStream extends InputStream implements DataI
return (pos < buf.length) ? (buf[pos++] & 0xff) : -1;
}
+ public int readOrIOException() throws IOException {
+ int rc = read();
+ if( rc == -1 ) {
+ throw new EOFException();
+ }
+ return rc;
+ }
+
/**
* Reads up to <code>len</code> bytes of data into an array of bytes from
* this input stream.
@@ -180,45 +185,48 @@ public final class DataByteArrayInputStream extends InputStream implements DataI
return n;
}
- public boolean readBoolean() {
- return read() != 0;
+ public boolean readBoolean() throws IOException {
+ return readOrIOException() != 0;
}
- public byte readByte() {
- return (byte)read();
+ public byte readByte() throws IOException {
+ return (byte)readOrIOException();
}
- public int readUnsignedByte() {
- return read();
+ public int readUnsignedByte() throws IOException {
+ return readOrIOException();
}
- public short readShort() {
- int ch1 = read();
- int ch2 = read();
+ public short readShort() throws IOException {
+ int ch1 = readOrIOException();
+ int ch2 = readOrIOException();
return (short)((ch1 << 8) + (ch2 << 0));
}
- public int readUnsignedShort() {
- int ch1 = read();
- int ch2 = read();
+ public int readUnsignedShort() throws IOException {
+ int ch1 = readOrIOException();
+ int ch2 = readOrIOException();
return (ch1 << 8) + (ch2 << 0);
}
- public char readChar() {
- int ch1 = read();
- int ch2 = read();
+ public char readChar() throws IOException {
+ int ch1 = readOrIOException();
+ int ch2 = readOrIOException();
return (char)((ch1 << 8) + (ch2 << 0));
}
- public int readInt() {
- int ch1 = read();
- int ch2 = read();
- int ch3 = read();
- int ch4 = read();
+ public int readInt() throws IOException {
+ int ch1 = readOrIOException();
+ int ch2 = readOrIOException();
+ int ch3 = readOrIOException();
+ int ch4 = readOrIOException();
return (ch1 << 24) + (ch2 << 16) + (ch3 << 8) + (ch4 << 0);
}
- public long readLong() {
+ public long readLong() throws IOException {
+ if (pos >= buf.length ) {
+ throw new EOFException();
+ }
long rc = ((long)buf[pos++] << 56) + ((long)(buf[pos++] & 255) << 48) + ((long)(buf[pos++] & 255) << 40) + ((long)(buf[pos++] & 255) << 32);
return rc + ((long)(buf[pos++] & 255) << 24) + ((buf[pos++] & 255) << 16) + ((buf[pos++] & 255) << 8) + ((buf[pos++] & 255) << 0);
}
[04/17] git commit: Add some methods for stoppng and restarting
broker during tests.
Posted by ha...@apache.org.
Add some methods for stoppng and restarting broker during tests.
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/4aae0328
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/4aae0328
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/4aae0328
Branch: refs/heads/activemq-5.9
Commit: 4aae0328989aba205371d3f45ee4e3aad6d449fb
Parents: 3444c89
Author: Timothy Bish <ta...@gmai.com>
Authored: Wed Nov 27 14:02:05 2013 -0500
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Wed Mar 12 13:11:04 2014 -0400
----------------------------------------------------------------------
.../transport/amqp/AmqpTestSupport.java | 30 +++++++++++++++++---
1 file changed, 26 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/4aae0328/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpTestSupport.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpTestSupport.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpTestSupport.java
index 5c5bcff..91eb6b2 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpTestSupport.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpTestSupport.java
@@ -70,12 +70,14 @@ public class AmqpTestSupport {
autoFailTestSupport.startAutoFailThread();
exceptions.clear();
startBroker();
+ this.numberOfMessages = 2000;
}
- public void startBroker() throws Exception {
+ private void createBroker(boolean deleteAllMessages) throws Exception {
brokerService = new BrokerService();
brokerService.setPersistent(false);
brokerService.setAdvisorySupport(false);
+ brokerService.setDeleteAllMessagesOnStartup(deleteAllMessages);
SSLContext ctx = SSLContext.getInstance("TLS");
ctx.init(new KeyManager[0], new TrustManager[]{new DefaultTrustManager()}, new SecureRandom());
@@ -93,8 +95,6 @@ public class AmqpTestSupport {
brokerService.setSslContext(sslContext);
addAMQPConnector();
- brokerService.start();
- this.numberOfMessages = 2000;
}
protected void addAMQPConnector() throws Exception {
@@ -112,12 +112,34 @@ public class AmqpTestSupport {
LOG.debug("Using amqp+nio+ssl port " + nioPlusSslPort);
}
- @After
+ public void startBroker() throws Exception {
+ if (brokerService != null) {
+ throw new IllegalStateException("Broker is already created.");
+ }
+
+ createBroker(true);
+ brokerService.start();
+ brokerService.waitUntilStarted();
+ }
+
+ public void restartBroker() throws Exception {
+ stopBroker();
+ createBroker(false);
+ brokerService.start();
+ brokerService.waitUntilStarted();
+ }
+
public void stopBroker() throws Exception {
if (brokerService != null) {
brokerService.stop();
+ brokerService.waitUntilStopped();
brokerService = null;
}
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ stopBroker();
autoFailTestSupport.stopAutoFailThread();
}
[12/17] git commit: https://issues.apache.org/jira/browse/AMQ-4889
Posted by ha...@apache.org.
https://issues.apache.org/jira/browse/AMQ-4889
Improve the stream close logic in the init method to ensure we don't
leak and resources.
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/720b8ace
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/720b8ace
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/720b8ace
Branch: refs/heads/activemq-5.9
Commit: 720b8aceca0bf25b150e8f80c6f22e1c365f7fae
Parents: f6ed548
Author: Timothy Bish <ta...@gmai.com>
Authored: Wed Dec 4 11:09:28 2013 -0500
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Wed Mar 12 13:13:20 2014 -0400
----------------------------------------------------------------------
.../activemq/transport/nio/NIOSSLTransport.java | 15 ++++++++++++---
1 file changed, 12 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/720b8ace/activemq-client/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransport.java
----------------------------------------------------------------------
diff --git a/activemq-client/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransport.java b/activemq-client/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransport.java
index 52c3e97..02789f3 100644
--- a/activemq-client/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransport.java
+++ b/activemq-client/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransport.java
@@ -34,7 +34,6 @@ import javax.net.ssl.SSLEngineResult;
import javax.net.ssl.SSLPeerUnverifiedException;
import javax.net.ssl.SSLSession;
-import org.apache.activemq.command.Command;
import org.apache.activemq.command.ConnectionInfo;
import org.apache.activemq.openwire.OpenWireFormat;
import org.apache.activemq.thread.TaskRunnerFactory;
@@ -75,6 +74,7 @@ public class NIOSSLTransport extends NIOTransport {
@Override
protected void initializeStreams() throws IOException {
+ NIOOutputStream outputStream = null;
try {
channel = socket.getChannel();
channel.configureBlocking(false);
@@ -119,7 +119,7 @@ public class NIOSSLTransport extends NIOTransport {
inputBuffer = ByteBuffer.allocate(sslSession.getPacketBufferSize());
inputBuffer.clear();
- NIOOutputStream outputStream = new NIOOutputStream(channel);
+ outputStream = new NIOOutputStream(channel);
outputStream.setEngine(sslEngine);
this.dataOut = new DataOutputStream(outputStream);
this.buffOut = outputStream;
@@ -127,6 +127,12 @@ public class NIOSSLTransport extends NIOTransport {
handshakeStatus = sslEngine.getHandshakeStatus();
doHandshake();
} catch (Exception e) {
+ try {
+ if(outputStream != null) {
+ outputStream.close();
+ }
+ super.closeStreams();
+ } catch (Exception ex) {}
throw new IOException(e);
}
}
@@ -143,10 +149,12 @@ public class NIOSSLTransport extends NIOTransport {
// listen for events telling us when the socket is readable.
selection = SelectorManager.getInstance().register(channel, new SelectorManager.Listener() {
+ @Override
public void onSelect(SelectorSelection selection) {
serviceRead();
}
+ @Override
public void onError(SelectorSelection selection, Throwable error) {
if (error instanceof IOException) {
onException((IOException) error);
@@ -158,6 +166,7 @@ public class NIOSSLTransport extends NIOTransport {
}
}
+ @Override
protected void serviceRead() {
try {
if (handshakeInProgress) {
@@ -272,7 +281,7 @@ public class NIOSSLTransport extends NIOTransport {
} else {
currentBuffer.flip();
Object command = wireFormat.unmarshal(new DataInputStream(new NIOInputStream(currentBuffer)));
- doConsume((Command) command);
+ doConsume(command);
nextFrameSize = -1;
currentBuffer = null;
}
[16/17] git commit: Fix for
https://issues.apache.org/jira/browse/AMQ-4918
Posted by ha...@apache.org.
Fix for https://issues.apache.org/jira/browse/AMQ-4918
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/3474f37e
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/3474f37e
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/3474f37e
Branch: refs/heads/activemq-5.9
Commit: 3474f37e74850355b0d95506c0b245e76a7f3434
Parents: cb3e854
Author: rajdavies <ra...@gmail.com>
Authored: Wed Dec 4 20:14:07 2013 +0000
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Wed Mar 12 13:14:14 2014 -0400
----------------------------------------------------------------------
.../activemq/broker/jmx/NetworkBridgeView.java | 18 ++++++++++++++++++
.../broker/jmx/NetworkBridgeViewMBean.java | 2 ++
.../broker/jmx/NetworkDestinationView.java | 12 +++++++++---
.../broker/jmx/NetworkDestinationViewMBean.java | 2 +-
.../activemq/network/MBeanBridgeDestination.java | 14 +++++++++-----
.../activemq/network/MBeanNetworkListener.java | 7 +++----
6 files changed, 42 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/3474f37e/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java
index 2201b30..795de3a 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java
@@ -16,12 +16,16 @@
*/
package org.apache.activemq.broker.jmx;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
import org.apache.activemq.network.NetworkBridge;
public class NetworkBridgeView implements NetworkBridgeViewMBean {
private final NetworkBridge bridge;
private boolean createByDuplex = false;
+ private List<NetworkDestinationView> networkDestinationViewList = new CopyOnWriteArrayList<NetworkDestinationView>();
public NetworkBridgeView(NetworkBridge bridge) {
this.bridge = bridge;
@@ -66,4 +70,18 @@ public class NetworkBridgeView implements NetworkBridgeViewMBean {
public void setCreateByDuplex(boolean createByDuplex) {
this.createByDuplex = createByDuplex;
}
+
+ public void resetStats(){
+ for (NetworkDestinationView networkDestinationView:networkDestinationViewList){
+ networkDestinationView.resetStats();
+ }
+ }
+
+ public void addNetworkDestinationView(NetworkDestinationView networkDestinationView){
+ networkDestinationViewList.add(networkDestinationView);
+ }
+
+ public void removeNetworkDestinationView(NetworkDestinationView networkDestinationView){
+ networkDestinationViewList.remove(networkDestinationView);
+ }
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/3474f37e/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeViewMBean.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeViewMBean.java b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeViewMBean.java
index ec7598c..b8733d1 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeViewMBean.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeViewMBean.java
@@ -34,4 +34,6 @@ public interface NetworkBridgeViewMBean extends Service {
boolean isCreatedByDuplex();
+ void resetStats();
+
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/3474f37e/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkDestinationView.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkDestinationView.java b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkDestinationView.java
index 4edfd37..26177d5 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkDestinationView.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkDestinationView.java
@@ -25,10 +25,12 @@ public class NetworkDestinationView implements NetworkDestinationViewMBean {
private TimeStatisticImpl timeStatistic = new TimeStatisticImpl("networkEnqueue","network messages enqueued");
private final String name;
+ private final NetworkBridgeView networkBridgeView;
private long lastTime = -1;
- public NetworkDestinationView(String name){
- this.name = name;
+ public NetworkDestinationView(NetworkBridgeView networkBridgeView, String name){
+ this.networkBridgeView = networkBridgeView;
+ this.name=name;
}
/**
* Returns the name of this destination
@@ -42,7 +44,7 @@ public class NetworkDestinationView implements NetworkDestinationViewMBean {
* Resets the managment counters.
*/
@Override
- public void resetStatistics() {
+ public void resetStats() {
timeStatistic.reset();
lastTime = -1;
}
@@ -70,4 +72,8 @@ public class NetworkDestinationView implements NetworkDestinationViewMBean {
timeStatistic.addTime(time);
lastTime=currentTime;
}
+
+ public void close(){
+ networkBridgeView.removeNetworkDestinationView(this);
+ }
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/3474f37e/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkDestinationViewMBean.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkDestinationViewMBean.java b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkDestinationViewMBean.java
index 7164cdd..96c57c0 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkDestinationViewMBean.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkDestinationViewMBean.java
@@ -28,7 +28,7 @@ public interface NetworkDestinationViewMBean {
* Resets the managment counters.
*/
@MBeanInfo("Resets statistics.")
- void resetStatistics();
+ void resetStats();
/**
* Returns the number of messages that have been sent to the destination.
http://git-wip-us.apache.org/repos/asf/activemq/blob/3474f37e/activemq-broker/src/main/java/org/apache/activemq/network/MBeanBridgeDestination.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/MBeanBridgeDestination.java b/activemq-broker/src/main/java/org/apache/activemq/network/MBeanBridgeDestination.java
index 666f11e..c718063 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/network/MBeanBridgeDestination.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/network/MBeanBridgeDestination.java
@@ -23,6 +23,7 @@ import javax.management.ObjectName;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.jmx.AnnotatedMBean;
import org.apache.activemq.broker.jmx.BrokerMBeanSupport;
+import org.apache.activemq.broker.jmx.NetworkBridgeView;
import org.apache.activemq.broker.jmx.NetworkDestinationView;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.Message;
@@ -33,13 +34,15 @@ public class MBeanBridgeDestination {
private static final Logger LOG = LoggerFactory.getLogger(MBeanBridgeDestination.class);
private final BrokerService brokerService;
private final NetworkBridge bridge;
+ private final NetworkBridgeView networkBridgeView;
private Map<ActiveMQDestination, ObjectName> destinationObjectNameMap = new ConcurrentHashMap<ActiveMQDestination, ObjectName>();
private Map<ActiveMQDestination, NetworkDestinationView> outboundDestinationViewMap = new ConcurrentHashMap<ActiveMQDestination, NetworkDestinationView>();
private Map<ActiveMQDestination, NetworkDestinationView> inboundDestinationViewMap = new ConcurrentHashMap<ActiveMQDestination, NetworkDestinationView>();
- public MBeanBridgeDestination(BrokerService brokerService, NetworkBridge bridge) {
+ public MBeanBridgeDestination(BrokerService brokerService, NetworkBridge bridge, NetworkBridgeView networkBridgeView) {
this.brokerService = brokerService;
this.bridge = bridge;
+ this.networkBridgeView = networkBridgeView;
}
@@ -48,11 +51,11 @@ public class MBeanBridgeDestination {
NetworkDestinationView networkDestinationView = outboundDestinationViewMap.get(destination);
if (networkDestinationView == null) {
synchronized (destinationObjectNameMap) {
- if (!destinationObjectNameMap.containsKey(destination)) {
+ if ((networkDestinationView = outboundDestinationViewMap.get(destination)) == null) {
ObjectName bridgeObjectName = bridge.getMbeanObjectName();
try {
ObjectName objectName = BrokerMBeanSupport.createNetworkOutBoundDestinationObjectName(bridgeObjectName, destination);
- networkDestinationView = new NetworkDestinationView(destination.getPhysicalName());
+ networkDestinationView = new NetworkDestinationView(networkBridgeView,destination.getPhysicalName());
AnnotatedMBean.registerMBean(brokerService.getManagementContext(), networkDestinationView, objectName);
destinationObjectNameMap.put(destination, objectName);
outboundDestinationViewMap.put(destination, networkDestinationView);
@@ -72,11 +75,12 @@ public class MBeanBridgeDestination {
NetworkDestinationView networkDestinationView = inboundDestinationViewMap.get(destination);
if (networkDestinationView == null) {
synchronized (destinationObjectNameMap) {
- if (!destinationObjectNameMap.containsKey(destination)) {
+ if ((networkDestinationView = inboundDestinationViewMap.get(destination)) == null) {
ObjectName bridgeObjectName = bridge.getMbeanObjectName();
try {
ObjectName objectName = BrokerMBeanSupport.createNetworkInBoundDestinationObjectName(bridgeObjectName, destination);
- networkDestinationView= new NetworkDestinationView(destination.getPhysicalName());
+ networkDestinationView= new NetworkDestinationView(networkBridgeView,destination.getPhysicalName());
+ networkBridgeView.addNetworkDestinationView(networkDestinationView);
AnnotatedMBean.registerMBean(brokerService.getManagementContext(), networkDestinationView, objectName);
destinationObjectNameMap.put(destination, objectName);
inboundDestinationViewMap.put(destination, networkDestinationView);
http://git-wip-us.apache.org/repos/asf/activemq/blob/3474f37e/activemq-broker/src/main/java/org/apache/activemq/network/MBeanNetworkListener.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/MBeanNetworkListener.java b/activemq-broker/src/main/java/org/apache/activemq/network/MBeanNetworkListener.java
index 0481f3d..c877ecd 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/network/MBeanNetworkListener.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/network/MBeanNetworkListener.java
@@ -25,7 +25,6 @@ import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.jmx.AnnotatedMBean;
import org.apache.activemq.broker.jmx.BrokerMBeanSupport;
import org.apache.activemq.broker.jmx.NetworkBridgeView;
-import org.apache.activemq.broker.jmx.NetworkBridgeViewMBean;
import org.apache.activemq.command.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -52,13 +51,13 @@ public class MBeanNetworkListener implements NetworkBridgeListener {
if (!brokerService.isUseJmx()) {
return;
}
- NetworkBridgeViewMBean view = new NetworkBridgeView(bridge);
- ((NetworkBridgeView)view).setCreateByDuplex(createdByDuplex);
+ NetworkBridgeView view = new NetworkBridgeView(bridge);
+ view.setCreateByDuplex(createdByDuplex);
try {
ObjectName objectName = createNetworkBridgeObjectName(bridge);
AnnotatedMBean.registerMBean(brokerService.getManagementContext(), view, objectName);
bridge.setMbeanObjectName(objectName);
- MBeanBridgeDestination mBeanBridgeDestination = new MBeanBridgeDestination(brokerService,bridge);
+ MBeanBridgeDestination mBeanBridgeDestination = new MBeanBridgeDestination(brokerService,bridge,view);
destinationObjectNameMap.put(bridge,mBeanBridgeDestination);
LOG.debug("registered: {} as: {}", bridge, objectName);
} catch (Throwable e) {
[15/17] git commit: Fix for
https://issues.apache.org/jira/browse/AMQ-4918
Posted by ha...@apache.org.
Fix for https://issues.apache.org/jira/browse/AMQ-4918
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/cb3e854d
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/cb3e854d
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/cb3e854d
Branch: refs/heads/activemq-5.9
Commit: cb3e854d26b2b0654e93b37cf9c4c2e30e0b64fa
Parents: 37e1fd7
Author: rajdavies <ra...@gmail.com>
Authored: Wed Dec 4 19:00:54 2013 +0000
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Wed Mar 12 13:14:00 2014 -0400
----------------------------------------------------------------------
.../activemq/broker/jmx/BrokerMBeanSupport.java | 15 +++
.../broker/jmx/NetworkDestinationView.java | 73 ++++++++++++
.../broker/jmx/NetworkDestinationViewMBean.java | 48 ++++++++
.../network/DemandForwardingBridgeSupport.java | 16 +++
.../network/MBeanBridgeDestination.java | 111 +++++++++++++++++++
.../activemq/network/MBeanNetworkListener.java | 34 +++++-
.../activemq/network/NetworkBridgeListener.java | 16 +++
.../activemq/management/SizeStatisticImpl.java | 2 +-
8 files changed, 312 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/cb3e854d/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/BrokerMBeanSupport.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/BrokerMBeanSupport.java b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/BrokerMBeanSupport.java
index 3ad4495..87605bc 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/BrokerMBeanSupport.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/BrokerMBeanSupport.java
@@ -173,6 +173,21 @@ public class BrokerMBeanSupport {
return new ObjectName(connectorName.getDomain(), map);
}
+ public static ObjectName createNetworkOutBoundDestinationObjectName(ObjectName networkName, ActiveMQDestination destination) throws MalformedObjectNameException {
+ String str = networkName.toString();
+ str += ",direction=outbound" + createDestinationProperties(destination);
+ return new ObjectName(str);
+
+ }
+
+ public static ObjectName createNetworkInBoundDestinationObjectName(ObjectName networkName, ActiveMQDestination destination) throws MalformedObjectNameException {
+ String str = networkName.toString();
+ str += ",direction=inbound" + createDestinationProperties(destination);
+ return new ObjectName(str);
+
+ }
+
+
public static ObjectName createProxyConnectorName(ObjectName brokerObjectName, String type, String name) throws MalformedObjectNameException {
return createProxyConnectorName(brokerObjectName.toString(), type, name);
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/cb3e854d/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkDestinationView.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkDestinationView.java b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkDestinationView.java
new file mode 100644
index 0000000..4edfd37
--- /dev/null
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkDestinationView.java
@@ -0,0 +1,73 @@
+/**
+ * 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.jmx;
+
+import org.apache.activemq.management.TimeStatisticImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class NetworkDestinationView implements NetworkDestinationViewMBean {
+ private static final Logger LOG = LoggerFactory.getLogger(NetworkDestinationViewMBean.class);
+ private TimeStatisticImpl timeStatistic = new TimeStatisticImpl("networkEnqueue","network messages enqueued");
+
+ private final String name;
+ private long lastTime = -1;
+
+ public NetworkDestinationView(String name){
+ this.name = name;
+ }
+ /**
+ * Returns the name of this destination
+ */
+ @Override
+ public String getName() {
+ return name;
+ }
+
+ /**
+ * Resets the managment counters.
+ */
+ @Override
+ public void resetStatistics() {
+ timeStatistic.reset();
+ lastTime = -1;
+ }
+
+
+ @Override
+ public long getCount() {
+ return timeStatistic.getCount();
+ }
+
+ @Override
+ public double getRate() {
+ return timeStatistic.getAveragePerSecond();
+ }
+
+ public void messageSent(){
+ long currentTime = System.currentTimeMillis();
+ long time = 0;
+ if (lastTime < 0){
+ time = 0;
+ lastTime = currentTime;
+ }else{
+ time = currentTime-lastTime;
+ }
+ timeStatistic.addTime(time);
+ lastTime=currentTime;
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq/blob/cb3e854d/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkDestinationViewMBean.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkDestinationViewMBean.java b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkDestinationViewMBean.java
new file mode 100644
index 0000000..7164cdd
--- /dev/null
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkDestinationViewMBean.java
@@ -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.jmx;
+
+public interface NetworkDestinationViewMBean {
+
+ /**
+ * Returns the name of this destination
+ */
+ @MBeanInfo("Name of this destination.")
+ String getName();
+
+ /**
+ * Resets the managment counters.
+ */
+ @MBeanInfo("Resets statistics.")
+ void resetStatistics();
+
+ /**
+ * Returns the number of messages that have been sent to the destination.
+ *
+ * @return The number of messages that have been sent to the destination.
+ */
+ @MBeanInfo("Number of messages that have been sent to the destination.")
+ long getCount();
+
+ /**
+ * Returns the rate of messages that have been sent to the destination.
+ *
+ * @return The rate of messages that have been sent to the destination.
+ */
+ @MBeanInfo("rate of messages sent across the network destination.")
+ double getRate();
+}
http://git-wip-us.apache.org/repos/asf/activemq/blob/cb3e854d/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java b/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
index af10a94..bf0b4f6 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
@@ -607,6 +607,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
} else {
duplexInboundLocalBroker.oneway(message);
}
+ serviceInboundMessage(message);
}
} else {
switch (command.getDataStructureType()) {
@@ -985,6 +986,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
sub.decrementOutstandingResponses();
}
}
+ serviceOutbound(message);
} else {
LOG.debug("No subscription registered with this network bridge for consumerId: {} for message: {}", md.getConsumerId(), md.getMessage());
}
@@ -1612,4 +1614,18 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
}
}
+ protected void serviceOutbound(Message message){
+ NetworkBridgeListener l = this.networkBridgeListener;
+ if (l != null){
+ l.onOutboundMessage(this,message);
+ }
+ }
+
+ protected void serviceInboundMessage(Message message){
+ NetworkBridgeListener l = this.networkBridgeListener;
+ if (l != null){
+ l.onInboundMessage(this,message);
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/cb3e854d/activemq-broker/src/main/java/org/apache/activemq/network/MBeanBridgeDestination.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/MBeanBridgeDestination.java b/activemq-broker/src/main/java/org/apache/activemq/network/MBeanBridgeDestination.java
new file mode 100644
index 0000000..666f11e
--- /dev/null
+++ b/activemq-broker/src/main/java/org/apache/activemq/network/MBeanBridgeDestination.java
@@ -0,0 +1,111 @@
+/**
+ * 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.network;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import javax.management.ObjectName;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.jmx.AnnotatedMBean;
+import org.apache.activemq.broker.jmx.BrokerMBeanSupport;
+import org.apache.activemq.broker.jmx.NetworkDestinationView;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.Message;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MBeanBridgeDestination {
+ private static final Logger LOG = LoggerFactory.getLogger(MBeanBridgeDestination.class);
+ private final BrokerService brokerService;
+ private final NetworkBridge bridge;
+ private Map<ActiveMQDestination, ObjectName> destinationObjectNameMap = new ConcurrentHashMap<ActiveMQDestination, ObjectName>();
+ private Map<ActiveMQDestination, NetworkDestinationView> outboundDestinationViewMap = new ConcurrentHashMap<ActiveMQDestination, NetworkDestinationView>();
+ private Map<ActiveMQDestination, NetworkDestinationView> inboundDestinationViewMap = new ConcurrentHashMap<ActiveMQDestination, NetworkDestinationView>();
+
+ public MBeanBridgeDestination(BrokerService brokerService, NetworkBridge bridge) {
+ this.brokerService = brokerService;
+ this.bridge = bridge;
+ }
+
+
+ public void onOutboundMessage(Message message) {
+ ActiveMQDestination destination = message.getDestination();
+ NetworkDestinationView networkDestinationView = outboundDestinationViewMap.get(destination);
+ if (networkDestinationView == null) {
+ synchronized (destinationObjectNameMap) {
+ if (!destinationObjectNameMap.containsKey(destination)) {
+ ObjectName bridgeObjectName = bridge.getMbeanObjectName();
+ try {
+ ObjectName objectName = BrokerMBeanSupport.createNetworkOutBoundDestinationObjectName(bridgeObjectName, destination);
+ networkDestinationView = new NetworkDestinationView(destination.getPhysicalName());
+ AnnotatedMBean.registerMBean(brokerService.getManagementContext(), networkDestinationView, objectName);
+ destinationObjectNameMap.put(destination, objectName);
+ outboundDestinationViewMap.put(destination, networkDestinationView);
+
+ } catch (Exception e) {
+ LOG.warn("Failed to register " + destination, e);
+ }
+ }
+ }
+ }
+ networkDestinationView.messageSent();
+ }
+
+
+ public void onInboundMessage(Message message) {
+ ActiveMQDestination destination = message.getDestination();
+ NetworkDestinationView networkDestinationView = inboundDestinationViewMap.get(destination);
+ if (networkDestinationView == null) {
+ synchronized (destinationObjectNameMap) {
+ if (!destinationObjectNameMap.containsKey(destination)) {
+ ObjectName bridgeObjectName = bridge.getMbeanObjectName();
+ try {
+ ObjectName objectName = BrokerMBeanSupport.createNetworkInBoundDestinationObjectName(bridgeObjectName, destination);
+ networkDestinationView= new NetworkDestinationView(destination.getPhysicalName());
+ AnnotatedMBean.registerMBean(brokerService.getManagementContext(), networkDestinationView, objectName);
+ destinationObjectNameMap.put(destination, objectName);
+ inboundDestinationViewMap.put(destination, networkDestinationView);
+ } catch (Exception e) {
+ LOG.warn("Failed to register " + destination, e);
+ }
+ }
+ }
+ }
+ networkDestinationView.messageSent();
+ }
+
+ public void close() {
+ if (!brokerService.isUseJmx()) {
+ return;
+ }
+
+ for (ObjectName objectName : destinationObjectNameMap.values()) {
+ try {
+ if (objectName != null) {
+ brokerService.getManagementContext().unregisterMBean(objectName);
+ }
+ } catch (Throwable e) {
+ LOG.debug("Network bridge could not be unregistered in JMX: {}", e.getMessage(), e);
+ }
+ }
+ destinationObjectNameMap.clear();
+ outboundDestinationViewMap.clear();
+ inboundDestinationViewMap.clear();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/activemq/blob/cb3e854d/activemq-broker/src/main/java/org/apache/activemq/network/MBeanNetworkListener.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/MBeanNetworkListener.java b/activemq-broker/src/main/java/org/apache/activemq/network/MBeanNetworkListener.java
index c9a9935..0481f3d 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/network/MBeanNetworkListener.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/network/MBeanNetworkListener.java
@@ -16,14 +16,17 @@
*/
package org.apache.activemq.network;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;
-
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.jmx.AnnotatedMBean;
import org.apache.activemq.broker.jmx.BrokerMBeanSupport;
import org.apache.activemq.broker.jmx.NetworkBridgeView;
import org.apache.activemq.broker.jmx.NetworkBridgeViewMBean;
+import org.apache.activemq.command.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -34,7 +37,7 @@ public class MBeanNetworkListener implements NetworkBridgeListener {
BrokerService brokerService;
ObjectName connectorName;
boolean createdByDuplex = false;
-
+ private Map<NetworkBridge,MBeanBridgeDestination> destinationObjectNameMap = new ConcurrentHashMap<NetworkBridge,MBeanBridgeDestination>();
public MBeanNetworkListener(BrokerService brokerService, ObjectName connectorName) {
this.brokerService = brokerService;
this.connectorName = connectorName;
@@ -55,6 +58,8 @@ public class MBeanNetworkListener implements NetworkBridgeListener {
ObjectName objectName = createNetworkBridgeObjectName(bridge);
AnnotatedMBean.registerMBean(brokerService.getManagementContext(), view, objectName);
bridge.setMbeanObjectName(objectName);
+ MBeanBridgeDestination mBeanBridgeDestination = new MBeanBridgeDestination(brokerService,bridge);
+ destinationObjectNameMap.put(bridge,mBeanBridgeDestination);
LOG.debug("registered: {} as: {}", bridge, objectName);
} catch (Throwable e) {
LOG.debug("Network bridge could not be registered in JMX: {}", e.getMessage(), e);
@@ -71,11 +76,17 @@ public class MBeanNetworkListener implements NetworkBridgeListener {
if (objectName != null) {
brokerService.getManagementContext().unregisterMBean(objectName);
}
+ MBeanBridgeDestination mBeanBridgeDestination = destinationObjectNameMap.remove(bridge);
+ if (mBeanBridgeDestination != null){
+ mBeanBridgeDestination.close();
+ }
} catch (Throwable e) {
LOG.debug("Network bridge could not be unregistered in JMX: {}", e.getMessage(), e);
}
}
+
+
protected ObjectName createNetworkBridgeObjectName(NetworkBridge bridge) throws MalformedObjectNameException {
return BrokerMBeanSupport.createNetworkBridgeObjectName(connectorName, bridge.getRemoteAddress());
}
@@ -83,4 +94,23 @@ public class MBeanNetworkListener implements NetworkBridgeListener {
public void setCreatedByDuplex(boolean createdByDuplex) {
this.createdByDuplex = createdByDuplex;
}
+
+
+
+ @Override
+ public void onOutboundMessage(NetworkBridge bridge,Message message) {
+ MBeanBridgeDestination mBeanBridgeDestination = destinationObjectNameMap.get(bridge);
+ if (mBeanBridgeDestination != null){
+ mBeanBridgeDestination.onOutboundMessage(message);
+ }
+ }
+
+ @Override
+ public void onInboundMessage(NetworkBridge bridge,Message message) {
+ MBeanBridgeDestination mBeanBridgeDestination = destinationObjectNameMap.get(bridge);
+ if (mBeanBridgeDestination != null){
+ mBeanBridgeDestination.onInboundMessage(message);
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/cb3e854d/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeListener.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeListener.java b/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeListener.java
index 7d49177..c30a999 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeListener.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeListener.java
@@ -16,6 +16,8 @@
*/
package org.apache.activemq.network;
+import org.apache.activemq.command.Message;
+
/**
* called when a bridge fails
*
@@ -38,4 +40,18 @@ public interface NetworkBridgeListener {
*/
void onStop(NetworkBridge bridge);
+ /**
+ * Called when message forwarded over the network
+ * @param bridge
+ * @param message
+ */
+ void onOutboundMessage (NetworkBridge bridge,Message message);
+
+ /**
+ * Called for when a message arrives over the network
+ * @param bridge
+ * @param message
+ */
+ void onInboundMessage (NetworkBridge bridge,Message message);
+
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/cb3e854d/activemq-client/src/main/java/org/apache/activemq/management/SizeStatisticImpl.java
----------------------------------------------------------------------
diff --git a/activemq-client/src/main/java/org/apache/activemq/management/SizeStatisticImpl.java b/activemq-client/src/main/java/org/apache/activemq/management/SizeStatisticImpl.java
index 14664d2..d3dbb50 100644
--- a/activemq-client/src/main/java/org/apache/activemq/management/SizeStatisticImpl.java
+++ b/activemq-client/src/main/java/org/apache/activemq/management/SizeStatisticImpl.java
@@ -154,7 +154,7 @@ public class SizeStatisticImpl extends StatisticImpl{
buffer.append(Long.toString(minSize));
buffer.append(" totalSize: ");
buffer.append(Long.toString(totalSize));
- buffer.append(" averageTime: ");
+ buffer.append(" averageSize: ");
buffer.append(Double.toString(getAverageSize()));
buffer.append(" averageTimeExMinMax: ");
buffer.append(Double.toString(getAveragePerSecondExcludingMinMax()));
[06/17] git commit: ignore test pending investigation on new
SuppressReply from
http://git-wip-us.apache.org/repos/asf/activemq/commit/b0e91d47
Posted by ha...@apache.org.
ignore test pending investigation on new SuppressReply from http://git-wip-us.apache.org/repos/asf/activemq/commit/b0e91d47
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/ead3d163
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/ead3d163
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/ead3d163
Branch: refs/heads/activemq-5.9
Commit: ead3d163a894056035dffc37a9dfd80bcf2d7c1f
Parents: 4b01fe3
Author: gtully <ga...@gmail.com>
Authored: Fri Nov 29 12:40:00 2013 +0000
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Wed Mar 12 13:11:38 2014 -0400
----------------------------------------------------------------------
.../activemq/store/LevelDBStorePerDestinationTest.java | 13 +++++++++++++
1 file changed, 13 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/ead3d163/activemq-unit-tests/src/test/java/org/apache/activemq/store/LevelDBStorePerDestinationTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/store/LevelDBStorePerDestinationTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/store/LevelDBStorePerDestinationTest.java
index 8907f9b..fe15be5 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/store/LevelDBStorePerDestinationTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/store/LevelDBStorePerDestinationTest.java
@@ -17,6 +17,8 @@
package org.apache.activemq.store;
import org.apache.activemq.leveldb.LevelDBStore;
+import org.junit.Test;
+import org.junit.Ignore;
import java.io.IOException;
@@ -32,4 +34,15 @@ public class LevelDBStorePerDestinationTest extends StorePerDestinationTest {
}
return store;
}
+
+ @Ignore("needs some investigation w.r.t to SuppressReplyException - broken by: https://git-wip-us.apache.org/repos/asf?p=activemq.git;a=commit;h=b0e91d47")
+ @Test
+ @Override
+ public void testRollbackRecovery() throws Exception {}
+
+ @Ignore("needs some investigation w.r.t to SuppressReplyException - broken by: https://git-wip-us.apache.org/repos/asf?p=activemq.git;a=commit;h=b0e91d47")
+ @Test
+ @Override
+ public void testCommitRecovery() throws Exception {}
+
}
\ No newline at end of file
[08/17] git commit: Fixing regression caused by changes to
SuppressRelyException in commit b0e91d47f5fced59c89a34d993f4d87c7986b04b.
Posted by ha...@apache.org.
Fixing regression caused by changes to SuppressRelyException in commit b0e91d47f5fced59c89a34d993f4d87c7986b04b.
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/0bf2233c
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/0bf2233c
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/0bf2233c
Branch: refs/heads/activemq-5.9
Commit: 0bf2233c8a3d61c2fda18dc29422c5d5813b5460
Parents: 30352d0
Author: Hiram Chirino <hi...@hiramchirino.com>
Authored: Mon Dec 2 12:29:21 2013 -0500
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Wed Mar 12 13:12:21 2014 -0400
----------------------------------------------------------------------
.../main/scala/org/apache/activemq/leveldb/LevelDBStore.scala | 4 ++--
.../apache/activemq/store/LevelDBStorePerDestinationTest.java | 3 ---
2 files changed, 2 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/0bf2233c/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala
----------------------------------------------------------------------
diff --git a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala
index e4c7a02..43238a9 100644
--- a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala
+++ b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala
@@ -437,13 +437,13 @@ class LevelDBStore extends LockableServiceSupport with BrokerServiceAware with P
def verify_running = {
if( isStopping || isStopped ) {
try {
- throw new SuppressReplyException("Not running")
+ throw new IOException("Not running")
} catch {
case e:IOException =>
if( broker_service!=null ) {
broker_service.handleIOException(e)
}
- throw e
+ throw new SuppressReplyException(e);
}
}
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/0bf2233c/activemq-unit-tests/src/test/java/org/apache/activemq/store/LevelDBStorePerDestinationTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/store/LevelDBStorePerDestinationTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/store/LevelDBStorePerDestinationTest.java
index fe15be5..b10786c 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/store/LevelDBStorePerDestinationTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/store/LevelDBStorePerDestinationTest.java
@@ -18,7 +18,6 @@ package org.apache.activemq.store;
import org.apache.activemq.leveldb.LevelDBStore;
import org.junit.Test;
-import org.junit.Ignore;
import java.io.IOException;
@@ -35,12 +34,10 @@ public class LevelDBStorePerDestinationTest extends StorePerDestinationTest {
return store;
}
- @Ignore("needs some investigation w.r.t to SuppressReplyException - broken by: https://git-wip-us.apache.org/repos/asf?p=activemq.git;a=commit;h=b0e91d47")
@Test
@Override
public void testRollbackRecovery() throws Exception {}
- @Ignore("needs some investigation w.r.t to SuppressReplyException - broken by: https://git-wip-us.apache.org/repos/asf?p=activemq.git;a=commit;h=b0e91d47")
@Test
@Override
public void testCommitRecovery() throws Exception {}
[05/17] git commit: https://issues.apache.org/jira/browse/AMQ-4907 -
sanity check on the index when checkForCorruptJournalFiles - test and check
that validates the orderindex makes sense
Posted by ha...@apache.org.
https://issues.apache.org/jira/browse/AMQ-4907 - sanity check on the index when checkForCorruptJournalFiles - test and check that validates the orderindex makes sense
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/4b01fe35
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/4b01fe35
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/4b01fe35
Branch: refs/heads/activemq-5.9
Commit: 4b01fe3512d9b384d2243547f01f9c67d7106ceb
Parents: 4aae032
Author: gtully <ga...@gmail.com>
Authored: Thu Nov 28 21:13:02 2013 +0000
Committer: Hadrian Zbarcea <ha...@apache.org>
Committed: Wed Mar 12 13:11:20 2014 -0400
----------------------------------------------------------------------
.../activemq/store/kahadb/MessageDatabase.java | 9 +++++++++
.../store/kahadb/KahaDBStoreRecoveryBrokerTest.java | 16 ++++++++++++++--
2 files changed, 23 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/4b01fe35/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
----------------------------------------------------------------------
diff --git a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
index b3c2d6f..8c8fe0f 100644
--- a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
+++ b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
@@ -303,6 +303,15 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
Entry<String, StoredDestination> entry = iterator.next();
StoredDestination sd = loadStoredDestination(tx, entry.getKey(), entry.getValue().subscriptions!=null);
storedDestinations.put(entry.getKey(), sd);
+
+ if (checkForCorruptJournalFiles) {
+ // sanity check the index also
+ if (!entry.getValue().locationIndex.isEmpty(tx)) {
+ if (entry.getValue().orderIndex.nextMessageId <= 0) {
+ throw new IOException("Detected uninitialized orderIndex nextMessageId with pending messages for " + entry.getKey());
+ }
+ }
+ }
}
}
});
http://git-wip-us.apache.org/repos/asf/activemq/blob/4b01fe35/activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBStoreRecoveryBrokerTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBStoreRecoveryBrokerTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBStoreRecoveryBrokerTest.java
index 2ce9533..3725572 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBStoreRecoveryBrokerTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBStoreRecoveryBrokerTest.java
@@ -46,7 +46,7 @@ public class KahaDBStoreRecoveryBrokerTest extends RecoveryBrokerTest {
public static final String KAHADB_DIR_BASE = "target/activemq-data/kahadb";
public static String kahaDbDirectoryName;
- enum CorruptionType { None, FailToLoad, LoadInvalid, LoadCorrupt };
+ enum CorruptionType { None, FailToLoad, LoadInvalid, LoadCorrupt, LoadOrderIndex0 };
public CorruptionType failTest = CorruptionType.None;
@Override
@@ -71,6 +71,7 @@ public class KahaDBStoreRecoveryBrokerTest extends RecoveryBrokerTest {
KahaDBStore kaha = new KahaDBStore();
kaha.setDirectory(new File(kahaDbDirectoryName));
kaha.deleteAllMessages();
+ kaha.setCheckForCorruptJournalFiles(failTest == CorruptionType.LoadOrderIndex0);
broker.setPersistenceAdapter(kaha);
return broker;
}
@@ -100,6 +101,16 @@ public class KahaDBStoreRecoveryBrokerTest extends RecoveryBrokerTest {
raf.seek(8*1024 + 57);
raf.writeLong(Integer.MAX_VALUE-10);
break;
+ case LoadOrderIndex0:
+ // loadable but invalid metadata
+ // location of order index default priority index size
+ // so looks like there are no ids in the order index
+ // picked up by setCheckForCorruptJournalFiles
+ raf.seek(12*1024 + 21);
+ raf.writeShort(0);
+ raf.writeChar(0);
+ raf.writeLong(-1);
+ break;
default:
}
raf.close();
@@ -107,6 +118,7 @@ public class KahaDBStoreRecoveryBrokerTest extends RecoveryBrokerTest {
// starting broker
BrokerService broker = new BrokerService();
KahaDBStore kaha = new KahaDBStore();
+ kaha.setCheckForCorruptJournalFiles(failTest == CorruptionType.LoadOrderIndex0);
// uncomment if you want to test archiving
//kaha.setArchiveCorruptedIndex(true);
kaha.setDirectory(new File(kahaDbDirectoryName));
@@ -123,7 +135,7 @@ public class KahaDBStoreRecoveryBrokerTest extends RecoveryBrokerTest {
}
public void initCombosForTestLargeQueuePersistentMessagesNotLostOnRestart() {
- this.addCombinationValues("failTest", new CorruptionType[]{CorruptionType.FailToLoad, CorruptionType.LoadInvalid, CorruptionType.LoadCorrupt} );
+ this.addCombinationValues("failTest", new CorruptionType[]{CorruptionType.FailToLoad, CorruptionType.LoadInvalid, CorruptionType.LoadCorrupt, CorruptionType.LoadOrderIndex0} );
}
public void testLargeQueuePersistentMessagesNotLostOnRestart() throws Exception {