<commit_before>/**
 * Copyright 2012 Comcast Corporation
 * 
 * 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.
 */
package com.comcast.cqs.persistence;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

import me.prettyprint.hector.api.exceptions.HTimedOutException;

import org.apache.log4j.Logger;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisShardInfo;
import redis.clients.jedis.ShardedJedis;
import redis.clients.jedis.ShardedJedisPool;
import redis.clients.jedis.Transaction;
import redis.clients.jedis.exceptions.JedisConnectionException;
import redis.clients.jedis.exceptions.JedisException;

import com.comcast.cmb.common.util.CMBProperties;
import com.comcast.cmb.common.util.PersistenceException;
import com.comcast.cmb.common.util.ValueAccumulator.AccumulatorName;
import com.comcast.cqs.controller.CQSCache;
import com.comcast.cqs.controller.CQSControllerServlet;
import com.comcast.cqs.controller.CQSMonitor;
import com.comcast.cqs.controller.CQSMonitor.CacheType;
import com.comcast.cqs.model.CQSMessage;
import com.comcast.cqs.model.CQSQueue;
import com.comcast.cqs.util.CQSConstants;
import com.comcast.cqs.util.Util;

/**
 * This class uses Redis as cache in front of our Cassandra access.
 * Currently we only cache message-ids and hidden/delayed state of messages.
 * The payloads are fetched from underlying persistence layer.
 * 
 * The class must exist as a singleton
 * @author aseem, bwolf, vvenkatraman
 * 
 * Class is thread-safe
 */

public class RedisCachedCassandraPersistence implements ICQSMessagePersistence, ICQSMessagePersistenceIdSequence {
	
    private static final Logger logger = Logger.getLogger(RedisCachedCassandraPersistence.class);
    private static final Random rand = new Random();
    
    private static RedisCachedCassandraPersistence Inst;
    
    public static ExecutorService executor;
    public static ExecutorService revisibilityExecutor;
    
    public final TestInterface testInterface = new TestInterface();
    
    /**
     * 
     * @return Singleton implementation of this object
     */
    public static RedisCachedCassandraPersistence getInstance() {
        return Inst;
    }
    
    private static JedisPoolConfig cfg = new JedisPoolConfig();
    private static ShardedJedisPool pool;
    static {
        initializeInstance();
        initializePool();
    }
    
    private static void initializeInstance() {
        CQSMessagePartitionedCassandraPersistence cassandraPersistence = new CQSMessagePartitionedCassandraPersistence();
        logger.info("event=initialize_redis");
        Inst = new RedisCachedCassandraPersistence(cassandraPersistence);            
    }
    
    private volatile ICQSMessagePersistence persistenceStorage; 
    
    private RedisCachedCassandraPersistence(ICQSMessagePersistence persistenceStorage) {
        this.persistenceStorage = persistenceStorage;
    }
    /**
     * Initialize the Redis connection pool
     */
    private static void initializePool() {
    	
        cfg.maxActive = CMBProperties.getInstance().getRedisConnectionsMaxActive();
        cfg.maxIdle = -1;
        List<JedisShardInfo> shardInfos = new LinkedList<JedisShardInfo>();
        String serverList = CMBProperties.getInstance().getRedisServerList();
        
        if (serverList == null) {
            throw new RuntimeException("Redis server list not specified");
        }
        
        String []arr = serverList.split(",");
        
        for (int i = 0; i < arr.length; i++) {
            
        	String []hostPort = arr[i].trim().split(":");
            
        	if (hostPort.length != 2) {
                throw new RuntimeException("Invalid redis server list: " + serverList);
            }
            
        	JedisShardInfo shardInfo = new JedisShardInfo(hostPort[0].trim(), Integer.parseInt(hostPort[1].trim()), 4000);
            shardInfos.add(shardInfo);
        }
        
        pool = new ShardedJedisPool(cfg, shardInfos);
        executor = Executors.newFixedThreadPool(CMBProperties.getInstance().getRedisFillerThreads());
        revisibilityExecutor = Executors.newFixedThreadPool(CMBProperties.getInstance().getRedisRevisibleThreads());
        
        logger.info("event=initialize_redis pools_size=" + shardInfos.size() + " max_active=" + cfg.maxActive + " server_list=" + serverList);
    }
    <commit_msg>allow redis server list to be initialized with default port if it isn't explicitly stated<commit_after>/**
 * Copyright 2012 Comcast Corporation
 * 
 * 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.
 */
package com.comcast.cqs.persistence;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

import me.prettyprint.hector.api.exceptions.HTimedOutException;

import org.apache.log4j.Logger;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisShardInfo;
import redis.clients.jedis.ShardedJedis;
import redis.clients.jedis.ShardedJedisPool;
import redis.clients.jedis.Transaction;
import redis.clients.jedis.exceptions.JedisConnectionException;
import redis.clients.jedis.exceptions.JedisException;

import com.comcast.cmb.common.util.CMBProperties;
import com.comcast.cmb.common.util.PersistenceException;
import com.comcast.cmb.common.util.ValueAccumulator.AccumulatorName;
import com.comcast.cqs.controller.CQSCache;
import com.comcast.cqs.controller.CQSControllerServlet;
import com.comcast.cqs.controller.CQSMonitor;
import com.comcast.cqs.controller.CQSMonitor.CacheType;
import com.comcast.cqs.model.CQSMessage;
import com.comcast.cqs.model.CQSQueue;
import com.comcast.cqs.util.CQSConstants;
import com.comcast.cqs.util.Util;

/**
 * This class uses Redis as cache in front of our Cassandra access.
 * Currently we only cache message-ids and hidden/delayed state of messages.
 * The payloads are fetched from underlying persistence layer.
 * 
 * The class must exist as a singleton
 * @author aseem, bwolf, vvenkatraman
 * 
 * Class is thread-safe
 */

public class RedisCachedCassandraPersistence implements ICQSMessagePersistence, ICQSMessagePersistenceIdSequence {
	
    private static final Logger logger = Logger.getLogger(RedisCachedCassandraPersistence.class);
    private static final Random rand = new Random();
    
    private static RedisCachedCassandraPersistence Inst;
    
    public static ExecutorService executor;
    public static ExecutorService revisibilityExecutor;
    
    public final TestInterface testInterface = new TestInterface();
    
    /**
     * 
     * @return Singleton implementation of this object
     */
    public static RedisCachedCassandraPersistence getInstance() {
        return Inst;
    }
    
    private static JedisPoolConfig cfg = new JedisPoolConfig();
    private static ShardedJedisPool pool;
    static {
        initializeInstance();
        initializePool();
    }
    
    private static void initializeInstance() {
        CQSMessagePartitionedCassandraPersistence cassandraPersistence = new CQSMessagePartitionedCassandraPersistence();
        logger.info("event=initialize_redis");
        Inst = new RedisCachedCassandraPersistence(cassandraPersistence);            
    }
    
    private volatile ICQSMessagePersistence persistenceStorage; 
    
    private RedisCachedCassandraPersistence(ICQSMessagePersistence persistenceStorage) {
        this.persistenceStorage = persistenceStorage;
    }
    /**
     * Initialize the Redis connection pool
     */
    private static void initializePool() {
    	
        cfg.maxActive = CMBProperties.getInstance().getRedisConnectionsMaxActive();
        cfg.maxIdle = -1;
        List<JedisShardInfo> shardInfos = new LinkedList<JedisShardInfo>();
        String serverList = CMBProperties.getInstance().getRedisServerList();
        
        if (serverList == null) {
            throw new RuntimeException("Redis server list not specified");
        }
        
        String []arr = serverList.split(",");
        
        for (int i = 0; i < arr.length; i++) {
            
        	String []hostPort = arr[i].trim().split(":");
        	JedisShardInfo shardInfo = null;
        	if (hostPort.length != 2) {
        		// use Redis default port if one wasn't specified
        		 shardInfo = new JedisShardInfo(hostPort[0].trim(), 6379, 4000);
            } else {
            	 shardInfo = new JedisShardInfo(hostPort[0].trim(), Integer.parseInt(hostPort[1].trim()), 4000);
            }
            shardInfos.add(shardInfo);
        }
        
        pool = new ShardedJedisPool(cfg, shardInfos);
        executor = Executors.newFixedThreadPool(CMBProperties.getInstance().getRedisFillerThreads());
        revisibilityExecutor = Executors.newFixedThreadPool(CMBProperties.getInstance().getRedisRevisibleThreads());
        
        logger.info("event=initialize_redis pools_size=" + shardInfos.size() + " max_active=" + cfg.maxActive + " server_list=" + serverList);
    }
    
{'fraction_non_alphanumeric': 0.0878415551072804, 'fraction_numerical': 0.004126169081239685, 'mean_word_length': 3.663103890551518, 'pattern_counts': {'":': 2, '<': 9, '<?xml version=': 0, '>': 7, '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': '2144113', 'n_tokens_mistral': 3179, 'n_tokens_neox': 3111, 'n_words': 745}