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.subgraph.iterator;
022
023import java.util.ArrayList;
024import java.util.Collection;
025import java.util.HashSet;
026import java.util.List;
027import java.util.NoSuchElementException;
028import java.util.Set;
029
030import cascading.flow.FlowElement;
031import cascading.flow.planner.PlannerContext;
032import cascading.flow.planner.graph.ElementGraph;
033import cascading.flow.planner.iso.ElementAnnotation;
034import cascading.flow.planner.iso.expression.ElementCapture;
035import cascading.flow.planner.iso.expression.ExpressionGraph;
036import cascading.flow.planner.iso.finder.GraphFinder;
037import cascading.flow.planner.iso.finder.Match;
038import cascading.flow.planner.iso.subgraph.SubGraphIterator;
039import cascading.flow.planner.iso.transformer.ContractedTransformer;
040import cascading.flow.planner.iso.transformer.Transformed;
041import cascading.util.EnumMultiMap;
042
043import static cascading.flow.planner.graph.ElementGraphs.asSubGraph;
044import static cascading.util.Util.createIdentitySet;
045
046/**
047 *
048 */
049public class ExpressionSubGraphIterator implements SubGraphIterator
050  {
051  private final PlannerContext plannerContext;
052  private final ElementGraph elementGraph;
053
054  private ContractedTransformer contractedTransformer;
055  private GraphFinder graphFinder;
056
057  private Set<FlowElement> elementExcludes = createIdentitySet();
058  private ElementGraph contractedGraph;
059  private Transformed<ElementGraph> contractedTransformed;
060
061  private boolean firstOnly = false; // false will continue to accumulate around the primary
062  private Match match;
063
064  private List<Match> matches = new ArrayList<>();
065
066  int count = 0;
067
068  public ExpressionSubGraphIterator( ExpressionGraph matchExpression, ElementGraph elementGraph )
069    {
070    this( new PlannerContext(), matchExpression, elementGraph );
071    }
072
073  public ExpressionSubGraphIterator( PlannerContext plannerContext, ExpressionGraph matchExpression, ElementGraph elementGraph )
074    {
075    this( plannerContext, null, matchExpression, elementGraph );
076    }
077
078  public ExpressionSubGraphIterator( PlannerContext plannerContext, ExpressionGraph contractionExpression, ExpressionGraph matchExpression, ElementGraph elementGraph )
079    {
080    this( plannerContext, contractionExpression, matchExpression, false, elementGraph );
081    }
082
083  public ExpressionSubGraphIterator( PlannerContext plannerContext, ExpressionGraph contractionExpression, ExpressionGraph matchExpression, ElementGraph elementGraph, Collection<FlowElement> elementExcludes )
084    {
085    this( plannerContext, contractionExpression, matchExpression, false, elementGraph, elementExcludes );
086    }
087
088  public ExpressionSubGraphIterator( PlannerContext plannerContext, ExpressionGraph contractionExpression, ExpressionGraph matchExpression, boolean firstOnly, ElementGraph elementGraph )
089    {
090    this( plannerContext, contractionExpression, matchExpression, firstOnly, elementGraph, null );
091    }
092
093  public ExpressionSubGraphIterator( PlannerContext plannerContext, ExpressionGraph contractionExpression, ExpressionGraph matchExpression, boolean firstOnly, ElementGraph elementGraph, Collection<FlowElement> elementExcludes )
094    {
095    this.plannerContext = plannerContext;
096    this.firstOnly = firstOnly;
097    this.elementGraph = elementGraph;
098
099    if( elementExcludes != null )
100      this.elementExcludes.addAll( elementExcludes );
101
102    if( contractionExpression != null )
103      contractedTransformer = new ContractedTransformer( contractionExpression );
104    else
105      contractedGraph = elementGraph;
106
107    graphFinder = new GraphFinder( matchExpression );
108    }
109
110  @Override
111  public ElementGraph getElementGraph()
112    {
113    return elementGraph;
114    }
115
116  public List<Match> getMatches()
117    {
118    return matches;
119    }
120
121  public ElementGraph getContractedGraph()
122    {
123    if( contractedGraph == null )
124      {
125      contractedTransformed = contractedTransformer.transform( plannerContext, elementGraph );
126      contractedGraph = contractedTransformed.getEndGraph();
127      }
128
129    return contractedGraph;
130    }
131
132  public Match getLastMatch()
133    {
134    if( matches.isEmpty() )
135      return null;
136
137    return matches.get( count - 1 );
138    }
139
140  @Override
141  public EnumMultiMap getAnnotationMap( ElementAnnotation[] annotations )
142    {
143    EnumMultiMap annotationsMap = new EnumMultiMap();
144
145    if( annotations.length == 0 )
146      return annotationsMap;
147
148    Match match = getLastMatch();
149
150    for( ElementAnnotation annotation : annotations )
151      annotationsMap.addAll( annotation.getAnnotation(), match.getCapturedElements( annotation.getCapture() ) );
152
153    return annotationsMap;
154    }
155
156  @Override
157  public boolean hasNext()
158    {
159    if( match == null )
160      {
161      match = graphFinder.findMatchesOnPrimary( plannerContext, getContractedGraph(), firstOnly, elementExcludes );
162
163      if( match.foundMatch() )
164        {
165        matches.add( match );
166        elementExcludes.addAll( match.getCapturedElements( ElementCapture.Primary ) ); // idempotent
167        count++;
168        }
169      }
170
171    return match.foundMatch();
172    }
173
174  @Override
175  public ElementGraph next()
176    {
177    try
178      {
179      if( !hasNext() )
180        throw new NoSuchElementException();
181
182      ElementGraph contractedMatchedGraph = match.getMatchedGraph();
183
184      Set<FlowElement> excludes = new HashSet<>( getContractedGraph().vertexSet() );
185
186      excludes.removeAll( contractedMatchedGraph.vertexSet() );
187
188      return asSubGraph( elementGraph, contractedMatchedGraph, excludes );
189      }
190    finally
191      {
192      match = null;
193      }
194    }
195
196  @Override
197  public void remove()
198    {
199    throw new UnsupportedOperationException();
200    }
201  }