Move graph related code to fluent core

azure-mgmt-resources/src/main/java/com/microsoft/azure/management/resources/fluentcore/dag/DAGNode.java

@@ -0,0 +1,126 @@
+/**
+ *
+ * Copyright (c) Microsoft Corporation. All rights reserved.
+ * Licensed under the MIT License. See License.txt in the project root for license information.
+ *
+ */
+
+package com.microsoft.azure.management.resources.fluentcore.dag;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * The type representing node in a {@link DAGraph}.
+ *
+ * @param <T> the type of the data stored in the node
+ */
+public class DAGNode<T> extends Node<T> {
+    private List<String> dependentKeys;
+    private int toBeResolved;
+    private boolean isPreparer;
+    private ReentrantLock lock;
+
+    /**
+     * Creates a DAG node.
+     *
+     * @param key unique id of the node
+     * @param data data to be stored in the node
+     */
+    public DAGNode(String key, T data) {
+        super(key, data);
+        dependentKeys = new ArrayList<>();
+        lock = new ReentrantLock();
+    }
+
+    /**
+     * @return the lock to be used while performing thread safe operation on this node.
+     */
+    public ReentrantLock lock() {
+        return this.lock;
+    }
+
+    /**
+     * @return a list of keys of nodes in {@link DAGraph} those are dependents on this node
+     */
+    List<String> dependentKeys() {
+        return Collections.unmodifiableList(this.dependentKeys);
+    }
+
+    /**
+     * Mark the node identified by the given key as dependent of this node.
+     *
+     * @param key the id of the dependent node
+     */
+    public void addDependent(String key) {
+        this.dependentKeys.add(key);
+    }
+
+    /**
+     * @return a list of keys of nodes in {@link DAGraph} that this node depends on
+     */
+    public List<String> dependencyKeys() {
+        return this.children();
+    }
+
+    /**
+     * Mark the node identified by the given key as this node's dependency.
+     *
+     * @param dependencyKey the id of the dependency node
+     */
+    public void addDependency(String dependencyKey) {
+        super.addChild(dependencyKey);
+    }
+
+    /**
+     * @return <tt>true</tt> if this node has any dependency
+     */
+    public boolean hasDependencies() {
+        return this.hasChildren();
+    }
+
+    /**
+     * Mark or un-mark this node as preparer.
+     *
+     * @param isPreparer <tt>true</tt> if this node needs to be marked as preparer, <tt>false</tt> otherwise.
+     */
+    public void setPreparer(boolean isPreparer) {
+        this.isPreparer = isPreparer;
+    }
+
+    /**
+     * @return <tt>true</tt> if this node is marked as preparer
+     */
+    public boolean isPreparer() {
+        return isPreparer;
+    }
+
+    /**
+     * Initialize the node so that traversal can be performed on the parent DAG.
+     */
+    public void initialize() {
+        this.toBeResolved = this.dependencyKeys().size();
+        this.dependentKeys.clear();
+    }
+
+    /**
+     * @return <tt>true</tt> if all dependencies of this node are ready to be consumed
+     */
+    boolean hasAllResolved() {
+        return toBeResolved == 0;
+    }
+
+    /**
+     * Reports that one of this node's dependency has been resolved and ready to be consumed.
+     *
+     * @param dependencyKey the id of the dependency node
+     */
+    void reportResolved(String dependencyKey) {
+        if (toBeResolved == 0) {
+            throw new RuntimeException("invalid state - " + this.key() + ": The dependency '" + dependencyKey + "' is already reported or there is no such dependencyKey");
+        }
+        toBeResolved--;
+    }
+}

azure-mgmt-resources/src/main/java/com/microsoft/azure/management/resources/fluentcore/dag/DAGraph.java

@@ -0,0 +1,193 @@
+/**
+ *
+ * Copyright (c) Microsoft Corporation. All rights reserved.
+ * Licensed under the MIT License. See License.txt in the project root for license information.
+ *
+ */
+
+package com.microsoft.azure.management.resources.fluentcore.dag;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * Type representing a DAG (directed acyclic graph).
+ * <p>
+ * each node in a DAG is represented by {@link DAGNode}
+ *
+ * @param <T> the type of the data stored in the graph nodes
+ * @param <U> the type of the nodes in the graph
+ */
+public class DAGraph<T, U extends DAGNode<T>> extends Graph<T, U> {
+    private ConcurrentLinkedQueue<String> queue;
+    private boolean hasParent;
+    private U rootNode;
+
+    /**
+     * Creates a new DAG.
+     *
+     * @param rootNode the root node of this DAG
+     */
+    public DAGraph(U rootNode) {
+        this.rootNode = rootNode;
+        this.queue = new ConcurrentLinkedQueue<>();
+        this.rootNode.setPreparer(true);
+        this.addNode(rootNode);
+    }
+
+    /**
+     * @return <tt>true</tt> if this DAG is merged with another DAG and hence has a parent
+     */
+    public boolean hasParent() {
+        return hasParent;
+    }
+
+    /**
+     * Checks whether the given node is root node of this DAG.
+     *
+     * @param node the node {@link DAGNode} to be checked
+     * @return <tt>true</tt> if the given node is root node
+     */
+    public boolean isRootNode(U node) {
+        return this.rootNode == node;
+    }
+
+    /**
+     * @return <tt>true</tt> if this dag is the preparer responsible for
+     * preparing the DAG for traversal.
+     */
+    public boolean isPreparer() {
+        return this.rootNode.isPreparer();
+    }
+
+    /**
+     * Merge this DAG with another DAG.
+     * <p>
+     * This will mark this DAG as a child DAG, the dependencies of nodes in this DAG will be merged
+     * with (copied to) the parent DAG
+     *
+     * @param parent the parent DAG
+     */
+    public void merge(DAGraph<T, U> parent) {
+        this.hasParent = true;
+        parent.rootNode.addDependency(this.rootNode.key());
+        for (Map.Entry<String, U> entry: graph.entrySet()) {
+            String key = entry.getKey();
+            if (!parent.graph.containsKey(key)) {
+                parent.graph.put(key, entry.getValue());
+            }
+        }
+    }
+
+    /**
+     * Prepares this DAG for traversal using getNext method, each call to getNext returns next node
+     * in the DAG with no dependencies.
+     */
+    public void prepare() {
+        if (isPreparer()) {
+            for (U node : graph.values()) {
+                // Prepare each node for traversal
+                node.initialize();
+                if (!this.isRootNode(node)) {
+                    // Mark other sub-DAGs as non-preparer
+                    node.setPreparer(false);
+                }
+            }
+            initializeDependentKeys();
+            initializeQueue();
+        }
+    }
+
+    /**
+     * Gets next node in the DAG which has no dependency or all of it's dependencies are resolved and
+     * ready to be consumed.
+     *
+     * @return next node or null if all the nodes have been explored or no node is available at this moment.
+     */
+    public U getNext() {
+        String nextItemKey = queue.poll();
+        if (nextItemKey == null) {
+            return null;
+        }
+        return graph.get(nextItemKey);
+    }
+
+    /**
+     * Gets the data stored in a graph node with a given key.
+     *
+     * @param key the key of the node
+     * @return the value stored in the node
+     */
+    public T getNodeData(String key) {
+       return graph.get(key).data();
+    }
+
+    /**
+     * Reports that a node is resolved hence other nodes depends on it can consume it.
+     *
+     * @param completed the node ready to be consumed
+     */
+    public void reportedCompleted(U completed) {
+        completed.setPreparer(true);
+        String dependency = completed.key();
+        for (String dependentKey : graph.get(dependency).dependentKeys()) {
+            DAGNode<T> dependent = graph.get(dependentKey);
+            dependent.lock().lock();
+            try {
+                dependent.reportResolved(dependency);
+                if (dependent.hasAllResolved()) {
+                    queue.add(dependent.key());
+                }
+            } finally {
+                dependent.lock().unlock();
+            }
+        }
+    }
+
+    /**
+     * Initializes dependents of all nodes.
+     * <p>
+     * The DAG will be explored in DFS order and all node's dependents will be identified,
+     * this prepares the DAG for traversal using getNext method, each call to getNext returns next node
+     * in the DAG with no dependencies.
+     */
+    private void initializeDependentKeys() {
+        visit(new Visitor<U>() {
+            @Override
+            public void visitNode(U node) {
+                if (node.dependencyKeys().isEmpty()) {
+                    return;
+                }
+
+                String dependentKey = node.key();
+                for (String dependencyKey : node.dependencyKeys()) {
+                    graph.get(dependencyKey)
+                            .addDependent(dependentKey);
+                }
+            }
+
+            @Override
+            public  void visitEdge(String fromKey, String toKey, EdgeType edgeType) {
+                if (edgeType == EdgeType.BACK) {
+                    throw new IllegalStateException("Detected circular dependency: " + findPath(fromKey, toKey));
+                }
+            }
+        });
+    }
+
+    /**
+     * Initializes the queue that tracks the next set of nodes with no dependencies or
+     * whose dependencies are resolved.
+     */
+    private void initializeQueue() {
+        this.queue.clear();
+        for (Map.Entry<String, U> entry: graph.entrySet()) {
+            if (!entry.getValue().hasDependencies()) {
+                this.queue.add(entry.getKey());
+            }
+        }
+        if (queue.isEmpty()) {
+            throw new RuntimeException("Found circular dependency");
+        }
+    }
+}