1 /* 2 * Copyright (c) 2012, 2013, Oracle and/or its affiliates. All rights reserved. 3 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. 4 * 5 * This code is free software; you can redistribute it and/or modify it 6 * under the terms of the GNU General Public License version 2 only, as 7 * published by the Free Software Foundation. Oracle designates this 8 * particular file as subject to the "Classpath" exception as provided 9 * by Oracle in the LICENSE file that accompanied this code. 10 * 11 * This code is distributed in the hope that it will be useful, but WITHOUT 12 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or 13 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License 14 * version 2 for more details (a copy is included in the LICENSE file that 15 * accompanied this code). 16 * 17 * You should have received a copy of the GNU General Public License version 18 * 2 along with this work; if not, write to the Free Software Foundation, 19 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. 20 * 21 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA 22 * or visit www.oracle.com if you need additional information or have any 23 * questions. 24 */ 25 package java.util.stream; 26 27 import java.util.Objects; 28 import java.util.function.Consumer; 29 import java.util.function.DoubleConsumer; 30 import java.util.function.IntConsumer; 31 import java.util.function.LongConsumer; 32 33 /** 34 * An extension of {@link Consumer} used to conduct values through the stages of 35 * a stream pipeline, with additional methods to manage size information, 36 * control flow, etc. Before calling the {@code accept()} method on a 37 * {@code Sink} for the first time, you must first call the {@code begin()} 38 * method to inform it that data is coming (optionally informing the sink how 39 * much data is coming), and after all data has been sent, you must call the 40 * {@code end()} method. After calling {@code end()}, you should not call 41 * {@code accept()} without again calling {@code begin()}. {@code Sink} also 42 * offers a mechanism by which the sink can cooperatively signal that it does 43 * not wish to receive any more data (the {@code cancellationRequested()} 44 * method), which a source can poll before sending more data to the 45 * {@code Sink}. 46 * 47 * <p>A sink may be in one of two states: an initial state and an active state. 48 * It starts out in the initial state; the {@code begin()} method transitions 49 * it to the active state, and the {@code end()} method transitions it back into 50 * the initial state, where it can be re-used. Data-accepting methods (such as 51 * {@code accept()} are only valid in the active state. 52 * 53 * @apiNote 54 * A stream pipeline consists of a source, zero or more intermediate stages 55 * (such as filtering or mapping), and a terminal stage, such as reduction or 56 * for-each. For concreteness, consider the pipeline: 57 * 58 * <pre>{@code 59 * int longestStringLengthStartingWithA 60 * = strings.stream() 61 * .filter(s -> s.startsWith("A")) 62 * .mapToInt(String::length) 63 * .max(); 64 * }</pre> 65 * 66 * <p>Here, we have three stages, filtering, mapping, and reducing. The 67 * filtering stage consumes strings and emits a subset of those strings; the 68 * mapping stage consumes strings and emits ints; the reduction stage consumes 69 * those ints and computes the maximal value. 70 * 71 * <p>A {@code Sink} instance is used to represent each stage of this pipeline, 72 * whether the stage accepts objects, ints, longs, or doubles. Sink has entry 73 * points for {@code accept(Object)}, {@code accept(int)}, etc, so that we do 74 * not need a specialized interface for each primitive specialization. (It 75 * might be called a "kitchen sink" for this omnivorous tendency.) The entry 76 * point to the pipeline is the {@code Sink} for the filtering stage, which 77 * sends some elements "downstream" -- into the {@code Sink} for the mapping 78 * stage, which in turn sends integral values downstream into the {@code Sink} 79 * for the reduction stage. The {@code Sink} implementations associated with a 80 * given stage is expected to know the data type for the next stage, and call 81 * the correct {@code accept} method on its downstream {@code Sink}. Similarly, 82 * each stage must implement the correct {@code accept} method corresponding to 83 * the data type it accepts. 84 * 85 * <p>The specialized subtypes such as {@link Sink.OfInt} override 86 * {@code accept(Object)} to call the appropriate primitive specialization of 87 * {@code accept}, implement the appropriate primitive specialization of 88 * {@code Consumer}, and re-abstract the appropriate primitive specialization of 89 * {@code accept}. 90 * 91 * <p>The chaining subtypes such as {@link ChainedInt} not only implement 92 * {@code Sink.OfInt}, but also maintain a {@code downstream} field which 93 * represents the downstream {@code Sink}, and implement the methods 94 * {@code begin()}, {@code end()}, and {@code cancellationRequested()} to 95 * delegate to the downstream {@code Sink}. Most implementations of 96 * intermediate operations will use these chaining wrappers. For example, the 97 * mapping stage in the above example would look like: 98 * 99 * <pre>{@code 100 * IntSink is = new Sink.ChainedReference<U>(sink) { 101 * public void accept(U u) { 102 * downstream.accept(mapper.applyAsInt(u)); 103 * } 104 * }; 105 * }</pre> 106 * 107 * <p>Here, we implement {@code Sink.ChainedReference<U>}, meaning that we expect 108 * to receive elements of type {@code U} as input, and pass the downstream sink 109 * to the constructor. Because the next stage expects to receive integers, we 110 * must call the {@code accept(int)} method when emitting values to the downstream. 111 * The {@code accept()} method applies the mapping function from {@code U} to 112 * {@code int} and passes the resulting value to the downstream {@code Sink}. 113 * 114 * @param <T> type of elements for value streams 115 * @since 1.8 116 * @hide Visible for CTS testing only (OpenJDK8 tests). 117 */ 118 // Android-changed: Made public for CTS tests only. 119 public interface Sink<T> extends Consumer<T> { 120 /** 121 * Resets the sink state to receive a fresh data set. This must be called 122 * before sending any data to the sink. After calling {@link #end()}, 123 * you may call this method to reset the sink for another calculation. 124 * @param size The exact size of the data to be pushed downstream, if 125 * known or {@code -1} if unknown or infinite. 126 * 127 * <p>Prior to this call, the sink must be in the initial state, and after 128 * this call it is in the active state. 129 */ begin(long size)130 default void begin(long size) {} 131 132 /** 133 * Indicates that all elements have been pushed. If the {@code Sink} is 134 * stateful, it should send any stored state downstream at this time, and 135 * should clear any accumulated state (and associated resources). 136 * 137 * <p>Prior to this call, the sink must be in the active state, and after 138 * this call it is returned to the initial state. 139 */ end()140 default void end() {} 141 142 /** 143 * Indicates that this {@code Sink} does not wish to receive any more data. 144 * 145 * @implSpec The default implementation always returns false. 146 * 147 * @return true if cancellation is requested 148 */ cancellationRequested()149 default boolean cancellationRequested() { 150 return false; 151 } 152 153 /** 154 * Accepts an int value. 155 * 156 * @implSpec The default implementation throws IllegalStateException. 157 * 158 * @throws IllegalStateException if this sink does not accept int values 159 */ accept(int value)160 default void accept(int value) { 161 throw new IllegalStateException("called wrong accept method"); 162 } 163 164 /** 165 * Accepts a long value. 166 * 167 * @implSpec The default implementation throws IllegalStateException. 168 * 169 * @throws IllegalStateException if this sink does not accept long values 170 */ accept(long value)171 default void accept(long value) { 172 throw new IllegalStateException("called wrong accept method"); 173 } 174 175 /** 176 * Accepts a double value. 177 * 178 * @implSpec The default implementation throws IllegalStateException. 179 * 180 * @throws IllegalStateException if this sink does not accept double values 181 */ accept(double value)182 default void accept(double value) { 183 throw new IllegalStateException("called wrong accept method"); 184 } 185 186 /** 187 * {@code Sink} that implements {@code Sink<Integer>}, re-abstracts 188 * {@code accept(int)}, and wires {@code accept(Integer)} to bridge to 189 * {@code accept(int)}. 190 */ 191 interface OfInt extends Sink<Integer>, IntConsumer { 192 @Override accept(int value)193 void accept(int value); 194 195 @Override accept(Integer i)196 default void accept(Integer i) { 197 if (Tripwire.ENABLED) 198 Tripwire.trip(getClass(), "{0} calling Sink.OfInt.accept(Integer)"); 199 accept(i.intValue()); 200 } 201 } 202 203 /** 204 * {@code Sink} that implements {@code Sink<Long>}, re-abstracts 205 * {@code accept(long)}, and wires {@code accept(Long)} to bridge to 206 * {@code accept(long)}. 207 */ 208 interface OfLong extends Sink<Long>, LongConsumer { 209 @Override accept(long value)210 void accept(long value); 211 212 @Override accept(Long i)213 default void accept(Long i) { 214 if (Tripwire.ENABLED) 215 Tripwire.trip(getClass(), "{0} calling Sink.OfLong.accept(Long)"); 216 accept(i.longValue()); 217 } 218 } 219 220 /** 221 * {@code Sink} that implements {@code Sink<Double>}, re-abstracts 222 * {@code accept(double)}, and wires {@code accept(Double)} to bridge to 223 * {@code accept(double)}. 224 */ 225 interface OfDouble extends Sink<Double>, DoubleConsumer { 226 @Override accept(double value)227 void accept(double value); 228 229 @Override accept(Double i)230 default void accept(Double i) { 231 if (Tripwire.ENABLED) 232 Tripwire.trip(getClass(), "{0} calling Sink.OfDouble.accept(Double)"); 233 accept(i.doubleValue()); 234 } 235 } 236 237 /** 238 * Abstract {@code Sink} implementation for creating chains of 239 * sinks. The {@code begin}, {@code end}, and 240 * {@code cancellationRequested} methods are wired to chain to the 241 * downstream {@code Sink}. This implementation takes a downstream 242 * {@code Sink} of unknown input shape and produces a {@code Sink<T>}. The 243 * implementation of the {@code accept()} method must call the correct 244 * {@code accept()} method on the downstream {@code Sink}. 245 */ 246 static abstract class ChainedReference<T, E_OUT> implements Sink<T> { 247 protected final Sink<? super E_OUT> downstream; 248 ChainedReference(Sink<? super E_OUT> downstream)249 public ChainedReference(Sink<? super E_OUT> downstream) { 250 this.downstream = Objects.requireNonNull(downstream); 251 } 252 253 @Override begin(long size)254 public void begin(long size) { 255 downstream.begin(size); 256 } 257 258 @Override end()259 public void end() { 260 downstream.end(); 261 } 262 263 @Override cancellationRequested()264 public boolean cancellationRequested() { 265 return downstream.cancellationRequested(); 266 } 267 } 268 269 /** 270 * Abstract {@code Sink} implementation designed for creating chains of 271 * sinks. The {@code begin}, {@code end}, and 272 * {@code cancellationRequested} methods are wired to chain to the 273 * downstream {@code Sink}. This implementation takes a downstream 274 * {@code Sink} of unknown input shape and produces a {@code Sink.OfInt}. 275 * The implementation of the {@code accept()} method must call the correct 276 * {@code accept()} method on the downstream {@code Sink}. 277 */ 278 static abstract class ChainedInt<E_OUT> implements Sink.OfInt { 279 protected final Sink<? super E_OUT> downstream; 280 ChainedInt(Sink<? super E_OUT> downstream)281 public ChainedInt(Sink<? super E_OUT> downstream) { 282 this.downstream = Objects.requireNonNull(downstream); 283 } 284 285 @Override begin(long size)286 public void begin(long size) { 287 downstream.begin(size); 288 } 289 290 @Override end()291 public void end() { 292 downstream.end(); 293 } 294 295 @Override cancellationRequested()296 public boolean cancellationRequested() { 297 return downstream.cancellationRequested(); 298 } 299 } 300 301 /** 302 * Abstract {@code Sink} implementation designed for creating chains of 303 * sinks. The {@code begin}, {@code end}, and 304 * {@code cancellationRequested} methods are wired to chain to the 305 * downstream {@code Sink}. This implementation takes a downstream 306 * {@code Sink} of unknown input shape and produces a {@code Sink.OfLong}. 307 * The implementation of the {@code accept()} method must call the correct 308 * {@code accept()} method on the downstream {@code Sink}. 309 */ 310 static abstract class ChainedLong<E_OUT> implements Sink.OfLong { 311 protected final Sink<? super E_OUT> downstream; 312 ChainedLong(Sink<? super E_OUT> downstream)313 public ChainedLong(Sink<? super E_OUT> downstream) { 314 this.downstream = Objects.requireNonNull(downstream); 315 } 316 317 @Override begin(long size)318 public void begin(long size) { 319 downstream.begin(size); 320 } 321 322 @Override end()323 public void end() { 324 downstream.end(); 325 } 326 327 @Override cancellationRequested()328 public boolean cancellationRequested() { 329 return downstream.cancellationRequested(); 330 } 331 } 332 333 /** 334 * Abstract {@code Sink} implementation designed for creating chains of 335 * sinks. The {@code begin}, {@code end}, and 336 * {@code cancellationRequested} methods are wired to chain to the 337 * downstream {@code Sink}. This implementation takes a downstream 338 * {@code Sink} of unknown input shape and produces a {@code Sink.OfDouble}. 339 * The implementation of the {@code accept()} method must call the correct 340 * {@code accept()} method on the downstream {@code Sink}. 341 */ 342 static abstract class ChainedDouble<E_OUT> implements Sink.OfDouble { 343 protected final Sink<? super E_OUT> downstream; 344 ChainedDouble(Sink<? super E_OUT> downstream)345 public ChainedDouble(Sink<? super E_OUT> downstream) { 346 this.downstream = Objects.requireNonNull(downstream); 347 } 348 349 @Override begin(long size)350 public void begin(long size) { 351 downstream.begin(size); 352 } 353 354 @Override end()355 public void end() { 356 downstream.end(); 357 } 358 359 @Override cancellationRequested()360 public boolean cancellationRequested() { 361 return downstream.cancellationRequested(); 362 } 363 } 364 } 365