mjhjmtu

I have the following code:

interface Device 
 // ...
 boolean isDisconnected();
 void reconnect();


interface Gateway 
 // ...
 List<Device> getDevices();


...

for (Gateway gateway : gateways) 
 for(Device device : gateway.getDevices())
 if(device.isDisconnected())
 device.reconnect();
 
 

I want to refactor the code using Stream API. My first attempt was like the following:

gateways
 .stream()
 .forEach(
 gateway -> 
 gateway
 .getDevices()
 .parallelStream()
 .filter(device -> device.isDisconnected())
 .forEach(device -> device.reconnect())
 ;
 
 )
;

I didn't like it so after some modifications I ended up with this code:

gateways
 .parallelStream()
 .map(gateway -> gateway.getDevices().parallelStream())
 .map(stream -> stream.filter(device -> device.isDisconnected()))
 .forEach(stream -> stream.forEach(device -> device.reconnect()))
;

My question is whether there is a way to avoid nested forEach.

You should flatten the stream of streams using flatMap instead of map:

gateways
 .parallelStream()
 .flatMap(gateway -> gateway.getDevices().parallelStream())
 .filter(device -> device.isDisconnected())
 .forEach(device -> device.reconnect());

I would improve it further by using method references instead of lambda expressions:

gateways
 .parallelStream()
 .map(Gateway::getDevices)
 .flatMap(List::stream)
 .filter(Device::isDisconnected)
 .forEach(Device::reconnect);

Don't refactor your code into using Streams. You gain no benefits and gain no advantages over doing it like this, since the code is now less readable and less idiomatic for future maintainers.

By not using streams, you avoid nested forEach statements.

Remember: streams are meant to be side-effect free for safer parallelization. forEach by definition introduces side-effects. You lose the benefit of streams and lose readability at the same time, making this less desirable to do at all.

I would try this with a sequential stream before using a parallel one:

gateways
 .stream()
 .flatMap(gateway -> gateway.getDevices().stream())
 .filter(device -> device.isDisconnected())
 .forEach(device -> device.reconnect())
;

The idea is to create a stream via gateways.stream() then flatten the sequences returned from gateway.getDevices() via flatMap.

Then we apply a filter operation which acts like the if statement in your code and finally, a forEach terminal operation enabling us to invoke reconnect on each and every device passing the filter operation.

see Should I always use a parallel stream when possible?