/*
 * Copyright (c) 2016-2017 Chris K Wensel <chris@wensel.net>. All Rights Reserved.
 * Copyright (c) 2007-2017 Xplenty, Inc. All Rights Reserved.
 *
 * Project and contact information: http://www.cascading.org/
 *
 * This file is part of the Cascading project.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package cascading.flow.planner.graph;

import java.io.Serializable;
import java.util.Collection;
import java.util.Collections;
import java.util.IdentityHashMap;
import java.util.List;
import java.util.Set;

import cascading.flow.FlowElement;
import cascading.flow.planner.Scope;
import cascading.util.Util;
import org.jgrapht.DirectedGraph;
import org.jgrapht.Graph;
import org.jgrapht.Graphs;
import org.jgrapht.graph.SimpleDirectedGraph;

/**
 *
 */
public abstract class BaseElementGraph implements ElementGraph, Serializable
  {
  public static final ElementGraph NULL = new BaseElementGraph( new SimpleDirectedGraph<>( Scope.class ) )
    {
    @Override
    public ElementGraph copyElementGraph()
      {
      return null;
      }
    };

  protected Graph<FlowElement, Scope> graph;

  public BaseElementGraph()
    {
    }

  public BaseElementGraph( DirectedGraph<FlowElement, Scope> graph )
    {
    this.graph = graph;
    }

  protected void copyFrom( ElementGraph elementGraph )
    {
    Graphs.addAllVertices( graph, elementGraph.vertexSet() );
    Graphs.addAllEdges( graph, ElementGraphs.directed( elementGraph ), elementGraph.edgeSet() );
    }

  public boolean containsEdge( FlowElement sourceVertex, FlowElement targetVertex )
    {
    return graph.containsEdge( sourceVertex, targetVertex );
    }

  public boolean removeAllEdges( Collection<? extends Scope> edges )
    {
    return graph.removeAllEdges( edges );
    }

  public Set<Scope> removeAllEdges( FlowElement sourceVertex, FlowElement targetVertex )
    {
    return graph.removeAllEdges( sourceVertex, targetVertex );
    }

  public boolean removeAllVertices( Collection<? extends FlowElement> vertices )
    {
    return graph.removeAllVertices( vertices );
    }

  public Set<Scope> getAllEdges( FlowElement sourceVertex, FlowElement targetVertex )
    {
    return graph.getAllEdges( sourceVertex, targetVertex );
    }

  public Scope getEdge( FlowElement sourceVertex, FlowElement targetVertex )
    {
    return graph.getEdge( sourceVertex, targetVertex );
    }

  public Scope addEdge( FlowElement sourceVertex, FlowElement targetVertex )
    {
//     prevent multiple edges from head or to tail
    if( !allowMultipleExtentEdges() && ( sourceVertex == Extent.head || targetVertex == Extent.tail ) && graph.containsEdge( sourceVertex, targetVertex ) )
      return graph.getEdge( sourceVertex, targetVertex );

    return graph.addEdge( sourceVertex, targetVertex );
    }

  public boolean addEdge( FlowElement sourceVertex, FlowElement targetVertex, Scope scope )
    {
    // prevent multiple edges from head or to tail
    if( !allowMultipleExtentEdges() && ( sourceVertex == Extent.head || targetVertex == Extent.tail ) && graph.containsEdge( sourceVertex, targetVertex ) )
      return true;

    return graph.addEdge( sourceVertex, targetVertex, scope );
    }

  protected boolean allowMultipleExtentEdges()
    {
    return true;
    }

  @Override
  public boolean addHeadVertex( FlowElement flowElement )
    {
    if( !graph.containsVertex( Extent.head ) )
      graph.addVertex( Extent.head );

    if( flowElement == Extent.head )
      return false;

    boolean result = true;

    if( !graph.containsVertex( flowElement ) )
      result = graph.addVertex( flowElement );

    return result && graph.addEdge( Extent.head, flowElement ) != null;
    }

  @Override
  public boolean addTailVertex( FlowElement flowElement )
    {
    if( !graph.containsVertex( Extent.tail ) )
      graph.addVertex( Extent.tail );

    if( flowElement == Extent.tail )
      return false;

    boolean result = true;

    if( !graph.containsVertex( flowElement ) )
      result = graph.addVertex( flowElement );

    return result && graph.addEdge( flowElement, Extent.tail ) != null;
    }

  public boolean addVertex( FlowElement flowElement )
    {
    return graph.addVertex( flowElement );
    }

  public FlowElement getEdgeSource( Scope scope )
    {
    return graph.getEdgeSource( scope );
    }

  public FlowElement getEdgeTarget( Scope scope )
    {
    return graph.getEdgeTarget( scope );
    }

  public boolean containsEdge( Scope scope )
    {
    return graph.containsEdge( scope );
    }

  public boolean containsVertex( FlowElement flowElement )
    {
    return graph.containsVertex( flowElement );
    }

  public Set<Scope> edgeSet()
    {
    return graph.edgeSet();
    }

  public Set<Scope> edgesOf( FlowElement vertex )
    {
    return graph.edgesOf( vertex );
    }

  public int inDegreeOf( FlowElement vertex )
    {
    return graph.inDegreeOf( vertex );
    }

  public Set<Scope> incomingEdgesOf( FlowElement vertex )
    {
    return graph.incomingEdgesOf( vertex );
    }

  public int outDegreeOf( FlowElement vertex )
    {
    return graph.outDegreeOf( vertex );
    }

  public Set<Scope> outgoingEdgesOf( FlowElement vertex )
    {
    return graph.outgoingEdgesOf( vertex );
    }

  public Scope removeEdge( FlowElement sourceVertex, FlowElement targetVertex )
    {
    return graph.removeEdge( sourceVertex, targetVertex );
    }

  public boolean removeEdge( Scope scope )
    {
    return graph.removeEdge( scope );
    }

  public boolean removeVertex( FlowElement flowElement )
    {
    return graph.removeVertex( flowElement );
    }

  public Set<FlowElement> vertexSet()
    {
    return graph.vertexSet();
    }

  @Override
  public Set<FlowElement> vertexSetCopy()
    {
    Set<FlowElement> result = Collections.newSetFromMap( new IdentityHashMap<FlowElement, Boolean>() );

    result.addAll( vertexSet() );

    return result;
    }

  @Override
  public List<FlowElement> predecessorListOf( FlowElement flowElement )
    {
    return Graphs.predecessorListOf( graph, flowElement );
    }

  @Override
  public List<FlowElement> successorListOf( FlowElement flowElement )
    {
    return Graphs.successorListOf( graph, flowElement );
    }

  @Override
  public ElementGraph bindExtents()
    {
    Set<FlowElement> vertexSet = vertexSetCopy();

    for( FlowElement flowElement : vertexSet )
      {
      if( inDegreeOf( flowElement ) == 0 )
        addHeadVertex( flowElement );

      if( outDegreeOf( flowElement ) == 0 )
        addTailVertex( flowElement );
      }

    // be sure to test underlying container
    if( !vertexSet().contains( Extent.head ) )
      throw new IllegalStateException( "did not bind head vertex to graph" );

    if( !vertexSet().contains( Extent.tail ) )
      throw new IllegalStateException( "did not bind tail vertex to graph" );

    return this;
    }

  @Override
  public void writeDOT( String filename )
    {
    boolean success = ElementGraphs.printElementGraph( filename, this, null );

    if( success )
      Util.writePDF( filename );
    }

  @Override
  public boolean equals( Object object )
    {
    return ElementGraphs.equals( this, (ElementGraph) object );
    }

  @Override
  public int hashCode()
    {
    int result = graph.hashCode();
    result = 31 * result; // parity with AnnotatedGraph types
    return result;
    }
  }