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 2018/09/27 21:29:54 UTC
[3/4] activemq-artemis git commit: ARTEMIS-1545 Support JMS 2.0
Completion Listener for Exceptions
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/e4ba48a3/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/ServerSessionPacketHandler.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/ServerSessionPacketHandler.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/ServerSessionPacketHandler.java
index 37564b5..f5756f2 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/ServerSessionPacketHandler.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/ServerSessionPacketHandler.java
@@ -41,6 +41,7 @@ import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.CreateQueu
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.CreateQueueMessage_V2;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.CreateSharedQueueMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.CreateSharedQueueMessage_V2;
+import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ActiveMQExceptionMessage_V2;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.NullResponseMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.RollbackMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionAcknowledgeMessage;
@@ -76,11 +77,13 @@ import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionXAG
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionXAJoinMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionXAPrepareMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionXAResponseMessage;
+import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionXAResponseMessage_V2;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionXAResumeMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionXARollbackMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionXASetTimeoutMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionXASetTimeoutResponseMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionXAStartMessage;
+import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.NullResponseMessage_V2;
import org.apache.activemq.artemis.core.remoting.CloseListener;
import org.apache.activemq.artemis.core.remoting.FailureListener;
import org.apache.activemq.artemis.core.remoting.impl.netty.NettyConnection;
@@ -313,7 +316,7 @@ public class ServerSessionPacketHandler implements ChannelHandler {
requiresResponse = message.isRequiresResponse();
sendContinuations(message.getPacketSize(), message.getMessageBodySize(), message.getBody(), message.isContinues());
if (requiresResponse) {
- response = new NullResponseMessage();
+ response = createNullResponseMessage(packet);
}
break;
}
@@ -342,7 +345,7 @@ public class ServerSessionPacketHandler implements ChannelHandler {
requiresResponse = request.isRequiresResponse();
session.createAddress(request.getAddress(), request.getRoutingTypes(), request.isAutoCreated());
if (requiresResponse) {
- response = new NullResponseMessage();
+ response = createNullResponseMessage(packet);
}
break;
}
@@ -351,7 +354,7 @@ public class ServerSessionPacketHandler implements ChannelHandler {
requiresResponse = request.isRequiresResponse();
session.createQueue(request.getAddress(), request.getQueueName(), RoutingType.MULTICAST, request.getFilterString(), request.isTemporary(), request.isDurable());
if (requiresResponse) {
- response = new NullResponseMessage();
+ response = createNullResponseMessage(packet);
}
break;
}
@@ -361,7 +364,7 @@ public class ServerSessionPacketHandler implements ChannelHandler {
session.createQueue(request.getAddress(), request.getQueueName(), request.getRoutingType(), request.getFilterString(), request.isTemporary(), request.isDurable(), request.getMaxConsumers(), request.isPurgeOnNoConsumers(),
request.isExclusive(), request.isLastValue(), request.isAutoCreated());
if (requiresResponse) {
- response = new NullResponseMessage();
+ response = createNullResponseMessage(packet);
}
break;
}
@@ -373,7 +376,7 @@ public class ServerSessionPacketHandler implements ChannelHandler {
session.createSharedQueue(request.getAddress(), request.getQueueName(), request.isDurable(), request.getFilterString());
}
if (requiresResponse) {
- response = new NullResponseMessage();
+ response = createNullResponseMessage(packet);
}
break;
}
@@ -385,7 +388,7 @@ public class ServerSessionPacketHandler implements ChannelHandler {
session.createSharedQueue(request.getAddress(), request.getQueueName(), request.getRoutingType(), request.getFilterString(), request.isDurable(), request.getMaxConsumers(), request.isPurgeOnNoConsumers(), request.isExclusive(), request.isLastValue());
}
if (requiresResponse) {
- response = new NullResponseMessage();
+ response = createNullResponseMessage(packet);
}
break;
}
@@ -393,7 +396,7 @@ public class ServerSessionPacketHandler implements ChannelHandler {
requiresResponse = true;
SessionDeleteQueueMessage request = (SessionDeleteQueueMessage) packet;
session.deleteQueue(request.getQueueName());
- response = new NullResponseMessage();
+ response = createNullResponseMessage(packet);
break;
}
case SESS_QUEUEQUERY: {
@@ -453,62 +456,62 @@ public class ServerSessionPacketHandler implements ChannelHandler {
case SESS_COMMIT: {
requiresResponse = true;
session.commit();
- response = new NullResponseMessage();
+ response = createNullResponseMessage(packet);
break;
}
case SESS_ROLLBACK: {
requiresResponse = true;
session.rollback(((RollbackMessage) packet).isConsiderLastMessageAsDelivered());
- response = new NullResponseMessage();
+ response = createNullResponseMessage(packet);
break;
}
case SESS_XA_COMMIT: {
requiresResponse = true;
SessionXACommitMessage message = (SessionXACommitMessage) packet;
session.xaCommit(message.getXid(), message.isOnePhase());
- response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
+ response = createSessionXAResponseMessage(packet);
break;
}
case SESS_XA_END: {
requiresResponse = true;
SessionXAEndMessage message = (SessionXAEndMessage) packet;
session.xaEnd(message.getXid());
- response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
+ response = createSessionXAResponseMessage(packet);
break;
}
case SESS_XA_FORGET: {
requiresResponse = true;
SessionXAForgetMessage message = (SessionXAForgetMessage) packet;
session.xaForget(message.getXid());
- response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
+ response = createSessionXAResponseMessage(packet);
break;
}
case SESS_XA_JOIN: {
requiresResponse = true;
SessionXAJoinMessage message = (SessionXAJoinMessage) packet;
session.xaJoin(message.getXid());
- response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
+ response = createSessionXAResponseMessage(packet);
break;
}
case SESS_XA_RESUME: {
requiresResponse = true;
SessionXAResumeMessage message = (SessionXAResumeMessage) packet;
session.xaResume(message.getXid());
- response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
+ response = createSessionXAResponseMessage(packet);
break;
}
case SESS_XA_ROLLBACK: {
requiresResponse = true;
SessionXARollbackMessage message = (SessionXARollbackMessage) packet;
session.xaRollback(message.getXid());
- response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
+ response = createSessionXAResponseMessage(packet);
break;
}
case SESS_XA_START: {
requiresResponse = true;
SessionXAStartMessage message = (SessionXAStartMessage) packet;
session.xaStart(message.getXid());
- response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
+ response = createSessionXAResponseMessage(packet);
break;
}
case SESS_XA_FAILED: {
@@ -521,14 +524,14 @@ public class ServerSessionPacketHandler implements ChannelHandler {
case SESS_XA_SUSPEND: {
requiresResponse = true;
session.xaSuspend();
- response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
+ response = createSessionXAResponseMessage(packet);
break;
}
case SESS_XA_PREPARE: {
requiresResponse = true;
SessionXAPrepareMessage message = (SessionXAPrepareMessage) packet;
session.xaPrepare(message.getXid());
- response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
+ response = createSessionXAResponseMessage(packet);
break;
}
case SESS_XA_INDOUBT_XIDS: {
@@ -557,14 +560,14 @@ public class ServerSessionPacketHandler implements ChannelHandler {
case SESS_STOP: {
requiresResponse = true;
session.stop();
- response = new NullResponseMessage();
+ response = createNullResponseMessage(packet);
break;
}
case SESS_CLOSE: {
requiresResponse = true;
session.close(false);
// removeConnectionListeners();
- response = new NullResponseMessage();
+ response = createNullResponseMessage(packet);
flush = true;
closeChannel = true;
break;
@@ -574,7 +577,7 @@ public class ServerSessionPacketHandler implements ChannelHandler {
requiresResponse = message.isRequiresResponse();
session.individualAcknowledge(message.getConsumerID(), message.getMessageID());
if (requiresResponse) {
- response = new NullResponseMessage();
+ response = createNullResponseMessage(packet);
}
break;
}
@@ -582,7 +585,7 @@ public class ServerSessionPacketHandler implements ChannelHandler {
requiresResponse = true;
SessionConsumerCloseMessage message = (SessionConsumerCloseMessage) packet;
session.closeConsumer(message.getConsumerID());
- response = new NullResponseMessage();
+ response = createNullResponseMessage(packet);
break;
}
case SESS_FORCE_CONSUMER_DELIVERY: {
@@ -591,7 +594,7 @@ public class ServerSessionPacketHandler implements ChannelHandler {
break;
}
case PacketImpl.SESS_ADD_METADATA: {
- response = new NullResponseMessage();
+ response = createNullResponseMessage(packet);
SessionAddMetaDataMessage message = (SessionAddMetaDataMessage) packet;
session.addMetaData(message.getKey(), message.getData());
break;
@@ -600,7 +603,7 @@ public class ServerSessionPacketHandler implements ChannelHandler {
requiresResponse = true;
SessionAddMetaDataMessageV2 message = (SessionAddMetaDataMessageV2) packet;
if (message.isRequiresConfirmations()) {
- response = new NullResponseMessage();
+ response = createNullResponseMessage(packet);
}
session.addMetaData(message.getKey(), message.getData());
break;
@@ -609,7 +612,7 @@ public class ServerSessionPacketHandler implements ChannelHandler {
requiresResponse = true;
SessionUniqueAddMetaDataMessage message = (SessionUniqueAddMetaDataMessage) packet;
if (session.addUniqueMetaData(message.getKey(), message.getData())) {
- response = new NullResponseMessage();
+ response = createNullResponseMessage(packet);
} else {
response = new ActiveMQExceptionMessage(ActiveMQMessageBundle.BUNDLE.duplicateMetadata(message.getKey(), message.getData()));
}
@@ -617,15 +620,15 @@ public class ServerSessionPacketHandler implements ChannelHandler {
}
}
} catch (ActiveMQIOErrorException e) {
- response = onActiveMQIOErrorExceptionWhileHandlePacket(e, requiresResponse, response, this.session);
+ response = onActiveMQIOErrorExceptionWhileHandlePacket(packet, e, requiresResponse, response, this.session);
} catch (ActiveMQXAException e) {
- response = onActiveMQXAExceptionWhileHandlePacket(e, requiresResponse, response);
+ response = onActiveMQXAExceptionWhileHandlePacket(packet, e, requiresResponse, response);
} catch (ActiveMQQueueMaxConsumerLimitReached e) {
- response = onActiveMQQueueMaxConsumerLimitReachedWhileHandlePacket(e, requiresResponse, response);
+ response = onActiveMQQueueMaxConsumerLimitReachedWhileHandlePacket(packet, e, requiresResponse, response);
} catch (ActiveMQException e) {
- response = onActiveMQExceptionWhileHandlePacket(e, requiresResponse, response);
+ response = onActiveMQExceptionWhileHandlePacket(packet, e, requiresResponse, response);
} catch (Throwable t) {
- response = onCatchThrowableWhileHandlePacket(t, requiresResponse, response, this.session);
+ response = onCatchThrowableWhileHandlePacket(packet, t, requiresResponse, response, this.session);
}
sendResponse(packet, response, flush, closeChannel);
} finally {
@@ -633,6 +636,26 @@ public class ServerSessionPacketHandler implements ChannelHandler {
}
}
+ private Packet createNullResponseMessage(Packet packet) {
+ final Packet response;
+ if (!packet.isResponseAsync() || channel.getConnection().isVersionBeforeAsyncResponseChange()) {
+ response = new NullResponseMessage();
+ } else {
+ response = new NullResponseMessage_V2(packet.getCorrelationID());
+ }
+ return response;
+ }
+
+ private Packet createSessionXAResponseMessage(Packet packet) {
+ Packet response;
+ if (packet.isResponseAsync()) {
+ response = new SessionXAResponseMessage_V2(packet.getCorrelationID(), false, XAResource.XA_OK, null);
+ } else {
+ response = new SessionXAResponseMessage(false, XAResource.XA_OK, null);
+ }
+ return response;
+ }
+
private void onSessionAcknowledge(Packet packet) {
this.storageManager.setContext(session.getSessionContext());
try {
@@ -643,18 +666,18 @@ public class ServerSessionPacketHandler implements ChannelHandler {
requiresResponse = message.isRequiresResponse();
this.session.acknowledge(message.getConsumerID(), message.getMessageID());
if (requiresResponse) {
- response = new NullResponseMessage();
+ response = createNullResponseMessage(packet);
}
} catch (ActiveMQIOErrorException e) {
- response = onActiveMQIOErrorExceptionWhileHandlePacket(e, requiresResponse, response, this.session);
+ response = onActiveMQIOErrorExceptionWhileHandlePacket(packet, e, requiresResponse, response, this.session);
} catch (ActiveMQXAException e) {
- response = onActiveMQXAExceptionWhileHandlePacket(e, requiresResponse, response);
+ response = onActiveMQXAExceptionWhileHandlePacket(packet, e, requiresResponse, response);
} catch (ActiveMQQueueMaxConsumerLimitReached e) {
- response = onActiveMQQueueMaxConsumerLimitReachedWhileHandlePacket(e, requiresResponse, response);
+ response = onActiveMQQueueMaxConsumerLimitReachedWhileHandlePacket(packet, e, requiresResponse, response);
} catch (ActiveMQException e) {
- response = onActiveMQExceptionWhileHandlePacket(e, requiresResponse, response);
+ response = onActiveMQExceptionWhileHandlePacket(packet, e, requiresResponse, response);
} catch (Throwable t) {
- response = onCatchThrowableWhileHandlePacket(t, requiresResponse, response, this.session);
+ response = onCatchThrowableWhileHandlePacket(packet, t, requiresResponse, response, this.session);
}
sendResponse(packet, response, false, false);
} finally {
@@ -672,18 +695,18 @@ public class ServerSessionPacketHandler implements ChannelHandler {
requiresResponse = message.isRequiresResponse();
this.session.send(EmbedMessageUtil.extractEmbedded(message.getMessage()), this.direct);
if (requiresResponse) {
- response = new NullResponseMessage();
+ response = createNullResponseMessage(packet);
}
} catch (ActiveMQIOErrorException e) {
- response = onActiveMQIOErrorExceptionWhileHandlePacket(e, requiresResponse, response, this.session);
+ response = onActiveMQIOErrorExceptionWhileHandlePacket(packet, e, requiresResponse, response, this.session);
} catch (ActiveMQXAException e) {
- response = onActiveMQXAExceptionWhileHandlePacket(e, requiresResponse, response);
+ response = onActiveMQXAExceptionWhileHandlePacket(packet, e, requiresResponse, response);
} catch (ActiveMQQueueMaxConsumerLimitReached e) {
- response = onActiveMQQueueMaxConsumerLimitReachedWhileHandlePacket(e, requiresResponse, response);
+ response = onActiveMQQueueMaxConsumerLimitReachedWhileHandlePacket(packet, e, requiresResponse, response);
} catch (ActiveMQException e) {
- response = onActiveMQExceptionWhileHandlePacket(e, requiresResponse, response);
+ response = onActiveMQExceptionWhileHandlePacket(packet, e, requiresResponse, response);
} catch (Throwable t) {
- response = onCatchThrowableWhileHandlePacket(t, requiresResponse, response, this.session);
+ response = onCatchThrowableWhileHandlePacket(packet, t, requiresResponse, response, this.session);
}
sendResponse(packet, response, false, false);
} finally {
@@ -700,15 +723,15 @@ public class ServerSessionPacketHandler implements ChannelHandler {
SessionRequestProducerCreditsMessage message = (SessionRequestProducerCreditsMessage) packet;
session.requestProducerCredits(message.getAddress(), message.getCredits());
} catch (ActiveMQIOErrorException e) {
- response = onActiveMQIOErrorExceptionWhileHandlePacket(e, requiresResponse, response, this.session);
+ response = onActiveMQIOErrorExceptionWhileHandlePacket(packet, e, requiresResponse, response, this.session);
} catch (ActiveMQXAException e) {
- response = onActiveMQXAExceptionWhileHandlePacket(e, requiresResponse, response);
+ response = onActiveMQXAExceptionWhileHandlePacket(packet, e, requiresResponse, response);
} catch (ActiveMQQueueMaxConsumerLimitReached e) {
- response = onActiveMQQueueMaxConsumerLimitReachedWhileHandlePacket(e, requiresResponse, response);
+ response = onActiveMQQueueMaxConsumerLimitReachedWhileHandlePacket(packet, e, requiresResponse, response);
} catch (ActiveMQException e) {
- response = onActiveMQExceptionWhileHandlePacket(e, requiresResponse, response);
+ response = onActiveMQExceptionWhileHandlePacket(packet, e, requiresResponse, response);
} catch (Throwable t) {
- response = onCatchThrowableWhileHandlePacket(t, requiresResponse, response, this.session);
+ response = onCatchThrowableWhileHandlePacket(packet, t, requiresResponse, response, this.session);
}
sendResponse(packet, response, false, false);
} finally {
@@ -725,15 +748,15 @@ public class ServerSessionPacketHandler implements ChannelHandler {
SessionConsumerFlowCreditMessage message = (SessionConsumerFlowCreditMessage) packet;
session.receiveConsumerCredits(message.getConsumerID(), message.getCredits());
} catch (ActiveMQIOErrorException e) {
- response = onActiveMQIOErrorExceptionWhileHandlePacket(e, requiresResponse, response, this.session);
+ response = onActiveMQIOErrorExceptionWhileHandlePacket(packet, e, requiresResponse, response, this.session);
} catch (ActiveMQXAException e) {
- response = onActiveMQXAExceptionWhileHandlePacket(e, requiresResponse, response);
+ response = onActiveMQXAExceptionWhileHandlePacket(packet, e, requiresResponse, response);
} catch (ActiveMQQueueMaxConsumerLimitReached e) {
- response = onActiveMQQueueMaxConsumerLimitReachedWhileHandlePacket(e, requiresResponse, response);
+ response = onActiveMQQueueMaxConsumerLimitReachedWhileHandlePacket(packet, e, requiresResponse, response);
} catch (ActiveMQException e) {
- response = onActiveMQExceptionWhileHandlePacket(e, requiresResponse, response);
+ response = onActiveMQExceptionWhileHandlePacket(packet, e, requiresResponse, response);
} catch (Throwable t) {
- response = onCatchThrowableWhileHandlePacket(t, requiresResponse, response, this.session);
+ response = onCatchThrowableWhileHandlePacket(packet, t, requiresResponse, response, this.session);
}
sendResponse(packet, response, false, false);
} finally {
@@ -742,50 +765,68 @@ public class ServerSessionPacketHandler implements ChannelHandler {
}
- private static Packet onActiveMQIOErrorExceptionWhileHandlePacket(ActiveMQIOErrorException e,
+ private static Packet onActiveMQIOErrorExceptionWhileHandlePacket(Packet packet,
+ ActiveMQIOErrorException e,
boolean requiresResponse,
Packet response,
ServerSession session) {
session.markTXFailed(e);
if (requiresResponse) {
logger.debug("Sending exception to client", e);
- response = new ActiveMQExceptionMessage(e);
+ response = convertToExceptionPacket(packet, e);
} else {
ActiveMQServerLogger.LOGGER.caughtException(e);
}
return response;
}
- private static Packet onActiveMQXAExceptionWhileHandlePacket(ActiveMQXAException e,
+ private static Packet onActiveMQXAExceptionWhileHandlePacket(Packet packet,
+ ActiveMQXAException e,
boolean requiresResponse,
Packet response) {
if (requiresResponse) {
logger.debug("Sending exception to client", e);
- response = new SessionXAResponseMessage(true, e.errorCode, e.getMessage());
+ if (packet.isResponseAsync()) {
+ response = new SessionXAResponseMessage_V2(packet.getCorrelationID(), true, e.errorCode, e.getMessage());
+ } else {
+ response = new SessionXAResponseMessage(true, e.errorCode, e.getMessage());
+ }
} else {
ActiveMQServerLogger.LOGGER.caughtXaException(e);
}
return response;
}
- private static Packet onActiveMQQueueMaxConsumerLimitReachedWhileHandlePacket(ActiveMQQueueMaxConsumerLimitReached e,
+ private static Packet onActiveMQQueueMaxConsumerLimitReachedWhileHandlePacket(Packet packet,
+ ActiveMQQueueMaxConsumerLimitReached e,
boolean requiresResponse,
Packet response) {
if (requiresResponse) {
logger.debug("Sending exception to client", e);
- response = new ActiveMQExceptionMessage(e);
+ response = convertToExceptionPacket(packet, e);
} else {
ActiveMQServerLogger.LOGGER.caughtException(e);
}
return response;
}
- private static Packet onActiveMQExceptionWhileHandlePacket(ActiveMQException e,
+ private static Packet convertToExceptionPacket(Packet packet, ActiveMQException e) {
+ Packet response;
+ if (packet.isResponseAsync()) {
+ response = new ActiveMQExceptionMessage_V2(packet.getCorrelationID(), e);
+ } else {
+ response = new ActiveMQExceptionMessage(e);
+ }
+ return response;
+ }
+
+ private static Packet onActiveMQExceptionWhileHandlePacket(Packet packet,
+ ActiveMQException e,
boolean requiresResponse,
Packet response) {
if (requiresResponse) {
logger.debug("Sending exception to client", e);
- response = new ActiveMQExceptionMessage(e);
+ response = convertToExceptionPacket(packet, e);
} else {
if (e.getType() == ActiveMQExceptionType.QUEUE_EXISTS) {
logger.debug("Caught exception", e);
@@ -796,7 +837,8 @@ public class ServerSessionPacketHandler implements ChannelHandler {
return response;
}
- private static Packet onCatchThrowableWhileHandlePacket(Throwable t,
+ private static Packet onCatchThrowableWhileHandlePacket(Packet packet,
+ Throwable t,
boolean requiresResponse,
Packet response,
ServerSession session) {
@@ -805,7 +847,7 @@ public class ServerSessionPacketHandler implements ChannelHandler {
ActiveMQServerLogger.LOGGER.sendingUnexpectedExceptionToClient(t);
ActiveMQException activeMQInternalErrorException = new ActiveMQInternalErrorException();
activeMQInternalErrorException.initCause(t);
- response = new ActiveMQExceptionMessage(activeMQInternalErrorException);
+ response = convertToExceptionPacket(packet, activeMQInternalErrorException);
} else {
ActiveMQServerLogger.LOGGER.caughtException(t);
}
@@ -827,12 +869,11 @@ public class ServerSessionPacketHandler implements ChannelHandler {
public void onError(final int errorCode, final String errorMessage) {
ActiveMQServerLogger.LOGGER.errorProcessingIOCallback(errorCode, errorMessage);
- ActiveMQExceptionMessage exceptionMessage = new ActiveMQExceptionMessage(ActiveMQExceptionType.createException(errorCode, errorMessage));
-
- doConfirmAndResponse(confirmPacket, exceptionMessage, flush, closeChannel);
+ Packet exceptionPacket = convertToExceptionPacket(confirmPacket, ActiveMQExceptionType.createException(errorCode, errorMessage));
+ doConfirmAndResponse(confirmPacket, exceptionPacket, flush, closeChannel);
if (logger.isTraceEnabled()) {
- logger.trace("ServerSessionPacketHandler::exception response sent::" + exceptionMessage);
+ logger.trace("ServerSessionPacketHandler::exception response sent::" + exceptionPacket);
}
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/e4ba48a3/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 36e5c33..7acfe0d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -126,7 +126,7 @@
<activemq.version.majorVersion>1</activemq.version.majorVersion>
<activemq.version.minorVersion>0</activemq.version.minorVersion>
<activemq.version.microVersion>0</activemq.version.microVersion>
- <activemq.version.incrementingVersion>129,128,127,126,125,124,123,122</activemq.version.incrementingVersion>
+ <activemq.version.incrementingVersion>130,129,128,127,126,125,124,123,122</activemq.version.incrementingVersion>
<activemq.version.versionTag>${project.version}</activemq.version.versionTag>
<ActiveMQ-Version>${project.version}(${activemq.version.incrementingVersion})</ActiveMQ-Version>
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/e4ba48a3/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/util/BackupSyncDelay.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/util/BackupSyncDelay.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/util/BackupSyncDelay.java
index e4afb5b..c7ed869 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/util/BackupSyncDelay.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/util/BackupSyncDelay.java
@@ -25,6 +25,7 @@ import org.apache.activemq.artemis.core.protocol.core.ChannelHandler;
import org.apache.activemq.artemis.core.protocol.core.CommandConfirmationHandler;
import org.apache.activemq.artemis.core.protocol.core.CoreRemotingConnection;
import org.apache.activemq.artemis.core.protocol.core.Packet;
+import org.apache.activemq.artemis.core.protocol.core.ResponseHandler;
import org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationResponseMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationResponseMessageV2;
@@ -315,6 +316,11 @@ public class BackupSyncDelay implements Interceptor {
}
@Override
+ public void setResponseHandler(ResponseHandler handler) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
public void flushConfirmations() {
throw new UnsupportedOperationException();
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/e4ba48a3/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/jms2client/JmsProducerCompletionListenerTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/jms2client/JmsProducerCompletionListenerTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/jms2client/JmsProducerCompletionListenerTest.java
index d3951f2..3020310 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/jms2client/JmsProducerCompletionListenerTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/jms2client/JmsProducerCompletionListenerTest.java
@@ -167,7 +167,24 @@ public class JmsProducerCompletionListenerTest extends JMSTestBase {
@Override
public void onException(Message message, Exception exception) {
- // TODO Auto-generated method stub
+ latch.countDown();
+ try {
+ switch (call) {
+ case 0:
+ context.rollback();
+ break;
+ case 1:
+ context.commit();
+ break;
+ case 2:
+ context.close();
+ break;
+ default:
+ throw new IllegalArgumentException("call code " + call);
+ }
+ } catch (Exception error1) {
+ this.error = error1;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/e4ba48a3/tests/jms-tests/src/test/java/org/apache/activemq/artemis/jms/tests/SecurityTest.java
----------------------------------------------------------------------
diff --git a/tests/jms-tests/src/test/java/org/apache/activemq/artemis/jms/tests/SecurityTest.java b/tests/jms-tests/src/test/java/org/apache/activemq/artemis/jms/tests/SecurityTest.java
index e83d815..0fd469c 100644
--- a/tests/jms-tests/src/test/java/org/apache/activemq/artemis/jms/tests/SecurityTest.java
+++ b/tests/jms-tests/src/test/java/org/apache/activemq/artemis/jms/tests/SecurityTest.java
@@ -16,12 +16,23 @@
*/
package org.apache.activemq.artemis.jms.tests;
+import static org.junit.Assert.fail;
+
+import javax.jms.CompletionListener;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
import javax.jms.IllegalStateException;
import javax.jms.JMSSecurityException;
+import javax.jms.Message;
+import javax.jms.MessageProducer;
import javax.jms.Session;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
import org.apache.activemq.artemis.jms.client.DefaultConnectionProperties;
import org.apache.activemq.artemis.jms.tests.util.ProxyAssertSupport;
@@ -68,9 +79,9 @@ public class SecurityTest extends JMSTestCase {
}
- /**
- * Login with no user, no password Should allow login (equivalent to guest)
- */
+ /**
+ * Login with no user, no password Should allow login (equivalent to guest)
+ */
@Test
public void testLoginNoUserNoPassword() throws Exception {
createConnection();
@@ -170,6 +181,71 @@ public class SecurityTest extends JMSTestCase {
}
}
+ /**
+ * Login with valid user and password
+ * But try send to address not authorised - Persistent
+ * Should not allow and should throw exception
+ */
+ @Test
+ public void testLoginValidUserAndPasswordButNotAuthorisedToSend() throws Exception {
+ ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
+ Connection connection = connectionFactory.createConnection("guest", "guest");
+ Session session = connection.createSession();
+ Destination destination = session.createQueue("guest.cannot.send");
+ MessageProducer messageProducer = session.createProducer(destination);
+ try {
+ messageProducer.send(session.createTextMessage("hello"));
+ fail("JMSSecurityException expected as guest is not allowed to send");
+ } catch (JMSSecurityException activeMQSecurityException) {
+ //pass
+ }
+ connection.close();
+ }
+
+ /**
+ * Login with valid user and password
+ * But try send to address not authorised - Non Persistent.
+ * Should have same behaviour as Persistent with exception on send.
+ */
+ @Test
+ public void testLoginValidUserAndPasswordButNotAuthorisedToSendNonPersistent() throws Exception {
+ ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
+ connectionFactory.setConfirmationWindowSize(100);
+ connectionFactory.setBlockOnDurableSend(false);
+ connectionFactory.setBlockOnNonDurableSend(false);
+ Connection connection = connectionFactory.createConnection("guest", "guest");
+ Session session = connection.createSession();
+ Destination destination = session.createQueue("guest.cannot.send");
+ MessageProducer messageProducer = session.createProducer(destination);
+ messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
+ try {
+ AtomicReference<Exception> e = new AtomicReference<>();
+ // messageProducer.send(session.createTextMessage("hello"));
+
+ CountDownLatch countDownLatch = new CountDownLatch(1);
+ messageProducer.send(session.createTextMessage("hello"), new CompletionListener() {
+ @Override
+ public void onCompletion(Message message) {
+ countDownLatch.countDown();
+ }
+
+ @Override
+ public void onException(Message message, Exception exception) {
+ e.set(exception);
+ countDownLatch.countDown();
+ }
+ });
+ countDownLatch.await(10, TimeUnit.SECONDS);
+ if (e.get() != null) {
+ throw e.get();
+ }
+ fail("JMSSecurityException expected as guest is not allowed to send");
+ } catch (JMSSecurityException activeMQSecurityException) {
+ activeMQSecurityException.printStackTrace();
+ }
+ connection.close();
+ }
+
/* Now some client id tests */
/**
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/e4ba48a3/tests/jms-tests/src/test/resources/broker.xml
----------------------------------------------------------------------
diff --git a/tests/jms-tests/src/test/resources/broker.xml b/tests/jms-tests/src/test/resources/broker.xml
index 733e8c3..644ce83 100644
--- a/tests/jms-tests/src/test/resources/broker.xml
+++ b/tests/jms-tests/src/test/resources/broker.xml
@@ -54,6 +54,16 @@
<permission type="browse" roles="guest,def"/>
<permission type="send" roles="guest,def"/>
</security-setting>
+
+ <security-setting match="guest.cannot.send">
+ <permission type="createDurableQueue" roles="guest,def"/>
+ <permission type="deleteDurableQueue" roles="guest,def"/>
+ <permission type="createNonDurableQueue" roles="guest,def"/>
+ <permission type="deleteNonDurableQueue" roles="guest,def"/>
+ <permission type="consume" roles="guest,def"/>
+ <permission type="browse" roles="guest,def"/>
+ <permission type="send" roles="def"/>
+ </security-setting>
</security-settings>
</core>
</configuration>
\ No newline at end of file