BUG-5280: add basic concept of ClientSnapshot

[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / databroker / actors / dds / AbstractClientHandle.java

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/AbstractClientHandle.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/AbstractClientHandle.java
new file mode 100644 (file)
index 0000000..b87819c
--- /dev/null
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/AbstractClientHandle.java
@@ -0,0 +1,119 @@
+/*
+ * Copyright (c) 2016 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.cluster.databroker.actors.dds;
+
+import com.google.common.annotations.Beta;
+import com.google.common.base.Preconditions;
+import java.util.Collection;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+import java.util.function.Function;
+import javax.annotation.Nullable;
+import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
+import org.opendaylight.yangtools.concepts.Identifiable;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Abstract superclass of both ClientSnapshot and ClientTransaction. Provided for convenience.
+ *
+ * @author Robert Varga
+ */
+@Beta
+public abstract class AbstractClientHandle<T extends AbstractProxyTransaction> extends LocalAbortable
+        implements Identifiable<TransactionIdentifier> {
+    /*
+     * Our state consist of the the proxy map, hence we just subclass ConcurrentHashMap directly.
+     */
+    private static final class State<T> extends ConcurrentHashMap<Long, T> {
+        private static final long serialVersionUID = 1L;
+    }
+
+    private static final Logger LOG = LoggerFactory.getLogger(AbstractClientHandle.class);
+    @SuppressWarnings("rawtypes")
+    private static final AtomicReferenceFieldUpdater<AbstractClientHandle, State> STATE_UPDATER =
+            AtomicReferenceFieldUpdater.newUpdater(AbstractClientHandle.class, State.class, "state");
+
+    private final TransactionIdentifier transactionId;
+    private final AbstractClientHistory parent;
+
+    private volatile State<T> state = new State<>();
+
+    // Hidden to prevent outside instantiation
+    AbstractClientHandle(final AbstractClientHistory parent, final TransactionIdentifier transactionId) {
+        this.transactionId = Preconditions.checkNotNull(transactionId);
+        this.parent = Preconditions.checkNotNull(parent);
+    }
+
+    @Override
+    public final TransactionIdentifier getIdentifier() {
+        return transactionId;
+    }
+
+    /**
+     * Release all state associated with this transaction.
+     *
+     * @return True if this transaction became closed during this call
+     */
+    public final boolean abort() {
+        if (commonAbort()) {
+            parent.onTransactionAbort(this);
+            return true;
+        }
+
+        return false;
+    }
+
+    private boolean commonAbort() {
+        final Collection<T> toClose = ensureClosed();
+        if (toClose == null) {
+            return false;
+        }
+
+        toClose.forEach(AbstractProxyTransaction::abort);
+        return true;
+    }
+
+    @Override
+    final void localAbort(final Throwable cause) {
+        LOG.debug("Local abort of transaction {}", getIdentifier(), cause);
+        commonAbort();
+    }
+
+    /**
+     * Make sure this snapshot is closed. If it became closed as the effect of this call, return a collection of
+     * {@link AbstractProxyTransaction} handles which need to be closed, too.
+     *
+     * @return null if this snapshot has already been closed, otherwise a collection of proxies, which need to be
+     *         closed, too.
+     */
+    @Nullable final Collection<T> ensureClosed() {
+        @SuppressWarnings("unchecked")
+        final State<T> local = STATE_UPDATER.getAndSet(this, null);
+        return local == null ? null : local.values();
+    }
+
+    final T ensureProxy(final YangInstanceIdentifier path, final Function<Long, T> createProxy) {
+        final Map<Long, T> local = getState();
+        final Long shard = parent.resolveShardForPath(path);
+
+        return local.computeIfAbsent(shard, createProxy);
+    }
+
+    final AbstractClientHistory parent() {
+        return parent;
+    }
+
+    private State<T> getState() {
+        final State<T> local = state;
+        Preconditions.checkState(local != null, "Transaction %s is closed", transactionId);
+        return local;
+    }
+}