opendaylight/md-sal/clustered-data-store/implementation/src/main/java/org/opendaylight/controller/datastore/internal/ClusteredDataStoreImpl.java

   1
   2 /*
   3  * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
   4  *
   5  * This program and the accompanying materials are made available under the
   6  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   7  * and is available at http://www.eclipse.org/legal/epl-v10.html
   8  */
   9
  10 package org.opendaylight.controller.datastore.internal;
  11
  12 import com.google.common.base.Preconditions;
  13
  14 import org.opendaylight.controller.clustering.services.CacheConfigException;
  15 import org.opendaylight.controller.clustering.services.CacheExistException;
  16 import org.opendaylight.controller.clustering.services.IClusterGlobalServices;
  17 import org.opendaylight.controller.clustering.services.IClusterServices;
  18 import org.opendaylight.controller.datastore.ClusteredDataStore;
  19 import org.opendaylight.controller.md.sal.common.api.data.DataModification;
  20 import org.opendaylight.controller.sal.common.util.Rpcs;
  21 import org.opendaylight.yangtools.yang.common.RpcError;
  22 import org.opendaylight.yangtools.yang.common.RpcResult;
  23 import org.opendaylight.yangtools.yang.data.api.CompositeNode;
  24 import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier;
  25 import org.slf4j.Logger;
  26 import org.slf4j.LoggerFactory;
  27
  28 import java.util.Collections;
  29 import java.util.EnumSet;
  30 import java.util.Set;
  31 import java.util.concurrent.ConcurrentMap;
  32
  33 /**
  34  * The ClusteredDataStoreImpl stores global data to be shared across a controller cluster. It uses Clustering Services.
  35  */
  36 public class ClusteredDataStoreImpl implements ClusteredDataStore {
  37
  38
  39     public static final String OPERATIONAL_DATA_CACHE = "clustered_data_store.operational_data_cache";
  40     public static final String CONFIGURATION_DATA_CACHE = "clustered_data_store.configuration_data_cache";
  41
  42     private final ConcurrentMap<InstanceIdentifier, CompositeNode> operationalDataCache;
  43     private final ConcurrentMap<InstanceIdentifier, CompositeNode> configurationDataCache;
  44
  45     private Logger logger = LoggerFactory.getLogger(ClusteredDataStoreImpl.class);
  46
  47     public ClusteredDataStoreImpl(IClusterGlobalServices clusterGlobalServices) throws CacheConfigException {
  48         logger.trace("Constructing clustered data store");
  49         Preconditions.checkNotNull(clusterGlobalServices, "clusterGlobalServices cannot be null");
  50
  51         operationalDataCache = getOrCreateCache(clusterGlobalServices, OPERATIONAL_DATA_CACHE);
  52
  53         Preconditions.checkNotNull(operationalDataCache, "operationalDataCache cannot be null");
  54
  55         configurationDataCache = getOrCreateCache(clusterGlobalServices, CONFIGURATION_DATA_CACHE);
  56
  57         Preconditions.checkNotNull(configurationDataCache, "configurationDataCache cannot be null");
  58     }
  59
  60     @Override
  61     public DataCommitTransaction<InstanceIdentifier, CompositeNode> requestCommit(DataModification<InstanceIdentifier, CompositeNode> modification) {
  62         return new ClusteredDataStoreTransaction(modification);
  63     }
  64
  65     @Override
  66     public CompositeNode readOperationalData(InstanceIdentifier path) {
  67         Preconditions.checkNotNull(path, "path cannot be null");
  68         return operationalDataCache.get(path);
  69     }
  70
  71     @Override
  72     public boolean containsConfigurationPath(InstanceIdentifier path) {
  73         return configurationDataCache.containsKey(path);
  74     }
  75
  76     @Override
  77     public boolean containsOperationalPath(InstanceIdentifier path) {
  78         return operationalDataCache.containsKey(path);
  79     }
  80
  81     @Override
  82     public Iterable<InstanceIdentifier> getStoredConfigurationPaths() {
  83         return configurationDataCache.keySet();
  84     }
  85
  86     @Override
  87     public Iterable<InstanceIdentifier> getStoredOperationalPaths() {
  88         return operationalDataCache.keySet();
  89     }
  90
  91
  92
  93     @Override
  94     public CompositeNode readConfigurationData(InstanceIdentifier path) {
  95         Preconditions.checkNotNull(path, "path cannot be null");
  96         return configurationDataCache.get(path);
  97     }
  98
  99     private RpcResult<Void> finish(final ClusteredDataStoreTransaction transaction) {
 100       final DataModification<InstanceIdentifier,CompositeNode> modification = transaction.getModification();
 101
 102       this.configurationDataCache.putAll(modification.getUpdatedConfigurationData());
 103       this.operationalDataCache.putAll(modification.getUpdatedOperationalData());
 104
 105       for (final InstanceIdentifier removal : modification.getRemovedConfigurationData()) {
 106         this.configurationDataCache.remove(removal);
 107       }
 108
 109       for (final InstanceIdentifier removal : modification.getRemovedOperationalData()) {
 110         this.operationalDataCache.remove(removal  );
 111       }
 112
 113       Set<RpcError> _emptySet = Collections.<RpcError>emptySet();
 114       return Rpcs.<Void>getRpcResult(true, null, _emptySet);
 115     }
 116
 117     private RpcResult<Void> rollback(final ClusteredDataStoreTransaction transaction) {
 118       Set<RpcError> _emptySet = Collections.<RpcError>emptySet();
 119       return Rpcs.<Void>getRpcResult(true, null, _emptySet);
 120     }
 121
 122
 123     private ConcurrentMap getOrCreateCache(IClusterGlobalServices clusterGlobalServices, String name) throws CacheConfigException {
 124         ConcurrentMap cache = clusterGlobalServices.getCache(name);
 125
 126         if(cache == null) {
 127             try {
 128                 cache = clusterGlobalServices.createCache(name, EnumSet.of(IClusterServices.cacheMode.TRANSACTIONAL));
 129             } catch (CacheExistException e) {
 130                 cache = clusterGlobalServices.getCache(name);
 131             }
 132         }
 133         return cache;
 134     }
 135
 136     private class ClusteredDataStoreTransaction implements DataCommitTransaction<InstanceIdentifier, CompositeNode> {
 137         private final DataModification<InstanceIdentifier,CompositeNode> modification;
 138
 139         public ClusteredDataStoreTransaction(DataModification<InstanceIdentifier,CompositeNode> modification){
 140             Preconditions.checkNotNull(modification, "modification cannot be null");
 141
 142             this.modification = modification;
 143         }
 144
 145         @Override
 146         public DataModification<InstanceIdentifier, CompositeNode> getModification() {
 147             return this.modification;
 148         }
 149
 150         @Override
 151         public RpcResult<Void> finish() throws IllegalStateException {
 152             return ClusteredDataStoreImpl.this.finish(this);
 153         }
 154
 155         @Override
 156         public RpcResult<Void> rollback() throws IllegalStateException {
 157             return ClusteredDataStoreImpl.this.rollback(this);
 158         }
 159     }
 160 }