* @param timeout seconds to wait for the next message if the pipe is empty. 
	 */
	DbLogListener(Connection conn, String key, Logger rootLogger, int timeout, String pipeName) throws SQLException {
		mapKey_ = key;
		
		setRootLoggerName( (rootLogger == null) ? "" : rootLogger.getName());
		
		pipeName_ = pipeName;

		stmt_ = getDbLogPipeReadStatement(conn, timeout, pipeName_);
	}
	
	/**
	 * If you want to ensure only one listener per user schema (connection) and pipe name combination, use the static
	 * getDblogListener() factory methods instead of the direct constructor call.
	 * 
	 * @param xmlNodeConfig XML configuration node, for connection properties and pipe name.
	 * @param dbLogger base logger to use.
	 */
	public DbLogListener(XMLNode xmlNodeConfig, Logger dbLogger) throws SQLException {
		XmlConfig lxmlConfig = new XmlConfig(xmlNodeConfig);
		
      String jdbcUrl = lxmlConfig.getXpathParam("database/source/connection/dburl/text()", "NotExiste");
      String user = lxmlConfig.getXpathParam("database/source/connection/username/text()", "NotExiste");
      String pwd = lxmlConfig.getXpathParam("database/source/connection/password/text()", "NotExiste");
      pipeName_ = lxmlConfig.getXpathParam("database/source/pipename/text()", "NotExiste");
		
    	OracleDataSource ods = new OracleDataSource();
    	ods.setURL(jdbcUrl);
    	ods.setUser(user);
    	ods.setPassword(pwd);
    	conn_ = ods.getConnection();
    	
    	setRootLoggerName(dbLogger.getName());
    	
    	stmt_ = getDbLogPipeReadStatement(conn_, DEFAULT_PIPE_TIMEOUT, pipeName_);
	}
	
	/**
	 * Populate this object with the next message from the Oracle pipe it is reading from.
	 * The PL/SQL function used to read the next log message has the following parameters and return value:
	 * <p>
	 * <pre>
	 * FUNCTION READ_MESSAGE(
	 *    pTIMEOUT    IN        NUMBER		-- seconds to wait for the next message.
	 *    pID         OUT       NUMBER		-- sequential ID of the log message
	 *    pLDATE      OUT       DATE			-- SYSDATE value for this message
	 *    pLHSECS     OUT       NUMBER		-- hundredths of seconds for the SYSDATE
	 *    pLLEVEL     OUT       NUMBER		-- Log4J log level for the message
	 *    pLSECTION   OUT       VARCHAR2	-- Log4Plsql section name, can be used as Log4J Logger name
	 *    pLUSER      OUT       VARCHAR2	-- database user sending the log message
	 * 	pCOMMAND    OUT       VARCHAR2   -- command name for future use, when multiple commands are possible
	 *    pLTEXTE     OUT       VARCHAR2	-- message text
	 * 	pMDC_KEYS	OUT		 VARCHAR2	-- delimited string of MDC keys
	 * 	pMDC_VALUES OUT		 VARCHAR2	-- delimited string of MDC values
	 * 	pMDC_SEPARATOR OUT	 VARCHAR2	-- MDC string delimiter
	 *    ) RETURN NUMBER -- DBMS_PIPE.RECEIVE_MESSAGE return value
	 * </pre>
	 * <p>
	 * <b>This method is only package visible for Unit Test purposes. Do not call it outside the Run Method.</b>
	 * <p>
	 * @return True if a message was read, False if the timeout period passed without a new message.
	 */
	boolean getNextMessage() throws SQLException {
		stmt_.execute();
		
		int retval = stmt_.getInt(1);
		
		/**
		 * Possible retval values:
		 * 0 = success, message received
		 * 1 = timeout, no message
		 * 2 = message in pipe too big for buffer (should not be possible)
		 * 3 = interrupt of some sort occurred
		 */
		if (retval == 0) {
			id_ = stmt_.getLong(3);
			Timestamp timestamp = stmt_.getTimestamp(4);
			// If the passed date is not null, add the hundredths of seconds passed as well by turning them into milliseconds.
			date_ = (timestamp != null) ? new Date(timestamp.getTime() + stmt_.getInt(5) * 10) : null;
			level_ = (Level) this.logLevels_.get(new Integer(stmt_.getInt(6)));
			section_ = stmt_.getString(7);
			user_ = stmt_.getString(8);
			command_ = stmt_.getString(9);
			message_ = stmt_.getString(10);
			mdcSeparator_ = stmt_.getString(13);
			
			mdcKeys_.clear();
			mdcValues_.clear();
			
			StringTokenizer keys = new StringTokenizer(stmt_.getString(11), mdcSeparator_);
			while (keys.hasMoreTokens()) mdcKeys_.add(keys.nextToken());
			
			StringTokenizer values = new StringTokenizer(stmt_.getString(12), mdcSeparator_);
			while (values.hasMoreTokens()) mdcValues_.add(values.nextToken());
			
			return true;
		}
		return false;
	}
	
	/**
	 * Log the current message. Default logging uses <code>section</code> as the Logger name, 
	 * <code>level</code> for the logging level, and puts the <code>date, ID,</code> and <code>User</code>
	 * in the Log4J MDC using the constants for these fields defined in this class.
	 * <p>
	 * The assumption is that Log4J has been configured previously to define this logger and attach some
	 * appender(s) to it.
	 * <p>
	 * All messages are logged.  The assumption is that Log4Plsql already determined it wanted these messages in
	 * the log output.  A new <code>LoggingEvent</code> is created and passed to Logger.callAppenders() directly,
	 * without evaluating the Log4Plsql level against the Logger level.
	 * <p>
	 * This also ensures that the database date stamp is the one used for the LoggingEvent, instead of the current
	 * Java system time, which will be later.  How much later depends on the pipe timeout, communication lag times,
	 * etc.
	 * <p>
	 * <b>This method is only package visible for Unit Test purposes. Do not call it outside the Run Method.</b>
	 * <p>
	 */
	void logMessage() {
		MDC.put(MDC_DATE, this.date_);
		MDC.put(MDC_USER, this.user_);
		MDC.put(MDC_ID, new Long(this.id_));
		for (int i=0; i < mdcKeys_.size(); i++) MDC.put((String) mdcKeys_.get(i), mdcValues_.get(i));
		getDbLogHelper().preLog();
		Logger logger = getDbLogHelper().getLogger();
		LoggingEvent logEvent = new LoggingEvent(logger.getClass().getName(), logger, getTimestamp(), this.level_, this.message_, null);
		logger.callAppenders(logEvent);
		getDbLogHelper().postLog();
		MDC.remove(MDC_DATE);
		MDC.remove(MDC_USER);
		MDC.remove(MDC_ID);
		for (int i=0; i < mdcKeys_.size(); i++) MDC.remove((String) mdcKeys_.get(i));
	}
	
	protected void close() {
		getDbLogHelper().close();
		if (stmt_ != null) try { stmt_.close(); } catch (SQLException e) {}
	}


	/**
	 * Starts the logging loop for this instance.
	 * If this instance is already looping on another thread, then the method exits immediately.
	 * When the looping is done (isStillWatching = false), this instance removes itself from the pool
	 * and performs any cleanup needed.
	 * 
	 */
	public void run() {
		
		synchronized (isRunningLock_) {
			if (isRunning_) return;
			else isRunning_ = true;
		}

		try {
			while (isStillWatching()) {
				
				if (getNextMessage()) {
					// Only need to log if there was a new message retrieved.
					logMessage();
				}
			}
		} catch (SQLException e) {
			Logger logger = null;
			if ("".equals(getRootLoggerName())) logger = Logger.getRootLogger();
			else logger = Logger.getLogger(getRootLoggerName());
			logger.error("Error getting next Oracle Pipe message on pipe '" + pipeName_ + "':", e);
		} finally {
			// unregister from map
			removeDbLogListener(mapKey_);
			close();
			synchronized (isRunningLock_) {
				isRunning_ = false;
			}
			// reset this in case the thread will be restarted at some point.
			setStillWatching(true);
		}
	}
	
	public boolean isStillWatching() {
		synchronized (stillWatchingLock_) {
			return this.stillWatching_;
		}
	}

	public boolean isRunning() {
		synchronized (isRunningLock_) {
			return this.isRunning_;
		}
	}

	public void setStillWatching(boolean keepWatching) {
		synchronized (stillWatchingLock_) {
			this.stillWatching_ = keepWatching;
		}
	}
	
	public String getRootLoggerName() { return rootLoggerName_; }
	private void setRootLoggerName(String rootLoggerName) { 
		this.rootLoggerName_ = rootLoggerName == null ? "": rootLoggerName.endsWith(".") ? rootLoggerName : rootLoggerName + ".";
	}
	
	/**
	 * Current Log command from the pipe.  Currently does nothing, as there is only one command.
	 */
	public String getCommand() {
		return this.command_;
	}
	/**
	 * Date stamp of the current log message.  Granularity to hundredths of seconds only.
	 */
	public Date getDate() {
		return this.date_;
	}
	/**
	 * Return 0 if the current message date is null, or Date.getTime() if not.
	 * @return
	 */
	public long getTimestamp() {
		return (this.date_ == null) ? 0 : this.date_.getTime();
	}
	/**
	 * Sequential identifier of the current log message - 0 = no current message.
	 */
	public long getId() {
		return this.id_;
	}
	/**
	 * Log message level - current value is looked up in the DB log level map.
	 */
	public Level getLevel() {
		return this.level_;
	}
	/**
	 * Text of the current log message.
	 */
	public String getMessage() {
		return this.message_;
	}
	/**
	 * Log4Plsql section for the current message, composed of period delimited strings.  
	 * This commonly maps to a Logger name.  The default implementation appends this to any
	 * root logger name given in the factory constructor to define the Logger name to use.
	 */
	public String getSection() {
		return this.section_;
	}
	/**
	 * Database user the current log message comes from.  May or may not be useful depending on the app architecture.
	 */
	public String getUser() {
		return this.user_;
	}
	/**
	 * Current Log command from the pipe.  Currently does nothing, as there is only one command.
	 */
	public void setCommand(String command) {
		this.command_ = command;
	}
	/**
	 * Date stamp of the current log message.  Granularity to hundredths of seconds only.
	 */
	public void setDate(Date date) {
		this.date_ = date;
	}
	/**
	 * Sequential identifier of the current log message - 0 = no current message.
	 */
	public void setId(long id) {
		this.id_ = id;
	}
	/**
	 * Log message level - current value is looked up in the DB log level map.
	 */
	public void setLevel(Level level) {
		this.level_ = level;
	}
	/**
	 * Text of the current log message.
	 */
	public void setMessage(String message) {
		this.message_ = message;
	}
	/**
	 * Log4Plsql section for the current message, composed of period delimited strings.  
	 * This commonly maps to a Logger name.  The default implementation appends this to any
	 * root logger name given in the factory constructor to define the Logger name to use.
	 */
	public void setSection(String section) {
		this.section_ = section;
	}
	/**
	 * Database user the current log message comes from.  May or may not be useful depending on the app architecture.
	 */
	public void setUser(String user) {
		this.user_ = user;
	}
	private DbLogHelper getDbLogHelper() {
		return this.dbLogHelper_;
	}
	/**
	 * Specify the instance of DbLogHelper or a sub-class to use when logging messages from this class.  Use subclasses to customize
	 * logic for deciding what Logger to use, re-configure Appenders, or modify Layouts for a given pipe message before it is logged
	 * and clean up anything needed after each message is logged.  The default case simply uses the base Logger name used when creating
	 * this instance concatenated with the current message section.  If this is sufficient, then no extra work is needed.
	 * <p>
	 * @param dbLogHelper
	 */
	public synchronized void setDbLogHelper(DbLogHelper dbLogHelper) {
		this.dbLogHelper_ = dbLogHelper;
	}
}