Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Implement --conn-anonymous-producer option for JMS clients #567

Open
wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Implement --conn-anonymous-producer option for JMS clients #567

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0759a02
Assorted jmslib and jakartalib fixups
jiridanek Mar 24, 2023
eb52263
Implement --conn-anonymous-producer option
jiridanek Mar 24, 2023
3995aa8
fixup
jiridanek Mar 25, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion cli-qpid-jms/src/main/java/com/redhat/mqe/jms/AacSenderOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,8 @@ public class AacSenderOptions extends AacClientOptions {
// TODO
new Option(SYNC_MODE, "", "SYNCMODE", "action", "synchronization mode: none/session/action/persistent/transient"),
new Option(CAPACITY, "", "CAPACITY", "-1", "sender|receiver capacity (no effect in jms atm)"),
new Option(ON_RELEASE, "", "ACTION", "fail", "fail|ignore|retry action to perform if message is released by receiver")
new Option(ON_RELEASE, "", "ACTION", "fail", "fail|ignore|retry action to perform if message is released by receiver"),
new Option(CONN_ANONYMOUS_PRODUCER, "", "ANONYMOUS", "no", "create anonymous (no queue specified) producer")
));
}

Expand Down
1 change: 1 addition & 0 deletions jakartalib/src/main/java/com/redhat/mqe/lib/ClientOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,6 +145,7 @@ void setOptions(ClientOptionManager clientOptionManager, @Args String[] args) {
public static final String CONN_TCP_NO_DELAY = "conn-tcp-no-delay"; // wireFormat.tcpNoDelayEnabled
public static final String CONN_TIGHT_ENCODING_ENA = "conn-tight-encoding-ena"; // wireFormat.tightEncodingEnabled
public static final String CONN_WATCH_TOPIC_ADVISORIES = "conn-watch-topic-advisories";
public static final String CONN_ANONYMOUS_PRODUCER = "conn-anonymous-producer";

// TODO Not implemented by client libraries
// static final String CON_SSL_PROTOCOL = "conn-ssl-protocol";
Expand Down
2 changes: 1 addition & 1 deletion jakartalib/src/main/java/com/redhat/mqe/lib/ReceiverClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -155,7 +155,7 @@ private void unsubscribe() {
try {
session.unsubscribe(durableSubscriberName);
} catch (JMSException e) {
LOG.error("Error while unsubscribing durable subscriptor " + durableSubscriberName);
LOG.error("Error while unsubscribing durable subscriber " + durableSubscriberName);
e.printStackTrace();
} finally {
close(session);
Expand Down
15 changes: 10 additions & 5 deletions jakartalib/src/main/java/com/redhat/mqe/lib/SenderClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,9 @@ public void startClient() {
Session session = (transaction == null || transaction.equals("none")) ?
this.createSession(senderOptions, connection, false) : this.createSession(senderOptions, connection, true);
connection.start();
MessageProducer msgProducer = session.createProducer(this.getDestination());

boolean anonymousProducer = Utils.convertOptionToBoolean(senderOptions.getOption(ClientOptions.CONN_ANONYMOUS_PRODUCER).getValue());
MessageProducer msgProducer = anonymousProducer ? session.createProducer(null) : session.createProducer(this.getDestination());
setMessageProducer(senderOptions, msgProducer);

// Calculate msg-rate from COUNT & DURATION
Expand All @@ -104,7 +106,11 @@ public void startClient() {

// Send messages
try {
msgProducer.send(message);
if (anonymousProducer) {
msgProducer.send(getDestination(), message);
} else {
msgProducer.send(message);
}
} catch (Exception e) {
switch (e.getCause().getClass().getName()) {
case "org.apache.qpid.jms.provider.exceptions.ProviderDeliveryReleasedException":
Expand Down Expand Up @@ -165,7 +171,7 @@ public void startClient() {
doTransaction(session, senderOptions.getOption(ClientOptions.TX_ENDLOOP_ACTION).getValue());
}
} catch (JMSException | IllegalArgumentException jmse) {
LOG.error("Error while sending a message!", jmse.getMessage());
LOG.error("Error while sending a message! {}", jmse.getMessage());
jmse.printStackTrace();
System.exit(1);
} finally {
Expand All @@ -185,8 +191,7 @@ public void startClient() {
protected static void setMessageProducer(ClientOptions senderOptions, MessageProducer producer) {
try {
// set delivery mode - durable/non-durable
String deliveryModeArg = senderOptions.getOption(ClientOptions.MSG_DURABLE).getValue().toLowerCase();
int deliveryMode = (deliveryModeArg.equals("true") || deliveryModeArg.equals("yes"))
int deliveryMode = Utils.convertOptionToBoolean(senderOptions.getOption(ClientOptions.MSG_DURABLE).getValue())
? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT;
producer.setDeliveryMode(deliveryMode);
// set time to live of message if provided
Expand Down
1 change: 1 addition & 0 deletions jakartalib/src/main/java/com/redhat/mqe/lib/SenderOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@ public class SenderOptions extends ClientOptions {
new Option(PROPERTY_TYPE, "", "PTYPE", "String", "specify the type of message property"),
new Option(MSG_PROPERTY, "", "KEY=PVALUE", "", "specify message property as KEY=VALUE (use '~' instead of '=' for auto-casting)"),
new Option(CONTENT_TYPE, "", "CTYPE", "String", "specify type of the actual content type"),
new Option(CONN_ANONYMOUS_PRODUCER, "", "ANONYMOUS", "no", "create anonymous (no queue specified) producer"),
new Option(MSG_CONTENT_TYPE, "", "MSGTYPE", "", "type of JMSMessageBody to use in header"),
new Option(MSG_CONTENT_FROM_FILE, "", "PATH", "", "specify filename to load content from"),
new Option(MSG_CONTENT, "", "CONTENT", "", "actual content fed to message body"),
Expand Down
1 change: 1 addition & 0 deletions jmslib/src/main/java/com/redhat/mqe/lib/ClientOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,6 +145,7 @@ void setOptions(ClientOptionManager clientOptionManager, @Args String[] args) {
public static final String CONN_TCP_NO_DELAY = "conn-tcp-no-delay"; // wireFormat.tcpNoDelayEnabled
public static final String CONN_TIGHT_ENCODING_ENA = "conn-tight-encoding-ena"; // wireFormat.tightEncodingEnabled
public static final String CONN_WATCH_TOPIC_ADVISORIES = "conn-watch-topic-advisories";
public static final String CONN_ANONYMOUS_PRODUCER = "conn-anonymous-producer";

// TODO Not implemented by client libraries
// static final String CON_SSL_PROTOCOL = "conn-ssl-protocol";
Expand Down
15 changes: 10 additions & 5 deletions jmslib/src/main/java/com/redhat/mqe/lib/SenderClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,9 @@ public void startClient() {
Session session = (transaction == null || transaction.equals("none")) ?
this.createSession(senderOptions, connection, false) : this.createSession(senderOptions, connection, true);
connection.start();
MessageProducer msgProducer = session.createProducer(this.getDestination());

boolean anonymousProducer = Utils.convertOptionToBoolean(senderOptions.getOption(ClientOptions.CONN_ANONYMOUS_PRODUCER).getValue());
MessageProducer msgProducer = anonymousProducer ? session.createProducer(null) : session.createProducer(this.getDestination());
setMessageProducer(senderOptions, msgProducer);

// Calculate msg-rate from COUNT & DURATION
Expand All @@ -104,7 +106,11 @@ public void startClient() {

// Send messages
try {
msgProducer.send(message);
if (anonymousProducer) {
msgProducer.send(getDestination(), message);
} else {
msgProducer.send(message);
}
} catch (Exception e) {
switch (e.getCause().getClass().getName()) {
case "org.apache.qpid.jms.provider.exceptions.ProviderDeliveryReleasedException":
Expand Down Expand Up @@ -165,7 +171,7 @@ public void startClient() {
doTransaction(session, senderOptions.getOption(ClientOptions.TX_ENDLOOP_ACTION).getValue());
}
} catch (JMSException | IllegalArgumentException jmse) {
LOG.error("Error while sending a message!", jmse.getMessage());
LOG.error("Error while sending a message! {}", jmse.getMessage());
jmse.printStackTrace();
System.exit(1);
} finally {
Expand All @@ -185,8 +191,7 @@ public void startClient() {
protected static void setMessageProducer(ClientOptions senderOptions, MessageProducer producer) {
try {
// set delivery mode - durable/non-durable
String deliveryModeArg = senderOptions.getOption(ClientOptions.MSG_DURABLE).getValue().toLowerCase();
int deliveryMode = (deliveryModeArg.equals("true") || deliveryModeArg.equals("yes"))
int deliveryMode = Utils.convertOptionToBoolean(senderOptions.getOption(ClientOptions.MSG_DURABLE).getValue())
? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT;
producer.setDeliveryMode(deliveryMode);
// set time to live of message if provided
Expand Down
1 change: 1 addition & 0 deletions jmslib/src/main/java/com/redhat/mqe/lib/SenderOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@ public class SenderOptions extends ClientOptions {
new Option(PROPERTY_TYPE, "", "PTYPE", "String", "specify the type of message property"),
new Option(MSG_PROPERTY, "", "KEY=PVALUE", "", "specify message property as KEY=VALUE (use '~' instead of '=' for auto-casting)"),
new Option(CONTENT_TYPE, "", "CTYPE", "String", "specify type of the actual content type"),
new Option(CONN_ANONYMOUS_PRODUCER, "", "ANONYMOUS", "no", "create anonymous (no queue specified) producer"),
new Option(MSG_CONTENT_TYPE, "", "MSGTYPE", "", "type of JMSMessageBody to use in header"),
new Option(MSG_CONTENT_FROM_FILE, "", "PATH", "", "specify filename to load content from"),
new Option(MSG_CONTENT, "", "CONTENT", "", "actual content fed to message body"),
Expand Down
5 changes: 5 additions & 0 deletions lib/src/main/java/com/redhat/mqe/lib/Utils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -294,4 +294,9 @@ public static Object getObjectValue(Class<?> clazz, String object, boolean allow
return myObj;
}


public static boolean convertOptionToBoolean(String optionStringValue) {
String optionValue = optionStringValue.toLowerCase();
return (optionValue.equals("true") || optionValue.equals("yes"));
}
}