Java Streams - forEach with pre-Action and post-Action

user_4945535 0
Viewed 229

When using Stream.forEach(), I was thinking if it is not possible to add a pre-action and post-action to be executed when the stream is not empty. For example, when printing a List, one could prepend something or write something else when the stream is empty.

Now I came up with something like

private static <T> void forEach(Stream<T> stream, Consumer<? super T> action,
    Runnable preAction, Runnable postAction, Runnable ifEmpty) {
    AtomicBoolean hasElements = new AtomicBoolean(false);
    Consumer<T> preActionConsumer = x -> {
        if (hasElements.compareAndSet(false, true)) {
            preAction.run();
        }
    };
    stream.forEach(preActionConsumer.andThen(action));
    if (hasElements.get()) {
        postAction.run();
    } else {
        ifEmpty.run();
    }
}

For sequential streams, this should work, should it not? Is this method correct, is it as "good idea" have such a method or are there any caveats?

This does not work for parallel streams, since the preAction might be slower than another thread executing the action, but implementing it correctly without resorting to synchronized or other concurrency utils which defeat the purpose of parallel streams will probably be not easy...

edit: adding use-case. Reading searching integers from a file using a regex and writing them to another file. Using this approach, I don't have to create a String in memory and after that write it to some file. (Obviously, for my real task, I am using more complicated regexes.)

public static void main(String[] args) throws IOException {
    Stream<String> lines = Files.lines(Paths.get("foo.txt"));

    Pattern findInts = Pattern.compile("(\\d+)");
    Path barFile = Paths.get("bar.txt");
    try (BufferedWriter writer = Files.newBufferedWriter(barFile , StandardOpenOption.CREATE_NEW)) {
        lines.flatMap(x -> findInts.matcher(x).results())
                .forEach(x-> convertCheckedIOException(() ->  {
                            writer.write(x.group(1));
                            writer.newLine();
                        })
                );
    }
}

public static void convertCheckedIOException(Run r) {
    try {
        r.run();
    } catch (IOException e) {
        throw new UncheckedIOException(e);
    }
}

interface Run {
    void run() throws IOException;
}
java-stream java
2 Answers

Use the right tool for your job. This task does not benefit from the Stream API.

Pattern intPattern = Pattern.compile("\\d+");
try(Scanner scanner = new Scanner(Paths.get("foo.txt"));
    BufferedWriter writer = Files.newBufferedWriter(Paths.get("bar.txt"), CREATE_NEW)) {

    String s = scanner.findWithinHorizon(intPattern, 0);
    if(s == null) {
        // perform empty action
    } else {
        // perform pre action
        do {
            writer.append(s);
            writer.newLine();
        } while( (s=scanner.findWithinHorizon(intPattern, 0)) != null);
        // perform post action
    }
}

You could still bring in a Stream operation, e.g.

Pattern intPattern = Pattern.compile("\\d+");
try(Scanner scanner = new Scanner(Paths.get("foo.txt"));
    BufferedWriter writer = Files.newBufferedWriter(Paths.get("bar.txt"), CREATE_NEW)) {

    String firstLine = scanner.findWithinHorizon(intPattern, 0);
    if(firstLine == null) {
        // perform empty action
    } else {
        // perform pre action
        Stream.concat(Stream.of(firstLine),
                      scanner.findAll(intPattern).map(MatchResult::group))
            .forEach(line -> convertCheckedIOException(() ->  {
                    writer.write(line);
                    writer.newLine();
                })
            );
        // perform post action
    }
}

but having to deal with the checked IOException just complicates the code for no benefit.

user_27114880

I like the idea of having a util like this. At first I thought using a second flag that is set/unset by the preaction and halts the action might be enough. But what makes it more complicated is the fact that preAction is put in front of every action call, not just the first one.

I came up with a synchronized solution that enforces the order pre, actions, post/empty. One caveat is that the first batch of parallel threads will have to wait for the first of them to finish as they will run into synchronized:

private static <T> void forEach(Stream<T> stream, Consumer<? super T> action, Runnable preAction, Runnable postAction, Runnable ifEmpty)
{
    AtomicBoolean hasElements = new AtomicBoolean(false);

    stream.forEach(new Consumer<T>()
    {
        private Consumer<? super T> delegate = new Consumer<T>()
        {
            private Consumer<? super T> delegate2 = new Consumer<T>()
            {
                @Override
                public void accept(T x)
                {
                    System.out.println("check");
                    hasElements.set(true);
                    preAction.run();
                    action.accept(x);
                    delegate2 = action; // rest of first batch won't run preAction anymore
                    delegate = action; // next batches won't even synchronize anymore
                }
            };

            @Override
            public void accept(T x)
            {
                synchronized (this)
                {
                    delegate2.accept(x);
                }
            }
        };

        @Override
        public void accept(T x)
        {
            delegate.accept(x);
        }
    });

    if (hasElements.get()) { postAction.run(); } else { ifEmpty.run(); }
}

public static void main(String[] args)
{
    Stream<Integer> s = Stream.generate(() -> 1).limit(1000).parallel();
    forEach(s, i -> System.out.println(Thread.currentThread().getId()), () -> System.out.println("pre"),
            () -> System.out.println("post"), () -> System.out.println("empty"));
}

Output:
check
pre
...
many thread IDs
...
post
user_76530730
Related