<commit_before>/*
 * $Log: ReceiverBase.java,v $
 * Revision 1.40  2007-06-21 07:07:06  europe\L190409
 * removed warnings about not transacted=true
 *
 * Revision 1.39  2007/06/19 12:07:32  <NAME> <<EMAIL>>
 * modifiy retryinterval handling
 *
 * Revision 1.38  2007/06/14 08:49:35  <NAME> <<EMAIL>>
 * catch less specific types of exception
 *
 * Revision 1.37  2007/06/12 11:24:04  <NAME> <<EMAIL>>
 * corrected typeSettings of transactional storages
 *
 * Revision 1.36  2007/06/08 12:49:03  <NAME> <<EMAIL>>
 * updated javadoc
 *
 * Revision 1.35  2007/06/08 12:17:40  <NAME> <<EMAIL>>
 * improved error handling
 * introduced retry mechanisme with increasing wait interval
 *
 * Revision 1.34  2007/06/08 07:49:13  <NAME> <<EMAIL>>
 * changed error to warning
 *
 * Revision 1.33  2007/06/07 15:22:44  <NAME> <<EMAIL>>
 * made stopping after receiving an exception configurable
 *
 * Revision 1.32  2007/05/23 09:25:17  <NAME> <<EMAIL>>
 * added support for attribute 'active' on transactional storages
 *
 * Revision 1.31  2007/05/21 12:22:47  <NAME> <<EMAIL>>
 * added setMessageLog()
 *
 * Revision 1.30  2007/05/02 11:37:51  <NAME> <<EMAIL>>
 * added attribute 'active'
 *
 * Revision 1.29  2007/02/12 14:03:45  <NAME> <<EMAIL>>
 * Logger from LogUtil
 *
 * Revision 1.28  2007/02/05 15:01:44  <NAME> <<EMAIL>>
 * configure inProcessStorage when it is present, not only when transacted
 *
 * Revision 1.27  2006/12/13 16:30:41  <NAME> <<EMAIL>>
 * added maxRetries to configuration javadoc
 *
 * Revision 1.26  2006/08/24 07:12:42  <NAME> <<EMAIL>>
 * documented METT tracing event numbers
 *
 * Revision 1.25  2006/06/20 14:10:43  <NAME> <<EMAIL>>
 * added stylesheet attribute
 *
 * Revision 1.24  2006/04/12 16:17:43  <NAME> <<EMAIL>>
 * retry after failed storing of message in inProcessStorage
 *
 * Revision 1.23  2006/02/20 15:42:41  <NAME> <<EMAIL>>
 * moved METT-support to single entry point for tracing
 *
 * Revision 1.22  2006/02/09 07:57:47  <NAME> <<EMAIL>>
 * METT tracing support
 *
 * Revision 1.21  2005/10/27 08:46:45  <NAME> <<EMAIL>>
 * introduced RunStateEnquiries
 *
 * Revision 1.20  2005/10/26 08:52:31  <NAME> <<EMAIL>>
 * allow for transacted="true" without inProcessStorage, (ohne Gewhr!)
 *
 * Revision 1.19  2005/10/17 11:29:24  <NAME> <<EMAIL>>
 * fixed nullpointerexception in startRunning
 *
 * Revision 1.18  2005/09/26 11:42:10  <NAME> <<EMAIL>>
 * added fileNameIfStopped attribute and replace from/to processing when stopped
 *
 * Revision 1.17  2005/09/13 15:42:14  <NAME> <<EMAIL>>
 * improved handling of non-serializable messages like Poison-messages
 *
 * Revision 1.16  2005/08/08 09:44:11  <NAME> <<EMAIL>>
 * start transactions if needed and not already started
 *
 * Revision 1.15  2005/07/19 15:27:14  <NAME> <<EMAIL>>
 * modified closing procedure
 * added errorStorage
 * modified implementation of transactionalStorage
 * allowed exceptions to bubble up
 * assume rawmessages to be serializable for transacted processing
 * added ibis42compatibility attribute, avoiding exception bubbling
 *
 * Revision 1.14  2005/07/05 12:54:38  <NAME> <<EMAIL>>
 * allow to set parameters from context for processRequest() methods
 *
 * Revision 1.13  2005/06/02 11:52:24  <NAME> <<EMAIL>>
 * limited number of actively polling threads to value of attriubte numThreadsPolling
 *
 * Revision 1.12  2005/04/13 12:53:09  <NAME> <<EMAIL>>
 * removed unused imports
 *
 * Revision 1.11  2005/03/31 08:22:49  <NAME> <<EMAIL>>
 * fixed bug in getIdleStatistics
 *
 * Revision 1.10  2005/03/07 11:04:36  <NAME> <<EMAIL>>
 * PipeLineSession became a extension of HashMap, using other iterator
 *
 * Revision 1.9  2005/03/04 08:53:29  <NAME> <<EMAIL>>
 * Fixed IndexOutOfBoundException in getProcessStatistics  due to multi threading.
 * Adjusted this too for getIdleStatistics
 *
 * Revision 1.8  2005/02/10 08:17:34  <NAME> <<EMAIL>>
 * included context dump in debug
 *
 * Revision 1.7  2005/01/13 08:56:04  <NAME> <<EMAIL>>
 * Make threadContext-attributes available in PipeLineSession
 *
 * Revision 1.6  2004/10/12 15:14:11  <NAME> <<EMAIL>>
 * removed unused code
 *
 * Revision 1.5  2004/08/25 09:11:33  unknown <<EMAIL>>
 * Add waitForRunstate with timeout
 *
 * Revision 1.4  2004/08/23 13:10:48  <NAME> <<EMAIL>>
 * updated JavaDoc
 *
 * Revision 1.3  2004/08/16 14:09:58  unknown <<EMAIL>>
 * Return returnIfStopped value in case adapter is stopped
 *
 * Revision 1.2  2004/08/09 13:46:52  <NAME> <<EMAIL>>
 * various changes
 *
 * Revision 1.1  2004/08/03 13:04:30  <NAME> <<EMAIL>>
 * introduction of GenericReceiver
 *
 */
package nl.nn.adapterframework.receivers;

import java.io.IOException;
import java.io.Serializable;
import java.net.URL;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;

import javax.transaction.Status;
import javax.transaction.UserTransaction;
import javax.xml.transform.Transformer;
import javax.xml.transform.TransformerConfigurationException;
import javax.xml.transform.TransformerException;

import nl.nn.adapterframework.configuration.ConfigurationException;
import nl.nn.adapterframework.core.HasPhysicalDestination;
import nl.nn.adapterframework.core.HasSender;
import nl.nn.adapterframework.core.IAdapter;
import nl.nn.adapterframework.core.IListener;
import nl.nn.adapterframework.core.IMessageHandler;
import nl.nn.adapterframework.core.INamedObject;
import nl.nn.adapterframework.core.IPullingListener;
import nl.nn.adapterframework.core.IPushingListener;
import nl.nn.adapterframework.core.IReceiver;
import nl.nn.adapterframework.core.IReceiverStatistics;
import nl.nn.adapterframework.core.ISender;
import nl.nn.adapterframework.core.ITransactionalStorage;
import nl.nn.adapterframework.core.IXAEnabled;
import nl.nn.adapterframework.core.IbisExceptionListener;
import nl.nn.adapterframework.core.ListenerException;
import nl.nn.adapterframework.core.PipeLineResult;
import nl.nn.adapterframework.core.PipeLineSession;
import nl.nn.adapterframework.core.SenderException;
import nl.nn.adapterframework.core.TransactionException;
import nl.nn.adapterframework.util.ClassUtils;
import nl.nn.adapterframework.util.Counter;
import nl.nn.adapterframework.util.DomBuilderException;
import nl.nn.adapterframework.util.JtaUtil;
import nl.nn.adapterframework.util.LogUtil;
import nl.nn.adapterframework.util.Misc;
import nl.nn.adapterframework.util.RunStateEnquiring;
import nl.nn.adapterframework.util.RunStateEnum;
import nl.nn.adapterframework.util.RunStateManager;
import nl.nn.adapterframework.util.Semaphore;
import nl.nn.adapterframework.util.StatisticsKeeper;
import nl.nn.adapterframework.util.TracingEventNumbers;
import nl.nn.adapterframework.util.TracingUtil;
import nl.nn.adapterframework.util.XmlUtils;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.SystemUtils;
import org.apache.commons.lang.builder.ToStringBuilder;
import org.apache.commons.lang.builder.ToStringStyle;
import org.apache.log4j.Logger;

/**
 * This {@link IReceiver Receiver} may be used as a base-class for developing receivers.
 *
 * <p><b>Configuration:</b>
 * <table border="1">
 * <tr><th>attributes</th><th>description</th><th>default</th></tr>
 * <tr><td>classname</td><td>name of the class, mostly a class that extends this class</td><td>&nbsp;</td></tr>
 * <tr><td>{@link #setName(String) name}</td>  <td>name of the receiver as known to the adapter</td><td>&nbsp;</td></tr>
 * <tr><td>{@link #setActive(boolean) active}</td>  <td>when set <code>false</code> or set to something else as "true", (even set to the empty string), the receiver is not included in the configuration</td><td>true</td></tr>
 * <tr><td>{@link #setNumThreads(int) numThreads}</td><td>the number of threads that may execute a pipeline concurrently (only for pulling listeners)</td><td>1</td></tr>
 * <tr><td>{@link #setNumThreadsPolling(int) numThreadsPolling}</td><td>the number of threads that are activily polling for messages concurrently. '0' means 'limited only by <code>numThreads</code>' (only for pulling listeners)</td><td>1</td></tr>
 * <tr><td>{@link #setStyleSheetName(String) styleSheetName}</td>  <td></td><td>&nbsp;</td></tr>
 * <tr><td>{@link #setOnError(String) onError}</td><td>one of 'continue' or 'close'. Controls the behaviour of the receiver when it encounters an error sending a reply or receives an exception asynchronously</td><td>continue</td></tr>
 * <tr><td>{@link #setTransacted(boolean) transacted}</td><td>if set to <code>true, messages will be received and processed under transaction control. If processing fails, messages will be sent to the error-sender. (see below)</code></td><td><code>false</code></td></tr>
 * <tr><td>{@link #setMaxRetries(int) maxRetries}</td><td>The number of times a pulling listening attempt is retried after an exception is caught</td><td>3</td></tr>
 * <tr><td>{@link #setIbis42compatibility(boolean) ibis42compatibility}</td><td>if set to <code>true, the result of a failed processing of a message is a formatted errormessage. Otherwise a listener specific error handling is performed</code></td><td><code>false</code></td></tr>
 * <tr><td>{@link #setBeforeEvent(int) beforeEvent}</td>      <td>METT eventnumber, fired just before a message is processed by this Receiver</td><td>-1 (disabled)</td></tr>
 * <tr><td>{@link #setAfterEvent(int) afterEvent}</td>        <td>METT eventnumber, fired just after message processing by this Receiver is finished</td><td>-1 (disabled)</td></tr>
 * <tr><td>{@link #setExceptionEvent(int) exceptionEvent}</td><td>METT eventnumber, fired when message processing by this Receiver resulted in an exception</td><td>-1 (disabled)</td></tr>
 * </table>
 * </p>
 * <p>
 * <table border="1">
 * <tr><th>nested elements (accessible in descender-classes)</th><th>description</th></tr>
 * <tr><td>{@link nl.nn.adapterframework.core.IPullingListener listener}</td><td>the listener used to receive messages from</td></tr>
 * <tr><td>{@link nl.nn.adapterframework.core.ITransactionalStorage inProcessStorage}</td><td>mandatory for {@link #setTransacted(boolean) transacted} receivers: place to store messages during processing.</td></tr>
 * <tr><td>{@link nl.nn.adapterframework.core.ITransactionalStorage errorStorage}</td><td>optional for {@link #setTransacted(boolean) transacted} receivers: place to store messages if message processing has gone wrong. If no errorStorage is specified, the inProcessStorage is used for errorStorage</td></tr>
 * <tr><td>{@link nl.nn.adapterframework.core.ISender errorSender}</td><td>optional for {@link #setTransacted(boolean) transacted} receviers: 
 * will be called to store messages that failed to process. If no errorSender is specified, failed messages will remain in inProcessStorage</td></tr>
 * </table>
 * </p>
 * <p><b>Transaction control</b><br>
 * If {@link #setTransacted(boolean) transacted} is set to <code>true, messages will be received and processed under transaction control.
 * This means that after a message has been read and processed and the transaction has ended, one of the following apply:
 * <ul>
 * <table border="1">
 * <tr><th>situation</th><th>input listener</th><th>Pipeline</th><th>inProcess storage</th><th>errorSender</th><th>summary of effect</th></tr>
 * <tr><td>successful</td><td>message read and committed</td><td>message processed</td><td>unchanged</td><td>unchanged</td><td>message processed</td></tr>
 * <tr><td>procesing failed</td><td>message read and committed</td><td>message processing failed and rolled back</td><td>unchanged</td><td>message sent</td><td>message only transferred from listener to errroSender</td></tr>
 * <tr><td>listening failed</td><td>unchanged: listening rolled back</td><td>no processing performed</td><td>unchanged</td><td>unchanged</td><td>no changes, input message remains on input available for listener</td></tr>
 * <tr><td>transfer to inprocess storage failed</td><td>unchanged: listening rolled back</td><td>no processing performed</td><td>unchanged</td><td>unchanged</td><td>no changes, input message remains on input available for listener</td></tr>
 * <tr><td>transfer to errorSender failed</td><td>message read and committed</td><td>message processing failed and rolled back</td><td>message present</td><td>unchanged</td><td>message only transferred from listener to inProcess storage</td></tr>
 * </table> 
 * If the application or the server crashes in the middle of one or more transactions, these transactions 
 * will be recovered and rolled back after the server/application is restarted. Then allways exactly one of 
 * the following applies for any message touched at any time by Ibis by a transacted receiver:
 * <ul>
 * <li>It is processed correctly by the pipeline and removed from the input-queue, 
 *     not present in inProcess storage and not send to the errorSender</li> 
 * <li>It is not processed at all by the pipeline, or processing by the pipeline has been rolled back; 
 *     the message is removed from the input queue and either (one of) still in inProcess storage <i>or</i> sent to the errorSender</li>
 * </ul>
 * </p>
 *
 * <p><b>commit or rollback</b><br>
 * If {@link #setTransacted(boolean) transacted} is set to <code>true</code>, messages will be either committed or rolled back.
 * All message-processing transactions are committed, unless one or more of the following apply:
 * <ul>
 * <li>The PipeLine is transacted and the exitState of the pipeline is not equal to {@link nl.nn.adapterframework.core.PipeLine#setCommitOnState(String) commitOnState} (that defaults to 'success')</li>
 * <li>a PipeRunException or another runtime-exception has been thrown by any Pipe or by the PipeLine</li>
 * <li>the setRollBackOnly() method has been called on the userTransaction (not accessible by Pipes)</li>
 * </ul>
 * </p>
 *
 * @version Id
 * @author     <NAME>el
 * @since 4.2
 */
public class ReceiverBase implements IReceiver, IReceiverStatistics, Runnable, IMessageHandler, IbisExceptionListener, HasSender, TracingEventNumbers {
	public static final String version="$RCSfile: ReceiverBase.java,v $ $Revision: 1.40 $ $Date: 2007-06-21 07:07:06 $";
	protected Logger log = LogUtil.getLogger(this);
 
	private String returnIfStopped="";
	private String fileNameIfStopped = null;
	private String replaceFrom = null;
	private String replaceTo = null;
	private String styleSheetName = null;

	public static final String ONERROR_CONTINUE = "continue";
	public static final String ONERROR_CLOSE = "close";

	private boolean active=true;

  	private String name;
  	private String onError = ONERROR_CONTINUE; 
    protected RunStateManager runState = new RunStateManager();
    
    private boolean ibis42compatibility=false;

	// the number of threads that may execute a pipeline concurrently (only for pulling listeners)
	private int numThreads = 1;
	// the number of threads that are activily polling for messages (concurrently, only for pulling listeners)
	private int numThreadsPolling = 1;
   
	private Counter threadsProcessing = new Counter(0);
	private Counter threadsRunning = new Counter(0);
	private Semaphore pollToken = null;
	        
	// number of messages received
    private Counter numReceived = new Counter(0);
	private ArrayList processStatistics = new ArrayList();
	private ArrayList idleStatistics = new ArrayList();

    // the adapter that handles the messages and initiates this listener
    private IAdapter adapter;

	private IListener listener;
    private ITransactionalStorage inProcessStorage=null;
    private ISender errorSender=null;
	private ITransactionalStorage errorStorage=null;
	private ISender sender=null; // answer-sender
	private ITransactionalStorage messageLog=null;
	
	private int maxRetries=3;
	private Counter retryCount = new Counter(0);
    
    private boolean transacted=false;
 
 	// METT event numbers
	private int beforeEvent=-1;
	private int afterEvent=-1;
	private int exceptionEvent=-1;

	int retryInterval=1;
    
	protected String getLogPrefix() {
		return "Receiver ["+getName()+"] "; 
	}	

	/** 
	 * sends an informational message to the log and to the messagekeeper of the adapter
	 */
	protected void info(String msg) {
		log.info(msg);
		if (adapter != null)
			adapter.getMessageKeeper().add(msg);
	}

	/** 
	 * sends a warning to the log and to the messagekeeper of the adapter
	 */
	protected void warn(String msg) {
		log.warn(msg);
		if (adapter != null)
			adapter.getMessageKeeper().add("WARNING: " + msg);
	}

	/** 
	 * sends a warning to the log and to the messagekeeper of the adapter
	 */
	protected void error(String msg, Throwable t) {
		log.error(msg, t);
		if (adapter != null)
			adapter.getMessageKeeper().add("ERROR: " + msg+": "+t.getMessage());
	}


	protected void openAllResources() throws ListenerException {	
		// on exit resouces must be in a state that runstate is or can be set to 'STARTED'
		try {
			if (getSender()!=null) {
				getSender().open();
			}
			if (getErrorSender()!=null) {
				getErrorSender().open();
			}
			if (getInProcessStorage()!=null) {
				getInProcessStorage().open();
			}
			if (getMessageLog()!=null) {
				getMessageLog().open();
			}
			if (isTransacted()) {
				getAdapter().getUserTransaction();
			}
		} catch (SenderException e) {
			throw new ListenerException(e);
		} catch (TransactionException e) {
			throw new ListenerException(e);
		}
		getListener().open();
		if (getListener() instanceof IPullingListener){
			// start all threads
			if (getNumThreads() > 1) {
				for (int i = 1; i <= getNumThreads(); i++) {
					addThread("[" + i+"]");
				}
			} else {
				addThread(null);
			}
		}
	}

	private void addThread(String nameSuffix) {
		if (getListener() instanceof IPullingListener){
			Thread t = new Thread(this, getName() + (nameSuffix==null ? "" : nameSuffix));
			t.start();
		}
	}


	protected void tellResourcesToStop() throws ListenerException {
		 // must lead to a 'closeAllResources()'
		 // runstate is 'STOPPING'
		 // default just calls 'closeAllResources()'
		 if (getListener() instanceof IPushingListener) {
			closeAllResources();
		 }
		 // IPullingListeners stop as their threads finish, as the runstate is set to stopping
	}
	protected void closeAllResources() {
		// on exit resouces must be in a state that runstate can be set to 'STOPPED'
		try {
			log.debug("closing Receiver ["+ getName()+ "]");
			getListener().close();
			if (getSender()!=null) {
				getSender().close();
			}
			if (getErrorSender()!=null) {
				getErrorSender().close();
			}
			if (getInProcessStorage()!=null) {
				getInProcessStorage().close();
			}
			if (getErrorStorage()!=null && getErrorStorage()!=getInProcessStorage()) {
				getInProcessStorage().close();
			}
			if (getMessageLog()!=null) {
				getMessageLog().close();
			}
	
			log.info("closed Receiver ["+ getName()+ "]");
		} catch (Exception e) {
			log.error(
				"Receiver [" + getName()+ "]: error closing connection", e);
		}
		runState.setRunState(RunStateEnum.STOPPED);
		info("Receiver [" + getName() + "] stopped");
	}
	 
	protected void propagateName() {
		IListener listener=getListener();
		if (listener!=null && StringUtils.isEmpty(listener.getName())) {
			listener.setName("listener of ["+getName()+"]");
		}
		ITransactionalStorage inProcess = getInProcessStorage();
		if (inProcess != null) {
			inProcess.setName("inProcessStorage of ["+getName()+"]");
		}
		ISender errorSender = getErrorSender();
		if (errorSender != null) {
			errorSender.setName("errorSender of ["+getName()+"]");
		}
		ITransactionalStorage errorStorage = getErrorStorage();
		if (errorStorage != null && errorStorage != inProcess) {
			errorStorage.setName("errorStorage of ["+getName()+"]");
		}
	}

	public void configure() throws ConfigurationException {		
 		try {
 			propagateName();
			if (getListener()==null) {
				throw new ConfigurationException("Receiver ["+getName()+"] has no listener");
			}
			if (getListener() instanceof IPushingListener) {
				IPushingListener pl = (IPushingListener)getListener();
				pl.setHandler(this);
				pl.setExceptionListener(this);
			}
			if (getListener() instanceof IPullingListener) {
				if (getNumThreadsPolling()>0 && getNumThreadsPolling()<getNumThreads()) {
					pollToken = new Semaphore(getNumThreadsPolling());
				}
			}
			getListener().configure();
			if (getListener() instanceof HasPhysicalDestination) {
				info("Receiver ["+getName()+"] has listener on "+((HasPhysicalDestination)getListener()).getPhysicalDestinationName());
			}
			if (getListener() instanceof HasSender) {
				// only informational
				ISender sender = ((HasSender)getListener()).getSender();
				if (sender instanceof HasPhysicalDestination) {
					info("Listener of receiver ["+getName()+"] has answer-sender on "+((HasPhysicalDestination)sender).getPhysicalDestinationName());
				}
			}
			ISender sender = getSender();
			if (sender!=null) {
				sender.configure();
				if (sender instanceof HasPhysicalDestination) {
					info("receiver ["+getName()+"] has answer-sender on "+((HasPhysicalDestination)sender).getPhysicalDestinationName());
				}
			}
			ITransactionalStorage inProcessStorage = getInProcessStorage();
			if (inProcessStorage!=null) {
				inProcessStorage.configure();
				if (inProcessStorage instanceof HasPhysicalDestination) {
					info("Receiver ["+getName()+"] has inProcessStorage in "+((HasPhysicalDestination)inProcessStorage).getPhysicalDestinationName());
				}
			}
			ISender errorSender = getErrorSender();
			if (errorSender!=null) {
				errorSender.configure();
				if (errorSender instanceof HasPhysicalDestination) {
					info("Receiver ["+getName()+"] has errorSender to "+((HasPhysicalDestination)errorSender).getPhysicalDestinationName());
				}
			}
			ITransactionalStorage errorStorage = getErrorStorage();
			if (errorStorage!=null && errorStorage != getInProcessStorage()) {
				errorStorage.configure();
				if (errorStorage instanceof HasPhysicalDestination) {
					info("Receiver ["+getName()+"] has errorStorage to "+((HasPhysicalDestination)errorStorage).getPhysicalDestinationName());
				}
			}
			ITransactionalStorage messageLog = getMessageLog();
			if (messageLog!=null) {
				messageLog.configure();
				if (messageLog instanceof HasPhysicalDestination) {
					info("Receiver ["+getName()+"] has messageLog in "+((HasPhysicalDestination)messageLog).getPhysicalDestinationName());
				}
			}
			if (isTransacted()) {
//				if (!(getListener() instanceof IXAEnabled && ((IXAEnabled)getListener()).isTransacted())) {
//					warn("Receiver ["+getName()+"] sets transacted=true, but listener not. Transactional integrity is not guaranteed"); 
//				}
				if (inProcessStorage==null && getListener() instanceof IPullingListener) {
//					throw new ConfigurationException("Receiver ["+getName()+"] sets transacted=true, but has no inProcessStorage.");
					warn("Receiver ["+getName()+"] sets transacted=true, but has no inProcessStorage. Transactional integrity is not guaranteed");
//				} else {
//					if (!(inProcessStorage instanceof IXAEnabled && ((IXAEnabled)inProcessStorage).isTransacted())) {
//						warn("Receiver ["+getName()+"] sets transacted=true, but inProcessStorage not. Transactional integrity is not guaranteed"); 
//					}
				}
				
				if (errorSender==null && errorStorage==null) {
					warn("Receiver ["+getName()+"] sets transacted=true, but has no errorSender or errorStorage. Messages processed with errors will be lost");
				} else {
//					if (errorSender!=null && !(errorSender instanceof IXAEnabled && ((IXAEnabled)errorSender).isTransacted())) {
//						warn("Receiver ["+getName()+"] sets transacted=true, but errorSender is not. Transactional integrity is not guaranteed"); 
//					}
//					if (errorStorage!=null && !(errorStorage instanceof IXAEnabled && ((IXAEnabled)errorStorage).isTransacted())) {
//						warn("Receiver ["+getName()+"] sets transacted=true, but errorStorage is not. Transactional integrity is not guaranteed"); 
//					}
					if (errorStorage==inProcessStorage) {
						info("Receiver ["+getName()+"] has errorStorage in inProcessStorage."); 
					}
				}
			} <commit_msg>set inProcessStorage type to 'E' if combined with errorStorage<commit_after>/*
 * $Log: ReceiverBase.java,v $
 * Revision 1.41  2007-06-26 06:56:59  europe\L190409
 * set inProcessStorage type to 'E' if combined with errorStorage
 *
 * Revision 1.40  2007/06/21 07:07:06  <NAME> <<EMAIL>>
 * removed warnings about not transacted=true
 *
 * Revision 1.39  2007/06/19 12:07:32  <NAME> <<EMAIL>>
 * modifiy retryinterval handling
 *
 * Revision 1.38  2007/06/14 08:49:35  <NAME> <<EMAIL>>
 * catch less specific types of exception
 *
 * Revision 1.37  2007/06/12 11:24:04  <NAME> <<EMAIL>>
 * corrected typeSettings of transactional storages
 *
 * Revision 1.36  2007/06/08 12:49:03  <NAME> <<EMAIL>>
 * updated javadoc
 *
 * Revision 1.35  2007/06/08 12:17:40  <NAME> <<EMAIL>>
 * improved error handling
 * introduced retry mechanisme with increasing wait interval
 *
 * Revision 1.34  2007/06/08 07:49:13  <NAME> <<EMAIL>>
 * changed error to warning
 *
 * Revision 1.33  2007/06/07 15:22:44  <NAME> <<EMAIL>>
 * made stopping after receiving an exception configurable
 *
 * Revision 1.32  2007/05/23 09:25:17  <NAME> <<EMAIL>>
 * added support for attribute 'active' on transactional storages
 *
 * Revision 1.31  2007/05/21 12:22:47  <NAME> <<EMAIL>>
 * added setMessageLog()
 *
 * Revision 1.30  2007/05/02 11:37:51  <NAME> <<EMAIL>>
 * added attribute 'active'
 *
 * Revision 1.29  2007/02/12 14:03:45  <NAME> <<EMAIL>>
 * Logger from LogUtil
 *
 * Revision 1.28  2007/02/05 15:01:44  <NAME> <<EMAIL>>
 * configure inProcessStorage when it is present, not only when transacted
 *
 * Revision 1.27  2006/12/13 16:30:41  <NAME> <<EMAIL>>
 * added maxRetries to configuration javadoc
 *
 * Revision 1.26  2006/08/24 07:12:42  <NAME> <<EMAIL>>
 * documented METT tracing event numbers
 *
 * Revision 1.25  2006/06/20 14:10:43  <NAME> <<EMAIL>>
 * added stylesheet attribute
 *
 * Revision 1.24  2006/04/12 16:17:43  <NAME> <<EMAIL>>
 * retry after failed storing of message in inProcessStorage
 *
 * Revision 1.23  2006/02/20 15:42:41  <NAME> <<EMAIL>>
 * moved METT-support to single entry point for tracing
 *
 * Revision 1.22  2006/02/09 07:57:47  <NAME> <<EMAIL>>
 * METT tracing support
 *
 * Revision 1.21  2005/10/27 08:46:45  <NAME> <<EMAIL>>
 * introduced RunStateEnquiries
 *
 * Revision 1.20  2005/10/26 08:52:31  <NAME> <<EMAIL>>
 * allow for transacted="true" without inProcessStorage, (ohne Gewhr!)
 *
 * Revision 1.19  2005/10/17 11:29:24  <NAME> <<EMAIL>>
 * fixed nullpointerexception in startRunning
 *
 * Revision 1.18  2005/09/26 11:42:10  <NAME> <<EMAIL>>
 * added fileNameIfStopped attribute and replace from/to processing when stopped
 *
 * Revision 1.17  2005/09/13 15:42:14  <NAME> <<EMAIL>>
 * improved handling of non-serializable messages like Poison-messages
 *
 * Revision 1.16  2005/08/08 09:44:11  <NAME> <<EMAIL>>
 * start transactions if needed and not already started
 *
 * Revision 1.15  2005/07/19 15:27:14  <NAME> <<EMAIL>>
 * modified closing procedure
 * added errorStorage
 * modified implementation of transactionalStorage
 * allowed exceptions to bubble up
 * assume rawmessages to be serializable for transacted processing
 * added ibis42compatibility attribute, avoiding exception bubbling
 *
 * Revision 1.14  2005/07/05 12:54:38  <NAME> <<EMAIL>>
 * allow to set parameters from context for processRequest() methods
 *
 * Revision 1.13  2005/06/02 11:52:24  <NAME> <<EMAIL>>
 * limited number of actively polling threads to value of attriubte numThreadsPolling
 *
 * Revision 1.12  2005/04/13 12:53:09  <NAME> <<EMAIL>>
 * removed unused imports
 *
 * Revision 1.11  2005/03/31 08:22:49  <NAME> <<EMAIL>>
 * fixed bug in getIdleStatistics
 *
 * Revision 1.10  2005/03/07 11:04:36  <NAME> <<EMAIL>>
 * PipeLineSession became a extension of HashMap, using other iterator
 *
 * Revision 1.9  2005/03/04 08:53:29  <NAME> <<EMAIL>>
 * Fixed IndexOutOfBoundException in getProcessStatistics  due to multi threading.
 * Adjusted this too for getIdleStatistics
 *
 * Revision 1.8  2005/02/10 08:17:34  <NAME> <<EMAIL>>
 * included context dump in debug
 *
 * Revision 1.7  2005/01/13 08:56:04  <NAME> <<EMAIL>>
 * Make threadContext-attributes available in PipeLineSession
 *
 * Revision 1.6  2004/10/12 15:14:11  <NAME> <<EMAIL>>
 * removed unused code
 *
 * Revision 1.5  2004/08/25 09:11:33  unknown <<EMAIL>>
 * Add waitForRunstate with timeout
 *
 * Revision 1.4  2004/08/23 13:10:48  <NAME> <<EMAIL>>
 * updated JavaDoc
 *
 * Revision 1.3  2004/08/16 14:09:58  unknown <<EMAIL>>
 * Return returnIfStopped value in case adapter is stopped
 *
 * Revision 1.2  2004/08/09 13:46:52  <NAME> <<EMAIL>>
 * various changes
 *
 * Revision 1.1  2004/08/03 13:04:30  <NAME> <<EMAIL>>
 * introduction of GenericReceiver
 *
 */
package nl.nn.adapterframework.receivers;

import java.io.IOException;
import java.io.Serializable;
import java.net.URL;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;

import javax.transaction.Status;
import javax.transaction.UserTransaction;
import javax.xml.transform.Transformer;
import javax.xml.transform.TransformerConfigurationException;
import javax.xml.transform.TransformerException;

import nl.nn.adapterframework.configuration.ConfigurationException;
import nl.nn.adapterframework.core.HasPhysicalDestination;
import nl.nn.adapterframework.core.HasSender;
import nl.nn.adapterframework.core.IAdapter;
import nl.nn.adapterframework.core.IListener;
import nl.nn.adapterframework.core.IMessageHandler;
import nl.nn.adapterframework.core.INamedObject;
import nl.nn.adapterframework.core.IPullingListener;
import nl.nn.adapterframework.core.IPushingListener;
import nl.nn.adapterframework.core.IReceiver;
import nl.nn.adapterframework.core.IReceiverStatistics;
import nl.nn.adapterframework.core.ISender;
import nl.nn.adapterframework.core.ITransactionalStorage;
import nl.nn.adapterframework.core.IXAEnabled;
import nl.nn.adapterframework.core.IbisExceptionListener;
import nl.nn.adapterframework.core.ListenerException;
import nl.nn.adapterframework.core.PipeLineResult;
import nl.nn.adapterframework.core.PipeLineSession;
import nl.nn.adapterframework.core.SenderException;
import nl.nn.adapterframework.core.TransactionException;
import nl.nn.adapterframework.util.ClassUtils;
import nl.nn.adapterframework.util.Counter;
import nl.nn.adapterframework.util.DomBuilderException;
import nl.nn.adapterframework.util.JtaUtil;
import nl.nn.adapterframework.util.LogUtil;
import nl.nn.adapterframework.util.Misc;
import nl.nn.adapterframework.util.RunStateEnquiring;
import nl.nn.adapterframework.util.RunStateEnum;
import nl.nn.adapterframework.util.RunStateManager;
import nl.nn.adapterframework.util.Semaphore;
import nl.nn.adapterframework.util.StatisticsKeeper;
import nl.nn.adapterframework.util.TracingEventNumbers;
import nl.nn.adapterframework.util.TracingUtil;
import nl.nn.adapterframework.util.XmlUtils;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.SystemUtils;
import org.apache.commons.lang.builder.ToStringBuilder;
import org.apache.commons.lang.builder.ToStringStyle;
import org.apache.log4j.Logger;

/**
 * This {@link IReceiver Receiver} may be used as a base-class for developing receivers.
 *
 * <p><b>Configuration:</b>
 * <table border="1">
 * <tr><th>attributes</th><th>description</th><th>default</th></tr>
 * <tr><td>classname</td><td>name of the class, mostly a class that extends this class</td><td>&nbsp;</td></tr>
 * <tr><td>{@link #setName(String) name}</td>  <td>name of the receiver as known to the adapter</td><td>&nbsp;</td></tr>
 * <tr><td>{@link #setActive(boolean) active}</td>  <td>when set <code>false</code> or set to something else as "true", (even set to the empty string), the receiver is not included in the configuration</td><td>true</td></tr>
 * <tr><td>{@link #setNumThreads(int) numThreads}</td><td>the number of threads that may execute a pipeline concurrently (only for pulling listeners)</td><td>1</td></tr>
 * <tr><td>{@link #setNumThreadsPolling(int) numThreadsPolling}</td><td>the number of threads that are activily polling for messages concurrently. '0' means 'limited only by <code>numThreads</code>' (only for pulling listeners)</td><td>1</td></tr>
 * <tr><td>{@link #setStyleSheetName(String) styleSheetName}</td>  <td></td><td>&nbsp;</td></tr>
 * <tr><td>{@link #setOnError(String) onError}</td><td>one of 'continue' or 'close'. Controls the behaviour of the receiver when it encounters an error sending a reply or receives an exception asynchronously</td><td>continue</td></tr>
 * <tr><td>{@link #setTransacted(boolean) transacted}</td><td>if set to <code>true, messages will be received and processed under transaction control. If processing fails, messages will be sent to the error-sender. (see below)</code></td><td><code>false</code></td></tr>
 * <tr><td>{@link #setMaxRetries(int) maxRetries}</td><td>The number of times a pulling listening attempt is retried after an exception is caught</td><td>3</td></tr>
 * <tr><td>{@link #setIbis42compatibility(boolean) ibis42compatibility}</td><td>if set to <code>true, the result of a failed processing of a message is a formatted errormessage. Otherwise a listener specific error handling is performed</code></td><td><code>false</code></td></tr>
 * <tr><td>{@link #setBeforeEvent(int) beforeEvent}</td>      <td>METT eventnumber, fired just before a message is processed by this Receiver</td><td>-1 (disabled)</td></tr>
 * <tr><td>{@link #setAfterEvent(int) afterEvent}</td>        <td>METT eventnumber, fired just after message processing by this Receiver is finished</td><td>-1 (disabled)</td></tr>
 * <tr><td>{@link #setExceptionEvent(int) exceptionEvent}</td><td>METT eventnumber, fired when message processing by this Receiver resulted in an exception</td><td>-1 (disabled)</td></tr>
 * </table>
 * </p>
 * <p>
 * <table border="1">
 * <tr><th>nested elements (accessible in descender-classes)</th><th>description</th></tr>
 * <tr><td>{@link nl.nn.adapterframework.core.IPullingListener listener}</td><td>the listener used to receive messages from</td></tr>
 * <tr><td>{@link nl.nn.adapterframework.core.ITransactionalStorage inProcessStorage}</td><td>mandatory for {@link #setTransacted(boolean) transacted} receivers: place to store messages during processing.</td></tr>
 * <tr><td>{@link nl.nn.adapterframework.core.ITransactionalStorage errorStorage}</td><td>optional for {@link #setTransacted(boolean) transacted} receivers: place to store messages if message processing has gone wrong. If no errorStorage is specified, the inProcessStorage is used for errorStorage</td></tr>
 * <tr><td>{@link nl.nn.adapterframework.core.ISender errorSender}</td><td>optional for {@link #setTransacted(boolean) transacted} receviers: 
 * will be called to store messages that failed to process. If no errorSender is specified, failed messages will remain in inProcessStorage</td></tr>
 * </table>
 * </p>
 * <p><b>Transaction control</b><br>
 * If {@link #setTransacted(boolean) transacted} is set to <code>true, messages will be received and processed under transaction control.
 * This means that after a message has been read and processed and the transaction has ended, one of the following apply:
 * <ul>
 * <table border="1">
 * <tr><th>situation</th><th>input listener</th><th>Pipeline</th><th>inProcess storage</th><th>errorSender</th><th>summary of effect</th></tr>
 * <tr><td>successful</td><td>message read and committed</td><td>message processed</td><td>unchanged</td><td>unchanged</td><td>message processed</td></tr>
 * <tr><td>procesing failed</td><td>message read and committed</td><td>message processing failed and rolled back</td><td>unchanged</td><td>message sent</td><td>message only transferred from listener to errroSender</td></tr>
 * <tr><td>listening failed</td><td>unchanged: listening rolled back</td><td>no processing performed</td><td>unchanged</td><td>unchanged</td><td>no changes, input message remains on input available for listener</td></tr>
 * <tr><td>transfer to inprocess storage failed</td><td>unchanged: listening rolled back</td><td>no processing performed</td><td>unchanged</td><td>unchanged</td><td>no changes, input message remains on input available for listener</td></tr>
 * <tr><td>transfer to errorSender failed</td><td>message read and committed</td><td>message processing failed and rolled back</td><td>message present</td><td>unchanged</td><td>message only transferred from listener to inProcess storage</td></tr>
 * </table> 
 * If the application or the server crashes in the middle of one or more transactions, these transactions 
 * will be recovered and rolled back after the server/application is restarted. Then allways exactly one of 
 * the following applies for any message touched at any time by Ibis by a transacted receiver:
 * <ul>
 * <li>It is processed correctly by the pipeline and removed from the input-queue, 
 *     not present in inProcess storage and not send to the errorSender</li> 
 * <li>It is not processed at all by the pipeline, or processing by the pipeline has been rolled back; 
 *     the message is removed from the input queue and either (one of) still in inProcess storage <i>or</i> sent to the errorSender</li>
 * </ul>
 * </p>
 *
 * <p><b>commit or rollback</b><br>
 * If {@link #setTransacted(boolean) transacted} is set to <code>true</code>, messages will be either committed or rolled back.
 * All message-processing transactions are committed, unless one or more of the following apply:
 * <ul>
 * <li>The PipeLine is transacted and the exitState of the pipeline is not equal to {@link nl.nn.adapterframework.core.PipeLine#setCommitOnState(String) commitOnState} (that defaults to 'success')</li>
 * <li>a PipeRunException or another runtime-exception has been thrown by any Pipe or by the PipeLine</li>
 * <li>the setRollBackOnly() method has been called on the userTransaction (not accessible by Pipes)</li>
 * </ul>
 * </p>
 *
 * @version Id
 * @author     <NAME>
 * @since 4.2
 */
public class ReceiverBase implements IReceiver, IReceiverStatistics, Runnable, IMessageHandler, IbisExceptionListener, HasSender, TracingEventNumbers {
	public static final String version="$RCSfile: ReceiverBase.java,v $ $Revision: 1.41 $ $Date: 2007-06-26 06:56:59 $";
	protected Logger log = LogUtil.getLogger(this);
 
	private String returnIfStopped="";
	private String fileNameIfStopped = null;
	private String replaceFrom = null;
	private String replaceTo = null;
	private String styleSheetName = null;

	public static final String ONERROR_CONTINUE = "continue";
	public static final String ONERROR_CLOSE = "close";

	private boolean active=true;

  	private String name;
  	private String onError = ONERROR_CONTINUE; 
    protected RunStateManager runState = new RunStateManager();
    
    private boolean ibis42compatibility=false;

	// the number of threads that may execute a pipeline concurrently (only for pulling listeners)
	private int numThreads = 1;
	// the number of threads that are activily polling for messages (concurrently, only for pulling listeners)
	private int numThreadsPolling = 1;
   
	private Counter threadsProcessing = new Counter(0);
	private Counter threadsRunning = new Counter(0);
	private Semaphore pollToken = null;
	        
	// number of messages received
    private Counter numReceived = new Counter(0);
	private ArrayList processStatistics = new ArrayList();
	private ArrayList idleStatistics = new ArrayList();

    // the adapter that handles the messages and initiates this listener
    private IAdapter adapter;

	private IListener listener;
    private ITransactionalStorage inProcessStorage=null;
    private ISender errorSender=null;
	private ITransactionalStorage errorStorage=null;
	private ISender sender=null; // answer-sender
	private ITransactionalStorage messageLog=null;
	
	private int maxRetries=3;
	private Counter retryCount = new Counter(0);
    
    private boolean transacted=false;
 
 	// METT event numbers
	private int beforeEvent=-1;
	private int afterEvent=-1;
	private int exceptionEvent=-1;

	int retryInterval=1;
    
	protected String getLogPrefix() {
		return "Receiver ["+getName()+"] "; 
	}	

	/** 
	 * sends an informational message to the log and to the messagekeeper of the adapter
	 */
	protected void info(String msg) {
		log.info(msg);
		if (adapter != null)
			adapter.getMessageKeeper().add(msg);
	}

	/** 
	 * sends a warning to the log and to the messagekeeper of the adapter
	 */
	protected void warn(String msg) {
		log.warn(msg);
		if (adapter != null)
			adapter.getMessageKeeper().add("WARNING: " + msg);
	}

	/** 
	 * sends a warning to the log and to the messagekeeper of the adapter
	 */
	protected void error(String msg, Throwable t) {
		log.error(msg, t);
		if (adapter != null)
			adapter.getMessageKeeper().add("ERROR: " + msg+": "+t.getMessage());
	}


	protected void openAllResources() throws ListenerException {	
		// on exit resouces must be in a state that runstate is or can be set to 'STARTED'
		try {
			if (getSender()!=null) {
				getSender().open();
			}
			if (getErrorSender()!=null) {
				getErrorSender().open();
			}
			if (getInProcessStorage()!=null) {
				getInProcessStorage().open();
			}
			if (getMessageLog()!=null) {
				getMessageLog().open();
			}
			if (isTransacted()) {
				getAdapter().getUserTransaction();
			}
		} catch (SenderException e) {
			throw new ListenerException(e);
		} catch (TransactionException e) {
			throw new ListenerException(e);
		}
		getListener().open();
		if (getListener() instanceof IPullingListener){
			// start all threads
			if (getNumThreads() > 1) {
				for (int i = 1; i <= getNumThreads(); i++) {
					addThread("[" + i+"]");
				}
			} else {
				addThread(null);
			}
		}
	}

	private void addThread(String nameSuffix) {
		if (getListener() instanceof IPullingListener){
			Thread t = new Thread(this, getName() + (nameSuffix==null ? "" : nameSuffix));
			t.start();
		}
	}


	protected void tellResourcesToStop() throws ListenerException {
		 // must lead to a 'closeAllResources()'
		 // runstate is 'STOPPING'
		 // default just calls 'closeAllResources()'
		 if (getListener() instanceof IPushingListener) {
			closeAllResources();
		 }
		 // IPullingListeners stop as their threads finish, as the runstate is set to stopping
	}
	protected void closeAllResources() {
		// on exit resouces must be in a state that runstate can be set to 'STOPPED'
		try {
			log.debug("closing Receiver ["+ getName()+ "]");
			getListener().close();
			if (getSender()!=null) {
				getSender().close();
			}
			if (getErrorSender()!=null) {
				getErrorSender().close();
			}
			if (getInProcessStorage()!=null) {
				getInProcessStorage().close();
			}
			if (getErrorStorage()!=null && getErrorStorage()!=getInProcessStorage()) {
				getInProcessStorage().close();
			}
			if (getMessageLog()!=null) {
				getMessageLog().close();
			}
	
			log.info("closed Receiver ["+ getName()+ "]");
		} catch (Exception e) {
			log.error(
				"Receiver [" + getName()+ "]: error closing connection", e);
		}
		runState.setRunState(RunStateEnum.STOPPED);
		info("Receiver [" + getName() + "] stopped");
	}
	 
	protected void propagateName() {
		IListener listener=getListener();
		if (listener!=null && StringUtils.isEmpty(listener.getName())) {
			listener.setName("listener of ["+getName()+"]");
		}
		ITransactionalStorage inProcess = getInProcessStorage();
		if (inProcess != null) {
			inProcess.setName("inProcessStorage of ["+getName()+"]");
		}
		ISender errorSender = getErrorSender();
		if (errorSender != null) {
			errorSender.setName("errorSender of ["+getName()+"]");
		}
		ITransactionalStorage errorStorage = getErrorStorage();
		if (errorStorage != null && errorStorage != inProcess) {
			errorStorage.setName("errorStorage of ["+getName()+"]");
		}
	}

	public void configure() throws ConfigurationException {		
 		try {
 			propagateName();
			if (getListener()==null) {
				throw new ConfigurationException("Receiver ["+getName()+"] has no listener");
			}
			if (getListener() instanceof IPushingListener) {
				IPushingListener pl = (IPushingListener)getListener();
				pl.setHandler(this);
				pl.setExceptionListener(this);
			}
			if (getListener() instanceof IPullingListener) {
				if (getNumThreadsPolling()>0 && getNumThreadsPolling()<getNumThreads()) {
					pollToken = new Semaphore(getNumThreadsPolling());
				}
			}
			getListener().configure();
			if (getListener() instanceof HasPhysicalDestination) {
				info("Receiver ["+getName()+"] has listener on "+((HasPhysicalDestination)getListener()).getPhysicalDestinationName());
			}
			if (getListener() instanceof HasSender) {
				// only informational
				ISender sender = ((HasSender)getListener()).getSender();
				if (sender instanceof HasPhysicalDestination) {
					info("Listener of receiver ["+getName()+"] has answer-sender on "+((HasPhysicalDestination)sender).getPhysicalDestinationName());
				}
			}
			ISender sender = getSender();
			if (sender!=null) {
				sender.configure();
				if (sender instanceof HasPhysicalDestination) {
					info("receiver ["+getName()+"] has answer-sender on "+((HasPhysicalDestination)sender).getPhysicalDestinationName());
				}
			}
			ITransactionalStorage inProcessStorage = getInProcessStorage();
			if (inProcessStorage!=null) {
				inProcessStorage.configure();
				if (inProcessStorage instanceof HasPhysicalDestination) {
					info("Receiver ["+getName()+"] has inProcessStorage in "+((HasPhysicalDestination)inProcessStorage).getPhysicalDestinationName());
				}
			}
			ISender errorSender = getErrorSender();
			if (errorSender!=null) {
				errorSender.configure();
				if (errorSender instanceof HasPhysicalDestination) {
					info("Receiver ["+getName()+"] has errorSender to "+((HasPhysicalDestination)errorSender).getPhysicalDestinationName());
				}
			}
			ITransactionalStorage errorStorage = getErrorStorage();
			if (errorStorage!=null && errorStorage != getInProcessStorage()) {
				errorStorage.configure();
				if (errorStorage instanceof HasPhysicalDestination) {
					info("Receiver ["+getName()+"] has errorStorage to "+((HasPhysicalDestination)errorStorage).getPhysicalDestinationName());
				}
			}
			ITransactionalStorage messageLog = getMessageLog();
			if (messageLog!=null) {
				messageLog.configure();
				if (messageLog instanceof HasPhysicalDestination) {
					info("Receiver ["+getName()+"] has messageLog in "+((HasPhysicalDestination)messageLog).getPhysicalDestinationName());
				}
			}
			if (isTransacted()) {
//				if (!(getListener() instanceof IXAEnabled && ((IXAEnabled)getListener()).isTransacted())) {
//					warn("Receiver ["+getName()+"] sets transacted=true, but listener not. Transactional integrity is not guaranteed"); 
//				}
				if (inProcessStorage==null && getListener() instanceof IPullingListener) {
//					throw new ConfigurationException("Receiver ["+getName()+"] sets transacted=true, but has no inProcessStorage.");
					warn("Receiver ["+getName()+"] sets transacted=true, but has no inProcessStorage. Transactional integrity is not guaranteed");
//				} else {
//					if (!(inProcessStorage instanceof IXAEnabled && ((IXAEnabled)inProcessStorage).isTransacted())) {
//						warn("Receiver ["+getName()+"] sets transacted=true, but inProcessStorage not. Transactional integrity is not guaranteed"); 
//					}
				}
				
				if (errorSender==null && errorStorage==null) {
					warn("Receiver ["+getName()+"] sets transacted=true, but has no errorSender or errorStorage. Messages processed with errors will be lost");
				} else {
//					if (errorSender!=null && !(errorSender instanceof IXAEnabled && ((IXAEnabled)errorSender).isTransacted())) {
//						warn("Receiver ["+getName()+"] sets transacted=true, but errorSender is not. Transactional integrity is not guaranteed"); 
//					}
//					if (errorStorage!=null && !(errorStorage instanceof IXAEnabled && ((IXAEnabled)errorStorage).isTransacted())) {
//						warn("Receiver ["+getName()+"] sets transacted=true, but errorStorage is not. Transactional integrity is not guaranteed"); 
//					}
					if (errorStorage==inProcessStorage) {
						info("Receiver ["+getName()+"] has errorStorage in inProcessStorage, setting inProcessStorage's type to 'errorStorage'");
						errorStorage.setType("E"); 
					}
				}
			} 
{'fraction_non_alphanumeric': 0.13270425966839422, 'fraction_numerical': 0.030455699861658855, 'mean_word_length': 4.707955215085445, 'pattern_counts': {'":': 2, '<': 802, '<?xml version=': 0, '>': 802, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 6}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '136307', 'n_tokens_mistral': 16421, 'n_tokens_neox': 14503, 'n_words': 4133}