ARTEMIS-3493 - expose User ID (JMS Message ID) in send tab of console
https://issues.apache.org/jira/browse/ARTEMIS-3493
This commit is contained in:
Andy Taylor
parent
7ce1006e9d
commit
e37175784c
|
|
@ -84,6 +84,7 @@ public class ConsumerThread extends Thread {
|
|||
}
|
||||
} else {
|
||||
if (verbose) {
|
||||
System.out.println("JMS Message ID:" + msg.getJMSMessageID());
|
||||
if (bytesAsText && (msg instanceof BytesMessage)) {
|
||||
long length = ((BytesMessage) msg).getBodyLength();
|
||||
byte[] bytes = new byte[(int) length];
|
||||
|
|
|
|||
|
|
@ -150,6 +150,26 @@ public interface AddressControl {
|
|||
@Parameter(name = "user", desc = "The user to authenticate with") String user,
|
||||
@Parameter(name = "password", desc = "The users password to authenticate with") String password) throws Exception;
|
||||
|
||||
/**
|
||||
* @param headers the message headers and properties to set. Can only
|
||||
* container Strings maped to primitive types.
|
||||
* @param body the text to send
|
||||
* @param durable
|
||||
* @param user
|
||||
* @param password @return
|
||||
* @param createMessageId whether or not to auto generate a Message ID
|
||||
* @throws Exception
|
||||
*/
|
||||
@Operation(desc = "Sends a TextMessage to a password-protected address.", impact = MBeanOperationInfo.ACTION)
|
||||
String sendMessage(@Parameter(name = "headers", desc = "The headers to add to the message") Map<String, String> headers,
|
||||
@Parameter(name = "type", desc = "A type for the message") int type,
|
||||
@Parameter(name = "body", desc = "The body (byte[]) of the message encoded as a string using Base64") String body,
|
||||
@Parameter(name = "durable", desc = "Whether the message is durable") boolean durable,
|
||||
@Parameter(name = "user", desc = "The user to authenticate with") String user,
|
||||
@Parameter(name = "password", desc = "The users password to authenticate with") String password,
|
||||
@Parameter(name = "createMessageId", desc = "whether or not to auto generate a Message ID") boolean createMessageId) throws Exception;
|
||||
|
||||
|
||||
/**
|
||||
* Pauses all the queues bound to this address.Messages are no longer delivered to all its bounded queues.
|
||||
* Newly added queue will be paused too until resume is called.
|
||||
|
|
|
|||
|
|
@ -565,6 +565,25 @@ public interface QueueControl {
|
|||
@Parameter(name = "user", desc = "The user to authenticate with") String user,
|
||||
@Parameter(name = "password", desc = "The users password to authenticate with") String password) throws Exception;
|
||||
|
||||
/**
|
||||
* @param headers the message headers and properties to set. Can only
|
||||
* container Strings maped to primitive types.
|
||||
* @param body the text to send
|
||||
* @param durable
|
||||
* @param user
|
||||
* @param password @return
|
||||
* @param createMessageId whether or not to auto generate a Message ID
|
||||
* @throws Exception
|
||||
*/
|
||||
@Operation(desc = "Sends a TextMessage to a password-protected destination.", impact = MBeanOperationInfo.ACTION)
|
||||
String sendMessage(@Parameter(name = "headers", desc = "The headers to add to the message") Map<String, String> headers,
|
||||
@Parameter(name = "type", desc = "A type for the message") int type,
|
||||
@Parameter(name = "body", desc = "The body (byte[]) of the message encoded as a string using Base64") String body,
|
||||
@Parameter(name = "durable", desc = "Whether the message is durable") boolean durable,
|
||||
@Parameter(name = "user", desc = "The user to authenticate with") String user,
|
||||
@Parameter(name = "password", desc = "The users password to authenticate with") String password,
|
||||
@Parameter(name = "createMessageId", desc = "whether or not to auto generate a Message ID") boolean createMessageId) throws Exception;
|
||||
|
||||
/**
|
||||
* Changes the message's priority corresponding to the specified message ID to the specified priority.
|
||||
*
|
||||
|
|
|
|||
|
|
@ -42,6 +42,20 @@ var Artemis;
|
|||
<div class="form-group">
|
||||
<label>Durable </label>
|
||||
<input id="durable" type="checkbox" ng-model="$ctrl.message.durable" value="true">
|
||||
<button type="button" class="btn btn-link jvm-title-popover"
|
||||
uib-popover-template="'durable-info.html'" popover-placement="bottom-left"
|
||||
popover-title="Durable" popover-trigger="'mouseenter'">
|
||||
<span class="pficon pficon-info"></span>
|
||||
</button>
|
||||
</div>
|
||||
<div class="form-group">
|
||||
<label>Create Message ID </label>
|
||||
<input id="messageID" type="checkbox" ng-model="$ctrl.message.messageID" value="true">
|
||||
<button type="button" class="btn btn-link jvm-title-popover"
|
||||
uib-popover-template="'message-id-info.html'" popover-placement="bottom-left"
|
||||
popover-title="Message ID" popover-trigger="'mouseenter'">
|
||||
<span class="pficon pficon-info"></span>
|
||||
</button>
|
||||
</div>
|
||||
</form>
|
||||
</div>
|
||||
|
|
@ -87,7 +101,7 @@ var Artemis;
|
|||
</form>
|
||||
|
||||
<p>
|
||||
<button type="button" class="btn btn-primary artemis-send-message-button" ng-click="$ctrl.message.sendMessage($ctrl.message.durable)">Send message</button>
|
||||
<button type="button" class="btn btn-primary artemis-send-message-button" ng-click="$ctrl.message.sendMessage($ctrl.message.durable, $ctrl.message.messageID)">Send message</button>
|
||||
</p>
|
||||
<script type="text/ng-template" id="send-message-instructions.html">
|
||||
<div>
|
||||
|
|
@ -98,7 +112,23 @@ var Artemis;
|
|||
be null.
|
||||
</p>
|
||||
</div>
|
||||
</script>
|
||||
</script>
|
||||
<script type="text/ng-template" id="message-id-info.html">
|
||||
<div>
|
||||
<p>
|
||||
The Message ID is an automatically generated UUID that is set on the Message by the broker before it is routed.
|
||||
If using a JMS client this would be the JMS Message ID on the JMS Message, this typically would not get
|
||||
set for non JMS clients. Historically and on some other tabs this is also referred to as the User ID.
|
||||
</p>
|
||||
</div>
|
||||
</script>
|
||||
<script type="text/ng-template" id="durable-info.html">
|
||||
<div>
|
||||
<p>
|
||||
If durable the message will be marked persistent and written to the brokers journal if the destination queue is durable.
|
||||
</p>
|
||||
</div>
|
||||
</script>
|
||||
`,
|
||||
controller: AddressSendMessageController
|
||||
})
|
||||
|
|
@ -110,7 +140,11 @@ var Artemis;
|
|||
'durable': {
|
||||
'value': true,
|
||||
'converter': Core.parseBooleanValue
|
||||
}
|
||||
},
|
||||
'messageID': {
|
||||
'value': true,
|
||||
'converter': Core.parseBooleanValue
|
||||
}
|
||||
});
|
||||
var ctrl = this;
|
||||
ctrl.messageCreator = messageCreator;
|
||||
|
|
|
|||
|
|
@ -42,6 +42,20 @@ var Artemis;
|
|||
<div class="form-group">
|
||||
<label>Durable </label>
|
||||
<input id="durable" type="checkbox" ng-model="$ctrl.message.durable" value="true">
|
||||
<button type="button" class="btn btn-link jvm-title-popover"
|
||||
uib-popover-template="'durable-info.html'" popover-placement="bottom-left"
|
||||
popover-title="Durable" popover-trigger="'mouseenter'">
|
||||
<span class="pficon pficon-info"></span>
|
||||
</button>
|
||||
</div>
|
||||
<div class="form-group">
|
||||
<label>Create Message ID </label>
|
||||
<input id="messageID" type="checkbox" ng-model="$ctrl.message.messageID" value="true">
|
||||
<button type="button" class="btn btn-link jvm-title-popover"
|
||||
uib-popover-template="'message-id-info.html'" popover-placement="bottom-left"
|
||||
popover-title="Message ID" popover-trigger="'mouseenter'">
|
||||
<span class="pficon pficon-info"></span>
|
||||
</button>
|
||||
</div>
|
||||
</form>
|
||||
</div>
|
||||
|
|
@ -87,7 +101,7 @@ var Artemis;
|
|||
</form>
|
||||
|
||||
<p>
|
||||
<button type="button" class="btn btn-primary artemis-send-message-button" ng-click="$ctrl.message.sendMessage($ctrl.durable)">Send message</button>
|
||||
<button type="button" class="btn btn-primary artemis-send-message-button" ng-click="$ctrl.message.sendMessage($ctrl.message.durable, $ctrl.message.messageID)">Send message</button>
|
||||
</p>
|
||||
<script type="text/ng-template" id="send-message-instructions.html">
|
||||
<div>
|
||||
|
|
@ -98,7 +112,23 @@ var Artemis;
|
|||
be null.
|
||||
</p>
|
||||
</div>
|
||||
</script>
|
||||
</script>
|
||||
<script type="text/ng-template" id="message-id-info.html">
|
||||
<div>
|
||||
<p>
|
||||
The Message ID is an automatically generated UUID that is set on the Message by the broker before it is routed.
|
||||
If using a JMS client this would be the JMS Message ID on the JMS Message, this typically would not get
|
||||
set for non JMS clients. Historically and on some other tabs this is also referred to as the User ID.
|
||||
</p>
|
||||
</div>
|
||||
</script>
|
||||
<script type="text/ng-template" id="durable-info.html">
|
||||
<div>
|
||||
<p>
|
||||
If durable the message will be marked persistent and written to the brokers journal if the destination queue is durable.
|
||||
</p>
|
||||
</div>
|
||||
</script>
|
||||
`,
|
||||
controller: SendMessageController
|
||||
})
|
||||
|
|
@ -110,6 +140,10 @@ var Artemis;
|
|||
'durable': {
|
||||
'value': true,
|
||||
'converter': Core.parseBooleanValue
|
||||
},
|
||||
'messageID': {
|
||||
'value': true,
|
||||
'converter': Core.parseBooleanValue
|
||||
}
|
||||
});
|
||||
var ctrl = this;
|
||||
|
|
|
|||
|
|
@ -30,6 +30,7 @@ var Artemis;
|
|||
function message(scope, location, route, localStorage, artemisMessage, workspace, element, timeout, jolokia) {
|
||||
this.noCredentials = false,
|
||||
this.durable = true,
|
||||
this.messageID = false;
|
||||
this.message = "",
|
||||
this.headers = [],
|
||||
this.scope = scope;
|
||||
|
|
@ -118,12 +119,12 @@ var Artemis;
|
|||
this.formatMessage = function () {
|
||||
CodeEditor.autoFormatEditor(this.scope.codeMirror);
|
||||
};
|
||||
this.sendMessage = function (durable) {
|
||||
this.sendMessage = function (durable, createMessageId) {
|
||||
var body = this.message;
|
||||
Artemis.log.debug(body);
|
||||
this.doSendMessage(this.durable, body);
|
||||
this.doSendMessage(this.durable, createMessageId, body);
|
||||
};
|
||||
this.doSendMessage = function(durable, body) {
|
||||
this.doSendMessage = function(durable, createMessageId, body) {
|
||||
var selection = this.workspace.selection;
|
||||
if (selection) {
|
||||
var mbean = selection.objectName;
|
||||
|
|
@ -151,7 +152,7 @@ var Artemis;
|
|||
Artemis.log.debug(type);
|
||||
Artemis.log.debug(body);
|
||||
Artemis.log.debug(durable);
|
||||
this.jolokia.execute(mbean, "sendMessage(java.util.Map, int, java.lang.String, boolean, java.lang.String, java.lang.String)", headers, type, body, durable, user, pwd, Core.onSuccess(this.operationSuccess(), { error: this.onError }));
|
||||
this.jolokia.execute(mbean, "sendMessage(java.util.Map, int, java.lang.String, boolean, java.lang.String, java.lang.String, boolean)", headers, type, body, durable, user, pwd, createMessageId, Core.onSuccess(this.operationSuccess(), { error: this.onError }));
|
||||
Core.$apply(this.scope);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -36,6 +36,7 @@ import org.apache.activemq.artemis.core.server.ServerSession;
|
|||
import org.apache.activemq.artemis.logs.AuditLogger;
|
||||
import org.apache.activemq.artemis.utils.Base64;
|
||||
import org.apache.activemq.artemis.utils.RunnableEx;
|
||||
import org.apache.activemq.artemis.utils.UUID;
|
||||
import org.apache.activemq.artemis.utils.UUIDGenerator;
|
||||
|
||||
public abstract class AbstractControl extends StandardMBean {
|
||||
|
|
@ -122,6 +123,7 @@ public abstract class AbstractControl extends StandardMBean {
|
|||
boolean durable,
|
||||
String user,
|
||||
String password,
|
||||
boolean createMessageId,
|
||||
Long...queueID) throws Exception {
|
||||
ManagementRemotingConnection fakeConnection = new ManagementRemotingConnection();
|
||||
ServerSession serverSession = server.createSession("management::" + UUIDGenerator.getInstance().generateStringUUID(), user, password,
|
||||
|
|
@ -159,6 +161,11 @@ public abstract class AbstractControl extends StandardMBean {
|
|||
message.putBytesProperty(Message.HDR_ROUTE_TO_IDS, buffer.array());
|
||||
}
|
||||
|
||||
if (createMessageId) {
|
||||
UUID userID = UUIDGenerator.getInstance().generateUUID();
|
||||
message.setUserID(userID);
|
||||
}
|
||||
|
||||
// There's no point on direct delivery using the management thread, use false here
|
||||
serverSession.send(message, false);
|
||||
return "" + message.getMessageID();
|
||||
|
|
|
|||
|
|
@ -379,11 +379,22 @@ public class AddressControlImpl extends AbstractControl implements AddressContro
|
|||
boolean durable,
|
||||
final String user,
|
||||
final String password) throws Exception {
|
||||
return sendMessage(headers, type, body, durable, user, password, false);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String sendMessage(final Map<String, String> headers,
|
||||
final int type,
|
||||
final String body,
|
||||
boolean durable,
|
||||
final String user,
|
||||
final String password,
|
||||
boolean createMessageId) throws Exception {
|
||||
if (AuditLogger.isBaseLoggingEnabled()) {
|
||||
AuditLogger.sendMessageThroughManagement(this, headers, type, body, durable, user, "****");
|
||||
}
|
||||
try {
|
||||
return sendMessage(addressInfo.getName(), server, headers, type, body, durable, user, password);
|
||||
return sendMessage(addressInfo.getName(), server, headers, type, body, durable, user, password, createMessageId);
|
||||
} catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
throw new IllegalStateException(e.getMessage());
|
||||
|
|
|
|||
|
|
@ -1308,11 +1308,22 @@ public class QueueControlImpl extends AbstractControl implements QueueControl {
|
|||
boolean durable,
|
||||
final String user,
|
||||
final String password) throws Exception {
|
||||
return sendMessage(headers, type, body, durable, user, password, false);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String sendMessage(final Map<String, String> headers,
|
||||
final int type,
|
||||
final String body,
|
||||
boolean durable,
|
||||
final String user,
|
||||
final String password,
|
||||
boolean createMessageId) throws Exception {
|
||||
if (AuditLogger.isBaseLoggingEnabled()) {
|
||||
AuditLogger.sendMessageThroughManagement(queue, headers, type, body, durable, user, "****");
|
||||
}
|
||||
try {
|
||||
String s = sendMessage(queue.getAddress(), server, headers, type, body, durable, user, password, queue.getID());
|
||||
String s = sendMessage(queue.getAddress(), server, headers, type, body, durable, user, password, createMessageId, queue.getID());
|
||||
if (AuditLogger.isResourceLoggingEnabled()) {
|
||||
AuditLogger.sendMessageSuccess(queue.getName().toString(), user);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -491,6 +491,35 @@ public class AddressControlTest extends ManagementTestBase {
|
|||
assertEquals("myValue2", message.getStringProperty("myProp2"));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSendMessageWithMessageId() throws Exception {
|
||||
SimpleString address = RandomUtil.randomSimpleString();
|
||||
session.createAddress(address, RoutingType.ANYCAST, false);
|
||||
|
||||
AddressControl addressControl = createManagementControl(address);
|
||||
Assert.assertEquals(0, addressControl.getQueueNames().length);
|
||||
session.createQueue(new QueueConfiguration(address).setRoutingType(RoutingType.ANYCAST));
|
||||
Assert.assertEquals(1, addressControl.getQueueNames().length);
|
||||
addressControl.sendMessage(null, Message.BYTES_TYPE, Base64.encodeBytes("test".getBytes()), false, null, null, true);
|
||||
addressControl.sendMessage(null, Message.BYTES_TYPE, Base64.encodeBytes("test".getBytes()), false, null, null, false);
|
||||
|
||||
Wait.waitFor(() -> addressControl.getMessageCount() == 2);
|
||||
Assert.assertEquals(2, addressControl.getMessageCount());
|
||||
|
||||
ClientConsumer consumer = session.createConsumer(address);
|
||||
ClientMessage message = consumer.receive(500);
|
||||
assertNotNull(message);
|
||||
assertNotNull(message.getUserID());
|
||||
byte[] buffer = new byte[message.getBodyBuffer().readableBytes()];
|
||||
message.getBodyBuffer().readBytes(buffer);
|
||||
assertEquals("test", new String(buffer));message = consumer.receive(500);
|
||||
assertNotNull(message);
|
||||
assertNull(message.getUserID());
|
||||
buffer = new byte[message.getBodyBuffer().readableBytes()];
|
||||
message.getBodyBuffer().readBytes(buffer);
|
||||
assertEquals("test", new String(buffer));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testGetCurrentDuplicateIdCacheSize() throws Exception {
|
||||
internalDuplicateIdTest(false);
|
||||
|
|
|
|||
|
|
@ -187,6 +187,17 @@ public class AddressControlUsingCoreTest extends AddressControlTest {
|
|||
String password) throws Exception {
|
||||
return (String) proxy.invokeOperation("sendMessage", headers, type, body, durable, user, password);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String sendMessage(Map<String, String> headers,
|
||||
int type,
|
||||
String body,
|
||||
boolean durable,
|
||||
String user,
|
||||
String password,
|
||||
boolean createMessageId) throws Exception {
|
||||
return (String) proxy.invokeOperation("sendMessage", headers, type, body, durable, user, password, createMessageId);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -3447,6 +3447,46 @@ public class QueueControlTest extends ManagementTestBase {
|
|||
Assert.assertEquals(new String(body), "theBody");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSendMessageWithMessageId() throws Exception {
|
||||
SimpleString address = RandomUtil.randomSimpleString();
|
||||
SimpleString queue = RandomUtil.randomSimpleString();
|
||||
|
||||
session.createQueue(new QueueConfiguration(queue).setAddress(address).setDurable(durable));
|
||||
|
||||
QueueControl queueControl = createManagementControl(address, queue);
|
||||
|
||||
queueControl.sendMessage(new HashMap<String, String>(), Message.BYTES_TYPE, Base64.encodeBytes("theBody".getBytes()), true, "myUser", "myPassword");
|
||||
queueControl.sendMessage(null, Message.BYTES_TYPE, Base64.encodeBytes("theBody".getBytes()), true, "myUser", "myPassword", true);
|
||||
|
||||
Wait.assertEquals(2, () -> getMessageCount(queueControl));
|
||||
|
||||
// the message IDs are set on the server
|
||||
CompositeData[] browse = queueControl.browse(null);
|
||||
|
||||
Assert.assertEquals(2, browse.length);
|
||||
|
||||
byte[] body = (byte[]) browse[0].get(BODY);
|
||||
|
||||
String messageID = (String) browse[0].get("userID");
|
||||
|
||||
Assert.assertEquals(0, messageID.length());
|
||||
|
||||
Assert.assertNotNull(body);
|
||||
|
||||
Assert.assertEquals(new String(body), "theBody");
|
||||
|
||||
body = (byte[]) browse[1].get(BODY);
|
||||
|
||||
messageID = (String) browse[1].get("userID");
|
||||
|
||||
Assert.assertTrue(messageID.length() > 0);
|
||||
|
||||
Assert.assertNotNull(body);
|
||||
|
||||
Assert.assertEquals(new String(body), "theBody");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSendMessageWithProperties() throws Exception {
|
||||
SimpleString address = RandomUtil.randomSimpleString();
|
||||
|
|
|
|||
|
|
@ -535,6 +535,17 @@ public class QueueControlUsingCoreTest extends QueueControlTest {
|
|||
return (String) proxy.invokeOperation("sendMessage", headers, type, body, durable, user, password);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String sendMessage(Map<String, String> headers,
|
||||
int type,
|
||||
String body,
|
||||
boolean durable,
|
||||
String user,
|
||||
String password,
|
||||
boolean createMessageId) throws Exception {
|
||||
return (String) proxy.invokeOperation("sendMessage", headers, type, body, durable, user, password, createMessageId);
|
||||
}
|
||||
|
||||
public void setDeadLetterAddress(final String deadLetterAddress) throws Exception {
|
||||
proxy.invokeOperation("setDeadLetterAddress", deadLetterAddress);
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue