001/* 002 * Copyright (c) 2007-2016 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.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 }