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

This closes #1801

This commit is contained in:
Clebert Suconic 2018-01-23 11:51:42 -05:00
parent f0d2fae4b3 56d68f0b72
commit 6ad29e4039
1 changed files with 36 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/amq/AMQSession.java
View File

@ -18,6 +18,7 @@ package org.apache.activemq.artemis.core.protocol.openwire.amq;
import javax.jms.InvalidDestinationException; import javax.jms.InvalidDestinationException;
import javax.jms.ResourceAllocationException; import javax.jms.ResourceAllocationException;
import java.io.IOException;
import java.util.List; import java.util.List;
import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
@ -27,6 +28,7 @@ import org.apache.activemq.advisory.AdvisorySupport;
import org.apache.activemq.artemis.api.core.ActiveMQQueueExistsException; import org.apache.activemq.artemis.api.core.ActiveMQQueueExistsException;
import org.apache.activemq.artemis.api.core.RoutingType; import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString; import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.io.IOCallback;
import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools; import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
import org.apache.activemq.artemis.core.paging.PagingStore; import org.apache.activemq.artemis.core.paging.PagingStore;
import org.apache.activemq.artemis.core.postoffice.RoutingStatus; import org.apache.activemq.artemis.core.postoffice.RoutingStatus;
@ -428,26 +430,42 @@ public class AMQSession implements SessionCallback {
this.connection.getContext().setDontSendReponse(false); this.connection.getContext().setDontSendReponse(false);
connection.sendException(exceptionToSend); connection.sendException(exceptionToSend);
} else { } else {
if (sendProducerAck) { server.getStorageManager().afterCompleteOperations(new IOCallback() {
try { @Override
ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), messageSend.getSize()); public void done() {
connection.dispatchAsync(ack); if (sendProducerAck) {
} catch (Exception e) { try {
this.connection.getContext().setDontSendReponse(false); ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), messageSend.getSize());
ActiveMQServerLogger.LOGGER.warn(e.getMessage(), e); connection.dispatchAsync(ack);
connection.sendException(e); } catch (Exception e) {
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);
}
}
} }
} else {
connection.getContext().setDontSendReponse(false); @Override
try { public void onError(int errorCode, String errorMessage) {
Response response = new Response(); try {
response.setCorrelationId(messageSend.getCommandId()); final IOException e = new IOException(errorMessage);
connection.dispatchAsync(response); ActiveMQServerLogger.LOGGER.warn(errorMessage);
} catch (Exception e) { connection.serviceException(e);
ActiveMQServerLogger.LOGGER.warn(e.getMessage(), e); } catch (Exception ex) {
connection.sendException(e); ActiveMQServerLogger.LOGGER.debug(ex);
}
} }
} });
} }
} }
})) { })) {