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

ARTEMIS-3875 - adding consumer and producer metrics 

https://issues.apache.org/jira/browse/ARTEMIS-3875
This commit is contained in:
Andy Taylor 2022-06-28 10:03:57 +01:00 committed by Clebert Suconic
parent bc1258ab25
commit b02002fc66
65 changed files with 3604 additions and 321 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientProducerImpl.java
View File

@ -44,6 +44,8 @@ public class ClientProducerImpl implements ClientProducerInternal {
private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private final int id;
private final SimpleString address;
private final ClientSessionInternal session;
@ -75,7 +77,10 @@ public class ClientProducerImpl implements ClientProducerInternal {
final boolean autoGroup,
final SimpleString groupID,
final int minLargeMessageSize,
final SessionContext sessionContext) {
final SessionContext sessionContext,
final int producerID) {
this.id = producerID;
this.sessionContext = sessionContext;
this.session = session;
@ -197,6 +202,11 @@ public class ClientProducerImpl implements ClientProducerInternal {
return producerCredits;
}
@Override
public int getID() {
return id;
}
private void doCleanup() {
if (address != null) {
session.returnCredits(address);
@ -204,6 +214,8 @@ public class ClientProducerImpl implements ClientProducerInternal {
session.removeProducer(this);
sessionContext.removeProducer(id);
closed = true;
}
@ -290,7 +302,7 @@ public class ClientProducerImpl implements ClientProducerInternal {
theCredits.acquireCredits(creditSize);
sessionContext.sendFullMessage(msgI, sendBlocking, handler, address);
sessionContext.sendFullMessage(msgI, sendBlocking, handler, address, id);
}
private void checkClosed() throws ActiveMQException {
@ -381,7 +393,7 @@ public class ClientProducerImpl implements ClientProducerInternal {
lastChunk = pos >= bodySize;
SendAcknowledgementHandler messageHandler = lastChunk ? handler : null;
int creditsUsed = sessionContext.sendServerLargeMessageChunk(msgI, -1, sendBlocking, lastChunk, bodyBuffer.array(), messageHandler);
int creditsUsed = sessionContext.sendServerLargeMessageChunk(msgI, -1, sendBlocking, lastChunk, bodyBuffer.array(), id, messageHandler);
credits.acquireCredits(creditsUsed);
}
@ -492,7 +504,7 @@ public class ClientProducerImpl implements ClientProducerInternal {
headerSent = true;
sendInitialLargeMessageHeader(msgI, credits);
}
int creditsSent = sessionContext.sendLargeMessageChunk(msgI, messageSize.get(), sendBlocking, true, buff, reconnectID, handler);
int creditsSent = sessionContext.sendLargeMessageChunk(msgI, messageSize.get(), sendBlocking, true, buff, reconnectID, id, handler);
credits.acquireCredits(creditsSent);
}
} else {
@ -501,7 +513,7 @@ public class ClientProducerImpl implements ClientProducerInternal {
sendInitialLargeMessageHeader(msgI, credits);
}
int creditsSent = sessionContext.sendLargeMessageChunk(msgI, messageSize.get(), sendBlocking, false, buff, reconnectID, handler);
int creditsSent = sessionContext.sendLargeMessageChunk(msgI, messageSize.get(), sendBlocking, false, buff, reconnectID, id, handler);
credits.acquireCredits(creditsSent);
}
}

2
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientProducerInternal.java
View File

@ -26,4 +26,6 @@ public interface ClientProducerInternal extends ClientProducer {
void cleanUp();
ClientProducerCredits getProducerCredits();
int getID();
}

14
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientSessionImpl.java
View File

@ -161,6 +161,8 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
private final CoreMessageObjectPools coreMessageObjectPools = new CoreMessageObjectPools();
private AtomicInteger producerIDs = new AtomicInteger();
ClientSessionImpl(final ClientSessionFactoryInternal sessionFactory,
final String name,
final String username,
@ -2031,10 +2033,20 @@ public final class ClientSessionImpl implements ClientSessionInternal, FailureLi
final int maxRate) throws ActiveMQException {
checkClosed();
ClientProducerInternal producer = new ClientProducerImpl(this, address, maxRate == -1 ? null : new TokenBucketLimiterImpl(maxRate, false), autoCommitSends && blockOnNonDurableSend, autoCommitSends && blockOnDurableSend, autoGroup, groupID == null ? null : new SimpleString(groupID), minLargeMessageSize, sessionContext);
ClientProducerInternal producer = new ClientProducerImpl(this,
address,
maxRate == -1 ? null : new TokenBucketLimiterImpl(maxRate, false),
autoCommitSends && blockOnNonDurableSend,
autoCommitSends && blockOnDurableSend,
autoGroup, groupID == null ? null : new SimpleString(groupID),
minLargeMessageSize,
sessionContext,
producerIDs.incrementAndGet());
addProducer(producer);
sessionContext.createProducer(producer);
return producer;
}

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

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

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

@ -58,6 +58,7 @@ import org.apache.activemq.artemis.core.client.impl.ClientConsumerInternal;
import org.apache.activemq.artemis.core.client.impl.ClientLargeMessageInternal;
import org.apache.activemq.artemis.core.client.impl.ClientMessageInternal;
import org.apache.activemq.artemis.core.client.impl.ClientProducerCredits;
import org.apache.activemq.artemis.core.client.impl.ClientProducerInternal;
import org.apache.activemq.artemis.core.client.impl.ClientSessionImpl;
import org.apache.activemq.artemis.core.message.impl.CoreMessage;
import org.apache.activemq.artemis.core.protocol.core.Channel;
@ -68,6 +69,7 @@ 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.CreateProducerMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.CreateQueueMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.CreateQueueMessage_V2;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.CreateSessionMessage;
@ -76,6 +78,7 @@ import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.Disconnect
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.DisconnectConsumerWithKillMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReattachSessionMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReattachSessionResponseMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.RemoveProducerMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.RollbackMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionAcknowledgeMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionAddMetaDataMessageV2;
@ -105,10 +108,12 @@ import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionRec
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.SessionSendContinuationMessage_V3;
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.SessionSendMessage_V3;
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;
@ -371,6 +376,20 @@ public class ActiveMQSessionContext extends SessionContext {
return remotingConnection.isWritable(callback);
}
@Override
public void createProducer(ClientProducerInternal producer) {
if (!sessionChannel.getConnection().isBeforeProducerMetricsChanged()) {
sessionChannel.send(new CreateProducerMessage(producer.getID(), producer.getAddress()));
}
}
@Override
public void removeProducer(int id) {
if (!sessionChannel.getConnection().isBeforeProducerMetricsChanged()) {
sessionChannel.send(new RemoveProducerMessage(id));
}
}
@Override
public ClientConsumerInternal createConsumer(SimpleString queueName,
SimpleString filterString,
@ -546,15 +565,19 @@ public class ActiveMQSessionContext extends SessionContext {
public void sendFullMessage(ICoreMessage msgI,
boolean sendBlocking,
SendAcknowledgementHandler handler,
SimpleString defaultAddress) throws ActiveMQException {
SimpleString defaultAddress,
int senderID) throws ActiveMQException {
final SessionSendMessage packet;
if (sessionChannel.getConnection().isVersionBeforeAddressChange()) {
packet = new SessionSendMessage_1X(msgI, sendBlocking, handler);
} else if (sessionChannel.getConnection().isVersionBeforeAsyncResponseChange()) {
packet = new SessionSendMessage(msgI, sendBlocking, handler);
} else {
} else if (sessionChannel.getConnection().isBeforeProducerMetricsChanged()) {
boolean responseRequired = confirmationWindow != -1 || sendBlocking;
packet = new SessionSendMessage_V2(msgI, responseRequired, handler);
} else {
boolean responseRequired = confirmationWindow != -1 || sendBlocking;
packet = new SessionSendMessage_V3(msgI, responseRequired, handler, senderID);
}
if (sendBlocking) {
sessionChannel.sendBlocking(packet, PacketImpl.NULL_RESPONSE);
@ -579,8 +602,9 @@ public class ActiveMQSessionContext extends SessionContext {
boolean lastChunk,
byte[] chunk,
int reconnectID,
int senderID,
SendAcknowledgementHandler messageHandler) throws ActiveMQException {
return sendSessionSendContinuationMessage(this.sessionChannel, msgI, messageBodySize, sendBlocking, lastChunk, chunk, messageHandler);
return sendSessionSendContinuationMessage(this.sessionChannel, msgI, messageBodySize, sendBlocking, lastChunk, chunk, senderID, messageHandler);
}
@Override
@ -589,8 +613,9 @@ public class ActiveMQSessionContext extends SessionContext {
boolean sendBlocking,
boolean lastChunk,
byte[] chunk,
int senderID,
SendAcknowledgementHandler messageHandler) throws ActiveMQException {
return sendSessionSendContinuationMessage(this.sessionChannel, msgI, messageBodySize, sendBlocking, lastChunk, chunk, messageHandler);
return sendSessionSendContinuationMessage(this.sessionChannel, msgI, messageBodySize, sendBlocking, lastChunk, chunk, senderID, messageHandler);
}
@Override
@ -1043,13 +1068,16 @@ public class ActiveMQSessionContext extends SessionContext {
boolean sendBlocking,
boolean lastChunk,
byte[] chunk,
int senderID,
SendAcknowledgementHandler messageHandler) throws ActiveMQException {
final boolean requiresResponse = lastChunk && sendBlocking;
final SessionSendContinuationMessage chunkPacket;
if (sessionChannel.getConnection().isVersionBeforeAsyncResponseChange()) {
chunkPacket = new SessionSendContinuationMessage(msgI, chunk, !lastChunk, requiresResponse, messageBodySize, messageHandler);
} else {
} else if (sessionChannel.getConnection().isBeforeProducerMetricsChanged()) {
chunkPacket = new SessionSendContinuationMessage_V2(msgI, chunk, !lastChunk, requiresResponse || confirmationWindow != -1, messageBodySize, messageHandler);
} else {
chunkPacket = new SessionSendContinuationMessage_V3(msgI, chunk, !lastChunk, requiresResponse || confirmationWindow != -1, messageBodySize, senderID, messageHandler);
}
//perform a weak form of flow control to avoid OOM on tight loops
final CoreRemotingConnection connection = channel.getConnection();

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

@ -79,6 +79,7 @@ import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionRec
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.SessionSendContinuationMessage_V3;
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;
@ -423,8 +424,10 @@ public abstract class PacketDecoder implements Serializable {
case SESS_SEND_CONTINUATION: {
if (connection.isVersionBeforeAsyncResponseChange()) {
packet = new SessionSendContinuationMessage();
} else {
} else if (connection.isBeforeProducerMetricsChanged()) {
packet = new SessionSendContinuationMessage_V2();
} else {
packet = new SessionSendContinuationMessage_V3();
}
break;
}

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

@ -47,6 +47,9 @@ public class PacketImpl implements Packet {
// 2.24.0
public static final int ARTEMIS_2_24_0_VERSION = 133;
// 2.28.0
public static final int ARTEMIS_2_28_0_VERSION = 134;
public static final SimpleString OLD_QUEUE_PREFIX = new SimpleString("jms.queue.");
public static final SimpleString OLD_TEMP_QUEUE_PREFIX = new SimpleString("jms.tempqueue.");
public static final SimpleString OLD_TOPIC_PREFIX = new SimpleString("jms.topic.");
@ -293,7 +296,9 @@ public class PacketImpl implements Packet {
public static final byte DISCONNECT_V3 = -19;
public static final byte CREATE_PRODUCER = -20;
public static final byte REMOVE_PRODUCER = -21;
public PacketImpl(final byte type) {
this.type = type;

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

@ -0,0 +1,83 @@
/*
* 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.SimpleString;
import org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl;
import java.util.Objects;
public class CreateProducerMessage extends PacketImpl {
protected int id;
protected SimpleString address;
public CreateProducerMessage() {
super(PacketImpl.CREATE_PRODUCER);
}
public CreateProducerMessage(int id, SimpleString address) {
super(PacketImpl.CREATE_PRODUCER);
this.id = id;
this.address = address;
}
public int getId() {
return id;
}
public void setId(int id) {
this.id = id;
}
public SimpleString getAddress() {
return address;
}
public void setAddress(SimpleString address) {
this.address = address;
}
@Override
public void encodeRest(final ActiveMQBuffer buffer) {
buffer.writeInt(id);
buffer.writeNullableSimpleString(address);
}
@Override
public void decodeRest(final ActiveMQBuffer buffer) {
id = buffer.readInt();
address = buffer.readNullableSimpleString();
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
if (!super.equals(o)) return false;
CreateProducerMessage that = (CreateProducerMessage) o;
return Objects.equals(id, that.id) && Objects.equals(address, that.address);
}
@Override
public int hashCode() {
return Objects.hash(super.hashCode(), id, address);
}
}

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

@ -0,0 +1,68 @@
/*
* 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;
import java.util.Objects;
public class RemoveProducerMessage extends PacketImpl {
protected int id;
public RemoveProducerMessage(int id) {
super(PacketImpl.REMOVE_PRODUCER);
this.id = id;
}
public RemoveProducerMessage() {
super(PacketImpl.REMOVE_PRODUCER);
}
public int getId() {
return id;
}
public void setId(int id) {
this.id = id;
}
@Override
public void encodeRest(final ActiveMQBuffer buffer) {
buffer.writeInt(id);
}
@Override
public void decodeRest(final ActiveMQBuffer buffer) {
id = buffer.readInt();
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
if (!super.equals(o)) return false;
RemoveProducerMessage that = (RemoveProducerMessage) o;
return Objects.equals(id, that.id);
}
@Override
public int hashCode() {
return Objects.hash(super.hashCode(), id);
}
}

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

@ -180,4 +180,8 @@ public class SessionSendContinuationMessage extends SessionContinuationMessage {
public SendAcknowledgementHandler getHandler() {
return handler;
}
public int getSenderID() {
return -1;
}
}

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

@ -0,0 +1,72 @@
/*
* 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_V3 extends SessionSendContinuationMessage_V2 {
private int senderID;
public SessionSendContinuationMessage_V3() {
super();
}
/**
* @param body
* @param continues
* @param requiresResponse
*/
public SessionSendContinuationMessage_V3(final Message message,
final byte[] body,
final boolean continues,
final boolean requiresResponse,
final long messageBodySize,
final int senderID,
final SendAcknowledgementHandler handler) {
super(message, body, continues, requiresResponse, messageBodySize, handler);
this.senderID = senderID;
}
@Override
public int expectedEncodeSize() {
return super.expectedEncodeSize() + DataConstants.SIZE_INT;
}
@Override
public void encodeRest(final ActiveMQBuffer buffer) {
super.encodeRest(buffer);
buffer.writeInt(senderID);
}
@Override
public void decodeRest(final ActiveMQBuffer buffer) {
super.decodeRest(buffer);
senderID = buffer.readInt();
}
@Override
public int getSenderID() {
return senderID;
}
}

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

@ -136,4 +136,7 @@ public class SessionSendMessage extends MessagePacket {
return true;
}
public int getSenderID() {
return -1;
}
}

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

@ -0,0 +1,63 @@
/*
* 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.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_V3 extends SessionSendMessage_V2 {
private int senderID;
/** This will be using the CoreMessage because it is meant for the core-protocol */
public SessionSendMessage_V3(final ICoreMessage message,
final boolean requiresResponse,
final SendAcknowledgementHandler handler,
final int senderID) {
super(message, requiresResponse, handler);
this.senderID = senderID;
}
public SessionSendMessage_V3(final CoreMessage message) {
super(message);
}
@Override
public void encodeRest(ActiveMQBuffer buffer) {
super.encodeRest(buffer);
buffer.writeInt(senderID);
}
@Override
public void decodeRest(final ActiveMQBuffer buffer) {
super.decodeRest(buffer);
senderID = buffer.readInt();
}
@Override
protected int fieldsEncodeSize() {
return super.fieldsEncodeSize() + DataConstants.SIZE_INT;
}
@Override
public int getSenderID() {
return senderID;
}
}

10
artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/remoting/SessionContext.java
View File

@ -37,6 +37,7 @@ import org.apache.activemq.artemis.core.client.impl.ClientConsumerInternal;
import org.apache.activemq.artemis.core.client.impl.ClientLargeMessageInternal;
import org.apache.activemq.artemis.core.client.impl.ClientMessageInternal;
import org.apache.activemq.artemis.core.client.impl.ClientProducerCredits;
import org.apache.activemq.artemis.core.client.impl.ClientProducerInternal;
import org.apache.activemq.artemis.core.client.impl.ClientSessionInternal;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionQueueQueryResponseMessage;
import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
@ -142,7 +143,8 @@ public abstract class SessionContext {
public abstract void sendFullMessage(ICoreMessage msgI,
boolean sendBlocking,
SendAcknowledgementHandler handler,
SimpleString defaultAddress) throws ActiveMQException;
SimpleString defaultAddress,
int senderID) throws ActiveMQException;
/**
* it should return the number of credits (or bytes) used to send this packet
@ -159,6 +161,7 @@ public abstract class SessionContext {
boolean lastChunk,
byte[] chunk,
int reconnectID,
int senderID,
SendAcknowledgementHandler messageHandler) throws ActiveMQException;
public abstract int sendServerLargeMessageChunk(Message msgI,
@ -166,6 +169,7 @@ public abstract class SessionContext {
boolean sendBlocking,
boolean lastChunk,
byte[] chunk,
int senderID,
SendAcknowledgementHandler messageHandler) throws ActiveMQException;
public abstract void setSendAcknowledgementHandler(SendAcknowledgementHandler handler);
@ -385,4 +389,8 @@ public abstract class SessionContext {
public abstract void linkFlowControl(SimpleString address, ClientProducerCredits clientProducerCredits);
public abstract boolean isWritable(ReadyListener callback);
public abstract void createProducer(ClientProducerInternal producer);
public abstract void removeProducer(int id);
}

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

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

47
artemis-hawtio/artemis-plugin/src/main/webapp/plugin/js/components/consumers.js
View File

@ -98,7 +98,15 @@ var Artemis;
{name: "Address", visible: true},
{name: "Remote Address", visible: true},
{name: "Local Address", visible: true},
{name: "Creation Time", visible: true}
{name: "Creation Time", visible: true},
{name: "Messages in Transit", visible: false},
{name: "Messages in Transit Size", visible: false},
{name: "Messages Delivered", visible: false},
{name: "Messages Delivered Size", visible: false},
{name: "Messages Acknowledged", visible: false},
{name: "Messages Acknowledged awaiting Commit", visible: false},
{name: "Last Delivered Time", visible: false},
{name: "Last Acknowledged Time", visible: false}
]
};
@ -130,7 +138,13 @@ var Artemis;
{id: 'queue', name: 'Queue'},
{id: 'protocol', name: 'Protocol'},
{id: 'localAddress', name: 'Local Address'},
{id: 'remoteAddress', name: 'Remote Address'}
{id: 'remoteAddress', name: 'Remote Address'},
{id: 'messagesInTransit', name: 'Messages in Transit'},
{id: 'messagesInTransitSize', name: 'Messages in Transit Size'},
{id: 'messagesDelivered', name: 'Messages Delivered'},
{id: 'messagesDeliveredSize', name: 'Messages Delivered Size'},
{id: 'messagesAcknowledged', name: 'Messages Acknowledged'},
{id: 'messagesAcknowledgedAwaitingCommit', name: 'Messages Acknowledged awaiting Commit'}
],
operationOptions: [
{id: 'EQUALS', name: 'Equals'},
@ -181,7 +195,15 @@ var Artemis;
{ header: 'Address', itemField: 'addressName' , htmlTemplate: 'consumers-anchor-column-template', colActionFn: (item) => selectAddress(item.idx) },
{ header: 'Remote Address', itemField: 'remoteAddress' },
{ header: 'Local Address', itemField: 'localAddress' },
{ header: 'Creation Time', itemField: 'creationTime' }
{ header: 'Creation Time', itemField: 'creationTime' },
{ header: 'Messages in Transit', itemField: 'messagesInTransit' },
{ header: 'Messages in Transit Size', itemField: 'messagesInTransitSize' },
{ header: 'Messages Delivered', itemField: 'messagesDelivered' },
{ header: 'Messages Delivered Size', itemField: 'messagesDeliveredSize' },
{ header: 'Messages Acknowledged', itemField: 'messagesAcknowledged' },
{ header: 'Messages Acknowledged awaiting Commit', itemField: 'messagesAcknowledgedAwaitingCommit' },
{ header: 'Last Delivered Time', itemField: 'lastDeliveredTime', templateFn: function(value) { return formatTimestamp(value);} },
{ header: 'Last Acknowledged Time', itemField: 'lastAcknowledgedTime' , templateFn: function(value) { return formatTimestamp(value);} }
];
ctrl.refresh = function () {
@ -279,6 +301,25 @@ var Artemis;
}
};
function formatTimestamp(timestamp) {
Artemis.log.info("in timestamp " + timestamp + " " + (typeof timestamp !== "number"));
if (isNaN(timestamp) || typeof timestamp !== "number") {
Artemis.log.info("returning timestamp " + timestamp + " " + (typeof timestamp !== "number"));
return timestamp;
}
if (timestamp === 0) {
return "N/A";
}
var d = new Date(timestamp);
// "yyyy-MM-dd HH:mm:ss"
//add 1 to month as getmonth returns the position not the actual month
return d.getFullYear() + "-" + pad2(d.getMonth() + 1) + "-" + pad2(d.getDate()) + " " + pad2(d.getHours()) + ":" + pad2(d.getMinutes()) + ":" + pad2(d.getSeconds());
}
function pad2(value) {
return (value < 10 ? '0' : '') + value;
}
ctrl.pagination.setOperation(ctrl.loadOperation);
function onError(response) {

13
artemis-hawtio/artemis-plugin/src/main/webapp/plugin/js/components/producers.js
View File

@ -72,6 +72,7 @@ var Artemis;
ordering: false,
columns: [
{name: "ID", visible: true},
{name: "Name", visible: true},
{name: "Session", visible: true},
{name: "Client ID", visible: true},
{name: "Protocol", visible: true},
@ -79,7 +80,10 @@ var Artemis;
{name: "Validated User", visible: false},
{name: "Address", visible: true},
{name: "Remote Address", visible: true},
{name: "Local Address", visible: true}
{name: "Local Address", visible: true},
{name: "Messages Sent", visible: false},
{name: "Messages Sent Size", visible: false},
{name: "Last Produced Message ID", visible: false}
]
};
@ -103,6 +107,7 @@ var Artemis;
ctrl.filter = {
fieldOptions: [
{id: 'id', name: 'ID'},
{id: 'name', name: 'Name'},
{id: 'session', name: 'Session'},
{id: 'clientID', name: 'Client ID'},
{id: 'user', name: 'User'},
@ -140,6 +145,7 @@ var Artemis;
};
ctrl.tableColumns = [
{ header: 'ID', itemField: 'id' },
{ header: 'Name', itemField: 'name' },
{ header: 'Session', itemField: 'session' , htmlTemplate: 'producers-anchor-column-template', colActionFn: (item) => selectSession(item.idx) },
{ header: 'Client ID', itemField: 'clientID' },
{ header: 'Protocol', itemField: 'protocol' },
@ -147,7 +153,10 @@ var Artemis;
{ header: 'Validated User', name: 'validatedUser'},
{ header: 'Address', itemField: 'addressName' , htmlTemplate: 'producers-anchor-column-template', colActionFn: (item) => selectAddress(item.idx) },
{ header: 'Remote Address', itemField: 'remoteAddress' },
{ header: 'Local Address', itemField: 'localAddress' }
{ header: 'Local Address', itemField: 'localAddress' },
{ header: 'Messages Sent', itemField: 'msgSent'},
{ header: 'Messages Sent Size', itemField: 'msgSizeSent'},
{ header: 'Last Produced Message ID', itemField: 'lastProducedMessageID'}
];
ctrl.refresh = function () {

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

@ -41,7 +41,6 @@ import org.apache.activemq.artemis.core.server.MessageReference;
import org.apache.activemq.artemis.core.server.QueueQueryResult;
import org.apache.activemq.artemis.core.server.RoutingContext;
import org.apache.activemq.artemis.core.server.ServerConsumer;
import org.apache.activemq.artemis.core.server.ServerProducer;
import org.apache.activemq.artemis.core.server.ServerSession;
import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
@ -202,14 +201,14 @@ public class AMQPSessionCallback implements SessionCallback {
false, // boolean autoCommitAcks,
false, // boolean preAcknowledge,
true, //boolean xa,
null, this, true, operationContext, manager.getPrefixes(), manager.getSecurityDomain());
null, this, true, operationContext, manager.getPrefixes(), manager.getSecurityDomain(), false);
} else {
serverSession = manager.getServer().createSession(name, connection.getUser(), connection.getPassword(), ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, protonSPI.getProtonConnectionDelegate(), // RemotingConnection remotingConnection,
false, // boolean autoCommitSends
false, // boolean autoCommitAcks,
false, // boolean preAcknowledge,
true, //boolean xa,
null, this, true, operationContext, manager.getPrefixes(), manager.getSecurityDomain(), connection.getValidatedUser());
null, this, true, operationContext, manager.getPrefixes(), manager.getSecurityDomain(), connection.getValidatedUser(), false);
}
}
@ -532,7 +531,7 @@ public class AMQPSessionCallback implements SessionCallback {
OperationContext oldContext = recoverContext();
try {
if (invokeIncoming((AMQPMessage) message, (ActiveMQProtonRemotingConnection) transportConnection.getProtocolConnection()) == null) {
serverSession.send(transaction, message, directDeliver, false, routingContext);
serverSession.send(transaction, message, directDeliver, receiver.getName(), false, routingContext);
afterIO(new IOCallback() {
@Override
@ -744,8 +743,8 @@ public class AMQPSessionCallback implements SessionCallback {
return protonSPI.invokeOutgoingInterceptors(message, connection);
}
public void addProducer(ServerProducer serverProducer) {
serverSession.addProducer(serverProducer);
public void addProducer(String name, String address) {
serverSession.addProducer(name, ProtonProtocolManagerFactory.AMQP_PROTOCOL_NAME, address);
}
public void removeProducer(String name) {

9
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPSessionContext.java
View File

@ -24,8 +24,6 @@ import java.util.concurrent.ConcurrentHashMap;
import org.apache.activemq.artemis.api.core.ActiveMQSecurityException;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.ServerProducer;
import org.apache.activemq.artemis.core.server.impl.ServerProducerImpl;
import org.apache.activemq.artemis.protocol.amqp.connect.mirror.AMQPMirrorControllerSource;
import org.apache.activemq.artemis.protocol.amqp.connect.mirror.AMQPMirrorControllerTarget;
import org.apache.activemq.artemis.protocol.amqp.broker.AMQPSessionCallback;
@ -160,7 +158,6 @@ public class AMQPSessionContext extends ProtonInitializable {
}
public void removeReceiver(Receiver receiver) {
sessionSPI.removeProducer(receiver.getName());
receivers.remove(receiver);
}
@ -229,8 +226,7 @@ public class AMQPSessionContext extends ProtonInitializable {
AMQPMirrorControllerTarget protonReceiver = new AMQPMirrorControllerTarget(sessionSPI, connection, this, receiver, server);
protonReceiver.initialize();
receivers.put(receiver, protonReceiver);
ServerProducer serverProducer = new ServerProducerImpl(receiver.getName(), "AMQP", receiver.getTarget().getAddress());
sessionSPI.addProducer(serverProducer);
sessionSPI.addProducer(receiver.getName(), receiver.getTarget().getAddress());
receiver.setContext(protonReceiver);
HashMap<Symbol, Object> brokerIDProperties = new HashMap<>();
brokerIDProperties.put(AMQPMirrorControllerSource.BROKER_ID, server.getNodeID().toString());
@ -255,8 +251,7 @@ public class AMQPSessionContext extends ProtonInitializable {
ProtonServerReceiverContext protonReceiver = new ProtonServerReceiverContext(sessionSPI, connection, this, receiver);
protonReceiver.initialize();
receivers.put(receiver, protonReceiver);
ServerProducer serverProducer = new ServerProducerImpl(receiver.getName(), "AMQP", receiver.getTarget().getAddress());
sessionSPI.addProducer(serverProducer);
sessionSPI.addProducer(receiver.getName(), receiver.getTarget().getAddress());
receiver.setContext(protonReceiver);
connection.runNow(() -> {
receiver.open();

1
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerReceiverContext.java
View File

@ -270,6 +270,7 @@ public class ProtonServerReceiverContext extends ProtonAbstractReceiver {
@Override
public void close(boolean remoteLinkClose) throws ActiveMQAMQPException {
super.close(remoteLinkClose);
sessionSPI.removeProducer(receiver.getName());
org.apache.qpid.proton.amqp.messaging.Target target = (org.apache.qpid.proton.amqp.messaging.Target) receiver.getRemoteTarget();
if (target != null && target.getDynamic() && (target.getExpiryPolicy() == TerminusExpiryPolicy.LINK_DETACH || target.getExpiryPolicy() == TerminusExpiryPolicy.SESSION_END)) {
try {

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

@ -169,7 +169,9 @@ public class MQTTConnectionManager {
MQTTUtil.SESSION_AUTO_CREATE_QUEUE,
server.newOperationContext(),
session.getProtocolManager().getPrefixes(),
session.getProtocolManager().getSecurityDomain(), validatedUser);
session.getProtocolManager().getSecurityDomain(),
validatedUser,
false);
return (ServerSessionImpl) serverSession;
}

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

@ -42,8 +42,10 @@ import org.apache.activemq.artemis.core.io.IOCallback;
import org.apache.activemq.artemis.core.protocol.mqtt.exceptions.DisconnectException;
import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.ServerConsumer;
import org.apache.activemq.artemis.core.server.ServerProducer;
import org.apache.activemq.artemis.core.server.impl.ServerSessionImpl;
import org.apache.activemq.artemis.core.transaction.Transaction;
import org.apache.activemq.artemis.utils.UUIDGenerator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.invoke.MethodHandles;
@ -72,6 +74,10 @@ public class MQTTPublishManager {
private SimpleString managementAddress;
private final String senderName = UUIDGenerator.getInstance().generateUUID().toString();
private boolean createProducer = true;
private ServerConsumer managementConsumer;
private MQTTSession session;
@ -173,6 +179,10 @@ public class MQTTPublishManager {
*/
void sendToQueue(MqttPublishMessage message, boolean internal) throws Exception {
synchronized (lock) {
if (createProducer) {
session.getServerSession().addProducer(senderName, MQTTProtocolManagerFactory.MQTT_PROTOCOL_NAME, ServerProducer.ANONYMOUS);
createProducer = false;
}
String topic = message.variableHeader().topicName();
if (session.getVersion() == MQTTVersion.MQTT_5) {
Integer alias = MQTTUtil.getProperty(Integer.class, message.variableHeader().properties(), TOPIC_ALIAS);
@ -206,7 +216,6 @@ public class MQTTPublishManager {
if (qos > 0) {
serverMessage.setDurable(MQTTUtil.DURABLE_MESSAGES);
}
int messageId = message.variableHeader().packetId();
if (qos < 2 || !state.getPubRec().contains(messageId)) {
if (qos == 2 && !internal)
@ -217,7 +226,7 @@ public class MQTTPublishManager {
if (session.getServer().getAddressInfo(address) == null && session.getServer().getAddressSettingsRepository().getMatch(coreAddress).isAutoCreateAddresses()) {
session.getServerSession().createAddress(address, RoutingType.MULTICAST, true);
}
session.getServerSession().send(tx, serverMessage, true, false);
session.getServerSession().send(tx, serverMessage, true, senderName, false);
if (message.fixedHeader().isRetain()) {
ByteBuf payload = message.payload();
@ -303,7 +312,7 @@ public class MQTTPublishManager {
if (ref != null) {
Message m = MQTTUtil.createPubRelMessage(session, getManagementAddress(), messageId);
//send the management message via the internal server session to bypass security.
session.getInternalServerSession().send(m, true);
session.getInternalServerSession().send(m, true, senderName);
session.getServerSession().individualAcknowledge(ref.getB(), ref.getA());
releaseFlowControl(ref.getB());
} else {

7
artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireConnection.java
View File

@ -837,7 +837,7 @@ public class OpenWireConnection extends AbstractRemotingConnection implements Se
}
private void createInternalSession(ConnectionInfo info) throws Exception {
internalSession = server.createSession(UUIDGenerator.getInstance().generateStringUUID(), context.getUserName(), info.getPassword(), -1, this, true, false, false, false, null, null, true, operationContext, protocolManager.getPrefixes(), protocolManager.getSecurityDomain(), validatedUser);
internalSession = server.createSession(UUIDGenerator.getInstance().generateStringUUID(), context.getUserName(), info.getPassword(), -1, this, true, false, false, false, null, null, true, operationContext, protocolManager.getPrefixes(), protocolManager.getSecurityDomain(), validatedUser, false);
}
//raise the refCount of context
@ -1261,6 +1261,10 @@ public class OpenWireConnection extends AbstractRemotingConnection implements Se
}
ss.addProducer(info);
getSession(info.getProducerId().getParentId()).getCoreSession().addProducer(
info.getProducerId().toString(),
OpenWireProtocolManagerFactory.OPENWIRE_PROTOCOL_NAME,
info.getDestination() != null ? info.getDestination().getPhysicalName() : null);
}
return null;
@ -1290,6 +1294,7 @@ public class OpenWireConnection extends AbstractRemotingConnection implements Se
}
synchronized (producerExchanges) {
producerExchanges.remove(id);
getSession(id.getParentId()).getCoreSession().removeProducer(id.toString());
}
return null;
}

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

@ -362,6 +362,7 @@ public class AMQConsumer {
if (ack.isIndividualAck() || ack.isStandardAck()) {
for (MessageReference ref : ackList) {
ref.acknowledge(transaction, serverConsumer);
serverConsumer.metricsAcknowledge(ref, transaction);
removeRolledback(ref);
}
} else if (ack.isPoisonAck()) {

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

@ -132,7 +132,7 @@ public class AMQSession implements SessionCallback {
// now
try {
coreSession = server.createSession(name, username, password, minLargeMessageSize, connection, true, false, false, false, null, this, true, connection.getOperationContext(), protocolManager.getPrefixes(), protocolManager.getSecurityDomain(), connection.getValidatedUser());
coreSession = server.createSession(name, username, password, minLargeMessageSize, connection, true, false, false, false, null, this, true, connection.getOperationContext(), protocolManager.getPrefixes(), protocolManager.getSecurityDomain(), connection.getValidatedUser(), false);
} catch (Exception e) {
logger.error("error init session", e);
}
@ -392,7 +392,6 @@ public class AMQSession implements SessionCallback {
assert clientId.toString().equals(this.connection.getState().getInfo().getClientId()) : "Session cached clientId must be the same of the connection";
originalCoreMsg.putStringProperty(MessageUtil.CONNECTION_ID_PROPERTY_NAME, clientId);
/* ActiveMQ failover transport will attempt to reconnect after connection failure. Any sent messages that did
* not receive acks will be resent. (ActiveMQ broker handles this by returning a last sequence id received to
* the client). To handle this in Artemis we use a duplicate ID cache. To do this we check to see if the
@ -438,7 +437,7 @@ public class AMQSession implements SessionCallback {
restoreAutoRead();
}
getCoreSession().send(coreMsg, false, dest.isTemporary());
getCoreSession().send(coreMsg, false, producerInfo.getProducerId().toString(), dest.isTemporary());
if (count == null || count.decrementAndGet() == 0) {
if (sendProducerAck) {
@ -462,7 +461,7 @@ public class AMQSession implements SessionCallback {
Exception exceptionToSend = null;
try {
getCoreSession().send(coreMsg, false, dest.isTemporary());
getCoreSession().send(coreMsg, false, producerInfo.getProducerId().toString(), dest.isTemporary());
} catch (Exception e) {
logger.debug("Sending exception to the client", e);
exceptionToSend = e;

2
artemis-protocols/artemis-openwire-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/openwire/amq/OpenWireConnectionTest.java
View File

@ -71,7 +71,7 @@ public class OpenWireConnectionTest {
ServerSession serverSession = Mockito.mock(ServerSession.class);
Mockito.when(serverSession.getName()).thenReturn("session");
Mockito.doReturn(serverSession).when(server).createSession(Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.anyInt(), Mockito.any(), Mockito.anyBoolean(), Mockito.anyBoolean(), Mockito.anyBoolean(),
Mockito.anyBoolean(), Mockito.any(), Mockito.any(), Mockito.anyBoolean(), Mockito.any(), Mockito.any(), Mockito.anyString(), Mockito.any());
Mockito.anyBoolean(), Mockito.any(), Mockito.any(), Mockito.anyBoolean(), Mockito.any(), Mockito.any(), Mockito.anyString(), Mockito.any(), Mockito.anyBoolean());
OpenWireProtocolManager openWireProtocolManager = new OpenWireProtocolManager(null, server,null, null);
openWireProtocolManager.setSecurityDomain("securityDomain");

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

@ -247,7 +247,7 @@ public class StompProtocolManager extends AbstractProtocolManager<StompFrame, St
stompSession = new StompSession(connection, this, server.getStorageManager().newContext(server.getExecutorFactory().getExecutor()));
String name = UUIDGenerator.getInstance().generateStringUUID();
final String validatedUser = server.validateUser(connection.getLogin(), connection.getPasscode(), connection, getSecurityDomain());
ServerSession session = server.createSession(name, connection.getLogin(), connection.getPasscode(), ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, connection, !transacted, false, false, false, null, stompSession, true, server.newOperationContext(), getPrefixes(), getSecurityDomain(), validatedUser);
ServerSession session = server.createSession(name, connection.getLogin(), connection.getPasscode(), ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, connection, !transacted, false, false, false, null, stompSession, true, server.newOperationContext(), getPrefixes(), getSecurityDomain(), validatedUser, false);
stompSession.setServerSession(session);
sessions.put(id, stompSession);
}

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

@ -41,6 +41,7 @@ import org.apache.activemq.artemis.core.server.LargeServerMessage;
import org.apache.activemq.artemis.core.server.MessageReference;
import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.ServerConsumer;
import org.apache.activemq.artemis.core.server.ServerProducer;
import org.apache.activemq.artemis.core.server.ServerSession;
import org.apache.activemq.artemis.core.server.impl.ServerSessionImpl;
import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
@ -55,11 +56,16 @@ 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;
public class StompSession implements SessionCallback {
private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private final String senderName = UUIDGenerator.getInstance().generateUUID().toString();
private boolean createProducer = true;
private final StompProtocolManager manager;
private final StompConnection connection;
@ -217,6 +223,7 @@ public class StompSession implements SessionCallback {
@Override
public void closed() {
session.removeProducer(senderName);
}
@Override
@ -379,10 +386,18 @@ public class StompSession implements SessionCallback {
}
public void sendInternal(Message message, boolean direct) throws Exception {
session.send(message, direct);
if (createProducer) {
session.addProducer(senderName, STOMP_PROTOCOL_NAME, ServerProducer.ANONYMOUS);
createProducer = false;
}
session.send(message, direct, senderName);
}
public void sendInternalLarge(CoreMessage message, boolean direct) throws Exception {
if (createProducer) {
session.addProducer(senderName, STOMP_PROTOCOL_NAME, ServerProducer.ANONYMOUS);
createProducer = false;
}
int headerSize = message.getHeadersAndPropertiesEncodeSize();
if (headerSize >= connection.getMinLargeMessageSize()) {
throw BUNDLE.headerTooBig();
@ -402,7 +417,7 @@ public class StompSession implements SessionCallback {
largeMessage.toMessage().putLongProperty(Message.HDR_LARGE_BODY_SIZE, bytes.length);
session.send(largeMessage.toMessage(), direct);
session.send(largeMessage.toMessage(), direct, senderName);
}
}

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

@ -122,7 +122,7 @@ public abstract class AbstractControl extends StandardMBean {
Integer.MAX_VALUE, fakeConnection,
true, true, false,
false, address.toString(), fakeConnection.callback,
false, new DummyOperationContext(), Collections.emptyMap(), null, validatedUser);
false, new DummyOperationContext(), Collections.emptyMap(), null, validatedUser, false);
try {
CoreMessage message = new CoreMessage(storageManager.generateID(), 50);
if (headers != null) {
@ -159,7 +159,7 @@ public abstract class AbstractControl extends StandardMBean {
}
// There's no point on direct delivery using the management thread, use false here
serverSession.send(message, false);
serverSession.send(message, false, null);
return "" + message.getMessageID();
} finally {
try {

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

@ -79,6 +79,7 @@ import org.apache.activemq.artemis.core.config.TransformerConfiguration;
import org.apache.activemq.artemis.core.filter.Filter;
import org.apache.activemq.artemis.core.management.impl.view.AddressView;
import org.apache.activemq.artemis.core.management.impl.view.ConnectionView;
import org.apache.activemq.artemis.core.management.impl.view.ConsumerField;
import org.apache.activemq.artemis.core.management.impl.view.ConsumerView;
import org.apache.activemq.artemis.core.management.impl.view.ProducerView;
import org.apache.activemq.artemis.core.management.impl.view.QueueView;
@ -107,6 +108,7 @@ import org.apache.activemq.artemis.core.server.ConnectorServiceFactory;
import org.apache.activemq.artemis.core.server.Consumer;
import org.apache.activemq.artemis.core.server.Divert;
import org.apache.activemq.artemis.core.server.JournalType;
import org.apache.activemq.artemis.core.server.MessageReference;
import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.ServerConsumer;
import org.apache.activemq.artemis.core.server.ServerProducer;
@ -2609,7 +2611,7 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
try {
Set<ServerProducer> producers = new HashSet<>();
for (ServerSession session : server.getSessions()) {
producers.addAll(session.getServerProducers().values());
producers.addAll(session.getServerProducers());
}
ProducerView view = new ProducerView(server);
view.setCollection(producers);
@ -2767,7 +2769,24 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
}
private JsonObject toJSONObject(ServerConsumer consumer) throws Exception {
JsonObjectBuilder obj = JsonLoader.createObjectBuilder().add("consumerID", consumer.getID()).add("connectionID", consumer.getConnectionID().toString()).add("sessionID", consumer.getSessionID()).add("queueName", consumer.getQueue().getName().toString()).add("browseOnly", consumer.isBrowseOnly()).add("creationTime", consumer.getCreationTime()).add("deliveringCount", consumer.getDeliveringMessages().size());
List<MessageReference> deliveringMessages = consumer.getDeliveringMessages();
JsonObjectBuilder obj = JsonLoader.createObjectBuilder()
.add(ConsumerField.ID.getAlternativeName(), consumer.getID())
.add(ConsumerField.CONNECTION.getAlternativeName(), consumer.getConnectionID().toString())
.add(ConsumerField.SESSION.getAlternativeName(), consumer.getSessionID())
.add(ConsumerField.QUEUE.getAlternativeName(), consumer.getQueue().getName().toString())
.add(ConsumerField.BROWSE_ONLY.getName(), consumer.isBrowseOnly())
.add(ConsumerField.CREATION_TIME.getName(), consumer.getCreationTime())
// deliveringCount is renamed as MESSAGES_IN_TRANSIT but left in json for backward compatibility
.add(ConsumerField.MESSAGES_IN_TRANSIT.getAlternativeName(), consumer.getMessagesInTransit())
.add(ConsumerField.MESSAGES_IN_TRANSIT.getName(), consumer.getMessagesInTransit())
.add(ConsumerField.MESSAGES_IN_TRANSIT_SIZE.getName(), consumer.getMessagesInTransitSize())
.add(ConsumerField.MESSAGES_DELIVERED.getName(), consumer.getMessagesDelivered())
.add(ConsumerField.MESSAGES_DELIVERED_SIZE.getName(), consumer.getMessagesDeliveredSize())
.add(ConsumerField.MESSAGES_ACKNOWLEDGED.getName(), consumer.getMessagesAcknowledged())
.add(ConsumerField.MESSAGES_ACKNOWLEDGED_AWAITING_COMMIT.getName(), consumer.getMessagesAcknowledgedAwaitingCommit())
.add(ConsumerField.LAST_DELIVERED_TIME.getName(), consumer.getLastDeliveredTime())
.add(ConsumerField.LAST_ACKNOWLEDGED_TIME.getName(), consumer.getLastAcknowledgedTime());
if (consumer.getFilter() != null) {
obj.add("filter", consumer.getFilter().getFilterString().toString());
}
@ -4407,14 +4426,6 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
}
}
public String[] listTargetAddresses(final String sessionID) {
ServerSession session = server.getSessionByID(sessionID);
if (session != null) {
return session.getTargetAddresses();
}
return new String[0];
}
@Override
public void onNotification(org.apache.activemq.artemis.core.server.management.Notification notification) {
if (!(notification.getType() instanceof CoreNotificationType))

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

@ -36,6 +36,7 @@ import org.apache.activemq.artemis.api.core.management.QueueControl;
import org.apache.activemq.artemis.api.core.management.ResourceNames;
import org.apache.activemq.artemis.core.filter.Filter;
import org.apache.activemq.artemis.core.filter.impl.FilterImpl;
import org.apache.activemq.artemis.core.management.impl.view.ConsumerField;
import org.apache.activemq.artemis.core.messagecounter.MessageCounter;
import org.apache.activemq.artemis.core.messagecounter.impl.MessageCounterHelper;
import org.apache.activemq.artemis.core.persistence.StorageManager;
@ -1826,9 +1827,20 @@ public class QueueControlImpl extends AbstractControl implements QueueControl {
if (consumer instanceof ServerConsumer) {
ServerConsumer serverConsumer = (ServerConsumer) consumer;
JsonObjectBuilder obj = JsonLoader.createObjectBuilder().add("consumerID", serverConsumer.getID()).add("connectionID", serverConsumer.getConnectionID().toString()).add("sessionID", serverConsumer.getSessionID()).add("browseOnly", serverConsumer.isBrowseOnly()).add("creationTime", serverConsumer.getCreationTime());
JsonObjectBuilder obj = JsonLoader.createObjectBuilder()
.add(ConsumerField.ID.getAlternativeName(), serverConsumer.getID())
.add(ConsumerField.CONNECTION.getAlternativeName(), serverConsumer.getConnectionID().toString())
.add(ConsumerField.SESSION.getAlternativeName(), serverConsumer.getSessionID())
.add(ConsumerField.BROWSE_ONLY.getName(), serverConsumer.isBrowseOnly())
.add(ConsumerField.CREATION_TIME.getName(), serverConsumer.getCreationTime())
.add(ConsumerField.MESSAGES_IN_TRANSIT.getName(), serverConsumer.getMessagesInTransit())
.add(ConsumerField.MESSAGES_IN_TRANSIT_SIZE.getName(), serverConsumer.getMessagesInTransitSize())
.add(ConsumerField.MESSAGES_DELIVERED.getName(), serverConsumer.getMessagesDelivered())
.add(ConsumerField.MESSAGES_DELIVERED_SIZE.getName(), serverConsumer.getMessagesDeliveredSize())
.add(ConsumerField.MESSAGES_ACKNOWLEDGED.getName(), serverConsumer.getMessagesAcknowledged())
.add(ConsumerField.MESSAGES_ACKNOWLEDGED_AWAITING_COMMIT.getName(), serverConsumer.getMessagesAcknowledgedAwaitingCommit())
.add(ConsumerField.LAST_DELIVERED_TIME.getName(), serverConsumer.getLastDeliveredTime())
.add(ConsumerField.LAST_ACKNOWLEDGED_TIME.getName(), serverConsumer.getLastAcknowledgedTime());
jsonArray.add(obj);
}

34
artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/view/ConsumerField.java
View File

@ -20,9 +20,10 @@ import java.util.Map;
import java.util.TreeMap;
public enum ConsumerField {
ID("id"),
SESSION("session"),
QUEUE("queue"),
ID("id", "consumerID"),
SESSION("session", "sessionID"),
CONNECTION("connection", "connectionID"),
QUEUE("queue", "queueName"),
FILTER("filter"),
ADDRESS("address"),
USER("user"),
@ -32,7 +33,16 @@ public enum ConsumerField {
LOCAL_ADDRESS("localAddress"),
REMOTE_ADDRESS("remoteAddress"),
QUEUE_TYPE("queueType"),
CREATION_TIME("creationTime");
BROWSE_ONLY("browseOnly"),
CREATION_TIME("creationTime"),
MESSAGES_IN_TRANSIT("messagesInTransit", "deliveringCount"),
MESSAGES_IN_TRANSIT_SIZE("messagesInTransitSize"),
MESSAGES_DELIVERED("messagesDelivered"),
MESSAGES_DELIVERED_SIZE("messagesDeliveredSize"),
MESSAGES_ACKNOWLEDGED("messagesAcknowledged"),
MESSAGES_ACKNOWLEDGED_AWAITING_COMMIT("messagesAcknowledgedAwaitingCommit"),
LAST_DELIVERED_TIME("lastDeliveredTime"),
LAST_ACKNOWLEDGED_TIME("lastAcknowledgedTime");
private static final Map<String, ConsumerField> lookup = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
@ -44,12 +54,28 @@ public enum ConsumerField {
private final String name;
private final String alternativeName;
public String getName() {
return name;
}
/**
* There is some inconsistency with some json objects returned for consumers because they were hard coded.
* This is just to track the differences and provide backward compatibility.
* @return the old alternative name
*/
public String getAlternativeName() {
return alternativeName;
}
ConsumerField(String name) {
this(name, "");
}
ConsumerField(String name, String alternativeName) {
this.name = name;
this.alternativeName = alternativeName;
}
public static ConsumerField valueOfName(String name) {

29
artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/view/ConsumerView.java
View File

@ -16,6 +16,7 @@
*/
package org.apache.activemq.artemis.core.management.impl.view;
import org.apache.activemq.artemis.core.management.impl.ActiveMQServerControlImpl;
import org.apache.activemq.artemis.json.JsonObjectBuilder;
import java.util.Date;
@ -46,7 +47,7 @@ public class ConsumerView extends ActiveMQAbstractView<ServerConsumer> {
@Override
public JsonObjectBuilder toJson(ServerConsumer consumer) {
ServerSession session = server.getSessionByID(consumer.getSessionID());
ActiveMQServerControlImpl serverControl = server.getActiveMQServerControl();
//if session is not available then consumer is not in valid state - ignore
if (session == null) {
return null;
@ -71,7 +72,15 @@ public class ConsumerView extends ActiveMQAbstractView<ServerConsumer> {
.add(ConsumerField.ADDRESS.getName(), toString(consumer.getQueueAddress()))
.add(ConsumerField.LOCAL_ADDRESS.getName(), toString(consumer.getConnectionLocalAddress()))
.add(ConsumerField.REMOTE_ADDRESS.getName(), toString(consumer.getConnectionRemoteAddress()))
.add(ConsumerField.CREATION_TIME.getName(), new Date(consumer.getCreationTime()).toString());
.add(ConsumerField.CREATION_TIME.getName(), new Date(consumer.getCreationTime()).toString())
.add(ConsumerField.MESSAGES_IN_TRANSIT.getName(), toString(consumer.getMessagesInTransit()))
.add(ConsumerField.MESSAGES_IN_TRANSIT_SIZE.getName(), toString(consumer.getMessagesInTransitSize()))
.add(ConsumerField.MESSAGES_DELIVERED.getName(), toString(consumer.getMessagesDelivered()))
.add(ConsumerField.MESSAGES_DELIVERED_SIZE.getName(), toString(consumer.getMessagesDeliveredSize()))
.add(ConsumerField.MESSAGES_ACKNOWLEDGED.getName(), toString(consumer.getMessagesAcknowledged()))
.add(ConsumerField.MESSAGES_ACKNOWLEDGED_AWAITING_COMMIT.getName(), toString(consumer.getMessagesAcknowledgedAwaitingCommit()))
.add(ConsumerField.LAST_DELIVERED_TIME.getName(), consumer.getLastDeliveredTime())
.add(ConsumerField.LAST_ACKNOWLEDGED_TIME.getName(), consumer.getLastAcknowledgedTime());
return obj;
}
@ -111,6 +120,22 @@ public class ConsumerView extends ActiveMQAbstractView<ServerConsumer> {
return consumer.getConnectionRemoteAddress();
case CREATION_TIME:
return new Date(consumer.getCreationTime());
case MESSAGES_IN_TRANSIT:
return consumer.getMessagesInTransit();
case MESSAGES_IN_TRANSIT_SIZE:
return consumer.getMessagesInTransitSize();
case MESSAGES_DELIVERED:
return consumer.getMessagesDelivered();
case MESSAGES_DELIVERED_SIZE:
return consumer.getMessagesDeliveredSize();
case MESSAGES_ACKNOWLEDGED:
return consumer.getMessagesAcknowledged();
case MESSAGES_ACKNOWLEDGED_AWAITING_COMMIT:
return consumer.getMessagesAcknowledgedAwaitingCommit();
case LAST_DELIVERED_TIME:
return consumer.getLastDeliveredTime();
case LAST_ACKNOWLEDGED_TIME:
return consumer.getLastAcknowledgedTime();
default:
throw new IllegalArgumentException("Unsupported field, " + fieldName);
}

26
artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/view/ProducerField.java
View File

@ -21,15 +21,20 @@ import java.util.TreeMap;
public enum ProducerField {
ID("id"),
SESSION("session"),
NAME("name"),
SESSION("session", "sessionID"),
CONNECTION_ID("connectionID"),
ADDRESS("address"), USER("user"),
ADDRESS("address", "destination"),
USER("user"),
VALIDATED_USER("validatedUser"),
PROTOCOL("protocol"),
CLIENT_ID("clientID"),
LOCAL_ADDRESS("localAddress"),
REMOTE_ADDRESS("remoteAddress"),
CREATION_TIME("creationTime");
CREATION_TIME("creationTime"),
MESSAGE_SENT("msgSent"),
MESSAGE_SENT_SIZE("msgSizeSent"),
LAST_PRODUCED_MESSAGE_ID("lastProducedMessageID");
private static final Map<String, ProducerField> lookup = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
@ -41,12 +46,27 @@ public enum ProducerField {
private final String name;
private final String alternativeName;
public String getName() {
return name;
}
/**
* There is some inconsistency with some json objects returned for consumers because they were hard coded.
* This is just to track the differences and provide backward compatibility.
* @return the old alternative name
*/
public String getAlternativeName() {
return alternativeName;
}
ProducerField(String name) {
this(name, "");
}
ProducerField(String name, String alternativeName) {
this.name = name;
this.alternativeName = alternativeName;
}
public static ProducerField valueOfName(String name) {

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

@ -59,6 +59,7 @@ public class ProducerView extends ActiveMQAbstractView<ServerProducer> {
JsonObjectBuilder obj = JsonLoader.createObjectBuilder()
.add(ProducerField.ID.getName(), toString(producer.getID()))
.add(ProducerField.NAME.getName(), toString(producer.getName()))
.add(ProducerField.SESSION.getName(), toString(session.getName()))
.add(ProducerField.CLIENT_ID.getName(), toString(sessionClientID))
.add(ProducerField.USER.getName(), toString(session.getUsername()))
@ -67,7 +68,10 @@ public class ProducerView extends ActiveMQAbstractView<ServerProducer> {
.add(ProducerField.ADDRESS.getName(), toString(producer.getAddress() != null ? producer.getAddress() : session.getDefaultAddress()))
.add(ProducerField.LOCAL_ADDRESS.getName(), toString(session.getRemotingConnection().getTransportConnection().getLocalAddress()))
.add(ProducerField.REMOTE_ADDRESS.getName(), toString(session.getRemotingConnection().getTransportConnection().getRemoteAddress()))
.add(ProducerField.CREATION_TIME.getName(), toString(producer.getCreationTime()));
.add(ProducerField.CREATION_TIME.getName(), toString(producer.getCreationTime()))
.add(ProducerField.MESSAGE_SENT.getName(), producer.getMessagesSent())
.add(ProducerField.MESSAGE_SENT_SIZE.getName(), producer.getMessagesSentSize())
.add(ProducerField.LAST_PRODUCED_MESSAGE_ID.getName(), toString(producer.getLastProducedMessageID()));
return obj;
}

12
artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/view/predicate/ConsumerFilterPredicate.java
View File

@ -59,6 +59,18 @@ public class ConsumerFilterPredicate extends ActiveMQFilterPredicate<ServerConsu
return matches(server.getSessionByID(consumer.getSessionID()).getRemotingConnection().getTransportConnection().getLocalAddress());
case REMOTE_ADDRESS:
return matches(server.getSessionByID(consumer.getSessionID()).getRemotingConnection().getTransportConnection().getRemoteAddress());
case MESSAGES_IN_TRANSIT:
return matches(consumer.getMessagesInTransit());
case MESSAGES_IN_TRANSIT_SIZE:
return matches(consumer.getMessagesInTransitSize());
case MESSAGES_DELIVERED:
return matches(consumer.getDeliveringMessages());
case MESSAGES_DELIVERED_SIZE:
return matches(consumer.getMessagesDeliveredSize());
case MESSAGES_ACKNOWLEDGED:
return matches(consumer.getMessagesAcknowledged());
case MESSAGES_ACKNOWLEDGED_AWAITING_COMMIT:
return matches(consumer.getMessagesAcknowledgedAwaitingCommit());
}
return true;
}

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

@ -29,11 +29,13 @@ import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.BackupRequ
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.BackupResponseMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ClusterConnectMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ClusterConnectReplyMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.CreateProducerMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.FederationDownstreamConnectMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.MessagePacketI;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.NodeAnnounceMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.QuorumVoteMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.QuorumVoteReplyMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.RemoveProducerMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationAddMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationAddTXMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ReplicationCommitMessage;
@ -58,16 +60,19 @@ import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionSen
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.SessionSendMessage_V3;
import org.apache.activemq.artemis.core.server.LargeServerMessage;
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;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.CLUSTER_CONNECT;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.CLUSTER_CONNECT_REPLY;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.CREATE_PRODUCER;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.FEDERATION_DOWNSTREAM_CONNECT;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.NODE_ANNOUNCE;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.QUORUM_VOTE;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.QUORUM_VOTE_REPLY;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.REMOVE_PRODUCER;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.REPLICATION_APPEND;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.REPLICATION_APPEND_TX;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.REPLICATION_COMMIT_ROLLBACK;
@ -106,8 +111,10 @@ public class ServerPacketDecoder extends ClientPacketDecoder {
sendMessage = new SessionSendMessage_1X(new CoreMessage(this.coreMessageObjectPools));
} else if (connection.isVersionBeforeAsyncResponseChange()) {
sendMessage = new SessionSendMessage(new CoreMessage(this.coreMessageObjectPools));
} else {
} else if (connection.isBeforeProducerMetricsChanged()) {
sendMessage = new SessionSendMessage_V2(new CoreMessage(this.coreMessageObjectPools));
} else {
sendMessage = new SessionSendMessage_V3(new CoreMessage(this.coreMessageObjectPools));
}
sendMessage.decode(in);
@ -269,6 +276,14 @@ public class ServerPacketDecoder extends ClientPacketDecoder {
packet = new FederationDownstreamConnectMessage();
break;
}
case CREATE_PRODUCER: {
packet = new CreateProducerMessage();
break;
}
case REMOVE_PRODUCER: {
packet = new RemoveProducerMessage();
break;
}
default: {
packet = super.decode(packetType, connection);
}

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

@ -21,6 +21,7 @@ import javax.transaction.xa.Xid;
import java.util.List;
import java.util.Objects;
import io.netty.util.collection.IntObjectHashMap;
import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.ActiveMQExceptionType;
import org.apache.activemq.artemis.api.core.ActiveMQIOErrorException;
@ -31,6 +32,7 @@ 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;
import org.apache.activemq.artemis.api.core.client.ActiveMQClient;
import org.apache.activemq.artemis.api.core.client.ClientSession;
import org.apache.activemq.artemis.core.exception.ActiveMQXAException;
import org.apache.activemq.artemis.core.io.IOCallback;
@ -39,12 +41,14 @@ import org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ActiveMQExceptionMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ActiveMQExceptionMessage_V2;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.CreateAddressMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.CreateProducerMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.CreateQueueMessage;
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.NullResponseMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.NullResponseMessage_V2;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.RemoveProducerMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.RollbackMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionAcknowledgeMessage;
import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.SessionAddMetaDataMessage;
@ -97,6 +101,7 @@ import org.apache.activemq.artemis.core.server.ServerSession;
import org.apache.activemq.artemis.logs.AuditLogger;
import org.apache.activemq.artemis.spi.core.protocol.EmbedMessageUtil;
import org.apache.activemq.artemis.spi.core.remoting.Connection;
import org.apache.activemq.artemis.utils.UUIDGenerator;
import org.apache.activemq.artemis.utils.pools.MpscPool;
import org.apache.activemq.artemis.utils.pools.Pool;
import org.apache.activemq.artemis.utils.SimpleFuture;
@ -108,11 +113,13 @@ import org.slf4j.LoggerFactory;
import java.lang.invoke.MethodHandles;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.CREATE_ADDRESS;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.CREATE_PRODUCER;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.CREATE_QUEUE;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.CREATE_QUEUE_V2;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.CREATE_SHARED_QUEUE;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.CREATE_SHARED_QUEUE_V2;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.DELETE_QUEUE;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.REMOVE_PRODUCER;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_ACKNOWLEDGE;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_BINDINGQUERY;
import static org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl.SESS_CLOSE;
@ -151,6 +158,8 @@ public class ServerSessionPacketHandler implements ChannelHandler {
private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private static final String PRODUCER_ID_PREFIX = "artemis:sender:ID:";
private final ServerSession session;
private final StorageManager storageManager;
@ -174,6 +183,8 @@ public class ServerSessionPacketHandler implements ChannelHandler {
private final Pool<NullResponseMessage_V2> poolNullResponseV2;
private final IntObjectHashMap<String> producers = new IntObjectHashMap<>();
public ServerSessionPacketHandler(final ActiveMQServer server,
final ServerSession session,
final Channel channel) {
@ -323,7 +334,8 @@ public class ServerSessionPacketHandler implements ChannelHandler {
case SESS_SEND_CONTINUATION: {
SessionSendContinuationMessage message = (SessionSendContinuationMessage) packet;
requiresResponse = message.isRequiresResponse();
sendContinuations(message.getPacketSize(), message.getMessageBodySize(), message.getBody(), message.isContinues());
int senderID = message.getSenderID();
sendContinuations(message.getPacketSize(), message.getMessageBodySize(), message.getBody(), message.isContinues(), senderID);
if (requiresResponse) {
response = createNullResponseMessage(packet);
}
@ -635,6 +647,28 @@ public class ServerSessionPacketHandler implements ChannelHandler {
}
break;
}
case CREATE_PRODUCER: {
CreateProducerMessage message = (CreateProducerMessage) packet;
if (!producers.containsKey(message.getId())) {
// this is used to create/destroy the producer so needs to be unique
String senderName = PRODUCER_ID_PREFIX + UUIDGenerator.getInstance().generateUUID();
producers.put(message.getId(), senderName);
session.addProducer(senderName, ActiveMQClient.DEFAULT_CORE_PROTOCOL, message.getAddress() != null ? message.getAddress().toString() : null);
} else {
ActiveMQServerLogger.LOGGER.producerAlreadyExists(message.getId(),session.getName(), remotingConnection.getRemoteAddress());
}
break;
}
case REMOVE_PRODUCER: {
RemoveProducerMessage message = (RemoveProducerMessage) packet;
String remove = producers.remove(message.getId());
if (remove != null) {
session.removeProducer(remove);
} else {
ActiveMQServerLogger.LOGGER.producerDoesNotExist(message.getId(), session.getName(), remotingConnection.getRemoteAddress());
}
break;
}
}
} catch (ActiveMQIOErrorException e) {
response = onActiveMQIOErrorExceptionWhileHandlePacket(packet, e, requiresResponse, response, this.session);
@ -747,7 +781,7 @@ public class ServerSessionPacketHandler implements ChannelHandler {
try {
final SessionSendMessage message = (SessionSendMessage) packet;
requiresResponse = message.isRequiresResponse();
this.session.send(EmbedMessageUtil.extractEmbedded(message.getMessage(), storageManager), this.direct);
this.session.send(EmbedMessageUtil.extractEmbedded(message.getMessage(), storageManager), this.direct, producers.get(message.getSenderID()));
if (requiresResponse) {
response = createNullResponseMessage(packet);
}
@ -1049,7 +1083,8 @@ public class ServerSessionPacketHandler implements ChannelHandler {
private void sendContinuations(final int packetSize,
final long messageBodySize,
final byte[] body,
final boolean continues) throws Exception {
final boolean continues,
final int senderID) throws Exception {
synchronized (largeMessageLock) {
if (currentLargeMessage == null) {
@ -1071,7 +1106,7 @@ public class ServerSessionPacketHandler implements ChannelHandler {
LargeServerMessage message = currentLargeMessage;
currentLargeMessage.setStorageManager(storageManager);
currentLargeMessage = null;
session.doSend(session.getCurrentTransaction(), EmbedMessageUtil.extractEmbedded((ICoreMessage)message.toMessage(), storageManager), null, false, false);
session.doSend(session.getCurrentTransaction(), EmbedMessageUtil.extractEmbedded((ICoreMessage)message.toMessage(), storageManager), null, false, producers.get(senderID), false);
}
}
}

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

@ -45,9 +45,7 @@ import org.apache.activemq.artemis.core.security.ActiveMQPrincipal;
import org.apache.activemq.artemis.core.server.ActiveMQMessageBundle;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
import org.apache.activemq.artemis.core.server.ServerProducer;
import org.apache.activemq.artemis.core.server.ServerSession;
import org.apache.activemq.artemis.core.server.impl.ServerProducerImpl;
import org.apache.activemq.artemis.core.version.Version;
import org.apache.activemq.artemis.logs.AuditLogger;
import org.slf4j.Logger;
@ -183,9 +181,8 @@ public class ActiveMQPacketHandler implements ChannelHandler {
Map<SimpleString, RoutingType> routingTypeMap = protocolManager.getPrefixes();
CoreSessionCallback sessionCallback = new CoreSessionCallback(request.getName(), protocolManager, channel, connection);
ServerSession session = server.createSession(request.getName(), activeMQPrincipal == null ? request.getUsername() : activeMQPrincipal.getUserName(), activeMQPrincipal == null ? request.getPassword() : activeMQPrincipal.getPassword(), request.getMinLargeMessageSize(), connection, request.isAutoCommitSends(), request.isAutoCommitAcks(), request.isPreAcknowledge(), request.isXA(), request.getDefaultAddress(), sessionCallback, true, sessionOperationContext, routingTypeMap, protocolManager.getSecurityDomain(), validatedUser);
ServerProducer serverProducer = new ServerProducerImpl(session.getName(), "CORE", request.getDefaultAddress());
session.addProducer(serverProducer);
boolean isLegacyProducer = request.getVersion() < PacketImpl.ARTEMIS_2_28_0_VERSION;
ServerSession session = server.createSession(request.getName(), activeMQPrincipal == null ? request.getUsername() : activeMQPrincipal.getUserName(), activeMQPrincipal == null ? request.getPassword() : activeMQPrincipal.getPassword(), request.getMinLargeMessageSize(), connection, request.isAutoCommitSends(), request.isAutoCommitAcks(), request.isPreAcknowledge(), request.isXA(), request.getDefaultAddress(), sessionCallback, true, sessionOperationContext, routingTypeMap, protocolManager.getSecurityDomain(), validatedUser, isLegacyProducer);
ServerSessionPacketHandler handler = new ServerSessionPacketHandler(server, session, channel);
channel.setHandler(handler);
sessionCallback.setSessionHandler(handler);

7
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java
View File

@ -349,7 +349,9 @@ public interface ActiveMQServer extends ServiceComponent {
boolean autoCreateQueues,
OperationContext context,
Map<SimpleString, RoutingType> prefixes,
String securityDomain, String validatedUser) throws Exception;
String securityDomain,
String validatedUser,
boolean isLegacyProducer) throws Exception;
/** This is to be used in places where security is bypassed, like internal sessions, broker connections, etc... */
ServerSession createInternalSession(String name,
@ -364,7 +366,8 @@ public interface ActiveMQServer extends ServiceComponent {
boolean autoCreateQueues,
OperationContext context,
Map<SimpleString, RoutingType> prefixes,
String securityDomain) throws Exception;
String securityDomain,
boolean isLegacyProducer) throws Exception;
/** should the server rebuild page counters upon startup.
* this will be useful on testing or an embedded broker scenario */

4
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
View File

@ -1220,7 +1220,11 @@ public interface ActiveMQServerLogger {
@LogMessage(id = 222308, value = "Unable to listen for incoming fail-back request because {} is null. Ensure the broker has the proper cluster-connection configuration.", level = LogMessage.Level.WARN)
void failBackCheckerFailure(String component);
@LogMessage(id = 222309, value = "Trying to remove a producer with ID {} that doesnt exist from session {} on Connection {}.", level = LogMessage.Level.WARN)
void producerDoesNotExist(int id, String session, String remoteAddress);
@LogMessage(id = 222310, value = "Trying to add a producer with ID {} that already exists to session {} on Connection {}.", level = LogMessage.Level.WARN)
void producerAlreadyExists(int id, String session, String remoteAddress);
@LogMessage(id = 224000, value = "Failure in initialisation", level = LogMessage.Level.ERROR)
void initializationError(Throwable e);

47
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ConsumerInfo.java
View File

@ -62,4 +62,51 @@ public interface ConsumerInfo {
*/
String getConnectionRemoteAddress();
/**
* Returns how many messages are out for delivery but not yet acknowledged
* @return delivering count
*/
int getMessagesInTransit();
/**
* Returns the combined size of all the messages out for delivery but not yet acknowledged
* @return the total size of all the messages
*/
long getMessagesInTransitSize();
/**
* Returns The total number of messages sent to a consumer including redeliveries that have been acknowledged
* @return the total number of messages delivered.
*/
long getMessagesDelivered();
/**
* Returns the total size of all the messages delivered to the consumer. This includes redelivered messages
* @return The total size of all the messages
*/
long getMessagesDeliveredSize();
/**
* Returns the number of messages acknowledged by this consumer since it was created
* @return messages acknowledged
*/
long getMessagesAcknowledged();
/**
* Returns the number of acknowledged messages that are awaiting commit in a transaction
* @return th eno acknowledgements awaiting commit
*/
int getMessagesAcknowledgedAwaitingCommit();
/**
* Returns the time in milliseconds that the last message was delivered to a consumer
* @return the time of the last message delivered
*/
long getLastDeliveredTime();
/**
* Returns the time in milliseconds that the last message was acknowledged by a consumer
* @return the time of the last message was acknowledged
*/
long getLastAcknowledgedTime();
}

7
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ServerConsumer.java
View File

@ -110,4 +110,11 @@ public interface ServerConsumer extends Consumer, ConsumerInfo {
long getCreationTime();
String getSessionID();
/**
* This is needed when some protocols (OW) handle the acks themselves and need to update the metrics
* @param ref the message reference
* @param transaction the tx
*/
void metricsAcknowledge(MessageReference ref, Transaction transaction);
}

16
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ServerProducer.java
View File

@ -17,7 +17,12 @@
package org.apache.activemq.artemis.core.server;
public interface ServerProducer {
String getAddress();
String ANONYMOUS = "ANONYMOUS";
long getID();
String getName();
String getAddress();;
String getProtocol();
@ -29,7 +34,14 @@ public interface ServerProducer {
String getConnectionID();
String getID();
long getCreationTime();
Object getLastProducedMessageID();
long getMessagesSent();
long getMessagesSentSize();
void updateMetrics(Object lastProducedMessageID, int encodeSize);
}

16
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ServerSession.java
View File

@ -18,6 +18,7 @@ package org.apache.activemq.artemis.core.server;
import org.apache.activemq.artemis.json.JsonArrayBuilder;
import javax.transaction.xa.Xid;
import java.util.Collection;
import java.util.EnumSet;
import java.util.List;
import java.util.Map;
@ -332,11 +333,13 @@ public interface ServerSession extends SecurityAuth {
RoutingStatus send(Transaction tx,
Message message,
boolean direct,
String senderName,
boolean noAutoCreateQueue) throws Exception;
RoutingStatus send(Transaction tx,
Message message,
boolean direct,
String senderName,
boolean noAutoCreateQueue,
RoutingContext routingContext) throws Exception;
@ -345,18 +348,20 @@ public interface ServerSession extends SecurityAuth {
Message msg,
SimpleString originalAddress,
boolean direct,
String senderName,
boolean noAutoCreateQueue) throws Exception;
RoutingStatus doSend(Transaction tx,
Message msg,
SimpleString originalAddress,
boolean direct,
String senderName,
boolean noAutoCreateQueue,
RoutingContext routingContext) throws Exception;
RoutingStatus send(Message message, boolean direct, boolean noAutoCreateQueue) throws Exception;
RoutingStatus send(Message message, boolean direct, String senderName, boolean noAutoCreateQueue) throws Exception;
RoutingStatus send(Message message, boolean direct) throws Exception;
RoutingStatus send(Message message, boolean direct, String senderName) throws Exception;
void forceConsumerDelivery(long consumerID, long sequence) throws Exception;
@ -376,8 +381,6 @@ public interface ServerSession extends SecurityAuth {
Map<String, String> getMetaData();
String[] getTargetAddresses();
/**
* Add all the producers detail to the JSONArray object.
* This is a method to be used by the management layer.
@ -387,7 +390,6 @@ public interface ServerSession extends SecurityAuth {
*/
void describeProducersInfo(JsonArrayBuilder objs) throws Exception;
String getLastSentMessageID(String address);
long getCreationTime();
@ -522,11 +524,11 @@ public interface ServerSession extends SecurityAuth {
Pair<SimpleString, EnumSet<RoutingType>> getAddressAndRoutingTypes(SimpleString address,
EnumSet<RoutingType> defaultRoutingTypes);
void addProducer(ServerProducer serverProducer);
void addProducer(String name, String protocol, String address);
void removeProducer(String ID);
Map<String, ServerProducer> getServerProducers();
Collection<ServerProducer> getServerProducers();
String getDefaultAddress();

15
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
View File

@ -1746,7 +1746,8 @@ public class ActiveMQServerImpl implements ActiveMQServer {
final OperationContext context,
final Map<SimpleString, RoutingType> prefixes,
final String securityDomain,
String validatedUser) throws Exception {
String validatedUser,
boolean isLegacyProducer) throws Exception {
if (validatedUser == null) {
validatedUser = validateUser(username, password, connection, securityDomain);
}
@ -1758,7 +1759,7 @@ public class ActiveMQServerImpl implements ActiveMQServer {
autoCommitAcks, preAcknowledge, xa, defaultAddress, callback, autoCreateQueues, prefixes);
}
final ServerSessionImpl session = internalCreateSession(name, username, password, validatedUser, minLargeMessageSize, connection, autoCommitSends, autoCommitAcks, preAcknowledge, xa, defaultAddress, callback, context, autoCreateQueues, prefixes, securityDomain);
final ServerSessionImpl session = internalCreateSession(name, username, password, validatedUser, minLargeMessageSize, connection, autoCommitSends, autoCommitAcks, preAcknowledge, xa, defaultAddress, callback, context, autoCreateQueues, prefixes, securityDomain, isLegacyProducer);
return session;
}
@ -1786,8 +1787,9 @@ public class ActiveMQServerImpl implements ActiveMQServer {
boolean autoCreateQueues,
OperationContext context,
Map<SimpleString, RoutingType> prefixes,
String securityDomain) throws Exception {
ServerSessionImpl session = internalCreateSession(name, null, null, null, minLargeMessageSize, connection, autoCommitSends, autoCommitAcks, preAcknowledge, xa, defaultAddress, callback, context, autoCreateQueues, prefixes, securityDomain);
String securityDomain,
boolean isLegacyProducer) throws Exception {
ServerSessionImpl session = internalCreateSession(name, null, null, null, minLargeMessageSize, connection, autoCommitSends, autoCommitAcks, preAcknowledge, xa, defaultAddress, callback, context, autoCreateQueues, prefixes, securityDomain, isLegacyProducer);
session.disableSecurity();
return session;
}
@ -1864,14 +1866,15 @@ public class ActiveMQServerImpl implements ActiveMQServer {
OperationContext context,
boolean autoCreateQueues,
Map<SimpleString, RoutingType> prefixes,
String securityDomain) throws Exception {
String securityDomain,
boolean isLegacyProducer) throws Exception {
if (hasBrokerSessionPlugins()) {
callBrokerSessionPlugins(plugin -> plugin.beforeCreateSession(name, username, minLargeMessageSize, connection,
autoCommitSends, autoCommitAcks, preAcknowledge, xa, defaultAddress, callback, autoCreateQueues, context, prefixes));
}
ServerSessionImpl session = new ServerSessionImpl(name, username, password, validatedUser, minLargeMessageSize, autoCommitSends, autoCommitAcks, preAcknowledge, configuration.isPersistDeliveryCountBeforeDelivery(), xa, connection, storageManager, postOffice, resourceManager, securityStore, managementService, this, configuration.getManagementAddress(), defaultAddress == null ? null : new SimpleString(defaultAddress), callback, context, pagingManager, prefixes, securityDomain);
ServerSessionImpl session = new ServerSessionImpl(name, username, password, validatedUser, minLargeMessageSize, autoCommitSends, autoCommitAcks, preAcknowledge, configuration.isPersistDeliveryCountBeforeDelivery(), xa, connection, storageManager, postOffice, resourceManager, securityStore, managementService, this, configuration.getManagementAddress(), defaultAddress == null ? null : new SimpleString(defaultAddress), callback, context, pagingManager, prefixes, securityDomain, isLegacyProducer);
sessions.put(name, session);

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

@ -26,7 +26,9 @@ import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import java.util.function.Function;
import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
@ -61,6 +63,8 @@ import org.apache.activemq.artemis.core.server.SlowConsumerDetectionListener;
import org.apache.activemq.artemis.core.server.management.ManagementService;
import org.apache.activemq.artemis.core.server.management.Notification;
import org.apache.activemq.artemis.core.transaction.Transaction;
import org.apache.activemq.artemis.core.transaction.TransactionOperationAbstract;
import org.apache.activemq.artemis.core.transaction.TransactionPropertyIndexes;
import org.apache.activemq.artemis.core.transaction.impl.TransactionImpl;
import org.apache.activemq.artemis.logs.AuditLogger;
import org.apache.activemq.artemis.spi.core.protocol.SessionCallback;
@ -153,14 +157,14 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
private AtomicLong messageConsumedSnapshot = new AtomicLong(0);
private long acks;
private boolean requiresLegacyPrefix = false;
private boolean anycast = false;
private boolean isClosed = false;
ServerConsumerMetrics metrics = new ServerConsumerMetrics();
public ServerConsumerImpl(final long id,
final ServerSession session,
@ -355,6 +359,11 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
return this.session.getName();
}
@Override
public void metricsAcknowledge(MessageReference ref, Transaction transaction) {
metrics.addAcknowledge(ref.getMessage().getEncodeSize(), transaction);
}
@Override
public List<MessageReference> getDeliveringMessages() {
synchronized (lock) {
@ -441,6 +450,8 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
deliveringRefs.add(ref);
}
metrics.addMessage(ref.getMessage().getEncodeSize());
ref.handled();
ref.setConsumerId(this.id);
@ -466,7 +477,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
if (preAcknowledge) {
// With pre-ack, we ack *before* sending to the client
ref.getQueue().acknowledge(ref, this);
acks++;
metrics.addAcknowledge(ref.getMessage().getEncodeSize(), null);
}
}
@ -704,9 +715,9 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
final List<MessageReference> refs = new ArrayList<>(deliveringRefs.size());
MessageReference ref;
while ((ref = deliveringRefs.poll()) != null) {
metrics.addAcknowledge(ref.getMessage().getEncodeSize(), tx);
if (performACK) {
ref.acknowledge(tx, this);
performACK = false;
} else {
refs.add(ref);
@ -911,8 +922,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
ref.acknowledge(tx, this);
ackedRefs.add(ref.getMessageID());
acks++;
metrics.addAcknowledge(ref.getMessage().getEncodeSize(), tx);
}
while (ref.getMessageID() != messageID);
@ -973,11 +983,9 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
tx.markAsRollbackOnly(ils);
throw ils;
}
metrics.addAcknowledge(ref.getMessage().getEncodeSize(), tx);
ref.acknowledge(tx, this);
acks++;
if (startedTransaction) {
tx.commit();
}
@ -1016,7 +1024,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
if (!failed) {
ref.decrementDeliveryCount();
}
metrics.addAcknowledge(ref.getMessage().getEncodeSize(), null);
ref.getQueue().cancel(ref, System.currentTimeMillis());
}
@ -1032,7 +1040,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
if (ref == null) {
return; // nothing to be done
}
metrics.addAcknowledge(ref.getMessage().getEncodeSize(), null);
ref.getQueue().sendToDeadLetterAddress(null, ref);
}
@ -1040,6 +1048,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
public synchronized void backToDelivering(MessageReference reference) {
synchronized (lock) {
deliveringRefs.addFirst(reference);
metrics.addMessage(reference.getMessage().getEncodeSize());
}
}
@ -1059,10 +1068,12 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
}
if (deliveringRefs.peek().getMessage().getMessageID() == messageID) {
return deliveringRefs.poll();
MessageReference ref = deliveringRefs.poll();
return ref;
}
//slow path in a separate method
return removeDeliveringRefById(messageID);
MessageReference ref = removeDeliveringRefById(messageID);
return ref;
}
}
@ -1114,6 +1125,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
public float getRate() {
float timeSlice = ((System.currentTimeMillis() - consumerRateCheckTime.getAndSet(System.currentTimeMillis())) / 1000.0f);
long acks = metrics.getMessagesAcknowledged();
if (timeSlice == 0) {
messageConsumedSnapshot.getAndSet(acks);
return 0.0f;
@ -1513,7 +1525,143 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
return this.session.getRemotingConnection().getTransportConnection().getRemoteAddress();
}
@Override
public long getMessagesInTransitSize() {
return metrics.getMessagesInTransitSize();
}
@Override
public int getMessagesInTransit() {
return deliveringRefs.size();
}
@Override
public long getLastDeliveredTime() {
return metrics.getLastDeliveredTime();
}
@Override
public long getLastAcknowledgedTime() {
return metrics.getLastAcknowledgedTime();
}
@Override
public long getMessagesAcknowledged() {
return metrics.getMessagesAcknowledged();
}
@Override
public long getMessagesDeliveredSize() {
return metrics.getMessagesDeliveredSize();
}
@Override
public long getMessagesDelivered() {
return metrics.getMessagesDelivered();
}
@Override
public int getMessagesAcknowledgedAwaitingCommit() {
return metrics.getMessagesAcknowledgedAwaitingCommit();
}
public SessionCallback getCallback() {
return callback;
}
static class ServerConsumerMetrics extends TransactionOperationAbstract {
/**
* Since messages can be delivered (incremented) and acknowledged (decremented) at the same time we have to protect
* the encode size and make it atomic. The other fields are ok since they are only accessed from a single thread.
*/
private static final AtomicLongFieldUpdater<ServerConsumerMetrics> messagesInTransitSizeUpdater = AtomicLongFieldUpdater.newUpdater(ServerConsumerMetrics.class, "messagesInTransitSize");
private volatile long messagesInTransitSize = 0;
private static final AtomicIntegerFieldUpdater<ServerConsumerMetrics> messagesAcknowledgedAwaitingCommitUpdater = AtomicIntegerFieldUpdater.newUpdater(ServerConsumerMetrics.class, "messagesAcknowledgedAwaitingCommit");
private volatile int messagesAcknowledgedAwaitingCommit = 0;
private static final AtomicLongFieldUpdater<ServerConsumerMetrics>messagesDeliveredSizeUpdater = AtomicLongFieldUpdater.newUpdater(ServerConsumerMetrics.class, "messagesDeliveredSize");
private volatile long messagesDeliveredSize = 0;
private volatile long lastDeliveredTime = 0;
private volatile long lastAcknowledgedTime = 0;
private static final AtomicLongFieldUpdater<ServerConsumerMetrics>messagesDeliveredUpdater = AtomicLongFieldUpdater.newUpdater(ServerConsumerMetrics.class, "messagesDelivered");
private volatile long messagesDelivered = 0;
private static final AtomicLongFieldUpdater<ServerConsumerMetrics> messagesAcknowledgedUpdater = AtomicLongFieldUpdater.newUpdater(ServerConsumerMetrics.class, "messagesAcknowledged");
private volatile long messagesAcknowledged = 0;
public long getMessagesInTransitSize() {
return messagesInTransitSizeUpdater.get(this);
}
public long getMessagesDeliveredSize() {
return messagesDeliveredSizeUpdater.get(this);
}
public long getLastDeliveredTime() {
return lastDeliveredTime;
}
public long getLastAcknowledgedTime() {
return lastAcknowledgedTime;
}
public long getMessagesDelivered() {
return messagesDeliveredUpdater.get(this);
}
public long getMessagesAcknowledged() {
return messagesAcknowledgedUpdater.get(this);
}
public int getMessagesAcknowledgedAwaitingCommit() {
return messagesAcknowledgedAwaitingCommitUpdater.get(this);
}
public void addMessage(int encodeSize) {
messagesInTransitSizeUpdater.addAndGet(this, encodeSize);
messagesDeliveredSizeUpdater.addAndGet(this, encodeSize);
messagesDeliveredUpdater.addAndGet(this, 1);
lastDeliveredTime = System.currentTimeMillis();
}
public void addAcknowledge(int encodeSize, Transaction tx) {
messagesInTransitSizeUpdater.addAndGet(this, -encodeSize);
messagesAcknowledgedUpdater.addAndGet(this, 1);
lastAcknowledgedTime = System.currentTimeMillis();
if (tx != null) {
addOperation(tx);
messagesAcknowledgedAwaitingCommitUpdater.addAndGet(this, 1);
}
}
@Override
public void afterCommit(Transaction tx) {
messagesAcknowledgedAwaitingCommitUpdater.set(this, 0);
}
@Override
public void afterRollback(Transaction tx) {
messagesAcknowledgedAwaitingCommitUpdater.set(this, 0);
}
public void addOperation(Transaction tx) {
Object property = tx.getProperty(TransactionPropertyIndexes.CONSUMER_METRICS_OPERATION);
if (property == null) {
tx.putProperty(TransactionPropertyIndexes.CONSUMER_METRICS_OPERATION, this);
tx.addOperation(this);
}
}
}
}

83
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerLegacyProducersImpl.java Normal file
View File

@ -0,0 +1,83 @@
/*
* 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.server.impl;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.core.server.ServerProducer;
import org.apache.activemq.artemis.core.server.ServerSession;
import org.apache.activemq.artemis.utils.collections.MaxSizeMap;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
public class ServerLegacyProducersImpl implements ServerProducers {
private static final int MAX_PRODUCER_METRIC_SIZE = 100;
protected final Map<String, ServerProducer> producers = Collections.synchronizedMap(new MaxSizeMap<>(MAX_PRODUCER_METRIC_SIZE));
private final String sessionName;
private final String connectionID;
public ServerLegacyProducersImpl(ServerSession session) {
this.sessionName = session.getName();
this.connectionID = session.getConnectionID() != null ? session.getConnectionID().toString() : null;
}
@Override
public Map<String, ServerProducer> cloneProducers() {
return new HashMap<>(producers);
}
@Override
public Collection<ServerProducer> getServerProducers() {
return new ArrayList<>(producers.values());
}
@Override
public ServerProducer getServerProducer(String senderName, Message msg, ServerSessionImpl serverSession) {
String address = msg.getAddress();
String name = sessionName + ":" + address;
ServerProducer producer = producers.get(name);
if (producer == null) {
producer = new ServerProducerImpl(name, "CORE", address);
producer.setSessionID(sessionName);
producer.setConnectionID(connectionID);
producers.put(name, producer);
}
return producer;
}
@Override
public void put(String id, ServerProducer producer) {
//never called as we track by address the old way
}
@Override
public void remove(String id) {
//never called as we track by address the old way
}
@Override
public void clear() {
producers.clear();
}
}

61
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerProducerImpl.java
View File

@ -18,23 +18,50 @@ package org.apache.activemq.artemis.core.server.impl;
import org.apache.activemq.artemis.core.server.ServerProducer;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
public class ServerProducerImpl implements ServerProducer {
private final String ID;
private static final AtomicLong PRODUCER_ID_GENERATOR = new AtomicLong();
private final long ID;
private final String name;
private final String protocol;
private final long creationTime;
private volatile long messagesSent = 0;
private volatile long messagesSentSize = 0;
private static final AtomicLongFieldUpdater<ServerProducerImpl> messagesSentUpdater = AtomicLongFieldUpdater.newUpdater(ServerProducerImpl.class, "messagesSent");
private static final AtomicLongFieldUpdater<ServerProducerImpl> messagesSentSizeUpdater = AtomicLongFieldUpdater.newUpdater(ServerProducerImpl.class, "messagesSentSize");
private final String address;
private volatile Object lastProducedMessageID;
private String sessionID;
private String connectionID;
public ServerProducerImpl(String ID, String protocol, String address) {
this.ID = ID;
public ServerProducerImpl(String name, String protocol, String address) {
this.ID = PRODUCER_ID_GENERATOR.incrementAndGet();
this.name = name;
this.protocol = protocol;
this.address = address;
this.creationTime = System.currentTimeMillis();
}
@Override
public long getID() {
return ID;
}
@Override
public String getName() {
return name;
}
@Override
public String getAddress() {
return address;
@ -52,7 +79,6 @@ public class ServerProducerImpl implements ServerProducer {
@Override
public void setConnectionID(String connectionID) {
this.connectionID = connectionID;
}
@ -66,13 +92,30 @@ public class ServerProducerImpl implements ServerProducer {
return connectionID;
}
@Override
public String getID() {
return ID;
}
@Override
public long getCreationTime() {
return creationTime;
}
@Override
public void updateMetrics(Object lastProducedMessageID, int encodeSize) {
messagesSentUpdater.addAndGet(this, 1);
messagesSentSizeUpdater.getAndAdd(this, encodeSize);
this.lastProducedMessageID = lastProducedMessageID;
}
@Override
public Object getLastProducedMessageID() {
return lastProducedMessageID;
}
@Override
public long getMessagesSent() {
return messagesSent;
}
@Override
public long getMessagesSentSize() {
return messagesSentSize;
}
}

37
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerProducers.java Normal file
View File

@ -0,0 +1,37 @@
/*
* 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.server.impl;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.core.server.ServerProducer;
import java.util.Collection;
import java.util.Map;
public interface ServerProducers {
Map<String, ServerProducer> cloneProducers();
Collection<ServerProducer> getServerProducers();
ServerProducer getServerProducer(String senderName, Message msg, ServerSessionImpl serverSession);
void put(String id, ServerProducer producer);
void remove(String id);
void clear();
}

63
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerProducersImpl.java Normal file
View File

@ -0,0 +1,63 @@
/*
* 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.server.impl;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.core.server.ServerProducer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
public class ServerProducersImpl implements ServerProducers {
protected final Map<String, ServerProducer> producers = new ConcurrentHashMap<>();
@Override
public Map<String, ServerProducer> cloneProducers() {
return new HashMap<>(producers);
}
@Override
public Collection<ServerProducer> getServerProducers() {
return new ArrayList<>(producers.values());
}
@Override
public ServerProducer getServerProducer(String senderName, Message msg, ServerSessionImpl serverSession) {
if (senderName != null) {
return producers.get(senderName);
}
return null;
}
@Override
public void put(String id, ServerProducer producer) {
producers.put(id, producer);
}
@Override
public void remove(String id) {
producers.remove(id);
}
@Override
public void clear() {
producers.clear();
}
}

125
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
View File

@ -16,6 +16,7 @@
*/
package org.apache.activemq.artemis.core.server.impl;
import org.apache.activemq.artemis.core.management.impl.view.ProducerField;
import org.apache.activemq.artemis.api.core.ActiveMQAddressExistsException;
import org.apache.activemq.artemis.api.core.ActiveMQQueueExistsException;
import org.apache.activemq.artemis.core.postoffice.Bindings;
@ -25,17 +26,16 @@ import java.security.cert.X509Certificate;
import javax.transaction.xa.XAException;
import javax.transaction.xa.Xid;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.activemq.artemis.Closeable;
import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
@ -102,7 +102,6 @@ import org.apache.activemq.artemis.utils.ByteUtil;
import org.apache.activemq.artemis.utils.CompositeAddress;
import org.apache.activemq.artemis.utils.JsonLoader;
import org.apache.activemq.artemis.utils.PrefixUtil;
import org.apache.activemq.artemis.utils.collections.MaxSizeMap;
import org.apache.activemq.artemis.utils.collections.TypedProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -115,7 +114,6 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private boolean securityEnabled = true;
private final String securityDomain;
@ -140,7 +138,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
protected final Map<Long, ServerConsumer> consumers = new ConcurrentHashMap<>();
protected final Map<String, ServerProducer> producers = new ConcurrentHashMap<>();
protected final ServerProducers serverProducers;
protected Transaction tx;
@ -184,9 +182,6 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
private final OperationContext context;
// Session's usage should be by definition single threaded, hence it's not needed to use a concurrentHashMap here
protected final Map<SimpleString, Pair<Object, AtomicLong>> targetAddressInfos = new MaxSizeMap<>(100);
private final long creationTime = System.currentTimeMillis();
// to prevent session from being closed twice.
@ -228,7 +223,8 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
final OperationContext context,
final PagingManager pagingManager,
final Map<SimpleString, RoutingType> prefixes,
final String securityDomain) throws Exception {
final String securityDomain,
boolean isLegacyProducer) throws Exception {
this.username = username;
this.password = password;
@ -289,6 +285,11 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
sendSessionNotification(CoreNotificationType.SESSION_CREATED);
this.securityDomain = securityDomain;
if (isLegacyProducer) {
serverProducers = new ServerLegacyProducersImpl(this);
} else {
serverProducers = new ServerProducersImpl();
}
}
// ServerSession implementation ---------------------------------------------------------------------------
@ -450,7 +451,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
}
consumers.clear();
producers.clear();
serverProducers.clear();
if (closeables != null) {
for (Closeable closeable : closeables) {
@ -1186,10 +1187,6 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
remotingConnection.removeFailureListener(cleaner);
}
if (server.getAddressInfo(unPrefixedQueueName) == null) {
targetAddressInfos.remove(queueToDelete);
}
}
@Override
@ -1790,29 +1787,32 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
@Override
public RoutingStatus send(final Message message, final boolean direct) throws Exception {
return send(message, direct, false);
public RoutingStatus send(final Message message, final boolean direct, final String senderName) throws Exception {
return send(message, direct, senderName,false);
}
@Override
public RoutingStatus send(final Message message,
final boolean direct,
final String senderName,
boolean noAutoCreateQueue) throws Exception {
return send(getCurrentTransaction(), message, direct, noAutoCreateQueue);
return send(getCurrentTransaction(), message, direct, senderName, noAutoCreateQueue);
}
@Override
public synchronized RoutingStatus send(Transaction tx,
Message msg,
final boolean direct,
final String senderName,
boolean noAutoCreateQueue) throws Exception {
return send(tx, msg, direct, noAutoCreateQueue, routingContext);
return send(tx, msg, direct, senderName, noAutoCreateQueue, routingContext);
}
@Override
public synchronized RoutingStatus send(Transaction tx,
Message messageParameter,
final boolean direct,
final String senderName,
boolean noAutoCreateQueue,
RoutingContext routingContext) throws Exception {
final Message message = LargeServerMessageImpl.checkLargeMessage(messageParameter, storageManager);
@ -1865,7 +1865,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
result = handleManagementMessage(tx, message, direct);
} else {
result = doSend(tx, message, address, direct, noAutoCreateQueue, routingContext);
result = doSend(tx, message, address, direct, senderName, noAutoCreateQueue, routingContext);
}
if (AuditLogger.isMessageLoggingEnabled()) {
@ -1975,30 +1975,6 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
return metaData;
}
@Override
public String[] getTargetAddresses() {
Map<SimpleString, Pair<Object, AtomicLong>> copy = cloneTargetAddresses();
Iterator<SimpleString> iter = copy.keySet().iterator();
int num = copy.keySet().size();
String[] addresses = new String[num];
int i = 0;
while (iter.hasNext()) {
addresses[i] = iter.next().toString();
i++;
}
return addresses;
}
@Override
public String getLastSentMessageID(String address) {
Pair<Object, AtomicLong> value = targetAddressInfos.get(SimpleString.toSimpleString(address));
if (value != null) {
return value.getA().toString();
} else {
return null;
}
}
@Override
public long getCreationTime() {
return this.creationTime;
@ -2010,14 +1986,22 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
@Override
public void describeProducersInfo(JsonArrayBuilder array) throws Exception {
Map<SimpleString, Pair<Object, AtomicLong>> targetCopy = cloneTargetAddresses();
for (Map.Entry<SimpleString, Pair<Object, AtomicLong>> entry : targetCopy.entrySet()) {
Map<String, ServerProducer> targetCopy = cloneProducers();
for (Map.Entry<String, ServerProducer> entry : targetCopy.entrySet()) {
String uuid = null;
if (entry.getValue().getA() != null) {
uuid = entry.getValue().getA().toString();
if (entry.getValue().getLastProducedMessageID() != null) {
uuid = entry.getValue().getLastProducedMessageID().toString();
}
JsonObjectBuilder producerInfo = JsonLoader.createObjectBuilder().add("connectionID", this.getConnectionID().toString()).add("sessionID", this.getName()).add("destination", entry.getKey().toString()).add("lastUUIDSent", uuid, JsonValue.NULL).add("msgSent", entry.getValue().getB().longValue());
JsonObjectBuilder producerInfo = JsonLoader.createObjectBuilder()
.add(ProducerField.ID.getName(), String.valueOf(entry.getValue().getID()))
.add(ProducerField.NAME.getName(), entry.getValue().getName())
.add(ProducerField.CONNECTION_ID.getName(), this.getConnectionID().toString())
.add(ProducerField.SESSION.getAlternativeName(), this.getName())
.add(ProducerField.CREATION_TIME.getName(), String.valueOf(entry.getValue().getCreationTime()))
.add(ProducerField.ADDRESS.getAlternativeName(), entry.getValue().getAddress())
.add(ProducerField.LAST_PRODUCED_MESSAGE_ID.getName(), uuid, JsonValue.NULL)
.add(ProducerField.MESSAGE_SENT.getName(), entry.getValue().getMessagesSent())
.add(ProducerField.MESSAGE_SENT_SIZE.getName(), entry.getValue().getMessagesSentSize());
array.add(producerInfo);
}
}
@ -2098,8 +2082,8 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
connectionFailed(me, failedOver);
}
public Map<SimpleString, Pair<Object, AtomicLong>> cloneTargetAddresses() {
return new HashMap<>(targetAddressInfos);
public Map<String, ServerProducer> cloneProducers() {
return serverProducers.cloneProducers();
}
private void setStarted(final boolean s) {
@ -2140,7 +2124,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
}
reply.setAddress(replyTo);
doSend(tx, reply, null, direct, false, routingContext);
doSend(tx, reply, null, direct, null, false, routingContext);
}
return RoutingStatus.OK;
}
@ -2202,8 +2186,9 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
final Message msg,
final SimpleString originalAddress,
final boolean direct,
final String senderName,
final boolean noAutoCreateQueue) throws Exception {
return doSend(tx, msg, originalAddress, direct, noAutoCreateQueue, routingContext);
return doSend(tx, msg, originalAddress, direct, senderName, noAutoCreateQueue, routingContext);
}
@ -2212,6 +2197,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
final Message msg,
final SimpleString originalAddress,
final boolean direct,
final String senderName,
final boolean noAutoCreateQueue,
final RoutingContext routingContext) throws Exception {
@ -2272,14 +2258,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
logger.debug("Routing result for {} = {}", msg, result);
Pair<Object, AtomicLong> value = targetAddressInfos.get(msg.getAddressSimpleString());
if (value == null) {
targetAddressInfos.put(msg.getAddressSimpleString(), new Pair<>(msg.getUserID(), new AtomicLong(1)));
} else {
value.setA(msg.getUserID());
value.getB().incrementAndGet();
}
updateProducerMetrics(msg, senderName);
} finally {
if (!routingContext.isReusable()) {
routingContext.clear();
@ -2395,20 +2374,21 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
}
@Override
public void addProducer(ServerProducer serverProducer) {
serverProducer.setSessionID(getName());
serverProducer.setConnectionID(getConnectionID().toString());
producers.put(serverProducer.getID(), serverProducer);
public void addProducer(String name, String protocol, String address) {
ServerProducer producer = new ServerProducerImpl(name, protocol, address != null ? address : ServerProducer.ANONYMOUS);
producer.setSessionID(getName());
producer.setConnectionID(getConnectionID() != null ? getConnectionID().toString() : null);
serverProducers.put(name, producer);
}
@Override
public void removeProducer(String ID) {
producers.remove(ID);
serverProducers.remove(ID);
}
@Override
public Map<String, ServerProducer> getServerProducers() {
return Collections.unmodifiableMap(new HashMap(producers));
public Collection<ServerProducer> getServerProducers() {
return serverProducers.getServerProducers();
}
@Override
@ -2436,4 +2416,13 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
public String toManagementString() {
return "ServerSession [id=" + getConnectionID() + ":" + getName() + "]";
}
private void updateProducerMetrics(Message msg, String senderName) {
ServerProducer serverProducer = serverProducers.getServerProducer(senderName, msg, this);
if (serverProducer != null) {
serverProducer.updateMetrics(msg.getUserID(), msg instanceof LargeServerMessageImpl ? ((LargeServerMessageImpl)msg).getBodyBufferSize() : msg.getEncodeSize());
}
}
}

2
artemis-server/src/main/java/org/apache/activemq/artemis/core/transaction/TransactionPropertyIndexes.java
View File

@ -35,4 +35,6 @@ public class TransactionPropertyIndexes {
public static final int PAGE_CURSOR_POSITIONS = 8;
public static final int EXPIRY_LOGGER = 9;
public static final int CONSUMER_METRICS_OPERATION = 10;
}

2
artemis-server/src/main/java/org/apache/activemq/artemis/core/transaction/impl/TransactionImpl.java
View File

@ -48,7 +48,7 @@ public class TransactionImpl implements Transaction {
private List<TransactionOperation> storeOperations;
private static final int INITIAL_NUM_PROPERTIES = 10;
private static final int INITIAL_NUM_PROPERTIES = 11;
private Object[] properties = null;

11
artemis-server/src/test/java/org/apache/activemq/artemis/tests/util/ActiveMQTestBase.java
View File

@ -65,6 +65,7 @@ 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.api.core.ICoreMessage;
import org.apache.activemq.artemis.api.core.JsonUtil;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.Pair;
import org.apache.activemq.artemis.api.core.QueueConfiguration;
@ -136,6 +137,7 @@ import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.core.transaction.impl.XidImpl;
import org.apache.activemq.artemis.jdbc.store.drivers.JDBCUtils;
import org.apache.activemq.artemis.jdbc.store.sql.SQLProvider;
import org.apache.activemq.artemis.json.JsonObject;
import org.apache.activemq.artemis.nativo.jlibaio.LibaioContext;
import org.apache.activemq.artemis.spi.core.security.ActiveMQJAASSecurityManager;
import org.apache.activemq.artemis.spi.core.security.ActiveMQSecurityManager;
@ -2627,6 +2629,15 @@ public abstract class ActiveMQTestBase extends Assert {
return !hadToInterrupt;
}
public String createJsonFilter(String fieldName, String operationName, String value) {
HashMap<String, Object> filterMap = new HashMap<>();
filterMap.put("field", fieldName);
filterMap.put("operation", operationName);
filterMap.put("value", value);
JsonObject jsonFilterObject = JsonUtil.toJsonObject(filterMap);
return jsonFilterObject.toString();
}
protected static ReplicationEndpoint getReplicationEndpoint(ActiveMQServer server) {
final Activation activation = server.getActivation();
if (activation instanceof SharedNothingBackupActivation) {

2
pom.xml
View File

@ -182,7 +182,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>133,132,131,130,129,128,127,126,125,124,123,122</activemq.version.incrementingVersion>
<activemq.version.incrementingVersion>134,133,132,131,130,129,128,127,126,125,124,123,122</activemq.version.incrementingVersion>
<activemq.version.versionTag>${project.version}</activemq.version.versionTag>
<ActiveMQ-Version>${project.version}(${activemq.version.incrementingVersion})</ActiveMQ-Version>

45
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cli/DummyServerConsumer.java
View File

@ -180,6 +180,26 @@ public class DummyServerConsumer implements ServerConsumer {
return null;
}
@Override
public void metricsAcknowledge(MessageReference ref, Transaction transaction) {
}
@Override
public long getMessagesInTransitSize() {
return 0;
}
@Override
public long getMessagesDeliveredSize() {
return 0;
}
@Override
public long getMessagesDelivered() {
return 0;
}
@Override
public void promptDelivery() {
@ -278,4 +298,29 @@ public class DummyServerConsumer implements ServerConsumer {
// TODO Auto-generated method stub
return null;
}
@Override
public int getMessagesInTransit() {
return 0;
}
@Override
public long getLastDeliveredTime() {
return 0;
}
@Override
public long getLastAcknowledgedTime() {
return 0;
}
@Override
public long getMessagesAcknowledged() {
return 0;
}
@Override
public int getMessagesAcknowledgedAwaitingCommit() {
return 0;
}
}

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

@ -595,8 +595,9 @@ public class HangConsumerTest extends ActiveMQTestBase {
OperationContext context,
boolean autoCreateQueue,
Map<SimpleString, RoutingType> prefixes,
String securityDomain) throws Exception {
return new ServerSessionImpl(name, username, password, validatedUser, minLargeMessageSize, autoCommitSends, autoCommitAcks, preAcknowledge, getConfiguration().isPersistDeliveryCountBeforeDelivery(), xa, connection, getStorageManager(), getPostOffice(), getResourceManager(), getSecurityStore(), getManagementService(), this, getConfiguration().getManagementAddress(), defaultAddress == null ? null : new SimpleString(defaultAddress), new MyCallback(callback), context, getPagingManager(), prefixes, securityDomain);
String securityDomain,
boolean isLegacyProducer) throws Exception {
return new ServerSessionImpl(name, username, password, validatedUser, minLargeMessageSize, autoCommitSends, autoCommitAcks, preAcknowledge, getConfiguration().isPersistDeliveryCountBeforeDelivery(), xa, connection, getStorageManager(), getPostOffice(), getResourceManager(), getSecurityStore(), getManagementService(), this, getConfiguration().getManagementAddress(), defaultAddress == null ? null : new SimpleString(defaultAddress), new MyCallback(callback), context, getPagingManager(), prefixes, securityDomain, isLegacyProducer);
}
}

75
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/LargeMessageTest.java
View File

@ -29,6 +29,7 @@ import java.io.InputStream;
import java.lang.management.ManagementFactory;
import java.lang.management.OperatingSystemMXBean;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.HashMap;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
@ -61,6 +62,7 @@ import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.LargeServerMessage;
import org.apache.activemq.artemis.core.server.MessageReference;
import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.ServerProducer;
import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.tests.integration.largemessage.LargeMessageTestBase;
@ -2570,6 +2572,79 @@ public class LargeMessageTest extends LargeMessageTestBase {
session.commit();
}
@Test
public void testSendServerMessageMetrics() throws Exception {
ActiveMQServer server = createServer(true, isNetty(), storeType);
server.start();
ClientSessionFactory sf = addSessionFactory(createSessionFactory(locator));
ClientSession session = sf.createSession(false, false);
LargeServerMessageImpl fileMessage = new LargeServerMessageImpl((JournalStorageManager) server.getStorageManager());
fileMessage.setMessageID(1005);
for (int i = 0; i < largeMessageSize; i++) {
fileMessage.addBytes(new byte[]{ActiveMQTestBase.getSamplebyte(i)});
}
// The server would be doing this
fileMessage.putLongProperty(Message.HDR_LARGE_BODY_SIZE, largeMessageSize);
fileMessage.releaseResources(false, false);
session.createQueue(new QueueConfiguration(ADDRESS));
ClientProducer prod = session.createProducer(ADDRESS);
prod.send(fileMessage);
fileMessage.deleteFile();
session.commit();
Collection<ServerProducer> serverProducers = server.getSessions().iterator().next().getServerProducers();
Assert.assertEquals(1, serverProducers.size());
ServerProducer producer = serverProducers.iterator().next();
Assert.assertEquals(1, producer.getMessagesSent());
Assert.assertEquals(largeMessageSize, producer.getMessagesSentSize());
fileMessage = new LargeServerMessageImpl((JournalStorageManager) server.getStorageManager());
fileMessage.setMessageID(1006);
for (int i = 0; i < largeMessageSize; i++) {
fileMessage.addBytes(new byte[]{ActiveMQTestBase.getSamplebyte(i)});
}
// The server would be doing this
fileMessage.putLongProperty(Message.HDR_LARGE_BODY_SIZE, largeMessageSize);
fileMessage.releaseResources(false, false);
prod.send(fileMessage);
fileMessage.deleteFile();
session.commit();
serverProducers = server.getSessions().iterator().next().getServerProducers();
Assert.assertEquals(1, serverProducers.size());
producer = serverProducers.iterator().next();
Assert.assertEquals(2, producer.getMessagesSent());
Assert.assertEquals(largeMessageSize * 2, producer.getMessagesSentSize());
}
@Override

22
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/client/TemporaryDestinationTest.java
View File

@ -279,7 +279,7 @@ public class TemporaryDestinationTest extends JMSTestBase {
producer.send(s.createMessage());
temporaryQueue.delete();
for (ServerSession serverSession : server.getSessions()) {
assertFalse(((ServerSessionImpl)serverSession).cloneTargetAddresses().containsKey(SimpleString.toSimpleString(temporaryQueue.getQueueName())));
assertFalse(((ServerSessionImpl)serverSession).cloneProducers().containsKey(temporaryQueue.getQueueName()));
}
Wait.assertTrue(() -> server.locateQueue(temporaryQueue.getQueueName()) == null, 1000, 100);
Wait.assertTrue(() -> server.getAddressInfo(SimpleString.toSimpleString(temporaryQueue.getQueueName())) == null, 1000, 100);
@ -323,26 +323,6 @@ public class TemporaryDestinationTest extends JMSTestBase {
Wait.assertTrue(() -> server.getAddressInfo(SimpleString.toSimpleString(temporaryQueue.getQueueName())) == null, 1000, 100);
}
@Test
public void testForTempQueueTargetInfosSizeLimit() throws Exception {
try {
conn = createConnection();
Session s = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
for (int i = 0; i < 200; i++) {
TemporaryQueue temporaryQueue = s.createTemporaryQueue();
MessageProducer producer = s.createProducer(temporaryQueue);
producer.send(s.createMessage());
}
for (ServerSession serverSession : server.getSessions()) {
assertTrue(((ServerSessionImpl)serverSession).cloneTargetAddresses().size() <= 100);
}
} finally {
if (conn != null) {
conn.close();
}
}
}
@Test
public void testForSecurityCacheLeak() throws Exception {
server.getSecurityStore().setSecurityEnabled(true);

1789
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ActiveMQServerControlTest.java
View File

File diff suppressed because it is too large Load Diff

1
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ActiveMQServerControlUsingCoreTest.java
View File

@ -31,6 +31,7 @@ public class ActiveMQServerControlUsingCoreTest extends ActiveMQServerControlTes
public ActiveMQServerControlUsingCoreTest(boolean legacyCreateQueue) {
super(legacyCreateQueue);
extraProducers = 1;
}

452
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/QueueControlTest.java
View File

@ -19,14 +19,24 @@ package org.apache.activemq.artemis.tests.integration.management;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.artemis.api.core.TransportConfiguration;
import org.apache.activemq.artemis.api.jms.ActiveMQJMSClient;
import org.apache.activemq.artemis.api.jms.JMSFactoryType;
import org.apache.activemq.artemis.core.management.impl.view.ConsumerField;
import org.apache.activemq.artemis.core.paging.impl.PagingManagerTestAccessor;
import org.apache.activemq.artemis.core.server.impl.QueueImplTestAccessor;
import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
import org.apache.activemq.artemis.json.JsonArray;
import org.apache.activemq.artemis.json.JsonObject;
import javax.jms.XAConnection;
import javax.jms.XAConnectionFactory;
import javax.jms.XASession;
import javax.management.Notification;
import javax.management.openmbean.CompositeData;
import javax.management.openmbean.CompositeDataSupport;
@ -466,6 +476,10 @@ public class QueueControlTest extends ManagementTestBase {
assertEquals(1, obj.size());
assertEquals(0, obj.get(0).asJsonObject().getInt(ConsumerField.LAST_DELIVERED_TIME.getName()));
assertEquals(0, obj.get(0).asJsonObject().getInt(ConsumerField.LAST_ACKNOWLEDGED_TIME.getName()));
consumer.close();
Assert.assertEquals(0, queueControl.getConsumerCount());
@ -476,6 +490,442 @@ public class QueueControlTest extends ManagementTestBase {
session.deleteQueue(queue);
}
@Test
public void testGetConsumerWithMessagesJSON() throws Exception {
long currentTime = System.currentTimeMillis();
SimpleString address = RandomUtil.randomSimpleString();
SimpleString queue = RandomUtil.randomSimpleString();
session.createQueue(new QueueConfiguration(queue).setAddress(address).setDurable(durable));
QueueControl queueControl = createManagementControl(address, queue);
ClientProducer producer = session.createProducer(address);
for (int i = 0; i < 10; i++) {
producer.send(session.createMessage(true));
}
Wait.assertEquals(0, () -> queueControl.getConsumerCount());
ClientConsumer consumer = session.createConsumer(queue);
Wait.assertEquals(1, () -> queueControl.getConsumerCount());
session.start();
ClientMessage clientMessage = null;
int size = 0;
for (int i = 0; i < 5; i++) {
clientMessage = consumer.receiveImmediate();
size += clientMessage.getEncodeSize();
}
JsonArray obj = JsonUtil.readJsonArray(queueControl.listConsumersAsJSON());
assertEquals(1, obj.size());
Wait.assertEquals(5, () -> JsonUtil.readJsonArray(queueControl.listConsumersAsJSON()).get(0).asJsonObject().getInt(ConsumerField.MESSAGES_IN_TRANSIT.getName()));
obj = JsonUtil.readJsonArray(queueControl.listConsumersAsJSON());
JsonObject jsonObject = obj.get(0).asJsonObject();
assertEquals(5, jsonObject.getInt(ConsumerField.MESSAGES_IN_TRANSIT.getName()));
assertEquals(size, jsonObject.getInt(ConsumerField.MESSAGES_IN_TRANSIT_SIZE.getName()));
assertEquals(5, jsonObject.getInt(ConsumerField.MESSAGES_DELIVERED.getName()));
assertEquals(size, jsonObject.getInt(ConsumerField.MESSAGES_DELIVERED_SIZE.getName()));
assertEquals(0, jsonObject.getInt(ConsumerField.MESSAGES_ACKNOWLEDGED.getName()));
long lastDelivered = jsonObject.getJsonNumber(ConsumerField.LAST_DELIVERED_TIME.getName()).longValue();
assertTrue(lastDelivered > currentTime);
assertEquals( 0, jsonObject.getInt(ConsumerField.LAST_ACKNOWLEDGED_TIME.getName()));
clientMessage.acknowledge();
session.commit();
Wait.assertEquals(5, () -> JsonUtil.readJsonArray(queueControl.listConsumersAsJSON()).get(0).asJsonObject().getInt(ConsumerField.MESSAGES_ACKNOWLEDGED.getName()));
obj = JsonUtil.readJsonArray(queueControl.listConsumersAsJSON());
jsonObject = obj.get(0).asJsonObject();
assertEquals(0, jsonObject.getInt(ConsumerField.MESSAGES_IN_TRANSIT.getName()));
long lastAcked = jsonObject.getJsonNumber(ConsumerField.LAST_ACKNOWLEDGED_TIME.getName()).longValue();
assertTrue("lastAcked = " + lastAcked + " lastDelivered = " + lastDelivered, lastAcked >= lastDelivered);
currentTime = System.currentTimeMillis();
//now make sure they fall between the test time window
assertTrue("currentTime = " + currentTime + " lastAcked = " + lastAcked,currentTime >= lastAcked);
consumer.close();
Assert.assertEquals(0, queueControl.getConsumerCount());
obj = JsonUtil.readJsonArray(queueControl.listConsumersAsJSON());
assertEquals(0, obj.size());
session.deleteQueue(queue);
}
@Test
public void testGetConsumerMessageCountsAutoAckCore() throws Exception {
ActiveMQConnectionFactory factory = ActiveMQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF, new TransportConfiguration(INVM_CONNECTOR_FACTORY));
factory.setConsumerWindowSize(1);
testGetConsumerMessageCountsAutoAck(factory, false);
}
@Test
public void testGetConsumerMessageCountsAutoAckAMQP() throws Exception {
testGetConsumerMessageCountsAutoAck(new JmsConnectionFactory("amqp://localhost:61616"), false);
}
@Test
public void testGetConsumerMessageCountsAutoAckOpenWire() throws Exception {
testGetConsumerMessageCountsAutoAck(new org.apache.activemq.ActiveMQConnectionFactory("tcp://localhost:61616"), false);
}
@Test
public void testGetConsumerMessageCountsAutoAckCoreIndividualAck() throws Exception {
ActiveMQConnectionFactory factory = ActiveMQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF, new TransportConfiguration(INVM_CONNECTOR_FACTORY));
testGetConsumerMessageCountsAutoAck(factory, true);
}
public void testGetConsumerMessageCountsAutoAck(ConnectionFactory factory, boolean usePriority) throws Exception {
SimpleString queueName = RandomUtil.randomSimpleString();
this.session.createQueue(new QueueConfiguration(queueName).setAddress(queueName).setDurable(durable).setRoutingType(RoutingType.ANYCAST));
QueueControl queueControl = createManagementControl(queueName, queueName, RoutingType.ANYCAST);
Connection connection = factory.createConnection();
try {
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
javax.jms.Queue queue = session.createQueue(queueName.toString());
MessageProducer producer = session.createProducer(queue);
for (int i = 0; i < 100; i++) {
javax.jms.Message message = session.createMessage();
if (usePriority) {
producer.send(message, DeliveryMode.PERSISTENT, 1, javax.jms.Message.DEFAULT_TIME_TO_LIVE);
} else {
producer.send(message);
}
}
MessageConsumer consumer = session.createConsumer(queue);
connection.start();
javax.jms.Message message = null;
for (int i = 0; i < 100; i++) {
message = consumer.receive(5000);
Assert.assertNotNull("message " + i + " not received",message);
}
JsonArray obj = JsonUtil.readJsonArray(queueControl.listConsumersAsJSON());
assertEquals(1, obj.size());
Wait.assertEquals(0, () -> JsonUtil.readJsonArray(queueControl.listConsumersAsJSON()).get(0).asJsonObject().getInt(ConsumerField.MESSAGES_IN_TRANSIT.getName()));
Wait.assertEquals(0, () -> JsonUtil.readJsonArray(queueControl.listConsumersAsJSON()).get(0).asJsonObject().getInt(ConsumerField.MESSAGES_ACKNOWLEDGED_AWAITING_COMMIT.getName()));
obj = JsonUtil.readJsonArray(queueControl.listConsumersAsJSON());
JsonObject jsonObject = obj.get(0).asJsonObject();
assertEquals(0, jsonObject.getInt(ConsumerField.MESSAGES_IN_TRANSIT.getName()));
assertEquals(0, jsonObject.getInt(ConsumerField.MESSAGES_IN_TRANSIT_SIZE.getName()));
assertEquals(100, jsonObject.getInt(ConsumerField.MESSAGES_DELIVERED.getName()));
assertTrue(jsonObject.getInt(ConsumerField.MESSAGES_DELIVERED_SIZE.getName()) > 0);
assertEquals(100, jsonObject.getInt(ConsumerField.MESSAGES_ACKNOWLEDGED.getName()));
assertEquals(0, jsonObject.getInt(ConsumerField.MESSAGES_ACKNOWLEDGED_AWAITING_COMMIT.getName()));
consumer.close();
} finally {
connection.close();
}
}
@Test
public void testGetConsumerMessageCountsClientAckCore() throws Exception {
ActiveMQConnectionFactory factory = ActiveMQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF, new TransportConfiguration(INVM_CONNECTOR_FACTORY));
factory.setConsumerWindowSize(1);
testGetConsumerMessageCountsClientAck(factory, false);
}
@Test
public void testGetConsumerMessageCountsClientAckAMQP() throws Exception {
testGetConsumerMessageCountsClientAck(new JmsConnectionFactory("amqp://localhost:61616"), false);
}
@Test
public void testGetConsumerMessageCountsClientAckOpenWire() throws Exception {
testGetConsumerMessageCountsClientAck(new org.apache.activemq.ActiveMQConnectionFactory("tcp://localhost:61616"), false);
}
public void testGetConsumerMessageCountsClientAck(ConnectionFactory factory, boolean usePriority) throws Exception {
SimpleString queueName = RandomUtil.randomSimpleString();
this.session.createQueue(new QueueConfiguration(queueName).setAddress(queueName).setDurable(durable).setRoutingType(RoutingType.ANYCAST));
QueueControl queueControl = createManagementControl(queueName, queueName, RoutingType.ANYCAST);
Connection connection = factory.createConnection();
try {
Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
javax.jms.Queue queue = session.createQueue(queueName.toString());
MessageProducer producer = session.createProducer(queue);
for (int i = 0; i < 100; i++) {
javax.jms.Message message = session.createMessage();
if (usePriority) {
producer.send(message, DeliveryMode.PERSISTENT, 1, javax.jms.Message.DEFAULT_TIME_TO_LIVE);
} else {
producer.send(message);
}
}
MessageConsumer consumer = session.createConsumer(queue);
connection.start();
javax.jms.Message message = null;
for (int i = 0; i < 100; i++) {
message = consumer.receive(5000);
Assert.assertNotNull("message " + i + " not received",message);
message.acknowledge();
}
JsonArray obj = JsonUtil.readJsonArray(queueControl.listConsumersAsJSON());
assertEquals(1, obj.size());
Wait.assertEquals(0, () -> JsonUtil.readJsonArray(queueControl.listConsumersAsJSON()).get(0).asJsonObject().getInt(ConsumerField.MESSAGES_ACKNOWLEDGED_AWAITING_COMMIT.getName()));
Wait.assertEquals(0, () -> JsonUtil.readJsonArray(queueControl.listConsumersAsJSON()).get(0).asJsonObject().getInt(ConsumerField.MESSAGES_IN_TRANSIT.getName()));
Wait.assertEquals(0, () -> JsonUtil.readJsonArray(queueControl.listConsumersAsJSON()).get(0).asJsonObject().getInt(ConsumerField.MESSAGES_IN_TRANSIT_SIZE.getName()));
obj = JsonUtil.readJsonArray(queueControl.listConsumersAsJSON());
JsonObject jsonObject = obj.get(0).asJsonObject();
assertEquals(0, jsonObject.getInt(ConsumerField.MESSAGES_IN_TRANSIT.getName()));
assertEquals(0, jsonObject.getInt(ConsumerField.MESSAGES_IN_TRANSIT_SIZE.getName()));
assertEquals(100, jsonObject.getInt(ConsumerField.MESSAGES_DELIVERED.getName()));
assertTrue(jsonObject.getInt(ConsumerField.MESSAGES_DELIVERED_SIZE.getName()) > 0);
assertEquals(100, jsonObject.getInt(ConsumerField.MESSAGES_ACKNOWLEDGED.getName()));
assertEquals(0, jsonObject.getInt(ConsumerField.MESSAGES_ACKNOWLEDGED_AWAITING_COMMIT.getName()));
consumer.close();
} finally {
connection.close();
}
}
@Test
public void testGetConsumerMessageCountsTransactedCore() throws Exception {
ActiveMQConnectionFactory factory = ActiveMQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF, new TransportConfiguration(INVM_CONNECTOR_FACTORY));
factory.setConsumerWindowSize(1);
testGetConsumerMessageCountsTransacted(factory, false);
}
@Test
public void testGetConsumerMessageCountsTransactedAMQP() throws Exception {
testGetConsumerMessageCountsTransacted(new JmsConnectionFactory("amqp://localhost:61616"), false);
}
@Test
public void testGetConsumerMessageCountsTransactedOpenWire() throws Exception {
testGetConsumerMessageCountsTransacted(new org.apache.activemq.ActiveMQConnectionFactory("tcp://localhost:61616"), false);
}
public void testGetConsumerMessageCountsTransacted(ConnectionFactory factory, boolean usePriority) throws Exception {
SimpleString queueName = RandomUtil.randomSimpleString();
this.session.createQueue(new QueueConfiguration(queueName).setAddress(queueName).setDurable(durable).setRoutingType(RoutingType.ANYCAST));
QueueControl queueControl = createManagementControl(queueName, queueName, RoutingType.ANYCAST);
Connection connection = factory.createConnection();
try {
Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
javax.jms.Queue queue = session.createQueue(queueName.toString());
MessageProducer producer = session.createProducer(queue);
for (int i = 0; i < 100; i++) {
javax.jms.Message message = session.createMessage();
if (usePriority) {
producer.send(message, DeliveryMode.PERSISTENT, 1, javax.jms.Message.DEFAULT_TIME_TO_LIVE);
} else {
producer.send(message);
}
}
session.commit();
MessageConsumer consumer = session.createConsumer(queue);
connection.start();
javax.jms.Message message = null;
for (int i = 0; i < 100; i++) {
message = consumer.receive(5000);
Assert.assertNotNull("message " + i + " not received",message);
}
session.commit();
JsonArray obj = JsonUtil.readJsonArray(queueControl.listConsumersAsJSON());
assertEquals(1, obj.size());
Wait.assertEquals(0, () -> JsonUtil.readJsonArray(queueControl.listConsumersAsJSON()).get(0).asJsonObject().getInt(ConsumerField.MESSAGES_ACKNOWLEDGED_AWAITING_COMMIT.getName()));
obj = JsonUtil.readJsonArray(queueControl.listConsumersAsJSON());
JsonObject jsonObject = obj.get(0).asJsonObject();
assertEquals(0, jsonObject.getInt(ConsumerField.MESSAGES_IN_TRANSIT.getName()));
assertEquals(0, jsonObject.getInt(ConsumerField.MESSAGES_IN_TRANSIT_SIZE.getName()));
assertEquals(100, jsonObject.getInt(ConsumerField.MESSAGES_DELIVERED.getName()));
assertTrue(jsonObject.getInt(ConsumerField.MESSAGES_DELIVERED_SIZE.getName()) > 0);
assertEquals(100, jsonObject.getInt(ConsumerField.MESSAGES_ACKNOWLEDGED.getName()));
assertEquals(0, jsonObject.getInt(ConsumerField.MESSAGES_ACKNOWLEDGED_AWAITING_COMMIT.getName()));
consumer.close();
} finally {
connection.close();
}
}
@Test
public void testGetConsumerMessageCountsTransactedXACore() throws Exception {
ActiveMQConnectionFactory factory = ActiveMQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF, new TransportConfiguration(INVM_CONNECTOR_FACTORY));
factory.setConsumerWindowSize(1);
testGetConsumerMessageCountsTransactedXA(factory, false);
}
@Test
public void testGetConsumerMessageCountsTransactedXAOpenWire() throws Exception {
testGetConsumerMessageCountsTransactedXA(new org.apache.activemq.ActiveMQXAConnectionFactory("tcp://localhost:61616"), false);
}
public void testGetConsumerMessageCountsTransactedXA(XAConnectionFactory factory, boolean usePriority) throws Exception {
SimpleString queueName = RandomUtil.randomSimpleString();
this.session.createQueue(new QueueConfiguration(queueName).setAddress(queueName).setDurable(durable).setRoutingType(RoutingType.ANYCAST));
QueueControl queueControl = createManagementControl(queueName, queueName, RoutingType.ANYCAST);
XAConnection connection = factory.createXAConnection();
try {
XASession session = connection.createXASession();
XidImpl xid = newXID();
javax.jms.Queue queue = session.createQueue(queueName.toString());
MessageProducer producer = session.createProducer(queue);
session.getXAResource().start(xid, XAResource.TMNOFLAGS);
for (int i = 0; i < 100; i++) {
javax.jms.Message message = session.createMessage();
if (usePriority) {
producer.send(message, DeliveryMode.PERSISTENT, 1, javax.jms.Message.DEFAULT_TIME_TO_LIVE);
} else {
producer.send(message);
}
}
session.getXAResource().end(xid, XAResource.TMSUCCESS);
session.getXAResource().commit(xid, true);
MessageConsumer consumer = session.createConsumer(queue);
connection.start();
xid = newXID();
javax.jms.Message message = null;
session.getXAResource().start(xid, XAResource.TMNOFLAGS);
for (int i = 0; i < 100; i++) {
message = consumer.receive(5000);
Assert.assertNotNull("message " + i + " not received",message);
}
session.getXAResource().end(xid, XAResource.TMSUCCESS);
JsonArray obj = JsonUtil.readJsonArray(queueControl.listConsumersAsJSON());
assertEquals(1, obj.size());
Wait.assertEquals(100, () -> JsonUtil.readJsonArray(queueControl.listConsumersAsJSON()).get(0).asJsonObject().getInt(ConsumerField.MESSAGES_ACKNOWLEDGED_AWAITING_COMMIT.getName()));
obj = JsonUtil.readJsonArray(queueControl.listConsumersAsJSON());
JsonObject jsonObject = obj.get(0).asJsonObject();
assertEquals(0, jsonObject.getInt(ConsumerField.MESSAGES_IN_TRANSIT.getName()));
assertEquals(0, jsonObject.getInt(ConsumerField.MESSAGES_IN_TRANSIT_SIZE.getName()));
assertEquals(100, jsonObject.getInt(ConsumerField.MESSAGES_DELIVERED.getName()));
assertTrue(jsonObject.getInt(ConsumerField.MESSAGES_DELIVERED_SIZE.getName()) > 0);
assertEquals(100, jsonObject.getInt(ConsumerField.MESSAGES_ACKNOWLEDGED.getName()));
assertEquals(100, jsonObject.getInt(ConsumerField.MESSAGES_ACKNOWLEDGED_AWAITING_COMMIT.getName()));
session.getXAResource().commit(xid, true);
obj = JsonUtil.readJsonArray(queueControl.listConsumersAsJSON());
assertEquals(1, obj.size());
Wait.assertEquals(0, () -> JsonUtil.readJsonArray(queueControl.listConsumersAsJSON()).get(0).asJsonObject().getInt(ConsumerField.MESSAGES_ACKNOWLEDGED_AWAITING_COMMIT.getName()));
obj = JsonUtil.readJsonArray(queueControl.listConsumersAsJSON());
jsonObject = obj.get(0).asJsonObject();
assertEquals(0, jsonObject.getInt(ConsumerField.MESSAGES_IN_TRANSIT.getName()));
assertEquals(0, jsonObject.getInt(ConsumerField.MESSAGES_IN_TRANSIT_SIZE.getName()));
assertEquals(100, jsonObject.getInt(ConsumerField.MESSAGES_DELIVERED.getName()));
assertTrue(jsonObject.getInt(ConsumerField.MESSAGES_DELIVERED_SIZE.getName()) > 0);
assertEquals(100, jsonObject.getInt(ConsumerField.MESSAGES_ACKNOWLEDGED.getName()));
assertEquals(0, jsonObject.getInt(ConsumerField.MESSAGES_ACKNOWLEDGED_AWAITING_COMMIT.getName()));
consumer.close();
} finally {
connection.close();
}
}
@Test
public void testGetMessageCount() throws Exception {
SimpleString address = RandomUtil.randomSimpleString();
@ -589,7 +1039,7 @@ public class QueueControlTest extends ManagementTestBase {
if (key.equals("StringProperties")) {
// these are very verbose composite data structures
assertTrue(value.toString().length() + " truncated? " + key, value.toString().length() <= 2048);
assertTrue(value.toString().length() + " truncated? " + key, value.toString().length() <= 3000);
} else {
assertTrue(value.toString().length() + " truncated? " + key, value.toString().length() <= 512);
}

61
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/MQTTTest.java
View File

@ -37,12 +37,14 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Pattern;
import org.apache.activemq.artemis.api.core.JsonUtil;
import org.apache.activemq.artemis.api.core.QueueConfiguration;
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.config.Configuration;
import org.apache.activemq.artemis.core.config.CoreAddressConfiguration;
import org.apache.activemq.artemis.core.management.impl.view.ProducerField;
import org.apache.activemq.artemis.core.postoffice.Binding;
import org.apache.activemq.artemis.core.postoffice.QueueBinding;
import org.apache.activemq.artemis.core.protocol.mqtt.MQTTUtil;
@ -50,6 +52,8 @@ import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.json.JsonArray;
import org.apache.activemq.artemis.json.JsonObject;
import org.apache.activemq.artemis.logs.AssertionLoggerHandler;
import org.apache.activemq.artemis.tests.util.Wait;
import org.apache.activemq.transport.amqp.client.AmqpClient;
@ -168,6 +172,63 @@ public class MQTTTest extends MQTTTestSupport {
publishProvider.disconnect();
}
@Test
public void testProducerMetrics() throws Exception {
final MQTTClientProvider publishProvider = getMQTTClientProvider();
initializeConnection(publishProvider);
for (int i = 0; i < NUM_MESSAGES; i++) {
String payload = "Message " + i;
publishProvider.publish("foo/bah", payload.getBytes(), AT_LEAST_ONCE);
}
String filterString = createJsonFilter("", "", "");
String producersAsJsonString = server.getActiveMQServerControl().listProducers(filterString, 1, 50);
JsonObject producersAsJsonObject = JsonUtil.readJsonObject(producersAsJsonString);
JsonArray array = (JsonArray) producersAsJsonObject.get("data");
Assert.assertEquals("number of producers returned from query", 1, array.size());
JsonObject producer = array.getJsonObject(0);
Assert.assertNotEquals(ProducerField.ID.getName(), "", producer.getString(ProducerField.ID.getName()));
Assert.assertNotEquals(ProducerField.SESSION.getName(), "", producer.getString(ProducerField.SESSION.getName()));
Assert.assertNotEquals(ProducerField.CLIENT_ID.getName(), "", producer.getString(ProducerField.CLIENT_ID.getName()));
Assert.assertEquals(ProducerField.USER.getName(), "", producer.getString(ProducerField.USER.getName()));
Assert.assertEquals(ProducerField.PROTOCOL.getAlternativeName(), "MQTT", producer.getString(ProducerField.PROTOCOL.getName()));
Assert.assertEquals(ProducerField.ADDRESS.getName(), "ANONYMOUS", producer.getString(ProducerField.ADDRESS.getName()));
Assert.assertNotEquals(ProducerField.LOCAL_ADDRESS.getName(), "", producer.getString(ProducerField.LOCAL_ADDRESS.getName()));
Assert.assertNotEquals(ProducerField.REMOTE_ADDRESS.getName(), "", producer.getString(ProducerField.REMOTE_ADDRESS.getName()));
Assert.assertNotEquals(ProducerField.CREATION_TIME.getName(), "", producer.getString(ProducerField.CREATION_TIME.getName()));
Assert.assertEquals(ProducerField.MESSAGE_SENT.getName(), NUM_MESSAGES, producer.getInt(ProducerField.MESSAGE_SENT.getName()));
Assert.assertEquals(ProducerField.LAST_PRODUCED_MESSAGE_ID.getName(), "", producer.getString(ProducerField.LAST_PRODUCED_MESSAGE_ID.getName()));
final MQTTClientProvider publishProvider2 = getMQTTClientProvider();
initializeConnection(publishProvider2);
for (int i = 0; i < NUM_MESSAGES; i++) {
String payload = "Message " + i;
publishProvider2.publish("foo/bah", payload.getBytes(), AT_LEAST_ONCE);
}
filterString = createJsonFilter("", "", "");
producersAsJsonString = server.getActiveMQServerControl().listProducers(filterString, 1, 50);
producersAsJsonObject = JsonUtil.readJsonObject(producersAsJsonString);
array = (JsonArray) producersAsJsonObject.get("data");
Assert.assertEquals("number of producers returned from query", 2, array.size());
publishProvider.disconnect();
publishProvider2.disconnect();
filterString = createJsonFilter("", "", "");
producersAsJsonString = server.getActiveMQServerControl().listProducers(filterString, 1, 50);
producersAsJsonObject = JsonUtil.readJsonObject(producersAsJsonString);
array = (JsonArray) producersAsJsonObject.get("data");
Assert.assertEquals("number of producers returned from query", 0, array.size());
}
@Test(timeout = 60 * 1000)
public void testDirectDeliverFalse() throws Exception {
final MQTTClientProvider subscriptionProvider = getMQTTClientProvider();

42
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/StompTest.java
View File

@ -34,6 +34,7 @@ import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
import org.apache.activemq.artemis.api.core.JsonUtil;
import org.apache.activemq.artemis.api.core.QueueConfiguration;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString;
@ -47,6 +48,7 @@ import org.apache.activemq.artemis.api.core.management.ActiveMQServerControl;
import org.apache.activemq.artemis.api.core.management.ManagementHelper;
import org.apache.activemq.artemis.api.core.management.ResourceNames;
import org.apache.activemq.artemis.api.jms.ActiveMQJMSClient;
import org.apache.activemq.artemis.core.management.impl.view.ProducerField;
import org.apache.activemq.artemis.core.postoffice.Binding;
import org.apache.activemq.artemis.core.postoffice.QueueBinding;
import org.apache.activemq.artemis.core.postoffice.impl.LocalQueueBinding;
@ -59,6 +61,8 @@ import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.jms.client.ActiveMQMessage;
import org.apache.activemq.artemis.json.JsonArray;
import org.apache.activemq.artemis.json.JsonObject;
import org.apache.activemq.artemis.logs.AssertionLoggerHandler;
import org.apache.activemq.artemis.reader.MessageUtil;
import org.apache.activemq.artemis.spi.core.remoting.Acceptor;
@ -163,6 +167,44 @@ public class StompTest extends StompTestBase {
assertTrue(latch.await(60, TimeUnit.SECONDS));
}
@Test
public void testProducerMetrics() throws Exception {
conn.connect(defUser, defPass);
send(conn, getQueuePrefix() + getQueueName(), null, "Hello World!", true);
String filterString = createJsonFilter("", "", "");
String producersAsJsonString = server.getActiveMQServerControl().listProducers(filterString, 1, 50);
JsonObject producersAsJsonObject = JsonUtil.readJsonObject(producersAsJsonString);
JsonArray array = (JsonArray) producersAsJsonObject.get("data");
Assert.assertEquals("number of producers returned from query", 1, array.size());
JsonObject producer = array.getJsonObject(0);
Assert.assertEquals(ProducerField.MESSAGE_SENT.getName(), 1, producer.getInt(ProducerField.MESSAGE_SENT.getName()));
send(conn, getQueuePrefix() + getQueueName(), null, "Hello World!", true);
producersAsJsonString = server.getActiveMQServerControl().listProducers(filterString, 1, 50);
producersAsJsonObject = JsonUtil.readJsonObject(producersAsJsonString);
array = (JsonArray) producersAsJsonObject.get("data");
Assert.assertEquals("number of producers returned from query", 1, array.size());
producer = array.getJsonObject(0);
Assert.assertEquals(ProducerField.MESSAGE_SENT.getName(), 2, producer.getInt(ProducerField.MESSAGE_SENT.getName()));
conn.closeTransport();
Wait.assertEquals(0, () -> ((JsonArray) JsonUtil.readJsonObject(server.getActiveMQServerControl().listProducers(filterString, 1, 50)).get("data")).size());
producersAsJsonString = server.getActiveMQServerControl().listProducers(filterString, 1, 50);
producersAsJsonObject = JsonUtil.readJsonObject(producersAsJsonString);
array = (JsonArray) producersAsJsonObject.get("data");
Assert.assertEquals("number of producers returned from query", 0, array.size());
}
@Test
public void testSendOverDiskFull() throws Exception {
AssertionLoggerHandler.startCapture();