rewritten connector
This commit is contained in:
Radek Davidek
parent
579246d772
commit
0833bcff06
@ -24,424 +24,423 @@ import java.util.Map;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
/**
|
||||
* IBM MQ connector using JMS client with Jakarta JMS API.
|
||||
* Supports multi-instance Queue Manager, SSL/TLS, and multiple message formats.
|
||||
* IBM MQ connector using JMS client with Jakarta JMS API. Supports
|
||||
* multi-instance Queue Manager, SSL/TLS, and multiple message formats.
|
||||
* <p>
|
||||
* Supported formats:
|
||||
* - JSON: JMS TextMessage with plain JSON string (default)
|
||||
* - XML: JMS TextMessage with XML string
|
||||
* - UTF-8 (CCSID 1208): JMS BytesMessage with UTF-8 encoding
|
||||
* - EBCDIC (CCSID 870): JMS BytesMessage with EBCDIC IBM-870 encoding
|
||||
* Supported formats: - JSON: JMS TextMessage with plain JSON string (default) -
|
||||
* XML: JMS TextMessage with XML string - UTF-8 (CCSID 1208): JMS BytesMessage
|
||||
* with UTF-8 encoding - EBCDIC (CCSID 870): JMS BytesMessage with EBCDIC
|
||||
* IBM-870 encoding
|
||||
*/
|
||||
public class IbmMqConnector implements Connector {
|
||||
|
||||
private static final Logger LOG = LogManager.getLogger(IbmMqConnector.class);
|
||||
private static final Logger LOG = LogManager.getLogger(IbmMqConnector.class);
|
||||
|
||||
private static final Charset EBCDIC_870 = Charset.forName("IBM870");
|
||||
private static final Charset UTF_8 = StandardCharsets.UTF_8;
|
||||
private static final Charset EBCDIC_870 = Charset.forName("IBM870");
|
||||
private static final Charset UTF_8 = StandardCharsets.UTF_8;
|
||||
|
||||
private static final long DEFAULT_POLL_INTERVAL_MS = 100;
|
||||
private static final long DEFAULT_MAX_POLL_INTERVAL_MS = 1000;
|
||||
private static final long DEFAULT_POLL_INTERVAL_MS = 100;
|
||||
private static final long DEFAULT_MAX_POLL_INTERVAL_MS = 1000;
|
||||
|
||||
private final MQConnectionFactory connectionFactory;
|
||||
private JMSContext jmsContext;
|
||||
private final String queueManager;
|
||||
private final MQConnectionFactory connectionFactory;
|
||||
private JMSContext jmsContext;
|
||||
private final String queueManager;
|
||||
private final String user;
|
||||
private final String password;
|
||||
|
||||
/**
|
||||
* Constructor with multi-instance Queue Manager support.
|
||||
*
|
||||
* @param connectionNameList Connection name list in format "host1(port1),host2(port2)"
|
||||
* @param channel MQ channel name
|
||||
* @param queueManager Queue Manager name
|
||||
* @param user Username for authentication
|
||||
* @param password Password for authentication
|
||||
* @param keystorePath Path to SSL keystore (can be null for non-SSL)
|
||||
* @param keystorePassword Password for SSL keystore
|
||||
* @param sslCipherSuite SSL cipher suite to use (e.g., "TLS_RSA_WITH_AES_256_CBC_SHA256")
|
||||
*/
|
||||
public IbmMqConnector(String connectionNameList, String channel, String queueManager,
|
||||
String user, String password,
|
||||
String keystorePath, String keystorePassword, String sslCipherSuite) {
|
||||
this.queueManager = queueManager;
|
||||
/**
|
||||
* Constructor with multi-instance Queue Manager support.
|
||||
*
|
||||
* @param connectionNameList Connection name list in format
|
||||
* "host1(port1),host2(port2)"
|
||||
* @param channel MQ channel name
|
||||
* @param queueManager Queue Manager name
|
||||
* @param user Username for authentication
|
||||
* @param password Password for authentication
|
||||
* @param keystorePath Path to SSL keystore (can be null for non-SSL)
|
||||
* @param keystorePassword Password for SSL keystore
|
||||
* @param sslCipherSuite SSL cipher suite to use (e.g.,
|
||||
* "TLS_RSA_WITH_AES_256_CBC_SHA256")
|
||||
*/
|
||||
public IbmMqConnector(String connectionNameList, String channel, String queueManager, String user, String password,
|
||||
String keystorePath, String keystorePassword, String sslCipherSuite) {
|
||||
this.queueManager = queueManager;
|
||||
this.user = user;
|
||||
this.password = password;
|
||||
|
||||
try {
|
||||
MQConnectionFactory cf = new MQConnectionFactory();
|
||||
try {
|
||||
if (keystorePath != null && !keystorePath.isBlank()) {
|
||||
System.setProperty("javax.net.ssl.keyStore", keystorePath);
|
||||
System.setProperty("javax.net.ssl.trustStore", keystorePath);
|
||||
if (keystorePassword != null) {
|
||||
System.setProperty("javax.net.ssl.keyStorePassword", keystorePassword);
|
||||
System.setProperty("javax.net.ssl.trustStorePassword", keystorePassword);
|
||||
}
|
||||
}
|
||||
|
||||
// Set connection name list for multi-instance QMGR
|
||||
cf.setChannel(channel);
|
||||
cf.setQueueManager(queueManager);
|
||||
cf.setConnectionNameList(connectionNameList);
|
||||
cf.setTransportType(WMQConstants.WMQ_CM_CLIENT);
|
||||
connectionFactory = new MQConnectionFactory();
|
||||
connectionFactory.setConnectionNameList(connectionNameList);
|
||||
connectionFactory.setQueueManager(queueManager);
|
||||
connectionFactory.setChannel(channel);
|
||||
connectionFactory.setTransportType(WMQConstants.WMQ_CM_CLIENT);
|
||||
if (user != null && !user.isBlank()) {
|
||||
connectionFactory.setStringProperty(WMQConstants.USERID, user);
|
||||
}
|
||||
if (password != null && !password.isBlank()) {
|
||||
connectionFactory.setStringProperty(WMQConstants.PASSWORD, password);
|
||||
}
|
||||
|
||||
// Set authentication
|
||||
cf.setStringProperty(WMQConstants.USERID, user);
|
||||
cf.setStringProperty(WMQConstants.PASSWORD, password);
|
||||
if (sslCipherSuite != null && !sslCipherSuite.isBlank()) {
|
||||
connectionFactory.setSSLCipherSuite(sslCipherSuite);
|
||||
}
|
||||
|
||||
// SSL configuration if keystore is provided
|
||||
if (StringUtils.isNotBlank(keystorePath)) {
|
||||
System.setProperty("javax.net.ssl.keyStore", keystorePath);
|
||||
System.setProperty("javax.net.ssl.trustStore", keystorePath);
|
||||
if (StringUtils.isNotBlank(keystorePassword)) {
|
||||
System.setProperty("javax.net.ssl.keyStorePassword", keystorePassword);
|
||||
System.setProperty("javax.net.ssl.trustStorePassword", keystorePassword);
|
||||
}
|
||||
if (StringUtils.isNotBlank(sslCipherSuite)) {
|
||||
cf.setSSLCipherSuite(sslCipherSuite);
|
||||
}
|
||||
}
|
||||
// Initialize JMS context
|
||||
connect();
|
||||
|
||||
this.connectionFactory = cf;
|
||||
} catch (Exception e) {
|
||||
throw new MessagingConnectionException("Failed to create IBM MQ connection to " + queueManager, e);
|
||||
}
|
||||
}
|
||||
|
||||
// Initialize JMS context
|
||||
connect();
|
||||
/**
|
||||
* Connect to IBM MQ.
|
||||
*/
|
||||
private void connect() {
|
||||
try {
|
||||
this.jmsContext = connectionFactory.createContext(user, password, JMSContext.AUTO_ACKNOWLEDGE);
|
||||
this.jmsContext.start();
|
||||
LOG.info("Connected to IBM MQ: {}", queueManager);
|
||||
} catch (Exception e) {
|
||||
throw new MessagingConnectionException(
|
||||
"Failed to connect to IBM MQ: " + queueManager + " - " + e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
|
||||
} catch (Exception e) {
|
||||
throw new MessagingConnectionException(
|
||||
"Failed to create IBM MQ connection to " + queueManager, e);
|
||||
}
|
||||
}
|
||||
/**
|
||||
* Send a JSON or XML message as TextMessage.
|
||||
*/
|
||||
private void sendTextMessage(String queueName, String payload, Map<String, String> properties) {
|
||||
javax.jms.Queue queue = getQueue(queueName);
|
||||
|
||||
/**
|
||||
* Connect to IBM MQ.
|
||||
*/
|
||||
private void connect() {
|
||||
try {
|
||||
this.jmsContext = connectionFactory.createContext();
|
||||
this.jmsContext.start();
|
||||
LOG.info("Connected to IBM MQ: {}", queueManager);
|
||||
} catch (Exception e) {
|
||||
throw new MessagingConnectionException(
|
||||
"Failed to connect to IBM MQ: " + queueManager + " - " + e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
TextMessage message = jmsContext.createTextMessage(payload);
|
||||
|
||||
/**
|
||||
* Send a JSON or XML message as TextMessage.
|
||||
*/
|
||||
private void sendTextMessage(String queueName, String payload, Map<String, String> properties) {
|
||||
javax.jms.Queue queue = getQueue(queueName);
|
||||
// Set JMS properties
|
||||
if (properties != null) {
|
||||
for (Map.Entry<String, String> entry : properties.entrySet()) {
|
||||
try {
|
||||
message.setStringProperty(entry.getKey(), entry.getValue());
|
||||
} catch (JMSException e) {
|
||||
LOG.warn("Failed to set property: {}", entry.getKey(), e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
TextMessage message = jmsContext.createTextMessage(payload);
|
||||
try {
|
||||
jmsContext.createProducer().send(queue, message);
|
||||
} catch (RuntimeException e) {
|
||||
throw new MessagingDestinationException("Failed to send message to queue: " + queueName, e);
|
||||
}
|
||||
LOG.debug("Sent JSON/XML message to queue: {}", queueName);
|
||||
}
|
||||
|
||||
// Set JMS properties
|
||||
if (properties != null) {
|
||||
for (Map.Entry<String, String> entry : properties.entrySet()) {
|
||||
try {
|
||||
message.setStringProperty(entry.getKey(), entry.getValue());
|
||||
} catch (JMSException e) {
|
||||
LOG.warn("Failed to set property: {}", entry.getKey(), e);
|
||||
}
|
||||
}
|
||||
}
|
||||
/**
|
||||
* Send a message as BytesMessage with specific encoding and CCSID.
|
||||
*/
|
||||
private void sendBytesMessage(String queueName, String payload, Charset charset, int ccsid,
|
||||
Map<String, String> properties) {
|
||||
javax.jms.Queue queue = getQueue(queueName);
|
||||
|
||||
try {
|
||||
jmsContext.createProducer().send(queue, message);
|
||||
} catch (RuntimeException e) {
|
||||
throw new MessagingDestinationException("Failed to send message to queue: " + queueName, e);
|
||||
}
|
||||
LOG.debug("Sent JSON/XML message to queue: {}", queueName);
|
||||
}
|
||||
BytesMessage message = jmsContext.createBytesMessage();
|
||||
|
||||
/**
|
||||
* Send a message as BytesMessage with specific encoding and CCSID.
|
||||
*/
|
||||
private void sendBytesMessage(String queueName, String payload, Charset charset,
|
||||
int ccsid, Map<String, String> properties) {
|
||||
javax.jms.Queue queue = getQueue(queueName);
|
||||
// Convert payload to bytes using specified charset
|
||||
byte[] bytes = payload.getBytes(charset);
|
||||
try {
|
||||
message.writeBytes(bytes);
|
||||
message.setIntProperty("CCSID", ccsid);
|
||||
} catch (JMSException e) {
|
||||
throw new MessagingDestinationException("Failed to create bytes message", e);
|
||||
}
|
||||
|
||||
BytesMessage message = jmsContext.createBytesMessage();
|
||||
// Set JMS properties
|
||||
if (properties != null) {
|
||||
for (Map.Entry<String, String> entry : properties.entrySet()) {
|
||||
try {
|
||||
message.setStringProperty(entry.getKey(), entry.getValue());
|
||||
} catch (JMSException e) {
|
||||
LOG.warn("Failed to set property: {}", entry.getKey(), e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Convert payload to bytes using specified charset
|
||||
byte[] bytes = payload.getBytes(charset);
|
||||
try {
|
||||
message.writeBytes(bytes);
|
||||
message.setIntProperty("CCSID", ccsid);
|
||||
} catch (JMSException e) {
|
||||
throw new MessagingDestinationException("Failed to create bytes message", e);
|
||||
}
|
||||
try {
|
||||
jmsContext.createProducer().send(queue, message);
|
||||
} catch (RuntimeException e) {
|
||||
throw new MessagingDestinationException("Failed to send message to queue: " + queueName, e);
|
||||
}
|
||||
LOG.debug("Sent {} message to queue: {}", charset, queueName);
|
||||
}
|
||||
|
||||
// Set JMS properties
|
||||
if (properties != null) {
|
||||
for (Map.Entry<String, String> entry : properties.entrySet()) {
|
||||
try {
|
||||
message.setStringProperty(entry.getKey(), entry.getValue());
|
||||
} catch (JMSException e) {
|
||||
LOG.warn("Failed to set property: {}", entry.getKey(), e);
|
||||
}
|
||||
}
|
||||
}
|
||||
/**
|
||||
* Send a message to a queue with specified format.
|
||||
*
|
||||
* @param queueName Queue name
|
||||
* @param payload Message payload
|
||||
* @param format Message format (JSON, XML, EBCDIC_870, UTF8_1208)
|
||||
* @param properties JMS properties to set
|
||||
*/
|
||||
public void send(String queueName, String payload, MqMessageFormat format, Map<String, String> properties) {
|
||||
switch (format) {
|
||||
case JSON, XML -> sendTextMessage(queueName, payload, properties);
|
||||
case EBCDIC_870 -> sendBytesMessage(queueName, payload, EBCDIC_870, 870, properties);
|
||||
case UTF8_1208 -> sendBytesMessage(queueName, payload, UTF_8, 1208, properties);
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
jmsContext.createProducer().send(queue, message);
|
||||
} catch (RuntimeException e) {
|
||||
throw new MessagingDestinationException("Failed to send message to queue: " + queueName, e);
|
||||
}
|
||||
LOG.debug("Sent {} message to queue: {}", charset, queueName);
|
||||
}
|
||||
/**
|
||||
* Receive a message from a queue with timeout.
|
||||
*
|
||||
* @param queueName Queue name
|
||||
* @param messageSelector JMS message selector (optional)
|
||||
* @param format Expected message format
|
||||
* @param timeout Timeout duration
|
||||
* @return Received message
|
||||
*/
|
||||
public ReceivedMessage receive(String queueName, String messageSelector, MqMessageFormat format,
|
||||
java.time.Duration timeout) {
|
||||
long timeoutMs = timeout.toMillis();
|
||||
|
||||
/**
|
||||
* Send a message to a queue with specified format.
|
||||
*
|
||||
* @param queueName Queue name
|
||||
* @param payload Message payload
|
||||
* @param format Message format (JSON, XML, EBCDIC_870, UTF8_1208)
|
||||
* @param properties JMS properties to set
|
||||
*/
|
||||
public void send(String queueName, String payload, MqMessageFormat format,
|
||||
Map<String, String> properties) {
|
||||
switch (format) {
|
||||
case JSON, XML -> sendTextMessage(queueName, payload, properties);
|
||||
case EBCDIC_870 -> sendBytesMessage(queueName, payload, EBCDIC_870, 870, properties);
|
||||
case UTF8_1208 -> sendBytesMessage(queueName, payload, UTF_8, 1208, properties);
|
||||
}
|
||||
}
|
||||
javax.jms.Queue queue = getQueue(queueName);
|
||||
MessageConsumer consumer = (MessageConsumer) (messageSelector == null || messageSelector.isBlank()
|
||||
? jmsContext.createConsumer(queue)
|
||||
: jmsContext.createConsumer(queue, messageSelector));
|
||||
|
||||
/**
|
||||
* Receive a message from a queue with timeout.
|
||||
*
|
||||
* @param queueName Queue name
|
||||
* @param messageSelector JMS message selector (optional)
|
||||
* @param format Expected message format
|
||||
* @param timeout Timeout duration
|
||||
* @return Received message
|
||||
*/
|
||||
public ReceivedMessage receive(String queueName, String messageSelector,
|
||||
MqMessageFormat format, java.time.Duration timeout) {
|
||||
long timeoutMs = timeout.toMillis();
|
||||
AtomicBoolean messageFound = new AtomicBoolean(false);
|
||||
ReceivedMessage received = null;
|
||||
|
||||
javax.jms.Queue queue = getQueue(queueName);
|
||||
MessageConsumer consumer = (MessageConsumer) (messageSelector == null || messageSelector.isBlank()
|
||||
? jmsContext.createConsumer(queue)
|
||||
: jmsContext.createConsumer(queue, messageSelector));
|
||||
long pollInterval = DEFAULT_POLL_INTERVAL_MS;
|
||||
long remainingTime = timeoutMs;
|
||||
|
||||
AtomicBoolean messageFound = new AtomicBoolean(false);
|
||||
ReceivedMessage received = null;
|
||||
try {
|
||||
while (remainingTime > 0 && !messageFound.get()) {
|
||||
Message message = consumer.receive(remainingTime);
|
||||
|
||||
long pollInterval = DEFAULT_POLL_INTERVAL_MS;
|
||||
long remainingTime = timeoutMs;
|
||||
if (message != null) {
|
||||
received = decodeMessage(message, queueName, format);
|
||||
messageFound.set(true);
|
||||
} else {
|
||||
// Exponential backoff
|
||||
pollInterval = Math.min(pollInterval * 2, DEFAULT_MAX_POLL_INTERVAL_MS);
|
||||
remainingTime -= pollInterval;
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
while (remainingTime > 0 && !messageFound.get()) {
|
||||
Message message = consumer.receive(remainingTime);
|
||||
if (received == null) {
|
||||
throw new MessagingTimeoutException("No message matching filter found on queue '" + queueName
|
||||
+ "' within " + timeout.toMillis() + "ms");
|
||||
}
|
||||
|
||||
if (message != null) {
|
||||
received = decodeMessage(message, queueName, format);
|
||||
messageFound.set(true);
|
||||
} else {
|
||||
// Exponential backoff
|
||||
pollInterval = Math.min(pollInterval * 2, DEFAULT_MAX_POLL_INTERVAL_MS);
|
||||
remainingTime -= pollInterval;
|
||||
}
|
||||
}
|
||||
return received;
|
||||
|
||||
if (received == null) {
|
||||
throw new MessagingTimeoutException(
|
||||
"No message matching filter found on queue '" + queueName +
|
||||
"' within " + timeout.toMillis() + "ms");
|
||||
}
|
||||
} catch (MessagingTimeoutException e) {
|
||||
throw e;
|
||||
} catch (Exception e) {
|
||||
throw new MessagingDestinationException("Failed to receive message from queue: " + queueName, e);
|
||||
} finally {
|
||||
try {
|
||||
consumer.close();
|
||||
} catch (JMSException e) {
|
||||
LOG.warn("Failed to close consumer", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return received;
|
||||
/**
|
||||
* Browse a queue (non-destructive read).
|
||||
*
|
||||
* @param queueName Queue name
|
||||
* @param messageSelector JMS message selector (optional)
|
||||
* @param format Expected message format
|
||||
* @param maxMessages Maximum number of messages to browse
|
||||
* @return List of received messages
|
||||
*/
|
||||
public List<ReceivedMessage> browse(String queueName, String messageSelector, MqMessageFormat format,
|
||||
int maxMessages) {
|
||||
List<ReceivedMessage> messages = new ArrayList<>();
|
||||
|
||||
} catch (MessagingTimeoutException e) {
|
||||
throw e;
|
||||
} catch (Exception e) {
|
||||
throw new MessagingDestinationException("Failed to receive message from queue: " + queueName, e);
|
||||
} finally {
|
||||
try {
|
||||
consumer.close();
|
||||
} catch (JMSException e) {
|
||||
LOG.warn("Failed to close consumer", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
javax.jms.Queue queue = getQueue(queueName);
|
||||
MessageConsumer consumer = (MessageConsumer) (messageSelector == null || messageSelector.isBlank()
|
||||
? jmsContext.createConsumer(queue)
|
||||
: jmsContext.createConsumer(queue, messageSelector));
|
||||
|
||||
/**
|
||||
* Browse a queue (non-destructive read).
|
||||
*
|
||||
* @param queueName Queue name
|
||||
* @param messageSelector JMS message selector (optional)
|
||||
* @param format Expected message format
|
||||
* @param maxMessages Maximum number of messages to browse
|
||||
* @return List of received messages
|
||||
*/
|
||||
public List<ReceivedMessage> browse(String queueName, String messageSelector,
|
||||
MqMessageFormat format, int maxMessages) {
|
||||
List<ReceivedMessage> messages = new ArrayList<>();
|
||||
int count = 0;
|
||||
try {
|
||||
while (count < maxMessages) {
|
||||
Message message = consumer.receiveNoWait();
|
||||
|
||||
javax.jms.Queue queue = getQueue(queueName);
|
||||
MessageConsumer consumer = (MessageConsumer) (messageSelector == null || messageSelector.isBlank()
|
||||
? jmsContext.createConsumer(queue)
|
||||
: jmsContext.createConsumer(queue, messageSelector));
|
||||
if (message == null) {
|
||||
break;
|
||||
}
|
||||
|
||||
int count = 0;
|
||||
try {
|
||||
while (count < maxMessages) {
|
||||
Message message = consumer.receiveNoWait();
|
||||
ReceivedMessage received = decodeMessage(message, queueName, format);
|
||||
messages.add(received);
|
||||
count++;
|
||||
}
|
||||
|
||||
if (message == null) {
|
||||
break;
|
||||
}
|
||||
return messages;
|
||||
|
||||
ReceivedMessage received = decodeMessage(message, queueName, format);
|
||||
messages.add(received);
|
||||
count++;
|
||||
}
|
||||
} catch (Exception e) {
|
||||
throw new MessagingDestinationException("Failed to browse queue: " + queueName, e);
|
||||
} finally {
|
||||
try {
|
||||
consumer.close();
|
||||
} catch (JMSException e) {
|
||||
LOG.warn("Failed to close consumer", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return messages;
|
||||
/**
|
||||
* Decode a JMS message to ReceivedMessage.
|
||||
*/
|
||||
private ReceivedMessage decodeMessage(Message jmsMessage, String queueName, MqMessageFormat format) {
|
||||
long timestamp;
|
||||
try {
|
||||
timestamp = jmsMessage.getJMSTimestamp();
|
||||
} catch (JMSException e) {
|
||||
timestamp = System.currentTimeMillis();
|
||||
}
|
||||
if (timestamp == 0) {
|
||||
timestamp = System.currentTimeMillis();
|
||||
}
|
||||
|
||||
} catch (Exception e) {
|
||||
throw new MessagingDestinationException("Failed to browse queue: " + queueName, e);
|
||||
} finally {
|
||||
try {
|
||||
consumer.close();
|
||||
} catch (JMSException e) {
|
||||
LOG.warn("Failed to close consumer", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
String body;
|
||||
MessageContentType contentType;
|
||||
Map<String, String> headers = new HashMap<>();
|
||||
|
||||
/**
|
||||
* Decode a JMS message to ReceivedMessage.
|
||||
*/
|
||||
private ReceivedMessage decodeMessage(Message jmsMessage, String queueName, MqMessageFormat format) {
|
||||
long timestamp;
|
||||
try {
|
||||
timestamp = jmsMessage.getJMSTimestamp();
|
||||
} catch (JMSException e) {
|
||||
timestamp = System.currentTimeMillis();
|
||||
}
|
||||
if (timestamp == 0) {
|
||||
timestamp = System.currentTimeMillis();
|
||||
}
|
||||
// Extract JMS properties as headers
|
||||
extractJmsProperties(jmsMessage, headers);
|
||||
|
||||
String body;
|
||||
MessageContentType contentType;
|
||||
Map<String, String> headers = new HashMap<>();
|
||||
if (jmsMessage instanceof TextMessage textMessage) {
|
||||
try {
|
||||
body = textMessage.getText();
|
||||
} catch (JMSException e) {
|
||||
throw new RuntimeException("Failed to read text message body", e);
|
||||
}
|
||||
contentType = switch (format) {
|
||||
case XML -> MessageContentType.XML;
|
||||
default -> MessageContentType.JSON;
|
||||
};
|
||||
} else if (jmsMessage instanceof BytesMessage bytesMessage) {
|
||||
int ccsid;
|
||||
try {
|
||||
ccsid = bytesMessage.getIntProperty("CCSID");
|
||||
} catch (JMSException e) {
|
||||
ccsid = 1208; // default UTF-8
|
||||
}
|
||||
body = decodeBytesMessage(bytesMessage, ccsid);
|
||||
contentType = MessageContentType.RAW_TEXT;
|
||||
} else {
|
||||
try {
|
||||
throw new IllegalArgumentException("Unsupported message type: " + jmsMessage.getJMSType());
|
||||
} catch (JMSException e) {
|
||||
throw new IllegalArgumentException("Unsupported message type", e);
|
||||
}
|
||||
}
|
||||
|
||||
// Extract JMS properties as headers
|
||||
extractJmsProperties(jmsMessage, headers);
|
||||
return new ReceivedMessage(body, contentType, headers, timestamp, queueName, null);
|
||||
}
|
||||
|
||||
if (jmsMessage instanceof TextMessage textMessage) {
|
||||
try {
|
||||
body = textMessage.getText();
|
||||
} catch (JMSException e) {
|
||||
throw new RuntimeException("Failed to read text message body", e);
|
||||
}
|
||||
contentType = switch (format) {
|
||||
case XML -> MessageContentType.XML;
|
||||
default -> MessageContentType.JSON;
|
||||
};
|
||||
} else if (jmsMessage instanceof BytesMessage bytesMessage) {
|
||||
int ccsid;
|
||||
try {
|
||||
ccsid = bytesMessage.getIntProperty("CCSID");
|
||||
} catch (JMSException e) {
|
||||
ccsid = 1208; // default UTF-8
|
||||
}
|
||||
body = decodeBytesMessage(bytesMessage, ccsid);
|
||||
contentType = MessageContentType.RAW_TEXT;
|
||||
} else {
|
||||
try {
|
||||
throw new IllegalArgumentException("Unsupported message type: " + jmsMessage.getJMSType());
|
||||
} catch (JMSException e) {
|
||||
throw new IllegalArgumentException("Unsupported message type", e);
|
||||
}
|
||||
}
|
||||
/**
|
||||
* Decode BytesMessage body based on CCSID.
|
||||
*/
|
||||
private String decodeBytesMessage(BytesMessage bytesMessage, int ccsid) {
|
||||
try {
|
||||
long bodyLength;
|
||||
try {
|
||||
bodyLength = bytesMessage.getBodyLength();
|
||||
} catch (JMSException e) {
|
||||
throw new RuntimeException("Failed to get message body length", e);
|
||||
}
|
||||
byte[] data = new byte[(int) bodyLength];
|
||||
bytesMessage.readBytes(data);
|
||||
|
||||
return new ReceivedMessage(body, contentType, headers, timestamp, queueName, null);
|
||||
}
|
||||
Charset charset = switch (ccsid) {
|
||||
case 870 -> EBCDIC_870;
|
||||
case 1208 -> UTF_8;
|
||||
default -> UTF_8;
|
||||
};
|
||||
|
||||
/**
|
||||
* Decode BytesMessage body based on CCSID.
|
||||
*/
|
||||
private String decodeBytesMessage(BytesMessage bytesMessage, int ccsid) {
|
||||
try {
|
||||
long bodyLength;
|
||||
try {
|
||||
bodyLength = bytesMessage.getBodyLength();
|
||||
} catch (JMSException e) {
|
||||
throw new RuntimeException("Failed to get message body length", e);
|
||||
}
|
||||
byte[] data = new byte[(int) bodyLength];
|
||||
bytesMessage.readBytes(data);
|
||||
return new String(data, charset);
|
||||
} catch (JMSException e) {
|
||||
throw new RuntimeException("Failed to read BytesMessage body", e);
|
||||
}
|
||||
}
|
||||
|
||||
Charset charset = switch (ccsid) {
|
||||
case 870 -> EBCDIC_870;
|
||||
case 1208 -> UTF_8;
|
||||
default -> UTF_8;
|
||||
};
|
||||
/**
|
||||
* Extract JMS properties as headers.
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
private void extractJmsProperties(Message message, Map<String, String> headers) {
|
||||
try {
|
||||
// Common JMS headers
|
||||
headers.put("JMSMessageID", message.getJMSMessageID());
|
||||
try {
|
||||
headers.put("JMSType", message.getJMSType() != null ? message.getJMSType() : "");
|
||||
} catch (JMSException e) {
|
||||
headers.put("JMSType", "");
|
||||
}
|
||||
try {
|
||||
headers.put("JMSDestination",
|
||||
message.getJMSDestination() != null ? message.getJMSDestination().toString() : "");
|
||||
} catch (JMSException e) {
|
||||
headers.put("JMSDestination", "");
|
||||
}
|
||||
try {
|
||||
headers.put("JMSDeliveryMode", String.valueOf(message.getJMSDeliveryMode()));
|
||||
} catch (JMSException e) {
|
||||
headers.put("JMSDeliveryMode", "");
|
||||
}
|
||||
try {
|
||||
headers.put("JMSPriority", String.valueOf(message.getJMSPriority()));
|
||||
} catch (JMSException e) {
|
||||
headers.put("JMSPriority", "");
|
||||
}
|
||||
try {
|
||||
headers.put("JMSTimestamp", String.valueOf(message.getJMSTimestamp()));
|
||||
} catch (JMSException e) {
|
||||
headers.put("JMSTimestamp", "");
|
||||
}
|
||||
|
||||
return new String(data, charset);
|
||||
} catch (JMSException e) {
|
||||
throw new RuntimeException("Failed to read BytesMessage body", e);
|
||||
}
|
||||
}
|
||||
// Extract custom properties
|
||||
Enumeration<String> propertyNames = (Enumeration<String>) message.getPropertyNames();
|
||||
while (propertyNames.hasMoreElements()) {
|
||||
String propName = propertyNames.nextElement();
|
||||
Object propValue = message.getObjectProperty(propName);
|
||||
if (propValue != null) {
|
||||
headers.put(propName, propValue.toString());
|
||||
}
|
||||
}
|
||||
} catch (JMSException e) {
|
||||
LOG.warn("Failed to extract JMS properties", e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Extract JMS properties as headers.
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
private void extractJmsProperties(Message message, Map<String, String> headers) {
|
||||
try {
|
||||
// Common JMS headers
|
||||
headers.put("JMSMessageID", message.getJMSMessageID());
|
||||
try {
|
||||
headers.put("JMSType", message.getJMSType() != null ? message.getJMSType() : "");
|
||||
} catch (JMSException e) {
|
||||
headers.put("JMSType", "");
|
||||
}
|
||||
try {
|
||||
headers.put("JMSDestination", message.getJMSDestination() != null ?
|
||||
message.getJMSDestination().toString() : "");
|
||||
} catch (JMSException e) {
|
||||
headers.put("JMSDestination", "");
|
||||
}
|
||||
try {
|
||||
headers.put("JMSDeliveryMode", String.valueOf(message.getJMSDeliveryMode()));
|
||||
} catch (JMSException e) {
|
||||
headers.put("JMSDeliveryMode", "");
|
||||
}
|
||||
try {
|
||||
headers.put("JMSPriority", String.valueOf(message.getJMSPriority()));
|
||||
} catch (JMSException e) {
|
||||
headers.put("JMSPriority", "");
|
||||
}
|
||||
try {
|
||||
headers.put("JMSTimestamp", String.valueOf(message.getJMSTimestamp()));
|
||||
} catch (JMSException e) {
|
||||
headers.put("JMSTimestamp", "");
|
||||
}
|
||||
/**
|
||||
* Get Queue object from queue name.
|
||||
*/
|
||||
private javax.jms.Queue getQueue(String queueName) {
|
||||
return jmsContext.createQueue(queueName);
|
||||
}
|
||||
|
||||
// Extract custom properties
|
||||
Enumeration<String> propertyNames = (Enumeration<String>) message.getPropertyNames();
|
||||
while (propertyNames.hasMoreElements()) {
|
||||
String propName = propertyNames.nextElement();
|
||||
Object propValue = message.getObjectProperty(propName);
|
||||
if (propValue != null) {
|
||||
headers.put(propName, propValue.toString());
|
||||
}
|
||||
}
|
||||
} catch (JMSException e) {
|
||||
LOG.warn("Failed to extract JMS properties", e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get Queue object from queue name.
|
||||
*/
|
||||
private javax.jms.Queue getQueue(String queueName) {
|
||||
return jmsContext.createQueue(queueName);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
if (jmsContext != null) {
|
||||
try {
|
||||
jmsContext.close();
|
||||
LOG.info("Closed connection to IBM MQ: {}", queueManager);
|
||||
} catch (Exception e) {
|
||||
LOG.error("Failed to close IBM MQ connection", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
@Override
|
||||
public void close() {
|
||||
if (jmsContext != null) {
|
||||
try {
|
||||
jmsContext.close();
|
||||
LOG.info("Closed connection to IBM MQ: {}", queueManager);
|
||||
} catch (Exception e) {
|
||||
LOG.error("Failed to close IBM MQ connection", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -51,8 +51,8 @@ public class ImqFirstVisionEndpoint implements Endpoint {
|
||||
//Credentials credentials = loadCredentialsFromVault(vaultPath);
|
||||
|
||||
// SSL configuration (optional)
|
||||
String keystorePath = null;
|
||||
String keystorePassword = null;
|
||||
String keystorePath = "/home/kamma/aa/mq-docker/truststore.jks";
|
||||
String keystorePassword = "changeit";
|
||||
|
||||
try {
|
||||
this.connector = new IbmMqConnector(
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user