BUG-1018 Implement BackwardsCompatible BI broker data notifications
[controller.git] / opendaylight / md-sal / sal-dom-broker / src / main / java / org / opendaylight / controller / md / sal / dom / broker / impl / DOMDataBrokerImpl.java
1 /*
2  * Copyright (c) 2014 Cisco Systems, 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 package org.opendaylight.controller.md.sal.dom.broker.impl;
9
10 import static com.google.common.base.Preconditions.checkNotNull;
11 import static com.google.common.base.Preconditions.checkState;
12
13 import java.util.Collections;
14 import java.util.List;
15 import java.util.Map.Entry;
16 import java.util.concurrent.Callable;
17 import java.util.concurrent.ExecutionException;
18 import java.util.concurrent.atomic.AtomicLong;
19
20 import org.opendaylight.controller.md.sal.common.api.TransactionStatus;
21 import org.opendaylight.controller.md.sal.common.api.data.AsyncTransaction;
22 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
23 import org.opendaylight.controller.md.sal.dom.api.DOMDataBroker;
24 import org.opendaylight.controller.md.sal.dom.api.DOMDataChangeListener;
25 import org.opendaylight.controller.md.sal.dom.api.DOMDataReadTransaction;
26 import org.opendaylight.controller.md.sal.dom.api.DOMDataReadWriteTransaction;
27 import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
28 import org.opendaylight.controller.sal.common.util.Rpcs;
29 import org.opendaylight.controller.sal.core.spi.data.DOMStore;
30 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
31 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
32 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
33 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransaction;
34 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
35 import org.opendaylight.yangtools.concepts.ListenerRegistration;
36 import org.opendaylight.yangtools.yang.common.RpcError;
37 import org.opendaylight.yangtools.yang.common.RpcResult;
38 import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier;
39 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
40 import org.slf4j.Logger;
41 import org.slf4j.LoggerFactory;
42
43 import com.google.common.base.Function;
44 import com.google.common.base.Optional;
45 import com.google.common.collect.ImmutableList;
46 import com.google.common.collect.ImmutableList.Builder;
47 import com.google.common.collect.ImmutableMap;
48 import com.google.common.util.concurrent.Futures;
49 import com.google.common.util.concurrent.ListenableFuture;
50 import com.google.common.util.concurrent.ListeningExecutorService;
51
52 public class DOMDataBrokerImpl implements DOMDataBroker, AutoCloseable {
53
54     private static final Logger LOG = LoggerFactory.getLogger(DOMDataBrokerImpl.class);
55     private static final Logger COORDINATOR_LOG = LoggerFactory.getLogger(CommitCoordination.class);
56     private final ImmutableMap<LogicalDatastoreType, DOMStore> datastores;
57     private final ListeningExecutorService executor;
58     private final AtomicLong txNum = new AtomicLong();
59
60     public DOMDataBrokerImpl(final ImmutableMap<LogicalDatastoreType, DOMStore> datastores,
61             final ListeningExecutorService executor) {
62         super();
63         this.datastores = datastores;
64         this.executor = executor;
65     }
66
67     private static final Function<Iterable<Boolean>, Boolean> AND_FUNCTION = new Function<Iterable<Boolean>, Boolean>() {
68
69         @Override
70         public Boolean apply(final Iterable<Boolean> input) {
71
72             for (Boolean value : input) {
73                 if (value == false) {
74                     return Boolean.FALSE;
75                 }
76             }
77             return Boolean.TRUE;
78         }
79     };
80
81     @Override
82     public DOMDataReadTransaction newReadOnlyTransaction() {
83         ImmutableMap.Builder<LogicalDatastoreType, DOMStoreReadTransaction> builder = ImmutableMap.builder();
84         for (Entry<LogicalDatastoreType, DOMStore> store : datastores.entrySet()) {
85             builder.put(store.getKey(), store.getValue().newReadOnlyTransaction());
86         }
87         return new ReadOnlyTransactionImpl(newTransactionIdentifier(), builder.build());
88     }
89
90     private Object newTransactionIdentifier() {
91         return "DOM-" + txNum.getAndIncrement();
92     }
93
94     @Override
95     public DOMDataReadWriteTransaction newReadWriteTransaction() {
96         ImmutableMap.Builder<LogicalDatastoreType, DOMStoreReadWriteTransaction> builder = ImmutableMap.builder();
97         for (Entry<LogicalDatastoreType, DOMStore> store : datastores.entrySet()) {
98             builder.put(store.getKey(), store.getValue().newReadWriteTransaction());
99         }
100         return new ReadWriteTransactionImpl(newTransactionIdentifier(), builder.build(), this);
101     }
102
103     @Override
104     public DOMDataWriteTransaction newWriteOnlyTransaction() {
105         ImmutableMap.Builder<LogicalDatastoreType, DOMStoreWriteTransaction> builder = ImmutableMap.builder();
106         for (Entry<LogicalDatastoreType, DOMStore> store : datastores.entrySet()) {
107             builder.put(store.getKey(), store.getValue().newWriteOnlyTransaction());
108         }
109         return new WriteTransactionImpl<DOMStoreWriteTransaction>(newTransactionIdentifier(), builder.build(), this);
110     }
111
112     @Override
113     public ListenerRegistration<DOMDataChangeListener> registerDataChangeListener(final LogicalDatastoreType store,
114             final InstanceIdentifier path, final DOMDataChangeListener listener, final DataChangeScope triggeringScope) {
115
116         DOMStore potentialStore = datastores.get(store);
117         checkState(potentialStore != null, "Requested logical data store is not available.");
118         return potentialStore.registerChangeListener(path, listener, triggeringScope);
119     }
120
121     private ListenableFuture<RpcResult<TransactionStatus>> submit(
122             final WriteTransactionImpl<? extends DOMStoreWriteTransaction> transaction) {
123         LOG.debug("Tx: {} is submitted for execution.", transaction.getIdentifier());
124         return executor.submit(new CommitCoordination(transaction));
125     }
126
127     private abstract static class AbstractCompositeTransaction<K, T extends DOMStoreTransaction> implements
128             AsyncTransaction<InstanceIdentifier, NormalizedNode<?, ?>> {
129
130         private final ImmutableMap<K, T> backingTxs;
131         private final Object identifier;
132
133         protected AbstractCompositeTransaction(final Object identifier, final ImmutableMap<K, T> backingTxs) {
134             this.identifier = checkNotNull(identifier, "Identifier should not be null");
135             this.backingTxs = checkNotNull(backingTxs, "Backing transactions should not be null");
136         }
137
138         protected T getSubtransaction(final K key) {
139             return backingTxs.get(key);
140         }
141
142         public Iterable<T> getSubtransactions() {
143             return backingTxs.values();
144         }
145
146         @Override
147         public Object getIdentifier() {
148             return identifier;
149         }
150
151         @Override
152         public void close() {
153             try {
154                 for (T subtransaction : backingTxs.values()) {
155                     subtransaction.close();
156                 }
157             } catch (Exception e) {
158                 throw new IllegalStateException("Uncaught exception occured during closing transaction.", e);
159             }
160         }
161
162     }
163
164     private static class ReadOnlyTransactionImpl extends
165             AbstractCompositeTransaction<LogicalDatastoreType, DOMStoreReadTransaction> implements
166             DOMDataReadTransaction {
167
168         protected ReadOnlyTransactionImpl(final Object identifier,
169                 final ImmutableMap<LogicalDatastoreType, DOMStoreReadTransaction> backingTxs) {
170             super(identifier, backingTxs);
171         }
172
173         @Override
174         public ListenableFuture<Optional<NormalizedNode<?, ?>>> read(final LogicalDatastoreType store,
175                 final InstanceIdentifier path) {
176             return getSubtransaction(store).read(path);
177         }
178
179     }
180
181     private static class WriteTransactionImpl<T extends DOMStoreWriteTransaction> extends
182             AbstractCompositeTransaction<LogicalDatastoreType, T> implements DOMDataWriteTransaction {
183
184         private final DOMDataBrokerImpl broker;
185         private ImmutableList<DOMStoreThreePhaseCommitCohort> cohorts;
186
187         protected WriteTransactionImpl(final Object identifier, final ImmutableMap<LogicalDatastoreType, T> backingTxs,
188                 final DOMDataBrokerImpl broker) {
189             super(identifier, backingTxs);
190             this.broker = broker;
191         }
192
193         public synchronized Iterable<DOMStoreThreePhaseCommitCohort> ready() {
194             checkState(cohorts == null, "Transaction was already marked as ready.");
195             ImmutableList.Builder<DOMStoreThreePhaseCommitCohort> cohortsBuilder = ImmutableList.builder();
196             for (DOMStoreWriteTransaction subTx : getSubtransactions()) {
197                 cohortsBuilder.add(subTx.ready());
198             }
199             cohorts = cohortsBuilder.build();
200             return cohorts;
201         }
202
203         protected ImmutableList<DOMStoreThreePhaseCommitCohort> getCohorts() {
204             return cohorts;
205         }
206
207         @Override
208         public void put(final LogicalDatastoreType store, final InstanceIdentifier path, final NormalizedNode<?, ?> data) {
209             getSubtransaction(store).write(path, data);
210         }
211
212         @Override
213         public void delete(final LogicalDatastoreType store, final InstanceIdentifier path) {
214             getSubtransaction(store).delete(path);
215         }
216
217         @Override
218         public void merge(final LogicalDatastoreType store, final InstanceIdentifier path,
219                 final NormalizedNode<?, ?> data) {
220             getSubtransaction(store).merge(path,data);
221         }
222
223         @Override
224         public void cancel() {
225             // TODO Auto-generated method stub
226
227         }
228
229         @Override
230         public ListenableFuture<RpcResult<TransactionStatus>> commit() {
231
232             ready();
233             return broker.submit(this);
234         }
235
236     }
237
238     private static class ReadWriteTransactionImpl extends WriteTransactionImpl<DOMStoreReadWriteTransaction> implements
239             DOMDataReadWriteTransaction {
240
241         protected ReadWriteTransactionImpl(final Object identifier,
242                 final ImmutableMap<LogicalDatastoreType, DOMStoreReadWriteTransaction> backingTxs,
243                 final DOMDataBrokerImpl broker) {
244             // super(identifier, backingTxs);
245             super(identifier, backingTxs, broker);
246         }
247
248         @Override
249         public ListenableFuture<Optional<NormalizedNode<?, ?>>> read(final LogicalDatastoreType store,
250                 final InstanceIdentifier path) {
251             return getSubtransaction(store).read(path);
252         }
253     }
254
255     private final class CommitCoordination implements Callable<RpcResult<TransactionStatus>> {
256
257         private final WriteTransactionImpl<? extends DOMStoreWriteTransaction> transaction;
258
259         public CommitCoordination(final WriteTransactionImpl<? extends DOMStoreWriteTransaction> transaction) {
260             this.transaction = transaction;
261         }
262
263         @Override
264         public RpcResult<TransactionStatus> call() throws Exception {
265
266             try {
267                 Boolean canCommit = canCommit().get();
268
269                 if (canCommit) {
270                     try {
271                         preCommit().get();
272                         try {
273                             commit().get();
274                             COORDINATOR_LOG.debug("Tx: {} Is commited.", transaction.getIdentifier());
275                             return Rpcs.getRpcResult(true, TransactionStatus.COMMITED,
276                                     Collections.<RpcError> emptySet());
277
278                         } catch (InterruptedException | ExecutionException e) {
279                             COORDINATOR_LOG.error("Tx: {} Error during commit", transaction.getIdentifier(), e);
280                         }
281
282                     } catch (InterruptedException | ExecutionException e) {
283                         COORDINATOR_LOG.warn("Tx: {} Error during preCommit, starting Abort",
284                                 transaction.getIdentifier(), e);
285                     }
286                 } else {
287                     COORDINATOR_LOG.info("Tx: {} Did not pass canCommit phase.", transaction.getIdentifier());
288                     abort().get();
289                 }
290             } catch (InterruptedException | ExecutionException e) {
291                 COORDINATOR_LOG.warn("Tx: {} Error during canCommit, starting Abort", transaction.getIdentifier(), e);
292
293             }
294             try {
295                 abort().get();
296             } catch (InterruptedException | ExecutionException e) {
297                 COORDINATOR_LOG.error("Tx: {} Error during abort", transaction.getIdentifier(), e);
298             }
299             return Rpcs.getRpcResult(false, TransactionStatus.FAILED, Collections.<RpcError> emptySet());
300         }
301
302         public ListenableFuture<Void> preCommit() {
303             COORDINATOR_LOG.debug("Transaction {}: PreCommit Started ", transaction.getIdentifier());
304             Builder<ListenableFuture<Void>> ops = ImmutableList.builder();
305             for (DOMStoreThreePhaseCommitCohort cohort : transaction.getCohorts()) {
306                 ops.add(cohort.preCommit());
307             }
308             return (ListenableFuture) Futures.allAsList(ops.build());
309         }
310
311         public ListenableFuture<Void> commit() {
312             COORDINATOR_LOG.debug("Transaction {}: Commit Started ", transaction.getIdentifier());
313             Builder<ListenableFuture<Void>> ops = ImmutableList.builder();
314             for (DOMStoreThreePhaseCommitCohort cohort : transaction.getCohorts()) {
315                 ops.add(cohort.commit());
316             }
317             return (ListenableFuture) Futures.allAsList(ops.build());
318         }
319
320         public ListenableFuture<Boolean> canCommit() {
321             COORDINATOR_LOG.debug("Transaction {}: CanCommit Started ", transaction.getIdentifier());
322             Builder<ListenableFuture<Boolean>> canCommitOperations = ImmutableList.builder();
323             for (DOMStoreThreePhaseCommitCohort cohort : transaction.getCohorts()) {
324                 canCommitOperations.add(cohort.canCommit());
325             }
326             ListenableFuture<List<Boolean>> allCanCommits = Futures.allAsList(canCommitOperations.build());
327             return Futures.transform(allCanCommits, AND_FUNCTION);
328         }
329
330         public ListenableFuture<Void> abort() {
331             COORDINATOR_LOG.debug("Transaction {}: Abort Started ", transaction.getIdentifier());
332             Builder<ListenableFuture<Void>> ops = ImmutableList.builder();
333             for (DOMStoreThreePhaseCommitCohort cohort : transaction.getCohorts()) {
334                 ops.add(cohort.abort());
335             }
336             return (ListenableFuture) Futures.allAsList(ops.build());
337         };
338
339     }
340
341     @Override
342     public void close() throws Exception {
343
344     }
345
346 }