mappingservice/clusterdao/src/main/java/org/opendaylight/lispflowmapping/clusterdao/ClusterDAOService.java

   1 /*
   2  * Copyright (c) 2014 Contextream, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.lispflowmapping.clusterdao;
  10
  11 import java.util.Date;
  12 import java.util.EnumSet;
  13 import java.util.Map;
  14 import java.util.concurrent.ConcurrentHashMap;
  15 import java.util.concurrent.ConcurrentMap;
  16 import java.util.concurrent.Executors;
  17 import java.util.concurrent.ScheduledExecutorService;
  18 import java.util.concurrent.TimeUnit;
  19
  20 import org.opendaylight.controller.clustering.services.CacheConfigException;
  21 import org.opendaylight.controller.clustering.services.CacheExistException;
  22 import org.opendaylight.controller.clustering.services.IClusterContainerServices;
  23 import org.opendaylight.controller.clustering.services.IClusterServices;
  24 import org.opendaylight.lispflowmapping.interfaces.dao.ILispDAO;
  25 import org.opendaylight.lispflowmapping.interfaces.dao.IRowVisitor;
  26 import org.opendaylight.lispflowmapping.interfaces.dao.MappingEntry;
  27 import org.opendaylight.lispflowmapping.interfaces.dao.SubKeys;
  28 import org.slf4j.Logger;
  29 import org.slf4j.LoggerFactory;
  30
  31 public class ClusterDAOService implements ILispDAO {
  32
  33     protected static final Logger LOG = LoggerFactory.getLogger(ClusterDAOService.class);
  34     private IClusterContainerServices clusterContainerService = null;
  35     private ConcurrentMap<Object, ConcurrentMap<String, Object>> data;
  36     private final String CACHE_NAME = "mappingServiceCache";
  37     private TimeUnit timeUnit = TimeUnit.SECONDS;
  38     private int recordTimeOut = 240;
  39     private int cleanInterval = 10;
  40     private ScheduledExecutorService scheduler;
  41
  42     void setClusterContainerService(IClusterContainerServices s) {
  43         this.clusterContainerService = s;
  44         allocateCache();
  45         retrieveCache();
  46         scheduler = Executors.newScheduledThreadPool(1);
  47         scheduler.scheduleAtFixedRate(new Runnable() {
  48
  49             public void run() {
  50                 cleanOld();
  51             }
  52         }, 0, cleanInterval, timeUnit);
  53     }
  54
  55     void unsetClusterContainerService(IClusterContainerServices s) {
  56         LOG.trace("Cluster Service unset");
  57         if (this.clusterContainerService == s) {
  58             this.clusterContainerService = null;
  59         }
  60         scheduler.shutdownNow();
  61     }
  62
  63     private void allocateCache() {
  64         if (this.clusterContainerService == null) {
  65             LOG.warn("un-initialized clusterContainerService, can't create cache");
  66             return;
  67         }
  68         LOG.trace("Creating Cache for ClusterDAOService");
  69         try {
  70             this.clusterContainerService.createCache(CACHE_NAME, EnumSet.of(IClusterServices.cacheMode.NON_TRANSACTIONAL));
  71         } catch (CacheConfigException cce) {
  72             LOG.warn("Cache couldn't be created for ClusterDAOService -  check cache mode");
  73         } catch (CacheExistException cce) {
  74             LOG.warn("Cache for ClusterDAOService already exists, destroy and recreate");
  75         }
  76         LOG.trace("Cache successfully created for ClusterDAOService");
  77     }
  78
  79     @SuppressWarnings({ "unchecked" })
  80     private void retrieveCache() {
  81         if (this.clusterContainerService == null) {
  82             LOG.warn("un-initialized clusterContainerService, can't retrieve cache");
  83             return;
  84         }
  85         LOG.trace("Retrieving cache for ClusterDAOService");
  86         data = (ConcurrentMap<Object, ConcurrentMap<String, Object>>) this.clusterContainerService.getCache(CACHE_NAME);
  87         if (data == null) {
  88             LOG.warn("Cache couldn't be retrieved for ClusterDAOService");
  89         }
  90         LOG.trace("Cache was successfully retrieved for ClusterDAOService");
  91     }
  92
  93     public void getAll(IRowVisitor visitor) {
  94         for (ConcurrentMap.Entry<Object, ConcurrentMap<String, Object>> keyEntry : data.entrySet()) {
  95             for (Map.Entry<String, Object> valueEntry : keyEntry.getValue().entrySet()) {
  96                 visitor.visitRow(keyEntry.getKey(), valueEntry.getKey(), valueEntry.getValue());
  97             }
  98         }
  99     }
 100
 101     public void put(Object key, MappingEntry<?>... values) {
 102         if (!data.containsKey(key)) {
 103             data.put(key, new ConcurrentHashMap<String, Object>());
 104         }
 105         for (MappingEntry<?> entry : values) {
 106             data.get(key).put(entry.getKey(), entry.getValue());
 107         }
 108     }
 109
 110     // TODO: this should be moved outside of DAO implementation
 111     public void cleanOld() {
 112         getAll(new IRowVisitor() {
 113             public void visitRow(Object keyId, String valueKey, Object value) {
 114                 if (value != null && valueKey instanceof String && ((String) valueKey).equals(SubKeys.REGDATE)) {
 115                     Date date = (Date) value;
 116                     if (isExpired(date)) {
 117                         removeSpecific(keyId, SubKeys.RECORD);
 118                     }
 119                 }
 120             }
 121
 122             private boolean isExpired(Date date) {
 123                 return System.currentTimeMillis() - date.getTime() > TimeUnit.MILLISECONDS.convert(recordTimeOut, timeUnit);
 124             }
 125         });
 126     }
 127
 128     public Object getSpecific(Object key, String valueKey) {
 129         Map<String, Object> keyToValues = data.get(key);
 130         if (keyToValues == null) {
 131             return null;
 132         }
 133         return keyToValues.get(valueKey);
 134     }
 135
 136     public Map<String, Object> get(Object key) {
 137         return data.get(key);
 138     }
 139
 140     public void remove(Object key) {
 141         data.remove(key);
 142     }
 143
 144     public void removeSpecific(Object key, String valueKey) {
 145         if (data.containsKey(key) && data.get(key).containsKey(valueKey)) {
 146             data.get(key).remove(valueKey);
 147         }
 148     }
 149
 150     public void removeAll() {
 151         data.clear();
 152     }
 153
 154     public TimeUnit getTimeUnit() {
 155         return timeUnit;
 156     }
 157
 158     public void setRecordTimeOut(int recordTimeOut) {
 159         this.recordTimeOut = recordTimeOut;
 160     }
 161
 162     public int getRecordTimeOut() {
 163         return recordTimeOut;
 164     }
 165
 166     public void setTimeUnit(TimeUnit timeUnit) {
 167         this.timeUnit = timeUnit;
 168     }
 169
 170     @Override
 171     public ILispDAO putNestedTable(Object key, String valueKey) {
 172         // TODO Auto-generated method stub
 173         return null;
 174     }
 175 }