Merge pull request #41 from vdoleans01/master

Possibility to reset connection to redis

Author: vdoleans01

session-replacement/src/main/java/com/amadeus/session/ErrorTracker.java

@@ -0,0 +1,71 @@
+package com.amadeus.session;
+
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * The class is used to determine if the application is in error or not
+ * 
+ * If the number of error stored during the delay reach the max , the application is in error
+ * 
+ * @param period
+ *          Time in milliseconds while items are kept
+ * @param max
+ *          If the number of items is more than the max value the Tracker is considered in error
+ *
+ */
+
+public class ErrorTracker {
+  public ErrorTracker(int period, int max) {
+    super();
+    this.period = period;
+    this.max = max;
+  }
+
+  private ConcurrentLinkedQueue<Long> list = new ConcurrentLinkedQueue<Long>();
+
+  /**
+   * When a new event it added , all the event that is older that ( event time - period ) are removed from the tracker
+   */
+  final int period;
+
+  /**
+   * After the limits the Tracker is considering in error
+   */
+  final int max;
+
+  /**
+   * The parameter is a time with the format millisecond from 1900 ( System.currentTimeMillis() ) This method add an
+   * event into the Tracker and remove all the old event with the followinf criteria
+   * 
+   * now - oldevent > period will be
+   * 
+   * removed
+   * 
+   * @param now
+   */
+
+  public void addError(long now) {
+    list.add(new Long(now));
+    boolean cont = true;
+    while (cont) {
+      Long last = list.peek();
+      if (now - last > period) {
+        list.poll();
+      } else {
+        cont = false;
+      }
+    }
+  }
+
+  public boolean reachLimits() {
+    return list.size() >= max;
+  }
+
+  public int size() {
+    return list.size();
+  }
+
+  public void reset() {
+    list.clear();
+  }
+}

session-replacement/src/main/java/com/amadeus/session/ResetManager.java

@@ -0,0 +1,87 @@
+package com.amadeus.session;
+
+import static com.codahale.metrics.MetricRegistry.name;
+
+import com.codahale.metrics.JmxReporter;
+import com.codahale.metrics.Meter;
+import com.codahale.metrics.MetricRegistry;
+
+public class ResetManager {
+  private final MetricRegistry monitoring;
+
+  private JmxReporter reporter;
+
+  private final ErrorTracker errorTracker;
+
+  private final Meter create_meter;
+
+  private final Meter notConnected_meter;
+
+  private final Meter connected_meter;
+
+  private final SessionConfiguration configuration;
+
+  protected final ExecutorFacade executors;
+
+  final String SESSIONS_METRIC_MANAGER_PREFIX = "com.amadeus.session.manager";
+
+  final String RESETMANAGER_CREATED_METRIC = name(SESSIONS_METRIC_MANAGER_PREFIX, "initialized");
+
+  final String RESETMANAGER_NOTCONNECTED_METRIC = name(SESSIONS_METRIC_MANAGER_PREFIX, "notConnected");
+
+  final String RESETMANAGER_CONNECTED_METRIC = name(SESSIONS_METRIC_MANAGER_PREFIX, "connected");
+
+  public ResetManager(ExecutorFacade executors, SessionConfiguration configuration) {
+    this.executors = executors;
+
+    monitoring = new MetricRegistry();
+
+    this.configuration = configuration;
+
+    create_meter = monitoring.meter(RESETMANAGER_CREATED_METRIC);
+    notConnected_meter = monitoring.meter(RESETMANAGER_NOTCONNECTED_METRIC);
+    connected_meter = monitoring.meter(RESETMANAGER_CONNECTED_METRIC);
+
+    errorTracker = new ErrorTracker(configuration.getTrackerInterval(), configuration.getTrackerLimits());
+
+    startMonitoring();
+
+    create_meter.mark();
+  }
+
+  /**
+   * Starts monitoring this session manager. The method will expose all metrics through JMX.
+   */
+  private void startMonitoring() {
+    executors.startMetrics(monitoring);
+    reporter = JmxReporter.forRegistry(monitoring).inDomain(getJmxDomain()).build();
+    reporter.start();
+  }
+
+  /**
+   * Returns JMX domain for metrics for this session manager.
+   *
+   * @return JMX domain for metrics
+   */
+  private String getJmxDomain() {
+    return "metrics.session." + configuration.getNamespace();
+  }
+
+  public ErrorTracker getErrorTracker() {
+    return errorTracker;
+  }
+
+  public void reset() {
+    errorTracker.reset();
+    create_meter.mark();
+  }
+
+  public void connected() {
+    connected_meter.mark();
+  }
+
+  public void notConnected() {
+    notConnected_meter.mark();
+  }
+
+}

session-replacement/src/main/java/com/amadeus/session/SessionConfiguration.java

@@ -130,6 +130,22 @@ public interface AttributeProvider {
     Object source();
   }
 
+  
+  
+  
+  
+  
+  /**
+   * Indicates the interval when error are count.
+   */
+  public static final int DEFAULT_TRACKER_ERROR_INTERVAL_MILLISECONDS_NUM = 60000;
+  public static final String TRACKER_ERROR_INTERVAL_MILLISECONDS_KEY = "com.amadeus.session.tracker.interval" ;
+  /**
+   * Indicates the maximun number of errors before request the reset of redis connection.
+   */
+  public static final int DEFAULT_TRACKER_ERROR_LIMITS_NUMBER = 50;
+  public static final String  TRACKER_ERROR_LIMITS_NUMBER_KEY = "com.amadeus.session.tracker.limits" ;
+  
   /**
    * Indicates if sessions can be distributed or not.
    */
@@ -303,6 +319,8 @@ public interface AttributeProvider {
    */
   public static final String DELEGATE_WRITER = "com.amadeus.session.delegate.writer";
 
+  private int trackerInterval;
+  private int trackerLimits;
   private int maxInactiveInterval;
   private boolean distributable;
   private boolean sticky;
@@ -362,10 +380,31 @@ public SessionConfiguration() {
       logger.error("`{}` system property was not an integer: {}, using default {}", DEFAULT_SESSION_TIMEOUT,
           inactiveValue, maxInactiveInterval);
     }
+    
+    trackerInterval  = init(TRACKER_ERROR_INTERVAL_MILLISECONDS_KEY,DEFAULT_TRACKER_ERROR_INTERVAL_MILLISECONDS_NUM);
+    trackerLimits  = init(TRACKER_ERROR_LIMITS_NUMBER_KEY,DEFAULT_TRACKER_ERROR_LIMITS_NUMBER);
+    logger.error("trackerInterval:" + trackerInterval );
+    logger.error("trackerLimits  :" + trackerLimits );
+    
+    
     node = initNode();
     setEncryptionKey(getPropertySecured(SESSION_ENCRYPTION_KEY, null));
   }
 
+  private int init( String key , int def) {
+    int intVal = def;
+    String strVal = getPropertySecured(key , String.valueOf(def));
+    try {
+      if (nonEmpty(strVal)) {
+        intVal = Integer.parseInt(strVal);
+      }
+    } catch (NumberFormatException e) {
+      logger.error("`{}` system property was not an integer: {}, using default {}", key,
+          strVal, intVal);
+    }
+    return intVal ;
+  }
+
   /**
    * Parses value as comma-separated list of session propagators. Used to
    * specify several propagator in order of priority.
@@ -435,19 +474,26 @@ public void initializeFrom(AttributeProvider provider) {
     if (nonEmpty(value)) {
       setNonCacheable(value);
     }
-    initMaxInactiveInterval(provider);
+    maxInactiveInterval = initInt(provider,DEFAULT_SESSION_TIMEOUT,maxInactiveInterval);
+    trackerLimits = initInt(provider,TRACKER_ERROR_LIMITS_NUMBER_KEY ,trackerLimits);
+    trackerInterval = initInt(provider,TRACKER_ERROR_INTERVAL_MILLISECONDS_KEY,trackerInterval);
+    
+    logger.error("trackerInterval:" + trackerInterval );
+    logger.error("trackerLimits  :" + trackerLimits );
+    
   }
 
-  private void initMaxInactiveInterval(AttributeProvider provider) {
-    String val = provider.getAttribute(DEFAULT_SESSION_TIMEOUT);
+  private int initInt(AttributeProvider provider , String name , int maxInactiveInterval) {
+    String val = provider.getAttribute(name);
     if (nonEmpty(val)) {
       try {
-        maxInactiveInterval = Integer.parseInt(val);
+        return Integer.parseInt(val);
       } catch (NumberFormatException e) {
-        logger.warn("`{}` configuration attribute was not an integer: {} for source {}", DEFAULT_SESSION_TIMEOUT, val,
+        logger.warn("`{}` configuration attribute was not an integer: {} for source {}", name, val,
             provider.source());
       }
     }
+    return maxInactiveInterval; 
   }
 
   private boolean read(String key, boolean defaultValue) {
@@ -1046,4 +1092,20 @@ public String toString() {
            .append(commitOnAllConcurrent).append(", timestamp=").append(timestampSufix).append("]");
     return builder.toString();
   }
+
+  public int getTrackerInterval() {
+    return trackerInterval;
+  }
+
+  public void setTrackerInterval(int trackerInterval) {
+    this.trackerInterval = trackerInterval;
+  }
+
+  public int getTrackerLimits() {
+    return trackerLimits;
+  }
+
+  public void setTrackerLimits(int trackerLimits) {
+    this.trackerLimits = trackerLimits;
+  }
 }

session-replacement/src/main/java/com/amadeus/session/SessionManager.java

@@ -55,7 +55,10 @@ public class SessionManager implements Closeable {
   private static final Logger logger = LoggerFactory.getLogger(SessionManager.class);
 
   static final String SESSIONS_METRIC_PREFIX = "com.amadeus.session";
+  
 
+  
+  
   private static final String COMMIT_TIMER_METRIC = name(SESSIONS_METRIC_PREFIX, "timer", "commit");
   private static final String FETCH_TIMER_METRIC = name(SESSIONS_METRIC_PREFIX, "timer", "fetch");
   private static final String CREATED_SESSIONS_METRIC = name(SESSIONS_METRIC_PREFIX, "created");
@@ -115,7 +118,7 @@ public class SessionManager implements Closeable {
    *          the class loader to use
    */
   public SessionManager(ExecutorFacade executors, SessionFactory factory, SessionRepository repository,
-      SessionTracking tracking, SessionNotifier notifier, SessionConfiguration configuration, ClassLoader classLoader) {
+      SessionTracking tracking, SessionNotifier notifier, SessionConfiguration configuration, ClassLoader classLoader ) {
 
     this.repository = repository;
     this.tracking = tracking;
@@ -126,6 +129,7 @@ public SessionManager(ExecutorFacade executors, SessionFactory factory, SessionR
     this.classLoader = classLoader;
 
     monitoring = new MetricRegistry();
+    
     createdSessions = monitoring.meter(CREATED_SESSIONS_METRIC);
     deletedSessions = monitoring.meter(DELETED_SESSIONS_METRIC);
     retrievedSessions = monitoring.meter(RETRIEVED_SESSIONS_METRIC);
@@ -134,7 +138,7 @@ public SessionManager(ExecutorFacade executors, SessionFactory factory, SessionR
     invalidationExpiryErrors = monitoring.meter(INVALIDATION_ON_EXPIRY_ERRORS_METRIC);
     commitTimer = monitoring.timer(COMMIT_TIMER_METRIC);
     fetchTimer = monitoring.timer(FETCH_TIMER_METRIC);
-
+    
     serializerDeserializer = configuration.isUsingEncryption() ?
         new EncryptingSerializerDeserializer() :
         new JdkSerializerDeserializer();
@@ -570,6 +574,21 @@ public String encodeUrl(RequestWithSession request, String url) {
     return tracking.encodeUrl(request, url);
   }
 
+ 
+  
+  /**
+   * Called to shutdown the session manager and perform needed cleanup.
+   */
+  
+  public void reset() {
+	  if (reporter != null) {
+		  reporter.stop();
+		  reporter.close();
+	  }
+	  
+	  repository.reset();	  
+	  executors.shutdown();
+  }  
   /**
    * Called to shutdown the session manager and perform needed cleanup.
    */
@@ -637,4 +656,9 @@ public void remove(SessionData sessionData) {
     markSessionDeletion( sessionData.getId() );
     getRepository().remove(sessionData);
   }
+  
+  public boolean isConnected(){
+    return this.repository.isConnected();
+  }
+  
 }

session-replacement/src/main/java/com/amadeus/session/SessionRepository.java

@@ -139,6 +139,11 @@ public interface SessionRepository extends Closeable {
    */
   Collection<String> getOwnedSessionIds();
 
+  /**
+   * Called to reset the repository and release resources.
+   */
+  public void reset() ;
+  
   /**
    * Called to shutdown the repository and release resources.
    */
@@ -210,6 +215,11 @@ interface CommitTransaction {
      *         data being distributed to remote store.
      */
     boolean isDistributing();
+    
+   
+    
   }
 
+  boolean isConnected();
+
 }