<commit_before>package org.jimmutable.cloud.cache.redis;

import java.net.SocketTimeoutException;
import java.nio.charset.StandardCharsets;
import java.util.Random;
import java.util.concurrent.ExecutorService;

import org.jimmutable.cloud.ApplicationId;
import org.jimmutable.cloud.cache.CacheKey;
import org.jimmutable.cloud.messaging.queue.QueueId;
import org.jimmutable.cloud.messaging.queue.QueueListener;
import org.jimmutable.cloud.messaging.signal.SignalListener;
import org.jimmutable.cloud.messaging.signal.SignalTopicId;
import org.jimmutable.core.objects.StandardObject;
import org.jimmutable.core.serialization.Format;
import org.jimmutable.core.threading.DaemonThreadFactory;
import org.jimmutable.core.utils.NetUtils;
import org.jimmutable.core.utils.Validator;
import org.slf4j.LoggerFactory;
import org.slf4j.Logger;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisPubSub;
import redis.clients.jedis.ScanParams;
import redis.clients.jedis.ScanResult;
import redis.clients.jedis.exceptions.JedisConnectionException;

/**
 * Low level driver class for Redis.
 * 
 * DO NOT MODIFY UNLESS YOU REALLY UNDERSTAND REDIS, THREADING *AND* WHAT YOU ARE DOING 
 * 
 * All methods are *fully* thread safe
 * 
 * @author kanej
 *
 */
public class LowLevelRedisDriver
{
	static public final String DEFAULT_HOST = "localhost";
	static public final int DEFAULT_PORT_REDIS = 6379;
	static public final Logger logger = LoggerFactory.getLogger(LowLevelRedisDriver.class);
	
	private JedisPool pool;
	
	private ExecutorService pool_send;
	private ExecutorService pool_receive;
	
	private RedisCache cache;
	private RedisSignal signal;
	private RedisQueue queue;
	
	public LowLevelRedisDriver()
	{
		String redis_address = System.getProperty("redis.address");
		String host = NetUtils.extractHostFromHostPortPair(redis_address, LowLevelRedisDriver.DEFAULT_HOST);
		int port = NetUtils.extractPortFromHostPortPair(redis_address, DEFAULT_PORT_REDIS);
		
		JedisPoolConfig config = new JedisPoolConfig();
		config.setMaxTotal(250);
		config.setMaxIdle(1000 * 60);
		config.setTestOnBorrow(false);
		
		logger.info("[LowLevelRedisDriver.init] starting LowLevelRedisDriver on host: " + host + ", port: " + port);
		
		pool = new JedisPool(config, host, port);
		cache = new RedisCache();
		queue = new RedisQueue();
		signal = new RedisSignal();
		
		pool_send = DaemonThreadFactory.createDaemonFixedThreadPool(1);
		pool_receive = DaemonThreadFactory.createDaemonFixedThreadPool(10);
		
	}
	
	public boolean isRedisUp()
	{
		try(Jedis jedis = pool.getResource();)
		{
			String str = jedis.ping();
			return str.equalsIgnoreCase("PONG");
		}
		catch(Exception e)
		{
			return false;
		}
	}
	
	/**
	 * Access the cache functionality of the Redis driver
	 * 
	 * @return
	 */
	public RedisCache getSimpleCache() { return cache; }
	
	/**
	 * Access the signal functionality of the Redis driver
	 * 
	 * @return
	 */
	public RedisSignal getSimpleSignal() { return signal; }
	
	/**
	 * Access the queue functionality of the Redis driver
	 * 
	 * @return
	 */
	public RedisQueue getSimpleQueue() { return queue; }
	
	public class RedisCache
	{
		private byte[] createKeyBytes(ApplicationId app, CacheKey key, byte default_value[])
		{
			if ( app == null || key == null ) return default_value;
			
			StringBuilder builder = new StringBuilder();
			builder.append(app.toString());
			builder.append("/");
			builder.append(key.toString());
			
			return builder.toString().getBytes(StandardCharsets.UTF_8);
		}
		
		/**
		 * Set the value of a key in the cache (optionally with a maximum time to live).
		 * Setting with null data has the effect of deleting the key from the cache.
		 * Null keys and app ids result in the function doing nothing
		 * 
		 * @param app
		 *            The application id
		 * @param key
		 *            The key to set
		 * @param data
		 *            The data to write
		 * @param max_ttl
		 *            The maximum length of time that the data will remain in the cache
		 *            (in ms). Zero or negative values have the meaning of "infinite"
		 */
		public void set(ApplicationId app, CacheKey key, byte data[], long max_ttl)
		{
			if ( data == null )
			{
				delete(app,key);
				return;
			}
			
			byte cache_key_bytes[] = createKeyBytes(app,key,null);
			if ( cache_key_bytes == null ) return;
			
			try(Jedis jedis = pool.getResource();)
			{
				String result = null;
				try
				{
					result = jedis.set(cache_key_bytes, data);
				}
				catch (Exception e)
				{
					// If the set operation fails, skip the time-to-live setting also
					logger.error(String.format("Redis set operation for cache key %s failed.", key.getSimpleValue()), e);
					return;
				}
				
				if ( max_ttl > 0 )
				{ 
					max_ttl /= 1000;
					try
					{
						jedis.expire(cache_key_bytes, (int)max_ttl);
					}
					catch (Exception e)
					{
						String msg = String.format("Attempt to expire Redis cache key %s failed.", key.getSimpleValue());
						msg += String.format(" Preceeding Redis set operation result was %s", result != null ? result : "null");
						logger.error(msg, e);
					}
				}
			}
		}
		
		/**
		 * Test for the existence of a key. Null app or key results in a false return
		 * value
		 * 
		 * @param app
		 *            The application id
		 * @param key
		 *            The cache key
		 * 
		 * @return<commit_msg>Update LowLevelRedisDriver to apply ttl in one operation<commit_after>package org.jimmutable.cloud.cache.redis;

import java.net.SocketTimeoutException;
import java.nio.charset.StandardCharsets;
import java.util.Random;
import java.util.concurrent.ExecutorService;

import org.jimmutable.cloud.ApplicationId;
import org.jimmutable.cloud.cache.CacheKey;
import org.jimmutable.cloud.messaging.queue.QueueId;
import org.jimmutable.cloud.messaging.queue.QueueListener;
import org.jimmutable.cloud.messaging.signal.SignalListener;
import org.jimmutable.cloud.messaging.signal.SignalTopicId;
import org.jimmutable.core.objects.StandardObject;
import org.jimmutable.core.serialization.Format;
import org.jimmutable.core.threading.DaemonThreadFactory;
import org.jimmutable.core.utils.NetUtils;
import org.jimmutable.core.utils.Validator;
import org.slf4j.LoggerFactory;
import org.slf4j.Logger;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisPubSub;
import redis.clients.jedis.ScanParams;
import redis.clients.jedis.ScanResult;
import redis.clients.jedis.exceptions.JedisConnectionException;

/**
 * Low level driver class for Redis.
 * 
 * DO NOT MODIFY UNLESS YOU REALLY UNDERSTAND REDIS, THREADING *AND* WHAT YOU ARE DOING 
 * 
 * All methods are *fully* thread safe
 * 
 * @author kanej
 *
 */
public class LowLevelRedisDriver
{
	static public final String DEFAULT_HOST = "localhost";
	static public final int DEFAULT_PORT_REDIS = 6379;
	static public final Logger logger = LoggerFactory.getLogger(LowLevelRedisDriver.class);
	
	private JedisPool pool;
	
	private ExecutorService pool_send;
	private ExecutorService pool_receive;
	
	private RedisCache cache;
	private RedisSignal signal;
	private RedisQueue queue;
	
	public LowLevelRedisDriver()
	{
		String redis_address = System.getProperty("redis.address");
		String host = NetUtils.extractHostFromHostPortPair(redis_address, LowLevelRedisDriver.DEFAULT_HOST);
		int port = NetUtils.extractPortFromHostPortPair(redis_address, DEFAULT_PORT_REDIS);
		
		JedisPoolConfig config = new JedisPoolConfig();
		config.setMaxTotal(250);
		config.setMaxIdle(1000 * 60);
		config.setTestOnBorrow(false);
		
		logger.info("[LowLevelRedisDriver.init] starting LowLevelRedisDriver on host: " + host + ", port: " + port);
		
		pool = new JedisPool(config, host, port);
		cache = new RedisCache();
		queue = new RedisQueue();
		signal = new RedisSignal();
		
		pool_send = DaemonThreadFactory.createDaemonFixedThreadPool(1);
		pool_receive = DaemonThreadFactory.createDaemonFixedThreadPool(10);
		
	}
	
	public boolean isRedisUp()
	{
		try(Jedis jedis = pool.getResource();)
		{
			String str = jedis.ping();
			return str.equalsIgnoreCase("PONG");
		}
		catch(Exception e)
		{
			return false;
		}
	}
	
	/**
	 * Access the cache functionality of the Redis driver
	 * 
	 * @return
	 */
	public RedisCache getSimpleCache() { return cache; }
	
	/**
	 * Access the signal functionality of the Redis driver
	 * 
	 * @return
	 */
	public RedisSignal getSimpleSignal() { return signal; }
	
	/**
	 * Access the queue functionality of the Redis driver
	 * 
	 * @return
	 */
	public RedisQueue getSimpleQueue() { return queue; }
	
	public class RedisCache
	{
		private byte[] createKeyBytes(ApplicationId app, CacheKey key, byte default_value[])
		{
			if ( app == null || key == null ) return default_value;
			
			StringBuilder builder = new StringBuilder();
			builder.append(app.toString());
			builder.append("/");
			builder.append(key.toString());
			
			return builder.toString().getBytes(StandardCharsets.UTF_8);
		}
		
		/**
		 * Set the value of a key in the cache (optionally with a maximum time to live).
		 * Setting with null data has the effect of deleting the key from the cache.
		 * Null keys and app ids result in the function doing nothing
		 * 
		 * @param app
		 *            The application id
		 * @param key
		 *            The key to set
		 * @param data
		 *            The data to write
		 * @param max_ttl
		 *            The maximum length of time that the data will remain in the cache
		 *            (in ms). Zero or negative values have the meaning of "infinite"
		 */
		public void set(ApplicationId app, CacheKey key, byte data[], long max_ttl)
		{
			if ( data == null )
			{
				delete(app,key);
				return;
			}
			
			byte cache_key_bytes[] = createKeyBytes(app,key,null);
			if ( cache_key_bytes == null ) return;
			
			try(Jedis jedis = pool.getResource();)
			{
				try
				{
					if ( max_ttl > 0 )
					{
						long ttl_seconds = max_ttl / 1000;
						String result = jedis.setex(cache_key_bytes, (int) ttl_seconds, data);
						logger.info("result:" + result);
					}
					else
					{
						jedis.set(cache_key_bytes, data);
					}
				}
				catch (Exception e)
				{
					logger.error(String.format("Redis set operation for cache key %s failed.", key.getSimpleValue()), e);
				}
			}
		}
		
		/**
		 * Test for the existence of a key. Null app or key results in a false return
		 * value
		 * 
		 * @param app
		 *            The application id
		 * @param key
		 *            The cache key
		 * 
		 * @return
{'fraction_non_alphanumeric': 0.0989010989010989, 'fraction_numerical': 0.004508312200619893, 'mean_word_length': 3.4422194409678766, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '3112423', 'n_tokens_mistral': 3831, 'n_tokens_neox': 3283, 'n_words': 876}