how to get count then run iteration for single Java Stream object

Question

Considering a Java Stream object generated by method:

// get Stream object
Stream<String> dataCollection = myFunctionToGenerateStreamObj(...)

// get count for logging purpose
int total = dataCollection.count()

// iterate each object and do something
dataCollection.forEach( obj -> {
    anotherFunctionToProcessObj(obj);
});

Obviously above code will throw

IllegalStateException: stream has already been operated upon or closed

on the line for iteration. I understand there a solution to use Supplier class. I just wondering what is the best practice to code above logic in most common and efficient way.

Answer 1

The point is that the act of creating and iterating is potentially expensive.

Consider this simple one: Files.lines. It opens the file and reads it, line by line, exposing that as a Stream<String>.

You can, of course, wrap this in a supplier: () -> Files.lines(somePath) and pass that around. Anybody you gave that supplier to can invoke it, which results in the OS opening a file handle, and asking the disk to go spit out some bits.

But, the point is, the cost here is the stream process. The actual thing you do with the stream? Dwarfed, utterly, by the disk access (though with fast SSDs this matters less. If it helps, replace that with a file loaded over a network connection. That is metered).

Some streams aren't anything like that. if you have an ArrayList and you call .stream() on that, the 'processing' costs of the stream itself (making the stream object, and streaming through the list's elements), is an ignorable cost, it's just forwarding a memory pointer, through contiguous memory no less. It doesn't even cache miss much.

But Stream is an abstraction - the point is that it sucks to have to care. It's bad to have to document "if you pass a supplier that supplies a stream with high setup and processing costs, this code is quite slow and if its loaded across a metered network it'll really rack up your bill".

So, don't. Which means, the best thing to do is to stream only once.

Unfortunately there is no simple way to just make 2 Stream objects that are powered by a single stream through.

One way to do this count thing would be:

AtomicInteger counter = new AtomicInteger();
Stream<String> dataCollection = ....;
dataCollection.peek(x -> counter.incrementAndGet()).forEach(...);

It's annoying to have to do it like this. A key problem here is that lambdas don't like changing state, and methods can only return one thing.

Answer 2

If it is not an unbounded stream, what about map?:

List<Integer> integers = Arrays.asList(1, 2, 3, 4, 5, 6);
long count = integers.stream().map(x -> {
    System.out.println("x = " + x);
    return 1;
}).count();
System.out.println("count = " + count);

Answer 3

I think what you have can be represented as special case of reduction. More specifically, Stream.reduce(identity, accumulator, combiner).

int totalCount = dataCollection
        .reduce(0, (currentCount, obj) -> {
          anotherFunctionToProcessObj(obj);
          return currentCount + 1;
        }, Integer::sum);
System.out.println(totalCount);

1. Identity is the initial value for the counter - 0
2. Accumulator function will process object and return current count increased by 1
3. Combiner will sum the result of 2 accumulators

As stated in the stream-docs, the preferred way is reduction:

Many computations where one might be tempted to use side effects can be more safely and efficiently expressed without side-effects, such as using reduction instead of mutable accumulators.

As pointed out in comments, if the stream is infinite, you won't be able to reduce it(it's not possible to get the total count anyway). In this case external counter to keep track of the number of processed elements can be used:

AtomicInteger counter = new AtomicInteger(0);
dataCollection.forEach(obj -> {
    anotherFunctionToProcessObj(obj);
    counter.incrementAndGet();
});

Keep in mind, this is ok for this particular corner case only, as changing external state via side effects is bad practice.

As another option, you can just get the stream's iterator and use simple while loop.

Iterator<String> iterator = dataCollection.iterator();
int count = 0;
while (iterator.hasNext()) {
  count++;
  anotherFunctionToProcessObj(iterator.next());
}
System.out.println(count);

Answer 4

The stream you're pulling data from should not care about how large it is. After all, streams can be unbounded in size. This is also important in the context of what it is you're streaming; if you're streaming whole objects then it's a lot different than if you were streaming individual characters from a data source that was whole terabytes in size.

So...the only place you could realistically keep track of what you've seen is where the data is processed. In anotherFunctionToProcess, since you're just...consuming the data (doesn't look like you have to worry about threads too much), you could keep a tally inside of that class.

Because you're leveraging forEach, this already creates a side effect, and while you shouldn't look to create side effects when operating in streams, there's not a better way to do this without simply disregarding the requirement to count the elements.

This way, you only ever stream the data once and you keep track of what you process in the process.