2 * Copyright (c) 2016 Ericsson India Global Services Pvt Ltd. and others. All rights reserved.
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
8 package org.opendaylight.netvirt.elan.l2gw.ha.merge;
10 import static org.opendaylight.controller.md.sal.binding.api.WriteTransaction.CREATE_MISSING_PARENTS;
11 import static org.opendaylight.genius.infra.Datastore.CONFIGURATION;
12 import static org.opendaylight.genius.infra.Datastore.OPERATIONAL;
14 import com.google.common.base.Optional;
15 import java.util.Collection;
16 import java.util.HashMap;
18 import java.util.Objects;
19 import java.util.concurrent.ExecutionException;
20 import java.util.function.BiPredicate;
22 import org.opendaylight.controller.md.sal.binding.api.DataObjectModification;
23 import org.opendaylight.genius.infra.Datastore;
24 import org.opendaylight.genius.infra.Datastore.Configuration;
25 import org.opendaylight.genius.infra.Datastore.Operational;
26 import org.opendaylight.genius.infra.TypedReadWriteTransaction;
27 import org.opendaylight.genius.utils.SuperTypeUtil;
28 import org.opendaylight.netvirt.elan.l2gw.ha.commands.LocalUcastCmd;
29 import org.opendaylight.netvirt.elan.l2gw.ha.commands.MergeCommand;
30 import org.opendaylight.netvirt.elan.l2gw.ha.commands.RemoteUcastCmd;
31 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
32 import org.opendaylight.yangtools.concepts.Builder;
33 import org.opendaylight.yangtools.yang.binding.DataObject;
34 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
35 import org.slf4j.Logger;
36 import org.slf4j.LoggerFactory;
38 public abstract class MergeCommandsAggregator<BuilderTypeT extends Builder, AugTypeT extends DataObject> {
40 private static final Logger LOG = LoggerFactory.getLogger(MergeCommandsAggregator.class);
42 protected Map<Class<?>, MergeCommand> commands = new HashMap<>();
44 private final BiPredicate<Class<? extends Datastore>, Class> skipCopy =
45 (dsType, cmdType) -> (Configuration.class.equals(dsType) ? commands.get(cmdType) instanceof LocalUcastCmd
46 : commands.get(cmdType) instanceof RemoteUcastCmd);
48 protected MergeCommandsAggregator() {
51 protected void addCommand(MergeCommand mergeCommand) {
52 commands.put(SuperTypeUtil.getTypeParameter(mergeCommand.getClass(), 0), mergeCommand);
55 public void mergeOperationalData(BuilderTypeT builder,
56 AugTypeT existingData,
58 InstanceIdentifier<Node> dstPath) {
59 for (MergeCommand cmd : commands.values()) {
60 if (skipCopy.negate().test(OPERATIONAL, cmd.getClass())) {
61 cmd.mergeOperationalData(builder, existingData, src, dstPath);
66 public void mergeConfigData(BuilderTypeT builder,
68 InstanceIdentifier<Node> dstPath) {
69 for (MergeCommand cmd : commands.values()) {
70 if (skipCopy.negate().test(CONFIGURATION, cmd.getClass())) {
71 cmd.mergeConfigData(builder, src, dstPath);
77 public void mergeConfigUpdate(InstanceIdentifier<Node> dstPath,
78 DataObjectModification mod,
79 TypedReadWriteTransaction<Configuration> tx) {
80 mergeUpdate(dstPath, mod, CONFIGURATION, tx);
83 public void mergeOpUpdate(InstanceIdentifier<Node> dstPath,
84 DataObjectModification mod,
85 TypedReadWriteTransaction<Operational> tx) {
86 mergeUpdate(dstPath, mod, OPERATIONAL, tx);
89 public <D extends Datastore> void mergeUpdate(InstanceIdentifier<Node> dstPath,
90 DataObjectModification mod,
91 Class<D> datastoreType,
92 TypedReadWriteTransaction<D> tx) {
96 Collection<DataObjectModification> modifications = mod.getModifiedChildren();
97 modifications.stream()
98 .filter(modification -> skipCopy.negate().test(datastoreType, modification.getDataType()))
99 .filter(modification -> commands.get(modification.getDataType()) != null)
100 .peek(modification -> LOG.debug("Received {} modification {} copy/delete to {}",
101 datastoreType, modification, dstPath))
102 .forEach(modification -> {
103 MergeCommand mergeCommand = commands.get(modification.getDataType());
104 DataObject dataAfter = modification.getDataAfter();
105 boolean create = dataAfter != null;
106 DataObject data = create ? dataAfter : modification.getDataBefore();
107 InstanceIdentifier<DataObject> transformedId = mergeCommand.generateId(dstPath, data);
108 DataObject transformedItem = mergeCommand.transform(dstPath, data);
110 Optional<DataObject> existingDataOptional = null;
112 existingDataOptional = tx.read(transformedId).get();
113 } catch (InterruptedException | ExecutionException ex) {
114 LOG.error("Failed to read data {} from {}", transformedId, datastoreType);
118 String destination = Configuration.class.equals(datastoreType) ? "child" : "parent";
120 if (isDataUpdated(existingDataOptional, transformedItem)) {
121 LOG.debug("Copy to {} {} {}", destination, datastoreType, transformedId);
122 tx.put(transformedId, transformedItem, CREATE_MISSING_PARENTS);
124 LOG.debug("Data not updated skip copy to {}", transformedId);
127 if (existingDataOptional.isPresent()) {
128 LOG.debug("Delete from {} {} {}", destination, datastoreType, transformedId);
129 tx.delete(transformedId);
131 LOG.debug("Delete skipped for {}", transformedId);
137 boolean isDataUpdated(Optional<DataObject> existingDataOptional, DataObject newData) {
138 return !existingDataOptional.isPresent() || !Objects.equals(existingDataOptional.get(), newData);