001/*
002 * Copyright (c) 2007-2015 Concurrent, Inc. All Rights Reserved.
003 *
004 * Project and contact information: http://www.cascading.org/
005 *
006 * This file is part of the Cascading project.
007 *
008 * Licensed under the Apache License, Version 2.0 (the "License");
009 * you may not use this file except in compliance with the License.
010 * You may obtain a copy of the License at
011 *
012 *     http://www.apache.org/licenses/LICENSE-2.0
013 *
014 * Unless required by applicable law or agreed to in writing, software
015 * distributed under the License is distributed on an "AS IS" BASIS,
016 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
017 * See the License for the specific language governing permissions and
018 * limitations under the License.
019 */
020
021package cascading.flow.planner.graph;
022
023import java.util.Set;
024
025import cascading.flow.FlowElement;
026import cascading.flow.planner.Scope;
027import cascading.util.EnumMultiMap;
028
029import static cascading.flow.planner.graph.ElementGraphs.directed;
030import static cascading.flow.planner.graph.Extent.head;
031import static cascading.flow.planner.graph.Extent.tail;
032
033/**
034 *
035 */
036public class BoundedElementMultiGraph extends ElementMultiGraph
037  {
038  public BoundedElementMultiGraph( ElementGraph parentElementGraph, ElementGraph subElementGraph, EnumMultiMap annotations )
039    {
040    graph = new DirectedMultiGraph( directed( subElementGraph ) );
041
042    addParentAnnotations( parentElementGraph );
043
044    getAnnotations().addAll( annotations );
045
046    bindHeadAndTail( parentElementGraph, subElementGraph );
047    }
048
049  @Override
050  protected void addParentAnnotations( ElementGraph parentElementGraph )
051    {
052    if( !( parentElementGraph instanceof AnnotatedGraph ) || !( (AnnotatedGraph) parentElementGraph ).hasAnnotations() )
053      return;
054
055    Set<FlowElement> vertexSet = vertexSet();
056
057    EnumMultiMap parentAnnotations = ( (AnnotatedGraph) parentElementGraph ).getAnnotations();
058
059    Set<Enum> allKeys = parentAnnotations.getKeys();
060
061    for( Enum annotation : allKeys )
062      {
063      Set<FlowElement> flowElements = (Set<FlowElement>) parentAnnotations.getValues( annotation );
064
065      for( FlowElement flowElement : flowElements )
066        {
067        if( vertexSet.contains( flowElement ) )
068          getAnnotations().addAll( annotation, flowElements );
069        }
070      }
071    }
072
073  protected void bindHeadAndTail( ElementGraph parentElementGraph, ElementGraph subElementGraph )
074    {
075    Set<FlowElement> sources = ElementGraphs.findSources( subElementGraph, FlowElement.class );
076    Set<FlowElement> sinks = ElementGraphs.findSinks( subElementGraph, FlowElement.class );
077
078    addVertex( head );
079    addVertex( tail );
080
081    Set<FlowElement> parentElements = parentElementGraph.vertexSet();
082
083    for( FlowElement source : sources )
084      {
085      if( !parentElements.contains( source ) )
086        continue;
087
088      Set<Scope> scopes = parentElementGraph.incomingEdgesOf( source );
089
090      for( Scope scope : scopes )
091        addEdge( head, source, scope );
092      }
093
094    for( FlowElement sink : sinks )
095      {
096      if( !parentElements.contains( sink ) )
097        continue;
098
099      Set<Scope> scopes = parentElementGraph.outgoingEdgesOf( sink );
100
101      for( Scope scope : scopes )
102        addEdge( sink, tail, scope );
103      }
104    }
105  }