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 22:03:31 UTC
[1/2] activemq-artemis git commit: Revert "ARTEMIS-1545 Support JMS
2.0 Completion Listener for Exceptions" Revert "ARTEMIS-1545 Adding HornetQ
2.4.7 on the mesh to validate send-acks"
Repository: activemq-artemis
Updated Branches:
refs/heads/master 386f09037 -> 7514e91ed
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/7514e91e/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/ServerSessionPacketHandler.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/ServerSessionPacketHandler.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/ServerSessionPacketHandler.java
index 15f4e42..e93dd94 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/ServerSessionPacketHandler.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/ServerSessionPacketHandler.java
@@ -40,7 +40,6 @@ 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;
@@ -64,7 +63,6 @@ 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;
@@ -82,7 +80,6 @@ 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;
@@ -122,9 +119,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_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;
@@ -269,10 +264,6 @@ public class ServerSessionPacketHandler implements ChannelHandler {
onSessionSend(packet);
break;
}
- case SESS_SEND_V2: {
- onSessionSend(packet);
- break;
- }
case SESS_ACKNOWLEDGE: {
onSessionAcknowledge(packet);
break;
@@ -320,15 +311,6 @@ 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();
@@ -623,15 +605,15 @@ public class ServerSessionPacketHandler implements ChannelHandler {
}
}
} catch (ActiveMQIOErrorException e) {
- response = onActiveMQIOErrorExceptionWhileHandlePacket(packet, e, requiresResponse, response, this.session);
+ response = onActiveMQIOErrorExceptionWhileHandlePacket(e, requiresResponse, response, this.session);
} catch (ActiveMQXAException e) {
response = onActiveMQXAExceptionWhileHandlePacket(e, requiresResponse, response);
} catch (ActiveMQQueueMaxConsumerLimitReached e) {
- response = onActiveMQQueueMaxConsumerLimitReachedWhileHandlePacket(packet, e, requiresResponse, response);
+ response = onActiveMQQueueMaxConsumerLimitReachedWhileHandlePacket(e, requiresResponse, response);
} catch (ActiveMQException e) {
- response = onActiveMQExceptionWhileHandlePacket(packet, e, requiresResponse, response);
+ response = onActiveMQExceptionWhileHandlePacket(e, requiresResponse, response);
} catch (Throwable t) {
- response = onCatchThrowableWhileHandlePacket(packet, t, requiresResponse, response, this.session);
+ response = onCatchThrowableWhileHandlePacket(t, requiresResponse, response, this.session);
}
sendResponse(packet, response, flush, closeChannel);
} finally {
@@ -652,15 +634,15 @@ public class ServerSessionPacketHandler implements ChannelHandler {
response = new NullResponseMessage();
}
} catch (ActiveMQIOErrorException e) {
- response = onActiveMQIOErrorExceptionWhileHandlePacket(packet, e, requiresResponse, response, this.session);
+ response = onActiveMQIOErrorExceptionWhileHandlePacket(e, requiresResponse, response, this.session);
} catch (ActiveMQXAException e) {
response = onActiveMQXAExceptionWhileHandlePacket(e, requiresResponse, response);
} catch (ActiveMQQueueMaxConsumerLimitReached e) {
- response = onActiveMQQueueMaxConsumerLimitReachedWhileHandlePacket(packet, e, requiresResponse, response);
+ response = onActiveMQQueueMaxConsumerLimitReachedWhileHandlePacket(e, requiresResponse, response);
} catch (ActiveMQException e) {
- response = onActiveMQExceptionWhileHandlePacket(packet, e, requiresResponse, response);
+ response = onActiveMQExceptionWhileHandlePacket(e, requiresResponse, response);
} catch (Throwable t) {
- response = onCatchThrowableWhileHandlePacket(packet, t, requiresResponse, response, this.session);
+ response = onCatchThrowableWhileHandlePacket(t, requiresResponse, response, this.session);
}
sendResponse(packet, response, false, false);
} finally {
@@ -678,22 +660,18 @@ public class ServerSessionPacketHandler implements ChannelHandler {
requiresResponse = message.isRequiresResponse();
this.session.send(EmbedMessageUtil.extractEmbedded(message.getMessage()), this.direct);
if (requiresResponse) {
- if (packet.isResponseAsync()) {
- response = new SuccessResponseMessage(packet.getCorrelationID());
- } else {
- response = new NullResponseMessage();
- }
+ response = new NullResponseMessage();
}
} catch (ActiveMQIOErrorException e) {
- response = onActiveMQIOErrorExceptionWhileHandlePacket(packet, e, requiresResponse, response, this.session);
+ response = onActiveMQIOErrorExceptionWhileHandlePacket(e, requiresResponse, response, this.session);
} catch (ActiveMQXAException e) {
response = onActiveMQXAExceptionWhileHandlePacket(e, requiresResponse, response);
} catch (ActiveMQQueueMaxConsumerLimitReached e) {
- response = onActiveMQQueueMaxConsumerLimitReachedWhileHandlePacket(packet, e, requiresResponse, response);
+ response = onActiveMQQueueMaxConsumerLimitReachedWhileHandlePacket(e, requiresResponse, response);
} catch (ActiveMQException e) {
- response = onActiveMQExceptionWhileHandlePacket(packet, e, requiresResponse, response);
+ response = onActiveMQExceptionWhileHandlePacket(e, requiresResponse, response);
} catch (Throwable t) {
- response = onCatchThrowableWhileHandlePacket(packet, t, requiresResponse, response, this.session);
+ response = onCatchThrowableWhileHandlePacket(t, requiresResponse, response, this.session);
}
sendResponse(packet, response, false, false);
} finally {
@@ -710,15 +688,15 @@ public class ServerSessionPacketHandler implements ChannelHandler {
SessionRequestProducerCreditsMessage message = (SessionRequestProducerCreditsMessage) packet;
session.requestProducerCredits(message.getAddress(), message.getCredits());
} catch (ActiveMQIOErrorException e) {
- response = onActiveMQIOErrorExceptionWhileHandlePacket(packet, e, requiresResponse, response, this.session);
+ response = onActiveMQIOErrorExceptionWhileHandlePacket(e, requiresResponse, response, this.session);
} catch (ActiveMQXAException e) {
response = onActiveMQXAExceptionWhileHandlePacket(e, requiresResponse, response);
} catch (ActiveMQQueueMaxConsumerLimitReached e) {
- response = onActiveMQQueueMaxConsumerLimitReachedWhileHandlePacket(packet, e, requiresResponse, response);
+ response = onActiveMQQueueMaxConsumerLimitReachedWhileHandlePacket(e, requiresResponse, response);
} catch (ActiveMQException e) {
- response = onActiveMQExceptionWhileHandlePacket(packet, e, requiresResponse, response);
+ response = onActiveMQExceptionWhileHandlePacket(e, requiresResponse, response);
} catch (Throwable t) {
- response = onCatchThrowableWhileHandlePacket(packet, t, requiresResponse, response, this.session);
+ response = onCatchThrowableWhileHandlePacket(t, requiresResponse, response, this.session);
}
sendResponse(packet, response, false, false);
} finally {
@@ -735,15 +713,15 @@ public class ServerSessionPacketHandler implements ChannelHandler {
SessionConsumerFlowCreditMessage message = (SessionConsumerFlowCreditMessage) packet;
session.receiveConsumerCredits(message.getConsumerID(), message.getCredits());
} catch (ActiveMQIOErrorException e) {
- response = onActiveMQIOErrorExceptionWhileHandlePacket(packet, e, requiresResponse, response, this.session);
+ response = onActiveMQIOErrorExceptionWhileHandlePacket(e, requiresResponse, response, this.session);
} catch (ActiveMQXAException e) {
response = onActiveMQXAExceptionWhileHandlePacket(e, requiresResponse, response);
} catch (ActiveMQQueueMaxConsumerLimitReached e) {
- response = onActiveMQQueueMaxConsumerLimitReachedWhileHandlePacket(packet, e, requiresResponse, response);
+ response = onActiveMQQueueMaxConsumerLimitReachedWhileHandlePacket(e, requiresResponse, response);
} catch (ActiveMQException e) {
- response = onActiveMQExceptionWhileHandlePacket(packet, e, requiresResponse, response);
+ response = onActiveMQExceptionWhileHandlePacket(e, requiresResponse, response);
} catch (Throwable t) {
- response = onCatchThrowableWhileHandlePacket(packet, t, requiresResponse, response, this.session);
+ response = onCatchThrowableWhileHandlePacket(t, requiresResponse, response, this.session);
}
sendResponse(packet, response, false, false);
} finally {
@@ -752,15 +730,14 @@ public class ServerSessionPacketHandler implements ChannelHandler {
}
- private static Packet onActiveMQIOErrorExceptionWhileHandlePacket(Packet packet,
- ActiveMQIOErrorException e,
+ private static Packet onActiveMQIOErrorExceptionWhileHandlePacket(ActiveMQIOErrorException e,
boolean requiresResponse,
Packet response,
ServerSession session) {
session.markTXFailed(e);
if (requiresResponse) {
logger.debug("Sending exception to client", e);
- response = convertToExceptionPacket(packet, e);
+ response = new ActiveMQExceptionMessage(e);
} else {
ActiveMQServerLogger.LOGGER.caughtException(e);
}
@@ -779,36 +756,24 @@ public class ServerSessionPacketHandler implements ChannelHandler {
return response;
}
- private static Packet onActiveMQQueueMaxConsumerLimitReachedWhileHandlePacket(Packet packet,
- ActiveMQQueueMaxConsumerLimitReached e,
+ private static Packet onActiveMQQueueMaxConsumerLimitReachedWhileHandlePacket(ActiveMQQueueMaxConsumerLimitReached e,
boolean requiresResponse,
Packet response) {
if (requiresResponse) {
logger.debug("Sending exception to client", e);
- response = convertToExceptionPacket(packet, e);
+ response = new ActiveMQExceptionMessage(e);
} else {
ActiveMQServerLogger.LOGGER.caughtException(e);
}
return response;
}
- 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,
+ private static Packet onActiveMQExceptionWhileHandlePacket(ActiveMQException e,
boolean requiresResponse,
Packet response) {
if (requiresResponse) {
logger.debug("Sending exception to client", e);
- response = convertToExceptionPacket(packet, e);
+ response = new ActiveMQExceptionMessage(e);
} else {
if (e.getType() == ActiveMQExceptionType.QUEUE_EXISTS) {
logger.debug("Caught exception", e);
@@ -819,8 +784,7 @@ public class ServerSessionPacketHandler implements ChannelHandler {
return response;
}
- private static Packet onCatchThrowableWhileHandlePacket(Packet packet,
- Throwable t,
+ private static Packet onCatchThrowableWhileHandlePacket(Throwable t,
boolean requiresResponse,
Packet response,
ServerSession session) {
@@ -829,7 +793,7 @@ public class ServerSessionPacketHandler implements ChannelHandler {
ActiveMQServerLogger.LOGGER.sendingUnexpectedExceptionToClient(t);
ActiveMQException activeMQInternalErrorException = new ActiveMQInternalErrorException();
activeMQInternalErrorException.initCause(t);
- response = convertToExceptionPacket(packet, activeMQInternalErrorException);
+ response = new ActiveMQExceptionMessage(activeMQInternalErrorException);
} else {
ActiveMQServerLogger.LOGGER.caughtException(t);
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/7514e91e/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 0ef8b59..1baad38 100644
--- a/pom.xml
+++ b/pom.xml
@@ -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>130,129,128,127,126,125,124,123,122</activemq.version.incrementingVersion>
+ <activemq.version.incrementingVersion>129,128,127,126,125,124,123,122</activemq.version.incrementingVersion>
<activemq.version.versionTag>${project.version}</activemq.version.versionTag>
<ActiveMQ-Version>${project.version}(${activemq.version.incrementingVersion})</ActiveMQ-Version>
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/7514e91e/tests/compatibility-tests/src/test/java/org/apache/activemq/artemis/tests/compatibility/SendAckTest.java
----------------------------------------------------------------------
diff --git a/tests/compatibility-tests/src/test/java/org/apache/activemq/artemis/tests/compatibility/SendAckTest.java b/tests/compatibility-tests/src/test/java/org/apache/activemq/artemis/tests/compatibility/SendAckTest.java
index 6b68304..2f21704 100644
--- a/tests/compatibility-tests/src/test/java/org/apache/activemq/artemis/tests/compatibility/SendAckTest.java
+++ b/tests/compatibility-tests/src/test/java/org/apache/activemq/artemis/tests/compatibility/SendAckTest.java
@@ -21,7 +21,6 @@ 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;
@@ -62,7 +61,6 @@ 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;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/7514e91e/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/util/BackupSyncDelay.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/util/BackupSyncDelay.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/util/BackupSyncDelay.java
index c7ed869..e4afb5b 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/util/BackupSyncDelay.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/util/BackupSyncDelay.java
@@ -25,7 +25,6 @@ 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;
@@ -316,11 +315,6 @@ public class BackupSyncDelay implements Interceptor {
}
@Override
- public void setResponseHandler(ResponseHandler handler) {
- throw new UnsupportedOperationException();
- }
-
- @Override
public void flushConfirmations() {
throw new UnsupportedOperationException();
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/7514e91e/tests/jms-tests/src/test/java/org/apache/activemq/artemis/jms/tests/SecurityTest.java
----------------------------------------------------------------------
diff --git a/tests/jms-tests/src/test/java/org/apache/activemq/artemis/jms/tests/SecurityTest.java b/tests/jms-tests/src/test/java/org/apache/activemq/artemis/jms/tests/SecurityTest.java
index 7e121f3..d7137ae 100644
--- a/tests/jms-tests/src/test/java/org/apache/activemq/artemis/jms/tests/SecurityTest.java
+++ b/tests/jms-tests/src/test/java/org/apache/activemq/artemis/jms/tests/SecurityTest.java
@@ -16,23 +16,12 @@
*/
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;
@@ -78,9 +67,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();
@@ -180,71 +169,6 @@ public class SecurityTest extends JMSTestCase {
}
}
- /**
- * Login with valid user and password
- * But try send to address not authorised - Persistent
- * Should not allow and should throw exception
- */
- @Test
- public void testLoginValidUserAndPasswordButNotAuthorisedToSend() throws Exception {
- ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
- Connection connection = connectionFactory.createConnection("guest", "guest");
- Session session = connection.createSession();
- Destination destination = session.createQueue("guest.cannot.send");
- MessageProducer messageProducer = session.createProducer(destination);
- try {
- messageProducer.send(session.createTextMessage("hello"));
- fail("JMSSecurityException expected as guest is not allowed to send");
- } catch (JMSSecurityException activeMQSecurityException) {
- //pass
- }
- connection.close();
- }
-
- /**
- * Login with valid user and password
- * But try send to address not authorised - Non Persistent.
- * Should have same behaviour as Persistent with exception on send.
- */
- @Test
- public void testLoginValidUserAndPasswordButNotAuthorisedToSendNonPersistent() throws Exception {
- ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
- connectionFactory.setConfirmationWindowSize(100);
- connectionFactory.setBlockOnDurableSend(false);
- connectionFactory.setBlockOnNonDurableSend(false);
- Connection connection = connectionFactory.createConnection("guest", "guest");
- Session session = connection.createSession();
- Destination destination = session.createQueue("guest.cannot.send");
- MessageProducer messageProducer = session.createProducer(destination);
- messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
- try {
- AtomicReference<Exception> e = new AtomicReference<>();
- // messageProducer.send(session.createTextMessage("hello"));
-
- CountDownLatch countDownLatch = new CountDownLatch(1);
- messageProducer.send(session.createTextMessage("hello"), new CompletionListener() {
- @Override
- public void onCompletion(Message message) {
- countDownLatch.countDown();
- }
-
- @Override
- public void onException(Message message, Exception exception) {
- e.set(exception);
- countDownLatch.countDown();
- }
- });
- countDownLatch.await(10, TimeUnit.SECONDS);
- if (e.get() != null) {
- throw e.get();
- }
- fail("JMSSecurityException expected as guest is not allowed to send");
- } catch (JMSSecurityException activeMQSecurityException) {
- activeMQSecurityException.printStackTrace();
- }
- connection.close();
- }
-
/* Now some client id tests */
/**
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/7514e91e/tests/jms-tests/src/test/resources/broker.xml
----------------------------------------------------------------------
diff --git a/tests/jms-tests/src/test/resources/broker.xml b/tests/jms-tests/src/test/resources/broker.xml
index 2ee1d86..28550ae 100644
--- a/tests/jms-tests/src/test/resources/broker.xml
+++ b/tests/jms-tests/src/test/resources/broker.xml
@@ -52,16 +52,6 @@
<permission type="browse" roles="guest,def"/>
<permission type="send" roles="guest,def"/>
</security-setting>
-
- <security-setting match="guest.cannot.send">
- <permission type="createDurableQueue" roles="guest,def"/>
- <permission type="deleteDurableQueue" roles="guest,def"/>
- <permission type="createNonDurableQueue" roles="guest,def"/>
- <permission type="deleteNonDurableQueue" roles="guest,def"/>
- <permission type="consume" roles="guest,def"/>
- <permission type="browse" roles="guest,def"/>
- <permission type="send" roles="def"/>
- </security-setting>
</security-settings>
</core>
</configuration>
\ No newline at end of file
[2/2] activemq-artemis git commit: Revert "ARTEMIS-1545 Support JMS
2.0 Completion Listener for Exceptions" Revert "ARTEMIS-1545 Adding HornetQ
2.4.7 on the mesh to validate send-acks"
Posted by cl...@apache.org.
Revert "ARTEMIS-1545 Support JMS 2.0 Completion Listener for Exceptions"
Revert "ARTEMIS-1545 Adding HornetQ 2.4.7 on the mesh to validate send-acks"
I'm reverting this as the testsuite is broken..
We will send it back once worked out.
This reverts commit 8f5b7a1e73164f7b7867872a77d74f13176a5af5.
This reverts commit 9b982b3e303067cec8cac277a19a3ab3c63a1a9f.
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/7514e91e
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/7514e91e
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/7514e91e
Branch: refs/heads/master
Commit: 7514e91ed1af3b0586903426c8532728ef3e0912
Parents: 386f090
Author: Clebert Suconic <cl...@apache.org>
Authored: Tue Dec 19 17:01:56 2017 -0500
Committer: Clebert Suconic <cl...@apache.org>
Committed: Tue Dec 19 17:03:00 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 | 33 -----
.../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 +-
.../tests/compatibility/SendAckTest.java | 2 -
.../cluster/util/BackupSyncDelay.java | 6 -
.../artemis/jms/tests/SecurityTest.java | 82 +------------
tests/jms-tests/src/test/resources/broker.xml | 10 --
32 files changed, 64 insertions(+), 1006 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/7514e91e/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 0f47536..c164f6c 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,11 +41,4 @@ 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/7514e91e/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 ac8be9b..64107b9 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,7 +234,4 @@ 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/7514e91e/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 56f8259..127a69a 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,9 +211,6 @@ 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/7514e91e/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 74d9847..b6a5d93 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,11 +36,6 @@ 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/7514e91e/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 b658090..1f40314 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,14 +41,6 @@ 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/7514e91e/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
deleted file mode 100644
index 21e9879..0000000
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/ResponseHandler.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.artemis.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/7514e91e/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 88c7d03..a9c34f7 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,7 +54,6 @@ 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;
@@ -63,7 +62,6 @@ 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;
@@ -91,11 +89,9 @@ 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;
@@ -168,11 +164,7 @@ public class ActiveMQSessionContext extends SessionContext {
sessionChannel.setHandler(handler);
if (confirmationWindow >= 0) {
- if (sessionChannel.getConnection().isVersionBeforeAsyncResponseChange()) {
- sessionChannel.setCommandConfirmationHandler(commandConfirmationHandler);
- } else {
- sessionChannel.setResponseHandler(responseHandler);
- }
+ sessionChannel.setCommandConfirmationHandler(confirmationHandler);
}
}
@@ -189,50 +181,28 @@ public class ActiveMQSessionContext extends SessionContext {
this.killed = true;
}
- private final CommandConfirmationHandler commandConfirmationHandler = new CommandConfirmationHandler() {
- @Override
- public void commandConfirmed(Packet packet) {
- responseHandler.responseHandler(packet, null);
- }
- };
-
- private final ResponseHandler responseHandler = new ResponseHandler() {
+ private final CommandConfirmationHandler confirmationHandler = new CommandConfirmationHandler() {
@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) {
+ public void commandConfirmed(final Packet packet) {
+ if (packet.getType() == PacketImpl.SESS_SEND) {
SessionSendMessage ssm = (SessionSendMessage) packet;
- callSendAck(ssm.getHandler(), ssm.getMessage(), activeMQException);
- } else if (packet.getType() == PacketImpl.SESS_SEND_CONTINUATION || packet.getType() == PacketImpl.SESS_SEND_CONTINUATION_V2) {
+ callSendAck(ssm.getHandler(), ssm.getMessage());
+ } else if (packet.getType() == PacketImpl.SESS_SEND_CONTINUATION) {
SessionSendContinuationMessage scm = (SessionSendContinuationMessage) packet;
if (!scm.isContinues()) {
- callSendAck(scm.getHandler(), scm.getMessage(), activeMQException);
+ callSendAck(scm.getHandler(), scm.getMessage());
}
}
}
- private void callSendAck(SendAcknowledgementHandler handler, final Message message, final Exception exception) {
+ private void callSendAck(SendAcknowledgementHandler handler, final Message message) {
if (handler != null) {
- if (exception == null) {
- handler.sendAcknowledged(message);
- } else {
- handler.sendFailed(message, exception);
- }
+ handler.sendAcknowledged(message);
} else if (sendAckHandler != null) {
- if (exception == null) {
- sendAckHandler.sendAcknowledged(message);
- } else {
- handler.sendFailed(message, exception);
- }
+ sendAckHandler.sendAcknowledged(message);
}
}
+
};
// Failover utility methods
@@ -269,11 +239,7 @@ public class ActiveMQSessionContext extends SessionContext {
@Override
public void setSendAcknowledgementHandler(final SendAcknowledgementHandler handler) {
- if (sessionChannel.getConnection().isVersionBeforeAsyncResponseChange()) {
- sessionChannel.setCommandConfirmationHandler(commandConfirmationHandler);
- } else {
- sessionChannel.setResponseHandler(responseHandler);
- }
+ sessionChannel.setCommandConfirmationHandler(confirmationHandler);
this.sendAckHandler = handler;
}
@@ -488,21 +454,15 @@ public class ActiveMQSessionContext extends SessionContext {
boolean sendBlocking,
SendAcknowledgementHandler handler,
SimpleString defaultAddress) throws ActiveMQException {
- final SessionSendMessage packet;
- final byte expectedPacket;
+ SessionSendMessage packet;
if (sessionChannel.getConnection().isVersionBeforeAddressChange()) {
packet = new SessionSendMessage_1X(msgI, sendBlocking, handler);
- 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;
+ packet = new SessionSendMessage(msgI, sendBlocking, handler);
}
+
if (sendBlocking) {
- sessionChannel.sendBlocking(packet, expectedPacket);
+ sessionChannel.sendBlocking(packet, PacketImpl.NULL_RESPONSE);
} else {
sessionChannel.sendBatched(packet);
}
@@ -880,20 +840,15 @@ public class ActiveMQSessionContext extends SessionContext {
}
}
- private int sendSessionSendContinuationMessage(Channel channel,
+ private static int sendSessionSendContinuationMessage(Channel channel,
Message msgI,
long messageBodySize,
boolean sendBlocking,
boolean lastChunk,
byte[] chunk,
SendAcknowledgementHandler messageHandler) throws ActiveMQException {
- 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 boolean requiresResponse = lastChunk && sendBlocking;
+ final SessionSendContinuationMessage chunkPacket = new SessionSendContinuationMessage(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();
@@ -910,11 +865,7 @@ public class ActiveMQSessionContext extends SessionContext {
}
if (requiresResponse) {
// When sending it blocking, only the last chunk will be blocking.
- if (sendBlocking) {
- channel.sendBlocking(chunkPacket, PacketImpl.SUCCESS_RESPONSE);
- } else {
- channel.send(chunkPacket);
- }
+ channel.sendBlocking(chunkPacket, PacketImpl.NULL_RESPONSE);
} else {
channel.send(chunkPacket);
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/7514e91e/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 6540755..4d73cf8 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,9 +36,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.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;
@@ -98,8 +96,6 @@ 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);
@@ -126,8 +122,6 @@ public final class ChannelImpl implements Channel {
private CommandConfirmationHandler commandConfirmationHandler;
- private ResponseHandler responseHandler;
-
private volatile boolean transferring;
private final List<Interceptor> interceptors;
@@ -144,10 +138,8 @@ public final class ChannelImpl implements Channel {
if (confWindowSize != -1) {
resendCache = new ConcurrentLinkedQueue<>();
- responseAsyncCache = new ResponseCache(confWindowSize);
} else {
resendCache = null;
- responseAsyncCache = null;
}
this.interceptors = interceptors;
@@ -184,12 +176,6 @@ 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;
}
@@ -284,10 +270,6 @@ 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);
}
@@ -309,10 +291,6 @@ public final class ChannelImpl implements Channel {
if (resendCache != null && packet.isRequiresConfirmations()) {
addResendPacket(packet);
}
-
- if (responseAsyncCache != null && packet.isRequiresResponse()) {
- addResponseExpectedPacket(packet);
- }
} finally {
lock.unlock();
}
@@ -406,14 +384,14 @@ public final class ChannelImpl implements Channel {
long start = System.currentTimeMillis();
- while (!closed && (response == null || (response.getType() != PacketImpl.EXCEPTION && response.getType() != PacketImpl.EXCEPTION_RESPONSE && response.getType() != expectedPacket)) && toWait > 0) {
+ while (!closed && (response == null || (response.getType() != PacketImpl.EXCEPTION && 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() != PacketImpl.EXCEPTION_RESPONSE && response.getType() != expectedPacket) {
+ if (response != null && response.getType() != PacketImpl.EXCEPTION && response.getType() != expectedPacket) {
ActiveMQClientLogger.LOGGER.packetOutOfOrder(response, new Exception("trace"));
}
@@ -446,16 +424,6 @@ 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();
}
@@ -510,18 +478,6 @@ 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);
@@ -639,12 +595,6 @@ 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()) {
@@ -697,7 +647,6 @@ public final class ChannelImpl implements Channel {
if (packet.isResponse()) {
confirm(packet);
- handleResponse(packet);
lock.lock();
try {
@@ -727,15 +676,6 @@ 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/7514e91e/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 1c27684..5e46848 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,7 +39,6 @@ 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;
@@ -72,7 +71,6 @@ 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;
@@ -90,7 +88,6 @@ 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;
@@ -108,7 +105,6 @@ 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;
@@ -140,7 +136,6 @@ 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;
@@ -163,7 +158,6 @@ 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 {
@@ -193,22 +187,10 @@ 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;
@@ -401,6 +383,10 @@ 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;
@@ -409,10 +395,6 @@ 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/7514e91e/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 48f6c6c..925d089 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
@@ -29,13 +29,8 @@ 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.");
@@ -68,10 +63,6 @@ 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;
@@ -276,11 +267,6 @@ 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) {
@@ -448,24 +434,5 @@ 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/7514e91e/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
deleted file mode 100644
index 19b3c42..0000000
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ResponseCache.java
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.artemis.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/7514e91e/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 38d8d5e..f592a4c 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,7 +66,6 @@ public class CreateAddressMessage extends PacketImpl {
return address;
}
- @Override
public boolean isRequiresResponse() {
return requiresResponse;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/7514e91e/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 985d5f4..2ebf147 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,7 +100,6 @@ public class CreateQueueMessage extends PacketImpl {
return temporary;
}
- @Override
public boolean isRequiresResponse() {
return requiresResponse;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/7514e91e/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 3c072e0..af25ae9 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,7 +80,6 @@ public class CreateSharedQueueMessage extends PacketImpl {
return filterString;
}
- @Override
public boolean isRequiresResponse() {
return requiresResponse;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/7514e91e/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
deleted file mode 100644
index 69c386e..0000000
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/ExceptionResponseMessage.java
+++ /dev/null
@@ -1,120 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.artemis.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/7514e91e/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 67d9f67..542c34c 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,7 +51,6 @@ public class SessionAcknowledgeMessage extends PacketImpl {
return messageID;
}
- @Override
public boolean isRequiresResponse() {
return requiresResponse;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/7514e91e/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 e07b50c..f09beeb 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,7 +71,6 @@ public class SessionCreateConsumerMessage extends QueueAbstractPacket {
return browseOnly;
}
- @Override
public boolean isRequiresResponse() {
return requiresResponse;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/7514e91e/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 3164c23..7d06081 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,7 +60,6 @@ public class SessionIndividualAcknowledgeMessage extends PacketImpl {
return messageID;
}
- @Override
public boolean isRequiresResponse() {
return requiresResponse;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/7514e91e/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 4105b11..26eedd7 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 {
- protected boolean requiresResponse;
+ private boolean requiresResponse;
// Used on confirmation handling
- protected Message message;
+ private 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
*/
- protected long messageBodySize = -1;
+ private long messageBodySize = -1;
// Static --------------------------------------------------------
@@ -54,11 +54,6 @@ public class SessionSendContinuationMessage extends SessionContinuationMessage {
handler = null;
}
- protected SessionSendContinuationMessage(byte type) {
- super(type);
- handler = null;
- }
-
/**
* @param body
* @param continues
@@ -77,31 +72,11 @@ 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/7514e91e/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
deleted file mode 100644
index 4afcba7..0000000
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendContinuationMessage_V2.java
+++ /dev/null
@@ -1,120 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.artemis.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/7514e91e/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 22a9503..b56ae30 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,7 +21,6 @@ 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 {
@@ -38,22 +37,6 @@ 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) {
@@ -69,7 +52,6 @@ public class SessionSendMessage extends MessagePacket {
// Public --------------------------------------------------------
- @Override
public boolean isRequiresResponse() {
return requiresResponse;
}
@@ -80,7 +62,7 @@ public class SessionSendMessage extends MessagePacket {
@Override
public int expectedEncodeSize() {
- return message.getEncodeSize() + PACKET_HEADERS_SIZE + DataConstants.SIZE_BOOLEAN;
+ return message.getEncodeSize() + PACKET_HEADERS_SIZE + 1;
}
@Override
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/7514e91e/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
deleted file mode 100644
index f87f442..0000000
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendMessage_V2.java
+++ /dev/null
@@ -1,113 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.artemis.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/7514e91e/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
deleted file mode 100644
index 9f045f8..0000000
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SuccessResponseMessage.java
+++ /dev/null
@@ -1,93 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.artemis.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/7514e91e/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 ff65ff9..a39b422 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,130
+activemq.version.compatibleVersionList=121,122,123,124,125,126,127,128,129
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/7514e91e/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 01deb0b..9f86e49 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,36 +579,6 @@ 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/7514e91e/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 e7f8bb0..0584476 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,7 +53,6 @@ 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;
@@ -81,7 +80,6 @@ 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 {
@@ -101,12 +99,6 @@ 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);
@@ -132,8 +124,6 @@ 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: