001/*
002 * Copyright (c) 2007-2017 Xplenty, 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.iso.transformer;
022
023import java.util.HashSet;
024import java.util.Set;
025
026import cascading.flow.FlowElement;
027import cascading.flow.planner.Scope;
028import cascading.flow.planner.graph.ElementGraph;
029import cascading.flow.planner.graph.ElementGraphs;
030import cascading.flow.planner.iso.expression.ElementCapture;
031import cascading.flow.planner.iso.expression.ExpressionGraph;
032import cascading.flow.planner.iso.finder.Match;
033
034/**
035 *
036 */
037public class InsertionGraphTransformer extends MutateGraphTransformer
038  {
039  public enum Insertion
040    {
041      Before,
042      After,
043      BeforeEachEdge,
044      AfterEachEdge
045    }
046
047  private Insertion insertion = Insertion.After;
048  private final String factoryName;
049  private ElementCapture capture = ElementCapture.Primary;
050
051  public InsertionGraphTransformer( ExpressionGraph expressionGraph, String factoryName )
052    {
053    this( expressionGraph, ElementCapture.Primary, factoryName );
054    }
055
056  public InsertionGraphTransformer( ExpressionGraph expressionGraph, String factoryName, Insertion insertion )
057    {
058    this( expressionGraph, ElementCapture.Primary, factoryName, insertion );
059    }
060
061  public InsertionGraphTransformer( ExpressionGraph expressionGraph, ElementCapture capture, String factoryName )
062    {
063    this( expressionGraph, capture, factoryName, Insertion.After );
064    }
065
066  public InsertionGraphTransformer( ExpressionGraph expressionGraph, ElementCapture capture, String factoryName, Insertion insertion )
067    {
068    super( expressionGraph );
069
070    this.insertion = insertion;
071
072    if( capture != null )
073      this.capture = capture;
074
075    this.factoryName = factoryName;
076
077    if( factoryName == null )
078      throw new IllegalArgumentException( "factoryName may not be null" );
079    }
080
081  public InsertionGraphTransformer( GraphTransformer graphTransformer, ExpressionGraph filter, String factoryName )
082    {
083    this( graphTransformer, filter, ElementCapture.Primary, factoryName );
084    }
085
086  public InsertionGraphTransformer( GraphTransformer graphTransformer, ExpressionGraph filter, ElementCapture capture, String factoryName )
087    {
088    this( graphTransformer, filter, capture, factoryName, Insertion.After );
089    }
090
091  public InsertionGraphTransformer( GraphTransformer graphTransformer, ExpressionGraph filter, ElementCapture capture, String factoryName, Insertion insertion )
092    {
093    super( graphTransformer, filter );
094
095    this.insertion = insertion;
096
097    if( capture != null )
098      this.capture = capture;
099
100    this.factoryName = factoryName;
101
102    if( factoryName == null )
103      throw new IllegalArgumentException( "factoryName may not be null" );
104    }
105
106  @Override
107  protected boolean transformGraphInPlaceUsing( Transformed<ElementGraph> transformed, ElementGraph graph, Match match )
108    {
109    Set<FlowElement> insertions = match.getCapturedElements( capture );
110
111    if( insertions.isEmpty() )
112      return false;
113
114    ElementFactory elementFactory = transformed.getPlannerContext().getElementFactoryFor( factoryName );
115
116    for( FlowElement flowElement : insertions )
117      {
118      switch( insertion )
119        {
120        case Before:
121
122          ElementGraphs.insertFlowElementBefore( graph, flowElement, elementFactory.create( graph, flowElement ) );
123          break;
124
125        case After:
126
127          ElementGraphs.insertFlowElementAfter( graph, flowElement, elementFactory.create( graph, flowElement ) );
128          break;
129
130        case BeforeEachEdge:
131
132          Set<Scope> incomingEdges = new HashSet<>( graph.incomingEdgesOf( flowElement ) );
133
134          for( Scope incomingEdge : incomingEdges )
135            {
136            FlowElement predecessor = graph.getEdgeSource( incomingEdge );
137            ElementGraphs.insertFlowElementBetweenEdge( graph, incomingEdge, elementFactory.create( graph, predecessor ) );
138            }
139
140          break;
141
142        case AfterEachEdge:
143
144          Set<Scope> outgoingEdges = new HashSet<>( graph.outgoingEdgesOf( flowElement ) );
145
146          for( Scope outgoingEdge : outgoingEdges )
147            {
148            FlowElement successor = graph.getEdgeTarget( outgoingEdge );
149            ElementGraphs.insertFlowElementBetweenEdge( graph, outgoingEdge, elementFactory.create( graph, successor ) );
150            }
151
152          break;
153        }
154      }
155
156    return true;
157    }
158  }