<commit_before>/*******************************************************************************
* Copyright (C) 2016 <NAME> <<EMAIL>>.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * which accompanies this distribution, and is available at
 * http://www.eclipse.org/legal/epl-v10.html
 * 
 * Contributors:
 *     <NAME> <<EMAIL>> - initial API and implementation
 ******************************************************************************/
package com.quakearts.syshub.log;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.Timer;
import java.util.TimerTask;

import javax.enterprise.inject.spi.CDI;
import javax.inject.Inject;
import javax.inject.Singleton;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.quakearts.appbase.cdi.annotation.Transactional;
import com.quakearts.appbase.cdi.annotation.Transactional.TransactionType;
import com.quakearts.syshub.core.Message;
import com.quakearts.syshub.core.Result;
import com.quakearts.syshub.core.utils.CacheManager;
import com.quakearts.syshub.core.utils.Serializer;
import com.quakearts.syshub.core.utils.SystemDataStoreManager;
import com.quakearts.syshub.model.AgentConfiguration;
import com.quakearts.syshub.model.AgentModule;
import com.quakearts.syshub.model.ProcessingLog;
import com.quakearts.syshub.model.ResultExceptionLog;
import com.quakearts.webapp.orm.query.helper.ParameterMapBuilder;
import com.quakearts.webapp.orm.stringconcat.OrmStringConcatUtil;
import com.quakearts.syshub.model.ProcessingLog.LogType;

import org.infinispan.Cache;

@Singleton
public class LoggerImpl implements MessageLogger, ResultExceptionLogger {
	@Inject
	private CacheManager cacheManager;
	@Inject
	private Serializer serializer;
	@Inject
	private SystemDataStoreManager storeManager;

	private Cache<String, ProcessingLog> save_log_cache;
	private Cache<String, ProcessingLog> update_log_cache;
	private Comparator<ProcessingLog> logComparator = (n1,n2)-> n1.getLogDt().compareTo(n2.getLogDt());
	private final Logger log = LoggerFactory.getLogger(LoggerImpl.class);
	private final Timer log_timer = new Timer(true);
	private Cache<String, ResultExceptionLog> result_exception_log_cache;
	private boolean shutdown;
		
	LoggerImpl() {
		log_timer.schedule(new TimerTask() {	
			@Override
			public void run() {
				try {
					if(shutdown)
						return;
					
					getInstance().pushLogsToDB();
				} catch (Throwable e) {
					log.error("Exception of type " + e.getClass().getName() + " was thrown. Message is " + e.getMessage()
							+ ". Exception occured whiles pushing logs to database.");
				}
			}
		}, 0, 500);
		
		Runtime.getRuntime().addShutdownHook(new Thread(()-> {
				shutdown = true;
				log_timer.cancel();
				for(String key:getSaveLogCache().keySet()){
					getSaveLogCache().evict(key);
				}
				
				for(String key:getUpdateLogCache().keySet()){
					getUpdateLogCache().evict(key);
				}

				for(String key:getResultExceptionLog().keySet()){
					getResultExceptionLog().evict(key);
				}
			}));
    }
    
	private static LoggerImpl instance;
	
	public static LoggerImpl getInstance() {
		if(instance == null){
			instance = CDI.current().select(LoggerImpl.class).get();
		}
		return instance;
	}

	@Transactional(TransactionType.SINGLETON)
	public synchronized void pushLogsToDB() throws Exception {
		if(shutdown)
			return;
		
		List<ProcessingLog> sortList = new ArrayList<>(getSaveLogCache().values());
		sortList.sort(logComparator);
		
		Queue<ProcessingLog> logQueue = new LinkedList<>(sortList);
		
		while (!logQueue.isEmpty() ) {
			ProcessingLog notificationLog = logQueue.poll();
			try {
				storeManager.getDataStore().save(notificationLog);
			} catch (Exception e) {
				log.error("Unable to persist notification log.");
				log.error("Message dump: ["+notificationLog.toString()+"]");
			} finally {
				getSaveLogCache().remove(notificationLog.getMid());
			}
		}
	
		if(shutdown)
			return;
		
		sortList = new ArrayList<>(getUpdateLogCache().values());
		sortList.sort(logComparator);
		
		logQueue = new LinkedList<>(sortList);
		
		while(!logQueue.isEmpty()){
			ProcessingLog notificationLog = logQueue.poll();
			try {
				storeManager.getDataStore().update(notificationLog);
			} catch (Exception e) {
				log.error("Unable to persist notification log.");
				log.error("Message dump: ["+notificationLog.toString()+"]");
			} finally {
				getUpdateLogCache().remove(notificationLog.getMid());
			}
		}
		
		if(shutdown)
			return;
		
		List<ResultExceptionLog> sortExceptionLogList = new ArrayList<>(getResultExceptionLog().values());
		sortExceptionLogList.sort((r1,r2)-> r1.getExceptionDt().compareTo(r2.getExceptionDt()));
		
		Queue<ResultExceptionLog> exceptionQueue = new LinkedList<>(sortExceptionLogList);
		while (!exceptionQueue.isEmpty()) {
			ResultExceptionLog exceptionLog = exceptionQueue.poll();
			try {
				storeManager.getDataStore().save(exceptionLog);
			} catch (Exception e) {
				log.error("Unable to persist result exception log.");
				log.error("Message dump: ["+exceptionLog.toString()+"]");
			} finally {
				getResultExceptionLog().remove(exceptionLog.hashCodeAsString());
			}
		}
	}

	@Override
	public void logResultException(AgentConfiguration agentConfiguration, AgentModule agentModule,
			Exception e, Result<?> result) {
		ResultExceptionLog exceptionLog = new ResultExceptionLog();
		exceptionLog.setAgentConfiguration(agentConfiguration);
		exceptionLog.setAgentModule(agentModule);
		exceptionLog.setExceptionType(e.getClass().getName());
		exceptionLog.setExceptionData(serializer.toByteArray(e));
		exceptionLog.setResultData(serializer.toByteArray(result));
		getResultExceptionLog().put(exceptionLog.hashCodeAsString(), exceptionLog);
	}
	
	@Override
	public List<ResultExceptionLog> getUnpersistedResultExceptionLogs() {
		ArrayList<ResultExceptionLog> unpersistedLogs = 
				new ArrayList<ResultExceptionLog>(getResultExceptionLog().values());
		return Collections.unmodifiableList(unpersistedLogs);
	}
	
	@Override
	public ResultExceptionLog getResultExceptionLogByID(long relID) {
		return storeManager.getDataStore().get(ResultExceptionLog.class, relID);
	}
	
	/* (non-Javadoc)
	 * @see com.quakearts.notification.log.MessageLogger#getUnpersistedLogs()
	 */
	@Override
	public List<ProcessingLog> getUnpersistedProcessingLogs(){
		ArrayList<ProcessingLog> unpersistedLogs = 
				new ArrayList<ProcessingLog>(getSaveLogCache().values());
		unpersistedLogs.addAll(getUpdateLogCache().values());
		return Collections.unmodifiableList(unpersistedLogs);
	}
	
	/* (non-Javadoc)
	 * @see com.quakearts.notification.log.MessageLogger#log(com.quakearts.notification.core.Message, java.lang.String, boolean)
	 */
	@Override
	public void logMessage(AgentConfiguration agentConfiguration, AgentModule agentModule, Message<?> mssg, String response, boolean isError){
		saveLog(agentConfiguration, agentModule, mssg, LogType.INFO, response,isError);
	}

	/* (non-Javadoc)
	 * @see com.quakearts.notification.log.MessageLogger#store(com.quakearts.notification.core.Message, java.lang.String, boolean)
	 */
	@Override
	public void storeMessage(AgentConfiguration agentConfiguration, AgentModule agentModule, Message<?> mssg, String details){
		saveLog(agentConfiguration, agentModule, mssg, LogType.STORED, details, false);
	}
	
	/* (non-Javadoc)
	 * @see com.quakearts.notification.log.MessageLogger#queue(com.quakearts.notification.core.Message, java.lang.String)
	 */
	@Override
	public void queueMessage(AgentConfiguration agentConfiguration, AgentModule agentModule, Message<?> mssg,String reason){
		saveLog(agentConfiguration, agentModule, mssg, LogType.QUEUED, reason, false);
	}

	/* (non-Javadoc)
	 * @see com.quakearts.notification.log.MessageLogger#getLogByID(long)
	 */
	@Override
	@Transactional(TransactionType.SINGLETON)
	public ProcessingLog getProcessingLogByID(long logID){
		return storeManager.getDataStore().get(ProcessingLog.class, Long.valueOf(logID));
	}
	
	/* (non-Javadoc)
	 * @see com.quakearts.notification.log.MessageLogger#findMessageLogByMid(java.lang.String)
	 */
	@Override
	@Transactional(TransactionType.SINGLETON)
	public ProcessingLog getProcessingLogByMid(String mid){
		
		ProcessingLog founndlog = getSaveLogCache().get(mid);
		if(founndlog!=null){
			return founndlog;
		}

		List<ProcessingLog> list = storeManager.getDataStore().list(ProcessingLog.class, ParameterMapBuilder
				.createParameters().add("mid",mid).build());
		
		if(list.size()>0){
			Object obj = list.get(list.size()-1);
			return ((ProcessingLog)obj);
		} else
			return null;
	}

	/* (non-Javadoc)
	 * @see com.quakearts.notification.log.MessageLogger#updateLog(com.quakearts.notification.model.NotificationLog)
	 */
	@Override
	public void updateLog(ProcessingLog notificationLog){
		getUpdateLogCache().put(notificationLog.getMid(), notificationLog);
	}

	private void saveLog(AgentConfiguration agentConfiguration, AgentModule agentModule, 
			Message<?> mssg, LogType type, String statusMessage, boolean isError){
		if(log.isTraceEnabled())
			log.trace("Saving message to queue. Type: "+type+". statusMessage: "+statusMessage);
		
		ProcessingLog notelog = new ProcessingLog();
		
		notelog.setStatusMessage(statusMessage);
		notelog.setMid(mssg.getId());
		notelog.setAgentConfiguration(agentConfiguration);
		notelog.setAgentModule(agentModule);
		String[] receipients = mssg.getRecipients(); 
		notelog.setRecipient(Arrays.toString(receipients));
		notelog.setType(type);
		notelog.setMessageData(serializer.toByteArray(mssg));
		notelog.setRetries(0);
		notelog.setError(isError);
		notelog.setLogDt(new Date());
		
		OrmStringConcatUtil.trimStrings(notelog);
		
		getSaveLogCache().put(notelog.getMid(), notelog);
	}

	/**
	 * @return the log_cache
	 */
	private Cache<String, ProcessingLog> getSaveLogCache() {
		if(save_log_cache==null){
			save_log_cache = cacheManager.getCache(ProcessingLog.class,".NEW");
		}
		return save_log_cache;
	}

	private Cache<String, ProcessingLog> getUpdateLogCache() {
		if(update_log_cache==null){
			update_log_cache = cacheManager.getCache(ProcessingLog.class,".UPDATE");
		}
		return update_log_cache;
	}
	
	private Cache<String, ResultExceptionLog> getResultExceptionLog(){
		if(result_exception_log_cache == null){
			result_exception_log_cache = cacheManager.getCache(ResultExceptionLog.class, null);
		}
		
		return result_exception_log_cache;
	}
}
<commit_msg>Bug fix: check update log cache for messages before DB<commit_after>/*******************************************************************************
* Copyright (C) 2016 <NAME> <<EMAIL>>.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * which accompanies this distribution, and is available at
 * http://www.eclipse.org/legal/epl-v10.html
 * 
 * Contributors:
 *     <NAME> <<EMAIL>> - initial API and implementation
 ******************************************************************************/
package com.quakearts.syshub.log;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.LinkedList;
import java.util.List;
import java.util.Optional;
import java.util.Queue;
import java.util.Timer;
import java.util.TimerTask;

import javax.enterprise.inject.spi.CDI;
import javax.inject.Inject;
import javax.inject.Singleton;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.quakearts.appbase.cdi.annotation.Transactional;
import com.quakearts.appbase.cdi.annotation.Transactional.TransactionType;
import com.quakearts.syshub.core.Message;
import com.quakearts.syshub.core.Result;
import com.quakearts.syshub.core.utils.CacheManager;
import com.quakearts.syshub.core.utils.Serializer;
import com.quakearts.syshub.core.utils.SystemDataStoreManager;
import com.quakearts.syshub.model.AgentConfiguration;
import com.quakearts.syshub.model.AgentModule;
import com.quakearts.syshub.model.ProcessingLog;
import com.quakearts.syshub.model.ResultExceptionLog;
import com.quakearts.webapp.orm.stringconcat.OrmStringConcatUtil;
import com.quakearts.syshub.model.ProcessingLog.LogType;

import org.infinispan.Cache;

@Singleton
public class LoggerImpl implements MessageLogger, ResultExceptionLogger {
	private static final String MESSAGE_DUMP = "Message dump: [{}]";
	@Inject
	private CacheManager cacheManager;
	@Inject
	private Serializer serializer;
	@Inject
	private SystemDataStoreManager storeManager;

	private Cache<String, ProcessingLog> saveLogCache;
	private Cache<String, ProcessingLog> updateLogCache;
	private Comparator<ProcessingLog> logComparator = (n1,n2)-> n1.getLogDt().compareTo(n2.getLogDt());
	private final Logger log = LoggerFactory.getLogger(LoggerImpl.class);
	private final Timer logTimer = new Timer(true);
	private Cache<String, ResultExceptionLog> resultExceptionLogCache;
	private boolean shutdown;
		
	LoggerImpl() {
		logTimer.schedule(new TimerTask() {	
			@Override
			public void run() {
				if(shutdown)
					return;
				
				getInstance().pushLogsToDB();
			}
		}, 0, 500);
		
		Runtime.getRuntime().addShutdownHook(new Thread(()-> {
				shutdown = true;
				logTimer.cancel();
				for(String key:getSaveLogCache().keySet()){
					getSaveLogCache().evict(key);
				}
				
				for(String key:getUpdateLogCache().keySet()){
					getUpdateLogCache().evict(key);
				}

				for(String key:getResultExceptionLog().keySet()){
					getResultExceptionLog().evict(key);
				}
			}));
    }
    
	private static LoggerImpl instance;
	
	public static LoggerImpl getInstance() {
		if(instance == null){
			instance = CDI.current().select(LoggerImpl.class).get();
		}
		return instance;
	}

	@Transactional(TransactionType.SINGLETON)
	public synchronized void pushLogsToDB() {
		if(shutdown)
			return;
		
		List<ProcessingLog> sortList = new ArrayList<>(getSaveLogCache().values());
		sortList.sort(logComparator);
		
		Queue<ProcessingLog> logQueue = new LinkedList<>(sortList);
		
		while (!logQueue.isEmpty() ) {
			ProcessingLog notificationLog = logQueue.poll();
			try {
				storeManager.getDataStore().save(notificationLog);
			} catch (Exception e) {
				log.error("Unable to persist notification log.");
				log.error(MESSAGE_DUMP, notificationLog);
			} finally {
				getSaveLogCache().remove(notificationLog.getMid());
			}
		}
	
		if(shutdown)
			return;
		
		sortList = new ArrayList<>(getUpdateLogCache().values());
		sortList.sort(logComparator);
		
		logQueue = new LinkedList<>(sortList);
		
		while(!logQueue.isEmpty()){
			ProcessingLog notificationLog = logQueue.poll();
			try {
				storeManager.getDataStore().update(notificationLog);
			} catch (Exception e) {
				log.error("Unable to persist notification log.");
				log.error(MESSAGE_DUMP, notificationLog);
			} finally {
				getUpdateLogCache().remove(notificationLog.getMid());
			}
		}
		
		if(shutdown)
			return;
		
		List<ResultExceptionLog> sortExceptionLogList = new ArrayList<>(getResultExceptionLog().values());
		sortExceptionLogList.sort((r1,r2)-> r1.getExceptionDt().compareTo(r2.getExceptionDt()));
		
		Queue<ResultExceptionLog> exceptionQueue = new LinkedList<>(sortExceptionLogList);
		while (!exceptionQueue.isEmpty()) {
			ResultExceptionLog exceptionLog = exceptionQueue.poll();
			try {
				storeManager.getDataStore().save(exceptionLog);
			} catch (Exception e) {
				log.error("Unable to persist result exception log.");
				log.error(MESSAGE_DUMP, exceptionLog);
			} finally {
				getResultExceptionLog().remove(exceptionLog.hashCodeAsString());
			}
		}
	}

	@Override
	public void logResultException(AgentConfiguration agentConfiguration, AgentModule agentModule,
			Exception e, Result<?> result) {
		ResultExceptionLog exceptionLog = new ResultExceptionLog();
		exceptionLog.setAgentConfiguration(agentConfiguration);
		exceptionLog.setAgentModule(agentModule);
		exceptionLog.setExceptionType(e.getClass().getName());
		exceptionLog.setExceptionData(serializer.toByteArray(e));
		exceptionLog.setResultData(serializer.toByteArray(result));
		getResultExceptionLog().put(exceptionLog.hashCodeAsString(), exceptionLog);
	}
	
	@Override
	public List<ResultExceptionLog> getUnpersistedResultExceptionLogs() {
		ArrayList<ResultExceptionLog> unpersistedLogs = 
				new ArrayList<>(getResultExceptionLog().values());
		return Collections.unmodifiableList(unpersistedLogs);
	}
	
	@Override
	public ResultExceptionLog getResultExceptionLogByID(long relID) {
		return storeManager.getDataStore().get(ResultExceptionLog.class, relID);
	}
	
	/* (non-Javadoc)
	 * @see com.quakearts.notification.log.MessageLogger#getUnpersistedLogs()
	 */
	@Override
	public List<ProcessingLog> getUnpersistedProcessingLogs(){
		ArrayList<ProcessingLog> unpersistedLogs = 
				new ArrayList<>(getSaveLogCache().values());
		unpersistedLogs.addAll(getUpdateLogCache().values());
		return Collections.unmodifiableList(unpersistedLogs);
	}
	
	/* (non-Javadoc)
	 * @see com.quakearts.notification.log.MessageLogger#log(com.quakearts.notification.core.Message, java.lang.String, boolean)
	 */
	@Override
	public void logMessage(AgentConfiguration agentConfiguration, AgentModule agentModule, Message<?> mssg, String response, boolean isError){
		saveLog(agentConfiguration, agentModule, mssg, LogType.INFO, response,isError);
	}

	/* (non-Javadoc)
	 * @see com.quakearts.notification.log.MessageLogger#store(com.quakearts.notification.core.Message, java.lang.String, boolean)
	 */
	@Override
	public void storeMessage(AgentConfiguration agentConfiguration, AgentModule agentModule, Message<?> mssg, String details){
		saveLog(agentConfiguration, agentModule, mssg, LogType.STORED, details, false);
	}
	
	/* (non-Javadoc)
	 * @see com.quakearts.notification.log.MessageLogger#queue(com.quakearts.notification.core.Message, java.lang.String)
	 */
	@Override
	public void queueMessage(AgentConfiguration agentConfiguration, AgentModule agentModule, Message<?> mssg,String reason){
		saveLog(agentConfiguration, agentModule, mssg, LogType.QUEUED, reason, false);
	}

	/* (non-Javadoc)
	 * @see com.quakearts.notification.log.MessageLogger#getLogByID(long)
	 */
	@Override
	@Transactional(TransactionType.SINGLETON)
	public ProcessingLog getProcessingLogByID(long logID){
		return storeManager.getDataStore().get(ProcessingLog.class, Long.valueOf(logID));
	}
	
	/* (non-Javadoc)
	 * @see com.quakearts.notification.log.MessageLogger#findMessageLogByMid(java.lang.String)
	 */
	@Override
	@Transactional(TransactionType.SINGLETON)
	public ProcessingLog getProcessingLogByMid(String mid){
		
		ProcessingLog foundLog = getUpdateLogCache().get(mid);
		if(foundLog!=null){
			return foundLog;
		}

		foundLog = getSaveLogCache().get(mid);
		if(foundLog!=null){
			return foundLog;
		}

		Optional<ProcessingLog> optionalLog = storeManager.getDataStore().find(ProcessingLog.class)
				.filterBy("mid").withAValueEqualTo(mid).thenGetFirst();
		
		if(optionalLog.isPresent()){
			return optionalLog.get();
		} else {
			return null;
		}
	}

	/* (non-Javadoc)
	 * @see com.quakearts.notification.log.MessageLogger#updateLog(com.quakearts.notification.model.NotificationLog)
	 */
	@Override
	public void updateLog(ProcessingLog notificationLog){
		getUpdateLogCache().put(notificationLog.getMid(), notificationLog);
	}

	private void saveLog(AgentConfiguration agentConfiguration, AgentModule agentModule, 
			Message<?> mssg, LogType type, String statusMessage, boolean isError){
		if(log.isTraceEnabled())
			log.trace("Saving message to queue. Type: {}. statusMessage: {}", type, statusMessage);
		
		ProcessingLog notelog = new ProcessingLog();
		
		notelog.setStatusMessage(statusMessage);
		notelog.setMid(mssg.getId());
		notelog.setAgentConfiguration(agentConfiguration);
		notelog.setAgentModule(agentModule);
		String[] receipients = mssg.getRecipients(); 
		notelog.setRecipient(Arrays.toString(receipients));
		notelog.setType(type);
		notelog.setMessageData(serializer.toByteArray(mssg));
		notelog.setRetries(0);
		notelog.setError(isError);
		notelog.setLogDt(new Date());
		
		OrmStringConcatUtil.trimStrings(notelog);
		
		getSaveLogCache().put(notelog.getMid(), notelog);
	}

	/**
	 * @return the log_cache
	 */
	private Cache<String, ProcessingLog> getSaveLogCache() {
		if(saveLogCache==null){
			saveLogCache = cacheManager.getCache(ProcessingLog.class,".NEW");
		}
		return saveLogCache;
	}

	private Cache<String, ProcessingLog> getUpdateLogCache() {
		if(updateLogCache==null){
			updateLogCache = cacheManager.getCache(ProcessingLog.class,".UPDATE");
		}
		return updateLogCache;
	}
	
	private Cache<String, ResultExceptionLog> getResultExceptionLog(){
		if(resultExceptionLogCache == null){
			resultExceptionLogCache = cacheManager.getCache(ResultExceptionLog.class, null);
		}
		
		return resultExceptionLogCache;
	}
}

{'fraction_non_alphanumeric': 0.12990060739922696, 'fraction_numerical': 0.0022087244616234127, 'mean_word_length': 4.59552008238929, 'pattern_counts': {'":': 0, '<': 73, '<?xml version=': 0, '>': 80, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, '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': '7320583', 'n_tokens_mistral': 7212, 'n_tokens_neox': 6419, 'n_words': 1112}