You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2016/12/21 17:06:56 UTC
[1/5] activemq-artemis git commit: NO-JIRA fixing a test
Repository: activemq-artemis
Updated Branches:
refs/heads/master 0b131bddf -> 66b24d16c
NO-JIRA fixing a test
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/5e4f4aea
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/5e4f4aea
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/5e4f4aea
Branch: refs/heads/master
Commit: 5e4f4aea8296f83f04d23b14b1bb5faf9af08fe2
Parents: 8941f2b
Author: Clebert Suconic <cl...@apache.org>
Authored: Wed Dec 21 11:44:52 2016 -0500
Committer: Clebert Suconic <cl...@apache.org>
Committed: Wed Dec 21 12:04:06 2016 -0500
----------------------------------------------------------------------
.../tests/integration/crossprotocol/OpenWireToAMQPTest.java | 6 ++++--
1 file changed, 4 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5e4f4aea/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/crossprotocol/OpenWireToAMQPTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/crossprotocol/OpenWireToAMQPTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/crossprotocol/OpenWireToAMQPTest.java
index 752e60a..d432a9a 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/crossprotocol/OpenWireToAMQPTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/crossprotocol/OpenWireToAMQPTest.java
@@ -34,6 +34,7 @@ import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.apache.qpid.jms.JmsConnectionFactory;
import org.junit.After;
+import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@@ -56,11 +57,11 @@ public class OpenWireToAMQPTest extends ActiveMQTestBase {
super.setUp();
server = createServer(true, true);
Configuration serverConfig = server.getConfiguration();
- serverConfig.getAddressesSettings().put("#", new AddressSettings().setAutoCreateQueues(false).setAutoCreateAddresses(false).setDeadLetterAddress(new SimpleString("ActiveMQ.DLQ")));
+ serverConfig.getAddressesSettings().put("#", new AddressSettings().setAutoCreateQueues(true).setAutoCreateAddresses(true).setDeadLetterAddress(new SimpleString("ActiveMQ.DLQ")));
serverConfig.setSecurityEnabled(false);
server.start();
coreQueue = new SimpleString(queueName);
- this.server.createQueue(coreQueue, RoutingType.MULTICAST, coreQueue, null, false, false);
+ this.server.createQueue(coreQueue, RoutingType.MULTICAST, coreQueue, null, false, false, -1, false, true);
qpidfactory = new JmsConnectionFactory("amqp://localhost:61616");
}
@@ -93,6 +94,7 @@ public class OpenWireToAMQPTest extends ActiveMQTestBase {
MessageConsumer consumer = session.createConsumer(queue);
connection.start();
ObjectMessage receive = (ObjectMessage) consumer.receive(5000);
+ Assert.assertNotNull(receive);
list = (ArrayList) receive.getObject();
assertEquals(list.get(0), "aString");
connection.close();
[5/5] activemq-artemis git commit: This closes #907
Posted by cl...@apache.org.
This closes #907
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/66b24d16
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/66b24d16
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/66b24d16
Branch: refs/heads/master
Commit: 66b24d16c4b8342c78a7334a33590e5c630a9e0a
Parents: 0b131bd 5e4f4ae
Author: Clebert Suconic <cl...@apache.org>
Authored: Wed Dec 21 12:05:07 2016 -0500
Committer: Clebert Suconic <cl...@apache.org>
Committed: Wed Dec 21 12:05:07 2016 -0500
----------------------------------------------------------------------
.../protocol/openwire/OpenWireConnection.java | 18 ++
.../core/protocol/openwire/amq/AMQSession.java | 170 +++++++++----------
.../crossprotocol/OpenWireToAMQPTest.java | 6 +-
.../integration/openwire/BasicOpenWireTest.java | 15 +-
.../openwire/amq/ProducerBlockingTtlTest.java | 147 ++++++++++++++++
.../openwire/amq/ProducerFlowControlTest.java | 2 +-
.../openwire/interop/CompressedInteropTest.java | 12 +-
7 files changed, 277 insertions(+), 93 deletions(-)
----------------------------------------------------------------------
[2/5] activemq-artemis git commit: ARTEMIS-883 fixing tests
Posted by cl...@apache.org.
ARTEMIS-883 fixing tests
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/8941f2b1
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/8941f2b1
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/8941f2b1
Branch: refs/heads/master
Commit: 8941f2b15d9721d76ee5c788eabf08cd302bb63f
Parents: 81df93e
Author: Clebert Suconic <cl...@apache.org>
Authored: Wed Dec 21 11:40:06 2016 -0500
Committer: Clebert Suconic <cl...@apache.org>
Committed: Wed Dec 21 12:04:06 2016 -0500
----------------------------------------------------------------------
.../tests/integration/openwire/BasicOpenWireTest.java | 8 ++++++--
.../openwire/amq/ProducerFlowControlTest.java | 2 +-
.../openwire/interop/CompressedInteropTest.java | 12 +++++++++++-
3 files changed, 18 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/8941f2b1/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/BasicOpenWireTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/BasicOpenWireTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/BasicOpenWireTest.java
index d01e237..66805d1 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/BasicOpenWireTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/BasicOpenWireTest.java
@@ -65,8 +65,7 @@ public class BasicOpenWireTest extends OpenWireTestBase {
public void setUp() throws Exception {
super.setUp();
System.setProperty("org.apache.activemq.transport.AbstractInactivityMonitor.keepAliveTime", "5");
- factory = new ActiveMQConnectionFactory(getConnectionUrl());
- xaFactory = new ActiveMQXAConnectionFactory(getConnectionUrl());
+ createFactories();
SimpleString coreQueue = new SimpleString(queueName);
this.server.createQueue(coreQueue, RoutingType.ANYCAST, coreQueue, null, false, false, -1, false, true);
testQueues.put(queueName, coreQueue);
@@ -84,6 +83,11 @@ public class BasicOpenWireTest extends OpenWireTestBase {
}
}
+ protected void createFactories() {
+ factory = new ActiveMQConnectionFactory(getConnectionUrl());
+ xaFactory = new ActiveMQXAConnectionFactory(getConnectionUrl());
+ }
+
protected String getConnectionUrl() {
return urlString;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/8941f2b1/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/amq/ProducerFlowControlTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/amq/ProducerFlowControlTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/amq/ProducerFlowControlTest.java
index d372c46..c085d0f 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/amq/ProducerFlowControlTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/amq/ProducerFlowControlTest.java
@@ -282,7 +282,7 @@ public class ProducerFlowControlTest extends BasicOpenWireTest {
protected void waitForBlockedOrResourceLimit(final AtomicBoolean done) throws InterruptedException {
while (true) {
- Thread.sleep(2000);
+ Thread.sleep(100);
System.out.println("check done: " + done.get() + " ex: " + gotResourceException.get());
// the producer is blocked once the done flag stays true or there is a
// resource exception
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/8941f2b1/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/interop/CompressedInteropTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/interop/CompressedInteropTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/interop/CompressedInteropTest.java
index 5f2c621..337027b 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/interop/CompressedInteropTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/interop/CompressedInteropTest.java
@@ -28,7 +28,9 @@ import javax.jms.StreamMessage;
import javax.jms.TextMessage;
import java.nio.charset.StandardCharsets;
+import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQMessageProducer;
+import org.apache.activemq.ActiveMQXAConnectionFactory;
import org.apache.activemq.artemis.tests.integration.openwire.BasicOpenWireTest;
import org.apache.activemq.command.ActiveMQDestination;
import org.junit.Before;
@@ -50,12 +52,20 @@ public class CompressedInteropTest extends BasicOpenWireTest {
@Before
@Override
public void setUp() throws Exception {
- factory.setUseCompression(true);
super.setUp();
connection.start();
assertTrue(connection.isUseCompression());
}
+
+ @Override
+ protected void createFactories() {
+ super.createFactories();
+ factory.setUseCompression(true);
+ xaFactory.setUseCompression(true);
+ }
+
+
@Test
public void testCoreReceiveOpenWireCompressedMessages() throws Exception {
//TextMessage
[4/5] activemq-artemis git commit: ARTEMIS-883 improving exceptions
sending
Posted by cl...@apache.org.
ARTEMIS-883 improving exceptions sending
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/81df93eb
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/81df93eb
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/81df93eb
Branch: refs/heads/master
Commit: 81df93ebcfb53a8fe1070420a66913319c5bcaa5
Parents: ae90edf
Author: Clebert Suconic <cl...@apache.org>
Authored: Wed Dec 21 11:16:15 2016 -0500
Committer: Clebert Suconic <cl...@apache.org>
Committed: Wed Dec 21 12:04:06 2016 -0500
----------------------------------------------------------------------
.../core/protocol/openwire/amq/AMQSession.java | 17 ++++++++++-------
1 file changed, 10 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/81df93eb/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/amq/AMQSession.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/amq/AMQSession.java b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/amq/AMQSession.java
index 006f05e..fbdefd0 100644
--- a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/amq/AMQSession.java
+++ b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/amq/AMQSession.java
@@ -55,8 +55,11 @@ import org.apache.activemq.command.Response;
import org.apache.activemq.command.SessionInfo;
import org.apache.activemq.openwire.OpenWireFormat;
import org.apache.activemq.wireformat.WireFormat;
+import org.jboss.logging.Logger;
public class AMQSession implements SessionCallback {
+ private final Logger logger = Logger.getLogger(AMQSession.class);
+
// ConsumerID is generated inside the session, 0, 1, 2, ... as many consumers as you have on the session
protected final IDGenerator consumerIDGenerator = new SimpleIDGenerator(0);
@@ -305,7 +308,6 @@ public class AMQSession implements SessionCallback {
final AtomicInteger count = new AtomicInteger(actualDestinations.length);
- final Exception[] anyException = new Exception[] {null};
if (shouldBlockProducer) {
connection.getContext().setDontSendReponse(true);
@@ -329,6 +331,8 @@ public class AMQSession implements SessionCallback {
this.connection.disableTtl();
if (shouldBlockProducer) {
if (!store.checkMemory(() -> {
+ Exception exceptionToSend = null;
+
try {
RoutingStatus result = getCoreSession().send(coreMsg, false, dest.isTemporary());
@@ -336,16 +340,15 @@ public class AMQSession implements SessionCallback {
throw new InvalidDestinationException("Cannot publish to a non-existent Destination: " + dest);
}
} catch (Exception e) {
- if (anyException[0] == null) {
- anyException[0] = e;
- }
+
+ logger.warn(e.getMessage(), e);
+ exceptionToSend = e;
}
connection.enableTtl();
if (count.decrementAndGet() == 0) {
- if (anyException[0] != null) {
+ if (exceptionToSend != null) {
this.connection.getContext().setDontSendReponse(false);
- ActiveMQServerLogger.LOGGER.warn(anyException[0].getMessage(), anyException[0]);
- connection.sendException(anyException[0]);
+ connection.sendException(exceptionToSend);
} else {
if (sendProducerAck) {
try {
[3/5] activemq-artemis git commit: ARTEMIS-883 Fix OpenWire
ProducerFlowControlTest
Posted by cl...@apache.org.
ARTEMIS-883 Fix OpenWire ProducerFlowControlTest
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/ae90edfd
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/ae90edfd
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/ae90edfd
Branch: refs/heads/master
Commit: ae90edfdb624e1cdf4a4b36cbf82dbc7a1be8295
Parents: 0b131bd
Author: Howard Gao <ho...@gmail.com>
Authored: Tue Dec 20 21:13:38 2016 +0800
Committer: Clebert Suconic <cl...@apache.org>
Committed: Wed Dec 21 12:04:06 2016 -0500
----------------------------------------------------------------------
.../protocol/openwire/OpenWireConnection.java | 18 ++
.../core/protocol/openwire/amq/AMQSession.java | 169 +++++++++----------
.../integration/openwire/BasicOpenWireTest.java | 11 +-
.../openwire/amq/ProducerBlockingTtlTest.java | 147 ++++++++++++++++
4 files changed, 255 insertions(+), 90 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ae90edfd/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireConnection.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireConnection.java b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireConnection.java
index f90c0b7..0bcff66 100644
--- a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireConnection.java
+++ b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireConnection.java
@@ -183,6 +183,8 @@ public class OpenWireConnection extends AbstractRemotingConnection implements Se
private final Set<SimpleString> knownDestinations = new ConcurrentHashSet<>();
+ private AtomicBoolean disableTtl = new AtomicBoolean(false);
+
// TODO-NOW: check on why there are two connections created for every createConnection on the client.
public OpenWireConnection(Connection connection,
ActiveMQServer server,
@@ -776,6 +778,14 @@ public class OpenWireConnection extends AbstractRemotingConnection implements Se
this.connectionEntry = connectionEntry;
}
+ @Override
+ public boolean checkDataReceived() {
+ if (disableTtl.get()) {
+ return true;
+ }
+ return super.checkDataReceived();
+ }
+
public void setUpTtl(final long inactivityDuration,
final long inactivityDurationInitialDelay,
final boolean useKeepAlive) {
@@ -818,6 +828,14 @@ public class OpenWireConnection extends AbstractRemotingConnection implements Se
}
}
+ public void disableTtl() {
+ disableTtl.set(true);
+ }
+
+ public void enableTtl() {
+ disableTtl.set(false);
+ }
+
class SlowConsumerDetection implements SlowConsumerDetectionListener {
@Override
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ae90edfd/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/amq/AMQSession.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/amq/AMQSession.java b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/amq/AMQSession.java
index 9c592ca..006f05e 100644
--- a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/amq/AMQSession.java
+++ b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/amq/AMQSession.java
@@ -40,9 +40,7 @@ import org.apache.activemq.artemis.core.server.ServerConsumer;
import org.apache.activemq.artemis.core.server.ServerMessage;
import org.apache.activemq.artemis.core.server.ServerSession;
import org.apache.activemq.artemis.core.server.SlowConsumerDetectionListener;
-import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
import org.apache.activemq.artemis.spi.core.protocol.SessionCallback;
-import org.apache.activemq.artemis.spi.core.remoting.Connection;
import org.apache.activemq.artemis.spi.core.remoting.ReadyListener;
import org.apache.activemq.artemis.utils.IDGenerator;
import org.apache.activemq.artemis.utils.SimpleIDGenerator;
@@ -53,12 +51,12 @@ import org.apache.activemq.command.Message;
import org.apache.activemq.command.MessageDispatch;
import org.apache.activemq.command.ProducerAck;
import org.apache.activemq.command.ProducerInfo;
+import org.apache.activemq.command.Response;
import org.apache.activemq.command.SessionInfo;
import org.apache.activemq.openwire.OpenWireFormat;
import org.apache.activemq.wireformat.WireFormat;
public class AMQSession implements SessionCallback {
-
// ConsumerID is generated inside the session, 0, 1, 2, ... as many consumers as you have on the session
protected final IDGenerator consumerIDGenerator = new SimpleIDGenerator(0);
@@ -303,108 +301,103 @@ public class AMQSession implements SessionCallback {
originalCoreMsg.putStringProperty(org.apache.activemq.artemis.api.core.Message.HDR_DUPLICATE_DETECTION_ID.toString(), messageSend.getMessageId().toString());
}
- Runnable runnable;
-
- if (sendProducerAck) {
- runnable = new Runnable() {
- @Override
- public void run() {
- try {
- ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), messageSend.getSize());
- connection.dispatchSync(ack);
- } catch (Exception e) {
- ActiveMQServerLogger.LOGGER.warn(e.getMessage(), e);
- connection.sendException(e);
- }
-
- }
- };
- } else {
- final Connection transportConnection = connection.getTransportConnection();
-
- if (transportConnection == null) {
- // I don't think this could happen, but just in case, avoiding races
- runnable = null;
- } else {
- runnable = new Runnable() {
- @Override
- public void run() {
- transportConnection.setAutoRead(true);
- }
- };
- }
- }
-
- internalSend(actualDestinations, originalCoreMsg, runnable);
- }
+ boolean shouldBlockProducer = producerInfo.getWindowSize() > 0 || messageSend.isResponseRequired();
- private void internalSend(ActiveMQDestination[] actualDestinations,
- ServerMessage originalCoreMsg,
- final Runnable onComplete) throws Exception {
+ final AtomicInteger count = new AtomicInteger(actualDestinations.length);
- Runnable runToUse;
+ final Exception[] anyException = new Exception[] {null};
- if (actualDestinations.length <= 1 || onComplete == null) {
- // if onComplete is null, this will be null ;)
- runToUse = onComplete;
- } else {
- final AtomicInteger count = new AtomicInteger(actualDestinations.length);
- runToUse = new Runnable() {
- @Override
- public void run() {
- if (count.decrementAndGet() == 0) {
- onComplete.run();
- }
- }
- };
+ if (shouldBlockProducer) {
+ connection.getContext().setDontSendReponse(true);
}
- SimpleString[] addresses = new SimpleString[actualDestinations.length];
- PagingStore[] pagingStores = new PagingStore[actualDestinations.length];
-
- // We fillup addresses, pagingStores and we will throw failure if that's the case
for (int i = 0; i < actualDestinations.length; i++) {
ActiveMQDestination dest = actualDestinations[i];
- addresses[i] = new SimpleString(dest.getPhysicalName());
- pagingStores[i] = server.getPagingManager().getPageStore(addresses[i]);
- if (pagingStores[i].getAddressFullMessagePolicy() == AddressFullMessagePolicy.FAIL && pagingStores[i].isFull()) {
- throw new ResourceAllocationException("Queue is full");
- }
- }
-
- for (int i = 0; i < actualDestinations.length; i++) {
-
+ SimpleString address = new SimpleString(dest.getPhysicalName());
ServerMessage coreMsg = originalCoreMsg.copy();
-
- coreMsg.setAddress(addresses[i]);
-
- PagingStore store = pagingStores[i];
-
- if (store.isFull()) {
- connection.getTransportConnection().setAutoRead(false);
- }
+ coreMsg.setAddress(address);
if (actualDestinations[i].isQueue()) {
checkAutoCreateQueue(new SimpleString(actualDestinations[i].getPhysicalName()), actualDestinations[i].isTemporary());
- }
-
- if (actualDestinations[i].isQueue()) {
coreMsg.putByteProperty(org.apache.activemq.artemis.api.core.Message.HDR_ROUTING_TYPE, RoutingType.ANYCAST.getType());
} else {
coreMsg.putByteProperty(org.apache.activemq.artemis.api.core.Message.HDR_ROUTING_TYPE, RoutingType.MULTICAST.getType());
}
- RoutingStatus result = getCoreSession().send(coreMsg, false, actualDestinations[i].isTemporary());
+ PagingStore store = server.getPagingManager().getPageStore(address);
- if (result == RoutingStatus.NO_BINDINGS && actualDestinations[i].isQueue()) {
- throw new InvalidDestinationException("Cannot publish to a non-existent Destination: " + actualDestinations[i]);
- }
- if (runToUse != null) {
- // if the timeout is >0, it will wait this much milliseconds
- // before running the the runToUse
- // this will eventually unblock blocked destinations
- // playing flow control
- store.checkMemory(runToUse);
+ this.connection.disableTtl();
+ if (shouldBlockProducer) {
+ if (!store.checkMemory(() -> {
+ try {
+ RoutingStatus result = getCoreSession().send(coreMsg, false, dest.isTemporary());
+
+ if (result == RoutingStatus.NO_BINDINGS && dest.isQueue()) {
+ throw new InvalidDestinationException("Cannot publish to a non-existent Destination: " + dest);
+ }
+ } catch (Exception e) {
+ if (anyException[0] == null) {
+ anyException[0] = e;
+ }
+ }
+ connection.enableTtl();
+ if (count.decrementAndGet() == 0) {
+ if (anyException[0] != null) {
+ this.connection.getContext().setDontSendReponse(false);
+ ActiveMQServerLogger.LOGGER.warn(anyException[0].getMessage(), anyException[0]);
+ connection.sendException(anyException[0]);
+ } else {
+ if (sendProducerAck) {
+ try {
+ ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), messageSend.getSize());
+ connection.dispatchAsync(ack);
+ } catch (Exception e) {
+ this.connection.getContext().setDontSendReponse(false);
+ ActiveMQServerLogger.LOGGER.warn(e.getMessage(), e);
+ connection.sendException(e);
+ }
+ } else {
+ connection.getContext().setDontSendReponse(false);
+ try {
+ Response response = new Response();
+ response.setCorrelationId(messageSend.getCommandId());
+ connection.dispatchAsync(response);
+ } catch (Exception e) {
+ ActiveMQServerLogger.LOGGER.warn(e.getMessage(), e);
+ connection.sendException(e);
+ }
+ }
+ }
+ }
+ })) {
+ this.connection.getContext().setDontSendReponse(false);
+ connection.enableTtl();
+ throw new ResourceAllocationException("Queue is full " + address);
+ }
+ } else {
+ //non-persistent messages goes here, by default we stop reading from
+ //transport
+ connection.getTransportConnection().setAutoRead(false);
+ if (!store.checkMemory(() -> {
+ connection.getTransportConnection().setAutoRead(true);
+ connection.enableTtl();
+ })) {
+ connection.getTransportConnection().setAutoRead(true);
+ connection.enableTtl();
+ throw new ResourceAllocationException("Queue is full " + address);
+ }
+
+ RoutingStatus result = getCoreSession().send(coreMsg, false, dest.isTemporary());
+ if (result == RoutingStatus.NO_BINDINGS && dest.isQueue()) {
+ throw new InvalidDestinationException("Cannot publish to a non-existent Destination: " + dest);
+ }
+
+ if (count.decrementAndGet() == 0) {
+ if (sendProducerAck) {
+ ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), messageSend.getSize());
+ connection.dispatchAsync(ack);
+ }
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ae90edfd/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/BasicOpenWireTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/BasicOpenWireTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/BasicOpenWireTest.java
index 6be92f8..d01e237 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/BasicOpenWireTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/BasicOpenWireTest.java
@@ -45,8 +45,8 @@ public class BasicOpenWireTest extends OpenWireTestBase {
public TestName name = new TestName();
protected static final String urlString = "tcp://" + OWHOST + ":" + OWPORT + "?wireFormat.cacheEnabled=true";
- protected ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(urlString);
- protected ActiveMQXAConnectionFactory xaFactory = new ActiveMQXAConnectionFactory(urlString);
+ protected ActiveMQConnectionFactory factory;
+ protected ActiveMQXAConnectionFactory xaFactory;
protected ActiveMQConnection connection;
protected String topicName = "amqTestTopic1";
@@ -64,6 +64,9 @@ public class BasicOpenWireTest extends OpenWireTestBase {
@Before
public void setUp() throws Exception {
super.setUp();
+ System.setProperty("org.apache.activemq.transport.AbstractInactivityMonitor.keepAliveTime", "5");
+ factory = new ActiveMQConnectionFactory(getConnectionUrl());
+ xaFactory = new ActiveMQXAConnectionFactory(getConnectionUrl());
SimpleString coreQueue = new SimpleString(queueName);
this.server.createQueue(coreQueue, RoutingType.ANYCAST, coreQueue, null, false, false, -1, false, true);
testQueues.put(queueName, coreQueue);
@@ -81,6 +84,10 @@ public class BasicOpenWireTest extends OpenWireTestBase {
}
}
+ protected String getConnectionUrl() {
+ return urlString;
+ }
+
@Override
@After
public void tearDown() throws Exception {
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ae90edfd/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/amq/ProducerBlockingTtlTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/amq/ProducerBlockingTtlTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/amq/ProducerBlockingTtlTest.java
new file mode 100644
index 0000000..8473d67
--- /dev/null
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/amq/ProducerBlockingTtlTest.java
@@ -0,0 +1,147 @@
+/*
+ * 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.artemis.tests.integration.openwire.amq;
+
+import org.apache.activemq.ActiveMQConnection;
+import org.apache.activemq.artemis.core.config.Configuration;
+import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.tests.integration.openwire.BasicOpenWireTest;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.transport.tcp.TcpTransport;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import javax.jms.Connection;
+import javax.jms.DeliveryMode;
+import javax.jms.JMSException;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class ProducerBlockingTtlTest extends BasicOpenWireTest {
+
+ ActiveMQQueue queueA = new ActiveMQQueue("QUEUE.A");
+ protected ActiveMQConnection flowControlConnection;
+
+ @Override
+ protected void extraServerConfig(Configuration serverConfig) {
+ String match = "#";
+ Map<String, AddressSettings> asMap = serverConfig.getAddressesSettings();
+ asMap.get(match).setMaxSizeBytes(1).setAddressFullMessagePolicy(AddressFullMessagePolicy.BLOCK);
+ }
+
+ @Override
+ @Before
+ public void setUp() throws Exception {
+ super.setUp();
+ this.makeSureCoreQueueExist("QUEUE.A");
+ }
+
+ @Override
+ @After
+ public void tearDown() throws Exception {
+ try {
+ if (flowControlConnection != null) {
+ TcpTransport t = flowControlConnection.getTransport().narrow(TcpTransport.class);
+ try {
+ flowControlConnection.getTransport().stop();
+ flowControlConnection.close();
+ } catch (Throwable ignored) {
+ }
+ }
+ } finally {
+ super.tearDown();
+ }
+ }
+
+ //set ttl to 1000
+ @Override
+ protected String getConnectionUrl() {
+ return urlString + "&wireFormat.maxInactivityDuration=1000&wireFormat.maxInactivityDurationInitalDelay=1000";
+ }
+
+ @Test
+ public void testProducerBlockWontGetTimeout() throws Exception {
+
+ flowControlConnection = (ActiveMQConnection) factory.createConnection();
+ Connection consumerConnection = factory.createConnection();
+ Thread fillThread = null;
+ AtomicBoolean keepGoing = new AtomicBoolean(true);
+ try {
+ flowControlConnection.start();
+
+ final Session session = flowControlConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ final MessageProducer producer = session.createProducer(queueA);
+ producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
+ final String text = "Hello World";
+ final int num = 10;
+
+ fillThread = new Thread("Fill thread.") {
+ @Override
+ public void run() {
+ try {
+ for (int i = 0; i < num && keepGoing.get(); i++) {
+ producer.send(session.createTextMessage(text + i));
+ }
+ } catch (JMSException e) {
+ }
+ }
+ };
+
+ fillThread.start();
+
+ //longer enough than TTL (1000)
+ Thread.sleep(4000);
+
+ //receive messages and unblock the producer
+ consumerConnection.start();
+ Session consumerSession = consumerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageConsumer consumer = consumerSession.createConsumer(queueA);
+
+ for (int i = 0; i < num; i++) {
+ TextMessage m = (TextMessage) consumer.receive(5000);
+ assertNotNull(m);
+ assertEquals("Hello World" + i, m.getText());
+ }
+ assertNull(consumer.receive(3));
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ } finally {
+
+ if (fillThread != null) {
+ keepGoing.set(false);
+ fillThread.interrupt();
+ fillThread.join();
+ }
+ try {
+ flowControlConnection.close();
+ flowControlConnection = null;
+ } catch (Throwable t) {
+ }
+ try {
+ consumerConnection.close();
+ } catch (Throwable t) {
+ }
+ }
+ }
+}