midsuit
/
core-jgi
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Stabilization of replication 

1.- fix the indentation
2.- fix the save exception
3.- fix close the JMS connection if an exception us catch.

kind regards
Victor Perez
www.e-evolution.com

Link to SF Tracker: http://sourceforge.net/support/tracker.php?aid=2936561
This commit is contained in:
vpj-cd 2010-02-08 01:43:46 +00:00
parent 4830d4c749
commit bec8791241
2 changed files with 78 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
serverRoot/src/main/server/org/adempiere/server/rpl/imp/TopicListener.java
View File

@ -279,7 +279,7 @@ public class TopicListener implements MessageListener {
TextMessage txtMessage = (TextMessage) message; TextMessage txtMessage = (TextMessage) message;
String text = txtMessage.getText(); String text = txtMessage.getText();
//log.finest("Received message: \n" + text ); log.finest("Received message: \n" + text );
Document documentToBeImported = XMLHelper.createDocumentFromString( text ); Document documentToBeImported = XMLHelper.createDocumentFromString( text );
StringBuffer result = new StringBuffer(); StringBuffer result = new StringBuffer();
@ -288,28 +288,45 @@ public class TopicListener implements MessageListener {
impHelper.importXMLDocument(result, documentToBeImported, trxName ); impHelper.importXMLDocument(result, documentToBeImported, trxName );
log.finest("Replicated ...");
MIMPProcessorLog pLog = new MIMPProcessorLog(replicationProcessor.getMImportProcessor(), "Imported Document!"); if(replicationProcessor != null)
//pLog.setReference("topicName = " + topicName ); {
if (text.length() > 2000 ) { MIMPProcessorLog pLog = new MIMPProcessorLog(replicationProcessor.getMImportProcessor(), "Imported Document!");
pLog.setTextMsg( text.substring(0, 1999) ); //pLog.setReference("topicName = " + topicName );
} else { if (text.length() > 2000 ) {
pLog.setTextMsg( text); pLog.setTextMsg( text.substring(0, 1999) );
} else {
pLog.setTextMsg( text);
}
pLog.saveEx();
} }
boolean resultSave = pLog.save();
log.finest("Result Save = " + resultSave);
session.commit(); session.commit();
} catch (Exception e) { }
replicationProcessor.setProcessRunning(false); catch (Exception e)
try { {
log.finest("Rollback = " + e.toString());
try
{
session.rollback(); session.rollback();
} catch (JMSException e1) { stop();
e1.printStackTrace(); //replicationProcessor.interrupt();
//replicationProcessor.join();
replicationProcessor.setProcessRunning(false);
}
/*catch (InterruptedException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}*/
catch (JMSException e2)
{
e2.printStackTrace();
} }
e.printStackTrace(); e.printStackTrace();
} }
} else { } else {

80
serverRoot/src/main/server/org/compiere/server/ReplicationProcessor.java
View File

@ -85,62 +85,66 @@ public class ReplicationProcessor extends AdempiereServer {
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
@Override @Override
protected void doWork() { protected void doWork()
if (isProcessRunning) { {
if (isProcessRunning)
{
// process is already started successfully! // process is already started successfully!
} else { }
// process is not started! else
{
// process is not started!
m_summary = new StringBuffer(); m_summary = new StringBuffer();
String trxName = mImportProcessor.get_TrxName(); String trxName = mImportProcessor.get_TrxName();
if ( trxName == null || "".equals(trxName) ) { if ( trxName == null || "".equals(trxName) )
{
// trxName = "ImportProcessor-" + System.currentTimeMillis(); // trxName = "ImportProcessor-" + System.currentTimeMillis();
} }
log.fine("trxName = " + trxName); log.fine("trxName = " + trxName);
log.fine("ImportProcessor = " + mImportProcessor); log.fine("ImportProcessor = " + mImportProcessor);
int IMP_ProcessorType_ID = 0; int IMP_ProcessorType_ID = 0;
IMP_ProcessorType_ID = mImportProcessor.getIMP_Processor_Type_ID(); IMP_ProcessorType_ID = mImportProcessor.getIMP_Processor_Type_ID();
X_IMP_Processor_Type impProcessor_Type = new X_IMP_Processor_Type(mImportProcessor.getCtx(), IMP_ProcessorType_ID, trxName ); X_IMP_Processor_Type impProcessor_Type = new X_IMP_Processor_Type(mImportProcessor.getCtx(), IMP_ProcessorType_ID, trxName );
log.fine("impProcessor_Type = " + impProcessor_Type); // TODO --- REMOVE log.fine("impProcessor_Type = " + impProcessor_Type); // TODO --- REMOVE
String javaClass = impProcessor_Type.getJavaClass(); String javaClass = impProcessor_Type.getJavaClass();
IImportProcessor importProcessor = null; IImportProcessor importProcessor = null;
try { try
Class clazz = Class.forName(javaClass); {
importProcessor = (IImportProcessor)clazz.newInstance(); Class clazz = Class.forName(javaClass);
importProcessor = (IImportProcessor)clazz.newInstance();
importProcessor.process(mImportProcessor.getCtx(), this, trxName ); importProcessor.process(mImportProcessor.getCtx(), this, trxName );
}
} catch (Exception e) { catch (Exception e)
{
isProcessRunning = false; isProcessRunning = false;
log.fine("ReplicationProcessor caught an exception !!!" ); log.fine("ReplicationProcessor caught an exception !!!" );
e.printStackTrace(); e.printStackTrace();
log.severe(e.getMessage());
log.severe( e.getMessage() );
MIMPProcessorLog pLog = new MIMPProcessorLog(mImportProcessor, e.getMessage() ); MIMPProcessorLog pLog = new MIMPProcessorLog(mImportProcessor, e.getMessage() );
pLog.setReference("#" + String.valueOf(p_runCount) + " - " + TimeUtil.formatElapsed(new Timestamp(p_startWork))); pLog.setReference("#" + String.valueOf(p_runCount) + " - " + TimeUtil.formatElapsed(new Timestamp(p_startWork)));
boolean resultSave = pLog.save(); pLog.saveEx();
try
try { {
importProcessor.stop(); importProcessor.stop();
} catch (Exception e1) {
e1.printStackTrace();
MIMPProcessorLog pLog2 = new MIMPProcessorLog(mImportProcessor, e1.getMessage() );
boolean resultSave2 = pLog2.save();
} }
} catch (Exception e1)
{
e1.printStackTrace();
MIMPProcessorLog pLog2 = new MIMPProcessorLog(mImportProcessor, e1.getMessage() );
pLog2.saveEx();
}
}
// //
int no = mImportProcessor.deleteLog(); int no = mImportProcessor.deleteLog();
m_summary.append("Logs Records deleted=").append(no).append("; "); m_summary.append("Logs Records deleted=").append(no).append("; ");
// //
MIMPProcessorLog pLog = new MIMPProcessorLog(mImportProcessor, m_summary.toString()); MIMPProcessorLog pLog = new MIMPProcessorLog(mImportProcessor, m_summary.toString());
pLog.setReference("#" + String.valueOf(p_runCount) + " - " + TimeUtil.formatElapsed(new Timestamp(p_startWork))); pLog.setReference("#" + String.valueOf(p_runCount) + " - " + TimeUtil.formatElapsed(new Timestamp(p_startWork)));
boolean resultSave = pLog.save(); pLog.saveEx();
} }
} }