path: root/feature-session-persistence/src/main/java/org/onap/policy/drools/persistence/PersistenceFeature.java

/*-
 * ============LICENSE_START=======================================================
 * feature-session-persistence
 * ================================================================================
 * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
 * ================================================================================
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *      http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * ============LICENSE_END=========================================================
 */

package org.onap.policy.drools.persistence;

import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import javax.persistence.EntityManagerFactory;
import javax.persistence.Persistence;

import org.eclipse.persistence.config.PersistenceUnitProperties;
import org.kie.api.KieServices;
import org.kie.api.runtime.Environment;
import org.kie.api.runtime.EnvironmentName;
import org.kie.api.runtime.KieSession;
import org.kie.api.runtime.KieSessionConfiguration;
import org.onap.policy.drools.core.PolicyContainer;
import org.onap.policy.drools.core.PolicySession;
import org.onap.policy.drools.core.PolicySessionFeatureAPI;
import org.onap.policy.drools.features.PolicyEngineFeatureAPI;
import org.onap.policy.drools.system.PolicyController;
import org.onap.policy.drools.system.PolicyEngine;
import org.onap.policy.drools.utils.PropertyUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import bitronix.tm.BitronixTransactionManager;
import bitronix.tm.Configuration;
import bitronix.tm.TransactionManagerServices;
import bitronix.tm.resource.jdbc.PoolingDataSource;

/**
 * If this feature is supported, there is a single instance of it. It adds
 * persistence to Drools sessions. In addition, if an active-standby feature
 * exists, then that is used to determine the active and last-active PDP. If it
 * does not exist, then the current host name is used as the PDP id.
 *
 * The bulk of the code here was once in other classes, such as
 * 'PolicyContainer' and 'Main'. It was moved here as part of making this a
 * separate optional feature.
 */
public class PersistenceFeature implements PolicySessionFeatureAPI, PolicyEngineFeatureAPI {

	private static final Logger logger = LoggerFactory.getLogger(PersistenceFeature.class);


	/**
	 * Standard factory used to get various items.
	 */
	private static Factory stdFactory = new Factory();

	/**
	 * Factory used to get various items.
	 */
	private Factory fact = stdFactory;

	/**
	 * KieService factory.
	 */
	private KieServices kieSvcFact;

	/**
	 * Host name.
	 */
	private String hostName;

	/**
	 * Persistence properties.
	 */
	private Properties persistProps;

	/**
	 * Whether or not the SessionInfo records should be cleaned out.
	 */
	private boolean sessInfoCleaned;

	/**
	 * SessionInfo timeout, in milli-seconds, as read from
	 * {@link #persistProps}.
	 */
	private long sessionInfoTimeoutMs;

	/**
	 * Object used to serialize cleanup of sessioninfo table.
	 */
	private Object cleanupLock = new Object();

	/**
	 * Sets the factory to be used during junit testing.
	 * 
	 * @param fact
	 *            factory to be used
	 */
	protected void setFactory(Factory fact) {
		this.fact = fact;
	}

	/**
	 * Lookup the adjunct for this feature that is associated with the specified
	 * PolicyContainer. If not found, create one.
	 * 
	 * @param policyContainer
	 *            the container whose adjunct we are looking up, and possibly
	 *            creating
	 * @return the associated 'ContainerAdjunct' instance, which may be new
	 */
	private ContainerAdjunct getContainerAdjunct(PolicyContainer policyContainer) {

		Object rval = policyContainer.getAdjunct(this);

		if (rval == null || !(rval instanceof ContainerAdjunct)) {
			// adjunct does not exist, or has the wrong type (should never
			// happen)
			rval = new ContainerAdjunct(policyContainer);
			policyContainer.setAdjunct(this, rval);
		}

		return (ContainerAdjunct) rval;
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public int getSequenceNumber() {
		return 1;
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public void globalInit(String args[], String configDir) {
		
		kieSvcFact = fact.getKieServices();

		initHostName();

		try {
			persistProps = fact.loadProperties(configDir + "/feature-session-persistence.properties");

		} catch (IOException e1) {
			logger.error("initializePersistence: ", e1);
		}

		sessionInfoTimeoutMs = getPersistenceTimeout();

		Configuration bitronixConfiguration = fact.getTransMgrConfig();
		bitronixConfiguration.setJournal(null);
		bitronixConfiguration.setServerId(hostName);
	}

	/**
	 * Creates a persistent KieSession, loading it from the persistent store, or
	 * creating one, if it does not exist yet.
	 */
	@Override
	public KieSession activatePolicySession(PolicyContainer policyContainer, String name, String kieBaseName) {

		if (isPersistenceEnabled(policyContainer, name)) {
			cleanUpSessionInfo();

			return getContainerAdjunct(policyContainer).newPersistentKieSession(name, kieBaseName);
		}

		return null;
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public PolicySession.ThreadModel selectThreadModel(PolicySession session) {
		PolicyContainer policyContainer = session.getPolicyContainer();
		if (isPersistenceEnabled(policyContainer, session.getName())) {
			return new PersistentThreadModel(session, getProperties(policyContainer));
		}
		return null;
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public void disposeKieSession(PolicySession policySession) {
		
		ContainerAdjunct contAdj = (ContainerAdjunct) policySession.getPolicyContainer().getAdjunct(this);
		if(contAdj != null) {
			contAdj.disposeKieSession( policySession.getName());
		}
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public void destroyKieSession(PolicySession policySession) {

		ContainerAdjunct contAdj = (ContainerAdjunct) policySession.getPolicyContainer().getAdjunct(this);
		if(contAdj != null) {
			contAdj.destroyKieSession( policySession.getName());
		}
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public boolean afterStart(PolicyEngine engine) {
		return false;
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public boolean beforeStart(PolicyEngine engine) {
		synchronized (cleanupLock) {
			sessInfoCleaned = false;
		}

		return false;
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public boolean beforeActivate(PolicyEngine engine) {
		synchronized (cleanupLock) {
			sessInfoCleaned = false;
		}

		return false;
	}

	/**
	 * {@inheritDoc}
	 */
	@Override
	public boolean afterActivate(PolicyEngine engine) {
		return false;
	}

	/* ============================================================ */

	/**
	 * Gets the persistence timeout value for sessioninfo records.
	 * 
	 * @return the timeout value, in milli-seconds, or {@code -1} if it is
	 *         unspecified or invalid
	 */
	private long getPersistenceTimeout() {
		String timeoutString = null;

		try {
			timeoutString = persistProps.getProperty(DroolsPersistenceProperties.DB_SESSIONINFO_TIMEOUT);

			if (timeoutString != null) {
				// timeout parameter is specified
				return Long.valueOf(timeoutString) * 1000;
			}

		} catch (NumberFormatException e) {
			logger.error("Invalid value for Drools persistence property persistence.sessioninfo.timeout: {}",
					 timeoutString, e);
		}

		return -1;
	}

	/**
	 * Initializes {@link #hostName}.
	 */
	private void initHostName() {

		try {
			hostName = fact.getHostName();

		} catch (UnknownHostException e) {
			throw new RuntimeException("cannot determine local hostname", e);
		}
	}

	/* ============================================================ */

	/**
	 * Each instance of this class is a logical extension of a 'PolicyContainer'
	 * instance. Its reference is stored in the 'adjuncts' table within the
	 * 'PolicyContainer', and will be garbage-collected with the container.
	 */
	protected class ContainerAdjunct {
		/**
		 * 'PolicyContainer' instance that this adjunct is extending.
		 */
		private PolicyContainer policyContainer;
		
		/**
		 * Maps a KIE session name to its data source.
		 */
		private Map<String,PoolingDataSource> name2ds = new HashMap<>();

		/**
		 * Constructor - initialize a new 'ContainerAdjunct'
		 *
		 * @param policyContainer
		 *            the 'PolicyContainer' instance this adjunct is extending
		 */
		private ContainerAdjunct(PolicyContainer policyContainer) {
			this.policyContainer = policyContainer;
		}

		/**
		 * Create a new persistent KieSession. If there is already a
		 * corresponding entry in the database, it is used to initialize the
		 * KieSession. If not, a completely new session is created.
		 * 
		 * @param name
		 *            the name of the KieSession (which is also the name of the
		 *            associated PolicySession)
		 * @param kieBaseName
		 *            the name of the 'KieBase' instance containing this session
		 * @return a new KieSession with persistence enabled
		 */
		private KieSession newPersistentKieSession(String name, String kieBaseName) {

			long desiredSessionId;

			DroolsSessionConnector conn = getDroolsSessionConnector("onapPU");

			desiredSessionId = getSessionId(conn, name);

			logger.info("\n\nThis controller is primary... coming up with session {} \n\n", desiredSessionId);

			// session does not exist -- attempt to create one
			logger.info("getPolicySession:session does not exist -- attempt to create one with name {}", name);

			System.getProperties().put("java.naming.factory.initial", "bitronix.tm.jndi.BitronixInitialContextFactory");

			Environment env = kieSvcFact.newEnvironment();
			String dsName = loadDataSource(name);

			configureKieEnv(name, env, dsName);

			KieSessionConfiguration kConf = kieSvcFact.newKieSessionConfiguration();

			KieSession kieSession = desiredSessionId >= 0 ? loadKieSession(kieBaseName, desiredSessionId, env, kConf)
					: null;

			if (kieSession == null) {
				// loadKieSession() returned null or desiredSessionId < 0
				logger.info("LOADING We cannot load session {}. Going to create a new one", desiredSessionId);

				kieSession = newKieSession(kieBaseName, env);
			}

			replaceSession(conn, name, kieSession);

			return kieSession;
		}

		/**
		 * Loads a data source into {@link #name2ds}, if one doesn't exist
		 * yet.
		 * @param sessName		session name
		 * @return the unique data source name
		 */
		private String loadDataSource(String sessName) {
			PoolingDataSource ds = name2ds.get(sessName);
			
			if(ds == null) {
				Properties props = new Properties();
				addOptProp(props, "URL", persistProps.getProperty(DroolsPersistenceProperties.DB_URL));
				addOptProp(props, "user", persistProps.getProperty(DroolsPersistenceProperties.DB_USER));
				addOptProp(props, "password", persistProps.getProperty(DroolsPersistenceProperties.DB_PWD));

				ds = fact.makePoolingDataSource();
				ds.setUniqueName("jdbc/BitronixJTADataSource/" + sessName);
				ds.setClassName(persistProps.getProperty(DroolsPersistenceProperties.DB_DATA_SOURCE));
				ds.setMaxPoolSize(3);
				ds.setIsolationLevel("SERIALIZABLE");
				ds.setAllowLocalTransactions(true);
				ds.getDriverProperties().putAll(props);
				ds.init();
				
				name2ds.put(sessName, ds);
			}
			
			return ds.getUniqueName();
		}

		/**
		 * Configures a Kie Environment
		 * 
		 * @param name
		 * 				session name
		 * @param env
		 * 				environment to be configured
		 * @param dsName 
		 * 				data source name
		 */
		private void configureKieEnv(String name, Environment env, String dsName) {
			Properties emfProperties = new Properties();
			emfProperties.setProperty(PersistenceUnitProperties.JTA_DATASOURCE, dsName);
			
			EntityManagerFactory emfact = fact.makeEntMgrFact("onapsessionsPU", emfProperties);

			env.set(EnvironmentName.ENTITY_MANAGER_FACTORY, emfact);
			env.set(EnvironmentName.TRANSACTION_MANAGER, fact.getTransMgr());
		}

		/**
		 * Loads an existing KieSession from the persistent store.
		 * 
		 * @param kieBaseName
		 *            the name of the 'KieBase' instance containing this session
		 * @param desiredSessionId
		 *            id of the desired KieSession
		 * @param env
		 *            Kie Environment for the session
		 * @param kConf
		 *            Kie Configuration for the session
		 * @return the persistent session, or {@code null} if it could not be
		 *         loaded
		 */
		private KieSession loadKieSession(String kieBaseName, long desiredSessionId, Environment env,
				KieSessionConfiguration kConf) {
			try {
				KieSession kieSession = kieSvcFact.getStoreServices().loadKieSession(desiredSessionId,
						policyContainer.getKieContainer().getKieBase(kieBaseName), kConf, env);

				logger.info("LOADING Loaded session {}", desiredSessionId);

				return kieSession;

			} catch (Exception e) {
				logger.error("loadKieSession error: ", e);
				return null;
			}
		}

		/**
		 * Creates a new, persistent KieSession.
		 * 
		 * @param kieBaseName
		 *            the name of the 'KieBase' instance containing this session
		 * @param env
		 *            Kie Environment for the session
		 * @return a new, persistent session
		 */
		private KieSession newKieSession(String kieBaseName, Environment env) {
			KieSession kieSession = kieSvcFact.getStoreServices()
					.newKieSession(policyContainer.getKieContainer().getKieBase(kieBaseName), null, env);

			logger.info("LOADING CREATED {}", kieSession.getIdentifier());

			return kieSession;
		}

		/**
		 * Closes the data source associated with a session.
		 * @param name	name of the session being destroyed
		 */
		private void destroyKieSession(String name) {
			closeDataSource(name);
		}

		/**
		 * Closes the data source associated with a session.
		 * @param name	name of the session being disposed of
		 */
		private void disposeKieSession(String name) {
			closeDataSource(name);
		}

		/**
		 * Closes the data source associated with a session.
		 * @param name	name of the session whose data source is to be closed
		 */
		private void closeDataSource(String name) {
			PoolingDataSource ds = name2ds.remove(name);
			if(ds != null) {
				ds.close();
			}
		}
	}

	/* ============================================================ */

	/**
	 * Removes "old" Drools 'sessioninfo' records, so they aren't used to
	 * restore data to Drools sessions. This also has the useful side-effect of
	 * removing abandoned records as well.
	 */
	private void cleanUpSessionInfo() {

		synchronized (cleanupLock) {

			if (sessInfoCleaned) {
				logger.info("Clean up of sessioninfo table: already done");
				return;
			}

			if (sessionInfoTimeoutMs < 0) {
				logger.info("Clean up of sessioninfo table: no timeout specified");
				return;
			}

			// get DB connection properties
			String url = persistProps.getProperty(DroolsPersistenceProperties.DB_URL);
			String user = persistProps.getProperty(DroolsPersistenceProperties.DB_USER);
			String password = persistProps.getProperty(DroolsPersistenceProperties.DB_PWD);

			if (url == null || user == null || password == null) {
				logger.error("Missing DB properties for clean up of sessioninfo table");
				return;
			}

			// now do the record deletion
			try (Connection connection = fact.makeDbConnection(url, user, password);
					PreparedStatement statement = connection.prepareStatement(
							"DELETE FROM sessioninfo WHERE timestampdiff(second,lastmodificationdate,now()) > ?")) {
				statement.setLong(1, sessionInfoTimeoutMs/1000);

				int count = statement.executeUpdate();
				logger.info("Cleaning up sessioninfo table -- {} records removed", count);

			} catch (SQLException e) {
				logger.error("Clean up of sessioninfo table failed", e);
			}

			// TODO: delete DroolsSessionEntity where sessionId not in
			// (sessinfo.xxx)

			sessInfoCleaned = true;
		}
	}

	/**
	 * Gets a connector for manipulating DroolsSession objects within the
	 * persistent store.
	 * 
	 * @param pu
	 * @return a connector for DroolsSession objects
	 */
	private DroolsSessionConnector getDroolsSessionConnector(String pu) {

		Properties propMap = new Properties();
		addOptProp(propMap, "javax.persistence.jdbc.driver",
				persistProps.getProperty(DroolsPersistenceProperties.DB_DRIVER));
		addOptProp(propMap, "javax.persistence.jdbc.url", persistProps.getProperty(DroolsPersistenceProperties.DB_URL));
		addOptProp(propMap, "javax.persistence.jdbc.user",
				persistProps.getProperty(DroolsPersistenceProperties.DB_USER));
		addOptProp(propMap, "javax.persistence.jdbc.password",
				persistProps.getProperty(DroolsPersistenceProperties.DB_PWD));

		return fact.makeJpaConnector(pu, propMap);
	}

	/**
	 * Adds an optional property to a set of properties.
	 * @param propMap	map into which the property should be added
	 * @param name		property name
	 * @param value		property value, or {@code null} if it should not
	 * 					be added
	 */
	private void addOptProp(Properties propMap, String name, String value) {
		if (value != null) {
			propMap.put(name, value);
		}
	}

	/**
	 * Gets a session's ID from the persistent store.
	 * 
	 * @param conn
	 *            persistence connector
	 * @param sessnm
	 *            name of the session
	 * @return the session's id, or {@code -1} if the session is not found
	 */
	private long getSessionId(DroolsSessionConnector conn, String sessnm) {
		DroolsSession sess = conn.get(sessnm);
		return sess != null ? sess.getSessionId() : -1;
	}

	/**
	 * Replaces a session within the persistent store, if it exists.  Adds
	 * it otherwise.
	 * 
	 * @param conn
	 *            persistence connector
	 * @param sessnm
	 *            name of session to be updated
	 * @param kieSession
	 *            new session information
	 */
	private void replaceSession(DroolsSessionConnector conn, String sessnm, KieSession kieSession) {

		DroolsSessionEntity sess = new DroolsSessionEntity();

		sess.setSessionName(sessnm);
		sess.setSessionId(kieSession.getIdentifier());

		conn.replace(sess);
	}

	/**
	 * Determine whether persistence is enabled for a specific container
	 * 
	 * @param container
	 *            container to be checked
	 * @param sessionName
	 *            name of the session to be checked
	 * @return {@code true} if persistence is enabled for this container, and
	 *         {@code false} if not
	 */
	private boolean isPersistenceEnabled(PolicyContainer container, String sessionName) {
		Properties properties = getProperties(container);
		boolean rval = false;

		if (properties != null) {
			// fetch the 'type' property
			String type = getProperty(properties, sessionName, "type");
			rval = "auto".equals(type) || "native".equals(type);
		}

		return rval;
	}

	/**
	 * Determine the controller properties associated with the policy container.
	 * 
	 * @param container
	 *            container whose properties are to be retrieved
	 * @return the container's properties, or {@code null} if not found
	 */
	private Properties getProperties(PolicyContainer container) {
		try {
			return fact.getPolicyContainer(container).getProperties();
		} catch (IllegalArgumentException e) {
			logger.error("getProperties exception: ", e);
			return null;
		}
	}

	/**
	 * Fetch the persistence property associated with a session. The name may
	 * have the form:
	 * <ul>
	 * <li>persistence.SESSION-NAME.PROPERTY</li>
	 * <li>persistence.PROPERTY</li>
	 * </ul>
	 * 
	 * @param properties
	 *            properties from which the value is to be retrieved
	 * @param sessionName
	 *            session name of interest
	 * @param property
	 *            property name of interest
	 * @return the property value, or {@code null} if not found
	 */
	private String getProperty(Properties properties, String sessionName, String property) {
		String value = properties.getProperty("persistence." + sessionName + "." + property);
		if (value == null) {
			value = properties.getProperty("persistence." + property);
		}

		return value;
	}

	/* ============================================================ */

	/**
	 * This 'ThreadModel' variant periodically calls
	 * 'KieSession.fireAllRules()', because the 'fireUntilHalt' method isn't
	 * compatible with persistence.
	 */
	public class PersistentThreadModel implements Runnable, PolicySession.ThreadModel {
		
		/**
		 * Session associated with this persistent thread.
		 */
		private final PolicySession session;

		/**
		 * The session thread.
		 */
		private final Thread thread;
		
		/**
		 * Used to indicate that processing should stop.
		 */
		private final CountDownLatch stopped = new CountDownLatch(1);
		
		/**
		 * Minimum time, in milli-seconds, that the thread should sleep
		 * before firing rules again.
		 */
		long minSleepTime = 100;
		
		/**
		 * Maximum time, in milli-seconds, that the thread should sleep
		 * before firing rules again.  This is a "half" time, so that
		 * we can multiply it by two without overflowing the word size.
		 */
		long halfMaxSleepTime = 5000L / 2L;

		/**
		 * Constructor - initialize variables and create thread
		 *
		 * @param session
		 *            the 'PolicySession' instance
		 * @param properties
		 *            may contain additional session properties
		 */
		public PersistentThreadModel(PolicySession session, Properties properties) {
			this.session = session;
			this.thread = new Thread(this, getThreadName());
			
			if (properties == null) {
				return;
			}
			
			// extract 'minSleepTime' and/or 'maxSleepTime'
			String name = session.getName();

			// fetch 'minSleepTime' value, and update if defined
			String sleepTimeString = getProperty(properties, name, "minSleepTime");
			if (sleepTimeString != null) {
				try {
					minSleepTime = Math.max(1, Integer.valueOf(sleepTimeString));
				} catch (Exception e) {
					logger.error(sleepTimeString + ": Illegal value for 'minSleepTime'", e);
				}
			}

			// fetch 'maxSleepTime' value, and update if defined
			long maxSleepTime = 2 * halfMaxSleepTime;
			sleepTimeString = getProperty(properties, name, "maxSleepTime");
			if (sleepTimeString != null) {
				try {
					maxSleepTime = Math.max(1, Integer.valueOf(sleepTimeString));
				} catch (Exception e) {
					logger.error(sleepTimeString + ": Illegal value for 'maxSleepTime'", e);
				}
			}

			// swap values if needed
			if (minSleepTime > maxSleepTime) {
				logger.error("minSleepTime(" + minSleepTime + ") is greater than maxSleepTime(" + maxSleepTime
						+ ") -- swapping");
				long tmp = minSleepTime;
				minSleepTime = maxSleepTime;
				maxSleepTime = tmp;
			}
			
			halfMaxSleepTime = Math.max(1, maxSleepTime/2);
		}

		/**
		 * @return the String to use as the thread name
		 */
		private String getThreadName() {
			return "Session " + session.getFullName() + " (persistent)";
		}

		/***************************/
		/* 'ThreadModel' interface */
		/***************************/

		/**
		 * {@inheritDoc}
		 */
		@Override
		public void start() {
			thread.start();
		}

		/**
		 * {@inheritDoc}
		 */
		@Override
		public void stop() {
			// tell the thread to stop
			stopped.countDown();
			
			// wait up to 10 seconds for the thread to stop
			try {
				thread.join(10000);
				
			} catch (InterruptedException e) {
				logger.error("stopThread exception: ", e);
				Thread.currentThread().interrupt();
			}
			
			// verify that it's done
			if(thread.isAlive()) {
				logger.error("stopThread: still running");
			}
		}

		/**
		 * {@inheritDoc}
		 */
		@Override
		public void updated() {
			// the container artifact has been updated -- adjust the thread name
			thread.setName(getThreadName());
		}

		/************************/
		/* 'Runnable' interface */
		/************************/

		/**
		 * {@inheritDoc}
		 */
		@Override
		public void run() {
			logger.info("PersistentThreadModel running");
			
			// set thread local variable
			session.setPolicySession();

			KieSession kieSession = session.getKieSession();
			long sleepTime = 2 * halfMaxSleepTime;

			// We want to continue, despite any exceptions that occur
			// while rules are fired.
			
			for(;;) {
			
				try {
					if (kieSession.fireAllRules() > 0) {
						// some rules fired -- reduce poll delay
						sleepTime = Math.max(minSleepTime, sleepTime/2);
					} else {
						// no rules fired -- increase poll delay
						sleepTime = 2 * Math.min(halfMaxSleepTime, sleepTime);
					}
				} catch (Exception | LinkageError e) {
					logger.error("Exception during kieSession.fireAllRules", e);
				}				
			
				try {
					if(stopped.await(sleepTime, TimeUnit.MILLISECONDS)) {
						break;
					}
					
				} catch (InterruptedException e) {
					logger.error("startThread exception: ", e);
					Thread.currentThread().interrupt();
					break;
				}
			}
			
			logger.info("PersistentThreadModel completed");
		}
	}

	/* ============================================================ */

	/**
	 * Factory for various items. Methods can be overridden for junit testing.
	 */
	protected static class Factory {

		/**
		 * Gets the configuration for the transaction manager.
		 * 
		 * @return the configuration for the transaction manager
		 */
		public Configuration getTransMgrConfig() {
			return TransactionManagerServices.getConfiguration();
		}

		/**
		 * Gets the transaction manager.
		 * 
		 * @return the transaction manager
		 */
		public BitronixTransactionManager getTransMgr() {
			return TransactionManagerServices.getTransactionManager();
		}

		/**
		 * Gets the KIE services.
		 * 
		 * @return the KIE services
		 */
		public KieServices getKieServices() {
			return KieServices.Factory.get();
		}

		/**
		 * Gets the current host name.
		 * 
		 * @return the current host name, associated with the IP address of the
		 *         local machine
		 * @throws UnknownHostException
		 */
		public String getHostName() throws UnknownHostException {
			return InetAddress.getLocalHost().getHostName();
		}

		/**
		 * Loads properties from a file.
		 * 
		 * @param filenm
		 *            name of the file to load
		 * @return properties, as loaded from the file
		 * @throws IOException
		 *             if an error occurs reading from the file
		 */
		public Properties loadProperties(String filenm) throws IOException {
			return PropertyUtil.getProperties(filenm);
		}

		/**
		 * Makes a connection to the DB.
		 * 
		 * @param url
		 *            DB URL
		 * @param user
		 *            user name
		 * @param pass
		 *            password
		 * @return a new DB connection
		 * @throws SQLException
		 */
		public Connection makeDbConnection(String url, String user, String pass) throws SQLException {

			return DriverManager.getConnection(url, user, pass);
		}

		/**
		 * Makes a new pooling data source.
		 * 
		 * @return a new pooling data source
		 */
		public PoolingDataSource makePoolingDataSource() {
			return new PoolingDataSource();
		}

		/**
		 * Makes a new JPA connector for drools sessions.
		 * 
		 * @param pu
		 *            PU for the entity manager factory
		 * @param propMap
		 *            properties with which the factory should be configured
		 * @return a new JPA connector for drools sessions
		 */
		public DroolsSessionConnector makeJpaConnector(String pu, Properties propMap) {

			EntityManagerFactory emf = makeEntMgrFact(pu, propMap);

			return new JpaDroolsSessionConnector(emf);
		}

		/**
		 * Makes a new entity manager factory.
		 * 
		 * @param pu
		 *            PU for the entity manager factory
		 * @param propMap
		 *            properties with which the factory should be configured
		 * @return a new entity manager factory
		 */
		public EntityManagerFactory makeEntMgrFact(String pu, Properties propMap) {
			return Persistence.createEntityManagerFactory(pu, propMap);
		}

		/**
		 * Gets the policy controller associated with a given policy container.
		 * 
		 * @param container
		 *            container whose controller is to be retrieved
		 * @return the container's controller
		 */
		public PolicyController getPolicyContainer(PolicyContainer container) {
			return PolicyController.factory.get(container.getGroupId(), container.getArtifactId());
		}
	}
}