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