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

ARTEMIS-4530 clean up SessionCallback interface

This commit is contained in:
Justin Bertram 2023-12-06 16:02:53 -06:00
parent 29a2b2a879
commit 159416bc09
9 changed files with 44 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java
View File

@ -16,6 +16,7 @@
*/
package org.apache.activemq.artemis.protocol.amqp.broker;
import java.lang.invoke.MethodHandles;
import java.util.Map;
import java.util.concurrent.Executor;
@ -30,9 +31,9 @@ import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.client.ActiveMQClient;
import org.apache.activemq.artemis.core.io.IOCallback;
import org.apache.activemq.artemis.core.persistence.CoreMessageObjectPools;
import org.apache.activemq.artemis.core.paging.PagingManager;
import org.apache.activemq.artemis.core.paging.PagingStore;
import org.apache.activemq.artemis.core.persistence.CoreMessageObjectPools;
import org.apache.activemq.artemis.core.persistence.OperationContext;
import org.apache.activemq.artemis.core.persistence.StorageManager;
import org.apache.activemq.artemis.core.security.CheckType;
@ -76,7 +77,6 @@ import org.apache.qpid.proton.engine.EndpointState;
import org.apache.qpid.proton.engine.Receiver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.invoke.MethodHandles;
public class AMQPSessionCallback implements SessionCallback {
@ -649,7 +649,7 @@ public class AMQPSessionCallback implements SessionCallback {
}
@Override
public int sendMessage(MessageReference ref, Message message, ServerConsumer consumer, int deliveryCount) {
public int sendMessage(MessageReference ref, ServerConsumer consumer, int deliveryCount) {
ProtonServerSenderContext plugSender = (ProtonServerSenderContext) consumer.getProtocolContext();
@ -667,7 +667,6 @@ public class AMQPSessionCallback implements SessionCallback {
@Override
public int sendLargeMessage(MessageReference ref,
Message message,
ServerConsumer consumer,
long bodySize,
int deliveryCount) {

13
artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSessionCallback.java
View File

@ -16,7 +16,6 @@
*/
package org.apache.activemq.artemis.core.protocol.mqtt;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.server.MessageReference;
import org.apache.activemq.artemis.core.server.ServerConsumer;
@ -44,14 +43,13 @@ public class MQTTSessionCallback implements SessionCallback {
}
@Override
public int sendMessage(MessageReference reference,
Message message,
public int sendMessage(MessageReference ref,
ServerConsumer consumer,
int deliveryCount) {
try {
session.getMqttPublishManager().sendMessage(message.toCore(), consumer, deliveryCount);
session.getMqttPublishManager().sendMessage(ref.getMessage().toCore(), consumer, deliveryCount);
} catch (Exception e) {
MQTTLogger.LOGGER.unableToSendMessage(reference, e);
MQTTLogger.LOGGER.unableToSendMessage(ref, e);
}
return 1;
}
@ -70,12 +68,11 @@ public class MQTTSessionCallback implements SessionCallback {
}
@Override
public int sendLargeMessage(MessageReference reference,
Message message,
public int sendLargeMessage(MessageReference ref,
ServerConsumer consumer,
long bodySize,
int deliveryCount) {
return sendMessage(reference, message, consumer, deliveryCount);
return sendMessage(ref, consumer, deliveryCount);
}
@Override

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

@ -291,19 +291,17 @@ public class AMQSession implements SessionCallback {
}
@Override
public int sendMessage(MessageReference reference,
org.apache.activemq.artemis.api.core.Message message,
public int sendMessage(MessageReference ref,
ServerConsumer consumer,
int deliveryCount) {
AMQConsumer theConsumer = (AMQConsumer) consumer.getProtocolData();
//clear up possible rolledback ids.
theConsumer.removeRolledback(reference);
return theConsumer.handleDeliver(reference, message.toCore());
theConsumer.removeRolledback(ref);
return theConsumer.handleDeliver(ref, ref.getMessage().toCore());
}
@Override
public int sendLargeMessage(MessageReference reference,
org.apache.activemq.artemis.api.core.Message message,
public int sendLargeMessage(MessageReference ref,
ServerConsumer consumerID,
long bodySize,
int deliveryCount) {

5
artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompSession.java
View File

@ -16,6 +16,7 @@
*/
package org.apache.activemq.artemis.core.protocol.stomp;
import java.lang.invoke.MethodHandles;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@ -53,7 +54,6 @@ import org.apache.activemq.artemis.utils.PendingTask;
import org.apache.activemq.artemis.utils.UUIDGenerator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.invoke.MethodHandles;
import static org.apache.activemq.artemis.core.protocol.stomp.ActiveMQStompProtocolMessageBundle.BUNDLE;
import static org.apache.activemq.artemis.core.protocol.stomp.StompProtocolManagerFactory.STOMP_PROTOCOL_NAME;
@ -153,7 +153,7 @@ public class StompSession implements SessionCallback {
}
@Override
public int sendMessage(MessageReference ref, Message serverMessage, final ServerConsumer consumer, int deliveryCount) {
public int sendMessage(MessageReference ref, final ServerConsumer consumer, int deliveryCount) {
ICoreMessage message = ref.getMessage().toCore();
try {
StompSubscription subscription = subscriptions.get(consumer.getID());
@ -206,7 +206,6 @@ public class StompSession implements SessionCallback {
@Override
public int sendLargeMessage(MessageReference ref,
Message msg,
ServerConsumer consumer,
long bodySize,
int deliveryCount) {

6
artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/ManagementRemotingConnection.java
View File

@ -23,7 +23,6 @@ import java.util.concurrent.Future;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.remoting.CloseListener;
import org.apache.activemq.artemis.core.remoting.FailureListener;
@ -229,13 +228,12 @@ public class ManagementRemotingConnection implements RemotingConnection {
}
@Override
public int sendMessage(MessageReference ref, Message message, ServerConsumer consumerID, int deliveryCount) {
public int sendMessage(MessageReference ref, ServerConsumer consumerID, int deliveryCount) {
return 0;
}
@Override
public int sendLargeMessage(MessageReference reference,
Message message,
public int sendLargeMessage(MessageReference ref,
ServerConsumer consumerID,
long bodySize,
int deliveryCount) {

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

@ -16,7 +16,6 @@
*/
package org.apache.activemq.artemis.core.protocol.core.impl;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.persistence.CoreMessageObjectPools;
import org.apache.activemq.artemis.core.protocol.core.Channel;
@ -88,11 +87,10 @@ public final class CoreSessionCallback implements SessionCallback {
@Override
public int sendLargeMessage(MessageReference ref,
Message message,
ServerConsumer consumer,
long bodySize,
int deliveryCount) {
Packet packet = new SessionReceiveLargeMessage(consumer.getID(), message, bodySize, deliveryCount);
Packet packet = new SessionReceiveLargeMessage(consumer.getID(), ref.getMessage(), bodySize, deliveryCount);
channel.send(packet);
@ -114,13 +112,13 @@ public final class CoreSessionCallback implements SessionCallback {
}
@Override
public int sendMessage(MessageReference ref, Message message, ServerConsumer consumer, int deliveryCount) {
public int sendMessage(MessageReference ref, ServerConsumer consumer, int deliveryCount) {
Packet packet;
if (channel.getConnection().isVersionBeforeAddressChange()) {
packet = new SessionReceiveMessage_1X(consumer.getID(), message.toCore(coreMessageObjectPools), deliveryCount);
packet = new SessionReceiveMessage_1X(consumer.getID(), ref.getMessage().toCore(coreMessageObjectPools), deliveryCount);
} else {
packet = new SessionReceiveMessage(consumer.getID(), message.toCore(coreMessageObjectPools), deliveryCount);
packet = new SessionReceiveMessage(consumer.getID(), ref.getMessage().toCore(coreMessageObjectPools), deliveryCount);
}
int size = 0;

42
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java
View File

@ -487,7 +487,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
// The deliverer will increase the usageUp, so the preAck has to be done after this is created
// otherwise we may have a removed message early on
if (message instanceof CoreLargeServerMessage && this.supportLargeMessage) {
largeMessageDeliverer = new CoreLargeMessageDeliverer((LargeServerMessage) message, ref);
largeMessageDeliverer = new CoreLargeMessageDeliverer(ref);
}
if (preAcknowledge) {
@ -507,8 +507,6 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
@Override
public void proceedDeliver(MessageReference reference) throws Exception {
try {
Message message = reference.getMessage();
if (AuditLogger.isMessageLoggingEnabled()) {
AuditLogger.coreConsumeMessage(session.getRemotingConnection().getSubject(), session.getRemotingConnection().getRemoteAddress(), getQueueName().toString(), reference.toString());
}
@ -516,18 +514,18 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
server.callBrokerMessagePlugins(plugin -> plugin.beforeDeliver(this, reference));
}
if (message instanceof CoreLargeServerMessage && supportLargeMessage) {
if (reference.getMessage() instanceof CoreLargeServerMessage && supportLargeMessage) {
if (largeMessageDeliverer == null) {
// This can't really happen as handle had already crated the deliverer
// This can't really happen as handle had already created the deliverer
// instead of throwing an exception in weird cases there is no problem on just go ahead and create it
// again here
largeMessageDeliverer = new CoreLargeMessageDeliverer((LargeServerMessage) message, reference);
largeMessageDeliverer = new CoreLargeMessageDeliverer(reference);
}
// The deliverer was prepared during handle, as we can't have more than one pending large message
// as it would return busy if there is anything pending
largeMessageDeliverer.deliver();
} else {
deliverStandardMessage(reference, message);
deliverStandardMessage(reference);
}
} finally {
pendingDelivery.countDown();
@ -681,15 +679,15 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
@Override
public void forceDelivery(final long sequence) {
forceDelivery(sequence, () -> {
Message forcedDeliveryMessage = new CoreMessage(storageManager.generateID(), 50);
Message forcedDeliveryMessage = new CoreMessage(storageManager.generateID(), 50)
.putLongProperty(ClientConsumerImpl.FORCED_DELIVERY_MESSAGE, sequence)
.setAddress(messageQueue.getName());
MessageReference reference = MessageReference.Factory.createReference(forcedDeliveryMessage, messageQueue);
reference.setDeliveryCount(0);
forcedDeliveryMessage.putLongProperty(ClientConsumerImpl.FORCED_DELIVERY_MESSAGE, sequence);
forcedDeliveryMessage.setAddress(messageQueue.getName());
applyPrefixForLegacyConsumer(forcedDeliveryMessage);
callback.sendMessage(reference, forcedDeliveryMessage, ServerConsumerImpl.this, 0);
callback.sendMessage(reference, ServerConsumerImpl.this, 0);
});
}
@ -1220,13 +1218,9 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
messageQueue.getExecutor().execute(resumeLargeMessageRunnable);
}
/**
* @param ref
* @param message
*/
private void deliverStandardMessage(final MessageReference ref, Message message) throws ActiveMQException {
applyPrefixForLegacyConsumer(message);
int packetSize = callback.sendMessage(ref, message, ServerConsumerImpl.this, ref.getDeliveryCount());
private void deliverStandardMessage(final MessageReference ref) {
applyPrefixForLegacyConsumer(ref.getMessage());
int packetSize = callback.sendMessage(ref, ServerConsumerImpl.this, ref.getDeliveryCount());
if (availableCredits != null) {
availableCredits.addAndGet(-packetSize);
@ -1301,13 +1295,13 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
private ByteBuffer chunkBytes;
private CoreLargeMessageDeliverer(final LargeServerMessage message, final MessageReference ref) throws Exception {
largeMessage = message;
private CoreLargeMessageDeliverer(final MessageReference ref) {
this.ref = ref;
largeMessage = (LargeServerMessage) ref.getMessage();
largeMessage.toMessage().usageUp();
this.ref = ref;
this.chunkBytes = null;
}
@ -1357,7 +1351,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
sentInitialPacket = true;
int packetSize = callback.sendLargeMessage(ref, currentLargeMessage.toMessage(), ServerConsumerImpl.this, context.getSize(), ref.getDeliveryCount());
int packetSize = callback.sendLargeMessage(ref, ServerConsumerImpl.this, context.getSize(), ref.getDeliveryCount());
if (availableCredits != null) {
final int credits = availableCredits.addAndGet(-packetSize);

12
artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/SessionCallback.java
View File

@ -16,7 +16,6 @@
*/
package org.apache.activemq.artemis.spi.core.protocol;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.server.MessageReference;
import org.apache.activemq.artemis.core.server.ServerConsumer;
@ -69,16 +68,9 @@ public interface SessionCallback {
void sendProducerCreditsFailMessage(int credits, SimpleString address);
// Note: don't be tempted to remove the parameter message
// Even though ref will contain the message in certain cases
// such as paging the message could be a SoftReference or WeakReference
// and I wanted to avoid re-fetching paged data in case of GCs on this specific case.
//
// Future developments may change this, but beware why I have chosen to keep the parameter separated here
int sendMessage(MessageReference ref, Message message, ServerConsumer consumerID, int deliveryCount);
int sendMessage(MessageReference ref, ServerConsumer consumerID, int deliveryCount);
int sendLargeMessage(MessageReference reference,
Message message,
int sendLargeMessage(MessageReference ref,
ServerConsumer consumerID,
long bodySize,
int deliveryCount);

10
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/HangConsumerTest.java
View File

@ -28,7 +28,6 @@ import java.util.concurrent.TimeUnit;
import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.Interceptor;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.QueueConfiguration;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString;
@ -518,7 +517,7 @@ public class HangConsumerTest extends ActiveMQTestBase {
* @see SessionCallback#sendJmsMessage(org.apache.activemq.artemis.core.server.ServerMessage, long, int)
*/
@Override
public int sendMessage(MessageReference ref, Message message, ServerConsumer consumer, int deliveryCount) {
public int sendMessage(MessageReference ref, ServerConsumer consumer, int deliveryCount) {
inCall.countDown();
try {
callbackSemaphore.acquire();
@ -528,7 +527,7 @@ public class HangConsumerTest extends ActiveMQTestBase {
}
try {
return targetCallback.sendMessage(ref, message, consumer, deliveryCount);
return targetCallback.sendMessage(ref, consumer, deliveryCount);
} finally {
callbackSemaphore.release();
inCall.countUp();
@ -539,12 +538,11 @@ public class HangConsumerTest extends ActiveMQTestBase {
* @see SessionCallback#sendLargeMessage(org.apache.activemq.artemis.core.server.ServerMessage, long, long, int)
*/
@Override
public int sendLargeMessage(MessageReference reference,
Message message,
public int sendLargeMessage(MessageReference ref,
ServerConsumer consumer,
long bodySize,
int deliveryCount) {
return targetCallback.sendLargeMessage(reference, message, consumer, bodySize, deliveryCount);
return targetCallback.sendLargeMessage(ref, consumer, bodySize, deliveryCount);
}
/* (non-Javadoc)