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.management.annotation.Property;
029 import cascading.management.annotation.PropertyDescription;
030 import cascading.management.annotation.Visibility;
031 import cascading.operation.BaseOperation;
032 import cascading.operation.Filter;
033 import cascading.operation.FilterCall;
034 import cascading.operation.Function;
035 import cascading.operation.FunctionCall;
036 import cascading.operation.OperationCall;
037 import cascading.tuple.Fields;
038 import 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 * <p/>
051 */
052 public class SetValue extends BaseOperation implements Function
053 {
054 /** Field filter */
055 private final Filter filter;
056 /** Field values */
057 private Tuple[] values = new Tuple[]{new Tuple( true ), new Tuple( false )};
058
059 /**
060 * Constructor SetValue creates a new SetValue instance.
061 *
062 * @param fieldDeclaration of type Fields
063 * @param filter of type Filter
064 */
065 @ConstructorProperties({"fieldDeclaration", "filter"})
066 public SetValue( Fields fieldDeclaration, Filter filter )
067 {
068 super( fieldDeclaration );
069 this.filter = filter;
070
071 verify();
072 }
073
074 /**
075 * Constructor SetValue creates a new SetValue instance.
076 *
077 * @param fieldDeclaration of type Fields
078 * @param filter of type Filter
079 * @param firstValue of type Serializable
080 * @param secondValue of type Serializable
081 */
082 @ConstructorProperties({"fieldDeclaration", "filter", "firstValue", "secondValue"})
083 public SetValue( Fields fieldDeclaration, Filter filter, Serializable firstValue, Serializable secondValue )
084 {
085 super( fieldDeclaration );
086 this.filter = filter;
087 this.values = new Tuple[]{new Tuple( firstValue ), new Tuple( secondValue )};
088
089 verify();
090 }
091
092 @Property(name = "firstValue", visibility = Visibility.PRIVATE)
093 public Serializable getFirstValue()
094 {
095 return (Serializable) values[ 0 ].getObject( 0 );
096 }
097
098 @Property(name = "secondValue", visibility = Visibility.PRIVATE)
099 public Serializable getSecondValue()
100 {
101 return (Serializable) values[ 1 ].getObject( 0 );
102 }
103
104 private void verify()
105 {
106 if( fieldDeclaration.size() != 1 )
107 throw new IllegalArgumentException( "fieldDeclaration may only declare one field, was " + fieldDeclaration.print() );
108
109 if( filter == null )
110 throw new IllegalArgumentException( "filter may not be null" );
111
112 if( values == null || values.length != 2 )
113 throw new IllegalArgumentException( "values argument must contain two values" );
114 }
115
116 @Override
117 public void prepare( FlowProcess flowProcess, OperationCall operationCall )
118 {
119 filter.prepare( flowProcess, operationCall );
120 }
121
122 @Override
123 public void operate( FlowProcess flowProcess, FunctionCall functionCall )
124 {
125 boolean isRemove = !filter.isRemove( flowProcess, (FilterCall) functionCall );
126
127 int pos = isRemove ? 0 : 1;
128
129 functionCall.getOutputCollector().add( values[ pos ] );
130 }
131
132 @Override
133 public void cleanup( FlowProcess flowProcess, OperationCall operationCall )
134 {
135 filter.cleanup( flowProcess, operationCall );
136 }
137
138 @Override
139 public boolean equals( Object object )
140 {
141 if( this == object )
142 return true;
143 if( !( object instanceof SetValue ) )
144 return false;
145 if( !super.equals( object ) )
146 return false;
147
148 SetValue setValue = (SetValue) object;
149
150 if( filter != null ? !filter.equals( setValue.filter ) : setValue.filter != null )
151 return false;
152 if( !Arrays.equals( values, setValue.values ) )
153 return false;
154
155 return true;
156 }
157
158 @Override
159 public int hashCode()
160 {
161 int result = super.hashCode();
162 result = 31 * result + ( filter != null ? filter.hashCode() : 0 );
163 result = 31 * result + ( values != null ? Arrays.hashCode( values ) : 0 );
164 return result;
165 }
166 }