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.expression;
022
023import cascading.flow.FlowElement;
024import cascading.flow.planner.PlannerContext;
025import cascading.flow.planner.graph.ElementGraph;
026
027/**
028 *
029 */
030public class TypeExpression<Type> extends ElementExpression
031  {
032  public enum Topo
033    {
034      Ignore,
035      Head,
036      Tail,
037      Linear,
038      LinearIn,
039      LinearOut,
040      Splice,
041      Split,
042      SpliceOnly,
043      SplitOnly,
044      SpliceSplit
045    }
046
047  boolean exact = false;
048  Class<? extends Type> type;
049  Topo topo = Topo.Ignore;
050
051  public TypeExpression( ElementCapture capture, boolean exact, Class<? extends Type> type, Topo topo )
052    {
053    super( capture );
054    this.exact = exact;
055    this.type = type;
056    this.topo = topo;
057    }
058
059  public TypeExpression( ElementCapture capture, boolean exact, Class<? extends Type> type )
060    {
061    super( capture );
062    this.exact = exact;
063    this.type = type;
064    }
065
066  public TypeExpression( ElementCapture capture, Class<? extends Type> type, Topo topo )
067    {
068    super( capture );
069    this.type = type;
070    this.topo = topo;
071    }
072
073  public TypeExpression( ElementCapture capture, Class<? extends Type> type )
074    {
075    super( capture );
076    this.type = type;
077    }
078
079  public TypeExpression( boolean exact, Class<? extends Type> type, Topo topo )
080    {
081    this.exact = exact;
082    this.type = type;
083    this.topo = topo;
084    }
085
086  public TypeExpression( boolean exact, Class<? extends Type> type )
087    {
088    this.exact = exact;
089    this.type = type;
090    }
091
092  public TypeExpression( Class<? extends Type> type, Topo topo )
093    {
094    this.type = type;
095    this.topo = topo;
096    }
097
098  public TypeExpression( Class<? extends Type> type )
099    {
100    this.type = type;
101    }
102
103  protected Class<? extends Type> getType( FlowElement flowElement )
104    {
105    return (Class<? extends Type>) flowElement.getClass();
106    }
107
108  @Override
109  public boolean applies( PlannerContext plannerContext, ElementGraph elementGraph, FlowElement flowElement )
110    {
111    boolean typeApplies = typeApplies( flowElement );
112
113    if( !typeApplies )
114      return false;
115
116    if( topo == Topo.Ignore )
117      return true;
118
119    // todo: make lazy
120    boolean isHead = elementGraph.inDegreeOf( flowElement ) == 0;
121    boolean isTail = elementGraph.outDegreeOf( flowElement ) == 0;
122    boolean isSplice = elementGraph.inDegreeOf( flowElement ) > 1;
123    boolean isSplit = elementGraph.outDegreeOf( flowElement ) > 1;
124
125    switch( topo )
126      {
127      case Head:
128        return isHead;
129      case Tail:
130        return isTail;
131      case Linear:
132        return !isSplice && !isSplit;
133      case LinearIn:
134        return !isSplice;
135      case LinearOut:
136        return !isSplit;
137      case Splice:
138        return isSplice;
139      case Split:
140        return isSplit;
141      case SpliceOnly:
142        return isSplice && !isSplit;
143      case SplitOnly:
144        return !isSplice && isSplit;
145      case SpliceSplit:
146        return isSplice && isSplit;
147      }
148
149    throw new IllegalStateException( "unknown switch, got: " + topo );
150    }
151
152  private boolean typeApplies( FlowElement flowElement )
153    {
154    Class<? extends Type> givenType = getType( flowElement );
155
156    if( givenType == null )
157      return false;
158
159    if( exact )
160      return givenType == type;
161
162    return type.isAssignableFrom( givenType );
163    }
164
165  @Override
166  public String toString()
167    {
168    final StringBuilder sb = new StringBuilder( getClass().getSimpleName() ).append( "{" );
169    sb.append( "exact=" ).append( exact );
170    sb.append( ", type=" ).append( type );
171    sb.append( ", topo=" ).append( topo );
172    sb.append( '}' );
173    return sb.toString();
174    }
175  }