Rinjae
606 lines
18 KiB
Java
606 lines
18 KiB
Java
package com.eactive.eai.flowcontrol.jms;
|
|
|
|
import java.util.HashMap;
|
|
import java.util.Properties;
|
|
|
|
import org.apache.commons.lang.SerializationUtils;
|
|
|
|
import com.eactive.eai.common.exception.ExceptionUtil;
|
|
import com.eactive.eai.common.logger.EAILogSender;
|
|
import com.eactive.eai.common.message.EAIMessage;
|
|
import com.eactive.eai.common.property.PropManager;
|
|
import com.eactive.eai.common.routing.ElinkESBProcessProxy;
|
|
import com.eactive.eai.common.routing.RouteKeys;
|
|
import com.eactive.eai.common.routing.RoutingManager;
|
|
import com.eactive.eai.common.routing.RoutingVO;
|
|
import com.eactive.eai.common.util.JMSSender;
|
|
import com.eactive.eai.common.util.Logger;
|
|
import com.eactive.eai.common.util.MessageUtil;
|
|
import com.ext.eai.common.stdmessage.STDMessageKeys;
|
|
import com.rabbitmq.client.Channel;
|
|
import com.rabbitmq.client.Connection;
|
|
import com.rabbitmq.client.ConnectionFactory;
|
|
import com.rabbitmq.client.GetResponse;
|
|
|
|
public class FCRabbitMqReceiver {
|
|
private static Logger logger = Logger.getLogger(Logger.LOGGER_DEFAULT);
|
|
|
|
private String receiverName;
|
|
private String queueName;
|
|
private String errorQueueName;
|
|
private int receiverCount;
|
|
private int errorDelayTimeMs;
|
|
private int retryCount;
|
|
private String messageSelector = "";
|
|
private String description;
|
|
|
|
// ------------------------------------
|
|
// Used for Test
|
|
// ------------------------------------
|
|
public static final boolean isSingleApplication = false;
|
|
|
|
// ------------------------------------
|
|
// Objects needed for RabbitMQ clients
|
|
// ------------------------------------
|
|
private ConnectionFactory factory;
|
|
private Connection[] connection;
|
|
private Channel[] channel;
|
|
private ReceiverObject[] receiverObj = null;
|
|
|
|
// private boolean isTx = false;
|
|
// private static final DecimalFormat dFormat = new DecimalFormat("#0.0");
|
|
|
|
// ------------------------------------
|
|
// Thread Runtime Info
|
|
// ------------------------------------
|
|
private Thread[] receiverThread = null;
|
|
private boolean[] threadDowned = null;
|
|
public boolean allThreadDowned = true;
|
|
|
|
// private static String instid = "11";
|
|
static {
|
|
// String serverName = System.getProperty(Keys.WLI_SERVER_KEY);
|
|
// if(serverName != null && serverName.length() > 2) {
|
|
// instid = serverName.substring(serverName.length()-2, serverName.length());
|
|
// }
|
|
}
|
|
|
|
public String toString() {
|
|
StringBuilder sb = new StringBuilder();
|
|
sb.append("[ FCRabbitMqReceiver INFO ]");
|
|
sb.append("\nReveciver Name = " + this.receiverName);
|
|
sb.append(", Queue Name = " + this.queueName);
|
|
sb.append(", Error Queue Name = " + this.errorQueueName);
|
|
sb.append(", Recever Count = " + this.receiverCount);
|
|
sb.append(", Delay Time(ms) = " + this.errorDelayTimeMs);
|
|
sb.append(", Retry Count = " + this.retryCount);
|
|
sb.append(", Selector = " + this.messageSelector);
|
|
sb.append(", Description = " + this.description);
|
|
|
|
return sb.toString();
|
|
}
|
|
|
|
public FCRabbitMqReceiver(String rcvName, String qname, String errorqname, int rcvCount, int delayMs, int retryCnt,
|
|
boolean isTx, int ackMode, String msgSelector, String descr) throws Exception {
|
|
|
|
queueName = qname;
|
|
errorQueueName = errorqname;
|
|
receiverName = rcvName;
|
|
receiverCount = rcvCount;
|
|
errorDelayTimeMs = delayMs;
|
|
retryCount = retryCnt;
|
|
messageSelector = msgSelector;
|
|
description = descr;
|
|
|
|
// trim space
|
|
if (queueName != null)
|
|
queueName = queueName.trim();
|
|
if (errorQueueName != null)
|
|
errorQueueName = errorQueueName.trim();
|
|
if (receiverName != null)
|
|
receiverName = receiverName.trim();
|
|
if (messageSelector != null)
|
|
messageSelector = messageSelector.trim();
|
|
|
|
threadDowned = new boolean[receiverCount];
|
|
connection = new Connection[receiverCount];
|
|
channel = new Channel[receiverCount];
|
|
receiverObj = new ReceiverObject[receiverCount];
|
|
receiverThread = new Thread[receiverCount];
|
|
|
|
String clientId = receiverName;
|
|
|
|
String connectionUrl = "";
|
|
try {
|
|
if (isSingleApplication) {
|
|
if (logger.isDebug())
|
|
logger.debug("Local Application Test");
|
|
connectionUrl = "amqp://guest:guest@localhost:5672";
|
|
} else {
|
|
PropManager propManager = PropManager.getInstance();
|
|
connectionUrl = propManager.getProperty(RouteKeys.GROUP_NAME,
|
|
RouteKeys.ROUTING_QUEUE_CONNECTION_FACTORY);
|
|
|
|
}
|
|
} catch (Exception e) {
|
|
if (logger.isError())
|
|
logger.error("FCQueueReceiver: get InitialContext Error : ", e);
|
|
throw e;
|
|
}
|
|
|
|
if (logger.isInfo())
|
|
logger.info("RabbitMQ Connection : " + connectionUrl);
|
|
|
|
try {
|
|
// Create the connections
|
|
for (int i = 0; i < receiverCount; i++) {
|
|
factory = new ConnectionFactory();
|
|
factory.setUri(connectionUrl);
|
|
// create each connection
|
|
connection[i] = factory.newConnection();
|
|
|
|
// clientId = receiverName + instid + "-" + i;
|
|
// clientId = receiverName + "-" + i;
|
|
|
|
channel[i] = connection[i].createChannel();
|
|
channel[i].queueDeclare(queueName, false, false, false, null);
|
|
|
|
receiverObj[i] = new ReceiverObject(rcvName, i, channel[i], queueName, this.messageSelector,
|
|
this.errorQueueName, this.errorDelayTimeMs, this.retryCount, connectionUrl);
|
|
receiverObj[i].setName(clientId);
|
|
if (logger.isDebug())
|
|
logger.debug("[" + new java.util.Date() + "] Receiver Thread : [" + clientId + "] " + queueName);
|
|
receiverThread[i] = new Thread(receiverObj[i]);
|
|
receiverThread[i].setDaemon(true);// avoids thread leaks if
|
|
}
|
|
} catch (Exception e) {
|
|
if (logger.isError())
|
|
logger.error(" Exception while in constructor : ", e);
|
|
close();
|
|
throw e;
|
|
}
|
|
|
|
}
|
|
|
|
public synchronized void startReceiving() throws Exception {
|
|
|
|
if (!allThreadDowned) {
|
|
throw new Exception("Thread Started Already !");
|
|
}
|
|
|
|
try {
|
|
allThreadDowned = false;
|
|
} catch (Exception jmse) {
|
|
if (logger.isError()) {
|
|
logger.error("FCQueueReceiver: JMSException while in startReceiving() method : ", jmse);
|
|
}
|
|
close();
|
|
throw jmse;
|
|
}
|
|
|
|
}
|
|
|
|
public synchronized void clearReceived() {
|
|
// Clear Counter
|
|
for (int i = 0; i < receiverObj.length; i++) {
|
|
receiverObj[i].clearCount();
|
|
}
|
|
}
|
|
|
|
public synchronized void stopReceiving() {
|
|
// Cleanup each ReceiverObj (stops sending
|
|
for (int i = 0; i < receiverObj.length; i++) {
|
|
// stops each ReceiverObject
|
|
if (receiverObj[i] != null)
|
|
receiverObj[i].stopReceiver();
|
|
receiverObj[i].stopThread();
|
|
}
|
|
}
|
|
|
|
public String[][] getStatus() {
|
|
String[][] receiveStatus = new String[receiverObj.length][4];
|
|
for (int i = 0; i < receiverObj.length; i++) {
|
|
if (receiverObj[i] != null) {
|
|
receiveStatus[i][0] = receiverObj[i].getName();
|
|
receiveStatus[i][1] = "" + receiverObj[i].msgCounter;
|
|
receiveStatus[i][2] = "" + receiverObj[i].msgSuccessCounter;
|
|
receiveStatus[i][3] = "" + receiverObj[i].receiverStatus;
|
|
}
|
|
}
|
|
return receiveStatus;
|
|
}
|
|
|
|
public String[] getMessageCount() {
|
|
String[] receiveStatus = new String[2];
|
|
int receivedTotal = 0;
|
|
int successTotal = 0;
|
|
|
|
for (int i = 0; i < receiverObj.length; i++) {
|
|
if (receiverObj[i] != null) {
|
|
receivedTotal += receiverObj[i].msgCounter;
|
|
successTotal += receiverObj[i].msgSuccessCounter;
|
|
}
|
|
}
|
|
receiveStatus[0] = "" + receivedTotal;
|
|
receiveStatus[1] = "" + successTotal;
|
|
|
|
return receiveStatus;
|
|
}
|
|
|
|
/**
|
|
* Closes all connections which in turn stops the receivers and exits JVM
|
|
*/
|
|
public void close() {
|
|
for (int k = 0; k < channel.length; k++) {
|
|
try {
|
|
if (channel[k] != null) {
|
|
channel[k].close();
|
|
}
|
|
} catch (Exception ex) {
|
|
if (logger.isError()) {
|
|
logger.error("Exceptionchannel " + k + " message is " + ex.getMessage());
|
|
}
|
|
}
|
|
}
|
|
|
|
// Close all the connections before exiting the application
|
|
for (int i = 0; i < connection.length; i++) {
|
|
try {
|
|
if (connection[i] != null)
|
|
connection[i].close();
|
|
} catch (Exception ex) {
|
|
if (logger.isError()) {
|
|
logger.error("In exit(): Caught Exception trying to close connection(s)", ex);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// ////////////////////////////////////////////////////////////////////////
|
|
/**
|
|
* Inner class: ReceiverObject is an ASYNCHRONOUS receiver.
|
|
*/
|
|
// ////////////////////////////////////////////////////////////////////////
|
|
private class ReceiverObject extends Thread {
|
|
private String recvName = null;
|
|
|
|
private String queueName = null;
|
|
// Channel
|
|
private Channel channel = null;
|
|
|
|
private int id; // identifies the ReceiverObject
|
|
|
|
private int msgCounter = 0; // keeps track of messages
|
|
|
|
private int msgSuccessCounter = 0;
|
|
|
|
private int receiverStatus = 0; // 0 - wait, 1 - running, 2 - retry
|
|
|
|
private Object lockCounter = new Object();// used to synchronize
|
|
// updates of the message
|
|
// counter
|
|
|
|
private boolean keepRunning = false; // used to exit the sending loop
|
|
// in the run method
|
|
|
|
private String messageSelector = "";
|
|
|
|
private String errorQueueName = "";
|
|
|
|
private int errorDelayTimeMs = 0;
|
|
|
|
private int retryCount = 0;
|
|
|
|
private String queueConFactory = "";
|
|
|
|
/**
|
|
* Construct a ReceiverObject param id - the id number should be unique for each
|
|
* ReceiverObject param sess - the session object under which the receiver is
|
|
* created param q - the Queue object to receiver from
|
|
*/
|
|
public ReceiverObject(String recvName, int id, Channel channel, String queueName, String messageSelector,
|
|
String errorQueueName, int errorDelayMs, int retryCnt, String connectionUri) throws Exception {
|
|
init(recvName, id, channel, queueName, messageSelector, errorQueueName, errorDelayMs, retryCnt,
|
|
connectionUri);
|
|
}
|
|
|
|
public void init(String recvName, int sid, Channel channel, String queueName, String selector,
|
|
String errQueueName, int deplayMs, int retryCnt, String qConFactory) throws Exception {
|
|
this.recvName = recvName;
|
|
this.id = sid;
|
|
this.channel = channel;
|
|
this.queueName = queueName;
|
|
this.messageSelector = selector;
|
|
this.errorQueueName = errQueueName;
|
|
this.errorDelayTimeMs = deplayMs;
|
|
this.retryCount = retryCnt;
|
|
this.queueConFactory = qConFactory;
|
|
|
|
try {
|
|
logger.info("recvName=" + this.recvName + ",messageSelector=" + this.messageSelector);
|
|
} catch (Exception jmse) {
|
|
if (logger.isError())
|
|
logger.error(this.getName()
|
|
+ "] In ReceiverObject constructor: JMSException while setting up receiverObj[" + id + "]",
|
|
jmse);
|
|
close();
|
|
throw jmse;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Get number of message received since last get
|
|
*
|
|
* return totaSinceLastGet total number of messages received since the last get
|
|
*/
|
|
public int getMsgCount() {
|
|
synchronized (lockCounter) {
|
|
return msgCounter;
|
|
}
|
|
}
|
|
|
|
public void setMsgCount(int msgCounter) {
|
|
synchronized (lockCounter) {
|
|
this.msgCounter = msgCounter;
|
|
}
|
|
}
|
|
|
|
public void clearCount() {
|
|
synchronized (lockCounter) {
|
|
if (logger.isError()) {
|
|
logger.error(this.getName() + "] Clearing id: [" + id + "] clearing: " + msgCounter);
|
|
}
|
|
// reset counter
|
|
msgCounter = 0;
|
|
msgSuccessCounter = 0;
|
|
}
|
|
}
|
|
|
|
public void stopReceiver() {
|
|
// stop Receiver from sending which causes thread to exit run() method
|
|
// and thus finish.
|
|
keepRunning = false;
|
|
}
|
|
|
|
public void stopThread() {
|
|
// super.stop();
|
|
}
|
|
|
|
@Override
|
|
public void run() {
|
|
if (Thread.currentThread().getName().startsWith("Thread-")) {
|
|
Thread.currentThread().setName(queueName + "-" + Thread.currentThread().getName());
|
|
}
|
|
keepRunning = true;
|
|
threadDowned[id] = false;
|
|
EAIMessage eaiMessage = null;
|
|
|
|
while (keepRunning) {
|
|
try {
|
|
receiverStatus = 0;
|
|
|
|
GetResponse response = channel.basicGet(queueName, false);
|
|
|
|
if (response == null) {
|
|
Thread.sleep(500);
|
|
continue;
|
|
}
|
|
|
|
synchronized (lockCounter) {
|
|
receiverStatus = 1;
|
|
msgCounter++;
|
|
}
|
|
|
|
byte[] body = response.getBody();
|
|
|
|
try {
|
|
eaiMessage = (EAIMessage) SerializationUtils.deserialize(body);
|
|
} catch (Exception e) {
|
|
throw new RuntimeException(this.getName() + "] EAI메시지가 아닙니다. - " + e.getMessage());
|
|
}
|
|
|
|
// String guidLogPrefix = "RabbitMQ Receiver] GUID["+eaiMessage.getKbMsg().getKBHeader().getGuIdNo()+"] UUID["+eaiMessage.getSvcOgNo()+"] ";
|
|
String guidLogPrefix = "RabbitMQ Receiver] GUID["
|
|
+ eaiMessage.getMapper().getGuid(eaiMessage.getStandardMessage()) + "] UUID["
|
|
+ eaiMessage.getSvcOgNo() + "] ";
|
|
|
|
if (logger.isDebug()) {
|
|
logger.debug(this.getName() + "] receiverObj[" + id + "] Executed - " + eaiMessage.getEAISvcCd()
|
|
+ ":" + eaiMessage.getSvcOgNo());
|
|
}
|
|
|
|
RoutingVO routingVO = RoutingManager.getInstance().getRoutingVO(eaiMessage.getFlwCntlRtnNm());
|
|
String serviceURI = routingVO.getSyncRoutingPath();
|
|
boolean isLocal = true;
|
|
|
|
HashMap<String, Object> headers = (HashMap<String, Object>) response.getProps().getHeaders();
|
|
|
|
Properties prop = new Properties();
|
|
try {
|
|
if (headers != null) {
|
|
for (String key : headers.keySet()) {
|
|
String propName = key;
|
|
String propValue = (String) headers.get(propName);
|
|
prop.setProperty(propName, propValue);
|
|
}
|
|
}
|
|
} catch (Exception e) {
|
|
throw new RuntimeException(this.getName() + "] 프라퍼티 설정에러 - " + e.getMessage());
|
|
}
|
|
|
|
EAIMessage resEAIMessage = null;
|
|
|
|
channel.basicAck(response.getEnvelope().getDeliveryTag(), false);
|
|
|
|
try {
|
|
// EAI 거래처리 Error 이고 retry count가 있으면 회수만큼 재시도
|
|
// errorDelayTimeMs > 0 면 sleep
|
|
int retry = 1;
|
|
// retry count 만큼 반복한다.
|
|
// -1 : 무한대 retry
|
|
// 0 : 처리되지 않는다.
|
|
|
|
if (this.retryCount == 0) {
|
|
if (logger.isWarn()) {
|
|
logger.warn(this.getName() + "] SKIP Execute EAISvcCd (" + eaiMessage.getEAISvcCd()
|
|
+ ") Retry Count - " + this.retryCount);
|
|
}
|
|
continue; // 처리를 중단한다.
|
|
} else if (this.retryCount > 1) {
|
|
receiverStatus = 2;
|
|
}
|
|
|
|
while (((retry <= this.retryCount) || (this.retryCount == -1)) && keepRunning) {
|
|
resEAIMessage = ElinkESBProcessProxy.callElinkESBProcess(serviceURI, eaiMessage, isLocal,
|
|
prop);
|
|
|
|
if (resEAIMessage == null) {
|
|
if (logger.isWarn()) {
|
|
logger.warn(guidLogPrefix + this.getName()
|
|
+ "] response EAIMessage is NULL. SKIP Retry Count - " + this.retryCount);
|
|
}
|
|
break;
|
|
}
|
|
|
|
if (MessageUtil.checkRspErrCd(resEAIMessage.getRspErrCd())) {
|
|
break;
|
|
} else {
|
|
if ((this.retryCount > 0) && (retry <= this.retryCount)) {
|
|
if (logger.isWarn()) {
|
|
logger.warn(guidLogPrefix + this.getName() + "] Call Error("
|
|
+ resEAIMessage.getRspErrCd() + ") Retry - " + retry + "/"
|
|
+ this.retryCount);
|
|
}
|
|
// 다음 처리전에 delay를 준다.
|
|
if (this.errorDelayTimeMs > 0) {
|
|
if (logger.isWarn()) {
|
|
logger.warn(guidLogPrefix + this.getName() + "] Delay Before Retry - "
|
|
+ this.errorDelayTimeMs);
|
|
}
|
|
try {
|
|
Thread.sleep(this.errorDelayTimeMs);
|
|
} catch (Exception e) {
|
|
}
|
|
}
|
|
} else {
|
|
if (logger.isWarn()) {
|
|
logger.warn(guidLogPrefix + this.getName() + "] FlowControl Call Error("
|
|
+ resEAIMessage.getRspErrCd() + ") No Retry");
|
|
}
|
|
}
|
|
}
|
|
retry++;
|
|
|
|
} // while(retry <= this.retryCount)
|
|
|
|
} catch (Exception e) {
|
|
if (resEAIMessage != null && logger.isError()) {
|
|
logger.error(guidLogPrefix + this.getName() + "] FlowControl Call Error("
|
|
+ resEAIMessage.getRspErrCd() + ") " + e.getMessage(), e);
|
|
}
|
|
}
|
|
|
|
if (response != null) {
|
|
// increment message counter
|
|
synchronized (lockCounter) {
|
|
msgSuccessCounter++;
|
|
}
|
|
response = null;
|
|
}
|
|
|
|
if (resEAIMessage != null) {
|
|
// EAI 거래처리 Error 면
|
|
// errorQueueName 이 있으면 redirect
|
|
// errorDelayTimeMs > 0 면 sleep
|
|
if (!MessageUtil.checkRspErrCd(resEAIMessage.getRspErrCd())) {
|
|
if (this.errorQueueName != null && this.errorQueueName.length() > 0) {
|
|
if (logger.isWarn()) {
|
|
logger.warn(guidLogPrefix + this.getName() + "] Redirect Messsage to - "
|
|
+ this.errorQueueName);
|
|
}
|
|
JMSSender.sendToQueue(eaiMessage, queueConFactory, this.errorQueueName, prop);
|
|
}
|
|
}
|
|
}
|
|
// 다음 처리 전에 delay를 준다.
|
|
if (this.errorDelayTimeMs > 0) {
|
|
if (logger.isWarn()) {
|
|
logger.warn(guidLogPrefix + this.getName() + "] Delay Before Receive Next Message - "
|
|
+ this.errorDelayTimeMs);
|
|
}
|
|
try {
|
|
Thread.sleep(this.errorDelayTimeMs);
|
|
} catch (Exception e) {
|
|
}
|
|
}
|
|
|
|
} catch (Exception ex) {
|
|
if (logger.isError()) {
|
|
logger.error(this.getName() + "] JMSException for receiverObj[" + id + "] - " + ex.getMessage(),
|
|
ex);
|
|
}
|
|
|
|
if (eaiMessage != null) {
|
|
// --------------------------------------------
|
|
// Set Error
|
|
String[] msgArgs = new String[1];
|
|
msgArgs[0] = eaiMessage.getEAISvcCd();
|
|
String rspErrorCode = "RECEAIFQR001"; // [EAI 서비스코드:{1}] 큐수신자가 플로우컨트롤 호출 중 오류 발생
|
|
String rspErrorMsg = ExceptionUtil.make(ex, rspErrorCode, msgArgs);
|
|
// eaiMessage.setRspErrCd(rspErrorCode);
|
|
// eaiMessage.setRspErrMsg(rspErrorMsg);
|
|
eaiMessage.setRspErr(rspErrorCode,rspErrorMsg);
|
|
// --------------------------------------------
|
|
// String mesgDmanDvcd = eaiMessage.getKbMsg().getKBHeader().getTelgmDmndDstcd();
|
|
// String mesgDmanDvcd = eaiMessage.getExtMsg().getSendRecv();
|
|
String mesgDmanDvcd = eaiMessage.getMapper()
|
|
.getSendRecvDivision(eaiMessage.getStandardMessage()); // S|R
|
|
if (STDMessageKeys.SEND_RECV_CD_RECV.equals(mesgDmanDvcd)) {
|
|
// 응답인 경우
|
|
eaiMessage.setLogPssSno(400);
|
|
} else {
|
|
// 요청 인경우
|
|
eaiMessage.setLogPssSno(200);
|
|
}
|
|
|
|
try {
|
|
EAILogSender.send(eaiMessage, null);
|
|
} catch (Exception e) {
|
|
if (logger.isError())
|
|
logger.error(this.getName() + "] ErrorLogging :" + e.getMessage());
|
|
}
|
|
}
|
|
// auto reconnect so wait 10 seconds
|
|
try {
|
|
Thread.sleep(10000);
|
|
} catch (Exception e) {
|
|
// Do nothing
|
|
}
|
|
}
|
|
} // end of while
|
|
|
|
// if Stop Event Occurred
|
|
if (!keepRunning) {
|
|
threadDowned[id] = true;
|
|
synchronized (this) {
|
|
boolean tmpDowned = false;
|
|
for (int j = 0; j < threadDowned.length; j++) {
|
|
if (threadDowned[j]) {
|
|
tmpDowned = true;
|
|
} else {
|
|
tmpDowned = false;
|
|
break;
|
|
}
|
|
}
|
|
|
|
if (tmpDowned) {
|
|
allThreadDowned = true;
|
|
}
|
|
}
|
|
}
|
|
|
|
} // end of run
|
|
|
|
}// end of inner class ReceiverObject
|
|
}
|