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 2017/12/19 20:09:06 UTC
[2/4] activemq-artemis git commit: ARTEMIS-1545 Support JMS 2.0
Completion Listener for Exceptions
ARTEMIS-1545 Support JMS 2.0 Completion Listener for Exceptions
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/9b982b3e
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/9b982b3e
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/9b982b3e
Branch: refs/heads/master
Commit: 9b982b3e303067cec8cac277a19a3ab3c63a1a9f
Parents: fb34214
Author: Michael André Pearce <mi...@me.com>
Authored: Thu Dec 14 07:47:30 2017 +0000
Committer: Clebert Suconic <cl...@apache.org>
Committed: Tue Dec 19 14:44:58 2017 -0500
----------------------------------------------------------------------
.../core/client/SendAcknowledgementHandler.java | 7 ++
.../client/ActiveMQClientMessageBundle.java | 3 +
.../artemis/core/protocol/core/Channel.java | 3 +
.../protocol/core/CoreRemotingConnection.java | 5 +
.../artemis/core/protocol/core/Packet.java | 8 ++
.../core/protocol/core/ResponseHandler.java | 30 +++++
.../core/impl/ActiveMQSessionContext.java | 89 ++++++++++----
.../core/protocol/core/impl/ChannelImpl.java | 64 +++++++++-
.../core/protocol/core/impl/PacketDecoder.java | 26 +++-
.../core/protocol/core/impl/PacketImpl.java | 30 +++++
.../core/protocol/core/impl/ResponseCache.java | 61 ++++++++++
.../impl/wireformat/CreateAddressMessage.java | 1 +
.../impl/wireformat/CreateQueueMessage.java | 1 +
.../wireformat/CreateSharedQueueMessage.java | 1 +
.../wireformat/ExceptionResponseMessage.java | 120 +++++++++++++++++++
.../wireformat/SessionAcknowledgeMessage.java | 1 +
.../SessionCreateConsumerMessage.java | 1 +
.../SessionIndividualAcknowledgeMessage.java | 1 +
.../SessionSendContinuationMessage.java | 31 ++++-
.../SessionSendContinuationMessage_V2.java | 120 +++++++++++++++++++
.../impl/wireformat/SessionSendMessage.java | 20 +++-
.../impl/wireformat/SessionSendMessage_V2.java | 113 +++++++++++++++++
.../impl/wireformat/SuccessResponseMessage.java | 93 ++++++++++++++
.../main/resources/activemq-version.properties | 2 +-
.../jms/client/ActiveMQMessageProducer.java | 30 +++++
.../core/protocol/ServerPacketDecoder.java | 10 ++
.../core/ServerSessionPacketHandler.java | 94 ++++++++++-----
pom.xml | 2 +-
.../cluster/util/BackupSyncDelay.java | 6 +
.../artemis/jms/tests/SecurityTest.java | 82 ++++++++++++-
tests/jms-tests/src/test/resources/broker.xml | 10 ++
31 files changed, 1001 insertions(+), 64 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/client/SendAcknowledgementHandler.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/client/SendAcknowledgementHandler.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/client/SendAcknowledgementHandler.java
index c164f6c..0f47536 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/client/SendAcknowledgementHandler.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/client/SendAcknowledgementHandler.java
@@ -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);
+ }
+
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/ActiveMQClientMessageBundle.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/ActiveMQClientMessageBundle.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/ActiveMQClientMessageBundle.java
index 64107b9..ac8be9b 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/ActiveMQClientMessageBundle.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/ActiveMQClientMessageBundle.java
@@ -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();
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/Channel.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/Channel.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/Channel.java
index 127a69a..56f8259 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/Channel.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/Channel.java
@@ -211,6 +211,9 @@ public interface Channel {
*/
void setCommandConfirmationHandler(CommandConfirmationHandler handler);
+ void setResponseHandler(ResponseHandler handler);
+
+
/**
* flushes any confirmations on to the connection.
*/
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/CoreRemotingConnection.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/CoreRemotingConnection.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/CoreRemotingConnection.java
index b6a5d93..74d9847 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/CoreRemotingConnection.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/CoreRemotingConnection.java
@@ -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
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/Packet.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/Packet.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/Packet.java
index 1f40314..b658090 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/Packet.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/Packet.java
@@ -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.
*
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/ResponseHandler.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/ResponseHandler.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/ResponseHandler.java
new file mode 100644
index 0000000..21e9879
--- /dev/null
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/ResponseHandler.java
@@ -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);
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ActiveMQSessionContext.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ActiveMQSessionContext.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ActiveMQSessionContext.java
index a9c34f7..88c7d03 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ActiveMQSessionContext.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ActiveMQSessionContext.java
@@ -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(Packet packet) {
+ responseHandler.responseHandler(packet, null);
+ }
+ };
+
+ private final ResponseHandler responseHandler = new ResponseHandler() {
@Override
- public void commandConfirmed(final Packet packet) {
- if (packet.getType() == PacketImpl.SESS_SEND) {
+ 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);
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ChannelImpl.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ChannelImpl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ChannelImpl.java
index 4d73cf8..6540755 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ChannelImpl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ChannelImpl.java
@@ -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();
}
@@ -478,6 +510,18 @@ public final class ChannelImpl implements Channel {
}
@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()) {
logger.trace("RemotingConnectionID=" + (connection == null ? "NULL" : connection.getID()) + " Setting handler on " + this + " as " + handler);
@@ -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;
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketDecoder.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketDecoder.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketDecoder.java
index 5e46848..1c27684 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketDecoder.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketDecoder.java
@@ -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;
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketImpl.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketImpl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketImpl.java
index 925d089..00e9613 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketImpl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketImpl.java
@@ -30,6 +30,8 @@ public class PacketImpl implements Packet {
public static final int ADDRESSING_CHANGE_VERSION = 129;
+ 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 +65,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 +273,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 +445,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) {
+ }
+
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ResponseCache.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ResponseCache.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ResponseCache.java
new file mode 100644
index 0000000..19b3c42
--- /dev/null
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ResponseCache.java
@@ -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;
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateAddressMessage.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateAddressMessage.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateAddressMessage.java
index f592a4c..38d8d5e 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateAddressMessage.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateAddressMessage.java
@@ -66,6 +66,7 @@ public class CreateAddressMessage extends PacketImpl {
return address;
}
+ @Override
public boolean isRequiresResponse() {
return requiresResponse;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateQueueMessage.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateQueueMessage.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateQueueMessage.java
index 2ebf147..985d5f4 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateQueueMessage.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateQueueMessage.java
@@ -100,6 +100,7 @@ public class CreateQueueMessage extends PacketImpl {
return temporary;
}
+ @Override
public boolean isRequiresResponse() {
return requiresResponse;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateSharedQueueMessage.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateSharedQueueMessage.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateSharedQueueMessage.java
index af25ae9..3c072e0 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateSharedQueueMessage.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/CreateSharedQueueMessage.java
@@ -80,6 +80,7 @@ public class CreateSharedQueueMessage extends PacketImpl {
return filterString;
}
+ @Override
public boolean isRequiresResponse() {
return requiresResponse;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/ExceptionResponseMessage.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/ExceptionResponseMessage.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/ExceptionResponseMessage.java
new file mode 100644
index 0000000..69c386e
--- /dev/null
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/ExceptionResponseMessage.java
@@ -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;
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionAcknowledgeMessage.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionAcknowledgeMessage.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionAcknowledgeMessage.java
index 542c34c..67d9f67 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionAcknowledgeMessage.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionAcknowledgeMessage.java
@@ -51,6 +51,7 @@ public class SessionAcknowledgeMessage extends PacketImpl {
return messageID;
}
+ @Override
public boolean isRequiresResponse() {
return requiresResponse;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionCreateConsumerMessage.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionCreateConsumerMessage.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionCreateConsumerMessage.java
index f09beeb..e07b50c 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionCreateConsumerMessage.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionCreateConsumerMessage.java
@@ -71,6 +71,7 @@ public class SessionCreateConsumerMessage extends QueueAbstractPacket {
return browseOnly;
}
+ @Override
public boolean isRequiresResponse() {
return requiresResponse;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionIndividualAcknowledgeMessage.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionIndividualAcknowledgeMessage.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionIndividualAcknowledgeMessage.java
index 7d06081..3164c23 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionIndividualAcknowledgeMessage.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionIndividualAcknowledgeMessage.java
@@ -60,6 +60,7 @@ public class SessionIndividualAcknowledgeMessage extends PacketImpl {
return messageID;
}
+ @Override
public boolean isRequiresResponse() {
return requiresResponse;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendContinuationMessage.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendContinuationMessage.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendContinuationMessage.java
index 26eedd7..4105b11 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendContinuationMessage.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendContinuationMessage.java
@@ -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;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendContinuationMessage_V2.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendContinuationMessage_V2.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendContinuationMessage_V2.java
new file mode 100644
index 0000000..4afcba7
--- /dev/null
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendContinuationMessage_V2.java
@@ -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;
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendMessage.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendMessage.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendMessage.java
index b56ae30..22a9503 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendMessage.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendMessage.java
@@ -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 {
@@ -37,6 +38,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,
final SendAcknowledgementHandler handler) {
@@ -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
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendMessage_V2.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendMessage_V2.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendMessage_V2.java
new file mode 100644
index 0000000..f87f442
--- /dev/null
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendMessage_V2.java
@@ -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;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SuccessResponseMessage.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SuccessResponseMessage.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SuccessResponseMessage.java
new file mode 100644
index 0000000..9f045f8
--- /dev/null
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SuccessResponseMessage.java
@@ -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;
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-core-client/src/main/resources/activemq-version.properties
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/resources/activemq-version.properties b/artemis-core-client/src/main/resources/activemq-version.properties
index a39b422..ff65ff9 100644
--- a/artemis-core-client/src/main/resources/activemq-version.properties
+++ b/artemis-core-client/src/main/resources/activemq-version.properties
@@ -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
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-jms-client/src/main/java/org/apache/activemq/artemis/jms/client/ActiveMQMessageProducer.java
----------------------------------------------------------------------
diff --git a/artemis-jms-client/src/main/java/org/apache/activemq/artemis/jms/client/ActiveMQMessageProducer.java b/artemis-jms-client/src/main/java/org/apache/activemq/artemis/jms/client/ActiveMQMessageProducer.java
index 9f86e49..01deb0b 100644
--- a/artemis-jms-client/src/main/java/org/apache/activemq/artemis/jms/client/ActiveMQMessageProducer.java
+++ b/artemis-jms-client/src/main/java/org/apache/activemq/artemis/jms/client/ActiveMQMessageProducer.java
@@ -579,6 +579,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 + ")";
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9b982b3e/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/ServerPacketDecoder.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/ServerPacketDecoder.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/ServerPacketDecoder.java
index 0584476..e7f8bb0 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/ServerPacketDecoder.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/ServerPacketDecoder.java
@@ -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: