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

Revert "ARTEMIS-539" 

This reverts commit a3efafd975.
This reverts commit cf3396a3a6.
This reverts commit 17ea05bce6.
This reverts commit af4aa9fcb6.
This commit is contained in:
Clebert Suconic 2016-06-06 14:11:48 -04:00
parent 29a0d273dd
commit 8fdd1f6496
4 changed files with 5 additions and 193 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientProducerCreditManagerImpl.java
View File

@ -79,10 +79,6 @@ public class ClientProducerCreditManagerImpl implements ClientProducerCreditMana
// while this is still sending requests causing a dead lock
if (needInit) {
credits.init(context);
// Setting default address on the session
session.setAddress(null, address);
}
return credits;

7
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientSessionImpl.java
View File

@ -1052,14 +1052,11 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
@Override
public void setAddress(final Message message, final SimpleString address) {
if (defaultAddress == null) {
logger.tracef("setAddress() Setting default address as %s", address);
defaultAddress = address;
if (message != null) {
message.setAddress(address);
}
message.setAddress(address);
}
else if (message != null) {
else {
if (!address.equals(defaultAddress)) {
message.setAddress(address);
}

16
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
View File

@ -1233,7 +1233,9 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
SimpleString address = message.getAddress();
checkDefaultAddress(address);
if (defaultAddress == null && address != null) {
defaultAddress = address;
}
if (address == null) {
if (message.isDurable()) {
@ -1267,14 +1269,6 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
return result;
}
protected void checkDefaultAddress(SimpleString address) {
logger.tracef("checkDefaultAddress %s, defaultAddress=%s", address, defaultAddress);
if (defaultAddress == null && address != null) {
logger.tracef("checkDefaultAddress setting as %s", address);
defaultAddress = address;
}
}
@Override
public void sendContinuations(final int packetSize,
final long messageBodySize,
@ -1304,10 +1298,6 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
@Override
public void requestProducerCredits(final SimpleString address, final int credits) throws Exception {
// When the client gets the producer credits it already has the intention of sending messages
// and it will already register the default address on the core protocol
// hence we need to set it here when we request credits as well
checkDefaultAddress(address);
PagingStore store = server.getPagingManager().getPageStore(address);
if (!store.checkMemory(new Runnable() {

171
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/MultipleProducersTest.java
View File

@ -1,171 +0,0 @@
/**
* 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.client;
import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.management.ResourceNames;
import org.apache.activemq.artemis.api.jms.management.JMSQueueControl;
import org.apache.activemq.artemis.core.settings.HierarchicalRepository;
import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.core.settings.impl.SlowConsumerPolicy;
import org.apache.activemq.artemis.tests.util.JMSTestBase;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
public class MultipleProducersTest extends JMSTestBase {
public Connection conn;
public Queue queueOne = null;
public Queue queueTwo = null;
public Session session = null;
public SimpleString dlq = new SimpleString("jms.queue.DLQ");
public SimpleString expiryQueue = new SimpleString("jms.queue.ExpiryQueue");
public SimpleString queueOneName = new SimpleString("jms.queue.queueOne");
public SimpleString queueTwoName = new SimpleString("jms.queue.queueTwo");
public JMSQueueControl control = null;
public long queueOneMsgCount = 0;
public long queueTwoMsgCount = 0;
@Before
public void iniTest() throws Exception {
}
@Test
public void wrongQueue() throws Exception {
conn = cf.createConnection();
session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
HierarchicalRepository<AddressSettings> repos = server.getAddressSettingsRepository();
AddressSettings addressSettings = new AddressSettings();
addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.FAIL);
addressSettings.setExpiryAddress(dlq);
addressSettings.setDeadLetterAddress(expiryQueue);
addressSettings.setRedeliveryDelay(0);
addressSettings.setMessageCounterHistoryDayLimit(2);
addressSettings.setLastValueQueue(false);
addressSettings.setMaxDeliveryAttempts(10);
addressSettings.setMaxSizeBytes(1048576);
addressSettings.setPageCacheMaxSize(5);
addressSettings.setPageSizeBytes(2097152);
addressSettings.setRedistributionDelay(-1);
addressSettings.setSendToDLAOnNoRoute(false);
addressSettings.setSlowConsumerCheckPeriod(5);
addressSettings.setSlowConsumerPolicy(SlowConsumerPolicy.NOTIFY);
addressSettings.setSlowConsumerThreshold(-1);
repos.setDefault(addressSettings);
queueOne = createQueue("queueOne");
queueTwo = createQueue("queueTwo");
try {
while (true) {
sendMessage(queueOne, session);
}
}
catch (Throwable t) {
// expected
}
session.close();
conn.close();
session = null;
conn = null;
conn = cf.createConnection();
session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
// send a message to a queue which is already full
// result an exception
try {
sendMessage(queueOne, session);
Assert.fail("Exception expected");
}
catch (Exception t) {
}
// send 5 message to queueTwo
// there shoudl be 5 messages on queueTwo
for (int i = 0; i < 5; i++) {
sendMessage(queueTwo, session);
}
// before sending any messages to queueOne it has to be drained.
// after draining queueOne send 5 message to queueOne
queueTwoMsgCount = server.locateQueue(queueTwoName).getMessageCount();
control = (JMSQueueControl) server.getManagementService().getResource(ResourceNames.JMS_QUEUE + queueOne.getQueueName());
control.removeMessages(null);
for (int i = 0; i < 5; i++) {
sendMessage(queueOne, session);
}
// at the end of the test there shoudl be 5 message on queueOne and 5 messages on queueTwo
session.close();
conn.close();
queueOneMsgCount = server.locateQueue(queueOneName).getMessageCount();
queueTwoMsgCount = server.locateQueue(queueTwoName).getMessageCount();
Assert.assertEquals("queueTwo message count", 5, queueTwoMsgCount);
Assert.assertEquals("queueOne message count", 5, queueOneMsgCount);
}
private void sendMessage(Queue queue, Session session) throws Exception {
MessageProducer mp = session.createProducer(queue);
try {
mp.setDisableMessageID(true);
mp.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
mp.setPriority(Message.DEFAULT_PRIORITY);
mp.setTimeToLive(Message.DEFAULT_TIME_TO_LIVE);
mp.send(session.createTextMessage("This is message for " + queue.getQueueName()));
}
finally {
mp.close();
}
}
}