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