001/*
002 * Copyright (c) 2016-2017 Chris K Wensel <chris@wensel.net>. All Rights Reserved.
003 * Copyright (c) 2007-2017 Xplenty, Inc. All Rights Reserved.
004 *
005 * Project and contact information: http://www.cascading.org/
006 *
007 * This file is part of the Cascading project.
008 *
009 * Licensed under the Apache License, Version 2.0 (the "License");
010 * you may not use this file except in compliance with the License.
011 * You may obtain a copy of the License at
012 *
013 *     http://www.apache.org/licenses/LICENSE-2.0
014 *
015 * Unless required by applicable law or agreed to in writing, software
016 * distributed under the License is distributed on an "AS IS" BASIS,
017 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
018 * See the License for the specific language governing permissions and
019 * limitations under the License.
020 */
021
022package cascading.operation.function;
023
024import java.beans.ConstructorProperties;
025import java.io.Serializable;
026import java.util.Arrays;
027
028import cascading.flow.FlowProcess;
029import cascading.management.annotation.Property;
030import cascading.management.annotation.Visibility;
031import cascading.operation.BaseOperation;
032import cascading.operation.Filter;
033import cascading.operation.FilterCall;
034import cascading.operation.Function;
035import cascading.operation.FunctionCall;
036import cascading.operation.OperationCall;
037import cascading.tuple.Fields;
038import cascading.tuple.Tuple;
039
040/**
041 * Class SetValue is a utility {@link Function} that allows for a Tuple value to be returned based on the outcome
042 * of a given {@link Filter} operation.
043 * <p>
044 * There are only two possible values, either {@link Filter#isRemove(cascading.flow.FlowProcess, cascading.operation.FilterCall)}
045 * returns {@code true} or {@code false}.
046 * <p>
047 * If {@code false} is returned, most commonly the {@link Filter} passed and the Tuple should be kept. SetValue will then return
048 * the first value in the given values array, by default {@code true}. If the Filter returns {@code true}, the second
049 * value in the values array will be returned, by default {@code false}.
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  }