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 #1722

This commit is contained in:
Clebert Suconic 2017-12-19 15:05:15 -05:00
parent fb34214a1a 8f5b7a1e73
commit 386f090372
32 changed files with 1006 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/client/SendAcknowledgementHandler.java
View File

@ -41,4 +41,11 @@ public interface SendAcknowledgementHandler {
* @param message message sent asynchronously
*/
void sendAcknowledged(Message message);
default void sendFailed(Message message, Exception e) {
//This is to keep old behaviour that would ack even if error,
// if anyone custom implemented this interface but doesnt update.
sendAcknowledged(message);
}
}

3
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/ActiveMQClientMessageBundle.java
View File

@ -234,4 +234,7 @@ public interface ActiveMQClientMessageBundle {
@Message(id = 119062, value = "Multi-packet transmission (e.g. Large Messages) interrupted because of a reconnection.")
ActiveMQInterruptedException packetTransmissionInterrupted();
@Message(id = 119063, value = "Cannot send a packet while response cache is full.")
IllegalStateException cannotSendPacketWhilstResponseCacheFull();
}

3
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/Channel.java
View File

@ -211,6 +211,9 @@ public interface Channel {
*/
void setCommandConfirmationHandler(CommandConfirmationHandler handler);
void setResponseHandler(ResponseHandler handler);
/**
* flushes any confirmations on to the connection.
*/

5
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/CoreRemotingConnection.java
View File

@ -36,6 +36,11 @@ public interface CoreRemotingConnection extends RemotingConnection {
return (version > 0 && version < PacketImpl.ADDRESSING_CHANGE_VERSION);
}
default boolean isVersionBeforeAsyncResponseChange() {
int version = getChannelVersion();
return (version > 0 && version < PacketImpl.ASYNC_RESPONSE_CHANGE_VERSION);
}
/**
* Sets the client protocol used on the communication. This will determine if the client has
* support for certain packet types

8
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/Packet.java
View File

@ -41,6 +41,14 @@ public interface Packet {
return INITIAL_PACKET_SIZE;
}
boolean isRequiresResponse();
boolean isResponseAsync();
long getCorrelationID();
void setCorrelationID(long correlationID);
/**
* Returns the channel id of the channel that should handle this packet.
*

30
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/ResponseHandler.java Normal file
View File

@ -0,0 +1,30 @@
/*
* 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.core.protocol.core;
/**
* A CommandConfirmationHandler is used by the channel to confirm confirmations of packets.
*/
public interface ResponseHandler {
/**
* called by channel after a confirmation has been received.
*
* @param packet the packet confirmed
*/
void responseHandler(Packet packet, Packet response);
}

89
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ActiveMQSessionContext.java
View File

@ -54,6 +54,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.wireformat.ActiveMQExceptionMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.CreateAddressMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.CreateQueueMessage;
@ -62,6 +63,7 @@ import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.CreateSess
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.CreateSharedQueueMessage_V2;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.DisconnectConsumerMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.DisconnectConsumerWithKillMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ExceptionResponseMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReattachSessionMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReattachSessionResponseMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.RollbackMessage;
@ -89,9 +91,11 @@ import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionRec
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionReceiveMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionRequestProducerCreditsMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionSendContinuationMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionSendContinuationMessage_V2;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionSendLargeMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionSendMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionSendMessage_1X;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionSendMessage_V2;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionUniqueAddMetaDataMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionXAAfterFailedMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionXACommitMessage;
@ -164,7 +168,11 @@ public class ActiveMQSessionContext extends SessionContext {
sessionChannel.setHandler(handler);
if (confirmationWindow >= 0) {
sessionChannel.setCommandConfirmationHandler(confirmationHandler);
if (sessionChannel.getConnection().isVersionBeforeAsyncResponseChange()) {
sessionChannel.setCommandConfirmationHandler(commandConfirmationHandler);
} else {
sessionChannel.setResponseHandler(responseHandler);
}
}
}
@ -181,28 +189,50 @@ public class ActiveMQSessionContext extends SessionContext {
this.killed = true;
}
private final CommandConfirmationHandler confirmationHandler = new CommandConfirmationHandler() {
private final CommandConfirmationHandler commandConfirmationHandler = new CommandConfirmationHandler() {
@Override
public void commandConfirmed(final Packet packet) {
if (packet.getType() == PacketImpl.SESS_SEND) {
public void commandConfirmed(Packet packet) {
responseHandler.responseHandler(packet, null);
}
};
private final ResponseHandler responseHandler = new ResponseHandler() {
@Override
public void responseHandler(Packet packet, Packet response) {
final ActiveMQException activeMQException;
if (response != null && response.getType() == PacketImpl.EXCEPTION_RESPONSE) {
ExceptionResponseMessage exceptionResponseMessage = (ExceptionResponseMessage) response;
activeMQException = exceptionResponseMessage.getException();
} else {
activeMQException = null;
}
if (packet.getType() == PacketImpl.SESS_SEND || packet.getType() == PacketImpl.SESS_SEND_V2) {
SessionSendMessage ssm = (SessionSendMessage) packet;
callSendAck(ssm.getHandler(), ssm.getMessage());
} else if (packet.getType() == PacketImpl.SESS_SEND_CONTINUATION) {
callSendAck(ssm.getHandler(), ssm.getMessage(), activeMQException);
} else if (packet.getType() == PacketImpl.SESS_SEND_CONTINUATION || packet.getType() == PacketImpl.SESS_SEND_CONTINUATION_V2) {
SessionSendContinuationMessage scm = (SessionSendContinuationMessage) packet;
if (!scm.isContinues()) {
callSendAck(scm.getHandler(), scm.getMessage());
callSendAck(scm.getHandler(), scm.getMessage(), activeMQException);
}
}
}
private void callSendAck(SendAcknowledgementHandler handler, final Message message) {
private void callSendAck(SendAcknowledgementHandler handler, final Message message, final Exception exception) {
if (handler != null) {
handler.sendAcknowledged(message);
if (exception == null) {
handler.sendAcknowledged(message);
} else {
handler.sendFailed(message, exception);
}
} else if (sendAckHandler != null) {
sendAckHandler.sendAcknowledged(message);
if (exception == null) {
sendAckHandler.sendAcknowledged(message);
} else {
handler.sendFailed(message, exception);
}
}
}
};
// Failover utility methods
@ -239,7 +269,11 @@ public class ActiveMQSessionContext extends SessionContext {
@Override
public void setSendAcknowledgementHandler(final SendAcknowledgementHandler handler) {
sessionChannel.setCommandConfirmationHandler(confirmationHandler);
if (sessionChannel.getConnection().isVersionBeforeAsyncResponseChange()) {
sessionChannel.setCommandConfirmationHandler(commandConfirmationHandler);
} else {
sessionChannel.setResponseHandler(responseHandler);
}
this.sendAckHandler = handler;
}
@ -454,15 +488,21 @@ public class ActiveMQSessionContext extends SessionContext {
boolean sendBlocking,
SendAcknowledgementHandler handler,
SimpleString defaultAddress) throws ActiveMQException {
SessionSendMessage packet;
final SessionSendMessage packet;
final byte expectedPacket;
if (sessionChannel.getConnection().isVersionBeforeAddressChange()) {
packet = new SessionSendMessage_1X(msgI, sendBlocking, handler);
} else {
expectedPacket = PacketImpl.NULL_RESPONSE;
} else if (sessionChannel.getConnection().isVersionBeforeAsyncResponseChange()) {
packet = new SessionSendMessage(msgI, sendBlocking, handler);
expectedPacket = PacketImpl.NULL_RESPONSE;
} else {
boolean responseRequired = confirmationWindow > 0 || sendBlocking;
packet = new SessionSendMessage_V2(msgI, responseRequired, handler);
expectedPacket = PacketImpl.SUCCESS_RESPONSE;
}
if (sendBlocking) {
sessionChannel.sendBlocking(packet, PacketImpl.NULL_RESPONSE);
sessionChannel.sendBlocking(packet, expectedPacket);
} else {
sessionChannel.sendBatched(packet);
}
@ -840,15 +880,20 @@ public class ActiveMQSessionContext extends SessionContext {
}
}
private static int sendSessionSendContinuationMessage(Channel channel,
private int sendSessionSendContinuationMessage(Channel channel,
Message msgI,
long messageBodySize,
boolean sendBlocking,
boolean lastChunk,
byte[] chunk,
SendAcknowledgementHandler messageHandler) throws ActiveMQException {
final boolean requiresResponse = lastChunk && sendBlocking;
final SessionSendContinuationMessage chunkPacket = new SessionSendContinuationMessage(msgI, chunk, !lastChunk, requiresResponse, messageBodySize, messageHandler);
final boolean requiresResponse = lastChunk || confirmationWindow > 0;
final SessionSendContinuationMessage chunkPacket;
if (sessionChannel.getConnection().isVersionBeforeAsyncResponseChange()) {
chunkPacket = new SessionSendContinuationMessage(msgI, chunk, !lastChunk, requiresResponse, messageBodySize, messageHandler);
} else {
chunkPacket = new SessionSendContinuationMessage_V2(msgI, chunk, !lastChunk, requiresResponse, messageBodySize, messageHandler);
}
final int expectedEncodeSize = chunkPacket.expectedEncodeSize();
//perform a weak form of flow control to avoid OOM on tight loops
final CoreRemotingConnection connection = channel.getConnection();
@ -865,7 +910,11 @@ public class ActiveMQSessionContext extends SessionContext {
}
if (requiresResponse) {
// When sending it blocking, only the last chunk will be blocking.
channel.sendBlocking(chunkPacket, PacketImpl.NULL_RESPONSE);
if (sendBlocking) {
channel.sendBlocking(chunkPacket, PacketImpl.SUCCESS_RESPONSE);
} else {
channel.send(chunkPacket);
}
} else {
channel.send(chunkPacket);
}

64
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ChannelImpl.java
View File

@ -36,7 +36,9 @@ 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.wireformat.ActiveMQExceptionMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ExceptionResponseMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.PacketsConfirmedMessage;
import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
import org.apache.activemq.artemis.utils.ConcurrentUtil;
@ -96,6 +98,8 @@ public final class ChannelImpl implements Channel {
private final java.util.Queue<Packet> resendCache;
private final ResponseCache responseAsyncCache;
private int firstStoredCommandID;
private final AtomicInteger lastConfirmedCommandID = new AtomicInteger(-1);
@ -122,6 +126,8 @@ public final class ChannelImpl implements Channel {
private CommandConfirmationHandler commandConfirmationHandler;
private ResponseHandler responseHandler;
private volatile boolean transferring;
private final List<Interceptor> interceptors;
@ -138,8 +144,10 @@ public final class ChannelImpl implements Channel {
if (confWindowSize != -1) {
resendCache = new ConcurrentLinkedQueue<>();
responseAsyncCache = new ResponseCache(confWindowSize);
} else {
resendCache = null;
responseAsyncCache = null;
}
this.interceptors = interceptors;
@ -176,6 +184,12 @@ public final class ChannelImpl implements Channel {
return version >= 129;
case PacketImpl.SESS_BINDINGQUERY_RESP_V4:
return version >= 129;
case PacketImpl.SESS_SEND_V2:
return version >= 130;
case PacketImpl.EXCEPTION_RESPONSE:
return version >= 130;
case PacketImpl.SUCCESS_RESPONSE:
return version >= 130;
default:
return true;
}
@ -270,6 +284,10 @@ public final class ChannelImpl implements Channel {
synchronized (sendLock) {
packet.setChannelID(id);
if (responseAsyncCache != null && packet.isRequiresResponse() && packet.isResponseAsync()) {
packet.setCorrelationID(responseAsyncCache.nextCorrelationID());
}
if (logger.isTraceEnabled()) {
logger.trace("RemotingConnectionID=" + (connection == null ? "NULL" : connection.getID()) + " Sending packet nonblocking " + packet + " on channelID=" + id);
}
@ -291,6 +309,10 @@ public final class ChannelImpl implements Channel {
if (resendCache != null && packet.isRequiresConfirmations()) {
addResendPacket(packet);
}
if (responseAsyncCache != null && packet.isRequiresResponse()) {
addResponseExpectedPacket(packet);
}
} finally {
lock.unlock();
}
@ -384,14 +406,14 @@ public final class ChannelImpl implements Channel {
long start = System.currentTimeMillis();
while (!closed && (response == null || (response.getType() != PacketImpl.EXCEPTION && response.getType() != expectedPacket)) && toWait > 0) {
while (!closed && (response == null || (response.getType() != PacketImpl.EXCEPTION && response.getType() != PacketImpl.EXCEPTION_RESPONSE && response.getType() != expectedPacket)) && toWait > 0) {
try {
sendCondition.await(toWait, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
throw new ActiveMQInterruptedException(e);
}
if (response != null && response.getType() != PacketImpl.EXCEPTION && response.getType() != expectedPacket) {
if (response != null && response.getType() != PacketImpl.EXCEPTION && response.getType() != PacketImpl.EXCEPTION_RESPONSE && response.getType() != expectedPacket) {
ActiveMQClientLogger.LOGGER.packetOutOfOrder(response, new Exception("trace"));
}
@ -424,6 +446,16 @@ public final class ChannelImpl implements Channel {
throw e;
}
if (response.getType() == PacketImpl.EXCEPTION_RESPONSE) {
final ExceptionResponseMessage mem = (ExceptionResponseMessage) response;
ActiveMQException e = mem.getException();
e.fillInStackTrace();
throw e;
}
} finally {
lock.unlock();
}
@ -477,6 +509,18 @@ public final class ChannelImpl implements Channel {
commandConfirmationHandler = handler;
}
@Override
public void setResponseHandler(final ResponseHandler responseHandler) {
if (confWindowSize < 0) {
final String msg = "You can't set responseHandler on a connection with confirmation-window-size < 0." + " Look at the documentation for more information.";
if (logger.isTraceEnabled()) {
logger.trace("RemotingConnectionID=" + (connection == null ? "NULL" : connection.getID()) + " " + msg);
}
throw new IllegalStateException(msg);
}
responseAsyncCache.setResponseHandler(responseHandler);
}
@Override
public void setHandler(final ChannelHandler handler) {
if (logger.isTraceEnabled()) {
@ -595,6 +639,12 @@ public final class ChannelImpl implements Channel {
}
}
public void handleResponse(Packet packet) {
if (responseAsyncCache != null && packet.isResponseAsync()) {
responseAsyncCache.handleResponse(packet);
}
}
@Override
public void confirm(final Packet packet) {
if (resendCache != null && packet.isRequiresConfirmations()) {
@ -647,6 +697,7 @@ public final class ChannelImpl implements Channel {
if (packet.isResponse()) {
confirm(packet);
handleResponse(packet);
lock.lock();
try {
@ -676,6 +727,15 @@ public final class ChannelImpl implements Channel {
}
}
private void addResponseExpectedPacket(Packet packet) {
if (packet.isResponseAsync()) {
responseAsyncCache.add(packet);
if (logger.isTraceEnabled()) {
logger.trace("RemotingConnectionID=" + (connection == null ? "NULL" : connection.getID()) + " ChannelImpl::addResendPacket adding packet " + packet + " stored correlationID=" + packet.getCorrelationID());
}
}
}
private void clearUpTo(final int lastReceivedCommandID) {
final int numberToClear = 1 + lastReceivedCommandID - firstStoredCommandID;

26
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketDecoder.java
View File

@ -39,6 +39,7 @@ import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.Disconnect
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.DisconnectConsumerWithKillMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.DisconnectMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.DisconnectMessage_V2;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ExceptionResponseMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.NullResponseMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.PacketsConfirmedMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.Ping;
@ -71,6 +72,7 @@ import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionQue
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionReceiveContinuationMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionRequestProducerCreditsMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionSendContinuationMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionSendContinuationMessage_V2;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionUniqueAddMetaDataMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionXAAfterFailedMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionXACommitMessage;
@ -88,6 +90,7 @@ import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionXAS
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionXAStartMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SubscribeClusterTopologyUpdatesMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SubscribeClusterTopologyUpdatesMessageV2;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SuccessResponseMessage;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.CHECK_FOR_FAILOVER;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.CLUSTER_TOPOLOGY;
@ -105,6 +108,7 @@ import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.DIS
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.DISCONNECT_CONSUMER;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.DISCONNECT_V2;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.EXCEPTION;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.EXCEPTION_RESPONSE;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.NULL_RESPONSE;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.PACKETS_CONFIRMED;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.PING;
@ -136,6 +140,7 @@ import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SES
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_RECEIVE_CONTINUATION;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_ROLLBACK;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_SEND_CONTINUATION;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_SEND_CONTINUATION_V2;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_START;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_STOP;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_UNIQUE_ADD_METADATA;
@ -158,6 +163,7 @@ import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SES
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_XA_SUSPEND;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SUBSCRIBE_TOPOLOGY;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SUBSCRIBE_TOPOLOGY_V2;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SUCCESS_RESPONSE;
public abstract class PacketDecoder implements Serializable {
@ -187,10 +193,22 @@ public abstract class PacketDecoder implements Serializable {
packet = new ActiveMQExceptionMessage();
break;
}
case EXCEPTION_RESPONSE: {
packet = new ExceptionResponseMessage();
break;
}
case PACKETS_CONFIRMED: {
packet = new PacketsConfirmedMessage();
break;
}
case NULL_RESPONSE: {
packet = new NullResponseMessage();
break;
}
case SUCCESS_RESPONSE: {
packet = new SuccessResponseMessage();
break;
}
case CREATESESSION: {
packet = new CreateSessionMessage();
break;
@ -383,10 +401,6 @@ public abstract class PacketDecoder implements Serializable {
packet = new SessionIndividualAcknowledgeMessage();
break;
}
case NULL_RESPONSE: {
packet = new NullResponseMessage();
break;
}
case SESS_RECEIVE_CONTINUATION: {
packet = new SessionReceiveContinuationMessage();
break;
@ -395,6 +409,10 @@ public abstract class PacketDecoder implements Serializable {
packet = new SessionSendContinuationMessage();
break;
}
case SESS_SEND_CONTINUATION_V2: {
packet = new SessionSendContinuationMessage_V2();
break;
}
case SESS_PRODUCER_REQUEST_CREDITS: {
packet = new SessionRequestProducerCreditsMessage();
break;

33
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketImpl.java
View File

@ -29,8 +29,13 @@ public class PacketImpl implements Packet {
// Constants -------------------------------------------------------------------------
// 2.0.0
public static final int ADDRESSING_CHANGE_VERSION = 129;
// 2.5.0
public static final int ASYNC_RESPONSE_CHANGE_VERSION = 130;
public static final SimpleString OLD_QUEUE_PREFIX = new SimpleString("jms.queue.");
public static final SimpleString OLD_TOPIC_PREFIX = new SimpleString("jms.topic.");
@ -63,6 +68,10 @@ public class PacketImpl implements Packet {
public static final byte PACKETS_CONFIRMED = 22;
public static final byte SUCCESS_RESPONSE = 23;
public static final byte EXCEPTION_RESPONSE = 24;
// Server
public static final byte CREATESESSION = 30;
@ -267,6 +276,11 @@ public class PacketImpl implements Packet {
public static final byte SESS_BINDINGQUERY_RESP_V4 = -15;
public static final byte SESS_SEND_V2 = -16;
public static final byte SESS_SEND_CONTINUATION_V2 = -17;
// Static --------------------------------------------------------
public PacketImpl(final byte type) {
@ -434,5 +448,24 @@ public class PacketImpl implements Packet {
return DataConstants.SIZE_BOOLEAN + (str != null ? stringEncodeSize(str) : 0);
}
@Override
public boolean isRequiresResponse() {
return false;
}
@Override
public boolean isResponseAsync() {
return false;
}
@Override
public long getCorrelationID() {
return -1;
}
@Override
public void setCorrelationID(long correlationID) {
}
}

61
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ResponseCache.java Normal file
View File

@ -0,0 +1,61 @@
/*
* 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.core.protocol.core.impl;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.activemq.artemis.core.client.ActiveMQClientMessageBundle;
import org.apache.activemq.artemis.core.protocol.core.Packet;
import org.apache.activemq.artemis.core.protocol.core.ResponseHandler;
import org.apache.activemq.artemis.utils.collections.ConcurrentLongHashMap;
public class ResponseCache {
private final AtomicLong sequence = new AtomicLong(0);
private final ConcurrentLongHashMap<Packet> store;
private final int size;
private ResponseHandler responseHandler;
public ResponseCache(int size) {
this.store = new ConcurrentLongHashMap<>(size);
this.size = size;
}
public long nextCorrelationID() {
return sequence.incrementAndGet();
}
public void add(Packet packet) {
if (store.size() + 1 > size) {
throw ActiveMQClientMessageBundle.BUNDLE.cannotSendPacketWhilstResponseCacheFull();
}
this.store.put(packet.getCorrelationID(), packet);
}
public void handleResponse(Packet response) {
long correlationID = response.getCorrelationID();
Packet packet = store.remove(correlationID);
if (packet != null) {
responseHandler.responseHandler(packet, response);
}
}
public void setResponseHandler(ResponseHandler responseHandler) {
this.responseHandler = responseHandler;
}
}

1
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateAddressMessage.java
View File

@ -66,6 +66,7 @@ public class CreateAddressMessage extends PacketImpl {
return address;
}
@Override
public boolean isRequiresResponse() {
return requiresResponse;
}

1
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateQueueMessage.java
View File

@ -100,6 +100,7 @@ public class CreateQueueMessage extends PacketImpl {
return temporary;
}
@Override
public boolean isRequiresResponse() {
return requiresResponse;
}

1
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateSharedQueueMessage.java
View File

@ -80,6 +80,7 @@ public class CreateSharedQueueMessage extends PacketImpl {
return filterString;
}
@Override
public boolean isRequiresResponse() {
return requiresResponse;
}

120
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/ExceptionResponseMessage.java Normal file
View File

@ -0,0 +1,120 @@
/*
* 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.core.protocol.core.impl.wireformat;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.ActiveMQExceptionType;
import org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl;
public class ExceptionResponseMessage extends PacketImpl {
private ActiveMQException exception;
private long correlationID;
// Static --------------------------------------------------------
// Constructors --------------------------------------------------
public ExceptionResponseMessage(final long correlationID, final ActiveMQException exception) {
super(EXCEPTION_RESPONSE);
this.correlationID = correlationID;
this.exception = exception;
}
public ExceptionResponseMessage() {
super(EXCEPTION_RESPONSE);
}
// Public --------------------------------------------------------
@Override
public boolean isResponse() {
return true;
}
public ActiveMQException getException() {
return exception;
}
@Override
public void encodeRest(final ActiveMQBuffer buffer) {
buffer.writeLong(correlationID);
buffer.writeInt(exception.getType().getCode());
buffer.writeNullableString(exception.getMessage());
}
@Override
public void decodeRest(final ActiveMQBuffer buffer) {
correlationID = buffer.readLong();
int code = buffer.readInt();
String msg = buffer.readNullableString();
exception = ActiveMQExceptionType.createException(code, msg);
}
@Override
public final boolean isResponseAsync() {
return true;
}
@Override
public long getCorrelationID() {
return this.correlationID;
}
@Override
public String toString() {
return getParentString() + ", exception= " + exception + "]";
}
@Override
public int hashCode() {
final int prime = 31;
int result = super.hashCode();
result = prime * result + ((exception == null) ? 0 : exception.hashCode());
return result;
}
@Override
public boolean equals(Object obj) {
if (this == obj) {
return true;
}
if (!super.equals(obj)) {
return false;
}
if (!(obj instanceof ExceptionResponseMessage)) {
return false;
}
ExceptionResponseMessage other = (ExceptionResponseMessage) obj;
if (correlationID != other.correlationID) {
return false;
}
if (exception == null) {
if (other.exception != null) {
return false;
}
} else if (!exception.equals(other.exception)) {
return false;
}
return true;
}
}

1
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionAcknowledgeMessage.java
View File

@ -51,6 +51,7 @@ public class SessionAcknowledgeMessage extends PacketImpl {
return messageID;
}
@Override
public boolean isRequiresResponse() {
return requiresResponse;
}

1
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionCreateConsumerMessage.java
View File

@ -71,6 +71,7 @@ public class SessionCreateConsumerMessage extends QueueAbstractPacket {
return browseOnly;
}
@Override
public boolean isRequiresResponse() {
return requiresResponse;
}

1
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionIndividualAcknowledgeMessage.java
View File

@ -60,6 +60,7 @@ public class SessionIndividualAcknowledgeMessage extends PacketImpl {
return messageID;
}
@Override
public boolean isRequiresResponse() {
return requiresResponse;
}

31
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendContinuationMessage.java
View File

@ -26,10 +26,10 @@ import org.apache.activemq.artemis.utils.DataConstants;
*/
public class SessionSendContinuationMessage extends SessionContinuationMessage {
private boolean requiresResponse;
protected boolean requiresResponse;
// Used on confirmation handling
private Message message;
protected Message message;
/**
* In case, we are using a different handler than the one set on the {@link org.apache.activemq.artemis.api.core.client.ClientSession}
* <br>
@ -43,7 +43,7 @@ public class SessionSendContinuationMessage extends SessionContinuationMessage {
/**
* to be sent on the last package
*/
private long messageBodySize = -1;
protected long messageBodySize = -1;
// Static --------------------------------------------------------
@ -54,6 +54,11 @@ public class SessionSendContinuationMessage extends SessionContinuationMessage {
handler = null;
}
protected SessionSendContinuationMessage(byte type) {
super(type);
handler = null;
}
/**
* @param body
* @param continues
@ -72,11 +77,31 @@ public class SessionSendContinuationMessage extends SessionContinuationMessage {
this.messageBodySize = messageBodySize;
}
/**
* @param body
* @param continues
* @param requiresResponse
*/
protected SessionSendContinuationMessage(final byte type,
final Message message,
final byte[] body,
final boolean continues,
final boolean requiresResponse,
final long messageBodySize,
SendAcknowledgementHandler handler) {
super(type, body, continues);
this.requiresResponse = requiresResponse;
this.message = message;
this.handler = handler;
this.messageBodySize = messageBodySize;
}
// Public --------------------------------------------------------
/**
* @return the requiresResponse
*/
@Override
public boolean isRequiresResponse() {
return requiresResponse;
}

120
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendContinuationMessage_V2.java Normal file
View File

@ -0,0 +1,120 @@
/*
* 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.core.protocol.core.impl.wireformat;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.client.SendAcknowledgementHandler;
import org.apache.activemq.artemis.utils.DataConstants;
/**
* A SessionSendContinuationMessage<br>
*/
public class SessionSendContinuationMessage_V2 extends SessionSendContinuationMessage {
private long correlationID;
// Static --------------------------------------------------------
// Constructors --------------------------------------------------
public SessionSendContinuationMessage_V2() {
super(SESS_SEND_CONTINUATION_V2);
}
/**
* @param body
* @param continues
* @param requiresResponse
*/
public SessionSendContinuationMessage_V2(final Message message,
final byte[] body,
final boolean continues,
final boolean requiresResponse,
final long messageBodySize,
SendAcknowledgementHandler handler) {
super(SESS_SEND_CONTINUATION_V2, message, body, continues, requiresResponse, messageBodySize, handler);
}
// Public --------------------------------------------------------
@Override
public int expectedEncodeSize() {
return super.expectedEncodeSize() + DataConstants.SIZE_LONG;
}
@Override
public void encodeRest(final ActiveMQBuffer buffer) {
super.encodeRest(buffer);
buffer.writeLong(correlationID);
}
@Override
public void decodeRest(final ActiveMQBuffer buffer) {
super.decodeRest(buffer);
correlationID = buffer.readLong();
}
@Override
public long getCorrelationID() {
return this.correlationID;
}
@Override
public void setCorrelationID(long correlationID) {
this.correlationID = correlationID;
}
@Override
public boolean isResponseAsync() {
return true;
}
@Override
public int hashCode() {
final int prime = 31;
int result = super.hashCode();
result = prime * result + (int) (correlationID ^ (correlationID >>> 32));
return result;
}
@Override
public String toString() {
StringBuffer buff = new StringBuffer(getParentString());
buff.append(", continues=" + continues);
buff.append(", message=" + message);
buff.append(", messageBodySize=" + messageBodySize);
buff.append(", requiresResponse=" + requiresResponse);
buff.append(", correlationID=" + correlationID);
buff.append("]");
return buff.toString();
}
@Override
public boolean equals(Object obj) {
if (this == obj)
return true;
if (!super.equals(obj))
return false;
if (!(obj instanceof SessionSendContinuationMessage_V2))
return false;
SessionSendContinuationMessage_V2 other = (SessionSendContinuationMessage_V2) obj;
if (correlationID != other.correlationID)
return false;
return true;
}
}

20
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendMessage.java
View File

@ -21,6 +21,7 @@ import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ICoreMessage;
import org.apache.activemq.artemis.api.core.client.SendAcknowledgementHandler;
import org.apache.activemq.artemis.core.message.impl.CoreMessage;
import org.apache.activemq.artemis.utils.DataConstants;
public class SessionSendMessage extends MessagePacket {
@ -36,6 +37,22 @@ public class SessionSendMessage extends MessagePacket {
*/
private final transient SendAcknowledgementHandler handler;
/** This will be using the CoreMessage because it is meant for the core-protocol */
protected SessionSendMessage(final byte id,
final ICoreMessage message,
final boolean requiresResponse,
final SendAcknowledgementHandler handler) {
super(id, message);
this.handler = handler;
this.requiresResponse = requiresResponse;
}
protected SessionSendMessage(final byte id,
final CoreMessage message) {
super(id, message);
this.handler = null;
}
/** This will be using the CoreMessage because it is meant for the core-protocol */
public SessionSendMessage(final ICoreMessage message,
final boolean requiresResponse,
@ -52,6 +69,7 @@ public class SessionSendMessage extends MessagePacket {
// Public --------------------------------------------------------
@Override
public boolean isRequiresResponse() {
return requiresResponse;
}
@ -62,7 +80,7 @@ public class SessionSendMessage extends MessagePacket {
@Override
public int expectedEncodeSize() {
return message.getEncodeSize() + PACKET_HEADERS_SIZE + 1;
return message.getEncodeSize() + PACKET_HEADERS_SIZE + DataConstants.SIZE_BOOLEAN;
}
@Override

113
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendMessage_V2.java Normal file
View File

@ -0,0 +1,113 @@
/*
* 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.core.protocol.core.impl.wireformat;
import io.netty.buffer.ByteBuf;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ICoreMessage;
import org.apache.activemq.artemis.api.core.client.SendAcknowledgementHandler;
import org.apache.activemq.artemis.core.message.impl.CoreMessage;
import org.apache.activemq.artemis.utils.DataConstants;
public class SessionSendMessage_V2 extends SessionSendMessage {
private long correlationID;
/** This will be using the CoreMessage because it is meant for the core-protocol */
public SessionSendMessage_V2(final ICoreMessage message,
final boolean requiresResponse,
final SendAcknowledgementHandler handler) {
super(SESS_SEND_V2, message, requiresResponse, handler);
}
public SessionSendMessage_V2(final CoreMessage message) {
super(SESS_SEND_V2, message);
}
@Override
public int expectedEncodeSize() {
return super.expectedEncodeSize() + DataConstants.SIZE_LONG;
}
@Override
public void encodeRest(ActiveMQBuffer buffer) {
message.sendBuffer(buffer.byteBuf(), 0);
buffer.writeLong(correlationID);
buffer.writeBoolean(requiresResponse);
}
@Override
public void decodeRest(final ActiveMQBuffer buffer) {
// Buffer comes in after having read standard headers and positioned at Beginning of body part
ByteBuf messageBuffer = copyMessageBuffer(buffer.byteBuf(), 1);
message.receiveBuffer(messageBuffer);
buffer.readerIndex(buffer.capacity() - DataConstants.SIZE_LONG - DataConstants.SIZE_BOOLEAN);
correlationID = buffer.readLong();
requiresResponse = buffer.readBoolean();
}
@Override
public long getCorrelationID() {
return this.correlationID;
}
@Override
public void setCorrelationID(long correlationID) {
this.correlationID = correlationID;
}
@Override
public boolean isResponseAsync() {
return true;
}
@Override
public int hashCode() {
final int prime = 31;
int result = super.hashCode();
result = prime * result + (int) (correlationID ^ (correlationID >>> 32));
return result;
}
@Override
public String toString() {
StringBuffer buff = new StringBuffer(getParentString());
buff.append(", correlationID=" + correlationID);
buff.append(", requiresResponse=" + super.isRequiresResponse());
buff.append("]");
return buff.toString();
}
@Override
public boolean equals(Object obj) {
if (this == obj)
return true;
if (!super.equals(obj))
return false;
if (!(obj instanceof SessionSendMessage_V2))
return false;
SessionSendMessage_V2 other = (SessionSendMessage_V2) obj;
if (correlationID != other.correlationID)
return false;
return true;
}
}

93
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SuccessResponseMessage.java Normal file
View File

@ -0,0 +1,93 @@
/*
* 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.core.protocol.core.impl.wireformat;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl;
public class SuccessResponseMessage extends PacketImpl {
private long correlationID;
public SuccessResponseMessage(final long correlationID) {
super(SUCCESS_RESPONSE);
this.correlationID = correlationID;
}
public SuccessResponseMessage() {
super(SUCCESS_RESPONSE);
}
// Public --------------------------------------------------------
@Override
public long getCorrelationID() {
return correlationID;
}
@Override
public void encodeRest(final ActiveMQBuffer buffer) {
buffer.writeLong(correlationID);
}
@Override
public void decodeRest(final ActiveMQBuffer buffer) {
correlationID = buffer.readLong();
}
@Override
public final boolean isResponse() {
return true;
}
@Override
public final boolean isResponseAsync() {
return true;
}
@Override
public String toString() {
return getParentString() + ", correlationID=" + correlationID + "]";
}
@Override
public int hashCode() {
final int prime = 31;
int result = super.hashCode();
result = prime * result + (int) (correlationID ^ (correlationID >>> 32));
return result;
}
@Override
public boolean equals(Object obj) {
if (this == obj) {
return true;
}
if (!super.equals(obj)) {
return false;
}
if (!(obj instanceof SuccessResponseMessage)) {
return false;
}
SuccessResponseMessage other = (SuccessResponseMessage) obj;
if (correlationID != other.correlationID) {
return false;
}
return true;
}
}

2
artemis-core-client/src/main/resources/activemq-version.properties
View File

@ -20,4 +20,4 @@ activemq.version.minorVersion=${activemq.version.minorVersion}
activemq.version.microVersion=${activemq.version.microVersion}
activemq.version.incrementingVersion=${activemq.version.incrementingVersion}
activemq.version.versionTag=${activemq.version.versionTag}
activemq.version.compatibleVersionList=121,122,123,124,125,126,127,128,129
activemq.version.compatibleVersionList=121,122,123,124,125,126,127,128,129,130

30
artemis-jms-client/src/main/java/org/apache/activemq/artemis/jms/client/ActiveMQMessageProducer.java
View File

@ -578,6 +578,36 @@ public class ActiveMQMessageProducer implements MessageProducer, QueueSender, To
}
}
@Override
public void sendFailed(org.apache.activemq.artemis.api.core.Message clientMessage, Exception exception) {
if (jmsMessage instanceof StreamMessage) {
try {
((StreamMessage) jmsMessage).reset();
} catch (JMSException e) {
// HORNETQ-1209 XXX ignore?
}
}
if (jmsMessage instanceof BytesMessage) {
try {
((BytesMessage) jmsMessage).reset();
} catch (JMSException e) {
// HORNETQ-1209 XXX ignore?
}
}
try {
producer.connection.getThreadAwareContext().setCurrentThread(true);
if (exception instanceof ActiveMQException) {
exception = JMSExceptionHelper.convertFromActiveMQException((ActiveMQException)exception);
} else if (exception instanceof ActiveMQInterruptedException) {
exception = JMSExceptionHelper.convertFromActiveMQException((ActiveMQInterruptedException) exception);
}
completionListener.onException(jmsMessage, exception);
} finally {
producer.connection.getThreadAwareContext().clearCurrentThread(true);
}
}
@Override
public String toString() {
return CompletionListenerWrapper.class.getSimpleName() + "( completionListener=" + completionListener + ")";

10
artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/ServerPacketDecoder.java
View File

@ -53,6 +53,7 @@ import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionReq
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionSendLargeMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionSendMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionSendMessage_1X;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionSendMessage_V2;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.BACKUP_REQUEST;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.BACKUP_REQUEST_RESPONSE;
@ -80,6 +81,7 @@ import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SES
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_PRODUCER_REQUEST_CREDITS;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_SEND;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_SEND_LARGE;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_SEND_V2;
public class ServerPacketDecoder extends ClientPacketDecoder {
@ -99,6 +101,12 @@ public class ServerPacketDecoder extends ClientPacketDecoder {
return sendMessage;
}
private static SessionSendMessage decodeSessionSendMessageV2(final ActiveMQBuffer in, CoreRemotingConnection connection) {
final SessionSendMessage_V2 sendMessage = new SessionSendMessage_V2(new CoreMessage());
sendMessage.decode(in);
return sendMessage;
}
private static SessionAcknowledgeMessage decodeSessionAcknowledgeMessage(final ActiveMQBuffer in, CoreRemotingConnection connection) {
final SessionAcknowledgeMessage acknowledgeMessage = new SessionAcknowledgeMessage();
acknowledgeMessage.decode(in);
@ -124,6 +132,8 @@ public class ServerPacketDecoder extends ClientPacketDecoder {
switch (packetType) {
case SESS_SEND:
return decodeSessionSendMessage(in, connection);
case SESS_SEND_V2:
return decodeSessionSendMessageV2(in, connection);
case SESS_ACKNOWLEDGE:
return decodeSessionAcknowledgeMessage(in, connection);
case SESS_PRODUCER_REQUEST_CREDITS:

94
artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/ServerSessionPacketHandler.java
View File

@ -40,6 +40,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.ExceptionResponseMessage;
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;
@ -63,6 +64,7 @@ import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionQue
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionQueueQueryResponseMessage_V3;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionRequestProducerCreditsMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionSendContinuationMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionSendContinuationMessage_V2;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionSendLargeMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionSendMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionUniqueAddMetaDataMessage;
@ -80,6 +82,7 @@ import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionXAR
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.SuccessResponseMessage;
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;
@ -119,7 +122,9 @@ import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SES
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_ROLLBACK;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_SEND;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_SEND_CONTINUATION;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_SEND_CONTINUATION_V2;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_SEND_LARGE;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_SEND_V2;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_START;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_STOP;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_XA_COMMIT;
@ -264,6 +269,10 @@ public class ServerSessionPacketHandler implements ChannelHandler {
onSessionSend(packet);
break;
}
case SESS_SEND_V2: {
onSessionSend(packet);
break;
}
case SESS_ACKNOWLEDGE: {
onSessionAcknowledge(packet);
break;
@ -311,6 +320,15 @@ public class ServerSessionPacketHandler implements ChannelHandler {
}
break;
}
case SESS_SEND_CONTINUATION_V2: {
SessionSendContinuationMessage_V2 message = (SessionSendContinuationMessage_V2) packet;
requiresResponse = message.isRequiresResponse();
sendContinuations(message.getPacketSize(), message.getMessageBodySize(), message.getBody(), message.isContinues());
if (requiresResponse) {
response = new SuccessResponseMessage();
}
break;
}
case SESS_CREATECONSUMER: {
SessionCreateConsumerMessage request = (SessionCreateConsumerMessage) packet;
requiresResponse = request.isRequiresResponse();
@ -605,15 +623,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);
} 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 {
@ -634,15 +652,15 @@ public class ServerSessionPacketHandler implements ChannelHandler {
response = new NullResponseMessage();
}
} 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);
} 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 {
@ -660,18 +678,22 @@ public class ServerSessionPacketHandler implements ChannelHandler {
requiresResponse = message.isRequiresResponse();
this.session.send(EmbedMessageUtil.extractEmbedded(message.getMessage()), this.direct);
if (requiresResponse) {
response = new NullResponseMessage();
if (packet.isResponseAsync()) {
response = new SuccessResponseMessage(packet.getCorrelationID());
} else {
response = new NullResponseMessage();
}
}
} 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);
} 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 {
@ -688,15 +710,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);
} 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 {
@ -713,15 +735,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);
} 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 {
@ -730,14 +752,15 @@ 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);
}
@ -756,24 +779,36 @@ public class ServerSessionPacketHandler implements ChannelHandler {
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 ExceptionResponseMessage(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);
@ -784,7 +819,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) {
@ -793,7 +829,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);
}

2
pom.xml
View File

@ -121,7 +121,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>

2
tests/compatibility-tests/src/test/java/org/apache/activemq/artemis/tests/compatibility/SendAckTest.java
View File

@ -21,6 +21,7 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import static org.apache.activemq.artemis.tests.compatibility.GroovyRun.HORNETQ_247;
import static org.apache.activemq.artemis.tests.compatibility.GroovyRun.ONE_FOUR;
import static org.apache.activemq.artemis.tests.compatibility.GroovyRun.SNAPSHOT;
import static org.apache.activemq.artemis.tests.compatibility.GroovyRun.TWO_FOUR;
@ -61,6 +62,7 @@ public class SendAckTest extends VersionedBaseTest {
// not every combination on two four would make sense.. as there's a compatibility issue between 2.4 and 1.4 when crossing consumers and producers
combinations.add(new Object[]{TWO_FOUR, SNAPSHOT, SNAPSHOT});
combinations.add(new Object[]{SNAPSHOT, TWO_FOUR, TWO_FOUR});
combinations.add(new Object[]{HORNETQ_247, SNAPSHOT, SNAPSHOT});
return combinations;
}

6
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/util/BackupSyncDelay.java
View File

@ -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;
@ -314,6 +315,11 @@ public class BackupSyncDelay implements Interceptor {
}
@Override
public void setResponseHandler(ResponseHandler handler) {
throw new UnsupportedOperationException();
}
@Override
public void flushConfirmations() {
throw new UnsupportedOperationException();

82
tests/jms-tests/src/test/java/org/apache/activemq/artemis/jms/tests/SecurityTest.java
View File

@ -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;
@ -67,9 +78,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();
@ -169,6 +180,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 */
/**

10
tests/jms-tests/src/test/resources/broker.xml
View File

@ -52,6 +52,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>