Java 8 streams conditional processing

user_1008666 0
Viewed 84370

I'm interested in separating a stream into two or more substreams, and processing the elements in different ways. For example, a (large) text file might contain lines of type A and lines of type B, in which case I'd like to do something like:

File.lines(path)
.filter(line -> isTypeA(line))
.forEachTrue(line -> processTypeA(line))
.forEachFalse(line -> processTypeB(line))

The previous is my attempt at abstracting the situation. In reality I have a very large text file where each line is testing against a regex; if the line passes, then it is processed, whereas if it is rejected, then I want to update a counter. This further processing on rejected strings is why I don't simply use filter.

Is there any reasonable way to do this with streams, or will I have to fallback to loops? (I would like this to run in parallel as well, so streams are my first choice).

java-stream java-8 java
8 Answers

Here's an approach (which ignores the cautions about forcing conditional processing into a stream) that wraps a predicate and consumer into a single predicate-with-side-effect:

public static class StreamProc {

    public static <T> Predicate<T> process( Predicate<T> condition, Consumer<T> operation ) {
        Predicate<T> p = t -> { operation.accept(t); return false; };
        return (t) -> condition.test(t) ? p.test(t) : true;
    }

}

Then filter the stream:

someStream
    .filter( StreamProc.process( cond1, op1 ) )
    .filter( StreamProc.process( cond2, op2 ) )
    ...
    .collect( ... )

Elements remaining in the stream have not yet been processed.

For example, a typical filesystem traversal using external iteration looks like

File[] files = dir.listFiles();
for ( File f : files ) {
    if ( f.isDirectory() ) {
        this.processDir( f );
    } else if ( f.isFile() ) {
        this.processFile( f );
    } else {
        this.processErr( f );
    }
}

With streams and internal iteration this becomes

Arrays.stream( dir.listFiles() )
    .filter( StreamProc.process( f -> f.isDirectory(), this::processDir ) )
    .filter( StreamProc.process( f -> f.isFile(), this::processFile ) )
    .forEach( f -> this::processErr );

I would like Stream to implement the process method directly. Then we could have

Arrays.stream( dir.listFiles() )
    .process( f -> f.isDirectory(), this::processDir ) )
    .process( f -> f.isFile(), this::processFile ) )
    .forEach( f -> this::processErr );

Thoughts?

user_72918470

It seems that in reality you do want to process each line, but process it differently based on some condition (type).

I think this is more or less functional way to implement it would be:

public static void main(String[] args) {
    Arrays.stream(new int[] {1,2,3,4}).map(i -> processor(i).get()).forEach(System.out::println);
}

static Supplier<Integer> processor(int i) {
    return tellType(i) ? () -> processTypeA(i) : () -> processTypeB(i);
}

static boolean tellType(int i) {
    return i % 2 == 0;
}

static int processTypeA(int i) {
    return i * 100;
}

static int processTypeB(int i) {
    return i * 10;
}
user_5133420

@tom

What about this:

Arrays.stream( dir.listFiles() )
    .peek(  f -> { if(f.isDirectory()) { processDir(f); }} )
    .peek(  f -> { if(f.isFile())      { processFile(f);}}) )
    .forEach( f -> this::processErr );
user_59069390
Related