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.Arrays;
024import java.util.Collection;
025import java.util.HashSet;
026import java.util.Set;
027
028import cascading.flow.FlowElement;
029import cascading.flow.planner.Scope;
030import org.jgrapht.DirectedGraph;
031import org.jgrapht.graph.DirectedMaskSubgraph;
032import org.jgrapht.graph.MaskFunctor;
033
034import static cascading.flow.planner.graph.ElementGraphs.directed;
035import static cascading.util.Util.createIdentitySet;
036
037/**
038 *
039 */
040public class ElementMaskSubGraph extends BaseElementGraph implements ElementGraph
041  {
042  private ElementGraph elementGraph;
043  private FlowElementMaskFunctor mask;
044
045  private static class FlowElementMaskFunctor implements MaskFunctor<FlowElement, Scope>
046    {
047    Set<FlowElement> maskedElements = createIdentitySet();
048    Set<Scope> maskedScopes = new HashSet<>();
049
050    public FlowElementMaskFunctor( Collection<FlowElement> flowElements, Collection<Scope> scopes )
051      {
052      this( flowElements );
053
054      if( scopes != null )
055        maskedScopes.addAll( scopes );
056      }
057
058    public FlowElementMaskFunctor( Collection<FlowElement> flowElements )
059      {
060      if( flowElements != null )
061        maskedElements.addAll( flowElements );
062      }
063
064    @Override
065    public boolean isEdgeMasked( Scope scope )
066      {
067      return maskedScopes.contains( scope );
068      }
069
070    @Override
071    public boolean isVertexMasked( FlowElement flowElement )
072      {
073      return maskedElements.contains( flowElement );
074      }
075    }
076
077  public ElementMaskSubGraph( ElementGraph elementGraph, FlowElement... maskedFlowElements )
078    {
079    this( elementGraph, new FlowElementMaskFunctor( Arrays.asList( maskedFlowElements ) ) );
080    }
081
082  public ElementMaskSubGraph( ElementGraph elementGraph, Collection<FlowElement> maskedFlowElements )
083    {
084    this( elementGraph, new FlowElementMaskFunctor( maskedFlowElements ) );
085    }
086
087  public ElementMaskSubGraph( ElementGraph elementGraph, Collection<FlowElement> maskedFlowElements, Collection<Scope> maskedScopes )
088    {
089    this( elementGraph, new FlowElementMaskFunctor( maskedFlowElements, maskedScopes ) );
090    }
091
092  public ElementMaskSubGraph( ElementMaskSubGraph graph )
093    {
094    this( graph.elementGraph, graph.mask );
095    }
096
097  protected ElementMaskSubGraph( ElementGraph elementGraph, FlowElementMaskFunctor flowElementMaskFunctor )
098    {
099    this.graph = new DirectedMaskSubGraph( directed( elementGraph ), flowElementMaskFunctor );
100
101    this.elementGraph = elementGraph;
102    this.mask = flowElementMaskFunctor;
103    }
104
105  @Override
106  public ElementGraph copyElementGraph()
107    {
108    return new ElementMaskSubGraph( ElementMaskSubGraph.this );
109    }
110
111  private class DirectedMaskSubGraph extends DirectedMaskSubgraph<FlowElement, Scope>
112    {
113    public DirectedMaskSubGraph( DirectedGraph<FlowElement, Scope> base, MaskFunctor<FlowElement, Scope> mask )
114      {
115      super( base, mask );
116      }
117    }
118  }