Hi.
I have a memory allocation problem with a ServiceAssembly in Servicemix 3.3.
A file based tranformation will be done by the following bean code. The
transformation allocates 1GB of memory. Once the transformation is finnished
successfully with a tranformation result file, the memory won't be
de-allocated.
Waiting for several hours or days the memory will still allocated and not
released. So, the result is that the Servicemix will shut down due an out of
memory exception. ;-(
What is the problem with my code, is there a bug in Servicemix or what I
should change in the bean?
Thanks a lot,
Andreas.
Here comes my bean code:
===========================
package de.mmsdresden.sportscheck.beans;
import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import javax.jbi.component.ComponentContext;
import javax.jbi.messaging.DeliveryChannel;
import javax.jbi.messaging.ExchangeStatus;
import javax.jbi.messaging.InOnly;
import javax.jbi.messaging.MessageExchange;
import javax.jbi.messaging.MessagingException;
import javax.jbi.messaging.NormalizedMessage;
import javax.jbi.messaging.MessageExchange.Role;
import javax.xml.transform.TransformerConfigurationException;
import javax.xml.transform.TransformerException;
import javax.xml.transform.stream.StreamResult;
import javax.xml.transform.stream.StreamSource;
import org.apache.commons.lang.StringEscapeUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.servicemix.MessageExchangeListener;
import org.apache.servicemix.client.ServiceMixClientFacade;
import org.apache.servicemix.components.util.DefaultFileMarshaler;
import org.apache.servicemix.jbi.api.Destination;
import org.apache.servicemix.jbi.api.ServiceMixClient;
import org.jdom.Document;
import org.jdom.Element;
import org.jdom.JDOMException;
import org.jdom.input.SAXBuilder;
import org.jdom.output.XMLOutputter;
import org.milyn.payload.JavaResult;
import org.xml.sax.SAXException;
import
de.mmsdresden.sportscheck.pojos.pricesbfs.demandware.input.BfsPriceInfo;
import de.mmsdresden.sportscheck.pojos.pricesbfs.demandware.input.Item;
import de.mmsdresden.sportscheck.pojos.pricesbfs.demandware.input.PriceBook;
import de.mmsdresden.sportscheck.utils.FileNameUtils;
import de.mmsdresden.sportscheck.utils.LoggingUtils;
import de.mmsdresden.sportscheck.utils.Mail;
import de.mmsdresden.sportscheck.utils.MessageExchangeCache;
import de.mmsdresden.sportscheck.utils.MessageExchangeCacheImpl;
import de.mmsdresden.sportscheck.utils.ServiceMixUtils;
import de.mmsdresden.sportscheck.utils.SmooksWrapper;
import de.mmsdresden.sportscheck.utils.XMLUtils;
import de.mmsdresden.sportscheck.utils.properties.PropertyLoader;
import
de.mmsdresden.sportscheck.utils.properties.factories.FilenamePropertyFactory;
import
de.mmsdresden.sportscheck.utils.properties.factories.MailPropertyFactory;
/**
* DW -> BFS Price Mapping
*
* @author Andreas Lutz
* @since 03.08.2009
*/
public class PriceBean implements MessageExchangeListener {
/**
* @see javax.jbi.messaging.DeliveryChannel
*/
@Resource
private DeliveryChannel channel;
/**
* @see javax.jbi.component.ComponentContext
*/
@Resource
private ComponentContext context;
private ServiceMixClient client;
private Destination fileoutProduction;
private Destination fileoutTest;
private static Log LOGGER = LogFactory.getLog(PriceBean.class);
private static final String SERVICE_NAMESPACE =
"service:http://sportscheck.com/services/pricesbfs/";;
private static final String FILEOUT_PRODUCTION_ENDPOINT =
SERVICE_NAMESPACE
+ "PricesOutProduction";
private static final String FILEOUT_TEST_ENDPOINT = SERVICE_NAMESPACE +
"PricesOutTest";
private static final String SMOOKS_PRICE_INPUT_CONF =
"smooks_conf/pricesbfs_dmware_input.xml";
private static final String SMOOKS_PRICE_OUTPUT_CONF =
"smooks_conf/pricesbfs_bfs_output.xml";
private MessageExchangeCache exchangeCache = new
MessageExchangeCacheImpl();
private PropertyLoader propsFilename = null;
private String fileOutName;
private String fileInName;
private static final String PROPERTIES_PATH_TEST =
"baur.import.test.prices";
private static final String PROPERTIES_PATH_PRODUCTION =
"baur.import.production.prices";
private static final String DEFAULT_XSL_PATH = "sportscheck_conf/XSL/";
private static final String DW_XSL = "DW/dw_prices.xsl";
private static String GLOBAL_ERROR;
private static String GLOBAL_PROCESS_INVALID_FILES;
static javax.xml.transform.TransformerFactory factory =
javax.xml.transform.TransformerFactory.newInstance();
// some Mail constants //
private static String MAIL_SENDMAIL;
private static String MAIL_SENDER;
private static String MAIL_RECEIVERS;
private static String[] MAIL_RECEIVERS_ARRAY;
private static Mail MAIL;
private static PropertyLoader props = null;
static {
try {
props = new MailPropertyFactory().getImplementation();
} catch (FileNotFoundException e) {
LOGGER.error(e);
} catch (IOException e) {
LOGGER.error(e);
}
GLOBAL_PROCESS_INVALID_FILES =
props.getValue("global.processInvalidFiles");
MAIL = new Mail();
MAIL.setSmtpHost(props.getValue("smtp_host"),
props.getValue("smtp_user"),
props.getValue("smtp_pass"));
MAIL_SENDMAIL =
props.getValue("esb.pricesbfs.validateGeneratedXML.invalid.sendmail");
}
/**
* This init method checks that the needed artifacts are initialized
*
* @throws IOException
* @throws FileNotFoundException
* @throws MessagingException
*/
@PostConstruct
public void init() throws FileNotFoundException, IOException,
MessagingException {
if (this.context == null || this.channel == null) {
throw new IllegalStateException("Bean not initialized");
}
this.propsFilename = new
FilenamePropertyFactory().getImplementation();
this.fileOutName = this.propsFilename.getValue("pricebooks");
/*
* Create client to get the destination (where to send the
files)
*/
this.client = new ServiceMixClientFacade(this.context);
this.fileoutProduction =
this.client.createDestination(FILEOUT_PRODUCTION_ENDPOINT);
this.fileoutTest =
this.client.createDestination(FILEOUT_TEST_ENDPOINT);
LOGGER.info("-----------------------------------------------");
LOGGER.info("-------- PriceBFS mapping initialized ---------");
LOGGER.info("-----------------------------------------------");
LOGGER.info("-> For more informations turn on debug logging.");
}
/*
* (non-Javadoc)
*
* @see
* org.apache.servicemix.MessageExchangeListener#onMessageExchange(javax
* .jbi.messaging.MessageExchange)
*/
public synchronized void onMessageExchange(MessageExchange exchange)
throws
MessagingException {
LOGGER.debug("Received Exchange: " + exchange.getExchangeId());
if (exchange.getRole().equals(Role.CONSUMER)) {
if
(this.exchangeCache.containsKey(exchange.getExchangeId())) {
boolean done = true;
boolean error = false;
LOGGER.debug("Exchange has status: " +
exchange.getStatus());
if
(exchange.getStatus().equals(ExchangeStatus.ACTIVE))
done = false;
if
(exchange.getStatus().equals(ExchangeStatus.ERROR))
error = true;
if (done && !error) {
LOGGER.debug("Setting exchange: " +
exchange.getExchangeId() + " to
DONE!");
ServiceMixUtils.sendMessage(exchange,
ExchangeStatus.DONE,
this.channel);
this.exchangeCache.removeFromCache(exchange.getExchangeId());
}
}
} else {
try {
process(exchange);
ServiceMixUtils.sendMessage(exchange,
ExchangeStatus.DONE,
this.channel);
} catch (IOException e) {
LOGGER.error(e);
} catch (SAXException e) {
LOGGER.error(e);
} catch (TransformerException e) {
LOGGER.error(e);
}
}
LOGGER.debug("Exchanges in cache: " +
this.exchangeCache.size());
}
private void process(MessageExchange exchange) throws
TransformerException,
IOException, SAXException, MessagingException {
InOnly inExchange = (InOnly) exchange;
NormalizedMessage message = inExchange.getInMessage();
String filePath = (String)
message.getProperty(DefaultFileMarshaler.FILE_PATH_PROPERTY);
fileInName = (String)
message.getProperty(DefaultFileMarshaler.FILE_NAME_PROPERTY);
StreamSource content =
ServiceMixUtils.getMessagePayload(message);
LOGGER.info("Processing Exchange Message, with incoming file
name " +
fileInName);
// fix DW XML (Smooks only works with unique tag-ids) //
LOGGER.info("Fixing DW XML file ...");
content = fixedXML(content);
// get XML into JAVA //
LOGGER.info("Mapping XML to Java ...");
List<PriceBook> in_prices = xmlTojava(content);
// unescape String //
LOGGER.info("Mapping escaped String to Java ...");
mapEscapedXMLString(in_prices);
// create XML from JAVA //
LOGGER.info("Creating XML from Java ...");
StreamResult out = javaToxml(in_prices);
// validate generated xml //
LOGGER.info("Validating generated XML ...");
boolean isvalid = validate(filePath, out);
if (!isvalid && MAIL_SENDMAIL.equals("1")) {
sendErrorMail(filePath);
}
// ouput XML //
if (isvalid || (!isvalid &&
GLOBAL_PROCESS_INVALID_FILES.equals("1"))) {
LOGGER.info("Saving XML to Filesystem ...");
outputToFile(filePath, out);
}
// Done //
LOGGER.info("Done processing exchange message ...");
}
private StreamSource fixedXML(StreamSource content) throws IOException,
TransformerConfigurationException, FileNotFoundException {
SAXBuilder sb = new SAXBuilder();
try {
Document doc = sb.build(content.getInputStream());
Element root = doc.getRootElement();
Element transformed = XMLUtils.modifyXML(root,
System.getProperty("user.dir") + File.separator + DEFAULT_XSL_PATH +
DW_XSL);
XMLOutputter xo = new XMLOutputter();
// xo.output(transformed, new
FileWriter("c:\\test.xml"));
return new StreamSource(new
ByteArrayInputStream(xo.outputString(transformed).getBytes()));
} catch (JDOMException e) {
LOGGER.error("Error parsing XML JDOM - " + e);
return null;
}
}
private StreamResult javaToxml(List<PriceBook> prices) throws
IOException,
SAXException {
SmooksWrapper xmlMapper = SmooksWrapper.getInstance();
Map<String, Object> outObjects = new HashMap<String, Object>();
outObjects.put("transformedPrices", prices);
StreamResult out =
xmlMapper.mapJavaToXML(SMOOKS_PRICE_OUTPUT_CONF,
outObjects);
return out;
}
private void mapEscapedXMLString(List<PriceBook> in_prices) throws
IOException, SAXException {
Iterator<PriceBook> book_iter = in_prices.iterator();
while (book_iter.hasNext()) {
PriceBook book = book_iter.next();
if (!book.getBfsPriceInfoString().equals("")) {
String unescaped =
StringEscapeUtils.unescapeXml(book.getBfsPriceInfoString());
unescaped = unescaped.replace("&null;", "");
SmooksWrapper xmlMapper =
SmooksWrapper.getInstance();
JavaResult result =
xmlMapper.mapXMLtoJava(SMOOKS_PRICE_INPUT_CONF, new
StreamSource(new ByteArrayInputStream(unescaped
.getBytes())));
book.setBfsPriceInfo((BfsPriceInfo)
result.getBean("bfsPriceInfo"));
// create map out of arraylist //
Iterator<Item> iter =
book.getBfsPriceInfo().getItemsList().iterator();
HashMap<String, Item> map = new HashMap<String,
Item>();
while (iter.hasNext()) {
Item item = iter.next();
map.put(item.getProductId(), item);
}
book.getBfsPriceInfo().setItemsMap(map);
}
}
}
@SuppressWarnings("unchecked")
private List<PriceBook> xmlTojava(StreamSource content) throws
IOException,
SAXException {
SmooksWrapper xmlMapper = SmooksWrapper.getInstance();
JavaResult result =
xmlMapper.mapXMLtoJava(SMOOKS_PRICE_INPUT_CONF,
content);
List<PriceBook> in_prices = (LinkedList<PriceBook>)
result.getBean("pricebooks");
return in_prices;
}
private boolean validate(String filePath, StreamResult out) {
String path = null;
if (ServiceMixUtils.isTestDir(filePath))
path = PROPERTIES_PATH_TEST;
else
path = PROPERTIES_PATH_PRODUCTION;
try {
XMLUtils.validateXML(out.getOutputStream(), path);
} catch (Exception e) {
LOGGER.error("Exception: " + e);
GLOBAL_ERROR = e.getMessage();
return false;
}
return true;
}
private void sendErrorMail(String filePath) {
MAIL_SENDER =
props.getValue("esb.pricesbfs.validateGeneratedXML.invalid.sender");
MAIL_RECEIVERS =
props.getValue("esb.pricesbfs.validateGeneratedXML.invalid.receivers");
// split receivers //
if (MAIL_RECEIVERS.indexOf("|") > -1) {
MAIL_RECEIVERS_ARRAY = MAIL_RECEIVERS.split("\\|");
} else {
MAIL_RECEIVERS_ARRAY = new String[1];
MAIL_RECEIVERS_ARRAY[0] = new String(MAIL_RECEIVERS);
}
MAIL.setContent(MAIL_SENDER, "Error while validating generated
Baur price
file");
MAIL.addRecipients(MAIL_RECEIVERS_ARRAY);
MAIL.addText("Error validating " + filePath + ":");
MAIL.addText(GLOBAL_ERROR);
if (!MAIL.sendMail()) {
LOGGER.error("Could not sent mail.\r\n Error: " +
MAIL.lastError);
}
}
private void outputToFile(String filePath, StreamResult out) throws
MessagingException, IOException {
MessageExchange[] exchanges = null;
String fileOutTotal =
FileNameUtils.getOutFilenameWithUniqueKey(this.fileOutName, fileInName,
".xml");
if (ServiceMixUtils.isTestDir(filePath)) {
exchanges = new MessageExchange[1];
// fileout test copy //
Properties fileOutProps = new Properties();
fileOutProps.put(DefaultFileMarshaler.FILE_NAME_PROPERTY, fileOutTotal);
exchanges[0] =
ServiceMixUtils.sendMessage(this.fileoutTest, out,
fileOutProps, this.client);
LOGGER.debug("Exchange created - Production file: " +
exchanges[0].getExchangeId() + ", with outgoing file name "
+ fileOutTotal +
LoggingUtils.printSize(out));
} else {
exchanges = new MessageExchange[1];
// fileout production copy //
Properties fileOutProps = new Properties();
fileOutProps.put(DefaultFileMarshaler.FILE_NAME_PROPERTY, fileOutTotal);
exchanges[0] =
ServiceMixUtils.sendMessage(this.fileoutProduction, out,
fileOutProps, this.client);
LOGGER.debug("Exchange created - file: " +
exchanges[0].getExchangeId() +
", with outgoing file name " + fileOutTotal
+ LoggingUtils.printSize(out));
}
}
}
--
View this message in context:
http://old.nabble.com/Memory-leak-or-no-de-allocation-of-memory-tp27613701p27613701.html
Sent from the ServiceMix - User mailing list archive at Nabble.com.
