3 * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
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
10 package org.opendaylight.controller.datastore.internal;
12 import com.google.common.base.Preconditions;
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;
28 import java.util.Collections;
29 import java.util.EnumSet;
31 import java.util.concurrent.ConcurrentMap;
34 * The ClusteredDataStoreImpl stores global data to be shared across a controller cluster. It uses Clustering Services.
36 public class ClusteredDataStoreImpl implements ClusteredDataStore {
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";
42 private final ConcurrentMap<InstanceIdentifier, CompositeNode> operationalDataCache;
43 private final ConcurrentMap<InstanceIdentifier, CompositeNode> configurationDataCache;
45 private Logger logger = LoggerFactory.getLogger(ClusteredDataStoreImpl.class);
47 public ClusteredDataStoreImpl(IClusterGlobalServices clusterGlobalServices) throws CacheConfigException {
48 logger.info("Constructing clustered data store");
49 Preconditions.checkNotNull(clusterGlobalServices, "clusterGlobalServices cannot be null");
51 operationalDataCache = getOrCreateCache(clusterGlobalServices, OPERATIONAL_DATA_CACHE);
53 Preconditions.checkNotNull(operationalDataCache, "operationalDataCache cannot be null");
55 configurationDataCache = getOrCreateCache(clusterGlobalServices, CONFIGURATION_DATA_CACHE);
57 Preconditions.checkNotNull(configurationDataCache, "configurationDataCache cannot be null");
61 public DataCommitTransaction<InstanceIdentifier, CompositeNode> requestCommit(DataModification<InstanceIdentifier, CompositeNode> modification) {
62 return new ClusteredDataStoreTransaction(modification);
66 public CompositeNode readOperationalData(InstanceIdentifier path) {
67 Preconditions.checkNotNull(path, "path cannot be null");
68 return operationalDataCache.get(path);
72 public boolean containsConfigurationPath(InstanceIdentifier path) {
73 return configurationDataCache.containsKey(path);
77 public boolean containsOperationalPath(InstanceIdentifier path) {
78 return operationalDataCache.containsKey(path);
82 public Iterable<InstanceIdentifier> getStoredConfigurationPaths() {
83 return configurationDataCache.keySet();
87 public Iterable<InstanceIdentifier> getStoredOperationalPaths() {
88 return operationalDataCache.keySet();
94 public CompositeNode readConfigurationData(InstanceIdentifier path) {
95 Preconditions.checkNotNull(path, "path cannot be null");
96 return configurationDataCache.get(path);
99 private RpcResult<Void> finish(final ClusteredDataStoreTransaction transaction) {
100 final DataModification<InstanceIdentifier,CompositeNode> modification = transaction.getModification();
102 this.configurationDataCache.putAll(modification.getUpdatedConfigurationData());
103 this.operationalDataCache.putAll(modification.getUpdatedOperationalData());
105 for (final InstanceIdentifier removal : modification.getRemovedConfigurationData()) {
106 this.configurationDataCache.remove(removal);
109 for (final InstanceIdentifier removal : modification.getRemovedOperationalData()) {
110 this.operationalDataCache.remove(removal );
113 Set<RpcError> _emptySet = Collections.<RpcError>emptySet();
114 return Rpcs.<Void>getRpcResult(true, null, _emptySet);
117 private RpcResult<Void> rollback(final ClusteredDataStoreTransaction transaction) {
118 Set<RpcError> _emptySet = Collections.<RpcError>emptySet();
119 return Rpcs.<Void>getRpcResult(true, null, _emptySet);
123 private ConcurrentMap getOrCreateCache(IClusterGlobalServices clusterGlobalServices, String name) throws CacheConfigException {
124 ConcurrentMap cache = clusterGlobalServices.getCache(name);
128 cache = clusterGlobalServices.createCache(name, EnumSet.of(IClusterServices.cacheMode.TRANSACTIONAL));
129 } catch (CacheExistException e) {
130 cache = clusterGlobalServices.getCache(name);
136 private class ClusteredDataStoreTransaction implements DataCommitTransaction<InstanceIdentifier, CompositeNode> {
137 private final DataModification<InstanceIdentifier,CompositeNode> modification;
139 public ClusteredDataStoreTransaction(DataModification<InstanceIdentifier,CompositeNode> modification){
140 Preconditions.checkNotNull(modification, "modification cannot be null");
142 this.modification = modification;
146 public DataModification<InstanceIdentifier, CompositeNode> getModification() {
147 return this.modification;
151 public RpcResult<Void> finish() throws IllegalStateException {
152 return ClusteredDataStoreImpl.this.finish(this);
156 public RpcResult<Void> rollback() throws IllegalStateException {
157 return ClusteredDataStoreImpl.this.rollback(this);