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