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.operation.function;
022
023import java.beans.ConstructorProperties;
024import java.io.Serializable;
025import java.util.Arrays;
026
027import cascading.flow.FlowProcess;
028import cascading.management.annotation.Property;
029import cascading.management.annotation.Visibility;
030import cascading.operation.BaseOperation;
031import cascading.operation.Filter;
032import cascading.operation.FilterCall;
033import cascading.operation.Function;
034import cascading.operation.FunctionCall;
035import cascading.operation.OperationCall;
036import cascading.tuple.Fields;
037import cascading.tuple.Tuple;
038
039/**
040 * Class SetValue is a utility {@link Function} that allows for a Tuple value to be returned based on the outcome
041 * of a given {@link Filter} operation.
042 * <p/>
043 * There are only two possible values, either {@link Filter#isRemove(cascading.flow.FlowProcess, cascading.operation.FilterCall)}
044 * returns {@code true} or {@code false}.
045 * <p/>
046 * If {@code false} is returned, most commonly the {@link Filter} passed and the Tuple should be kept. SetValue will then return
047 * the first value in the given values array, by default {@code true}. If the Filter returns {@code true}, the second
048 * value in the values array will be returned, by default {@code false}.
049 * <p/>
050 */
051public class SetValue extends BaseOperation implements Function
052  {
053  /** Field filter */
054  private final Filter filter;
055  /** Field values */
056  private Tuple[] values = new Tuple[]{new Tuple( true ), new Tuple( false )};
057
058  /**
059   * Constructor SetValue creates a new SetValue instance.
060   *
061   * @param fieldDeclaration of type Fields
062   * @param filter           of type Filter
063   */
064  @ConstructorProperties({"fieldDeclaration", "filter"})
065  public SetValue( Fields fieldDeclaration, Filter filter )
066    {
067    super( fieldDeclaration );
068    this.filter = filter;
069
070    verify();
071    }
072
073  /**
074   * Constructor SetValue creates a new SetValue instance.
075   *
076   * @param fieldDeclaration of type Fields
077   * @param filter           of type Filter
078   * @param firstValue       of type Serializable
079   * @param secondValue      of type Serializable
080   */
081  @ConstructorProperties({"fieldDeclaration", "filter", "firstValue", "secondValue"})
082  public SetValue( Fields fieldDeclaration, Filter filter, Serializable firstValue, Serializable secondValue )
083    {
084    super( fieldDeclaration );
085    this.filter = filter;
086    this.values = new Tuple[]{new Tuple( firstValue ), new Tuple( secondValue )};
087
088    verify();
089    }
090
091  @Property(name = "firstValue", visibility = Visibility.PRIVATE)
092  public Serializable getFirstValue()
093    {
094    return (Serializable) values[ 0 ].getObject( 0 );
095    }
096
097  @Property(name = "secondValue", visibility = Visibility.PRIVATE)
098  public Serializable getSecondValue()
099    {
100    return (Serializable) values[ 1 ].getObject( 0 );
101    }
102
103  private void verify()
104    {
105    if( fieldDeclaration.size() != 1 )
106      throw new IllegalArgumentException( "fieldDeclaration may only declare one field, was " + fieldDeclaration.print() );
107
108    if( filter == null )
109      throw new IllegalArgumentException( "filter may not be null" );
110
111    if( values == null || values.length != 2 )
112      throw new IllegalArgumentException( "values argument must contain two values" );
113    }
114
115  @Override
116  public void prepare( FlowProcess flowProcess, OperationCall operationCall )
117    {
118    filter.prepare( flowProcess, operationCall );
119    }
120
121  @Override
122  public void operate( FlowProcess flowProcess, FunctionCall functionCall )
123    {
124    boolean isRemove = !filter.isRemove( flowProcess, (FilterCall) functionCall );
125
126    int pos = isRemove ? 0 : 1;
127
128    functionCall.getOutputCollector().add( values[ pos ] );
129    }
130
131  @Override
132  public void cleanup( FlowProcess flowProcess, OperationCall operationCall )
133    {
134    filter.cleanup( flowProcess, operationCall );
135    }
136
137  @Override
138  public boolean equals( Object object )
139    {
140    if( this == object )
141      return true;
142    if( !( object instanceof SetValue ) )
143      return false;
144    if( !super.equals( object ) )
145      return false;
146
147    SetValue setValue = (SetValue) object;
148
149    if( filter != null ? !filter.equals( setValue.filter ) : setValue.filter != null )
150      return false;
151    if( !Arrays.equals( values, setValue.values ) )
152      return false;
153
154    return true;
155    }
156
157  @Override
158  public int hashCode()
159    {
160    int result = super.hashCode();
161    result = 31 * result + ( filter != null ? filter.hashCode() : 0 );
162    result = 31 * result + ( values != null ? Arrays.hashCode( values ) : 0 );
163    return result;
164    }
165  }